大模型Agentic RAG工作流实战：自主检索增强生成架构设计

摘要

• Agentic RAG是2026年RAG的终极形态：从"检索-生成"到"规划-检索-推理-验证"闭环，准确率提升40%+
• 4大核心能力：自主检索规划、多步推理链、工具调用增强、自我反思验证
• 3种Agentic RAG架构：单Agent路由、多Agent协作、分层Agent编排，各有最佳场景
• 生产关键：检索质量评估、幻觉检测、成本控制，3大指标缺一不可
• 本文提供LangGraph+Agentic RAG完整实现与生产部署方案

---

目录

• Agentic RAG：RAG的终极形态
• 4大核心能力
• 3种Agentic RAG架构
• LangGraph实现Agentic RAG
• 检索质量与幻觉检测
• 生产部署与成本控制
• 总结与引流

---

Agentic RAG：RAG的终极形态

传统RAG vs Agentic RAG

维度	传统RAG	Agentic RAG
检索方式	单次查询	多轮自主检索
推理深度	1步生成	多步推理链
工具使用	仅检索	搜索+计算+API
自我纠错	无	反思+重试
复杂查询	差	强
准确率	60-70%	85-95%

Agentic RAG演进路线

┌──────────────────────────────────────────────────────────────┐
│              RAG演进路线                                       │
│                                                                │
│  RAG 1.0 (2023)                                              │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Query → 检索 → 生成                                  │    │
│  │  简单管道，无反思，无规划                              │    │
│  └──────────────────────────────────────────────────────┘    │
│                         ↓                                     │
│  RAG 2.0 (2024)                                              │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Query → 改写 → 混合检索 → 重排 → 生成               │    │
│  │  查询增强+多路检索+重排序                             │    │
│  └──────────────────────────────────────────────────────┘    │
│                         ↓                                     │
│  Agentic RAG (2025-2026)                                     │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Query → 规划 → 检索 → 推理 → 验证 → (循环) → 生成   │    │
│  │  自主规划+多步推理+自我反思+工具调用                   │    │
│  └──────────────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────────────┘

2026年Agentic RAG框架对比

框架	语言	核心特点	Agent支持	生产就绪
LangGraph	Python	图编排+状态机	强	是
CrewAI	Python	多Agent协作	强	中
AutoGen	Python	多Agent对话	强	中
LlamaIndex	Python	RAG生态最好	中	是
Haystack	Python	管道式编排	中	是
Dify	Python/Go	可视化+低代码	中	是

---

4大核心能力

能力1：自主检索规划

from pydantic import BaseModel
from typing import List, Optional

class RetrievalPlan(BaseModel):
    queries: List[str]
    tools: List[str]
    priority: int
    max_iterations: int

class RetrievalPlanner:
    def __init__(self, llm):
        self.llm = llm
    
    def plan(self, question: str) -> RetrievalPlan:
        prompt = f"""分析以下问题，制定检索计划：

问题：{question}

请输出：
1. 需要检索的子问题列表（按优先级排序）
2. 每个子问题需要的检索工具
3. 最大检索轮数

格式：
- 子问题1 [工具: vector_search]
- 子问题2 [工具: web_search]
- ..."""
        
        response = self.llm.invoke(prompt)
        return self._parse_plan(response)
    
    def _parse_plan(self, response: str) -> RetrievalPlan:
        queries = []
        tools = []
        
        for line in response.strip().split("\n"):
            if "[" in line and "]" in line:
                query = line.split("[")[0].strip("- ").strip()
                tool = line.split("[")[1].split("]")[0].replace("工具:", "").strip()
                queries.append(query)
                tools.append(tool)
        
        return RetrievalPlan(
            queries=queries,
            tools=tools,
            priority=1,
            max_iterations=3,
        )

能力2：多步推理链

