一、为什么分块是 RAG 的 原罪1.1 固定大小分块的三大致命缺陷现在使用的固定大小分块如 512token / 块是最简单也是最常用的分块方法但它存在三个无法解决的根本性问题这也是 90% 的 RAG 系统回答质量差的根源。缺陷 1上下文断裂这是最常见也最严重的问题。当一个完整的语义单元被分割到两个不同的块中时检索到任何一个块都无法获得完整的信息。示例文档原文检索增强生成技术的核心优势在于它可以利用外部知识库来增强大模型的回答能力从而有效减少幻觉问题。分块结果块 1检索增强生成技术的核心优势在于它可以利用外部知识库来增强大模型的回答能力块 2从而有效减少幻觉问题。当用户问 RAG 如何解决幻觉问题 时系统可能只能检索到块 1大模型看到的信息是不完整的自然无法给出准确的回答。缺陷 2语义不完整固定大小分块会将不同主题的内容强行合并到同一个块中导致块的语义混杂影响检索的准确性。示例一个 512token 的块中可能同时包含 JVM 内存结构 和 JVM 垃圾回收算法 两个完全不同的主题当用户查询 JVM 内存结构 时这个块会被检索到但其中一半的内容是无关的这些无关内容会作为噪声进入上下文降低大模型的回答质量缺陷 3粒度不匹配不同的查询需要不同粒度的信息。有些查询只需要一个句子就能回答而有些查询需要整个段落甚至多个段落的信息。固定大小分块无法适应这种粒度差异。示例查询 什么是 RAG只需要一个定义句子查询 RAG 的工作流程是什么需要一个完整的段落查询 如何构建一个生产级 RAG 系统需要多个段落甚至整个章节固定大小分块要么提供的信息太少无法回答复杂问题要么提供的信息太多包含大量噪声。1.2 工业界三大主流高级检索架构为了解决固定大小分块的问题工业界经过多年的实践总结出了三种主流的高级检索架构它们各有优缺点适用于不同的场景。检索架构核心思想解决的问题优点缺点适用场景句子窗口检索用细粒度的句子进行检索返回句子前后 N 个句子的上下文上下文断裂实现简单、效果好、速度快窗口大小固定无法自适应大多数通用场景父文档检索将文档分成小的子块用于检索返回整个父块语义不完整、粒度不匹配可以获得完整的语义单元父块大小固定可能包含无关内容结构化文档、技术手册分层检索先检索文档摘要再检索具体内容块大规模文档库的检索效率可以处理百万级以上的文档实现复杂、需要额外生成摘要大规模知识库、企业文档库1.3 句子窗口检索最实用的高级检索方法句子窗口检索是目前工业界最常用的高级检索方法它的实现最简单效果提升最明显几乎适用于所有场景。工作原理分块阶段将文档分割成单个句子或非常小的块如 128token检索阶段用这些细粒度的句子进行检索找到最相关的句子返回阶段返回该句子前后 N 个句子的上下文而不仅仅是句子本身示例分块将文档分割成 100 个句子检索找到最相关的句子是第 35 句返回返回第 30 句到第 40 句的上下文窗口大小为 10核心优势完美解决上下文断裂问题相关句子的上下文会被完整返回检索精度更高细粒度的句子检索可以更准确地找到相关信息实现简单只需要在现有分块和检索逻辑上做少量修改速度快细粒度的向量检索速度比粗粒度更快关键参数窗口大小窗口大小是句子窗口检索最重要的参数它决定了返回的上下文长度。窗口太小仍然会有上下文断裂问题窗口太大会引入过多的无关内容工业界最佳实践窗口大小设置为 5-10 个句子或 512-1024 个 token1.4 父文档检索结构化文档的最佳选择父文档检索特别适合处理结构化文档如技术手册、API 文档、书籍等这些文档有明确的章节和段落结构。工作原理分块阶段将文档分成大的父块如每个段落或每个小节512-2048token将每个父块再分成小的子块如每个句子或 128token建立子块到父块的映射关系检索阶段用细粒度的子块进行检索找到最相关的子块返回阶段返回子块对应的整个父块而不仅仅是子块本身示例父块一个完整的段落1000token子块将这个段落分成 5 个句子每个句子约 200token检索找到最相关的子块是第 3 个句子返回返回整个段落1000token核心优势获得完整的语义单元返回的是一个完整的段落或小节语义完整粒度匹配更好用细粒度的子块检索用粗粒度的父块回答适合结构化文档可以利用文档的天然结构进行分块关键参数父子块大小比父子块大小比是父文档检索最重要的参数它决定了检索精度和上下文完整性的平衡。工业界最佳实践父子块大小比设置为 4:1 到 8:1例如父块 1024token子块 128-256token1.5 分层检索大规模文档库的解决方案分层检索主要用于处理百万级以上的大规模文档库它通过多级检索来提升检索效率和效果。工作原理索引阶段为每个文档生成一个摘要构建摘要层向量库和内容层向量库检索阶段第一层检索摘要层找到最相关的几个文档第二层在这些相关文档的内容层中进行细粒度检索返回阶段返回内容层的检索结果核心优势检索效率高先在摘要层进行粗筛大大减少了需要处理的文档数量可以处理大规模文档库支持百万级甚至千万级的文档效果更好摘要层可以更好地捕捉文档的整体主题局限性实现复杂需要额外生成文档摘要维护两个向量库摘要质量影响整体效果如果摘要生成得不好会导致相关文档被过滤掉不适合小规模文档库对于几千个文档的小规模场景反而会增加复杂度二、从零构建高级检索架构2.1 重构文档分块器支持父子块关联结构高级检索架构的基础是支持父子块关联的分块器。我们需要重构原来的简单分块器使其能够生成父子块结构并建立子块到父块的映射关系。第一步实现句子级分块首先我们需要实现一个能够将文档分割成句子的分块器。我们将使用nltk库的句子分割器它对中文和英文都有很好的支持。import nltk from nltk.tokenize import sent_tokenize import uuid # 下载nltk的句子分割模型第一次运行需要 nltk.download(punkt) def split_into_sentences(text: str) - list: 将文本分割成句子 # 处理中文句子分割 text text.replace(。, 。\n).replace(, \n).replace(, \n) sentences sent_tokenize(text) # 过滤空句子 sentences [s.strip() for s in sentences if s.strip()] return sentences第二步实现父子块分块器接下来我们实现一个通用的父子块分块器它可以将文档分割成父块和子块并建立它们之间的映射关系。def chunk_document_with_parent_child( text: str, parent_chunk_size: int 1024, child_chunk_size: int 128, overlap: int 0, metadata: dict None ) - tuple[list, dict]: 将文档分割成父子块结构 :param text: 文档文本 :param parent_chunk_size: 父块大小token数 :param child_chunk_size: 子块大小token数 :param overlap: 块重叠大小 :param metadata: 文档元数据 :return: (所有块列表, 子块到父块的映射字典) if metadata is None: metadata {} # 第一步将文档分割成句子 sentences split_into_sentences(text) # 第二步将句子组合成父块 parent_chunks [] current_parent [] current_parent_length 0 for sentence in sentences: sentence_length len(sentence) // 4 # 简单估算token数1个token≈4个字符 if current_parent_length sentence_length parent_chunk_size and current_parent: # 父块已满保存并新建 parent_text .join(current_parent) parent_chunks.append(parent_text) current_parent [] current_parent_length 0 current_parent.append(sentence) current_parent_length sentence_length # 保存最后一个父块 if current_parent: parent_text .join(current_parent) parent_chunks.append(parent_text) # 第三步将每个父块分割成子块并建立映射 all_chunks [] child_to_parent {} for parent_idx, parent_text in enumerate(parent_chunks): parent_id str(uuid.uuid4()) # 添加父块到所有块列表 parent_chunk { id: parent_id, text: parent_text, metadata: { **metadata, chunk_type: parent, parent_idx: parent_idx } } all_chunks.append(parent_chunk) # 将父块分割成子块 child_sentences split_into_sentences(parent_text) current_child [] current_child_length 0 for child_sentence in child_sentences: child_sentence_length len(child_sentence) // 4 if current_child_length child_sentence_length child_chunk_size and current_child: # 子块已满保存并新建 child_text .join(current_child) child_id str(uuid.uuid4()) child_chunk { id: child_id, text: child_text, metadata: { **metadata, chunk_type: child, parent_id: parent_id, parent_idx: parent_idx } } all_chunks.append(child_chunk) # 建立子块到父块的映射 child_to_parent[child_id] parent_id current_child [] current_child_length 0 current_child.append(child_sentence) current_child_length child_sentence_length # 保存最后一个子块 if current_child: child_text .join(current_child) child_id str(uuid.uuid4()) child_chunk { id: child_id, text: child_text, metadata: { **metadata, chunk_type: child, parent_id: parent_id, parent_idx: parent_idx } } all_chunks.append(child_chunk) child_to_parent[child_id] parent_id return all_chunks, child_to_parent代码解释我们首先将文档分割成句子这是所有高级分块的基础然后将句子组合成大的父块默认 1024token再将每个父块分割成小的子块默认 128token为每个块生成唯一的 ID并在子块的元数据中记录对应的父块 ID最后返回所有块的列表和子块到父块的映射字典2.2 实现句子窗口检索器句子窗口检索器的核心是检索到相关的子块后返回该子块前后 N 个句子的上下文。def sentence_window_retrieval( self, query: str, top_k: int 5, window_size: int 5 ) - list: 句子窗口检索 :param query: 用户查询 :param top_k: 返回结果数量 :param window_size: 窗口大小前后各window_size个句子 :return: 检索结果列表 # 第一步检索最相关的子块 child_results self.semantic_search(query, top_ktop_k*2) # 第二步为每个子块扩展上下文窗口 window_results [] for child_result in child_results: # 获取子块所在的父块 parent_id child_result[metadata][parent_id] parent_chunk next((c for c in self.chunks if c[id] parent_id), None) if not parent_chunk: continue # 将父块分割成句子 parent_sentences split_into_sentences(parent_chunk[text]) # 找到子块在父块中的位置 child_text child_result[text] child_sentence_idx -1 for i, sentence in enumerate(parent_sentences): if child_text in sentence or sentence in child_text: child_sentence_idx i break if child_sentence_idx -1: continue # 计算窗口的起始和结束位置 start_idx max(0, child_sentence_idx - window_size) end_idx min(len(parent_sentences), child_sentence_idx window_size 1) # 提取窗口内的句子 window_sentences parent_sentences[start_idx:end_idx] window_text .join(window_sentences) # 创建窗口结果 window_result { id: child_result[id], text: window_text, metadata: { **child_result[metadata], window_size: window_size, original_sentence: child_text, sentence_idx: child_sentence_idx }, score: child_result[score] } window_results.append(window_result) # 去重并按得分排序 seen_texts set() unique_results [] for result in window_results: if result[text] not in seen_texts: seen_texts.add(result[text]) unique_results.append(result) # 按得分降序排序 unique_results.sort(keylambda x: x[score], reverseTrue) return unique_results[:top_k]代码解释首先用细粒度的子块进行检索返回 Top-2*K 个结果为了去重对于每个检索到的子块找到它所在的父块将父块分割成句子找到子块对应的句子在父块中的位置提取该句子前后window_size个句子的上下文最后去重并按得分排序返回 Top-K 个结果2.3 实现父文档检索器父文档检索器的核心是检索到相关的子块后返回该子块对应的整个父块。def parent_document_retrieval( self, query: str, top_k: int 5 ) - list: 父文档检索 :param query: 用户查询 :param top_k: 返回结果数量 :return: 检索结果列表 # 第一步检索最相关的子块 child_results self.semantic_search(query, top_ktop_k*2) # 第二步收集对应的父块 parent_results [] seen_parent_ids set() for child_result in child_results: parent_id child_result[metadata][parent_id] if parent_id in seen_parent_ids: continue seen_parent_ids.add(parent_id) # 获取父块 parent_chunk next((c for c in self.chunks if c[id] parent_id), None) if not parent_chunk: continue # 创建父块结果 parent_result { id: parent_id, text: parent_chunk[text], metadata: { **parent_chunk[metadata], child_id: child_result[id], child_score: child_result[score] }, score: child_result[score] } parent_results.append(parent_result) if len(parent_results) top_k: break return parent_results代码解释首先用细粒度的子块进行检索返回 Top-2*K 个结果对于每个检索到的子块找到它对应的父块去重避免同一个父块被多次返回最后返回 Top-K 个父块结果2.4 统一检索接口为了方便使用和对比不同的检索策略我们需要统一检索接口支持一键切换不同的检索方法。在HybridRetriever类中添加以下方法def advanced_search( self, query: str, top_k: int 5, retrieval_method: str sentence_window, window_size: int 5 ) - list: 统一高级检索接口 :param query: 用户查询 :param top_k: 返回结果数量 :param retrieval_method: 检索方法sentence_window / parent_document / hybrid :param window_size: 句子窗口大小仅sentence_window方法有效 :return: 检索结果列表 if retrieval_method sentence_window: return self.sentence_window_retrieval(query, top_k, window_size) elif retrieval_method parent_document: return self.parent_document_retrieval(query, top_k) elif retrieval_method hybrid: # 混合使用句子窗口和父文档检索 sw_results self.sentence_window_retrieval(query, top_k//2, window_size) pd_results self.parent_document_retrieval(query, top_k//2) # 合并结果并去重 all_results sw_results pd_results all_results.sort(keylambda x: x[score], reverseTrue) seen_texts set() unique_results [] for result in all_results: if result[text] not in seen_texts: seen_texts.add(result[text]) unique_results.append(result) return unique_results[:top_k] else: raise ValueError(f不支持的检索方法{retrieval_method})2.5 集成到 RAG 核心类最后我们需要将高级检索集成到 RAG 核心类中替换原来的简单检索。打开rag_core.py找到调用检索的地方修改为def query(self, question, top_k5, streamFalse, max_context_tokens12000): processed_question self._preprocess_query(question) if not processed_question: return 请输入有效的问题。 # 使用高级检索 retrieved_docs self.retriever.advanced_search( queryprocessed_question, top_ktop_k, retrieval_methodsentence_window, # 可以切换为parent_document或hybrid window_size5 ) # 重排序 reranked_docs self.retriever.reranker.rerank(processed_question, retrieved_docs, top_ktop_k) print(f\n 最终送入大模型的文档数量{len(reranked_docs)}) if not reranked_docs: return 抱歉知识库中没有找到相关信息无法回答您的问题。 # 后续的上下文拼接和大模型生成代码保持不变 # ...
