L10 Agentic RAG：让 Agent 自己决定"什么时候检索、检索什么"

• 难度 ⭐⭐⭐（中级）
• 预计阅读 30 分钟
• 代码量 含完整 Agentic RAG 实现
• 先修 L4（Tool Use）+ L5（LangGraph）+ L9（框架对比）

课程目标：理解传统 RAG 的局限性，掌握 Agentic RAG 的核心模式（Maker-Checker 迭代循环），能用 LangGraph 实现一个 Agentic RAG 系统。

一、传统 RAG 怎么了？#

RAG（Retrieval Augmented Generation）是 2023-2024 年 LLM 应用的主流架构：

用户问题 → Embedding → 向量检索 → 拿 Top-K 文档 → LLM 生成答案

这个流程有一个致命假设："先检索后生成"是一成不变的。

反例 1：模糊查询#

用户问："公司上季度表现怎么样？"

传统 RAG 检索出 10 个文档片段，但第一个文档提到的是 2024 Q3，第二个是 2025 Q1——用户问的"上季度"是哪个？

传统 RAG 不知道。它只会把所有片段塞给 LLM，让 LLM 自己推理（结果经常搞错）。

反例 2：需要多步推理的问题#

用户问："对比我们公司和竞品 X 在 AI Agent 战略上的差异"

传统 RAG 检索出"我们公司"和"竞品 X"的文档——但没有"对比"。

LLM 只能基于两堆独立文档做表面对比，无法生成深度分析。

反例 3：检索失败时无能为力#

传统 RAG 检索出 0 个相关文档——它会直接告诉 LLM "没找到"，LLM 只能瞎答。

没有"换个关键词再试"的机制。

二、用一个比喻：传统 RAG vs Agentic RAG#

把"用知识库回答问题"比作"图书馆查资料"：

传统 RAG = 图书管理员（机械执行）#

• 你说："我想要关于 X 的书"

• 管理员按你给的关键词检索，不管关键词对不对

• 找到 5 本相关 / 不相关的书给你

• 不够？你自己想办法

Agentic RAG = 研究助理（主动思考）#

• 你说："我想要关于 X 的书"

• 助理先分析你的需求："他想要的是 X 还是 Y？时间范围呢？"

• 检索第一次：找到 5 本，发现不够

• 反思：关键词太宽，改成更具体的词

• 检索第二次：找到 8 本，这次相关

• 整合、对比、写出报告

核心区别：Agentic RAG 把"检索"从固定流水线升级成由 LLM 主导的控制循环。

三、传统 RAG vs Agentic RAG 对比#

四、Agentic RAG 的核心：Maker-Checker 迭代循环#

这是 Agentic RAG 的灵魂（来源：Anthropic 2025 多 Agent 研究系统）：

        用户问题
           ↓
       [Maker] 生成查询
           ↓
       [Tool] 调用检索
           ↓
       [Checker] 评估证据
           ↓
    证据足够？ ─── 否 ──→ 改写查询 → 回到 Maker
           │
          是
           ↓
       生成最终答案

Maker 的职责#

• 基于用户问题生成第一个检索查询

• 生成答案草稿

Checker 的职责#

• 评估检索到的证据是否足够

• 如果不够，指出哪里不够（"还缺 X 时间段的数据"）

• 给出改写建议（"下次试试搜 'Q4 2024 财报'"）

终止条件#

• 证据足够 → 进入最终生成

• 达到最大迭代次数（如 3-5 次）→ 强制结束

• 超出时间预算 → 强制结束

五、完整实现：基于 LangGraph 的 Agentic RAG#

我们构建一个能回答"公司财报"问题的 Agentic RAG 系统。

完整代码（agentic_rag.py）#

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# ============ 1. 准备知识库 ============

# 加载文档
loader = TextLoader("company_reports.txt")
documents = loader.load()

# 切分成 chunks
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

# 建立向量索引
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# ============ 2. 定义 State ============

class RAGState(TypedDict):
    messages: Annotated[list, add_messages]
    original_query: str       # 原始用户问题
    current_query: str        # 当前检索查询（会被改写）
    retrieved_docs: list[str] # 已检索的文档
    iterations: int           # 已迭代次数
    is_sufficient: bool       # 证据是否足够

