从 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是多少？”
...
