RAG 文档切分策略实战：块大小不是拍脑袋定的

RAG 文档切分策略实战块大小不是拍脑袋定的一、默认 chunk_size500准确率 62%调到 800反而降到 55%文档切分Chunking是 RAG 最基础的步骤。也是最容易被忽视的步骤。很多人用默认的 500 字符切分从未调过。块太小一个完整概念被切碎。Go 语言的并发模型基于 CSP通过 goroutine 和 channel 实现通信。如果切成两半语义完整性丢失。块太大一个块包含多个不相关的概念。检索命中后模型需要在 2000 字中找到用户真正关心的 50 字。召回精度下降。没有万能块大小。最佳大小取决于文档类型、问题和模型上下文窗口。为什么块大小的影响是非线性的因为 RAG 检索本质上是一个匹配精度 vs 上下文丰富度的权衡。块太小 → 匹配精度高因为每块只包含一个明确的概念但上下文不完整前后文被切断了。块太大 → 上下文丰富一整段完整描述但匹配精度低查询词可能在段落中的任意位置向量被稀释。在我们的测试中技术文档的甜蜜点在 500-800 字符之间但代码文档的最佳块大小是按函数/类切分可能一个函数就 2000 字符法律合同则偏好按条款编号切分。文档类型决定了最佳切分粒度而不是一个固定的字符数。这个发现意味着RAG 系统不应该对整个知识库使用统一的块大小。不同来源、不同类型的文档应该使用不同的切分策略甚至同一个文档内部也应该根据内容结构使用混合策略。二、分块策略的决策树flowchart TB A[文档类型] -- B{结构化文档?} B --|是| C[按章节/段落分块] B --|否| D{代码文档?} D --|是| E[按函数/类分块] D --|否| F{对话/QA?} F --|是| G[按 QA 对分块] F --|否| H[语义分块] C -- I[块大小: 500-1000] E -- J[块大小: 按函数长度] G -- K[块大小: QA对] H -- L[按相似度切分]三种核心策略固定大小滑动窗口最简单适合通用文本。递归字符分割按段落→句子→词的优先级切分。语义分块用 embedding 相似度判断切分点。固定大小滑动窗口是最暴力的方案——给一个 chunk_size 和 overlap直接按字符数切。优点是简单缺点是可能把CSP和goroutine切进两个块让基于 CSP 的并发模型这个完整概念变成碎片。递归字符分割是改进方案——优先在自然边界段落、句子切分只有在段落本身过长时才降级到句子级或字符级切分。这是我们最推荐的默认方案它对 80% 的文档类型都有合理的效果。语义分块是最先进的方案——计算相邻句子的 Embedding 相似度在相似度骤降点话题切换切分。效果最好但成本最高需要对所有句子做 Embedding适合对准确度要求极高的场景。三、Python 分块策略实现下面的代码实现了三种分块器递归分割适合通用文本语义分块适合长文档代码分块适合程序代码。create_chunker函数根据文档类型自动选择合适的分块策略。 chunker.py - RAG 文档切分策略 import re import hashlib from typing import List, Dict, Optional, Iterator from dataclasses import dataclass import numpy as np dataclass class Chunk: 文档块 id: str text: str metadata: Dict start_pos: int end_pos: int class RecursiveChunker: 递归字符分割器 优先在段落边界切分其次句子最后强制切割 def __init__( self, chunk_size: int 500, chunk_overlap: int 50, separators: List[str] None, ): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.separators separators or [ \n\n, # 段落 \n, # 换行 。, # 中文句号 . , # 英文句号 , # 分号 , # 空格 , # 字符级 ] def split(self, text: str) - List[Chunk]: 递归分割文本 chunks [] self._split_recursive(text, 0, chunks) return chunks def _split_recursive( self, text: str, start_offset: int, chunks: List[Chunk] ): if len(text) self.chunk_size: if text.strip(): chunk_id self._make_id(text, start_offset) chunks.append(Chunk( idchunk_id, texttext.strip(), metadata{length: len(text)}, start_posstart_offset, end_posstart_offset len(text), )) return # 尝试用当前分隔符切分 for sep in self.separators: if sep : # 强制字符级切割 self._force_split(text, start_offset, chunks) return parts text.split(sep) if len(parts) 1: continue # 合并小段落切割大段落 current for part in parts: if len(current) len(part) len(sep) self.chunk_size: current part sep else: if current.strip(): chunk_id self._make_id(current, start_offset) chunks.append(Chunk( idchunk_id, textcurrent.strip(), metadata{length: len(current)}, start_posstart_offset, end_posstart_offset len(current), )) start_offset len(current) current part sep if current.strip(): chunk_id self._make_id(current, start_offset) chunks.append(Chunk( idchunk_id, textcurrent.strip(), metadata{length: len(current)}, start_posstart_offset, end_posstart_offset len(current), )) return def _force_split( self, text: str, start_offset: int, chunks: List[Chunk] ): 强制按字符切割 step self.chunk_size - self.chunk_overlap for i in range(0, len(text), step): chunk_text text[i:i self.chunk_size] if chunk_text.strip(): chunk_id self._make_id(chunk_text, start_offset i) chunks.append(Chunk( idchunk_id, textchunk_text.strip(), metadata{length: len(chunk_text)}, start_posstart_offset i, end_posstart_offset i len(chunk_text), )) def _make_id(self, text: str, offset: int) - str: h hashlib.md5(text.encode()).hexdigest()[:8] return fchunk_{offset}_{h} class SemanticChunker: 语义分块器 当相邻句子的向量相似度低于阈值时切分 def __init__( self, embedding_fn, similarity_threshold: float 0.5, min_chunk_size: int 100, max_chunk_size: int 1000, ): self.embed embedding_fn self.threshold similarity_threshold self.min_size min_chunk_size self.max_size max_chunk_size def split(self, text: str) - List[Chunk]: 语义分块 sentences self._split_sentences(text) if len(sentences) 1: return [Chunk( idself._make_id(text, 0), texttext.strip(), metadata{length: len(text)}, start_pos0, end_poslen(text), )] # 计算句子向量 embeddings self.embed(sentences) # 找切分点 cut_points self._find_cut_points(embeddings) # 生成块 chunks [] offset 0 start 0 for cut in cut_points [len(sentences)]: chunk_text .join(sentences[start:cut]) if chunk_text.strip(): chunk_id self._make_id(chunk_text, offset) chunks.append(Chunk( idchunk_id, textchunk_text.strip(), metadata{length: len(chunk_text)}, start_posoffset, end_posoffset len(chunk_text), )) offset len(chunk_text) start cut return chunks def _find_cut_points(self, embeddings: List) - List[int]: 通过余弦相似度找切分点 cut_points [] for i in range(1, len(embeddings)): similarity self._cosine_sim(embeddings[i-1], embeddings[i]) if similarity self.threshold: cut_points.append(i) return cut_points def _cosine_sim(self, a, b) - float: a, b np.array(a), np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) 1e-8) def _split_sentences(self, text: str) - List[str]: return re.split(r(?[。.!?])\s*, text) def _make_id(self, text: str, offset: int) - str: h hashlib.md5(text.encode()).hexdigest()[:8] return fchunk_{offset}_{h} class CodeChunker: 代码分块器按函数/类切分 def split(self, code: str, language: str python) - List[Chunk]: 按代码结构分块 chunks [] offset 0 if language python: pattern r(def |class ) elif language go: pattern r(func |type ) else: pattern r(function |class ) parts re.split(f({pattern}), code) current for part in parts: if re.match(pattern, part): if current.strip(): chunks.append(Chunk( idself._make_id(current, offset), textcurrent.strip(), metadata{language: language}, start_posoffset, end_posoffset len(current), )) offset len(current) current part else: current part if current.strip(): chunks.append(Chunk( idself._make_id(current, offset), textcurrent.strip(), metadata{language: language}, start_posoffset, end_posoffset len(current), )) return chunks def _make_id(self, text: str, offset: int) - str: h hashlib.md5(text.encode()).hexdigest()[:8] return fcode_{offset}_{h} # ---- 策略选择函数 ---- def create_chunker( doc_type: str, chunk_size: int 500, embedding_fnNone, ) - object: 根据文档类型选择分块策略 if doc_type code: return CodeChunker() elif doc_type semantic: if embedding_fn is None: raise ValueError(语义分块需要 embedding_fn) return SemanticChunker(embedding_fn) else: return RecursiveChunker(chunk_sizechunk_size)RecursiveChunker的分隔符优先级设计是整个分块器的精髓。它优先在段落边界\n\n切分降级到句子边界。再降级到空格最后才到字符级强制切分。这种逐级降级策略保证了大块优先按照语义边界切只有在段落本身超过 chunk_size 时才动用更细粒度的切分。四、分块调优的实验方法用一批标注过的问答对评估不同块大小。评估指标Top-3 召回率正确答案在前 3 条中的比例。对几种块大小做网格搜索200, 400, 500, 800, 1000, 1500。选择召回率最高的配置。不同文档类型需要不同策略技术文档按标题层级递归分割。法律合同按条款编号分割。聊天记录按对话轮次分割。评估分块效果不是靠直觉而是靠标注数据。建议准备 50-100 个标注问答对query 正确答案所在的 chunk用不同的分块策略处理后看正确答案的召回率。这是一个需要反复迭代的过程——调参数、跑评估、看结果、再调参数。建议把这个流程做成一个可重复跑的评估脚本每次改分块逻辑后都能快速验证效果。还有一个进阶技巧混合检索 分块策略。块大小和检索方式也有联动关系——大块适合 BM25 关键词检索因为关键词密度高小块适合向量检索因为语义集中。如果你的 RAG 使用了混合检索可以对不同类型的分块使用不同的检索权重进一步优化效果。五、总结文档分块是 RAG 的基础工程直接影响召回质量。递归分割适合通用文本按段落-语句优先级切割。语义分块适合长文档通过向量相似度判断切点。代码分块按函数/类级别保持结构完整性。最佳块大小需针对具体文档做网格搜索确定。最后一条建议不要花超过 20% 的优化时间在分块上。分块确实重要但 RAG 的优化空间是一个多层次系统——Embedding 模型选择、检索策略向量/BM25/混合、重排序Rerank、查询重写等。分块是基础中的基础做好它然后把精力投入在更高杠杆的优化方向上。