从"查一次就生成"到"想清楚再回答"——让检索增强生成学会自主思考

如果你在项目里用过 RAG（检索增强生成），一定经历过这个时刻：用户问了一个复杂问题，你的 pipeline 老老实实地检索了 top-5 文档块，塞进 prompt，生成答案——结果牛头不对马嘴。

问题不在模型质量，而在于传统的 RAG 假设"一次检索就够了"。但现实世界的复杂问题——跨文档推理、多跳问答、需要先收集信息再判断的问题——一锤子检索永远搞不定。

Agentic RAG正是为了解决这个痛点而诞生：把检索从"一次性查询"升级为"自主推理闭环"。Agent 不再被动地接收检索结果然后生成，而是像一个研究员一样——拆解问题、多次检索、评估信息质量、重写查询、决定何时停止、何时换个角度再来一次。

本文会从以下四个维度逐一拆解 Agentic RAG 的核心机制，并给出可直接运行的代码实现：

• 路由 Agent：根据问题类型选择最优检索策略
• 查询重写：让模糊问题变精确，让检索命中率翻倍
• 迭代检索 + 自检：多次检索直到信息充分，而不是一次就停
• 多源融合与评分：从多个知识源收集证据并交叉验证

本文适合：已有基础 RAG 实践经验的开发者，想从"能跑"到"跑得好"。读完你将能实现一个具备自主检索决策能力的 Agentic RAG 系统。

一、RAG 进化简史：从 Naive 到 Agentic

先厘清概念。RAG 的演进大致分四个阶段：

本文重点拆解Agentic RAG这一层。它不是 Multi-Agent RAG 的平替，而是后者的基础单元——先把一个 Agent 的自主检索能力做扎实，再说多 Agent 协作。

⚡ 关键思维转变：在 Naive RAG 中，你问"怎么检索最相关的文档？"在 Agentic RAG 中，你问"Agent 需要什么信息才能回答这个问题？它要怎么找？找了之后够了吗？"——检索从机械动作变成了推理过程的一部分。

二、Agentic RAG 的核心架构

Agentic RAG 的心智模型可以用一个循环来描述：

用户问题 → [路由] → [查询重写] → [检索] → [评估] → ──不够──┐
↑ │
└──────────── [重写查询、换策略] ──┘
↓ 够了
[融合 + 生成最终答案]

路由 Agent

判断问题类型（事实型/推理型/对比型），选择最强检索策略，避免"用同一套路回答所有问题"。

✏️

查询重写

将用户自然语言查询转换为更适合检索的精确表达，包括关键词提取、同义词扩展、子问题拆分。

迭代检索 + 自检

检索后先自检：信息够了吗？相关吗？不够就换个角度再搜。这是 Agentic 区别于 Naive 的核心差异。

多源融合

向量库、关键词索引、SQL 数据库、Web API——不同工具擅长不同查询，Agent 自主选择调用哪个。

三、路由 Agent：不同问题，不同策略

传统 RAG 对所有问题一视同仁：向量化 → 搜 top-k → 生成。但"特斯拉 Model Y 的续航是多少？"和"比较特斯拉和小鹏的自动驾驶技术路线"是本质不同的两类问题。前者一句话就能回答，后者需要多轮检索和深度合成。

路由 Agent在检索开始前先做一次分类，将问题路由到最合适的检索策略：

