📦 第一部分：分块策略 —— 如何优雅地切分长文档
分块（Chunking）是RAG的"地基工程"——分块太大，检索噪声多；分块太小，语义不完整。核心目标是在保持语义完整性和控制检索粒度之间找到最佳平衡。
🎯 五大主流分块策略
1. 固定字符分块（最基础）
deffixed_size_chunking(text,chunk_size=500,overlap=50):""" 按固定字符数切分，带重叠区域防止语义断裂 """chunks=[]start=0whilestart<len(text):end=start+chunk_size chunks.append(text[start:end])start=end-overlap# 重叠区域，保留上下文returnchunks- 优点：实现简单，计算开销小
- 缺点：可能切断句子或段落，语义不完整
- 适用场景：对质量要求不高的快速原型
2. 递归字符分块（LangChain默认方式）
fromlangchain.text_splitterimportRecursiveCharacterTextSplitter splitter=RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=50,separators=["\n\n","\n","。","！","？","，"," ",""]# 优先级递减)chunks=splitter.split_text(document)- 核心思想：优先在段落、句子边界切分，保持自然语义单元
- 优点：比固定切分更"智能"，保留更多语义完整性
- 适用场景：通用场景的首选方案
3. 语义分块（基于句子嵌入）
fromsentence_transformersimportSentenceTransformerimportnumpyasnpdefsemantic_chunking(sentences,similarity_threshold=0.7):""" 计算句子间的语义相似度，相似度低的切分 """model=SentenceTransformer('all-MiniLM-L6-v2')embeddings=model.encode(sentences)chunks=[]current_chunk=[sentences[0]]foriinrange(1,len(sentences)):# 计算当前句子与上一句的相似度sim=np.dot(embeddings[i],embeddings[i-1])/(np.linalg.norm(embeddings[i])*np.linalg.norm(embeddings[i-1]))ifsim<similarity_threshold:chunks.append(" ".join(current_chunk))current_chunk=[sentences[i]]else:current_chunk.append(sentences[i])ifcurrent_chunk:chunks.append(" ".join(current_chunk))returnchunks- 优点：能智能识别主题边界，分块质量高
- 缺点：计算开销大，需要调用模型
- 适用场景：对检索精度要求高的生产环境
4. 文档结构感知分块（针对特定格式）
defmarkdown_aware_chunking(markdown_text):""" 按Markdown的标题层级切分，保持文档结构 """importre# 按标题（# ## ###）分割pattern=r'(#{1,3}\s+[^\n]+\n)'sections=re.split(pattern,markdown_text)chunks=[]current_header=""current_content=[]foriteminsections:ifre.match(r'^#{1,3}\s+',item):# 是标题，保存之前的块ifcurrent_headerandcurrent_content:chunks.append(current_header+"\n".join(current_content))current_header=item current_content=[]else:current_content.append(item)ifcurrent_headerandcurrent_content:chunks.append(current_header+"\n".join(current_content))returnchunks- 适用场景：技术文档、法律文书、财报等结构化文档
5. 大小块混合策略（进阶方案）
defhybrid_chunking(document):""" 同时建立大块（用于上下文）和小块（用于检索） """# 大块：提供完整上下文large_chunks=RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=100).split_text(document)# 小块：用于精确检索small_chunks=RecursiveCharacterTextSplitter(chunk_size=200,chunk_overlap=50).split_text(document)return{"large_chunks":large_chunks,"small_chunks":small_chunks,"strategy":"父文档检索（Parent Document Retrieval）"}- 核心思想：检索时用小块精确定位，返回答案时附带大块上下文
- 优点：兼顾检索精度和回答质量
📊 分块策略选择决策表
| 文档类型 | 推荐策略 | chunk_size | overlap |
|---|---|---|---|
| 短文章（<500字） | 不切分，整文使用 | - | - |
| 技术博客/文章 | 递归字符分块 | 500-800 | 50-100 |
| 法律/政策文档 | 结构感知分块 | 按章节 | - |
| 财报/年报 | 语义分块 | 动态 | - |
| 多主题综合文档 | 混合策略 | 200/1000 | 50/100 |
| 用户评论/对话 | 固定字符分块 | 200-300 | 30-50 |
🎯 第二部分：检索优化 —— 让AI找到最相关的信息
分块只是第一步，更关键的是如何让检索系统精准找到用户问题的相关片段。
🔧 检索优化策略矩阵
策略1：多路召回 + 重排序（Reranking）
核心思想：先快速召回多个候选，再用更精细的模型重新排序。
fromsentence_transformersimportCrossEncoderimportnumpyasnpdefrerank_search(query,chunks,top_k=10):""" 多路召回 + 交叉编码器重排 """# 1. 第一轮：快速召回20个候选（用普通嵌入）embeddings=OpenAIEmbeddings()query_emb=embeddings.embed_query(query)chunk_embs=[embeddings.embed_query(chunk)forchunkinchunks]# 计算余弦相似度，取前20similarities=np.dot(chunk_embs,query_emb)top_indices=np.argsort(similarities)[-20:][::-1]candidates=[chunks[i]foriintop_indices]# 2. 第二轮：用交叉编码器精细排序reranker=CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')scores=reranker.predict([(query,chunk)forchunkincandidates])# 返回排序后的top_kreranked=sorted(zip(candidates,scores),key=lambdax:x[1],reverse=True)return[chunkforchunk,scoreinreranked[:top_k]]- 性能提升：检索精度提升20-30%
- 适用场景：大规模知识库（>10万文档）
策略2：HyDE（假设文档嵌入）
核心思想：让大模型先"想象"一个理想答案，然后用这个假想答案去检索。
defhyde_search(query,llm,knowledge_base):""" HyDE：先让AI生成假设答案，再检索 """# 1. 生成假设答案（不要求准确，只求相关概念）hypothesis=llm.invoke(f"假设你是专家，请针对以下问题写一段可能的答案草稿：{query}")# 2. 用假设答案进行检索（而非原始问题）hypothesis_embed=embeddings.embed_query(hypothesis.content)results=knowledge_base.similarity_search_by_vector(hypothesis_embed)returnresults- 优点：能召回与问题语义相近但关键词不同的文档
- 缺点：增加LLM调用开销（约100-300ms）
策略3：多查询查询（Multi-Query）
核心思想：将用户问题改写为多个不同视角的问题，分别检索后合并去重。
defmulti_query_search(query,llm,knowledge_base):""" 生成5个不同视角的问题，合并检索结果 """# 1. 生成多个查询变体prompt=f"请将以下问题从不同角度改写为5个不同的问题（保持原意）：{query}"variants=llm.invoke(prompt).content.split('\n')[:5]# 2. 每个变体分别检索all_results=[]forqin[query]+variants:results=knowledge_base.similarity_search(q,k=3)all_results.extend(results)# 3. 去重并合并（基于内容相似度）unique_results=deduplicate_documents(all_results)returnunique_results- 适用场景：用户问题模糊、多义性强的场景
策略4：元数据过滤（Metadata Filtering）
核心思想：为文档块附加元数据（来源、日期、类型等），检索时先按条件过滤。
defmetadata_aware_retrieval(query,metadata_filters,knowledge_base):""" 带元数据过滤的检索 """# 例：只检索2024年以后的文档，且类型为"政策"filters={"year":{"$gte":2024},"doc_type":"policy"}results=knowledge_base.similarity_search(query,k=5,filter=filters# 先过滤再检索)returnresults- 优点：大幅减少无关候选，提升精度
- 适用场景：结构化知识库（如企业文档库）
策略5：自适应检索（Adaptive Retrieval）
核心思想：根据问题的复杂度动态调整检索策略。
defadaptive_search(query,llm,knowledge_base):""" 先判断问题复杂度，再选择策略 """# 1. 问题复杂度判断complexity_prompt=f"请判断以下问题的复杂度（简单/中等/复杂）：{query}"level=llm.invoke(complexity_prompt).content.strip()# 2. 动态策略选择iflevel=="简单":returnknowledge_base.similarity_search(query,k=3)eliflevel=="中等":# 使用多路召回returnmulti_query_search(query,llm,knowledge_base)else:# 复杂# 使用HyDE或链式检索returnhyde_search(query,llm,knowledge_base)🚀 实战：完整优化方案代码示例
fromlangchain.retrieversimportContextualCompressionRetrieverfromlangchain.retrievers.document_compressorsimportLLMChainExtractordefbuild_optimized_rag():""" 构建一个整合了多种优化策略的RAG系统 """# 1. 优化分块策略text_splitter=RecursiveCharacterTextSplitter(chunk_size=600,chunk_overlap=80,separators=["\n\n","\n","。","！","？","，"," ",""])docs=text_splitter.split_documents(documents)# 2. 创建向量库embeddings=OpenAIEmbeddings()vectorstore=Chroma.from_documents(docs,embeddings,collection_metadata={"hnsw:space":"cosine"}# 使用余弦相似度)# 3. 基础检索器（多路召回）base_retriever=vectorstore.as_retriever(search_type="mmr",# MMR算法平衡相关性和多样性search_kwargs={"k":20,"fetch_k":50})# 4. 重排序压缩（LLM链式压缩）llm=ChatOpenAI(model="gpt-3.5-turbo")compressor=LLMChainExtractor.from_llm(llm)# 5. 最终检索器final_retriever=ContextualCompressionRetriever(base_compressor=compressor,base_retriever=base_retriever)returnfinal_retriever# 使用优化后的检索器retriever=build_optimized_rag()relevant_docs=retriever.get_relevant_documents("公司年假政策是什么？")📈 优化效果与性能权衡
| 优化策略 | 精度提升 | 延迟增加 | 实现难度 | 推荐指数 |
|---|---|---|---|---|
| 递归字符分块 | +10% | +5ms | ⭐ | ⭐⭐⭐⭐⭐ |
| 语义分块 | +15% | +50ms | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| 多路召回+重排 | +20-30% | +100-200ms | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| HyDE | +15-25% | +200-300ms | ⭐⭐ | ⭐⭐⭐⭐ |
| 元数据过滤 | +25% | +10ms | ⭐⭐ | ⭐⭐⭐⭐⭐ |
| 自适应检索 | +30% | +50-300ms | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
💡 最佳实践建议
- 从小规模开始：先在1000个文档的规模测试分块策略
- 建立评测集：准备50-100个标准问题及期望答案，量化评估每个优化策略的效果
- 分层优化：先优化分块（影响基础），再优化检索（锦上添花）
- 监控关键指标：关注召回的NDCG@10和回答的BLEU/ROUGE分数
- 渐进式部署：在生产环境先用简单策略，逐步迭代复杂优化
分块和检索优化没有"银弹"，最佳实践是根据你的文档类型、用户查询模式、性能要求做AB测试，找到最适合的组合。