# ============ 3. 准备 LLM ============

llm = ChatOpenAI(model="gpt-4o-mini")

# ============ 4. Maker 节点：生成查询 + 检索 ============

def maker_node(state: RAGState):
    """Maker：生成或改写检索查询"""
    original = state.get("original_query") or state["messages"][-1].content
    last_iteration = state.get("iterations", 0)
    
    if last_iteration == 0:
        # 第一次：直接用原始查询
        query = original
    else:
        # 后续：基于上次反馈改写查询
        reflection = state["messages"][-1].content
        prompt = f"""基于以下反馈，改写检索查询以获取更好的结果：

        原始问题：{original}
        上次检索反馈：{reflection}
        
        请输出新的检索查询（只输出查询本身，不要其他内容）："""
        query = llm.invoke(prompt).content.strip()
    
    # 执行检索
    docs = retriever.invoke(query)
    doc_texts = [doc.page_content for doc in docs]
    
    return {
        "original_query": original,
        "current_query": query,
        "retrieved_docs": state.get("retrieved_docs", []) + doc_texts,
        "iterations": last_iteration + 1,
        "messages": [llm.invoke(f"已检索：{query}，获得 {len(doc_texts)} 个文档")]
    }

# ============ 5. Checker 节点：评估证据 ============

def checker_node(state: RAGState):
    """Checker：评估检索证据是否足够"""
    original = state["original_query"]
    docs = state["retrieved_docs"]
    docs_text = "\n\n".join([f"[文档{i+1}] {d}" for i, d in enumerate(docs[-6:])])  # 只看最近 6 个
    
    prompt = f"""请评估以下证据是否能回答用户问题：

    用户问题：{original}
    
    当前检索证据：
    {docs_text}
    
    评估标准：
    1. 证据是否直接相关？
    2. 信息是否足够完整？
    3. 是否需要更多信息？
    
    输出格式：
    SUFFICIENT: yes/no
    REASONING: <简短理由>
    NEXT_QUERY_HINT: <如果不够，下一次检索应该关注什么>
    """
    
    response = llm.invoke(prompt).content
    is_sufficient = "SUFFICIENT: yes" in response
    
    return {
        "is_sufficient": is_sufficient,
        "messages": [response]  # 把评估结果作为下一轮 Maker 的反馈
    }

# ============ 6. 答案生成节点 ============

def generator_node(state: RAGState):
    """基于所有检索到的文档生成最终答案"""
    original = state["original_query"]
    docs = state["retrieved_docs"]
    
    docs_text = "\n\n".join([f"[文档{i+1}] {d}" for i, d in enumerate(docs)])
    
    prompt = f"""请基于以下证据回答用户问题：

    问题：{original}
    
    证据：
    {docs_text}
    
    要求：
    1. 答案必须有证据支撑
    2. 标注引用来源（用 [文档X] 格式）
    3. 如果证据有冲突，说明并选择最权威的"""
    
    answer = llm.invoke(prompt).content
    return {"messages": [answer]}

# ============ 7. 条件边：决定是继续迭代还是生成答案 ============

def should_continue(state: RAGState) -> str:
    if state.get("is_sufficient", False):
        return "generate"
    if state.get("iterations", 0) >= 3:
        return "generate"  # 强制结束
    return "iterate"

# ============ 8. 装配图 ============

graph_builder = StateGraph(RAGState)

graph_builder.add_node("maker", maker_node)
graph_builder.add_node("checker", checker_node)
graph_builder.add_node("generator", generator_node)

graph_builder.add_edge(START, "maker")
graph_builder.add_edge("maker", "checker")
graph_builder.add_conditional_edges(
    "checker",
    should_continue,
    {
        "iterate": "maker",      # 证据不够，重新检索
        "generate": "generator"   # 证据足够，生成答案
    }
)
graph_builder.add_edge("generator", END)

graph = graph_builder.compile()

运行示例#

result = graph.invoke({
    "messages": [{"role": "user", "content": "公司上季度营收增长情况如何？与去年同期对比呢？"}]
})
print(result["messages"][-1].content)

