Agentic RAG

引言 检索增强生成（RAG）是2026年最实用的大模型应用架构之一。从基础的"检索-生成"流水线，到Agentic RAG的智能检索代理，RAG技术正在经历深刻的架构演进。本文系统梳理RAG的发展脉络和技术选型。 RAG 的演进阶段 第一阶段：基础 RAG（Naive RAG） 最原始的RAG架构，包含三个步骤： 将文档切分为片段 对片段进行向量化并存储 检索最相关的片段，拼接后输入模型 优点： 简单、快速、易于实现 缺点： 检索质量依赖单一向量相似度，无法处理复杂查询 第二阶段：增强 RAG（Enhanced RAG） 在基础RAG之上增加优化： 查询优化： 查询重写：将用户问题转化为更适合检索的形式 查询分解：将复杂问题拆分为多个子查询 多路检索：同时使用向量检索和关键词检索 后处理： 重排序（Reranking）：对检索结果进行精排 上下文压缩：去除冗余片段，保留关键信息 自适应窗口：根据问题类型调整上下文长度 第三阶段：Agentic RAG Agentic RAG的核心思想：让Agent自主决定如何检索、何时检索、检索什么。 ┌─────────────────────────────────────────────┐ │ Agentic RAG 架构 │ │ │ │ ┌──────────┐ ┌──────────┐ │ │ │ Agent │───▶│ 检索 │ │ │ │ 规划器 │ │ 策略 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ │ │ │ 反思 │◀───│ 多路 │ │ │ │ 与调整 │ │ 检索 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────┐ │ │ │ 最终生成与验证 │ │ │ └──────────────────────────┘ │ └─────────────────────────────────────────────┘ 主流 RAG 框架对比 LangChain / LangGraph 特点： 生态最丰富，组件最全 适合： 快速原型开发、复杂RAG管线 ...

