RAG 文档分块策略：从原理到最佳实践

本文系统梳理了 RAG（检索增强生成）系统中 12 种主流文档分块策略，涵盖固定大小分块、结构感知分块、语义感知分块和层级多粒度分块四大类别。每种策略均包含原理剖析、完整代码示例、优劣势对比和适用场景分析。此外，本文还深入探讨了 chunk 参数选择方法论、"Lost in the Middle"问题与排序策略、分块质量评估体系、多模态文档处理以及生产环境工程实践，帮助开发者在实际项目中做出最优选择。

目录

  1. 引言：为什么文档分块是 RAG 系统的基石
  2. 第一类：固定大小分块
  3. 第二类：结构感知分块
  4. 第三类：语义感知分块
  5. 第四类：层级多粒度分块
  6. 十二种方案综合对比
  7. Chunk 参数选择深度指南
  8. Lost in the Middle 问题与 Chunk 排序策略
  9. Chunk 质量评估体系
  10. 多模态文档的分块考量
  11. Chunk 元数据增强策略
  12. 业界最佳实践：分层组合策略
  13. 真实场景案例研究
  14. 生产环境工程实践
  15. 总结与建议

1. 引言：为什么文档分块是 RAG 系统的基石

1.1 RAG 的核心矛盾

RAG（Retrieval-Augmented Generation，检索增强生成）系统的核心工作流程可以概括为：

用户提问 → 检索相关文档片段 → 将片段注入 LLM 上下文 → LLM 生成回答

这个流程中存在一个根本性的矛盾：

  • Embedding 模型擅长处理短文本（通常 512 tokens 以内效果最佳），文本越长，向量表示越"模糊"——这被称为"表示稀释"（Representation Dilution）现象
  • LLM 需要充分的上下文才能给出准确回答，上下文越完整，回答质量越高
  • LLM 的上下文窗口虽然越来越大（GPT-4 支持 128K tokens，Claude 3 支持 200K tokens），但输入越长，成本越高、速度越慢、注意力越分散

文档分块（Chunking）就是解决这个矛盾的关键技术：将长文档切分成适当大小的片段，使得每个片段既能被 Embedding 模型精确编码，又能为 LLM 提供足够的上下文信息。

1.2 分块质量如何影响 RAG 系统

分块策略的好坏直接影响 RAG 系统的三个核心指标：

指标 分块过大的影响 分块过小的影响
召回率（Recall） 向量表示模糊，检索不准，相关文档可能漏掉 信息碎片化，关键信息分散在多个 chunk，单个 chunk 匹配度低
精确率（Precision） 包含大量无关信息，噪音多，检索结果不精确 上下文不足，LLM 无法准确理解用户意图
答案质量（Answer Quality） LLM 注意力分散，可能遗漏关键信息或产生幻觉 信息不完整，LLM 可能基于不完整信息产生错误推理

一个理想的分块策略应该做到：

  1. 语义完整性：每个 chunk 是一个自包含的语义单元，不依赖外部上下文即可理解
  2. 大小适中：既不超过 Embedding 模型的最佳输入长度，也不因过短而丢失上下文
  3. 边界合理：在自然语义边界处切分，而非在句子中间硬切断
  4. 结构感知：充分利用文档自身的结构信息（标题、段落、表格等）
  5. 信息密度均衡：每个 chunk 包含的信息量大致相当，避免某些 chunk 信息过载而另一些空洞

1.3 十二种分块策略全景图

                          文档分块策略
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
   固定大小分块            结构感知分块           语义感知分块
   (Size-based)        (Structure-aware)      (Semantic-aware)
        │                     │                     │
   ┌────┼────┐          ┌────┼────┐          ┌────┼────┐
   │    │    │          │    │    │          │    │    │
 字符  Token 滑动    章节  HTML  文档     Embedding LLM  命题
 切分  切分  窗口    标题 标记  特定     相似度   驱动  提取
                           解析  解析              分块
        │                     │                     │
        └─────────────────────┼─────────────────────┘
                              │
                     层级多粒度分块
                   (Hierarchical)
                              │
                        ┌────┼────┐
                        │         │
                     父子文档   层级分块

1.4 阅读建议

  • 初学者：建议从第 2 章开始顺序阅读，理解每种策略的原理和适用场景
  • 有经验的开发者：可以直接跳到第 6 章查看综合对比，然后阅读第 7-12 章的进阶内容
  • 架构师/技术负责人：重点关注第 7 章（参数选择）、第 9 章（评估体系）、第 12 章（组合策略）和第 14 章（工程实践）

2. 第一类：固定大小分块

固定大小分块是最基础的分块策略，核心思想是按照预设的固定长度（字符数或 token 数）对文档进行切分，不考虑文档的内容和结构。

2.1 固定字符数切分（Character Splitter）

原理

按固定字符数对文本进行硬切分，配合 overlap（重叠）参数来缓解边界处的信息断裂问题。这是最原始、最简单的分块方式。

代码示例
from langchain_text_splitters import CharacterTextSplitter

text = "今天天气很好，我们去公园散步。公园里有很多花，非常漂亮。下午我们还去了咖啡馆，喝了一杯拿铁。"

splitter = CharacterTextSplitter(
    separator="",       # 不指定分隔符，按字符硬切
    chunk_size=20,      # 每个 chunk 最多 20 个字符
    chunk_overlap=5,    # 相邻 chunk 重叠 5 个字符
    length_function=len,
)

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}: {chunk}")

输出结果：

Chunk 1: 今天天气很好，我们去公园散
Chunk 2: 园散步。公园里有很多花，非
Chunk 3: 很多花，非常漂亮。下午我们
Chunk 4: 下午我们还去了咖啡馆，喝了
Chunk 5: 喝了一杯拿铁。
问题分析

从输出可以清晰看到固定字符切分的致命缺陷：

  • Chunk 1 和 Chunk 2 在"公园散"和"园散步"之间硬切断，一个完整的词被拆成两半
  • Chunk 3 和 Chunk 4 的 overlap 导致"下午我们"重复出现，浪费存储空间
  • 句子"公园里有很多花，非常漂亮"被拆散到 Chunk 2 和 Chunk 3 中，语义破碎
  • 中文没有空格分隔词，硬切造成的破坏比英文更严重
综合评价
维度 评分 说明
实现难度 ★☆☆☆☆ 一行代码即可实现
分块质量 ★☆☆☆☆ 句子中间硬切断，语义破碎严重
处理速度 ★★★★★ 纯字符串操作，毫秒级完成
存储效率 ★★☆☆☆ overlap 机制导致存储膨胀
检索精度 ★★☆☆☆ 语义碎片化，检索效果差
适用场景

仅适用于原型验证阶段或对检索精度要求极低的场景。生产环境中几乎不会单独使用。

2.2 递归字符切分（RecursiveCharacterTextSplitter）

原理

递归字符切分是 LangChain 的默认分块器，也是目前使用最广泛的通用分块方案。它的核心思想是设定一个分隔符优先级列表，逐级尝试在自然边界处断开：

分隔符优先级（从高到低）：
\n\n  →  \n  →  。  →  ！  →  ？  →  .  →  !  →  ?  →  空格  →  字符
段落    换行    句号   感叹   问号   英文   英文   英文   词边界   兜底

切分流程如下：

  1. 首先尝试在段落边界（\n\n）处切分
  2. 如果切出的段落仍然超过 chunk_size，降级到换行符（\n）
  3. 如果仍然超长，继续降级到句号（。）
  4. 以此类推，直到在某个级别成功切分
  5. 如果所有分隔符都无法切分（极端情况），最终按字符硬切

这种"逐级降级"机制是递归字符切分区别于固定字符切分的核心优势。

代码示例
from langchain_text_splitters import RecursiveCharacterTextSplitter

text = """第一章 总则

第一条 本条例适用于全体员工，包括正式员工、试用期员工和实习生。

第二条 员工应当遵守公司各项规章制度，维护公司利益。

第二章 考勤管理

第三条 工作时间为上午9:00至下午6:00，午休时间为12:00至13:00。

第四条 员工上下班需打卡记录考勤，忘记打卡需在当日补卡。"""

splitter = RecursiveCharacterTextSplitter(
    chunk_size=80,
    chunk_overlap=15,
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    length_function=len,
)

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} ({len(chunk)} 字符) ---")
    print(chunk)
    print()

输出结果：

--- Chunk 1 (12 字符) ---
第一章 总则

--- Chunk 2 (42 字符) ---
第一条 本条例适用于全体员工，包括正式员工、试用期员工和实习生

--- Chunk 3 (30 字符) ---
第二条 员工应当遵守公司各项规章制度，维护公司利益

--- Chunk 4 (12 字符) ---
第二章 考勤管理

--- Chunk 5 (42 字符) ---
第三条 工作时间为上午9:00至下午6:00，午休时间为12:00至13:00

--- Chunk 6 (38 字符) ---
第四条 员工上下班需打卡记录考勤，忘记打卡需在当日补卡
优势分析

相比固定字符切分，递归字符切分有显著改进：

  1. 优先在段落边界切分：保留了文档的段落结构
  2. 逐级降级机制：不会在词中间硬切断
  3. 中文友好：支持中文标点（。！？）作为分隔符
  4. 通用性强：适用于绝大多数纯文本文档
局限性
  1. 不理解文档结构：不知道哪里是"章节"、哪里是"条款"，只是机械地按分隔符切
  2. 对无换行文档退化：如果文档没有换行符，会直接降级到句号甚至字符级别
  3. overlap 浪费：重叠部分可能包含不完整的信息
  4. 分隔符顺序固定：无法根据文档类型动态调整优先级
综合评价
维度 评分 说明
实现难度 ★★☆☆☆ LangChain 内置，开箱即用
分块质量 ★★★☆☆ 尽量在自然边界切，但不理解语义
处理速度 ★★★★☆ 很快，纯文本处理
存储效率 ★★★☆☆ overlap 可控
检索精度 ★★★☆☆ 比纯字符切好很多
适用场景

通用文档的兜底方案。当其他高级分块策略无法处理时，递归字符切分是最可靠的 fallback。在实际项目中，通常作为组合拳策略的最后一环。

2.3 Token 级别切分（TokenTextSplitter）

原理

Token 级别切分不按字符数而是按 token 数来切分文本。这是因为 LLM 的计费和上下文限制都是按 token 计算的，而非字符数。

关键认知：不同语言的 token 密度差异巨大。

语言 1000 字符 ≈ ? tokens 说明
英文 250-300 tokens 一个单词通常 1-2 个 token
中文 600-700 tokens 一个汉字通常 1-2 个 token
日文 500-600 tokens 假名和汉字混合
代码 200-400 tokens 取决于语言和格式