class MultiStepReasoner:
    def __init__(self, llm, max_steps=5):
        self.llm = llm
        self.max_steps = max_steps
    
    def reason(self, question: str, context: List[str]) -> dict:
        steps = []
        current_question = question
        accumulated_context = list(context)
        
        for step in range(self.max_steps):
            prompt = f"""基于以下上下文，逐步推理回答问题。

已收集上下文：
{chr(10).join(accumulated_context)}

当前问题：{current_question}

请输出：
1. 基于已有信息的推理步骤
2. 是否需要更多信息（是/否）
3. 如果需要，下一个检索查询是什么
4. 当前推理结论"""
            
            response = self.llm.invoke(prompt)
            step_result = self._parse_step(response)
            steps.append(step_result)
            
            if not step_result["need_more_info"]:
                break
            
            current_question = step_result["next_query"]
            new_context = self._retrieve(current_question)
            accumulated_context.extend(new_context)
        
        return {
            "steps": steps,
            "final_answer": steps[-1]["conclusion"],
            "total_steps": len(steps),
        }
    
    def _parse_step(self, response: str) -> dict:
        lines = response.strip().split("\n")
        return {
            "reasoning": lines[0] if len(lines) > 0 else "",
            "need_more_info": "是" in (lines[1] if len(lines) > 1 else ""),
            "next_query": lines[2] if len(lines) > 2 else "",
            "conclusion": lines[3] if len(lines) > 3 else "",
        }

能力3：工具调用增强

from typing import Callable, Dict, Any

class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, Callable] = {}
        self.tool_descriptions: Dict[str, str] = {}
    
    def register(self, name: str, description: str, func: Callable):
        self.tools[name] = func
        self.tool_descriptions[name] = description
    
    def execute(self, name: str, **kwargs) -> Any:
        if name not in self.tools:
            raise ValueError(f"Tool {name} not found")
        return self.tools[name](**kwargs)
    
    def get_descriptions(self) -> str:
        return "\n".join([
            f"- {name}: {desc}" 
            for name, desc in self.tool_descriptions.items()
        ])

def setup_tools(registry: ToolRegistry):
    registry.register(
        "vector_search",
        "在知识库中搜索相关文档",
        lambda query, top_k=5: vector_search_impl(query, top_k),
    )
    registry.register(
        "web_search",
        "在互联网上搜索最新信息",
        lambda query, num_results=5: web_search_impl(query, num_results),
    )
    registry.register(
        "calculator",
        "执行数学计算",
        lambda expression: eval(expression),
    )
    registry.register(
        "sql_query",
        "查询数据库",
        lambda sql: sql_query_impl(sql),
    )
    registry.register(
        "api_call",
        "调用外部API",
        lambda url, params: api_call_impl(url, params),
    )

能力4：自我反思验证

class SelfReflector:
    def __init__(self, llm):
        self.llm = llm
    
    def verify(self, question: str, answer: str, context: List[str]) -> dict:
        prompt = f"""验证以下回答的正确性：

问题：{question}
回答：{answer}
参考上下文：
{chr(10).join(context)}

请评估：
1. 回答是否完全基于上下文？(grounded: 是/否)
2. 回答是否完整回答了问题？(complete: 是/否)
3. 回答中是否有幻觉内容？(hallucination: 是/否)
4. 置信度评分 (0-1)
5. 改进建议"""
        
        response = self.llm.invoke(prompt)
        result = self._parse_verification(response)
        
        return result
    
    def _parse_verification(self, response: str) -> dict:
        lines = response.strip().split("\n")
        grounded = "是" in (lines[0] if len(lines) > 0 else "")
        complete = "是" in (lines[1] if len(lines) > 1 else "")
        hallucination = "是" in (lines[2] if len(lines) > 2 else "")
        
        return {
            "grounded": grounded,
            "complete": complete,
            "has_hallucination": hallucination,
            "confidence": 0.8,
            "needs_retry": not grounded or not complete or hallucination,
        }

---

3种Agentic RAG架构

架构对比

┌──────────────────────────────────────────────────────────────┐
│              3种Agentic RAG架构                                │
│                                                                │
│  架构1: 单Agent路由                                          │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Query → Router Agent → [检索|计算|API] → 生成       │    │
│  │  适合: 简单场景，快速部署                              │    │
│  └──────────────────────────────────────────────────────┘    │
│                                                                │
│  架构2: 多Agent协作                                          │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Query → Planner → [Retriever|Reasoner|Verifier]     │    │
│  │                ← 协作循环 →                           │    │
│  │  适合: 复杂查询，高准确率                              │    │
│  └──────────────────────────────────────────────────────┘    │
│                                                                │
│  架构3: 分层Agent编排                                        │
│  ┌──────────────────────────────────────────────────────┐    │
│  │  Orchestrator → [Sub-Orchestrator → [Workers]]       │    │
│  │  适合: 企业级，多租户                                  │    │
│  └──────────────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────────────┘

