返回文章列表

单独处理md的chunk分割,注入更多metadata

4 min read

在写md文件的chunker之前,先将之前的fileloader修改一下 UnstructuredMarkdownLoader修改成TextLoader,因为UnstructuredMarkdownLoader会将 Markdown 格式标记抽离,所以换成langchain的MarkdownHeaderTextSplitter比较好

PYTHON
左右滑动查看完整代码
import hashlib
from dataclasses import dataclass
from typing import Callable, List

from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter


@dataclass
class MarkdownSection:
    """Markdown 标题切分后的章节块。"""

    title_path: str
    section_title: str
    heading_level: int
    section_index: int
    content: str


class MarkdownChunker:
    """
    Markdown 专用切分器。

    Markdown 和普通纯文本不一样,它的 `# / ## / ###` 标题本身就是天然的父级上下文。
    标题解析交给 LangChain 的 MarkdownHeaderTextSplitter 处理
    这里主要负责标题路径增强和语义二次切分。
    """

    def __init__(
        self,
        semantic_split_text: Callable[[str], List[str]],
        max_chunk_size: int,
    ) -> None:
        """
        Args:
            semantic_split_text: 复用现有语义切分函数,避免在 MarkdownChunker 里重复造轮子。
            max_chunk_size: section 超过该长度时,再做语义二次切分。
        """
        self.semantic_split_text = semantic_split_text
        self.max_chunk_size = max_chunk_size
        self.header_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[
                ("#", "h1"),
                ("##", "h2"),
                ("###", "h3"),
                ("####", "h4"),
                ("#####", "h5"),
                ("######", "h6"),
            ],
            strip_headers=False,
        )

    def split_document(self, document: Document) -> List[Document]:
        """把单个 Markdown 文档切成带标题上下文的 chunk。"""
        sections = self._split_sections(document.page_content)
        if not sections:
            return []

        chunks: List[Document] = []
        global_chunk_index = 0
        for section in sections:
            section_chunks = self._split_section_content(section.content)
            section_id = self._build_section_id(document, section)
            for child_chunk_index, chunk_text in enumerate(section_chunks):
                metadata = dict(document.metadata)
                metadata.update(
                    {
                        "title_path": section.title_path,
                        "section_title": section.section_title,
                        "heading_level": section.heading_level,
                        "section_index": section.section_index,
                        "section_id": section_id,
                        "child_chunk_index": child_chunk_index,
                        "chunk_index": global_chunk_index,
                    }
                )

                page_content = self._build_chunk_content(section.title_path, chunk_text)
                metadata["chunk_size"] = len(page_content)
                chunks.append(Document(page_content=page_content, metadata=metadata))
                global_chunk_index += 1

        return chunks

    def _split_sections(self, text: str) -> List[MarkdownSection]:
        """用 LangChain 按 Markdown 标题切 section,并计算统一的标题路径。"""
        sections: List[MarkdownSection] = []
        header_docs = self.header_splitter.split_text(text)

        for section_index, header_doc in enumerate(header_docs):
            content = header_doc.page_content.strip()
            if not content:
                continue

            # MarkdownHeaderTextSplitter 会把命中的标题层级放到 metadata,
            # 例如 {"h1": "宫保鸡丁的做法", "h2": "计算"}。
            # 我们把它转成统一的 title_path,方便检索、重排和 prompt 展示。
            title_items = [
                (int(key[1:]), value)
                for key, value in header_doc.metadata.items()
                if key.startswith("h") and key[1:].isdigit()
            ]
            title_items.sort(key=lambda item: item[0])
            title_path_parts = [title for _, title in title_items]
            title_path = " > ".join(title_path_parts) if title_path_parts else "无标题"
            heading_level = title_items[-1][0] if title_items else 0
            section_title = title_path_parts[-1] if title_path_parts else "无标题"

            sections.append(
                MarkdownSection(
                    title_path=title_path,
                    section_title=section_title,
                    heading_level=heading_level,
                    section_index=section_index,
                    content=content,
                )
            )
        return sections

    def _split_section_content(self, content: str) -> List[str]:
        """section 内部太长时再语义切分,短 section 保持完整。"""
        if len(content) <= self.max_chunk_size:
            return [content]

        # 标题切分负责保留 Markdown 结构,语义切分负责控制 chunk 大小。
        # 这样既不会把大章节整个塞进向量库,也不容易把列表项切成没上下文的孤儿句。
        return self.semantic_split_text(content)

    def _build_chunk_content(self, title_path: str, chunk_text: str) -> str:
        """把标题路径写进 page_content,让 embedding、BM25、reranker 都能看到结构信息。"""
        # 短 chunk 单独看语义很弱,比如“莴笋 = 约 250g”。
        # 加上标题路径后,它就变成“宫保鸡丁 > 计算”下面的用量信息,相关性会稳定很多。
        return f"标题路径:{title_path}\n内容:\n{chunk_text.strip()}"

    def _build_section_id(self, document: Document, section: MarkdownSection) -> str:
        """生成同一 section 的稳定标识,后续可用于同 section 父子扩展。"""
        source = document.metadata.get("source_file") or document.metadata.get("file_name", "")
        raw_key = f"{source}|{section.section_index}|{section.title_path}"
        return hashlib.md5(raw_key.encode("utf-8")).hexdigest()