这意味着：如果按字符数设置 chunk_size=1000，英文文档的 chunk 约 250 tokens（远低于 LLM 限制），而中文文档的 chunk 约 650 tokens（接近限制）。使用 Token 切分可以精确控制每个 chunk 的 token 消耗。

代码示例
from langchain_text_splitters import TokenTextSplitter

text = """RAG（Retrieval-Augmented Generation）是一种结合检索和生成的 AI 技术架构。
它通过从外部知识库中检索相关文档片段，将其作为上下文注入大语言模型，
从而生成更准确、更可靠的回答。这种架构有效解决了 LLM 的知识截止日期问题和幻觉问题。"""

# 使用 GPT-4 的 tokenizer（cl100k_base）
splitter = TokenTextSplitter(
    chunk_size=30,          # 每个 chunk 最多 30 个 token
    chunk_overlap=5,        # 重叠 5 个 token
    encoding_name="cl100k_base",
)

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} ---")
    print(chunk)
    print()
不同 Tokenizer 的选择
# OpenAI 模型
TokenTextSplitter(encoding_name="cl100k_base")   # GPT-4, GPT-3.5-turbo, text-embedding-3
TokenTextSplitter(encoding_name="p50k_base")     # GPT-3 (davinci)
TokenTextSplitter(encoding_name="o200k_base")    # GPT-4o, o1

# 通过 transformers 使用任意 tokenizer
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-7B")
splitter = TokenTextSplitter.from_huggingface_tokenizer(
    tokenizer=tokenizer,
    chunk_size=512,
    chunk_overlap=50,
)
综合评价
维度 评分 说明
实现难度 ★★☆☆☆ LangChain 内置
分块质量 ★★★☆☆ 精确控制 LLM 输入长度
处理速度 ★★★★☆ 需要 tokenizer，但很快
存储效率 ★★★☆☆ 与递归字符切类似
检索精度 ★★★☆☆ 与递归字符切类似
适用场景
  • 需要精确控制 LLM token 消耗的场景
  • 英文文档为主的项目（token 和字符差异大）
  • 中文场景收益相对较小（token 密度高，差异不大）

2.4 滑动窗口切分（Sliding Window）

原理

滑动窗口切分不是将文档切成互不重叠的块，而是用一个固定大小的窗口在文档上滑动，每个窗口位置生成一个 chunk。相邻 chunk 之间存在大量重叠。

原文：ABCDEFGHIJKLMNOPQRSTUVWXYZ
窗口大小 = 8，步长 = 3

窗口1: [ABCDEFGH]
窗口2:    [DEFGHIJK]
窗口3:       [GHIJKLMN]
窗口4:          [JKLMNOPQ]
窗口5:             [MNOPQRST]
窗口6:                [PQRSTUVW]
窗口7:                   [STUVWXYZ]
代码示例
def sliding_window_chunk(text: str, window_size: int, step: int) -> list:
    """简单的滑动窗口分块实现"""
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + window_size, len(text))
        chunk = text[start:end]
        chunks.append(chunk)
        start += step
    return chunks

text = "今天天气很好，我们去公园散步。公园里有很多花，非常漂亮。"
chunks = sliding_window_chunk(text, window_size=12, step=5)

for i, chunk in enumerate(chunks):
    print(f"窗口 {i+1}: [{chunk}]")

输出结果：

窗口 1: [今天天气很好，我们去公园]
窗口 2: [很好，我们去公园散步。]
窗口 3: [去公园散步。公园里有很]
窗口 4: [步。公园里有很多花，非]
窗口 5: [有很多花，非常漂亮。]
核心参数
参数 含义 典型值 影响
window_size 窗口大小（字符数） 256-512 越大上下文越完整，但检索精度越低
step 滑动步长（字符数） window_size / 3 越小重叠越多，召回率越高但存储越大
综合评价
维度 评分 说明
实现难度 ★★☆☆☆ 简单循环即可实现
分块质量 ★★★☆☆ 不会漏掉任何信息
处理速度 ★★★☆☆ 生成大量冗余 chunk
存储效率 ★☆☆☆☆ 存储膨胀 3-5 倍
检索精度 ★★★★☆ 高重叠带来高召回率
适用场景
  • 对召回率要求极高、不在乎存储成本的场景
  • 短文档（几百字以内）的精细检索
  • 实时流式文本处理（如会议转录）

3. 第二类：结构感知分块

结构感知分块利用文档自身的结构信息（标题层级、HTML 标签、文档格式元数据等）来指导切分，使得每个 chunk 对应一个完整的结构单元。这是性价比最高的分块策略类别。

3.1 章节标题感知切分（Section-Aware / Heading Splitter）

原理

识别文档中的标题结构，在标题边界处切分，保证每个 chunk 是一个完整的章节或条款。这是处理结构化文档（法律、制度、合同、技术手册）的最佳策略。

标题识别模式：

模式 正则表达式 匹配示例
中文章节 第[一二三...\d]+[章节条款] 第一章、第二条、第三节
数字编号 \d+\.\s+.+ 1. 总则、2.1 适用范围
多级编号 \d+\.\d+\.?\s+.+ 1.1.1 子条款
Markdown ^#{1,3}\s+.+ ## 概述、### 详细说明
括号编号 （[一二三\d]+） （一）、（1）
罗马数字 [ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+\.? Ⅰ.、Ⅱ.
代码示例
import re
from typing import List

class SectionAwareSplitter:
    """按章节结构切分文档，优先在标题处断开"""

    # 四级标题匹配模式（优先级从高到低）
    SECTION_PATTERNS = [
        # 一级：中文章节标题 + 数字编号
        re.compile(
            r"^(第?[一二三四五六七八九十百零\d]+[章节条款]"
            r"|[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+\.?"
            r"|\d+\.)\s+.+",
            re.MULTILINE
        ),
        # 二级：1.1 格式
        re.compile(r"^\d+\.\d+\.?\s+.+", re.MULTILINE),
        # 三级：1.1.1 格式
        re.compile(r"^\d+\.\d+\.\d+\.?\s+.+", re.MULTILINE),
        # 四级：Markdown 标题
        re.compile(r"^#{1,3}\s+.+", re.MULTILINE),
    ]

    def __init__(self, chunk_size: int = 800, chunk_overlap: int = 150):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_text(self, text: str) -> List[str]:
        """主切分逻辑"""
        if not text or len(text) <= self.chunk_size:
            return [text] if text else []

        # 第一轮：按一级标题切分
        sections = self._split_by_pattern(text, level=0)
        chunks = []

        for section in sections:
            if len(section) <= self.chunk_size:
                chunks.append(section)
            else:
                # 第二轮：超长段落按二级标题再切
                sub_sections = self._split_by_pattern(section, level=1)
                for sub in sub_sections:
                    if len(sub) <= self.chunk_size:
                        chunks.append(sub)
                    else:
                        # 第三轮：仍超长则递归字符切兜底
                        chunks.extend(self._recursive_fallback(sub))

        return self._merge_short_chunks(chunks)

    def _split_by_pattern(self, text: str, level: int) -> List[str]:
        """按指定级别的标题模式切分"""
        if level >= len(self.SECTION_PATTERNS):
            return [text]

        pattern = self.SECTION_PATTERNS[level]
        matches = list(pattern.finditer(text))

        if not matches:
            return [text]

        sections = []
        # 保留第一个标题之前的前言内容
        if matches[0].start() > 0:
            preamble = text[:matches[0].start()].strip()
            if preamble:
                sections.append(preamble)

        for i, match in enumerate(matches):
            start = match.start()
            end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
            sections.append(text[start:end].strip())

        return sections

    def _recursive_fallback(self, text: str) -> List[str]:
        """兜底：递归字符切分"""
        from langchain_text_splitters import RecursiveCharacterTextSplitter
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
        )
        return splitter.split_text(text)

    def _merge_short_chunks(self, chunks: List[str]) -> List[str]:
        """合并过短的 chunk（< 30% chunk_size）"""
        if not chunks:
            return chunks
        merged = []
        buffer = chunks[0]
        for chunk in chunks[1:]:
            if (len(buffer) < self.chunk_size * 0.3
                    and len(buffer) + len(chunk) <= self.chunk_size * 1.5):
                buffer = buffer + "\n\n" + chunk
            else:
                merged.append(buffer)
                buffer = chunk
        merged.append(buffer)
        return merged
切分效果演示

输入文档：

1. 总则
第一条 本条例适用于全体员工，包括正式员工、试用期员工和实习生。
第二条 员工应当遵守公司各项规章制度，维护公司利益。

2. 考勤管理
第三条 工作时间为上午9:00至下午6:00，午休时间为12:00至13:00。
第四条 员工上下班需打卡记录考勤，忘记打卡需在当日补卡。

切分结果（chunk_size=100）：

Chunk 1: "1. 总则\n第一条 本条例适用于全体员工...\n第二条 员工应当遵守..."
Chunk 2: "2. 考勤管理\n第三条 工作时间为上午9:00至下午6:00...\n第四条 员工上下班需打卡..."

每个 chunk 对应一个完整的章节，条款信息完整无缺。

综合评价
维度 评分 说明
实现难度 ★★★☆☆ 需要编写和维护正则表达式
分块质量 ★★★★★ 制度/法律/合同类文档的最优方案
处理速度 ★★★★☆ 正则匹配非常快
存储效率 ★★★★☆ 按章节切分，几乎没有浪费
检索精度 ★★★★★ 条款级精准定位
适用场景
  • 法律条文、公司制度、合同协议
  • 技术文档、API 文档、产品手册
  • 任何有明确标题层级结构的文档
局限性
  • 依赖正则表达式，文档格式变化时需要调整
  • 对无标题结构的文档（如小说、散文）完全无效
  • 不同语言/文化背景的文档需要不同的正则模式
  • 标题编号不规范时（如手写编号、非标准格式）可能漏识别

3.2 HTML/Markdown 结构解析切分

原理

利用 HTML 的 DOM 树结构或 Markdown 的 AST（抽象语法树）进行切分。与正则匹配不同，这种方式能精确理解文档的层级关系，并且可以在 metadata 中保留完整的层级路径。

HTML 切分
from langchain_text_splitters import HTMLHeaderTextSplitter

html_content = """
<html>
  <body>
    <h1>产品介绍</h1>
    <p>本产品是一款智能办公助手，支持语音输入和文字输入两种交互方式。</p>
    <h2>核心功能</h2>
    <h3>语音识别</h3>
    <p>支持中英文混合识别，准确率高达98%。</p>
    <h3>智能问答</h3>
    <p>基于大语言模型，可回答各类办公相关问题。</p>
    <h2>使用方式</h2>
    <p>下载安装后，使用企业微信扫码登录即可使用。</p>
  </body>
</html>
"""

