RAG检索后优化：重排序、上下文压缩与提示工程实战

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_fp16=True) # 使用半精度加速 # 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(key=lambda x: x[0], reverse=True) # 降序排列，分数越高越相关 # 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”, temperature=0) summary_prompt = PromptTemplate.from_template( “””请用一段话简洁地总结以下文档的核心内容，保留与问题‘{query}’相关的事实和信息： 文档：{document} 简洁总结：””” ) summary_chain = LLMChain(llm=summary_llm, prompt=summary_prompt) compressed_docs = [] for doc in reranked_docs: # 假设使用重排序后的文档 summary = summary_chain.run(query=user_query, document=doc.page_content) # 创建一个新的Document对象，元数据可以继承或修改 from langchain.schema import Document compressed_doc = Document(page_content=summary, metadata=doc.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(temperature=0) compressor = LLMChainExtractor.from_llm(llm) # 使用ContextualCompressionRetriever进行包装 compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=base_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, threshold=0.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_keep

5. 关键技术三：提示工程优化——给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”文档[{i+1}] (来源：{doc.metadata.get(‘source’， ‘N/A’)}):\n{doc.page_content}” for i, doc in enumerate(compressed_docs)]) # 格式化提示 formatted_prompt = prompt.format_messages(context=context, question=user_question)

设计要点：

1. 强调“严格基于上下文”：明确要求模型不要自由发挥，这是减少幻觉的关键。
2. 允许“不知道”：赋予模型拒绝回答的权利，这比让它胡编乱造要好。
3. 结构化上下文：为每个文档片段添加编号和来源（元数据），有助于模型区分不同来源，也便于后期追溯答案依据。
4. 分隔符清晰：使用---等明显分隔符，帮助模型区分指令、上下文和问题。

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_fp16=True) # 4. 初始化用于压缩/摘要的LLM (可选，按需) # self.summary_llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0) 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(key=lambda x: x[0], reverse=True) 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_content=new_content, metadata=doc.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”【片段{i+1} - 来自 {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”]

流水线设计心得：

1. 模块化：每个步骤（检索、重排序、压缩、提示构建）都是独立的函数或类，方便调试、替换和升级。例如，你可以轻松地将BGE重排序器换成Cohere的API。
2. 可观测性：在每个关键步骤打印日志或记录中间结果（如检索到的原始文档数、重排序后的文档数），这对于排查问题至关重要。例如，如果最终答案不好，你可以检查重排序后的文档是否真的相关。
3. 配置化：将top_k、rerank_top_n、压缩方法、提示模板等作为参数，便于针对不同场景进行调整优化。
4. 异步处理：重排序和LLM调用通常是耗时的IO操作，使用异步（asyncio）可以显著提升吞吐量，尤其是在处理多个并发查询时。

7. 评估与迭代：如何衡量优化效果？

优化不是一次性的工作，需要评估其效果。仅仅感觉“答案变好了”是不够的，需要一些可量化的指标。

7.1 常用评估指标

• 答案相关性：生成的答案与问题的匹配程度。可以通过人工评分，或者使用“评估LLM”（如GPT-4）对其他LLM的答案进行评分。
• 事实准确性/忠实度：答案中的陈述是否严格源自提供的上下文，有没有幻觉。可以计算答案中的关键事实与上下文的匹配比例。
• 上下文利用率：检查最终答案是否用到了重排序后排名靠前的文档。如果答案主要依据排名第五的文档，而忽略了前四，可能说明重排序或提示工程还有问题。
• 检索精度@K：在重排序前后，计算前K个结果中真正相关文档的比例。这是衡量重排序效果最直接的指标。
• 延迟：从发起查询到获得答案的总时间。优化环节会增加延迟，需要评估其带来的精度提升是否值得。

7.2 构建一个简单的评估脚本

你可以针对一组标准问答对（Q&A 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应用给出的答案，会变得更加可靠、精准、令人信服。