架构选型

场景	推荐架构	复杂度	准确率	延迟
FAQ/简单查询	单Agent路由	低	80%	2s
研究分析	多Agent协作	中	92%	8s
企业知识库	分层Agent	高	95%	5s
实时决策	单Agent+缓存	低	85%	1s

---

LangGraph实现Agentic RAG

完整实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Any, Annotated
import operator

class AgentState(TypedDict):
    question: str
    plan: Dict[str, Any]
    documents: List[str]
    reasoning_steps: List[Dict]
    answer: str
    verification: Dict[str, Any]
    iteration: int
    max_iterations: int

class AgenticRAGGraph:
    def __init__(self, llm, retriever, reflector):
        self.llm = llm
        self.retriever = retriever
        self.reflector = reflector
        self.graph = self._build_graph()
    
    def _build_graph(self):
        workflow = StateGraph(AgentState)
        
        workflow.add_node("plan", self._plan_node)
        workflow.add_node("retrieve", self._retrieve_node)
        workflow.add_node("reason", self._reason_node)
        workflow.add_node("generate", self._generate_node)
        workflow.add_node("verify", self._verify_node)
        
        workflow.set_entry_point("plan")
        workflow.add_edge("plan", "retrieve")
        workflow.add_edge("retrieve", "reason")
        workflow.add_edge("reason", "generate")
        workflow.add_edge("generate", "verify")
        
        workflow.add_conditional_edges(
            "verify",
            self._should_retry,
            {
                "retry": "retrieve",
                "finish": END,
            },
        )
        
        return workflow.compile()
    
    def _plan_node(self, state: AgentState) -> AgentState:
        planner = RetrievalPlanner(self.llm)
        plan = planner.plan(state["question"])
        state["plan"] = plan.dict()
        state["iteration"] = state.get("iteration", 0) + 1
        return state
    
    def _retrieve_node(self, state: AgentState) -> AgentState:
        documents = []
        plan = state["plan"]
        
        for query in plan.get("queries", [state["question"]]):
            docs = self.retriever.search(query, top_k=5)
            documents.extend([d.page_content for d in docs])
        
        state["documents"] = list(set(documents))
        return state
    
    def _reason_node(self, state: AgentState) -> AgentState:
        reasoner = MultiStepReasoner(self.llm)
        result = reasoner.reason(state["question"], state["documents"])
        state["reasoning_steps"] = result["steps"]
        return state
    
    def _generate_node(self, state: AgentState) -> AgentState:
        context = "\n".join(state["documents"])
        reasoning = "\n".join([
            f"步骤{i+1}: {s['conclusion']}" 
            for i, s in enumerate(state["reasoning_steps"])
        ])
        
        prompt = f"""基于以下信息回答问题。

上下文：
{context}

推理过程：
{reasoning}

问题：{state["question"]}

请给出完整、准确的回答："""
        
        state["answer"] = self.llm.invoke(prompt)
        return state
    
    def _verify_node(self, state: AgentState) -> AgentState:
        result = self.reflector.verify(
            state["question"], state["answer"], state["documents"]
        )
        state["verification"] = result
        return state
    
    def _should_retry(self, state: AgentState) -> str:
        if (
            state["verification"]["needs_retry"]
            and state["iteration"] < state.get("max_iterations", 3)
        ):
            return "retry"
        return "finish"
    
    def run(self, question: str, max_iterations: int = 3) -> dict:
        initial_state = {
            "question": question,
            "plan": {},
            "documents": [],
            "reasoning_steps": [],
            "answer": "",
            "verification": {},
            "iteration": 0,
            "max_iterations": max_iterations,
        }
        
        result = self.graph.invoke(initial_state)
        return result

---

检索质量与幻觉检测

检索质量评估

指标	计算方式	目标值
召回率(Recall)	相关文档/总相关	>90%
精确率(Precision)	相关文档/检索文档	>80%
MRR	正确答案排名倒数均值	>0.7
NDCG@10	归一化折损累积增益	>0.8