# 按 h1, h2, h3 标题层级切分
splitter = HTMLHeaderTextSplitter(
    headers_to_split_on=[
        ("h1", "一级标题"),
        ("h2", "二级标题"),
        ("h3", "三级标题"),
    ]
)

chunks = splitter.split_text(html_content)
for chunk in chunks:
    print(f"元数据: {chunk.metadata}")
    print(f"内容: {chunk.page_content[:80]}...")
    print()

每个 chunk 的 metadata 会记录其所属的标题层级，检索时可以精确定位到"产品介绍 > 核心功能 > 语音识别"这样的路径。

Markdown 切分
from langchain_text_splitters import MarkdownHeaderTextSplitter

markdown_content = """
# 员工手册

## 第一章 入职指南

### 1.1 入职材料
- 身份证原件及复印件
- 学历证书复印件
- 离职证明（如有）

### 1.2 入职流程
1. 提交材料至 HR 部门
2. 签订劳动合同
3. 领取办公设备

## 第二章 考勤制度

### 2.1 工作时间
标准工作时间为周一至周五 9:00-18:00。
"""

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ],
    strip_headers=False,  # 保留标题在 chunk 内容中
)

chunks = splitter.split_text(markdown_content)
for chunk in chunks:
    print(f"层级: {chunk.metadata}")
    print(f"内容: {chunk.page_content[:100]}")
    print("---")
综合评价
维度 评分 说明
实现难度 ★★☆☆☆ LangChain 内置，开箱即用
分块质量 ★★★★☆ 保留完整结构信息和层级元数据
处理速度 ★★★★☆ 解析器效率高
存储效率 ★★★★☆ 结构化的 chunk 几乎没有浪费
检索精度 ★★★★☆ 层级元数据可用于精确过滤
适用场景
  • 网页抓取内容的处理
  • Markdown 格式的技术文档
  • 需要保留文档层级关系的场景

3.3 文档格式特定切分

原理

不同文档格式有各自的原生结构信息，利用这些信息进行切分比通用方案更精准。每种格式的"自然边界"不同，需要针对性的解析策略。

各格式切分策略详解
文档格式 原生结构 切分策略 推荐工具 典型 chunk 大小
PDF 页面、书签、段落 按页切 + 段落合并 PyPDF2, pdfplumber, PyMuPDF 1-3 页/chunk
Word (.docx) 标题样式、段落、表格 按标题样式切 python-docx, docx2txt 按章节
Excel (.xlsx) 行、列、工作表 按行切（每行一个 chunk） openpyxl, xlrd, pandas 1 行/chunk
PPT (.pptx) 幻灯片、文本框 按幻灯片切 python-pptx 1 幻灯片/chunk
代码文件 函数、类、模块 AST 解析按函数/类切 tree-sitter, ast 1 函数/chunk
图片 无文本结构 OCR → 文本 → 段落切 PaddleOCR, Tesseract 按段落
数据库 行、表 按行切 sqlite3, SQLAlchemy 1 行/chunk
JSON/XML 键值对、嵌套结构 按顶层元素切 json, lxml 1 元素/chunk
邮件 (.eml) 发件人、主题、正文 按邮件切 email 标准库 1 邮件/chunk
PDF 切分详解

PDF 是最复杂的文档格式之一，因为它可能包含多种内容类型（文本、图片、表格），且内部结构不统一。

import fitz  # PyMuPDF

def load_pdf_as_chunks(
    file_path: str,
    pages_per_chunk: int = 2,
    overlap_pages: int = 0
) -> list:
    """
    将 PDF 按页切分为 chunk，支持页间重叠

    Args:
        file_path: PDF 文件路径
        pages_per_chunk: 每个 chunk 包含的页数
        overlap_pages: 相邻 chunk 重叠的页数
    """
    doc = fitz.open(file_path)
    documents = []
    step = pages_per_chunk - overlap_pages

    for start_page in range(0, len(doc), step):
        end_page = min(start_page + pages_per_chunk, len(doc))
        chunk_text = ""
        page_range = []

        for page_num in range(start_page, end_page):
            page = doc[page_num]
            chunk_text += page.get_text() + "\n"
            page_range.append(page_num + 1)

        if chunk_text.strip():
            documents.append(Document(
                page_content=chunk_text.strip(),
                metadata={
                    "source": file_path,
                    "pages": f"{page_range[0]}-{page_range[-1]}",
                    "total_pages": len(doc),
                }
            ))

    doc.close()
    return documents
代码切分详解

代码切分需要理解编程语言的语法结构，在函数、类、方法等逻辑边界处断开。

from langchain_text_splitters import (
    Language,
    RecursiveCharacterTextSplitter,
)

python_code = """
import os
from typing import List, Optional

class DocumentProcessor:
    \"\"\"文档处理器\"\"\"

    def __init__(self, chunk_size: int = 800):
        self.chunk_size = chunk_size

    def load_document(self, path: str) -> str:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()

    def split_text(self, text: str) -> List[str]:
        if len(text) <= self.chunk_size:
            return [text]
        return self._recursive_split(text)

def main():
    processor = DocumentProcessor()
    content = processor.load_document("example.txt")
    chunks = processor.split_text(content)
    print(f"共 {len(chunks)} 个 chunk")
"""

# Python 代码专用切分器
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=200,
    chunk_overlap=30,
)

chunks = python_splitter.split_text(python_code)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} ---")
    print(chunk)
    print()

LangChain 支持的语言包括：Python、JavaScript、TypeScript、Java、Go、Rust、C++、Ruby、PHP 等 20+ 种编程语言。每种语言有各自的分隔符优先级，例如 Python 的优先级为：

class → def → 空行 → 缩进块 → 行
Excel 按行切分详解
import openpyxl
from langchain_core.documents import Document

def load_excel_as_chunks(file_path: str) -> list:
    """将 Excel 文件的每一行作为一个独立的 chunk"""
    wb = openpyxl.load_workbook(file_path, read_only=True)
    documents = []

    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
        headers = None
        for row_idx, row in enumerate(ws.iter_rows(values_only=True), 1):
            if row_idx == 1:
                headers = [str(cell) if cell else f"列{i}"
                           for i, cell in enumerate(row, 1)]
                continue

            # 将每行转换为 "列名: 值" 格式
            row_text = " | ".join(
                f"{headers[i]}: {cell}"
                for i, cell in enumerate(row)
                if cell is not None
            )

            if row_text.strip():
                doc = Document(
                    page_content=row_text,
                    metadata={
                        "source": file_path,
                        "sheet": sheet_name,
                        "row": row_idx,
                    }
                )
                documents.append(doc)

    wb.close()
    return documents
综合评价
维度 评分 说明
实现难度 ★★★☆☆ 每种格式需要不同的解析器
分块质量 ★★★★★ 利用格式原生结构，质量最高
处理速度 ★★★☆☆ 取决于格式复杂度和解析器效率
存储效率 ★★★★☆ 结构化的 chunk 几乎没有浪费
检索精度 ★★★★★ 精确到行/页/幻灯片级别

4. 第三类：语义感知分块

语义感知分块不再依赖文档的表面结构，而是通过理解文本的语义内容来决定切分边界。这是目前最前沿的分块方向，也是学术界研究的热点。

4.1 Embedding 相似度切分（Semantic Chunking）

原理

Embedding 相似度切分的核心思想是：同一主题的句子在向量空间中距离近，不同主题的句子距离远。通过计算相邻句子的 Embedding 向量相似度，在相似度骤降处（语义转折点）进行切分。

工作流程
步骤1：将文本拆分为句子列表
  "年假申请需要提前3天提交。"           → 句子1
  "通过OA系统提交申请单。"               → 句子2
  "审批通过后方可休假。"                 → 句子3
  "办公用品每月1号开放申请。"           → 句子4
  "需填写物品申领表。"                   → 句子5

步骤2：对每个句子计算 Embedding 向量
  句子1 → [0.12, -0.34, 0.56, ...]  (1024维)
  句子2 → [0.15, -0.31, 0.52, ...]
  句子3 → [0.10, -0.36, 0.58, ...]
  句子4 → [-0.42, 0.28, -0.15, ...]  ← 向量方向突变！
  句子5 → [-0.45, 0.25, -0.18, ...]

步骤3：计算相邻句子余弦相似度
  句子1↔句子2: 0.91  ← 同一主题（年假）
  句子2↔句子3: 0.88  ← 同一主题（年假）
  句子3↔句子4: 0.28  ← 主题切换！断点！
  句子4↔句子5: 0.87  ← 同一主题（办公用品）

步骤4：在断点处切分
  Chunk1: 句子1 + 句子2 + 句子3  (年假相关)
  Chunk2: 句子4 + 句子5          (办公用品相关)
断点检测策略
策略 算法 特点 适用场景
百分位数 相似度低于 P10 的边界切断 自适应，不依赖绝对阈值 通用场景（推荐）
固定阈值 相似度 < 0.5 就切 简单直观 文档主题明确的场景
相对下降 相邻相似度下降 > 50% 就切 对突变敏感 主题切换频繁的文档
标准差 低于均值 - 1σ 就切 统计稳定 长文档
局部最小值 在相似度曲线的局部极小值处切 自然边界 主题渐变的文档
代码示例
import numpy as np
from typing import List
from langchain_huggingface import HuggingFaceEmbeddings