还需要修改一下之前chain里面的格式化函数,将metadata的标题路径注入到返回的doc里面,这样单独的chunk

就算很小,也会带上路径信息,这样小而关键的chunk就不会被忽略了

PYTHON
左右滑动查看完整代码
def _format_docs(self, docs: List[Document]) -> str:
        """
        格式化检索到的文档
        
        Args:
            docs: 文档列表
            
        Returns:
            str: 格式化后的文档文本
        """
        formatted = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("file_name", "未知来源")
            title_path = doc.metadata.get("title_path")
            content = doc.page_content.strip()
            if title_path:
                formatted.append(f"[文档 {i}] (来源: {source}, 标题路径: {title_path})\n{content}")
            else:
                formatted.append(f"[文档 {i}] (来源: {source})\n{content}")

看一下效果

PYTHON
左右滑动查看完整代码
请输入您的问题: 我需要为家人准备一顿饭,包括用极光咖啡壶做咖啡和制作宫保鸡丁。请告诉我:1. 咖啡壶首次使用需要注意什么?2. 宫保鸡丁复杂版本需要哪些
配料和用量?3. .两个操.作过程中有哪些安全注意事项?

 正在思考...

2026-05-27 11:22:39.703 | INFO     | multi_functional_chain:ask:217 - 收到问题:我需要为家人准备一顿饭,包括用极光咖啡壶做咖啡和制作宫保鸡丁。请告
诉我:1. 咖啡壶首次使用需要注意什么?2. 宫保鸡丁复杂版本需要哪些配料和用量?3. .两个操.作过程中有哪些安全注意事项?
2026-05-27 11:22:42.643 | INFO     | multi_functional_chain:ask:221 - 生成的多路查询(含原始问题):['我需要为家人准备一顿饭,包括用极光咖啡壶做咖
啡和制作宫保鸡丁。请告诉我:1. 咖啡壶首次使用需要注意什么?2. 宫保鸡丁复杂版本需要哪些配料和用量?3. .两个操.作过程中有哪些安全注意事项?', '如何
正确使用新的极光咖啡壶以及制作经典版宫保鸡丁的详细步骤和安全提示', '首次使用极光咖啡机的注意事项与复杂版宫保鸡丁的配料清单及操作安全建议', '用极光
咖啡壶煮咖啡前的准备事项和正宗宫保鸡丁的食材用量及烹饪过程中的安全须知']
2026-05-27 11:22:42.885 | INFO     | vector:similarity_search_with_score:75 - 相似度搜索完成: query='我需要为家人准备一顿饭,包括用极光咖啡壶做咖
啡和制作宫保鸡丁。请告诉我:1. 咖啡壶首次使用需要注意什么?2. 宫保鸡丁复杂版本需要哪些配料和用量?3. .两个操.作过程中有哪些安全注意事项?', result
s=21
2026-05-27 11:22:42.885 | INFO     | hybrid_retriever:search:266 - 混合检索完成: BM25=21, 向量=21, 融合后=10
2026-05-27 11:22:43.063 | INFO     | vector:similarity_search_with_score:75 - 相似度搜索完成: query='如何正确使用新的极光咖啡壶以及制作经典版宫保
鸡丁的详细步骤和安全提示', results=21
2026-05-27 11:22:43.063 | INFO     | hybrid_retriever:search:266 - 混合检索完成: BM25=20, 向量=21, 融合后=10
2026-05-27 11:22:43.229 | INFO     | vector:similarity_search_with_score:75 - 相似度搜索完成: query='首次使用极光咖啡机的注意事项与复杂版宫保鸡丁
的配料清单及操作安全建议', results=21
2026-05-27 11:22:43.229 | INFO     | hybrid_retriever:search:266 - 混合检索完成: BM25=20, 向量=21, 融合后=10
2026-05-27 11:22:43.392 | INFO     | vector:similarity_search_with_score:75 - 相似度搜索完成: query='用极光咖啡壶煮咖啡前的准备事项和正宗宫保鸡丁
的食材用量及烹饪过程中的安全须知', results=21
2026-05-27 11:22:43.393 | INFO     | hybrid_retriever:search:266 - 混合检索完成: BM25=20, 向量=21, 融合后=10
Warning: You are sending unauthenticated requests to the HF Hub. Please set a HF_TOKEN to enable higher rate limits and faster downloads.
Loading weights: 100%|██████████████████████████████████████████████████████████████
████████████████████████| 201/201 [00:00<00:00, 11167.62it/s]
2026-05-27 11:22:52.026 | INFO     | reranker:_load_model:53 - 重排序模型加载完成: BAAI/bge-reranker-base
2026-05-27 11:22:55.262 | INFO     | reranker:rerank:84 - 重排序完成: 14 -> 3 个文档, 最高分=0.9976
 回答:
根据提供的文档,我无法回答关于极光咖啡壶首次使用需要注意的事项,因为相关的信息并未包含在参考文档中。以下是对宫保鸡丁复杂版本所需配料和用量的回答,
以及操作过程中的安全注意事项。

---

### **1. 宫保鸡丁复杂版本需要哪些配料和用量?**

根据文档内容,宫保鸡丁复杂版本的配料可以分为**必须配料****进阶配料****可选配料**三类:

#### 必须配料:
- 手枪腿(或鸡胸脯肉):1 支(约 350g
- 大葱:1 根(约 180g
- 熟花生:150g
- 姜片:10g
- 干辣椒(或二荆条):10g(若选择二荆条,则需要大约 4 支)
- 生抽酱油:10g
- 白糖:2g
- 盐:2g
- 植物油:20g
- 淀粉:15g
- 料酒:15g

#### 进阶配料:
- 老抽酱油:5g
- 花椒:5g
- 香醋:5g
- 鸡精:2g
- 芝麻油:10g
- 淀粉(用于勾芡):10g
- 豆瓣酱:10g

#### 可选配料:
- 莴笋:约 250g
- 油泼辣子:5g

这些配料可以根据个人口味进行调整,例如增加辣味可以选择多加干辣椒或油泼辣子。

---

### **2. 操作过程中有哪些安全注意事项?**

#### 宫保鸡丁制作的安全注意事项:
1. **热油操作**
   - 文档提到“转大火,倒入 20g 植物油,7 成热(竹筷子起泡)下入鸡丁”。加热油时需特别注意火候,避免油温过高导致油溅或起火。
   - 在翻炒过程中,尽量使用长柄锅铲,保持手部与热源的距离。

2. **高温焖煮**
   - 文档提到“盖上锅盖,转中小火焖 2 分钟”。焖煮时需确保锅盖稳固,防止蒸汽烫伤。

3. **淀粉勾芡**
   - 文档提到“淀粉 10g50g 清水调成水淀粉”。调制水淀粉时需搅拌均匀,避免结块导致烹饪过程中糊锅。

4. **熟花生处理**
   - 熟花生应在最后阶段加入,避免长时间高温加热导致口感变差。

#### 极光咖啡壶使用的安全注意事项:
由于文档未提及咖啡壶的具体信息,以下是通用的安全建议:
1. **首次使用清洁**
   - 首次使用前应彻底清洗咖啡壶内部,尤其是滤网和壶体,以去除可能存在的生产残留物。

2. **加热安全**
   - 使用电热咖啡壶时,确保电源线完好无损,避免过热或短路。
   - 如果是明火加热的咖啡壶,需注意火焰不要超出壶底范围,以免烧坏壶体。

3. **防烫措施**
   - 咖啡壶在加热后会变得非常烫手,操作时需佩戴隔热手套或使用防烫工具。

---

### **总结**
1. **宫保鸡丁复杂版本**需要准备上述列出的必须、进阶和可选配料,并根据口味调整用量。
2. **安全注意事项**包括热油操作、高温焖煮、淀粉勾芡等环节的防护,以及咖啡壶使用时的清洁、加热和防烫措施。

如果需要更详细的咖啡壶使用说明,请提供相关文档或具体型号信息以便进一步解答。

可以了