引言基础 RAG 的天花板标准 RAGRetrieval-Augmented Generation的工作模式很简单用户问题 → 向量检索 → 拼接上下文 → 生成回答。这个模式解决了 LLM 知识截止的问题但在复杂查询场景下暴露出明显局限-单次检索不足复杂问题需要多轮信息收集一次检索无法覆盖所有需要的信息-查询不够精准用户原始问题直接用于检索往往匹配不到最相关的文档-无法推理整合检索到的多个文档片段模型难以自主判断哪些关联性更强-缺乏迭代能力发现信息不足时不能主动补充检索Agentic RAG 通过引入 Agent 的规划与迭代机制让检索系统真正具备思考能力。—## 一、Agentic RAG 的核心架构### 1.1 从被动检索到主动推理基础 RAG 用户问题 → [一次检索] → 生成答案 Agentic RAG 用户问题 → Agent 分析 → [规划检索策略] → 执行检索① → 评估充分性 ↓ 不充分 [调整检索策略] → 执行检索② ↓ 充分 整合信息 → 生成答案### 1.2 三种 Agentic RAG 模式**模式一查询规划型Query Planning**将复杂问题分解为子查询分别检索后综合pythonfrom openai import OpenAIimport jsonclient OpenAI()def plan_queries(user_question: str) - list[str]: 将复杂问题分解为多个检索子查询 response client.chat.completions.create( modelgpt-4o, messages[ { role: system, content: 你是一个查询规划专家。将用户问题分解为2-4个具体的检索子查询。 每个子查询应该独立完整能直接用于文档检索。 输出 JSON 格式{queries: [子查询1, 子查询2, ...]} }, {role: user, content: user_question} ], response_format{type: json_object} ) result json.loads(response.choices[0].message.content) return result[queries]# 示例question 对比分析 GPT-4o 和 Claude 3.5 Sonnet 在代码生成和长文本理解方面的差异sub_queries plan_queries(question)# 输出# [GPT-4o 代码生成能力评测, Claude 3.5 Sonnet 代码生成能力, # GPT-4o 长文本理解与处理, Claude 3.5 Sonnet 长上下文性能]**模式二迭代检索型Iterative Retrieval**基于当前检索结果决定是否继续检索pythondef iterative_rag(question: str, retriever, max_iterations: int 3) - str: 迭代式 RAG每轮评估信息充分性不足则继续检索 collected_docs [] current_query question for iteration in range(max_iterations): # 检索文档 new_docs retriever.search(current_query, top_k3) collected_docs.extend(new_docs) # 评估当前信息是否足以回答问题 context \n\n.join([d.content for d in collected_docs]) evaluation_response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: 评估当前收集的信息是否足以回答用户问题。输出 JSON{\sufficient\: true/false, \missing\: \缺失的关键信息\, \next_query\: \下一步检索建议\} }, { role: user, content: f问题{question}\n\n已收集信息\n{context} } ], response_format{type: json_object} ) eval_result json.loads(evaluation_response.choices[0].message.content) if eval_result[sufficient] or iteration max_iterations - 1: break # 基于评估结果生成更精准的下一步查询 current_query eval_result[next_query] # 基于所有收集的文档生成最终答案 return generate_answer(question, collected_docs)**模式三工具调用型Tool-based RAG**将检索作为工具由 Agent 自主决定何时调用pythonretrieval_tools [ { type: function, function: { name: semantic_search, description: 在知识库中进行语义搜索找到与查询最相关的文档片段, parameters: { type: object, properties: { query: { type: string, description: 搜索查询使用自然语言描述需要找的信息 }, top_k: { type: integer, description: 返回结果数量1-10之间默认5, default: 5 }, filter_category: { type: string, description: 限定搜索的文档类别可选 } }, required: [query] } } }, { type: function, function: { name: keyword_search, description: 精确关键词搜索适合查找特定术语、产品名称、版本号等, parameters: { type: object, properties: { keywords: { type: array, items: {type: string}, description: 关键词列表 } }, required: [keywords] } } }]—## 二、查询改写与扩展### 2.1 HyDE假设文档嵌入用模型生成假设答案再用答案嵌入检索提升召回率pythondef hyde_retrieve(question: str, retriever, embedder) - list: HyDE先生成假设答案再用答案做检索 # Step 1: 生成假设答案不需要真实准确只需语义接近 hypothetical_response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: 根据问题生成一个假设性的详细答案。内容可以不完全准确但要语义完整、专业。 }, {role: user, content: question} ] ) hypothetical_doc hypothetical_response.choices[0].message.content # Step 2: 用假设答案的嵌入进行检索比直接用问题检索召回率更高 hyp_embedding embedder.encode(hypothetical_doc) docs retriever.search_by_vector(hyp_embedding, top_k5) return docs### 2.2 查询扩展与多向检索pythondef multi_query_retrieve(question: str, retriever) - list: 多角度查询扩展提高覆盖率 # 生成多个角度的查询变体 response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: 从3个不同角度改写用户查询增加检索覆盖面。 输出 JSON{variants: [变体1, 变体2, 变体3]} 角度建议同义词替换、更具体的表述、更通用的表述 }, {role: user, content: question} ], response_format{type: json_object} ) variants json.loads(response.choices[0].message.content)[variants] all_queries [question] variants # 并行检索所有变体 import concurrent.futures all_docs [] seen_ids set() with concurrent.futures.ThreadPoolExecutor() as executor: futures [executor.submit(retriever.search, q, top_k3) for q in all_queries] for future in concurrent.futures.as_completed(futures): for doc in future.result(): if doc.id not in seen_ids: all_docs.append(doc) seen_ids.add(doc.id) return all_docs—## 三、检索后处理Reranking 与过滤### 3.1 交叉编码器重排序pythonfrom sentence_transformers import CrossEncoderreranker CrossEncoder(BAAI/bge-reranker-v2-m3)def rerank_documents(query: str, docs: list, top_k: int 5) - list: 使用交叉编码器重排序提升检索精度 if len(docs) top_k: return docs # 构建 (query, doc) 对 pairs [(query, doc.content) for doc in docs] # 计算相关性分数 scores reranker.predict(pairs) # 按分数排序 ranked sorted(zip(docs, scores), keylambda x: x[1], reverseTrue) return [doc for doc, score in ranked[:top_k]]### 3.2 LLM 辅助过滤pythondef llm_filter_documents(query: str, docs: list) - list: 使用 LLM 过滤无关文档 filtered [] for doc in docs: response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: 判断文档是否与查询相关。输出 JSON{\relevant\: true/false, \reason\: \简要说明\} }, { role: user, content: f查询{query}\n\n文档{doc.content[:500]} } ], response_format{type: json_object} ) result json.loads(response.choices[0].message.content) if result[relevant]: filtered.append(doc) return filtered—## 四、完整 Agentic RAG 系统实现pythonclass AgenticRAGSystem: def __init__(self, vector_store, llm_client, rerankerNone): self.vector_store vector_store self.client llm_client self.reranker reranker self.max_iterations 3 def answer(self, question: str) - dict: 完整的 Agentic RAG 推理流程 # 1. 查询规划 sub_queries self._plan_queries(question) # 2. 并行多查询检索 all_docs self._parallel_retrieve(sub_queries) # 3. 重排序 if self.reranker: all_docs self._rerank(question, all_docs) # 4. 评估与迭代 final_docs, iterations self._iterative_refine(question, all_docs) # 5. 生成答案 answer self._generate_answer(question, final_docs) return { answer: answer, source_docs: [d.metadata for d in final_docs], iterations: iterations, sub_queries: sub_queries } def _plan_queries(self, question: str) - list[str]: # 查询规划逻辑见上文 pass def _parallel_retrieve(self, queries: list[str]) - list: import concurrent.futures all_docs [] seen set() with concurrent.futures.ThreadPoolExecutor() as executor: futures [executor.submit(self.vector_store.search, q, 5) for q in queries] for f in concurrent.futures.as_completed(futures): for doc in f.result(): if doc.id not in seen: all_docs.append(doc) seen.add(doc.id) return all_docs—## 五、性能与成本权衡| 策略 | 准确率提升 | 延迟增加 | 成本增加 | 适用场景 ||------|----------|---------|---------|---------|| 查询规划 | 15-25% | 500ms | 30% | 复杂多跳问题 || HyDE | 10-20% | 300ms | 20% | 专业领域问答 || 迭代检索 | 20-30% | 1-3s | 50-100% | 深度研究类问题 || Reranking | 10-15% | 200ms | 较小 | 通用场景 |实践建议- 简单事实性问题标准 RAG 即可- 中等复杂问题查询规划 Reranking- 复杂分析性问题完整 Agentic RAG 流程—## 结语Agentic RAG 不是对基础 RAG 的否定而是在其基础上增加了主动思考的能力。关键的工程选择在于什么场景下值得付出额外的延迟和成本换取更高质量的答案。对于大多数企业知识库问答场景从查询规划 多路检索 Reranking这三个改进出发是性价比最高的第一步。