class SemanticChunker:
    """基于 Embedding 相似度的语义分块器"""

    def __init__(
        self,
        embedding_model: str = "BAAI/bge-m3",
        chunk_size: int = 800,
        chunk_overlap: int = 150,
        similarity_percentile: float = 90.0,
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.similarity_percentile = similarity_percentile
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={"device": "cuda"},
            encode_kwargs={"normalize_embeddings": True},
        )

    def split_text(self, text: str) -> List[str]:
        if len(text) <= self.chunk_size:
            return [text] if text else []

        # 1. 拆分为句子
        sentences = self._split_sentences(text)
        if len(sentences) <= 1:
            return [text]

        # 2. 计算句子 Embedding（批量编码）
        vectors = np.array(self.embeddings.embed_documents(sentences))

        # 3. 计算相邻句子余弦相似度（向量已归一化，点积即余弦相似度）
        similarities = [
            float(np.dot(vectors[i], vectors[i + 1]))
            for i in range(len(vectors) - 1)
        ]

        # 4. 找到断点
        breakpoints = self._find_breakpoints(similarities)

        # 5. 在断点处切分
        return self._group_sentences(sentences, breakpoints)

    def _split_sentences(self, text: str) -> List[str]:
        """按标点拆分句子"""
        import re
        paragraphs = text.split("\n")
        sentences = []
        for para in paragraphs:
            para = para.strip()
            if not para:
                continue
            # 在句末标点处切分
            parts = re.split(r"(?<=[。！？；.!?;])", para)
            for part in parts:
                part = part.strip()
                if part and len(part) >= 5:
                    sentences.append(part)
        return sentences

    def _find_breakpoints(self, similarities: List[float]) -> set:
        """找到语义断点"""
        if not similarities:
            return set()

        # 计算分位数阈值
        threshold = np.percentile(
            similarities,
            100 - self.similarity_percentile
        )
        # 不低于 0.3，避免切得太碎
        threshold = max(threshold, 0.3)

        return {i + 1 for i, sim in enumerate(similarities) if sim < threshold}

    def _group_sentences(
        self, sentences: List[str], breakpoints: set
    ) -> List[str]:
        """按断点分组句子，同时控制 chunk 大小"""
        chunks = []
        current = []
        current_len = 0

        for i, sent in enumerate(sentences):
            sent_len = len(sent)
            should_break = (
                i in breakpoints
                or (current_len + sent_len > self.chunk_size
                    and current_len > self.chunk_size * 0.3)
            )

            if should_break and current:
                chunks.append("\n".join(current))
                # overlap: 保留最后一句
                current = [current[-1]] if self.chunk_overlap > 0 else []
                current_len = len(current[0]) if current else 0

            current.append(sent)
            current_len += sent_len

        if current:
            chunks.append("\n".join(current))

        return chunks
综合评价
维度 评分 说明
实现难度 ★★★★☆ 需要 Embedding 模型和向量计算
分块质量 ★★★★★ 主题自然聚合，语义完整
处理速度 ★★☆☆☆ 每个句子都要 Embedding 编码
存储效率 ★★★★☆ 语义边界切分，几乎没有浪费
检索精度 ★★★★★ 每个 chunk 是完整的语义单元
适用场景
  • 无结构文档（网页、会议纪要、报告、新闻）
  • 主题混合的长文档
  • 无法用正则匹配标题的文档
性能优化建议
  1. 批量编码：一次性对所有句子编码，而非逐个编码
  2. 缓存 Embedding：相同句子不重复编码
  3. GPU 加速：使用 CUDA 设备进行向量计算
  4. 句子过滤：过滤过短的句子（< 5 字符），减少计算量
  5. 增量计算：对于长文档，可以分段计算相似度

4.2 LLM 驱动分块（Agentic Chunking）

原理

让 LLM 直接阅读全文并输出分块方案。这是最智能但也最昂贵的方式。LLM 能够理解文档的深层语义结构，识别主题切换、论证层次和逻辑关系。

与 Embedding 相似度切分不同，LLM 驱动分块不是基于"相邻句子的向量距离"这种局部信号，而是基于对全文的全局理解来做决策。

基础实现
from langchain_openai import ChatOpenAI

def llm_chunking(text: str, llm: ChatOpenAI) -> list:
    """使用 LLM 对文档进行分块"""

    prompt = f"""你是一个专业的文档结构化助手。请将以下文档按主题分为若干段落。

要求：
1. 每个段落保持语义完整，在主题切换处断开
2. 不要改变原文的任何文字
3. 用 "---CHUNK---" 作为段落分隔符
4. 输出纯文本，不要添加任何解释

文档内容：
{text}

分块结果："""

    response = llm.invoke(prompt)
    chunks = response.content.split("---CHUNK---")
    return [chunk.strip() for chunk in chunks if chunk.strip()]
进阶变体：带摘要的分块
def llm_chunking_with_summary(text: str, llm: ChatOpenAI) -> list:
    """LLM 分块并生成每个 chunk 的摘要"""

    prompt = f"""请将以下文档按主题分为若干段落，并为每个段落生成一句话摘要。

输出格式（JSON）：
[
  {{"summary": "段落摘要", "content": "段落原文"}},
  ...
]

文档内容：
{text}"""

    import json
    response = llm.invoke(prompt)
    chunks = json.loads(response.content)
    return chunks
综合评价
维度 评分 说明
实现难度 ★★★★★ 需要精心设计 Prompt
分块质量 ★★★★★ 理论最优，LLM 深度理解语义
处理速度 ★☆☆☆☆ 极慢，每篇文档需要一次 LLM 调用
成本 ★☆☆☆☆ 极贵，按文档长度计费
检索精度 ★★★★★ 分块质量最高
适用场景
  • 少量高价值文档（如核心制度、重要合同）
  • 对分块质量要求极高的场景
  • 预算充足的项目
成本估算

以 GPT-4o 为例，处理一篇 10000 字的文档：

输入 token：约 7000 tokens（文档 + Prompt）
输出 token：约 8000 tokens（分块结果）
成本：7000 × $2.5/1M + 8000 × $10/1M ≈ $0.10

100 篇文档 ≈ $10
1000 篇文档 ≈ $100

4.3 命题提取分块（Proposition-Based Chunking）

原理

命题提取是 LLM 驱动分块的一个特化变体。它不按段落切分，而是将文档拆解为"原子事实命题"——每个命题是一个独立的、可验证的陈述。这是目前学术界公认的检索精度最高的分块方式。

核心概念
传统分块：文档 → 段落 → chunk（可能包含多个事实）
命题分块：文档 → 原子命题列表（每个命题一个独立事实）
示例
原文：
"公司为入职满一年的员工提供每年5天年假，年假可分次使用，
但每次不少于半天。未休完的年假可延期至次年3月31日。"

命题提取结果：
  命题1: "入职满一年的员工享有年假"
  命题2: "年假天数为每年5天"
  命题3: "年假可分次使用"
  命题4: "年假每次使用不少于半天"
  命题5: "未休完的年假可延期至次年3月31日"
代码示例
def extract_propositions(text: str, llm: ChatOpenAI) -> list:
    """使用 LLM 从文档中提取原子命题"""

    prompt = f"""你是一个知识提取专家。请将以下文档拆解为原子事实命题。

规则：
1. 每个命题只包含一个独立的事实陈述
2. 命题必须完整、可独立理解（不依赖上下文）
3. 保留原文中的具体数字、日期、条件等关键信息
4. 每行输出一个命题，不要编号

文档内容：
{text}

原子命题列表："""

    response = llm.invoke(prompt)
    propositions = [
        line.strip()
        for line in response.content.strip().split("\n")
        if line.strip()
    ]
    return propositions
为什么命题分块检索精度最高

传统分块的问题：

用户问："年假每次最少请多久？"

传统 chunk（包含多个事实）：
"公司为入职满一年的员工提供每年5天年假，年假可分次使用，
但每次不少于半天。未休完的年假可延期至次年3月31日。"
→ Embedding 向量混合了 5 个事实的信息 → 检索信号被稀释

命题 chunk（单一事实）：
"年假每次使用不少于半天"
→ Embedding 向量精确编码这一个事实 → 检索信号集中
综合评价
维度 评分 说明
实现难度 ★★★★★ 需要精心设计提取 Prompt
分块质量 ★★★★★ 原子级精度，理论最优
处理速度 ★☆☆☆☆ 极慢
成本 ★☆☆☆☆ 极贵（输出 token 远多于输入）
检索精度 ★★★★★ 精确到原子事实

5. 第四类：层级多粒度分块

层级多粒度分块的核心思想是：不同阶段使用不同粒度的 chunk。检索阶段用小 chunk 提高精度，回答阶段用大 chunk 提供完整上下文。

5.1 父子文档分块（Parent-Child / Small-to-Big Retrieval）

原理

同时维护两种粒度的 chunk：

  • 子文档（小 chunk）：用于向量检索，粒度细、匹配精准
  • 父文档（大 chunk）：用于 LLM 回答，上下文完整

检索时用小 chunk 匹配，返回时用对应的大 chunk 作为上下文。

父文档（大 chunk，1000 字符）：
┌─────────────────────────────────────────────┐
│ 第三章 考勤管理                               │
│                                              │
│ 第三条 工作时间为上午9:00至下午6:00，        │
│ 午休12:00-13:00。                            │
│                                              │
│ 第四条 迟到30分钟以内扣款50元，              │
│ 超过30分钟按旷工半天处理。                    │
│                                              │
│ 第五条 加班需提前申请，经部门负责人审批。     │
└─────────────────────────────────────────────┘
        │ 切分为 3 个子文档
        ▼
┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ 子1: 第三条      │ │ 子2: 第四条      │ │ 子3: 第五条      │
│ 工作时间9:00-    │ │ 迟到30分钟内     │ │ 加班需提前申请   │
│ 18:00，午休      │ │ 扣款50元，超过    │ │ 经部门负责人审批 │
│ 12:00-13:00      │ │ 30分钟按旷工     │ │                  │
│                  │ │ 半天处理         │ │                  │
└──────────────────┘ └──────────────────┘ └──────────────────┘

检索流程：
  用户问："迟到怎么罚？"
  → 子文档检索：子2 匹配度最高
  → 返回父文档：完整的第三章内容（包含上下文）
LangChain 实现
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 父文档切分器（大 chunk）
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=100,
)

# 子文档切分器（小 chunk）
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=30,
)

# 向量存储（存子文档）
vectorstore = Chroma(
    collection_name="child_docs",
    embedding_function=embeddings,
)

# 文档存储（存父文档）
docstore = InMemoryStore()

# 创建父子检索器
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 添加文档
retriever.add_documents(documents)

# 检索：自动用小 chunk 匹配，返回大 chunk
results = retriever.invoke("迟到怎么罚款？")
综合评价
维度 评分 说明
实现难度 ★★★★☆ 需要维护双份索引
分块质量 ★★★★★ 兼顾检索精度和回答完整性
处理速度 ★★★☆☆ 检索小 chunk，返回大 chunk
存储效率 ★★☆☆☆ 双份存储（子 + 父）
检索精度 ★★★★★ 业界公认的最佳实践之一
适用场景
  • 对回答质量要求高的 RAG 系统
  • 文档篇幅较长、需要完整上下文的场景
  • 法律、医疗等对准确性要求极高的领域

5.2 层级分块（Hierarchical Chunking）

原理

按文档的层级结构建立多层索引，检索时从粗到细逐层定位。

文档: 员工手册.pdf
  │
  ├── 层级1（文档级）: "员工手册.pdf"
  │     summary: "包含入职、考勤、休假、薪酬等制度"
  │
  ├── 层级2（章节级）:
  │     ├── "第三章 考勤管理"
  │     │     summary: "工作时间、打卡、迟到处理等规定"
  │     └── "第四章 休假制度"
  │           summary: "年假、病假、事假等申请流程"
  │
  ├── 层级3（段落级）:
  │     ├── "第三条 工作时间"
  │     │     content: "工作时间为9:00-18:00，午休12:00-13:00"
  │     └── "第四条 迟到处理"
  │           content: "迟到30分钟内扣款50元，超过按旷工半天处理"
  │
  └── 层级4（句子级）:
        ├── "工作时间为9:00-18:00"
        └── "午休时间为12:00-13:00"