幻觉检测方法

class HallucinationDetector:
    def __init__(self, llm, embedder):
        self.llm = llm
        self.embedder = embedder
    
    def detect(self, answer: str, context: List[str]) -> dict:
        claim_score = self._claim_verification(answer, context)
        consistency_score = self._self_consistency(answer)
        similarity_score = self._context_similarity(answer, context)
        
        overall = (
            0.4 * claim_score
            + 0.3 * consistency_score
            + 0.3 * similarity_score
        )
        
        return {
            "hallucination_risk": 1 - overall,
            "claim_verification": claim_score,
            "self_consistency": consistency_score,
            "context_similarity": similarity_score,
            "is_reliable": overall > 0.7,
        }
    
    def _claim_verification(self, answer: str, context: List[str]) -> float:
        prompt = f"""逐句验证回答中的每个论断是否有上下文支持。

上下文：{chr(10).join(context)}
回答：{answer}

对每个论断标注：支持/不支持/无法判断"""
        
        response = self.llm.invoke(prompt)
        supported = response.count("支持")
        total = supported + response.count("不支持") + response.count("无法判断")
        return supported / max(total, 1)
    
    def _self_consistency(self, answer: str) -> float:
        variations = []
        for _ in range(3):
            var = self.llm.invoke(f"用不同方式复述：{answer}")
            variations.append(var)
        
        embeddings = self.embedder.embed(variations + [answer])
        similarities = [
            cosine_similarity(embeddings[-1], emb) for emb in embeddings[:-1]
        ]
        return sum(similarities) / len(similarities)
    
    def _context_similarity(self, answer: str, context: List[str]) -> float:
        answer_emb = self.embedder.embed([answer])[0]
        context_emb = self.embedder.embed(context)
        max_sim = max(
            cosine_similarity(answer_emb, emb) for emb in context_emb
        )
        return max_sim

---

生产部署与成本控制

成本优化策略

策略	成本节省	准确率影响
检索结果缓存	40-60%	无
小模型路由	30-50%	<2%
批量推理	20-30%	无
上下文压缩	15-25%	<1%
动态迭代控制	10-20%	<3%

动态迭代控制

class DynamicIterationController:
    def __init__(self, max_iterations=3, confidence_threshold=0.8):
        self.max_iterations = max_iterations
        self.confidence_threshold = confidence_threshold
    
    def should_continue(self, state: dict) -> bool:
        if state["iteration"] >= self.max_iterations:
            return False
        
        if state.get("verification", {}).get("confidence", 0) >= self.confidence_threshold:
            return False
        
        if state["iteration"] > 1:
            prev_conf = state.get("prev_confidence", 0)
            curr_conf = state.get("verification", {}).get("confidence", 0)
            if curr_conf - prev_conf < 0.05:
                return False
        
        return True

Agentic RAG性能基准

指标	传统RAG	Agentic RAG	提升
简单查询准确率	85%	92%	+7%
复杂查询准确率	55%	88%	+33%
幻觉率	15%	5%	-67%
平均延迟	2s	5s	-60%
平均token消耗	500	2000	-75%

---

总结与引流

关键要点回顾

1. Agentic RAG是RAG的终极形态：规划-检索-推理-验证闭环，准确率提升40%+
2. 4大核心能力：自主规划、多步推理、工具调用、自我反思
3. 3种架构：单Agent路由(简单)、多Agent协作(复杂)、分层Agent(企业)
4. 生产关键：检索质量评估+幻觉检测+成本控制

Agentic RAG方案推荐

场景	推荐方案	预期效果
快速验证	单Agent+LangGraph	准确率85%+
生产部署	多Agent+缓存+量化	准确率90%+
企业级	分层Agent+监控	准确率95%+

需要处理RAG数据的格式转换？试试我们的JSON转YAML工具和文本差异对比，快速处理检索数据。

延伸阅读

• MCP协议实战：构建AI Agent工具链 — Agent工具调用
• AI Agent多轮记忆实战 — Agent记忆管理
• GraphRAG实战：知识图谱增强RAG — 知识图谱RAG
• LangGraph官方文档 — LangGraph使用指南
• Agentic RAG论文 — Agentic RAG方法论