from typing import Literal from langgraph.graph import StateGraph, END from pydantic import BaseModel, Field # --- 定义路由类别 --- class RouteDecision(BaseModel): category: Literal["factual", "comparison", "multi_hop", "analytical"] = Field( description="问题类型：factual=简单事实, comparison=对比, multi_hop=多跳推理, analytical=深度分析" ) reasoning: str = Field(description="分类理由") class AgentState(BaseModel): query: str route: RouteDecision | None = None retrieved_docs: list[str] = [] evaluation: dict = {} final_answer: str = "" # --- 路由节点：让 LLM 决定走哪条策略 --- async def router_node(state: AgentState) -> AgentState: prompt = f"""分析以下用户问题，决定最佳检索策略： 问题：{state.query} - factual：简单事实查询，一次精确检索即可 - comparison：对比两个或多个对象，需要分别检索并比较 - multi_hop：需要多步推理，上一步结论影响下一步检索 - analytical：需要深度分析，可能涉及多轮检索和综合判断 请以 JSON 格式返回：{{"category": "...", "reasoning": "..."}}""" response = await llm.invoke(prompt) state.route = RouteDecision.model_validate_json(response) return state # --- 四种检索策略 --- async def factual_search(state: AgentState) -> AgentState: """高精度单次检索，top-k 取小，threshold 取高""" docs = await vector_store.similarity_search( state.query, k=3, score_threshold=0.85 ) state.retrieved_docs = [d.page_content for d in docs] return state async def comparison_search(state: AgentState) -> AgentState: """拆分比较对象，各自检索，再合成""" # Step 1: 让 LLM 提取比较对象 entities_prompt = f"请从以下问题中提取需要对比的对象列表（JSON数组）：\n{state.query}" entities = json.loads(await llm.invoke(entities_prompt)) all_docs = [] for entity in entities: docs = await vector_store.similarity_search( f"{entity} 核心特性 技术路线", k=5 ) all_docs.extend([f"[{entity}] {d.page_content}" for d in docs]) state.retrieved_docs = all_docs return state async def multi_hop_search(state: AgentState) -> AgentState: """[检索→推理→生成新查询→再检索] 循环""" for round_num in range(3): docs = await vector_store.similarity_search(state.query, k=5) state.retrieved_docs.extend([d.page_content for d in docs]) # 基于已有信息，生成下一个查询 next_query_prompt = f"""基于已有信息：{state.retrieved_docs[-3:]} 原始问题：{state.query} 如果想完整回答原始问题，还需要了解什么？生成一个新的检索查询。 如果信息已经足够，回复 "ENOUGH"。""" next_query = await llm.invoke(next_query_prompt) if "ENOUGH" in next_query: break state.query = next_query.strip() return state async def analytical_search(state: AgentState) -> AgentState: """多角度多源检索：向量 + 关键词 + 可能涉及工具调用""" # 向量检索 vector_docs = await vector_store.similarity_search(state.query, k=8) # 关键词检索（BM25）补充精确匹配 keyword_docs = bm25_index.search(state.query, k=8) # 去重 + 融合（Reciprocal Rank Fusion） fused = reciprocal_rank_fusion(vector_docs, keyword_docs, k=10) # 重排序提纯 state.retrieved_docs = await reranker.rerank( state.query, [d.content for d in fused], top_n=5 ) return state

路由的价值在于精准匹配策略与问题类型。对于简单事实查询走 factual 策略，一次高质量检索就够，延迟低、成本小。而对于推理型问题走 multi_hop，多花点时间但答案质量显著提升。这是一个"该省省该花花"的调度智慧。

四、查询重写：把"大白话"翻译成"检索语"

用户问"最近有什么大新闻"，你不可能直接把这个句子向量化然后去搜——太模糊了。"大新闻"对向量而言就是 noise。查询重写就是把自然语言转成更适合检索的精确表达。

4.1 三种重写策略

4.2 子问题拆分：最实用的重写策略

from pydantic import BaseModel, Field class SubQueries(BaseModel): queries: list[str] = Field(description="子问题列表，每个应能独立检索") class QueryRewriter: def __init__(self, llm, max_sub_queries: int = 4): self.llm = llm self.max_sub_queries = max_sub_queries async def rewrite(self, query: str, context: list[str] | None = None) -> list[str]: context_str = "\n".join(context) if context else "无" prompt = f"""将以下问题拆分为 {self.max_sub_queries} 个以内可独立检索的子问题。 每个子问题应： 1. 独立完整（不是对上一问的追问） 2. 使用适合检索的关键词表达 3. 覆盖原问题的所有关键信息点 对话上下文：{context_str} 用户问题：{query} 以 JSON 格式返回：{{"queries": ["子问题1", "子问题2", ...]}}""" response = await self.llm.invoke(prompt) result = SubQueries.model_validate_json(response) return result.queries # --- 使用示例 --- rewriter = QueryRewriter(llm) queries = await rewriter.rewrite( "Spring AI 2.0 相比 1.x 有哪些改进？对现有项目升级有什么注意事项？" ) # 输出： # ["Spring AI 2.0 新特性变更列表", # "Spring AI 1.x 到 2.0 迁移指南 破坏性变更", # "Spring AI 2.0 升级注意事项 兼容性"]

为什么子问题拆分效果显著？经验数据表明：将复合问题拆分为 3-4 个子问题分别检索，答案完整度提升 35-50%。原因是向量检索本质上是"找最相似的单一信息点"——一个涵盖多主题的 query embedding，不如多个单一主题的 query embedding 各自精准。

五、迭代检索 + 自检：Agentic 的灵魂