检索流程：
  用户问："迟到扣多少钱？"
  → 层级1 检索：定位到"员工手册.pdf"
  → 层级2 检索：定位到"第三章 考勤管理"
  → 层级3 检索：定位到"第四条 迟到处理"
  → 返回：层级3 的完整段落 + 层级2 的章节标题作为上下文
实现思路
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class HierarchicalChunk:
    """层级 chunk 数据结构"""
    content: str
    level: int              # 层级深度（1=文档, 2=章节, 3=段落, 4=句子）
    parent: Optional[str]   # 父 chunk ID
    children: List[str]     # 子 chunk ID 列表
    metadata: dict

class HierarchicalRetriever:
    """层级检索器"""

    def __init__(self, embeddings):
        self.embeddings = embeddings
        self.level_stores = {
            1: None,  # 文档级向量库
            2: None,  # 章节级向量库
            3: None,  # 段落级向量库
            4: None,  # 句子级向量库
        }

    def retrieve(self, query: str, target_level: int = 3):
        """
        层级检索：从粗到细逐层定位

        Args:
            query: 用户查询
            target_level: 目标层级（默认段落级）
        """
        current_level = 1
        current_chunks = self._search_level(query, current_level, top_k=3)

        while current_level < target_level:
            # 获取下一层的候选 chunk
            next_chunks = []
            for chunk in current_chunks:
                next_chunks.extend(self._get_children(chunk))
            # 在下一层中检索
            current_chunks = self._rerank(query, next_chunks)
            current_level += 1

        return current_chunks
综合评价
维度 评分 说明
实现难度 ★★★★★ 需要维护多层索引和层级关系
分块质量 ★★★★★ 多粒度覆盖，信息不丢失
处理速度 ★★★☆☆ 多层检索增加延迟
存储效率 ★★☆☆☆ 多层存储，空间开销大
检索精度 ★★★★★ 精确到任意层级

6. 十二种方案综合对比

6.1 核心指标对比

# 方案 分块质量 处理速度 实现难度 存储效率 检索精度 成本
1 固定字符切 ★☆☆☆☆ ★★★★★ ★☆☆☆☆ ★★☆☆☆ ★★☆☆☆ 免费
2 递归字符切 ★★★☆☆ ★★★★☆ ★★☆☆☆ ★★★☆☆ ★★★☆☆ 免费
3 Token 切 ★★★☆☆ ★★★★☆ ★★☆☆☆ ★★★☆☆ ★★★☆☆ 免费
4 滑动窗口 ★★★☆☆ ★★★☆☆ ★★☆☆☆ ★☆☆☆☆ ★★★★☆ 免费
5 章节标题切 ★★★★★ ★★★★☆ ★★★☆☆ ★★★★☆ ★★★★★ 免费
6 HTML/MD 解析 ★★★★☆ ★★★★☆ ★★☆☆☆ ★★★★☆ ★★★★☆ 免费
7 格式特定切 ★★★★★ ★★★☆☆ ★★★☆☆ ★★★★☆ ★★★★★ 免费
8 Embedding 语义 ★★★★★ ★★☆☆☆ ★★★★☆ ★★★★☆ ★★★★★ Embedding
9 LLM 驱动 ★★★★★ ★☆☆☆☆ ★★★★★ ★★★★☆ ★★★★★ LLM 贵
10 命题提取 ★★★★★ ★☆☆☆☆ ★★★★★ ★★★★☆ ★★★★★ LLM 贵
11 父子文档 ★★★★★ ★★★☆☆ ★★★★☆ ★★☆☆☆ ★★★★★ 存储高
12 层级分块 ★★★★★ ★★★☆☆ ★★★★★ ★★☆☆☆ ★★★★★ 存储高

6.2 中文适配度对比

方案 中文适配 说明
固定字符切 差 中文词之间无空格，硬切破坏更严重
递归字符切 中 支持中文标点分隔符
Token 切 中 中文 token 密度高，收益不如英文
滑动窗口 中 与语言无关
章节标题切 好 中文标题模式（第X章、第X条）识别精准
HTML/MD 解析 好 与语言无关
格式特定切 好 与语言无关
Embedding 语义 好 BGE-M3 等中文 Embedding 模型成熟
LLM 驱动 好 取决于 LLM 的中文能力
命题提取 好 取决于 LLM 的中文能力
父子文档 好 与语言无关
层级分块 好 与语言无关

6.3 不同 Embedding 模型对分块策略的影响

不同的 Embedding 模型有不同的最佳输入长度和语义理解能力，这直接影响分块策略的选择：

Embedding 模型 最大输入长度 最佳 chunk 大小 推荐分块策略
BGE-M3 8192 tokens 512-1024 tokens 语义分块 / 章节切分
text-embedding-3-small 8191 tokens 256-512 tokens 递归字符切
text-embedding-3-large 8191 tokens 512-1024 tokens 语义分块
BGE-large-zh 512 tokens 256-384 tokens 章节切分（中文）
Jina-embeddings-v3 8192 tokens 512-1024 tokens 语义分块
Cohere Embed v3 512 tokens 256-384 tokens 递归字符切

7. Chunk 参数选择深度指南

7.1 chunk_size 的选择方法论

chunk_size 是最关键的分块参数。选择时需要综合考虑以下因素：

因素一：Embedding 模型的最佳输入长度

每个 Embedding 模型都有一个"甜蜜点"（sweet spot），在这个长度范围内向量表示质量最高。

BGE-M3 性能曲线（示意）：
  向量质量
    ↑
    │     ┌──────────────┐
    │    ╱                ╲
    │   ╱                  ╲
    │  ╱                    ╲_____
    │ ╱                            ╲___
    └──────────────────────────────────→ 文本长度 (tokens)
          ↑                    ↑
      128 tokens          2048 tokens
      (太短，信息不足)     (太长，表示稀释)

  甜蜜点：256 - 1024 tokens
因素二：文档类型
文档类型 推荐 chunk_size 原因
制度/法律条文 300-500 字符 单条条款通常在这个范围
技术文档 500-800 字符 一个完整的功能说明
网页文章 800-1200 字符 一个完整的段落组
学术论文 1000-1500 字符 一个完整的论证段落
对话/聊天记录 200-400 字符 单轮对话的长度
代码 500-1000 字符 一个完整的函数/类
因素三：LLM 上下文窗口

如果使用父子文档策略，父文档的 chunk_size 需要考虑 LLM 的上下文窗口：

LLM 上下文窗口预算分配（以 128K 窗口为例）：
  - System Prompt: 1K tokens
  - 用户问题: 0.5K tokens
  - 检索到的 chunks: 100K tokens（主要部分）
  - 回答预留: 26.5K tokens

如果检索 top_k=5，则每个 chunk 约 20K tokens
如果检索 top_k=10，则每个 chunk 约 10K tokens
因素四：检索精度 vs 回答完整性
chunk_size 小（100-300 tokens）：
  ✓ 检索精度高，向量表示精确
  ✓ 检索速度快
  ✗ 上下文不完整，LLM 可能无法理解
  ✗ 需要检索更多 chunk 才能覆盖完整信息

chunk_size 大（1000-2000 tokens）：
  ✓ 上下文完整，LLM 回答质量高
  ✓ 需要检索的 chunk 数量少
  ✗ 检索精度低，向量表示模糊
  ✗ 包含更多无关信息

7.2 chunk_overlap 的选择方法论

chunk_overlap 用于在相邻 chunk 之间保留重叠内容，缓解边界处的信息断裂。

推荐值
场景 推荐 overlap 说明
通用场景 chunk_size 的 10-20% 平衡存储和召回
高精度场景 chunk_size 的 20-30% 确保不丢失边界信息
存储敏感场景 chunk_size 的 5-10% 最小化存储开销
滑动窗口模式 chunk_size 的 50-70% 最大化召回率
计算公式
def calculate_overlap(chunk_size: int, strategy: str = "balanced") -> int:
    """根据策略计算推荐的 overlap 值"""
    strategies = {
        "minimal": int(chunk_size * 0.05),    # 5%
        "balanced": int(chunk_size * 0.15),   # 15%
        "high": int(chunk_size * 0.25),       # 25%
        "sliding": int(chunk_size * 0.6),     # 60%
    }
    return strategies.get(strategy, int(chunk_size * 0.15))

7.3 参数组合推荐

场景 chunk_size chunk_overlap 分块策略 说明
快速原型 1000 字符 200 字符 递归字符切 最简单，快速验证
企业知识库 800 字符 150 字符 组合拳 覆盖多种文档类型
法律文档 500 字符 100 字符 章节标题切 条款级精度
客服问答 300 字符 50 字符 语义分块 短问答匹配
学术研究 1500 字符 200 字符 父子文档 完整论证上下文
代码搜索 800 字符 100 字符 代码 AST 切 函数级精度

8. Lost in the Middle 问题与 Chunk 排序策略

8.1 什么是 “Lost in the Middle”

“Lost in the Middle” 是 2023 年由 Stanford、UC Berkeley 等机构在论文中提出的现象：LLM 对输入上下文中间位置的信息关注度最低，对开头和结尾的信息关注度最高。

LLM 注意力分布（U 型曲线）：
  注意力强度
    ↑
    │ ╲
    │  ╲
    │   ╲    ___________
    │    ╲__╱           ╲
    │                    ╲
    │                     ╲___
    └──────────────────────────→ 上下文位置
    开头                    结尾
    (高注意力)   中间区域    (高注意力)
                (低注意力)

这意味着：即使检索到了正确的 chunk，如果它被放在 LLM 输入上下文的中间位置，LLM 也可能"忽略"它。

8.2 Chunk 排序策略

为了对抗 “Lost in the Middle” 问题，需要对检索到的 chunk 进行智能排序：

策略一：相关性排序（Relevance Ranking）

将最相关的 chunk 放在开头和结尾：

def relevance_aware_ordering(chunks: list, scores: list) -> list:
    """
    相关性感知排序：
    最相关的放开头，次相关的放结尾，其余按相关性降序放中间
    """
    if len(chunks) <= 2:
        return chunks

    # 按相关性排序
    sorted_pairs = sorted(
        zip(chunks, scores), key=lambda x: x[1], reverse=True
    )

    result = []
    # 最相关的放开头
    result.append(sorted_pairs[0][0])
    # 中间部分：从第三相关开始到倒数第二相关
    middle = [chunk for chunk, _ in sorted_pairs[2:-1]]
    result.extend(middle)
    # 次相关的放结尾
    if len(sorted_pairs) > 1:
        result.append(sorted_pairs[1][0])

    return result
