RAG与Agent性能调优4.切片语义割裂怎么办Gitee地址https://gitee.com/agiforgagaplus/OptiRAGAgent文章详情目录RAG与Agent性能调优上一节第3节领域术语种混淆构建精准数语库提升检索一致性下一节待更新什么是滑动窗口我们可以把文档想象成一条连绵不断的河流而滑动窗口就像是一个移动观察框。它每次只关注一小段然后逐步向前滑动依次扫描整个文档避免切片过短确保每个片段都有一定的信息量同时合理设置窗口大小和布长实现对全文覆盖式处理关键词如何发挥作用关键词通常是文档中的核心概念、专业术语或高频语义单元它们承载着文本的主要信息特点如果某个窗口的边界恰好切断了一个关键词或者打断了一个完整的语义单元我们就可以根据关键词位置动态调整切片的起始或结束位置从而保证语语义的完整性滑动窗口关键词智能切片将两者结合我们可以构建一套更加智能的切片流程初步切片使用滑动窗口对文档进行基础分段关键词检测分析每个切片的边界是否存在关键词或语义单元中断动态调整根据关键词的位置和上下文语义微调切片边缘确保语义完整且不冗余实战演练均为伪代码详细代码进仓库查看Token切片优点控制输出长度适用于小模型实现简单缺点容易造成语义割裂依赖overlap补偿效果有限不适合复杂语义任务import os from llama_index.core import VectorStoreIndex, Settings, Document from llama_index.core.node_parser import SentenceWindowNodeParser, SemanticSplitterNodeParser, TokenTextSplitter from llama_index.core.postprocessor import MetadataReplacementPostProcessor # 导入 OpenAILike LLM (用于 DashScope 兼容模式Qwen 模型) from llama_index.llms.openai_like import OpenAILike # 导入 DashScopeEmbedding (用于阿里云 DashScope 嵌入模型) from llama_index.embeddings.dashscope import DashScopeEmbedding # 1. 配置 LLM (使用 DashScope 的 qwen-plus 模型通过 OpenAILike 调用) # 2. 配置嵌入模型 (使用 DashScopeEmbedding 类) def evaluate_splitter(splitter, documents, question, splitter_name): 评测不同文档切片方法的效果 手动打印召回结果方便直接对比切分效果。 print(f\n{*50}) print(f正在使用 {splitter_name} 方法进行测试...) print(f{*50}\n) # 显示 raw chunks generated by the splitter print(f【{splitter_name}】生成的原始文档切片 (Nodes):) raw_nodes splitter.get_nodes_from_documents(documents) for i, node in enumerate(raw_nodes, 1): print(f\n 切片 {i}:) if isinstance(splitter, SentenceWindowNodeParser): original_text node.metadata.get(original_text, node.get_content()) window_context node.metadata.get(window, N/A - 窗口内容未生成) print(f 核心内容: \{original_text}\) print(f 完整窗口上下文(供LLM用): \{window_context}\) else: print(f 内容: \{node.get_content()}\) # Add metadata for debugging if needed # print(f 元数据: {node.metadata}) print( - * 40) print(\n *50) # --- 开始测试不同的切片策略 --- # Token 切片 (Character/Token-based) token_splitter TokenTextSplitter( chunk_size30, # Small chunk size to demonstrate forced breaks chunk_overlap0 # No overlap for clear distinct chunks ) evaluate_splitter(token_splitter, documents, question, Token 切片 (chunk_size30)) token_splitter TokenTextSplitter( chunk_size30, # Small chunk size to demonstrate forced breaks chunk_overlap10 # No overlap for clear distinct chunks ) evaluate_splitter(token_splitter, documents, question, Token 切片 (chunk_size30,chunk_overlap10 ))句子切片特性TokenTextSplitterSentenceSplitter切分单位Token句子chunk_size硬性上限强制截断软性目标优先保持句子完整chunk_overlapToken 级别重叠可能包含不完整句子句子级别重叠确保上下文自然衔接sentence_splitter SentenceSplitter( chunk_size512, chunk_overlap50 ) evaluate_splitter(sentence_splitter, documents, question, Sentence)句子窗口切片切分单元句子核心机制为每个句子附加上下文窗口检索方式基于句子的精准召回生成方式使用上下文窗口提升语义完整性适用场景精准问答摘要生成解释型问答等对上下文敏感的任务# 句子窗口切片 (Sentence Window) sentence_window_splitter SentenceWindowNodeParser.from_defaults( window_size3, window_metadata_keywindow, original_text_metadata_keyoriginal_text ) evaluate_splitter(sentence_window_splitter, documents, question, Sentence Window)句子的滑动窗口切片特性句子切片句子窗口切片基于句子的滑动窗口保持句子完整性✅✅✅切块之间重叠❌❌✅上下文丰富程度一般强需后处理强天然包含检索准确性一般一般强适用场景基础文本处理精准问答、解释型任务复杂语义匹配、长文本检索def demonstrate_sliding_window_splitter(documents, chunk_size, chunk_overlap): 演示 LlamaIndex 中保持句子完整性的滑动窗口切片。 Args: documents (list[Document]): 待切分的文档列表。 chunk_size (int): 每个切块的目标 Token 数量。 chunk_overlap (int): 相邻切块之间重叠的 Token 数量。 print(f\n{*50}) print(f正在演示【滑动窗口切片】...) print(f切块大小 (chunk_size): {chunk_size}) print(f重叠大小 (chunk_overlap): {chunk_overlap}) print(f{*50}\n) # --- 第一步创建切分器 --- # SentenceSplitter 优先保持句子完整性再考虑大小 splitter SentenceSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap ) # --- 第二步执行切分 --- # 获取切分后的节点切块 nodes splitter.get_nodes_from_documents(documents) # --- 第三步打印切分结果展示重叠效果 --- print(\n--- 切分后生成的原始切块---) print(f文档被切分为 {len(nodes)} 个切块。) for i, node in enumerate(nodes, 1): content node.get_content().strip() print(f\n【切块 {i}】 (长度: {len(content)} 字符):) print(- * 50) print(f内容:\n\{content}\) print(- * 50) # --- 简单的切分效果分析观察相邻切块的重叠部分 --- print(\n--- 关键点观察相邻切块的重叠部分 ---) if len(nodes) 1: # 为了更好地展示重叠我们只截取重叠部分的内容 # 由于是句子级别的切分重叠部分是完整的句子 overlap_content_end_of_chunk1 nodes[0].get_content()[-chunk_overlap:].strip() overlap_content_start_of_chunk2 nodes[1].get_content()[:chunk_overlap].strip() print(f切块 1 的末尾 ({chunk_overlap} 字符): \...{overlap_content_end_of_chunk1}\) print(f切块 2 的开头 ({chunk_overlap} 字符): \{overlap_content_start_of_chunk2}...\) print(f\n你可以看到切块 1 的末尾与切块 2 的开头存在重叠这就是 chunk_overlap 的作用。) else: print(文档太短未能生成多个切块。请使用更长的文档以观察效果。) print(f\n滑动窗口切片测试完成。) print(f{*50}\n) # --- 调用滑动窗口切片演示函数 --- # 调整 chunk_size 和 chunk_overlap 观察不同效果 demonstrate_sliding_window_splitter(documents, chunk_size150, chunk_overlap50)语义切片# --- 3. 定义一个能处理中文的自定义分句函数 --- # 这个函数本身就是 SemanticSplitterNodeParser 所需要的句子切分器 def chinese_sentence_tokenizer(text: str) - list[str]: sentences re.findall(r[^。…\n][。…\n]?, text) return [s.strip() for s in sentences if s.strip()] def plot_similarity_and_chunks(splitter: SemanticSplitterNodeParser, title: str): # 使用我们自己的函数进行可视化部分的句子切分这部分逻辑是正确的 sentences chinese_sentence_tokenizer(document.get_content()) if len(sentences) 2: print(f错误只找到了 {len(sentences)} 个句子无法计算句子间的相似度。) return print(f正在为 {len(sentences)} 个句子生成嵌入向量...) embeddings Settings.embed_model.get_text_embedding_batch(sentences, show_progressTrue) def cosine_similarity(v1, v2): return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)) similarities [cosine_similarity(embeddings[i], embeddings[i1]) for i in range(len(embeddings) - 1)] breakpoint_threshold_val np.percentile(similarities, splitter.breakpoint_percentile_threshold) print(f计算出的相似度阈值为: {breakpoint_threshold_val:.4f}) # --- 可视化 --- plt.figure(figsize(12, 6)) plt.plot(similarities, markero, linestyle-, label相邻句子相似度) plt.axhline(ybreakpoint_threshold_val, colorr, linestyle--, labelf切分阈值 ({splitter.breakpoint_percentile_threshold}百分位)) plt.title(title, fontsize16) plt.xlabel(句子连接处索引) plt.ylabel(余弦相似度) plt.legend() try: plt.rcParams[font.sans-serif] [SimHei] plt.rcParams[axes.unicode_minus] False except Exception: print(无法设置中文字体 SimHei图表中的中文可能显示为乱码。) plt.grid(True) plt.show() # --- 实际切分 --- # 这部分现在应该可以正常工作了 nodes splitter.get_nodes_from_documents([document]) print(\n--- 切分结果 ---) for i, node in enumerate(nodes): print(f 节点 {i1} (长度: {len(llama_tokenizer(node.get_content()))} tokens) ) print(node.get_content().strip()) print(- * 20) # --- 5. 直接将我们编写的函数传递给 SemanticSplitterNodeParser # 实验一使用保守阈值 (95) print(*20 实验一使用保守阈值 (95) *20) conservative_splitter SemanticSplitterNodeParser( buffer_size1, breakpoint_percentile_threshold95, embed_modelSettings.embed_model, sentence_splitterchinese_sentence_tokenizer ) plot_similarity_and_chunks(conservative_splitter, 相似度与切分点 (阈值95%)) # 实验二使用激进阈值 (5) print(\n *20 实验二使用激进阈值 (5) *20) aggressive_splitter SemanticSplitterNodeParser( buffer_size1, breakpoint_percentile_threshold5, embed_modelSettings.embed_model, sentence_splitterchinese_sentence_tokenizer ) plot_similarity_and_chunks(aggressive_splitter, 相似度与切分点 (阈值5%))滑动窗口关键词语义切片策略类型优点缺点滑动窗口 (SentenceSplitter)1.上下文保留通过chunk_overlap确保切块边界的信息不会丢失有效缓解了“答案在切块边缘”的问题。 2.大小可控可以大致控制切块的chunk_size对LLM的上下文窗口友好。 3.速度快不依赖昂贵的Embedding模型调用计算成本低。1.语义盲目完全不理解文本内容可能在逻辑最紧密的地方强行切分导致“语义割裂”。 2.冗余度高产生大量重叠内容增加了索引的存储成本和检索时的计算量。语义切分 (SemanticSplitter)1.语义完整性切块的边界就是语义的边界每个块都是一个高度内聚的、完整的逻辑单元。 2.信噪比高提供给LLM的上下文非常“干净”几乎没有无关信息。 3.动态大小切块大小自适应文本的逻辑结构。1.边界上下文丢失如果用户的答案恰好需要结合两个语义块边界的信息可能会因为没有重叠而检索不全。 2.大小不可控可能产生非常大的语义块超出LLM能处理的上下文长度限制。 3.成本高需要为每个句子计算嵌入向量速度慢且有API调用成本。DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) # --- 自定义混合解析器类 (已更新为带详细打印的版本) --- class HybridNodeParser(NodeParser): primary_parser: NodeParser secondary_parser: NodeParser max_chunk_size: int 1024 tokenizer: Callable Field(default_factoryget_tokenizer, excludeTrue) def _parse_nodes(self, documents: List[Document], **kwargs) - List[Document]: print(--- 开始执行【混合切分】... ---) primary_nodes self.primary_parser.get_nodes_from_documents(documents) print(f\n{*25} 第一步语义切分结果 {*25}) print(f初步切分出 {len(primary_nodes)} 个语义段落。) for i, p_node in enumerate(primary_nodes, 1): print(f\n【原始语义段落 {i}】 (大小: {len(self.tokenizer(p_node.get_content()))} tokens)) print(- * 60) print(textwrap.indent(p_node.get_content().strip(), )) print(- * 60) print(f\n{*25} 第二步检查与二次切分过程 {*25}) final_nodes [] for i, node in enumerate(primary_nodes, 1): node_size len(self.tokenizer(node.get_content())) print(f\n 正在检查【原始语义段落 {i}】 (大小: {node_size} tokens)...) if node_size self.max_chunk_size: print(f └── 结果: 大小合适 ( {self.max_chunk_size} tokens)直接采纳。) final_nodes.append(node) else: print(f └── 结果: 段落过大 ( {self.max_chunk_size} tokens)将使用滑动窗口进行二次切分。) print(\n 【即将被切分的原始内容】) print( -*50) print(textwrap.indent(node.get_content().strip(), | )) print( -*50) sub_nodes self.secondary_parser.get_nodes_from_documents([Document(textnode.get_content())]) print(f\n 【二次切分结果】: 被切分成了 {len(sub_nodes)} 个重叠的子切块。) for j, s_node in enumerate(sub_nodes, 1): print(f\n 【子切块 {i}.{j}】 (大小: {len(self.tokenizer(s_node.get_content()))} tokens)) print( -*40) print(textwrap.indent(s_node.get_content().strip(), | )) print( -*40) final_nodes.extend(sub_nodes) print(\n--- 【混合切分】完成---) return final_nodes classmethod def from_defaults(cls, **kwargs): raise NotImplementedError(请直接实例化此类不要使用 from_defaults) # --- 实例化两个基础的解析器 --- semantic_parser SemanticSplitterNodeParser( buffer_size1, breakpoint_percentile_threshold95, sentence_splitterchinese_sentence_tokenizer, embed_modelSettings.embed_model ) window_parser SentenceSplitter( chunk_size256, chunk_overlap50 ) # --- 实例化并使用我们的混合解析器 --- hybrid_parser HybridNodeParser( primary_parsersemantic_parser, secondary_parserwindow_parser, max_chunk_size300, tokenizerget_tokenizer() ) # 执行混合切分 final_nodes hybrid_parser.get_nodes_from_documents([long_document])