从Naive RAG到Agentic RAG的演进 RAG技术的发展经历了三个阶段： Naive RAG (2023) → Advanced RAG (2024) → Agentic RAG (2025-2026) 朴素检索 增强检索 智能体检索 单次查询 多次优化 自主决策 固定管道 模块化 动态规划 传统RAG是一个"固定管道"：查询→检索→拼接→生成，一锤子买卖。Agentic RAG将LLM作为"大脑"，让它自主决定是否需要检索、检索什么、检索几次、何时停止。这就像从"查字典"升级为"咨询一个会查资料的研究员"。 Agentic RAG的核心架构 ┌─────────────────────────────────────────────────────┐ │ Agentic RAG Architecture │ ├─────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────────────────────────┐ │ │ │ User │───▶│ Agent Controller │ │ │ │ Query │ │ (LLM + Planning + Memory) │ │ │ └──────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌─────────┼─────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌───────┐ ┌─────────┐ │ │ │ Router │ │ Query │ │ Evaluator│ │ │ │ (路由) │ │Rewrite│ │ (评估器) │ │ │ └────┬────┘ └───┬───┘ └────┬────┘ │ │ │ │ │ │ │ ┌─────────┼──────────┼──────────┘ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌────────┐ ┌──────────┐ ┌─────────┐ │ │ │ Vector │ │ Web │ │ SQL │ │ Graph │ │ │ │ Store │ │ Search │ │ DB │ │ Store │ │ │ └──────────┘ └────────┘ └──────────┘ └─────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 四大核心组件 组件 职责 关键能力 Agent Controller 任务规划与决策 分解问题、决定工具调用顺序 Router 工具路由 选择合适的数据源（向量库/搜索引擎/数据库） Query Rewriter 查询改写 将用户问题改写为更有效的检索查询 Evaluator 结果评估 判断检索结果是否充分，决定是否需要追加检索 核心模式解析 模式1：迭代检索（Iterative Retrieval） from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): question: str retrieved_docs: Annotated[List[str], operator.add] answer: str iteration: int sufficient: bool def retrieve_node(state: AgentState) -> dict: """基于当前问题+已有信息生成查询并检索""" query = query_rewriter(state["question"], state["retrieved_docs"]) docs = vector_store.similarity_search(query, k=5) return {"retrieved_docs": docs, "iteration": state["iteration"] + 1} def evaluate_node(state: AgentState) -> dict: """评估检索结果是否足以回答问题""" if state["iteration"] >= 3: return {"sufficient": True} prompt = f"""基于以下检索到的文档，能否完整回答用户问题？ 问题: {state['question']} 文档: {state['retrieved_docs']} 如果信息充分回答"SUFFICIENT"，否则回答"INSUFFICIENT"并说明缺什么。""" response = llm.invoke(prompt) sufficient = "SUFFICIENT" in response return {"sufficient": sufficient} def answer_node(state: AgentState) -> dict: """基于所有检索文档生成最终回答""" prompt = f"问题: {state['question']}\n\n参考资料:\n{state['retrieved_docs']}" return {"answer": llm.invoke(prompt)} # 构建工作流 workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve_node) workflow.add_node("evaluate", evaluate_node) workflow.add_node("answer", answer_node) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "evaluate") workflow.add_conditional_edges( "evaluate", lambda state: "answer" if state["sufficient"] else "retrieve" ) workflow.add_edge("answer", END) agent = workflow.compile() 模式2：多路检索（Multi-Route Retrieval） def route_query(state: AgentState) -> str: """根据问题类型选择检索路径""" routing_prompt = f""" 分析以下问题，选择最合适的数据源： 问题: {state['question']} 选项: - "vector": 适合从内部文档库检索事实性信息 - "web": 适合需要最新信息的时效性问题 - "sql": 适合结构化数据查询 - "graph": 适合关系推理和多方关联 返回JSON: {{"route": "...", "reason": "..."}} """ decision = llm.invoke(routing_prompt) return json.loads(decision)["route"] 模式3：自我纠正检索（Self-Corrective RAG） def grade_documents(state: AgentState) -> dict: """对检索文档打分，过滤低质量结果""" graded_docs = [] for doc in state["retrieved_docs"]: score_prompt = f"""评估文档与问题的相关性(0-10): 问题: {state['question']} 文档: {doc.page_content[:500]} 只返回数字。""" score = int(llm.invoke(score_prompt).strip()) if score >= 6: graded_docs.append(doc) # 如果所有文档都不相关，触发查询改写 if not graded_docs: new_query = llm.invoke(f"改写这个查询以获得更好的检索结果: {state['question']}") return {"retrieved_docs": [], "question": new_query} return {"retrieved_docs": graded_docs} 工程实现：基于LangGraph的完整Agentic RAG from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 定义工具 @tool def vector_search(query: str) -> str: """从内部知识库检索相关文档""" docs = vector_store.similarity_search(query, k=5) return "\n\n".join([d.page_content for d in docs]) @tool def web_search(query: str) -> str: """从网络搜索最新信息""" results = search_api.search(query, num_results=5) return "\n\n".join([r["content"] for r in results]) @tool def sql_query(question: str) -> str: """查询业务数据库""" sql = text2sql(question) return db.execute(sql) # Agent决策循环 tools = [vector_search, web_search, sql_query] llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_decision(state): messages = state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response]} def should_continue(state): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END # 构建Graph graph = StateGraph(dict) graph.add_node("agent", agent_decision) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue) graph.add_edge("tools", "agent") # 工具执行后回到agent做下一轮决策 app = graph.compile() # 执行 result = app.invoke({ "messages": [HumanMessage(content="对比2026年Q1三大云厂商的AI服务收入")] }) 性能基准对比 我们在三个数据集上对比了不同RAG架构的表现： ...