策略二：多样性排序（Diversity Ranking）

避免连续放置内容相似的 chunk，增加信息多样性：

def diversity_aware_ordering(chunks: list, embeddings: list) -> list:
    """
    多样性感知排序：
    相邻 chunk 的内容尽量不重复
    """
    if len(chunks) <= 2:
        return chunks

    result = [chunks[0]]
    remaining = list(range(1, len(chunks)))

    while remaining:
        last_emb = embeddings[chunks.index(result[-1])]
        # 选择与上一个 chunk 最不相似的
        max_dist = -1
        best_idx = remaining[0]
        for idx in remaining:
            dist = 1 - cosine_similarity(last_emb, embeddings[idx])
            if dist > max_dist:
                max_dist = dist
                best_idx = idx
        result.append(chunks[best_idx])
        remaining.remove(best_idx)

    return result
策略三：层级排序（Hierarchical Ordering）

按文档的层级结构排序，保持逻辑连贯性：

def hierarchical_ordering(chunks: list) -> list:
    """
    层级排序：
    按文档的原始层级结构排序，保持逻辑连贯性
    同一章节的 chunk 放在一起，章节之间按原始顺序排列
    """
    # 按 (source, section, chunk_index) 排序
    return sorted(
        chunks,
        key=lambda c: (
            c.metadata.get("source", ""),
            c.metadata.get("section_order", 0),
            c.metadata.get("chunk_index", 0),
        )
    )
策略四：混合排序（Hybrid Ordering）

结合相关性和层级结构：

def hybrid_ordering(
    chunks: list,
    scores: list,
    diversity_weight: float = 0.3
) -> list:
    """
    混合排序策略：
    综合考虑相关性、多样性和层级结构

    Args:
        chunks: 检索到的 chunk 列表
        scores: 相关性分数列表
        diversity_weight: 多样性权重（0=纯相关性，1=纯多样性）
    """
    if len(chunks) <= 2:
        return chunks

    result = []
    remaining = list(range(len(chunks)))

    # 第一个：最相关的
    best_idx = remaining.pop(scores.index(max(scores)))
    result.append(chunks[best_idx])

    while remaining:
        best_score = -float("inf")
        best_idx = remaining[0]

        for idx in remaining:
            # 相关性分数（归一化）
            rel_score = scores[idx] / max(scores) if max(scores) > 0 else 0
            # 多样性分数（与已选 chunk 的最小相似度）
            div_score = min(
                1 - cosine_similarity(embeddings[idx], embeddings[sel])
                for sel in [chunks.index(c) for c in result]
            ) if result else 1.0
            # 综合分数
            combined = (1 - diversity_weight) * rel_score + diversity_weight * div_score
            if combined > best_score:
                best_score = combined
                best_idx = idx

        result.append(chunks[best_idx])
        remaining.remove(best_idx)

    return result

8.3 排序策略选择指南

场景 推荐排序策略 原因
通用问答 相关性排序 确保最相关信息在注意力高峰位置
多文档检索 多样性排序 避免重复信息浪费上下文窗口
结构化文档 层级排序 保持逻辑连贯性
高精度场景 混合排序 综合多种因素

9. Chunk 质量评估体系

9.1 为什么需要评估分块质量

分块策略的选择直接影响 RAG 系统的最终效果。没有评估，就无法知道：

  • 当前分块策略是否适合你的文档
  • 参数调整是否带来了实际改进
  • 不同策略之间的真实差距

9.2 评估指标体系

指标一：语义完整性（Semantic Completeness）

衡量每个 chunk 是否是一个自包含的语义单元。

def evaluate_semantic_completeness(chunks: list, llm) -> dict:
    """
    评估 chunk 的语义完整性

    方法：让 LLM 判断每个 chunk 是否可以独立理解
    """
    results = {"complete": 0, "partial": 0, "incomplete": 0}

    for chunk in chunks:
        prompt = f"""请判断以下文本片段是否可以独立理解（不需要额外上下文）：

文本：
{chunk.page_content[:500]}

回答格式：
- "完整"：可以完全独立理解
- "部分"：可以部分理解，但缺少一些上下文
- "不完整"：无法独立理解，需要额外上下文

判断结果："""

        response = llm.invoke(prompt).content.strip()
        if "完整" in response:
            results["complete"] += 1
        elif "部分" in response:
            results["partial"] += 1
        else:
            results["incomplete"] += 1

    total = len(chunks)
    return {
        "complete_rate": results["complete"] / total,
        "partial_rate": results["partial"] / total,
        "incomplete_rate": results["incomplete"] / total,
    }
指标二：信息密度（Information Density）

衡量每个 chunk 包含的有效信息量。

def evaluate_information_density(chunks: list) -> dict:
    """
    评估 chunk 的信息密度

    方法：计算有效内容占比（排除空白、标点等）
    """
    import re

    densities = []
    for chunk in chunks:
        text = chunk.page_content
        # 有效字符（中文字符 + 英文单词 + 数字）
        chinese_chars = len(re.findall(r"[\u4e00-\u9fff]", text))
        english_words = len(re.findall(r"[a-zA-Z]+", text))
        digits = len(re.findall(r"\d+", text))
        effective = chinese_chars + english_words + digits
        total = len(text.strip())
        density = effective / max(total, 1)
        densities.append(density)

    return {
        "mean_density": sum(densities) / len(densities),
        "min_density": min(densities),
        "max_density": max(densities),
        "std_density": (
            sum((d - sum(densities) / len(densities)) ** 2
                for d in densities) / len(densities)
        ) ** 0.5,
    }
指标三：大小分布（Size Distribution）

衡量 chunk 大小的均匀程度。