这是 Agentic RAG 区别于传统 RAG 的最核心差异。Agent 在检索后不是直接生成，而是先自检：

1. 这些文档真的回答了用户的问题吗？
2. 缺少哪些关键信息？
3. 换个查询角度是不是能搜到更好的结果？

如果答案不满意，Agent 会重写查询、换个检索策略、甚至调用不同的工具，直到信息充分或达到最大迭代次数。

import asyncio from dataclasses import dataclass, field @dataclass class IterativeAgentState: query: str retrieved_docs: list[str] = field(default_factory=list) iteration: int = 0 sufficiency_score: float = 0.0 missing_info: str = "" class IterativeRAGAgent: def __init__(self, llm, retriever, max_iterations: int = 5, sufficiency_threshold: float = 0.8): self.llm = llm self.retriever = retriever self.max_iterations = max_iterations self.sufficiency_threshold = sufficiency_threshold async def run(self, query: str) -> str: state = IterativeAgentState(query=query) while state.iteration < self.max_iterations: # Step 1: 检索 docs = await self.retriever.search(state.query, k=5) state.retrieved_docs.extend([d.content for d in docs]) # Step 2: 自检 —— 信息够了吗？ sufficiency = await self.self_check(state) state.sufficiency_score = sufficiency["score"] state.missing_info = sufficiency.get("missing", "") if state.sufficiency_score >= self.sufficiency_threshold: break # Step 3: 不够 → 重写查询，换个角度搜 state.query = await self.rewrite_query(state) state.iteration += 1 # Step 4: 最终生成 return await self.generate_final(state) async def self_check(self, state: IterativeAgentState) -> dict: """让 LLM 自评已有信息是否足以回答问题""" prompt = f"""评估当前检索到的信息是否足以回答原始问题。 原始问题：{state.query} 已检索到的文档片段（共 {len(state.retrieved_docs)} 条）： {chr(10).join(f"- {doc[:300]}..." for doc in state.retrieved_docs[-5:])} 请给出： 1. 充分性评分（0.0-1.0）：1.0 = 完全足够，0.0 = 完全不够 2. 如果不够，缺少什么关键信息？ 3. 建议下一轮搜索的关键词 以 JSON 格式返回：{{"score": 0.0-1.0, "missing": "描述缺少的信息", "suggested_query": "建议的搜索词"}}""" response = await self.llm.invoke(prompt) return json.loads(response) async def rewrite_query(self, state: IterativeAgentState) -> str: prompt = f"""已有文档未完全覆盖原始问题，需要换个角度搜索。 原始问题：{state.query} 已有信息但还缺少：{state.missing_info} 请生成一个新的检索查询，重点搜索缺失的信息。 直接返回查询字符串，不要加其他内容。""" return (await self.llm.invoke(prompt)).strip()

⚠ 关键配置：充分性阈值与最大迭代次数是 Agentic RAG 最需要调优的两个参数。阈值太低（0.6）容易过早停止、答案不完整；阈值太高（0.95）会导致无限循环。建议从 0.8 开始，并结合 RAGAS 指标（Faithfulness ≥ 0.9, Answer Relevancy ≥ 0.85）进行系统性评估后再微调。最大迭代次数建议 5，实测中超过 5 轮的边际收益极低。

六、多源融合：向量库不够时要学会"借工具"

纯向量检索只擅长语义相似匹配，面对精确查询（"2026年Q1营收"）、实时数据（"今天的股价"）、结构化数据时就力不从心了。Agentic RAG 的 Agent 应该能自主决定调用哪种工具。

from langgraph.prebuilt import ToolNode # --- 定义多种检索工具 --- async def vector_search_tool(query: str) -> str: """语义搜索：适合概念性、描述性问题""" docs = await vector_store.similarity_search(query, k=5) return format_docs(docs) async def keyword_search_tool(query: str) -> str: """关键词搜索：适合精确术语、型号、API 名称""" docs = bm25_index.search(query, k=5) return format_docs(docs) async def sql_query_tool(query: str) -> str: """SQL查询：适合统计数据、聚合信息""" # Agent 写 SQL → 执行 → 返回结果 sql = await llm.invoke( f"生成SQL查询以下问题（表结构见schema）:\n{query}" ) return await db.execute(sql) async def web_search_tool(query: str) -> str: """网络搜索：适合实时信息、最新动态""" results = await search_api.search(query, num=5) return format_results(results) # --- 将这些工具注册给 Agent --- tools = [ vector_search_tool, keyword_search_tool, sql_query_tool, web_search_tool, ] # LangGraph 中，Agent 自主选择调用哪个工具 agent = create_react_agent(llm, tools) result = await agent.invoke({ "messages": [ HumanMessage(content="对比2025年和2026年Q1的三个主要云服务商市场份额变化") ] })

