Agentic RAG实战：构建自主推理检索Agent的5大核心模式
AI与大数据
传统RAG已经不够用了
你精心搭建的RAG系统，用户问了一句"RAG和微调哪个更适合我的场景？"，结果返回了一堆关于RAG的基础介绍和微调的定义——完全没回答问题。这不是检索不准，而是传统RAG根本不具备推理能力。它只能做"问什么搜什么"，遇到需要多步推理、跨知识源综合、自我纠正的复杂问题，就只能原地打转。
2026年，Agentic RAG成为破局关键。它不再是被动的"检索-生成"管道，而是让Agent自主决定：要不要检索？去哪检索？检索结果够不够？要不要换个策略重新搜？这种"自主决策+循环迭代"的范式，才是复杂问题的正确打开方式。
核心概念速查
| 概念 | 说明 | 典型实现 |
|---|---|---|
| Agentic RAG | Agent自主决策检索策略的RAG范式 | LangGraph StateGraph、LlamaIndex Agent |
| 自主检索 | Agent根据问题自主决定是否检索、检索什么 | ReAct循环、工具调用 |
| 多跳推理 | 将复杂问题分解为多个子问题逐步检索 | 问题分解、链式检索 |
| 自我反思 | Agent评估检索结果质量，决定是否重新搜索 | 文档评分、查询重写 |
| 工具调用 | Agent调用外部工具（搜索、计算、数据库）增强能力 | Function Calling、Tool Use |
| LangGraph | 构建有状态Agent工作流的框架 | StateGraph、条件边、检查点 |
| ReAct模式 | 推理（Reasoning）+行动（Acting）交替进行 | Thought-Action-Observation循环 |
问题分析：传统RAG的5大局限
| # | 局限 | 具体表现 | 影响 |
|---|---|---|---|
| 1 | 单次检索 | 只能做一次向量搜索，无法迭代优化 | 检索不全面，遗漏关键信息 |
| 2 | 无推理能力 | 无法分解复杂问题为子问题 | 多跳问题回答质量差 |
| 3 | 无法自我纠正 | 检索到无关文档也无法识别和重试 | 幻觉严重，答非所问 |
| 4 | 缺乏规划 | 没有检索策略，无法决定先搜什么后搜什么 | 效率低下，资源浪费 |
| 5 | 无法处理多跳问题 | "A和B哪个更适合C场景？"类问题完全无解 | 用户体验极差，弃用率高 |
传统RAG的本质问题：它是一个无状态的管道，而不是一个有智能的Agent。Agentic RAG的核心转变，就是从"管道"到"Agent"——让检索过程具备自主决策、循环迭代和自我纠错的能力。
模式1：路由Agent — 智能分发
最简单的Agentic RAG模式：根据问题类型，将查询路由到不同的知识库或检索策略。比统一向量搜索精准得多。
from enum import Enum
from typing import Optional
class QueryType(Enum):
TECHNICAL = "technical"
BUSINESS = "business"
COMPARISON = "comparison"
TUTORIAL = "tutorial"
class RouterAgent:
def __init__(self, llm, retrievers: dict[str, object]):
self.llm = llm
self.retrievers = retrievers
def classify(self, question: str) -> QueryType:
prompt = f"""Classify this question into one of:
- technical: code/API/implementation questions
- business: pricing/ROI/strategy questions
- comparison: comparing two or more options
- tutorial: how-to/step-by-step guides
Question: {question}
Type:"""
result = self.llm.invoke(prompt).strip().lower()
return QueryType(result)
def route(self, question: str) -> list[dict]:
query_type = self.classify(question)
retriever = self.retrievers.get(query_type.value)
if not retriever:
retriever = self.retrievers["technical"]
docs = retriever.similarity_search(question, k=5)
return [{"content": d.page_content, "source": d.metadata.get("source", "")} for d in docs]
retrievers = {
"technical": tech_vector_store,
"business": biz_vector_store,
"comparison": compare_vector_store,
"tutorial": tutorial_vector_store,
}
router = RouterAgent(llm, retrievers)
result = router.route("TiDB和CockroachDB在分布式事务方面有什么区别？")
路由Agent适合知识库分类明确的场景，实现简单但效果立竿见影。
模式2：多跳推理Agent
复杂问题需要分解。比如"比较TiDB和CockroachDB在分布式事务方面的差异，给出性能基准数据"，需要先分别检索两个数据库的信息，再检索性能基准，最后综合回答。
from langchain.agents import create_react_agent
tools = [
search_knowledge_base,
search_web,
calculate_metrics,
query_database,
]
agent = create_react_agent(llm, tools, prompt)
result = agent.invoke({
"input": "比较TiDB和CockroachDB在分布式事务方面的差异，给出性能基准数据"
})
更细粒度的多跳实现：
from dataclasses import dataclass
from typing import Optional
@dataclass
class SubQuestion:
question: str
answer: Optional[str] = None
sources: list[str] = None
class MultiHopAgent:
def __init__(self, llm, retriever):
self.llm = llm
self.retriever = retriever
def decompose(self, question: str) -> list[SubQuestion]:
prompt = f"""Break down this complex question into 2-4 sub-questions.
Each sub-question should be independently answerable.
Question: {question}
Sub-questions (one per line):"""
result = self.llm.invoke(prompt)
lines = [l.strip() for l in result.strip().split("\n") if l.strip()]
return [SubQuestion(question=line) for line in lines]
def answer_sub_question(self, sub_q: SubQuestion) -> SubQuestion:
docs = self.retriever.similarity_search(sub_q.question, k=3)
context = "\n".join([d.page_content for d in docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {sub_q.question}"
)
sub_q.answer = answer
sub_q.sources = [d.metadata.get("source", "") for d in docs]
return sub_q
def synthesize(self, question: str, sub_answers: list[SubQuestion]) -> str:
evidence = "\n".join([f"- {sq.question}: {sq.answer}" for sq in sub_answers])
return self.llm.invoke(
f"Based on the following evidence, answer the original question.\n\n"
f"Original: {question}\n\nEvidence:\n{evidence}\n\nAnswer:"
)
def run(self, question: str) -> dict:
sub_questions = self.decompose(question)
answered = [self.answer_sub_question(sq) for sq in sub_questions]
final = self.synthesize(question, answered)
return {
"answer": final,
"sub_questions": [
{"q": sq.question, "a": sq.answer} for sq in answered
],
}
agent = MultiHopAgent(llm, vector_store)
result = agent.run("RAG和微调哪个更适合我的场景？请从成本、效果、适用数据量三个维度分析")
模式3：自我反思搜索Agent
Agent检索到文档后，自己评估质量——如果不够好，就重写查询再搜。这是Agentic RAG最核心的能力：自我纠错。
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ReflectionState:
question: str
documents: list = field(default_factory=list)
filtered_docs: list = field(default_factory=list)
answer: str = ""
iterations: int = 0
max_iterations: int = 3
need_more_search: bool = False
class SelfReflectiveAgent:
def __init__(self, llm, retriever, min_relevant: int = 2):
self.llm = llm
self.retriever = retriever
self.min_relevant = min_relevant
def retrieve(self, state: ReflectionState) -> dict:
docs = self.retriever.similarity_search(state.question, k=5)
return {"documents": docs}
def grade_documents(self, state: ReflectionState) -> dict:
filtered = []
for doc in state.documents:
score = self.llm.invoke(
f"Is this document relevant to '{state.question}'? "
f"Answer yes or no:\n{doc.page_content[:500]}"
)
if "yes" in score.lower():
filtered.append(doc)
return {
"filtered_docs": filtered,
"need_more_search": len(filtered) < self.min_relevant,
}
def rewrite_query(self, state: ReflectionState) -> dict:
better = self.llm.invoke(
f"Rewrite this question for better search results: {state.question}"
)
return {"question": better, "iterations": state.iterations + 1}
def generate(self, state: ReflectionState) -> dict:
context = "\n".join([d.page_content for d in state.filtered_docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {state.question}"
)
return {"answer": answer}
def run(self, question: str) -> dict:
state = ReflectionState(question=question)
self.retrieve(state)
while state.iterations < state.max_iterations:
self.grade_documents(state)
if not state.need_more_search:
break
self.rewrite_query(state)
self.retrieve(state)
self.generate(state)
return {"answer": state.answer, "iterations": state.iterations}
agent = SelfReflectiveAgent(llm, vector_store)
result = agent.run("如何在生产环境中部署Agentic RAG？")
模式4：工具增强Agent
Agent不只检索文档，还能调用搜索引擎、计算器、数据库等外部工具。这极大扩展了Agent的能力边界。
from typing import Annotated
def search_web(query: str) -> str:
"""Search the web for up-to-date information."""
results = web_search_engine.search(query, max_results=3)
return "\n".join([r.snippet for r in results])
def query_database(sql: str) -> str:
"""Execute SQL query against the product database."""
rows = db.execute(sql)
return str(rows[:20])
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: invalid expression"
return str(eval(expression))
def search_knowledge_base(query: str) -> str:
"""Search internal knowledge base for documentation."""
docs = vector_store.similarity_search(query, k=3)
return "\n".join([d.page_content for d in docs])
tools = [search_web, query_database, calculate, search_knowledge_base]
tool_descriptions = "\n".join(
[f"- {t.__name__}: {t.__doc__}" for t in tools]
)
class ToolAugmentedAgent:
def __init__(self, llm, tools: list):
self.llm = llm
self.tools = {t.__name__: t for t in tools}
def run(self, question: str, max_steps: int = 5) -> dict:
messages = [{"role": "user", "content": question}]
steps = []
for _ in range(max_steps):
response = self.llm.invoke(
f"Available tools:\n{tool_descriptions}\n\n"
f"To use a tool, respond: USE: tool_name(arg)\n"
f"To give final answer, respond: ANSWER: your answer\n\n"
f"Question: {messages[-1]['content']}"
)
if response.startswith("ANSWER:"):
return {"answer": response[7:].strip(), "steps": steps}
if response.startswith("USE:"):
tool_call = response[4:].strip()
tool_name = tool_call.split("(")[0]
tool_arg = tool_call.split("(")[1].rstrip(")")
tool_fn = self.tools.get(tool_name)
if tool_fn:
observation = tool_fn(tool_arg)
steps.append({
"tool": tool_name,
"arg": tool_arg,
"result": observation[:200],
})
messages.append({
"role": "assistant",
"content": f"Used {tool_name}, got: {observation}",
})
return {"answer": "Max steps reached", "steps": steps}
agent = ToolAugmentedAgent(llm, tools)
result = agent.run("我们产品的月活用户是多少？和竞品相比如何？")
模式5：LangGraph工作流编排
最完整的Agentic RAG实现——用LangGraph的StateGraph定义完整的检索-评估-重写-生成循环，支持条件分支和检查点。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
class AgentState(TypedDict):
question: str
documents: Annotated[list, operator.add]
answer: str
iterations: int
need_more_search: bool
def retrieve(state: AgentState) -> dict:
docs = vector_store.similarity_search(state["question"], k=5)
return {"documents": docs}
def grade_documents(state: AgentState) -> dict:
question = state["question"]
docs = state["documents"]
filtered = []
for doc in docs:
score = llm.invoke(f"Is this doc relevant to '{question}'? Answer yes/no: {doc.page_content}")
if "yes" in score.lower():
filtered.append(doc)
return {"documents": filtered, "need_more_search": len(filtered) < 2}
def generate(state: AgentState) -> dict:
context = "\n".join([d.page_content for d in state["documents"]])
answer = llm.invoke(f"Based on:\n{context}\n\nAnswer: {state['question']}")
return {"answer": answer}
def decide_to_search(state: AgentState) -> str:
if state["need_more_search"] and state["iterations"] < 3:
return "rewrite"
return "generate"
def rewrite_query(state: AgentState) -> dict:
better_query = llm.invoke(f"Rewrite this question for better search: {state['question']}")
return {"question": better_query, "iterations": state["iterations"] + 1}
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite_query)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", decide_to_search, {
"rewrite": "rewrite",
"generate": "generate",
})
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)
app = workflow.compile()
result = app.invoke({"question": "RAG和微调哪个更适合我的场景？", "iterations": 0})
LangGraph的优势在于：可视化工作流、支持检查点恢复、条件分支灵活、易于调试。生产环境首选。
避坑指南：5大常见陷阱
| # | 陷阱 | 表现 | 解决方案 |
|---|---|---|---|
| 1 | 无限循环 | Agent不断重写查询，永远不满意 | 设置max_iterations硬上限（3-5次） |
| 2 | 过度检索 | 每个问题都触发多轮搜索，延迟飙升 | 简单问题走快速路径，复杂问题才走Agentic |
| 3 | 工具幻觉 | Agent编造不存在的工具名调用 | 严格校验工具名，白名单机制 |
| 4 | 上下文爆炸 | 多轮检索的文档全部塞进Prompt | 每轮只保留评分最高的Top-K文档 |
| 5 | 评分偏差 | LLM对文档相关性评分不稳定 | 用结构化输出（JSON Schema）约束评分格式 |
报错排查：10大常见错误
| # | 错误信息 | 原因 | 解决方案 |
|---|---|---|---|
| 1 | GraphRecursionError: Recursion limit reached |
Agent循环超过LangGraph默认步数限制 | 设置recursion_limit参数，或优化条件边逻辑 |
| 2 | KeyError: 'documents' |
State中缺少预期字段 | 检查节点返回值是否包含所有必需字段 |
| 3 | ToolCallingError: Unknown tool |
Agent调用了未注册的工具 | 检查tools列表，确保工具名与描述一致 |
| 4 | TokenLimitExceeded |
多轮检索累积文档超过上下文窗口 | 在grade节点过滤文档，限制传入generate的文档数 |
| 5 | ValidationError: State schema mismatch |
State类型定义与实际数据不匹配 | 检查TypedDict字段类型，Annotated用法是否正确 |
| 6 | TimeoutError: LLM invocation timed out |
评分/生成步骤LLM响应慢 | 设置timeout参数，考虑异步调用 |
| 7 | ImportError: No module named 'langgraph' |
未安装LangGraph | pip install langgraph langchain-core |
| 8 | JSONDecodeError in structured output |
LLM返回非标准JSON | 使用with_structured_output()强制结构化 |
| 9 | RateLimitError: Too many requests |
多轮检索触发API限流 | 加exponential backoff重试，或换用本地模型 |
| 10 | AttributeError: 'NoneType' has no attribute |
某节点返回None | 确保每个节点都返回dict，而非None |
进阶优化技巧
1. 混合路由：简单问题快速路径
不是所有问题都需要Agentic流程。加一个分类器，简单问题走Naive RAG，复杂问题才走Agentic RAG，平均延迟降低60%。
class HybridRouter:
def __init__(self, llm, naive_rag, agentic_rag):
self.llm = llm
self.naive_rag = naive_rag
self.agentic_rag = agentic_rag
def is_simple(self, question: str) -> bool:
result = self.llm.invoke(
f"Is this a simple factual question? Answer yes/no: {question}"
)
return "yes" in result.lower()
def run(self, question: str) -> dict:
if self.is_simple(question):
return self.naive_rag.query(question)
return self.agentic_rag.invoke({"question": question, "iterations": 0})
2. 检查点恢复：长时间工作流不丢失
LangGraph支持检查点（Checkpoint），Agent中断后可从上次状态恢复。
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"question": "分析我们的技术选型", "iterations": 0}, config=config)
3. 并行检索：多知识源同时搜索
import asyncio
async def parallel_retrieve(question: str, stores: list) -> list:
tasks = [store.asimilarity_search(question, k=3) for store in stores]
results = await asyncio.gather(*tasks)
return [doc for batch in results for doc in batch]
4. 缓存热点查询
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_retrieve(query_hash: str, question: str) -> list:
return vector_store.similarity_search(question, k=5)
5. 可观测性：追踪Agent每一步决策
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agentic_rag")
def traced_node(func):
def wrapper(state):
logger.info(f"Node: {func.__name__}, Input: {str(state)[:200]}")
result = func(state)
logger.info(f"Node: {func.__name__}, Output: {str(result)[:200]}")
return result
return wrapper
对比分析：Agentic RAG vs Naive RAG vs Advanced RAG vs Modular RAG
| 维度 | Naive RAG | Advanced RAG | Modular RAG | Agentic RAG |
|---|---|---|---|---|
| 检索策略 | 单次向量搜索 | 查询重写+混合检索 | 可插拔检索模块 | Agent自主决策检索 |
| 推理能力 | ✗ | ✗ | ✗ | ✓ 多跳推理 |
| 自我纠错 | ✗ | △ 重排序 | ✗ | ✓ 反思+重写 |
| 工具调用 | ✗ | ✗ | △ | ✓ 多工具协同 |
| 工作流 | 线性管道 | 线性+预处理 | 模块化管道 | 有状态图 |
| 实现复杂度 | ★☆☆ | ★★☆ | ★★★ | ★★★★ |
| 延迟 | ~1s | ~2s | ~2s | ~3-8s |
| 准确率 | 60-70% | 75-85% | 80-88% | 85-95% |
| 适用场景 | 简单FAQ | 标准问答 | 定制化场景 | 复杂多跳问题 |
★越多表示实现复杂度越高；✓支持 △部分支持 ✗不支持；延迟和准确率为参考值
在线工具推荐
- JSON格式化 — 格式化Agent State和检索结果的JSON结构，调试必备
- 哈希计算 — 计算文档去重和查询缓存的MD5/SHA哈希值
- Curl转代码 — 将LLM API调试curl命令转为Python代码
总结与展望
"The future of RAG is not better retrieval — it's smarter agents that know when and how to retrieve."
"RAG的未来不是更好的检索，而是更聪明的Agent——它知道何时检索、如何检索。"
Agentic RAG正在从"实验性架构"走向"生产标配"。2026年的关键趋势：
- 原生Agent支持：LangGraph、LlamaIndex等框架将Agentic RAG作为一等公民
- 多Agent协作：多个专业Agent协同完成复杂检索任务
- 自适应策略：Agent根据问题难度自动选择检索深度
- 可解释性增强：每一步检索决策都有清晰的推理链可追溯
- 成本优化：智能路由+缓存，让Agentic RAG的成本接近Naive RAG
选择Agentic RAG方案的原则：从路由Agent开始，逐步升级。先用路由解决知识库分发问题，再加自我反思提升检索质量，最后用LangGraph编排完整工作流。别一上来就搞最复杂的——先让Agent跑起来，再让它变聪明。
延伸阅读
本站提供浏览器本地工具，免注册即可试用 →
#Agentic RAG#Agent RAG#自主检索Agent#LLM Agent推理#RAG Agent工作流#2026#LangGraph Agentic RAG