def evaluate_size_distribution(chunks: list) -> dict:
    """评估 chunk 大小分布"""
    sizes = [len(c.page_content) for c in chunks]

    return {
        "count": len(sizes),
        "mean": sum(sizes) / len(sizes),
        "min": min(sizes),
        "max": max(sizes),
        "median": sorted(sizes)[len(sizes) // 2],
        "std": (
            sum((s - sum(sizes) / len(sizes)) ** 2
                for s in sizes) / len(sizes)
        ) ** 0.5,
        "too_small_rate": sum(1 for s in sizes if s < 50) / len(sizes),
        "too_large_rate": sum(1 for s in sizes if s > 2000) / len(sizes),
    }
指标四：检索命中率（Retrieval Hit Rate）

衡量分块策略对检索效果的实际影响。

def evaluate_retrieval_hit_rate(
    chunks: list,
    test_queries: list,
    ground_truth: dict,  # {query: [expected_chunk_indices]}
    retriever,
) -> dict:
    """
    评估检索命中率

    Args:
        chunks: 所有 chunk
        test_queries: 测试查询列表
        ground_truth: 每个查询对应的正确答案（chunk 索引列表）
        retriever: 检索器
    """
    hits = []
    for query in test_queries:
        results = retriever.invoke(query)
        retrieved_indices = [
            chunks.index(r) for r in results
            if r in chunks
        ]
        expected = ground_truth.get(query, [])
        hit_count = len(set(retrieved_indices) & set(expected))
        hits.append(hit_count / max(len(expected), 1))

    return {
        "mean_hit_rate": sum(hits) / len(hits),
        "min_hit_rate": min(hits),
        "max_hit_rate": max(hits),
        "perfect_rate": sum(1 for h in hits if h == 1.0) / len(hits),
    }

9.3 评估流程建议

1. 准备测试集
   - 选择 20-50 个代表性查询
   - 人工标注每个查询的正确答案（golden chunks）

2. 基线测试
   - 使用默认参数（chunk_size=800, overlap=150）建立基线

3. 参数扫描
   - chunk_size: [200, 400, 600, 800, 1000, 1500]
   - chunk_overlap: [5%, 10%, 15%, 20%, 25%]
   - 记录每个参数组合的评估指标

4. 策略对比
   - 对比不同分块策略在相同参数下的表现
   - 重点关注检索命中率和答案质量

5. 选择最优方案
   - 综合考虑检索精度、处理速度和存储成本

10. 多模态文档的分块考量

10.1 多模态文档的挑战

现实世界中的文档往往不是纯文本，而是包含多种内容类型：

  • 图文混排：PDF 中的文字 + 图片 + 图表
  • 表格嵌入：Word 文档中的表格
  • 公式：学术论文中的数学公式
  • 代码块：技术文档中的代码示例

这些非文本内容在传统分块中往往被忽略或错误处理。

10.2 图片处理策略

策略一：图片描述生成

使用多模态 LLM 为图片生成文字描述，将描述嵌入到 chunk 中：

def process_image_with_caption(image_path: str, llm) -> str:
    """使用多模态 LLM 为图片生成描述"""
    import base64

    with open(image_path, "rb") as f:
        image_base64 = base64.b64encode(f.read()).decode()

    prompt = """请详细描述这张图片的内容，包括：
1. 图片类型（图表/照片/截图/示意图）
2. 主要内容
3. 关键数据和文字
4. 与文档主题的关联"""

    response = llm.invoke([
        {"type": "text", "text": prompt},
        {"type": "image_url", "image_url": {
            "url": f"data:image/png;base64,{image_base64}"
        }},
    ])

    return response.content
策略二：图片位置保留

在 chunk 中保留图片的位置标记，检索时可以定位到原始图片：

def chunk_with_image_markers(text: str, images: list) -> list:
    """
    在 chunk 中保留图片位置标记

    将 "[IMAGE:图片描述]" 插入到原文中图片的原始位置
    """
    for img in images:
        placeholder = f"[IMAGE:{img['description']}]"
        text = text.replace(img["original_text"], placeholder, 1)

    # 然后对包含图片标记的文本进行正常分块
    return splitter.split_text(text)

10.3 表格处理策略

表格在分块时需要特殊处理，因为表格的语义依赖于行列结构：

def process_table_for_chunking(table_html: str) -> str:
    """
    将表格转换为适合分块的文本格式

    策略：
    1. 保留表头
    2. 每行数据独立成句
    3. 保留行列关系
    """
    from bs4 import BeautifulSoup

    soup = BeautifulSoup(table_html, "html.parser")
    rows = soup.find_all("tr")

    if not rows:
        return ""

    # 提取表头
    headers = [th.get_text(strip=True) for th in rows[0].find_all(["th", "td"])]

    # 转换每行数据
    result = []
    for row in rows[1:]:
        cells = [td.get_text(strip=True) for td in row.find_all(["td", "th"])]
        row_text = " | ".join(
            f"{headers[i]}: {cells[i]}"
            for i in range(min(len(headers), len(cells)))
        )
        result.append(row_text)

    return "\n".join(result)

10.4 多模态分块最佳实践

内容类型 处理策略 工具
图片 生成文字描述 + 保留位置标记 GPT-4V / Claude 3
表格 转换为 “列名: 值” 格式 BeautifulSoup / pandas
公式 LaTeX 转 Unicode / 文字描述 latex2unicode
代码块 保留原始格式 + 语言标记 代码 AST 切分器
图表 提取数据 + 生成文字摘要 OCR + LLM

11. Chunk 元数据增强策略

11.1 为什么需要元数据增强

原始的 chunk 只包含文本内容，缺少上下文信息。元数据增强可以为每个 chunk 添加丰富的上下文标签，显著提升检索精度。

11.2 元数据增强维度

维度一：结构元数据
def enrich_structure_metadata(chunk: Document, original_doc: Document) -> Document:
    """添加结构相关的元数据"""
    chunk.metadata.update({
        "doc_title": original_doc.metadata.get("title", ""),
        "section_path": original_doc.metadata.get("section_path", ""),
        "heading_level": original_doc.metadata.get("heading_level", 0),
        "position_in_doc": f"{chunk.metadata.get('chunk_index', 0) + 1}"
                           f"/{chunk.metadata.get('total_chunks', 1)}",
    })
    return chunk
维度二：内容元数据
def enrich_content_metadata(chunk: Document, llm) -> Document:
    """添加内容相关的元数据"""
    prompt = f"""请为以下文本片段生成元数据标签：

文本：
{chunk.page_content[:500]}

请输出 JSON 格式：
{{
    "summary": "一句话摘要",
    "keywords": ["关键词1", "关键词2", ...],
    "category": "分类（如：制度/技术/财务/人事）",
    "entities": ["实体1", "实体2", ...]
}}"""

    import json
    response = llm.invoke(prompt)
    try:
        meta = json.loads(response.content)
        chunk.metadata.update(meta)
    except json.JSONDecodeError:
        pass

    return chunk
维度三：时间元数据
def enrich_temporal_metadata(chunk: Document) -> Document:
    """添加时间相关的元数据"""
    import re
    from datetime import datetime

    text = chunk.page_content

    # 提取日期
    date_patterns = [
        r"(\d{4}[-/]\d{1,2}[-/]\d{1,2})",
        r"(\d{4}年\d{1,2}月\d{1,2}日)",
    ]

    dates = []
    for pattern in date_patterns:
        dates.extend(re.findall(pattern, text))

    chunk.metadata["mentioned_dates"] = dates
    chunk.metadata["has_temporal_info"] = len(dates) > 0

    return chunk

11.3 元数据在检索中的应用

def metadata_aware_retrieval(
    query: str,
    vectorstore,
    metadata_filter: dict = None,
    top_k: int = 5
) -> list:
    """
    利用元数据进行精确检索

    示例：
    - 只检索"制度"类文档：{"category": "制度"}
    - 只检索特定章节：{"section_path": "第三章 考勤管理"}
    - 只检索包含日期的 chunk：{"has_temporal_info": True}
    """
    if metadata_filter:
        return vectorstore.similarity_search(
            query,
            k=top_k,
            filter=metadata_filter,
        )
    return vectorstore.similarity_search(query, k=top_k)

12. 业界最佳实践：分层组合策略

12.1 三层组合架构

在实际生产环境中，单一分块策略往往无法应对多样化的文档类型。业界推荐采用分层组合策略：

第一层：格式路由（自动检测 + 智能分发）
  ├── 表格类（xls/xlsx/csv/sqlite）→ 按行切分
  ├── 制度/合同类（有章节结构）  → 章节标题切分
  ├── 网页/报告类（无结构长文本）→ Embedding 语义切分
  └── 其他/短文本                → 递归字符切分（兜底）

第二层：父子文档（可选，进阶）
  小 chunk 检索 + 大 chunk 回答

第三层：命题提取（可选，高成本）
  关键文档用 LLM 提取原子命题

12.2 第一层实现：自动文档分类 + 路由

import logging
from pathlib import Path
from typing import List
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

logger = logging.getLogger(__name__)


class HybridDocumentProcessor:
    """组合拳文档处理器：自动检测文档类型并选择最优分块器"""

    def __init__(self, chunk_size=800, chunk_overlap=150):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        # 初始化所有分块器
        self.section_splitter = SectionAwareSplitter(chunk_size, chunk_overlap)
        self.semantic_splitter = SemanticChunker(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
        )
        self.fallback_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
        )

    def classify(self, doc: Document) -> str:
        """自动分类文档类型"""
        source = doc.metadata.get("source", "")
        text = doc.page_content

        # 1. 表格类
        if Path(source).suffix.lower() in {
            ".xlsx", ".xls", ".csv", ".db", ".sqlite"
        }:
            return "tabular"

        # 2. 有章节结构
        if self._has_section_structure(text):
            return "section"

        # 3. 长文本用语义分块
        if len(text) > self.chunk_size:
            return "semantic"

        # 4. 兜底
        return "fallback"

    def _has_section_structure(self, text: str) -> bool:
        """检测是否有章节结构"""
        import re
        patterns = [
            r"^(第?[一二三四五六七八九十百零\d]+[章节条款])",
            r"^\d+\.\s+.+",
            r"^\d+\.\d+\.?\s+.+",
            r"^#{1,3}\s+.+",
        ]
        total = sum(
            len(re.findall(p, text, re.MULTILINE))
            for p in patterns
        )
        return total >= 2 and len(text) / max(total, 1) <= 2000

    def process(self, documents: List[Document]) -> List[Document]:
        """处理文档：自动分类 + 路由分块"""
        stats = {"section": 0, "semantic": 0, "tabular": 0, "fallback": 0}
        all_chunks = []

        for doc in documents:
            strategy = self.classify(doc)
            stats[strategy] += 1

            if strategy == "tabular":
                all_chunks.append(self._wrap_chunk(doc, "tabular（表格按行）"))
            elif strategy == "section":
                chunks = self._split_with(
                    doc, self.section_splitter, "section（章节感知）"
                )
                all_chunks.extend(chunks)
            elif strategy == "semantic":
                chunks = self._split_with(
                    doc, self.semantic_splitter, "semantic（语义边界）"
                )
                all_chunks.extend(chunks)
            else:
                chunks = self._split_with(
                    doc, self.fallback_splitter, "fallback（递归字符）"
                )
                all_chunks.extend(chunks)

        logger.info(
            f"组合拳分块完成：{len(documents)} 篇文档 -> {len(all_chunks)} 个 Chunks"
            f"（章节感知={stats['section']}篇, 语义边界={stats['semantic']}篇, "
            f"表格按行={stats['tabular']}篇, 递归兜底={stats['fallback']}篇）"
        )
        return all_chunks

    def _wrap_chunk(self, doc: Document, method: str) -> Document:
        """将单个文档包装为 chunk（用于表格等不需要再切的文档）"""
        return Document(
            page_content=doc.page_content,
            metadata={
                **doc.metadata,
                "chunk_index": 0,
                "total_chunks": 1,
                "chunk_method": method,
            },
        )

    def _split_with(self, doc: Document, splitter, method: str) -> List[Document]:
        """使用指定分块器切分文档，并添加元数据"""
        raw_text = doc.page_content

        try:
            if hasattr(splitter, "split_text"):
                chunk_texts = splitter.split_text(raw_text)
            else:
                chunk_texts = splitter.split_text(raw_text)
        except Exception as e:
            logger.warning(f"{method} 分块失败，回退到递归字符切：{e}")
            chunk_texts = self.fallback_splitter.split_text(raw_text)
            method = "fallback（异常回退）"

        chunks = []
        for i, text in enumerate(chunk_texts):
            chunk_doc = Document(
                page_content=text,
                metadata={
                    **doc.metadata,
                    "chunk_index": i,
                    "total_chunks": len(chunk_texts),
                    "chunk_method": method,
                },
            )
            chunks.append(chunk_doc)

        return chunks

12.3 不同阶段的策略选择

项目阶段 推荐策略 原因
原型验证 递归字符切 快速搭建，验证流程
MVP 上线 组合拳（格式路由） 覆盖主流文档类型
质量优化 组合拳 + 父子文档 提升回答完整性和准确性
极致精度 组合拳 + 父子文档 + 命题提取 关键文档原子级精度

13. 真实场景案例研究

13.1 案例一：企业制度知识库

场景描述：
某 500 人企业需要构建内部制度知识库，包含 200+ 份制度文档（员工手册、考勤制度、报销制度等），格式为 Word 和 PDF。

文档特征：

  • 有明确的章节结构（第X章、第X条）
  • 包含表格（考勤时间表、报销标准表）
  • 部分文档包含流程图

分块方案：

第一层：格式路由
  - Word 文档 → 章节标题切分（chunk_size=500, overlap=100）
  - PDF 文档 → 按页切 + 章节标题切
  - 表格 → 按行切分

第二层：父子文档
  - 子 chunk: 单条条款（200-300 字符）
  - 父 chunk: 完整章节（800-1000 字符）

第三层：元数据增强
  - 制度分类标签（人事/财务/行政）
  - 生效日期
  - 版本号

效果对比：

指标 递归字符切 组合拳方案 提升
检索命中率 72% 94% +22%
答案准确率 68% 91% +23%
平均响应时间 2.1s 2.4s -0.3s

13.2 案例二：技术文档问答

场景描述：
某开源项目需要构建技术文档问答系统，文档为 Markdown 格式，包含代码示例、API 参考和架构说明。

文档特征：

  • Markdown 格式，有标题层级
  • 大量代码块
  • API 文档有固定格式

分块方案：

第一层：Markdown 结构解析
  - 按 ## 和 ### 标题切分
  - 代码块保持完整（不切割代码块内部）

第二层：代码特殊处理
  - 代码块使用 AST 切分器
  - 保留代码语言标记

第三层：语义增强
  - 为每个 API 文档 chunk 生成功能摘要
  - 提取函数签名作为关键词

效果对比：

指标 通用递归切 Markdown 结构切 提升
代码检索准确率 45% 89% +44%
API 检索命中率 52% 93% +41%
答案代码正确率 38% 85% +47%

13.3 案例三：客服知识库

场景描述：
某电商平台需要构建客服知识库，包含 FAQ、产品说明、退换货政策等，格式多样。

文档特征：

  • 短文本为主（FAQ 通常 50-200 字）
  • 格式多样（Excel FAQ 表、Word 政策文档、网页抓取）
  • 更新频繁（促销政策经常变化）