这个例子中，Agent 会自动推理："我需要市场数据（走 SQL 工具），也需要行业分析（走向量工具），还需要最近动态（走 Web 搜索）"——然后自主编排调用序列并合成答案。

工具选择的"门当户对"原则：
• 概念/原理类查询 → 向量语义搜索
• 精确术语/型号/版本号 → BM25 关键词搜索
• 统计/聚合/对比数据 → SQL 结构化查询
• 实时动态/最新信息 → Web Search API
不要把所有查询都扔进向量库。Agent 的智能体现在"选对合适的工具"。

七、最终生成：从"堆文档"到"结构化输出"

经过多轮检索，Agent 手里有了一堆来自不同来源的文档片段。最后一关是合成——不是简单的拼接，而是：去重、排序、甄别矛盾信息、补全逻辑缺口。

class FinalGenerator: def __init__(self, llm): self.llm = llm async def generate(self, query: str, docs: list[str], max_context_tokens: int = 6000) -> str: # Step 1: 去重 + 重排序（把最相关的放在最前面） unique_docs = deduplicate_by_similarity(docs, threshold=0.92) ranked_docs = await reranker.rerank(query, unique_docs, top_n=10) # Step 2: 动态窗口：优先保留高分文档，超出 token 限制则截断低分文档 context_window = build_context_window(ranked_docs, max_tokens=max_context_tokens) # Step 3: 带引用的最终生成 prompt = f"""基于以下检索到的资料，回答用户问题。 要求： 1. 如果资料中有矛盾信息，指出并说明你的判断依据 2. 对于关键事实，在文末列出引用来源（[1] [2]...） 3. 如果信息不足，明确指出哪些部分无法确认 参考资料： {context_window} 用户问题：{query}""" return await self.llm.invoke(prompt)

✨ 带引用的生成 vs 不带引用：Agentic RAG 的优势不仅在于回答更准，还在于可追溯。每个关键断言都能对应回检索到的文档——这对企业知识库、法律、金融等场景至关重要。单纯"模型说对了"不够，"模型能告诉我它为什么说对了"才是目标。

八、走向生产：Agentic RAG 落地 Checklist

⏱️

延迟控制

Agentic RAG 的延迟是 Naive RAG 的 3-8 倍。给简单查询设置快速路径（路由 Agent 判为 factual 直接单轮生成不上循环），控制最大迭代次数。

成本管理

每次自检都消耗一次 LLM 调用。用轻量模型（如 GPT-4o-mini）做自检和查询重写，只在与用户交互和最终生成时用强模型。

质量评估

RAGAS 四指标：Faithfulness、Answer Relevancy、Context Precision、Context Recall。上 Agentic 后 Context Recall 通常提升 20-35%。

️

安全边界

Agent 能调用 SQL 和 API，必须做好权限控制和请求频率限制。LLM 生成 SQL 时做语法校验，禁止 DDL 操作。

一句话总结 Agentic RAG 的价值：它不是让检索更"准"，而是让系统"知道什么时候不准了然后想办法变准"。这种元认知能力——对自己的知识边界有意识——正是 Agentic RAG 超越传统 RAG 的根本原因。

写在最后

RAG 的演进路线非常清晰：Naive RAG → Advanced RAG（混合检索 + 重排序）→ Agentic RAG（自主推理循环）→ Multi-Agent RAG（多智能体协作）。当前行业的主流正从 Advanced 向 Agentic 过渡。

对开发者来说，关键不在于追逐最新的框架或论文，而在于理解这背后的设计哲学：让检索从"一次性操作"变成"推理过程的一部分"。路由 Agent 帮你选策略，查询重写帮你把问题问清楚，迭代自检帮你确保信息充分——这三个机制组合起来，就能让一个跑 demo 的 RAG pipeline 变成能处理真实复杂问题的工作系统。

技术的进步不在于让模型变得更大，而在于让它变得更聪明地利用已有信息。Agentic RAG 就是这条路上的一块重要拼图。

从一个路由 Agent 开始，给简单问题快速通道、给复杂问题多加两轮思考——不需要一步到位，一步步迭代即可。

#Agentic RAG #检索增强生成 #LangGraph #Query Rewriting #Routing Agent #Self-Check #多源检索 #RAGAS #ReAct #AI Agent