从 RAG 到 Agentic RAG 的范式转移 传统 RAG 是一个线性的流水线：检索→拼接→生成。它被动地执行预定义的步骤，没有自主决策能力。而 Agentic RAG 将 LLM Agent 作为核心控制器，赋予 RAG 系统"思考"和"行动"的能力。 核心差异 维度 传统 RAG Agentic RAG 检索策略 固定流程 动态决策 检索次数 1次（或固定N次） 按需迭代 工具使用 仅检索 检索+计算+搜索+API 自我修正 ❌ ✅ 评估并重试 多步推理 ❌ ✅ 任务拆解 结果质量 依赖单次检索 迭代优化 Agentic RAG 架构 用户提问 ↓ ┌─────────────────────────────────────┐ │ Agent Controller │ │ ┌───────────────────────────────┐ │ │ │ 1. 意图理解 & 任务规划 │ │ │ │ 2. 工具选择 & 执行 │ │ │ │ 3. 结果评估 & 决策 │ │ │ │ 4. 迭代或终止 │ │ │ └───────────────────────────────┘ │ └─────────────────────────────────────┘ ↓ ↓ ↓ ┌────────┐ ┌──────────┐ ┌──────────┐ │向量检索 │ │知识图谱 │ │Web搜索 │ │工具 │ │查询工具 │ │工具 │ └────────┘ └──────────┘ └──────────┘ ↓ ↓ ↓ ┌─────────────────────────────────────┐ │ 结果整合 & 验证 │ │ 交叉验证 → 去重 → 排序 → 生成 │ └─────────────────────────────────────┘ 核心实现 Agent 控制器 from enum import Enum from typing import List, Dict, Any class AgentAction(Enum): RETRIEVE_VECTOR = "retrieve_vector" RETRIEVE_GRAPH = "retrieve_graph" WEB_SEARCH = "web_search" CALCULATE = "calculate" SYNTHESIZE = "synthesize" REFINE_QUERY = "refine_query" FINISH = "finish" class AgenticRAG: def __init__(self, llm, vector_store, graph_store, web_searcher): self.llm = llm self.tools = { AgentAction.RETRIEVE_VECTOR: self._retrieve_vector, AgentAction.RETRIEVE_GRAPH: self._retrieve_graph, AgentAction.WEB_SEARCH: self._web_search, AgentAction.CALCULATE: self._calculate, } self.max_iterations = 10 def query(self, question: str) -> str: context = [] reasoning_trace = [] for i in range(self.max_iterations): # Agent 决策：下一步做什么？ decision = self._decide_action(question, context, reasoning_trace) reasoning_trace.append({ "iteration": i, "thought": decision.thought, "action": decision.action.value, "action_input": decision.action_input }) if decision.action == AgentAction.FINISH: # 生成最终答案 return self._generate_answer(question, context, reasoning_trace) if decision.action == AgentAction.REFINE_QUERY: question = decision.action_input["refined_query"] continue # 执行工具 tool_result = self.tools[decision.action](**decision.action_input) context.append(tool_result) # 评估结果质量 evaluation = self._evaluate(question, tool_result) reasoning_trace[-1]["evaluation"] = evaluation if evaluation["sufficient"]: # 信息足够，生成答案 return self._generate_answer(question, context, reasoning_trace) return self._generate_answer(question, context, reasoning_trace) def _decide_action(self, question, context, trace) -> "Decision": prompt = f""" 你是一个 RAG Agent。根据当前状态决定下一步行动。 问题：{question} 已收集的上下文： {self._format_context(context)} 推理历史： {self._format_trace(trace)} 可选行动： 1. retrieve_vector - 向量检索（语义相关信息） 2. retrieve_graph - 图谱查询（实体关系） 3. web_search - 网络搜索（实时信息） 4. calculate - 计算（数值处理） 5. refine_query - 优化查询 6. finish - 信息足够，生成答案 请输出 JSON： {{"thought": "思考过程", "action": "行动名称", "action_input": {{...}}}} """ result = self.llm.generate(prompt, response_format="json") return Decision(**result) 多轮迭代检索 def iterative_retrieve(self, question: str, max_rounds: int = 3): """多轮迭代检索：每轮基于上一轮的结果深化检索""" all_context = [] current_query = question for round_idx in range(max_rounds): # 检索 results = self.vector_store.search( embed_model.encode(current_query), top_k=10 ) # 评估检索结果 relevance = self._assess_relevance(question, results) if relevance["score"] > 0.8: all_context.extend(results) break # 生成 follow-up query current_query = self._generate_followup_query( question, results, all_context ) all_context.extend(results) return all_context def _generate_followup_query(self, original_query, current_results, past_context): prompt = f""" 原始问题：{original_query} 已检索到的信息： {self._format_results(current_results)} 已有上下文： {self._format_results(past_context)} 信息缺口分析：还有哪些信息需要检索？ 请生成一个更精确的 follow-up 查询。 """ return self.llm.generate(prompt).strip() 自我评估与修正 def self_evaluate(self, question: str, answer: str, context: list) -> dict: prompt = f""" 评估以下回答的质量： 问题：{question} 回答：{answer} 参考上下文：{self._format_context(context)} 请从以下维度评估（0-1分）： 1. 准确性：回答是否与上下文一致？ 2. 完整性：是否回答了问题的所有方面？ 3. 引用性：关键论断是否有引用支撑？ 4. 幻觉率：回答中是否有上下文不支持的内容？ 输出 JSON：{{"accuracy": 0.0, "completeness": 0.0, "citation": 0.0, "hallucination": 0.0, "needs_retry": false, "reason": "..."}} """ return self.llm.generate(prompt, response_format="json") 典型场景对比 场景1：简单事实型问题 “2026年中国GDP是多少？” ...