分块方案：

第一层：格式路由
  - Excel FAQ → 按行切分（每行一个 QA 对）
  - Word 政策 → 章节标题切分
  - 网页抓取 → Embedding 语义切分

第二层：短文本优化
  - FAQ 类短文本不切分（直接作为一个 chunk）
  - 设置最小 chunk 阈值（< 100 字符不切）

第三层：增量更新
  - 按文档来源追踪变更
  - 只重新处理变更的文档

效果对比：

指标 统一递归切 组合拳方案 提升
FAQ 匹配率 61% 96% +35%
首次响应准确率 55% 88% +33%
更新延迟 全量 30min 增量 2min -93%

14. 生产环境工程实践

14.1 分块缓存策略

分块是一个计算密集型操作（尤其是语义分块），在生产环境中需要合理的缓存策略：

import hashlib
import json
from functools import lru_cache

class ChunkCache:
    """分块结果缓存"""

    def __init__(self, cache_dir: str = "./chunk_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def _get_cache_key(self, content: str, params: dict) -> str:
        """生成缓存键"""
        content_hash = hashlib.md5(content.encode()).hexdigest()
        params_hash = hashlib.md5(
            json.dumps(params, sort_keys=True).encode()
        ).hexdigest()
        return f"{content_hash}_{params_hash}"

    def get(self, content: str, params: dict) -> list:
        """从缓存获取分块结果"""
        key = self._get_cache_key(content, params)
        cache_file = self.cache_dir / f"{key}.json"

        if cache_file.exists():
            with open(cache_file, "r", encoding="utf-8") as f:
                return json.load(f)
        return None

    def set(self, content: str, params: dict, chunks: list):
        """缓存分块结果"""
        key = self._get_cache_key(content, params)
        cache_file = self.cache_dir / f"{key}.json"

        with open(cache_file, "w", encoding="utf-8") as f:
            json.dump(chunks, f, ensure_ascii=False)

14.2 并发处理

对于大量文档，需要并发处理以提高吞吐量：

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_chunking(
    documents: list,
    processor: HybridDocumentProcessor,
    max_workers: int = 4,
    batch_size: int = 10,
) -> list:
    """
    并发分块处理

    Args:
        documents: 文档列表
        processor: 分块处理器
        max_workers: 最大并发数
        batch_size: 批处理大小
    """
    all_chunks = []

    # 分批处理
    batches = [
        documents[i:i + batch_size]
        for i in range(0, len(documents), batch_size)
    ]

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(processor.process, batch): batch_idx
            for batch_idx, batch in enumerate(batches)
        }

        for future in as_completed(futures):
            try:
                chunks = future.result()
                all_chunks.extend(chunks)
            except Exception as e:
                logger.error(f"批次 {futures[future]} 处理失败: {e}")

    return all_chunks

14.3 增量更新

当文档发生变化时，不需要重新处理所有文档：

class IncrementalChunkManager:
    """增量分块管理器"""

    def __init__(self, vectorstore, processor):
        self.vectorstore = vectorstore
        self.processor = processor
        self.doc_registry = {}  # {doc_id: {"hash": "...", "chunk_ids": [...]}}

    def update_document(self, doc_id: str, new_content: str):
        """更新单个文档的分块"""
        new_hash = hashlib.md5(new_content.encode()).hexdigest()

        # 检查是否真的发生了变化
        if doc_id in self.doc_registry:
            old_info = self.doc_registry[doc_id]
            if old_info["hash"] == new_hash:
                logger.info(f"文档 {doc_id} 未变化，跳过")
                return

            # 删除旧 chunk
            self.vectorstore.delete(ids=old_info["chunk_ids"])

        # 重新分块
        new_doc = Document(page_content=new_content, metadata={"doc_id": doc_id})
        new_chunks = self.processor.process([new_doc])

        # 添加到向量库
        chunk_ids = self.vectorstore.add_documents(new_chunks)

        # 更新注册表
        self.doc_registry[doc_id] = {
            "hash": new_hash,
            "chunk_ids": chunk_ids,
        }

14.4 监控与告警

class ChunkingMonitor:
    """分块过程监控"""

    def __init__(self):
        self.metrics = {
            "total_documents": 0,
            "total_chunks": 0,
            "total_time": 0,
            "errors": 0,
            "strategy_distribution": {},
        }

    def record_batch(self, doc_count: int, chunk_count: int,
                     elapsed: float, stats: dict):
        """记录一批处理结果"""
        self.metrics["total_documents"] += doc_count
        self.metrics["total_chunks"] += chunk_count
        self.metrics["total_time"] += elapsed

        for strategy, count in stats.items():
            self.metrics["strategy_distribution"][strategy] = (
                self.metrics["strategy_distribution"].get(strategy, 0) + count
            )

    def get_report(self) -> dict:
        """生成监控报告"""
        return {
            **self.metrics,
            "avg_chunks_per_doc": (
                self.metrics["total_chunks"]
                / max(self.metrics["total_documents"], 1)
            ),
            "avg_time_per_doc": (
                self.metrics["total_time"]
                / max(self.metrics["total_documents"], 1)
            ),
        }

14.5 Chunk 去重

在分块过程中，可能会产生重复或高度相似的 chunk，需要去重：

def deduplicate_chunks(
    chunks: list,
    similarity_threshold: float = 0.95,
    embeddings=None,
) -> list:
    """
    去除重复或高度相似的 chunk

    Args:
        chunks: chunk 列表
        similarity_threshold: 相似度阈值（超过此值视为重复）
        embeddings: Embedding 模型
    """
    if len(chunks) <= 1:
        return chunks

    # 计算所有 chunk 的 Embedding
    texts = [c.page_content for c in chunks]
    vectors = embeddings.embed_documents(texts)

    # 标记重复
    keep = [True] * len(chunks)
    for i in range(len(chunks)):
        if not keep[i]:
            continue
        for j in range(i + 1, len(chunks)):
            if not keep[j]:
                continue
            sim = cosine_similarity(vectors[i], vectors[j])
            if sim >= similarity_threshold:
                keep[j] = False  # 标记为重复

    return [chunks[i] for i in range(len(chunks)) if keep[i]]

15. 总结与建议

15.1 核心结论

  1. 没有万能的分块策略。不同文档类型、不同业务场景需要不同的分块方案。理解每种策略的原理和适用场景，比记住具体参数更重要。

  2. 结构感知分块是性价比最高的方案。对于有结构的文档（制度、合同、技术手册），章节标题切分在零额外成本下达到接近最优的效果。

  3. 语义分块是通用性最强的方案。Embedding 相似度切分不依赖文档格式，适用于任何类型的文档，但需要额外的计算资源。

  4. LLM 驱动分块是质量最高但成本最高的方案。适用于少量高价值文档，不适合大规模文档库。

  5. 父子文档检索是业界公认的最佳实践。用小 chunk 检索、大 chunk 回答，兼顾精度和完整性。

  6. “Lost in the Middle” 问题不可忽视。即使检索到了正确的 chunk，排序不当也会导致 LLM 忽略关键信息。

  7. 分块策略需要持续评估和迭代。建立评估体系，用数据驱动分块策略的优化。

15.2 选型决策树

开始
  │
  ├── 文档有明确标题结构？
  │     └── 是 → 章节标题切分（方案5）
  │
  ├── 文档是 HTML/Markdown？
  │     └── 是 → HTML/MD 解析切分（方案6）
  │
  ├── 文档是表格/代码等特定格式？
  │     └── 是 → 格式特定切分（方案7）
  │
  ├── 文档是无结构长文本？
  │     └── 是 → Embedding 语义切分（方案8）
  │
  ├── 预算充足 + 文档量少 + 质量要求极高？
  │     └── 是 → LLM 驱动 / 命题提取（方案9/10）
  │
  ├── 需要兼顾检索精度和回答完整性？
  │     └── 是 → 父子文档（方案11）
  │
  └── 兜底 → 递归字符切分（方案2）

15.3 推荐技术栈

组件 推荐方案 说明
Embedding 模型 BGE-M3 中文效果好，支持 8192 tokens，多语言
向量数据库 Chroma / Milvus Chroma 适合小规模（< 100K chunks），Milvus 适合大规模
分块框架 LangChain 内置多种分块器，生态完善
Reranker bge-reranker-v2-m3 与 BGE-M3 配套，8192 tokens
LLM DeepSeek-V3 / GPT-4o DeepSeek 中文强且便宜，GPT-4o 通用强
文档解析 Unstructured / PyMuPDF 支持多种格式，提取质量高

15.4 实施路线图

第1周：基线建立
  - 使用递归字符切分建立基线
  - 准备测试集和评估脚本
  - 记录基线指标

第2周：策略探索
  - 分析文档类型分布
  - 为每种文档类型选择最优策略
  - 实现组合拳分块器

第3周：参数调优
  - 对 chunk_size 和 chunk_overlap 进行网格搜索
  - 对比不同参数组合的效果
  - 选择最优参数

第4周：进阶优化
  - 实现父子文档检索
  - 添加元数据增强
  - 实现 Chunk 排序策略

第5周：生产部署
  - 添加缓存和并发处理
  - 实现增量更新
  - 部署监控和告警

15.5 常见误区

误区 正确做法
“chunk_size 越大越好” 过大的 chunk 导致向量表示模糊，检索精度下降
“overlap 越大越好” 过大的 overlap 浪费存储，且不提升检索效果
“一种策略处理所有文档” 不同文档类型需要不同的分块策略
“分块策略一次设定就不改” 需要根据评估结果持续迭代优化
“只关注检索精度，忽略排序” “Lost in the Middle” 问题会严重影响最终效果
“忽略元数据的作用” 元数据过滤可以大幅提升检索精度

作者注：本文基于实际 RAG 系统开发经验总结，涵盖了从基础到进阶的完整分块策略体系。文档分块是 RAG 系统的基石，投入时间选择合适的策略，远比后期调参收益更大。建议读者根据自身文档特点和业务需求，选择最适合的组合方案，并建立持续的评估和迭代机制。

Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

OpenAI 又往前走了一步：Codex Sites 正在把“开发网站”变成一句话的事

avatar AtomGit开源社区
cover

会议录音秒变播客？用这个AI工具半小时搞定

avatar AtomGit开源社区

遥感图像地块类型土地类型识别分割数据集labelme格式5704张6类别

标注类别名称:["background","barren_land","unknown","urban_land","vegetation","water"]重要说明：可以将数据集用labelme打开编辑，json数据集需自己转成mask或者yolo格式或者coco格式作语义分割或者实例分割。数据集格式：labelme格式(不包含mask文件，仅仅包含jpg图片和对应的json文件)特别声明：本数

avatar AtomGit开源社区