典型运行过程：

=== 迭代 1 ===
Maker: 检索 "上季度 营收 增长"
检索到 3 个文档（部分相关，但缺少"去年同期"数据）

Checker: SUFFICIENT: no
         缺少去年同期对比数据
         NEXT_QUERY_HINT: 应该检索 2024 Q4 财报

=== 迭代 2 ===
Maker: 检索 "2024 Q4 财报 营收"
检索到 3 个文档（包含去年同期数据）

Checker: SUFFICIENT: yes
         证据齐全

=== 生成最终答案 ===
综合两轮检索的 6 个文档，生成包含对比的详细答案。

六、三大生产模式（递增复杂度）#

模式 1：迭代查询改写（最简单）#

架构：单检索工具 + 查询改写循环

用户问题 → Maker（生成查询）→ 检索 → Checker（评估）→ 不够就改写 → 回到 Maker

适用：FAQ 增强、知识库检索 实现：上面那个 LangGraph 代码就是

模式 2：多工具编排（中等）#

架构：Agent 在多个检索工具中选择

用户问题 → Agent 决策：
             ├─ 向量检索（语义相似）
             ├─ 关键词检索（精确匹配）
             ├─ SQL 查询（结构化数据）
             └─ API 调用（外部数据源）
           → 综合所有结果

实现：用 LangGraph + 多个 retriever 节点 + 路由函数

典型场景：企业知识库 + 数据库 + 第三方 API 综合查询

模式 3：层次化 Agent 分解（最复杂）#

架构：Planner Agent 拆子任务 → 子 Agent 并行检索 → Aggregator 综合

用户问题
   ↓
[Planner Agent] 拆分子问题
   ↓
[SubAgent 1] 检索子问题 A
[Subagent 2] 检索子问题 B  ← 并行
[Subagent 3] 检索子问题 C
   ↓
[Aggregator Agent] 综合所有子答案

适用：复杂研究任务（如 Anthropic 的 multi-agent research）

成本：是模式 1 的 5-15 倍 token（参考 Anthropic 2025 报告）

七、和传统 RAG 的成本对比#

来自 Anthropic 2025 的实测数据：

结论：Token 换准确率——每提升一个档位，token 涨 4 倍，准确率提升 15-30%。

八、何时用 Agentic RAG vs 传统 RAG#

九、和 L1-L9 的连接#

十、动手时间：45 分钟亲手实现#

步骤 1：准备测试文档（10 分钟）#

找 5-10 个文本文件（如公司财报、产品说明），用 TextLoader 加载。如果没现成文件，可以编 3-5 段假数据。

步骤 2：跑通上面的完整代码（15 分钟）#

pip install langchain langchain-openai langchain-community faiss-cpu

把代码复制到 agentic_rag.py，替换文档路径，跑通。

步骤 3：观察迭代过程（20 分钟）#

故意问一个需要多步检索的问题，观察：

• 第一次检索用了什么关键词

• Checker 怎么判断"证据不够"

• Maker 怎么改写查询

• 几轮后生成最终答案

修改 max_iterations=3 改成 5，观察效果差异。

十一、小结#

十二、Read next#

下一节 L13 可信赖 Agent 安全，我们学习生产环境必须解决的安全问题：

• Prompt Injection 防御：恶意用户怎么攻击 Agent

• Meta System Message 框架：微软的官方安全防护模式

• 敏感操作拦截：Agent 不能直接执行的危险操作

主要参考：

• src-001 §05-agentic-rag — 微软 Agentic RAG 课程

• Anthropic 多 Agent 研究系统 — 工业级 Agentic RAG 实践

十三、Sources#

• src-001 — Microsoft AI Agents for Beginners · T1 · 05-agentic-rag

• Anthropic 官方博客 · T1 · Multi-agent research system

• LangGraph RAG 教程 · T1 · Adaptive RAG

💬 有疑问？ 试试问我："Agentic RAG 怎么避免无限循环？"、"传统 RAG 和 Agentic RAG 该怎么选？"、"Agentic RAG 的检索质量不行怎么办？"——我会用代码或比喻为你解答。