1. 项目概述为什么检索后优化是RAG的“临门一脚”如果你已经跟着前面的系列从零搭建了一个基础的RAG系统可能会发现一个现象系统确实能根据你的问题从文档库里找到一些相关的片段但最终生成的答案有时候还是“差点意思”。要么是答案里夹杂着无关信息显得啰嗦要么是关键的证据没被用上导致回答不够精准甚至有时候模型会基于检索到的片段“自由发挥”说出一些文档里根本没有的内容。这背后的核心问题往往不在于检索本身而在于检索之后、生成答案之前我们缺少了一个关键的“优化”环节。这就是“检索后优化”要解决的问题。你可以把它想象成一位经验丰富的编辑。检索系统比如BM25或向量检索就像一位初级的资料搜集员它从浩如烟海的档案室里抱出来一堆可能相关的文件。但这些文件堆在桌上杂乱无章有的内容重复有的只是擦边相关有的甚至页码都不对。而“检索后优化”这位编辑就需要在这些原始材料送交“总编”大语言模型进行最终撰稿前对它们进行筛选、排序、去重、甚至提炼确保送到总编手里的是最精华、最相关、最可靠的那部分信息。在Advanced RAG的架构里检索后优化是连接“检索”与“生成”的桥梁是提升答案准确性、相关性和可靠性的决定性步骤。它直接决定了你的大模型是基于一堆“矿石”工作还是基于提炼过的“精矿”工作。本次我们就深入这个环节拆解其核心技术与实战策略。2. 检索后优化的核心目标与常见问题在动手优化之前我们必须明确我们要解决哪些具体问题。检索后优化并非一个模糊的概念它针对的是检索结果中几种典型的“病症”。2.1 目标一提升相关性——解决“检索不准”的遗留问题即使我们使用了混合检索BM25向量也无法保证Top-K个返回的片段个个都与问题强相关。总会混入一些相关性较弱但因为某些关键词匹配或语义沾边而被召回的内容。这些“噪音片段”如果直接喂给LLM会干扰其判断可能导致答案偏离核心或者让模型困惑于如何处理矛盾信息。优化目标对初步检索到的片段进行重排序将最相关的片段排到前面甚至过滤掉相关性低于某个阈值的片段。2.2 目标二增强信息密度——解决“信息冗余与碎片化”RAG中常见的文档切片策略可能导致一个完整的答案被切分到多个相邻的片段中。直接检索可能会返回多个高度重叠或连续的片段。此外不同片段可能从不同角度阐述了同一事实造成信息重复。直接将这些片段拼接成上下文会浪费宝贵的上下文窗口Token限额并可能让LLM陷入细节重复。优化目标对检索结果进行去重、合并或摘要消除冗余信息将多个相关片段融合成一个信息密度更高、更连贯的上下文块。2.3 目标三促进答案一致性——解决“信息冲突与缺失”当检索到的多个片段在事实上存在冲突或者关于某个关键点信息不全时LLM可能会生成模棱两可或错误的答案。例如一个片段说某事件发生在A年另一个片段说发生在B年。优化环节需要识别这种冲突并尝试解决或至少标注出来。优化目标通过交叉验证、证据聚合等技术识别并解决检索片段间的信息冲突补全关键信息缺口为LLM提供更一致、更完整的证据集。2.4 目标四引导生成方向——为LLM提供“任务指令”有时候我们不仅需要提供资料还需要告诉LLM如何处理这些资料。例如要求它“严格基于以下文档回答”、“如果文档中没有明确提及请回答‘不知道’”、“请先总结再分点论述”。优化目标在整合好的上下文前面添加清晰、具体的系统提示或指令明确约束LLM的生成行为降低其“幻觉”的可能性。3. 关键技术一重排序——让最相关的信息脱颖而出重排序是检索后优化最直接、最有效的技术之一。它的思路很简单用一个更精细但可能更耗资源的模型对初步检索召回得到的粗粒度结果列表进行重新打分和排序。3.1 为什么需要重排序第一阶段的检索如BM25、向量检索追求的是“召回率”即尽可能把所有相关的文档都找出来避免遗漏。因此它们通常速度很快但排序的精度是相对粗糙的。BM25依赖于词频统计向量检索依赖于嵌入模型的语义表示能力它们都无法完美理解查询与文档之间复杂的语义匹配和逻辑关系。重排序模型则像一个专业的评审它能够更深入地理解查询的意图和文档的细节进行一对一的精细匹配判断。常用的重排序模型如BGE-Reranker、Cohere Rerank等都是经过大量查询文档相关性对训练出来的它们输出的分数更能代表“答案质量”。3.2 实战使用BGE-Reranker进行本地化重排序假设我们已经通过LangChain的Retriever获取了10个初步的文档片段。接下来我们使用BAAI/bge-reranker-base这个开源模型进行重排序。from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 注意LangChain也支持集成的Reranker这里演示更底层的调用以理解原理 # 假设我们有一个基础检索器 base_retriever # from langchain.vectorstores import Chroma # vectorstore Chroma(...) # base_retriever vectorstore.as_retriever(search_kwargs{k: 10}) # 1. 获取初始检索结果 raw_docs base_retriever.get_relevant_documents(你的问题是什么) # 2. 初始化重排序模型这里以使用FlagEmbedding库为例 from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16True) # 使用半精度加速 # 3. 准备重排序数据对 pairs [[“你的问题是什么”, doc.page_content] for doc in raw_docs] # 4. 计算相关性分数 scores reranker.compute_score(pairs) # 返回一个分数列表 # 5. 根据分数对文档和原始元数据进行排序 scored_docs list(zip(scores, raw_docs)) scored_docs.sort(keylambda x: x[0], reverseTrue) # 降序排列分数越高越相关 # 6. 取出Top-N个优化后的文档 top_n 5 reranked_docs [doc for _, doc in scored_docs[:top_n]]关键细节与避坑指南计算成本重排序模型需要将每个查询文档对都进行一次前向计算。如果初始检索的k值很大比如50且文档内容很长计算开销会显著增加。需要在召回质量和计算延迟之间做权衡。通常先使用廉价检索器召回20-30个再用重排序精选出5-8个是性价比很高的策略。模型选择bge-reranker-base是平衡精度与速度的好选择。如果追求极致精度可以考虑bge-reranker-large但速度会慢很多。对于中文场景BAAI/bge-reranker-v2-m3是专门为多语言优化的版本效果更好。分数阈值可以设置一个相关性分数阈值例如只保留分数大于0.5的文档。这能有效过滤掉弱相关片段。但这个阈值需要在自己的业务数据上进行少量测试来确定没有通用值。与向量库集成一些向量数据库如Weaviate, Qdrant已经内置了重排序支持可以在查询时直接指定重排序模型更为方便。Milvus等也支持通过插件方式接入。4. 关键技术二上下文压缩与提炼——从“一堆资料”到“一份简报”重排序解决了“哪个更好”的问题但没解决“内容太多太杂”的问题。上下文压缩的目标就是在不损失核心信息的前提下缩短上下文长度。4.1 文档提炼用LLM总结长文档对于检索到的长文档我们可以先用一个快速的LLM比如GPT-3.5-Turbo或更小的本地模型对其进行摘要然后将摘要而非原文送入最终的生成LLM。from langchain.chat_models import ChatOpenAI from langchain.chains import LLMChain from langchain.prompts import PromptTemplate summary_llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) summary_prompt PromptTemplate.from_template( “””请用一段话简洁地总结以下文档的核心内容保留与问题‘{query}’相关的事实和信息 文档{document} 简洁总结””” ) summary_chain LLMChain(llmsummary_llm, promptsummary_prompt) compressed_docs [] for doc in reranked_docs: # 假设使用重排序后的文档 summary summary_chain.run(queryuser_query, documentdoc.page_content) # 创建一个新的Document对象元数据可以继承或修改 from langchain.schema import Document compressed_doc Document(page_contentsummary, metadatadoc.metadata) compressed_docs.append(compressed_doc)注意事项二次幻觉风险总结模型也可能在摘要过程中引入错误或遗漏关键细节。这对于要求高事实准确性的场景是风险。成本与延迟对每个检索片段都进行摘要意味着额外的LLM API调用成本和耗时。通常只对最长的或最重要的几个片段进行摘要。更佳实践可以考虑“提取式摘要”即直接从原文中找出最重要的句子例如通过嵌入相似度找出与查询最相关的原文句子而不是“生成式摘要”。这样能绝对保证事实准确性。4.2 上下文过滤只提取最相关的句子这是更轻量、更安全的压缩方法。LLMChainExtractor是LangChain提供的一个经典工具它利用LLM的能力根据查询从文档中提取出相关的句子并丢弃其他部分。from langchain.chat_models import ChatOpenAI from langchain.retrievers.document_compressors import LLMChainExtractor llm ChatOpenAI(temperature0) compressor LLMChainExtractor.from_llm(llm) # 使用ContextualCompressionRetriever进行包装 compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) # 现在这个retriever返回的就是被压缩过滤后的文档 compressed_docs compression_retriever.get_relevant_documents(“你的问题”)这个链的工作原理是让LLM针对原始文档和查询输出一个只包含相关句子的新文档。它比全文摘要更精准但同样有LLM调用开销。4.3 简单去重基于嵌入或哈希对于明显重复或高度重叠的片段可以在前期用更廉价的方法处理。基于文本哈希的去重对文档内容计算MD5或SimHash完全相同的或几乎相同的可以直接去重。但这种方法无法处理语义重复。基于嵌入聚类的去重计算所有片段嵌入的余弦相似度如果某些片段彼此非常相似可以只保留其中一个代表。这需要计算一个相似度矩阵对于片段数不多的情况是可行的。import numpy as np from sklearn.metrics.pairwise import cosine_similarity def deduplicate_by_embedding(docs, threshold0.95): “””基于嵌入相似度去重””” embeddings [get_embedding(doc.page_content) for doc in docs] # 假设有嵌入函数 sim_matrix cosine_similarity(embeddings) to_keep [] visited_indices set() for i in range(len(docs)): if i in visited_indices: continue to_keep.append(docs[i]) # 找到所有与i高度相似的索引 duplicate_indices np.where(sim_matrix[i] threshold)[0] visited_indices.update(duplicate_indices) return to_keep5. 关键技术三提示工程优化——给LLM清晰的“任务清单”检索后优化也包括对最终提交给LLM的提示进行精心设计。一段好的提示能极大地约束模型行为提升答案质量。5.1 结构化上下文与指令不要简单地把所有优化后的文档用\n\n拼接起来就扔给LLM。应该构建一个结构清晰、指令明确的提示模板。from langchain.prompts import ChatPromptTemplate system_template “””你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。 如果上下文中的信息不足以回答问题请直接说“根据提供的信息我无法回答这个问题”。不要编造信息。 请保持答案简洁、准确。 上下文信息 {context} “”” human_template “问题{question}” prompt ChatPromptTemplate.from_messages([ (“system”, system_template), (“human”, human_template) ]) # 构建最终上下文 context “\n\n---\n\n”.join([f”文档[{i1}] (来源{doc.metadata.get(‘source’ ‘N/A’)}):\n{doc.page_content}” for i, doc in enumerate(compressed_docs)]) # 格式化提示 formatted_prompt prompt.format_messages(contextcontext, questionuser_question)设计要点强调“严格基于上下文”明确要求模型不要自由发挥这是减少幻觉的关键。允许“不知道”赋予模型拒绝回答的权利这比让它胡编乱造要好。结构化上下文为每个文档片段添加编号和来源元数据有助于模型区分不同来源也便于后期追溯答案依据。分隔符清晰使用---等明显分隔符帮助模型区分指令、上下文和问题。5.2 少样本示例Few-Shot引导对于复杂任务可以在系统提示中提供一两个输入输出的例子引导模型遵循你期望的格式和推理方式。系统提示 你是一个根据法律条文回答问题的助手。请根据提供的法律条款上下文回答问题并引用相关条款号。 示例 上下文 [文档1] 《XX法》第10条当事人订立合同应当具有相应的民事权利能力和民事行为能力。 [文档2] 《XX法》第11条当事人可以委托代理人订立合同。 问题一个未成年人可以自己签订购买手机的合同吗 回答根据《XX法》第10条订立合同需要具有相应的民事行为能力。未成年人通常被视为限制民事行为能力人或无民事行为能力人因此一般不能独立签订此类合同可能需要法定代理人同意或追认。 现在请根据以下上下文回答真实问题 ...5.3 后处理指令你还可以在提示中要求模型对答案进行后处理比如“请用中文回答。”“请先给出是或否的结论再解释原因。”“请将答案组织成要点列表。”“如果涉及多个方面请分点论述。”这些指令能让你更可控地得到格式规整、符合需求的答案。6. 综合实战构建一个完整的检索后优化流水线现在我们将上述技术组合起来构建一个从原始检索到最终提示的完整流水线。这个流水线将包含混合检索 - 重排序 - 上下文压缩 - 提示工程。import asyncio from typing import List from langchain.schema import Document from langchain.retrievers import EnsembleRetriever from langchain.vectorstores import Chroma from langchain.retrievers import BM25Retriever from FlagEmbedding import FlagReranker class AdvancedRAGPipeline: def __init__(self, vector_store, text_corpus, reranker_model‘BAAI/bge-reranker-base’): “”” 初始化流水线。 :param vector_store: 已加载的向量库对象如Chroma实例 :param text_corpus: 用于BM25检索的原始文本列表与向量库文档对应 “”” # 1. 初始化基础检索器 self.vector_retriever vector_store.as_retriever(search_kwargs{“k”: 20}) self.bm25_retriever BM25Retriever.from_texts(text_corpus) # 需预先拟合 self.bm25_retriever.k 20 # 2. 初始化混合检索器 self.ensemble_retriever EnsembleRetriever( retrievers[self.bm25_retriever, self.vector_retriever], weights[0.4, 0.6] # 可以调整权重 ) # 3. 初始化重排序模型 self.reranker FlagReranker(reranker_model, use_fp16True) # 4. 初始化用于压缩/摘要的LLM (可选按需) # self.summary_llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) async def retrieve_and_rerank(self, query: str, top_k: int 20, rerank_top_n: int 8) - List[Document]: “””执行混合检索与重排序””” print(f“开始混合检索查询: ‘{query}’”) # 第一步混合检索 raw_docs self.ensemble_retriever.get_relevant_documents(query) print(f“混合检索到 {len(raw_docs)} 个初始文档”) if not raw_docs: return [] # 第二步准备重排序对 pairs [[query, doc.page_content] for doc in raw_docs] # 第三步计算重排序分数 (注意这里可能是计算密集型考虑异步或批量) # 为了演示我们同步计算。生产环境可考虑批量或使用GPU加速。 scores self.reranker.compute_score(pairs) # 第四步排序并选取 scored_docs list(zip(scores, raw_docs)) scored_docs.sort(keylambda x: x[0], reverseTrue) reranked_docs [doc for _, doc in scored_docs[:rerank_top_n]] print(f“重排序后保留 {len(reranked_docs)} 个文档”) return reranked_docs def compress_context(self, query: str, docs: List[Document], method: str “extract”) - List[Document]: “””上下文压缩””” compressed [] if method “extract”: # 简易版只取每个文档的前N个字符或句子根据查询相关性这里简化处理 # 更复杂的实现可以用LLM进行提取如之前提到的LLMChainExtractor for doc in docs: # 这里只是一个示例简单截断。实际应用中应使用更智能的方法。 content doc.page_content # 假设我们找到包含查询关键词的第一段简化逻辑 sentences content.split(‘。’) relevant_sentences [s for s in sentences if any(kw in s for kw in query.split())] if relevant_sentences: new_content ‘。’.join(relevant_sentences[:3]) ‘。’ # 最多取三句 else: new_content content[:500] # 否则取前500字符 new_doc Document(page_contentnew_content, metadatadoc.metadata) compressed.append(new_doc) # 可以扩展其他压缩方法如‘summarize’ return compressed or docs # 如果压缩失败返回原文档 def build_prompt(self, query: str, context_docs: List[Document]) - str: “””构建最终提示””” context_parts [] for i, doc in enumerate(context_docs): source doc.metadata.get(‘source’ ‘未知来源’) context_parts.append(f”【片段{i1} - 来自 {source}】\n{doc.page_content}\n”) full_context “\n” “-”*50 “\n”.join(context_parts) “-”*50 prompt_template f”””你是一个准确、可靠的AI助手。请严格根据以下提供的上下文信息来回答问题。 上下文信息是来自知识库的片段可能包含不相关的内容请自行判断。 如果上下文信息不足以回答请明确告知“根据现有信息无法回答”。 请保持答案客观不要添加未提及的信息。 上下文信息{full_context} 问题{query} 请基于上下文回答””” return prompt_template async def run(self, query: str) - dict: “””执行完整流水线””” # 1. 检索与重排序 reranked_docs await self.retrieve_and_rerank(query) if not reranked_docs: return {“answer”: “未检索到相关文档无法回答。”, “sources”: []} # 2. 上下文压缩 compressed_docs self.compress_context(query, reranked_docs) # 3. 构建提示 final_prompt self.build_prompt(query, compressed_docs) # 4. 调用LLM生成答案 (这里需要接入你的LLM如OpenAI, 本地模型等) # final_answer await self.call_llm(final_prompt) # 为演示我们返回提示和来源 sources [doc.metadata.get(‘source’ ‘N/A’) for doc in compressed_docs] return { “optimized_prompt”: final_prompt, # 实际使用中这里应该是answer “retrieved_sources”: sources, “context_docs”: compressed_docs } # 使用示例 # 假设已有vector_store和corpus # pipeline AdvancedRAGPipeline(vector_store, corpus) # result asyncio.run(pipeline.run(“你的问题”)) # print(result[“optimized_prompt”]) # 或 result[“answer”]流水线设计心得模块化每个步骤检索、重排序、压缩、提示构建都是独立的函数或类方便调试、替换和升级。例如你可以轻松地将BGE重排序器换成Cohere的API。可观测性在每个关键步骤打印日志或记录中间结果如检索到的原始文档数、重排序后的文档数这对于排查问题至关重要。例如如果最终答案不好你可以检查重排序后的文档是否真的相关。配置化将top_k、rerank_top_n、压缩方法、提示模板等作为参数便于针对不同场景进行调整优化。异步处理重排序和LLM调用通常是耗时的IO操作使用异步asyncio可以显著提升吞吐量尤其是在处理多个并发查询时。7. 评估与迭代如何衡量优化效果优化不是一次性的工作需要评估其效果。仅仅感觉“答案变好了”是不够的需要一些可量化的指标。7.1 常用评估指标答案相关性生成的答案与问题的匹配程度。可以通过人工评分或者使用“评估LLM”如GPT-4对其他LLM的答案进行评分。事实准确性/忠实度答案中的陈述是否严格源自提供的上下文有没有幻觉。可以计算答案中的关键事实与上下文的匹配比例。上下文利用率检查最终答案是否用到了重排序后排名靠前的文档。如果答案主要依据排名第五的文档而忽略了前四可能说明重排序或提示工程还有问题。检索精度K在重排序前后计算前K个结果中真正相关文档的比例。这是衡量重排序效果最直接的指标。延迟从发起查询到获得答案的总时间。优化环节会增加延迟需要评估其带来的精度提升是否值得。7.2 构建一个简单的评估脚本你可以针对一组标准问答对QA Pair进行自动化测试。import asyncio from your_rag_pipeline import AdvancedRAGPipeline # 导入你的流水线 eval_questions [ “什么是机器学习”, “RAG的主要优势是什么”, # ... 更多问题 ] ground_truth_answers { “什么是机器学习”: “机器学习是人工智能的一个分支…标准答案” # ... } async def evaluate_pipeline(pipeline, questions, ground_truth): results [] for q in questions: print(f“评估问题: {q}”) start_time asyncio.get_event_loop().time() # 运行你的流水线 pipeline_result await pipeline.run(q) generated_answer pipeline_result[“answer”] # 假设你的流水线返回答案 end_time asyncio.get_event_loop().time() latency end_time - start_time # 这里需要实现一个评分函数可以是调用GPT-4进行评分也可以是简单的字符串匹配 # 例如使用语义相似度通过嵌入计算作为相关性分数 relevance_score calculate_similarity(generated_answer, ground_truth.get(q, “”)) # 检查幻觉简化版可以检查生成答案中的命名实体是否出现在上下文中 context_text “ ”.join([doc.page_content for doc in pipeline_result.get(“context_docs”, [])]) hallucination_flag check_hallucination(generated_answer, context_text) results.append({ “question”: q, “generated_answer”: generated_answer, “latency”: latency, “relevance_score”: relevance_score, “has_hallucination”: hallucination_flag, “retrieved_sources”: pipeline_result.get(“retrieved_sources”, []) }) return results # 计算平均分 def analyze_results(results): avg_latency sum(r[“latency”] for r in results) / len(results) avg_relevance sum(r[“relevance_score”] for r in results) / len(results) hallucination_rate sum(1 for r in results if r[“has_hallucination”]) / len(results) print(f“平均延迟: {avg_latency:.2f}秒”) print(f“平均相关性分数: {avg_relevance:.2f}”) print(f“幻觉发生率: {hallucination_rate:.2%}”)通过对比增加优化步骤前后的评估结果你可以清晰地看到重排序、压缩等环节带来的具体收益如相关性提升、幻觉减少以及付出的代价延迟增加。这是指导你下一步优化方向例如是否要换一个更快的重排序模型是否要调整压缩强度的关键依据。检索后优化是打磨RAG系统、使其从“能用”到“好用”的必经之路。它没有一成不变的银弹需要你根据自身数据的特性、业务对准确性与速度的要求灵活选择和组合这些技术。核心思想始终是让高质量、高相关、高密度的信息以最清晰的方式呈现在大语言模型面前。当你把这些细节都做到位你会发现你的RAG应用给出的答案会变得更加可靠、精准、令人信服。