从传统 RAG 到 Agentic RAG：一次范式跃迁 检索增强生成（Retrieval-Augmented Generation, RAG）在过去两年里几乎成了大模型应用的标配方案。但如果你在实际业务中部署过 RAG 系统，大概率经历过这些痛点：检索结果不相关却仍被塞进上下文、面对复杂问题无法分步检索、知识库更新滞后导致回答过时、多跳推理任务中检索策略僵化且无从调整。这些问题的根源在于传统 RAG 本质上是一个线性的、一次性的管道——查询进来，检索一轮，拼上下文，生成回答，结束。 Agentic RAG 的核心思路是把这个线性管道升级为一个闭环的、自适应的智能体系统。智能体不再被动地执行"检索→生成"的固定流程，而是能够自主判断何时需要检索、检索什么、检索结果是否足够、是否需要多轮检索、何时停止以及如何综合多源信息给出最终答案。这就像是把一个只会按食谱做菜的厨师，升级成了一个能根据食材、客人喜好和现场情况即兴创作的大厨。 Agentic RAG 的核心架构组件 1. 智能体控制器（Agent Controller） 整个系统的大脑。它负责理解用户意图，制定检索策略，决定下一步行动。与传统 RAG 中固定的流程不同，控制器是一个基于 LLM 的决策引擎，能够在"检索"、“推理”、“工具调用”、“回答"之间动态切换。 控制器的工作循环可以这样描述： 用户查询 → 意图分析 → 策略制定 → 行动执行 → 结果评估 → ├─ 信息充分 → 生成回答 ├─ 信息不足 → 重新检索（调整查询） └─ 需要工具 → 调用外部工具 这个循环本质上是一个 ReAct（Reasoning + Acting）模式的变体，但加入了专门的检索策略模块。 2. 多级检索引擎（Multi-Stage Retrieval Engine） 传统 RAG 通常只有一层向量检索。Agentic RAG 则部署了多级检索策略： 语义检索层：基于向量数据库的稠密检索，处理模糊语义匹配 关键词检索层：基于 BM25 等算法的稀疏检索，处理精确术语匹配 结构化检索层：知识图谱查询，处理实体关系推理 实时检索层：联网搜索 API，处理最新信息获取 智能体控制器会根据查询类型动态选择检索策略。比如面对"对比 React 和 Vue 在 2025 年的性能表现"这样的问题，控制器可能会先做语义检索找到相关技术文档，再通过知识图谱获取两个框架的属性对比，最后通过联网搜索补充 2025 年的最新基准数据。 ...