RAG 进阶

Agentic RAG

固定 RAG 流水线解决不了多源、多跳、需要判断下一步的问题时，把检索变成 Agent 可调用的工具。

适合阶段：复杂问答与多源检索核心机制：Plan / Tool / Observe / Stop面试重点：可控性与评估

Agentic RAG 让模型决定何时查、查哪里、查几次。

面试官想考什么

读完这篇你要能正面回答下面这些题。每题后面括号里是面试官真正想看你答出什么。

• Agentic RAG 和普通 RAG 最大区别是什么？
  考固定 pipeline 和动态 tool loop 的差异。
• 什么场景值得上 Agentic RAG？什么场景不该上？
  考工程克制，不是所有问答都需要 Agent。
• Agentic RAG 的典型循环怎么设计？如何避免无限检索？
  考 max steps、stop condition、evidence grading。
• 检索工具应该怎么设计 schema？返回整段文本还是结构化证据？
  考工具契约和可引用性。
• Agent 什么时候应该重写 query、换检索源、或者拒答？
  考分支控制，而不是无脑 tool calling。
• Agentic RAG 怎么评估？为什么比普通 RAG 更难测？
  考轨迹评估、工具调用准确率和端到端质量。
• Agentic RAG 和 GraphRAG、Self-RAG、ReAct 是什么关系？
  考概念辨析和架构分层。

---

为什么需要 Agentic RAG

普通 RAG 是固定流水线：

query -> retrieve top-K -> rerank -> generate

这个流程适合单次检索能拿到证据的问题。比如"XPay 退款接口 timeout 最大多少秒？"，检索接口文档就够了。

但复杂问题经常不是一次检索能解决的：

XPay 退款接口 timeout 最大多少？这个限制是谁维护的？最近有没有相关投诉？

这其实是三个问题：

• 参数范围在 API 文档里。
• 维护团队可能在服务目录或组织文档里。
• 投诉信息可能在工单库里。

固定 RAG 只能提前决定"查哪个库、查几次、怎么组合"。Agentic RAG 把这些动作变成工具，让模型或控制器根据中间结果决定下一步：先查 API 文档，发现实体是 XPay 退款接口，再查服务目录，最后查投诉工单；证据足够就回答，不够就继续或拒答。

LangChain 的 retrieval 文档把 Agentic RAG 描述成：LLM-powered agent decides when and how to retrieve during reasoning。LlamaIndex 的 agentic strategies 也强调，可以把已有 RAG query engine 作为工具交给 agent 做更复杂决策。重点不在更换检索算法，重点在把检索从固定组件变成可规划动作。

Agentic RAG 是怎么工作的

它实际是一个有上限的工具循环。

流程图

flowchart LR
    Q["用户问题"] --> A["Planner / Router"]
    A --> B{"选择动作"}
    B -->|查政策库| C["policy_retriever"]
    B -->|查 API 文档| D["api_retriever"]
    B -->|查工单| E["ticket_search"]
    B -->|重写问题| F["query_rewrite"]
    C --> G["Observe evidence"]
    D --> G
    E --> G
    F --> A
    G --> H["Evidence grading"]
    H -->|不足且未超步数| A
    H -->|足够或超步数| I["Answer / Refuse with citations"]

这个结构和 ReAct 很像。ReAct 论文 Synergizing Reasoning and Acting 讲的是模型交替生成 reasoning trace 和 actions；Toolformer 论文 Language Models Can Teach Themselves to Use Tools 讨论模型何时调用 API、传什么参数、如何使用返回结果。Agentic RAG 可以看成这些工具使用思想在检索任务上的应用。

工程上不要让 agent 自由奔跑。一个生产级 Agentic RAG 至少要有：

• 可枚举工具列表和参数 schema
• 每步的最大候选数和最大 token 数
• evidence grading 和停止条件
• max steps / max cost / timeout
• 每步 trace、引用来源和权限过滤

核心原理 / 关键设计

1. 检索工具要小而清楚

不要给 agent 一个过宽的 search_everything(query)。工具边界越清楚，模型越容易选对。

TOOLS = {
    "search_api_docs": "查 API 参数、错误码、接口限制",
    "search_policy": "查员工制度、合同、权限政策",
    "search_tickets": "查近期工单、投诉、故障记录",
}

工具返回也要结构化。只返回一段长文本，后面无法做引用、去重、权限审计。

{
  "source": "api/xpay/refund.md",
  "section": "timeout",
  "text": "timeout 参数范围为 1-300 秒。",
  "score": 0.82,
  "updated_at": "2026-05-12"
}

这看起来像小事，但面试很容易问。Agentic RAG 的可靠性很大程度来自工具契约，而不是 prompt 写得多漂亮。

2. Planner 不应该直接生成最终答案

Agent 每一步会做两类事：选工具、读观察结果。最终答案最好由单独的 answer step 生成，并且只允许使用已收集证据。

bad:
Planner 一边想一边答，工具结果和模型猜测混在一起。

better:
Planner 只产出 action；
Tool 返回 evidence；
Evaluator 判断够不够；
Answerer 基于 evidence 输出答案和引用。

这样做能降低幻觉，也方便评估每个环节。LangGraph 的 Agentic RAG 教程里也把 retrieval、grade documents、rewrite、generate 拆成不同节点，本质就是为了让控制流可观察。

3. Stop condition 比多查一次更重要

Agentic RAG 最常见的故障：查了一步又一步，仍不给停止理由。

for step in range(max_steps):
    action = planner(state)
    observation = run_tool(action)
    state.add(observation)
    if evidence_is_enough(state):
        break
else:
    return refuse("检索次数达到上限，证据仍不足")

停止条件不能只靠"模型觉得够了"。至少要组合几类信号：是否命中必须字段、是否有来源引用、是否覆盖子问题、是否达到 max steps、是否超出成本预算。

4. Agentic RAG 更需要权限和版本控制

固定 RAG 通常只查一个库，权限过滤容易集中处理。Agentic RAG 会跨多个工具：政策库、工单库、CRM、代码库、网页搜索。只要一个工具忘了 tenant/user filter，就可能泄露数据。

每个工具调用必须带：
tenant_id
user_id / role
allowed_sources
document_version_policy

工具层要强制权限，不要只靠 planner 自觉。Agent 选择"查工单"时，工单检索工具必须自己做 ACL；不能让上层 prompt 说一句"不要越权"就算完。

5. Agent 轨迹本身就是评估对象

普通 RAG 可以只看输入输出和引用。Agentic RAG 还要看中间步骤：

用户问题 -> 选了哪些工具 -> 每个工具参数是什么 -> 返回哪些证据
-> 是否重写 query -> 是否覆盖子问题 -> 为什么停止

没有 trace，你只能看到最终错了，不知道是 planner 选错工具、query 写错、工具没返回、还是 evaluator 误判证据足够。

怎么用：标准库写一个可控 Agentic RAG loop

下面这段代码不用外部模型，演示 Agentic RAG 的控制流：工具路由、多步检索、证据检查、停止和引用。真实项目里，把 plan_next_action() 换成 LLM tool calling，把 search_* 换成真实 retriever。

from dataclasses import dataclass, field

API_DOCS = {
    "api-1": "XPay 退款接口 timeout 参数范围为 1-300 秒，超过 300 秒会被网关拒绝。",
    "api-2": "XPay 查询订单接口默认 timeout 为 30 秒。",
}
SERVICE_DOCS = {
    "svc-1": "XPay 退款接口由支付网关团队维护，负责人轮值在服务目录中更新。",
}
TICKETS = {
    "ticket-1": "近 7 天支付网关团队收到 12 起退款接口 timeout 超限投诉。",
}


@dataclass
class Evidence:
    source: str
    text: str


@dataclass
class State:
    question: str
    actions: list[str] = field(default_factory=list)
    evidence: list[Evidence] = field(default_factory=list)


def keyword_search(query: str, corpus: dict[str, str]) -> list[Evidence]:
    terms = [t for t in query.lower().replace("？", " ").split() if len(t) > 1]
    rows = []
    for source, text in corpus.items():
        score = sum(1 for term in terms if term in text.lower())
        if score:
            rows.append((score, Evidence(source, text)))
    return [e for _, e in sorted(rows, key=lambda x: x[0], reverse=True)]


def search_api_docs(query: str) -> list[Evidence]:
    return keyword_search(query, API_DOCS)


def search_service_catalog(query: str) -> list[Evidence]:
    return keyword_search(query, SERVICE_DOCS)


def search_tickets(query: str) -> list[Evidence]:
    return keyword_search(query, TICKETS)


def plan_next_action(state: State) -> str:
    text = "\n".join(e.text for e in state.evidence)
    q = state.question.lower()

    if "api" not in state.actions and ("timeout" in q or "接口" in q):
        return "api"
    if "service" not in state.actions and ("维护" in state.question or "谁" in state.question or "团队" in state.question):
        return "service"
    if "tickets" not in state.actions and ("投诉" in state.question or "最近" in state.question):
        return "tickets"
    if "timeout 参数范围" in text and ("维护" not in state.question and "投诉" not in state.question):
        return "done"
    return "done"


def run_action(action: str, question: str) -> list[Evidence]:
    if action == "api":
        return search_api_docs(question)
    if action == "service":
        return search_service_catalog(question)
    if action == "tickets":
        return search_tickets(question)
    return []


def enough_evidence(state: State) -> bool:
    text = "\n".join(e.text for e in state.evidence)
    checks = []
    if "timeout" in state.question:
        checks.append("timeout 参数范围" in text)
    if "谁" in state.question or "维护" in state.question:
        checks.append("维护" in text and "团队" in text)
    if "投诉" in state.question or "最近" in state.question:
        checks.append("投诉" in text)
    return bool(checks) and all(checks)


def agentic_rag(question: str, max_steps: int = 4) -> State:
    state = State(question=question)
    for _ in range(max_steps):
        action = plan_next_action(state)
        if action == "done" or enough_evidence(state):
            break
        state.actions.append(action)
        state.evidence.extend(run_action(action, question))
    return state


if __name__ == "__main__":
    state = agentic_rag("XPay 退款接口 timeout 最大多少？谁维护？最近有投诉吗？")
    print("actions:", state.actions)
    for e in state.evidence:
        print(f"[{e.source}] {e.text}")

这个 demo 里，Agent 不会一次性查所有库，而是根据问题和已有证据选择下一步。生产系统要把 actions、工具参数、返回证据、停止原因全部写进 trace。

容易踩的坑

坑 1：把 Agentic RAG 用在简单 FAQ

现象：原来 300ms 的知识库问答，变成几秒，还多了不稳定工具调用。

根因：简单单跳问题不需要规划。Agent loop 增加了 LLM 调用、分支和状态。

修法：做 query classifier。单点事实问题走普通 RAG；多源、多跳、需要补查的问题才走 Agentic RAG。

坑 2：工具描述太像，模型乱选

现象：该查 API 文档时查了工单库，该查政策库时查了网页。

根因：工具边界不清，description 写成"搜索相关信息"这种泛话。模型无法根据语义区分工具。

修法：工具名和描述写具体：search_api_docs、search_hr_policy、search_recent_tickets；给参数 schema 和反例；用离线问题集测 tool selection accuracy。

坑 3：没有 max steps 和成本上限

现象：Agent 一直检索、重写、再检索，线上延迟和成本失控。

根因：没有硬停止条件，或者把停止完全交给模型判断。

修法：设置 max steps、timeout、max tokens、max tool calls；evidence grading 达标才生成；未达标就拒答或请求澄清。

坑 4：工具返回不可引用

现象：答案看起来对，但没法知道来自哪份文档；用户追问来源时系统说不清。

根因：工具只返回纯文本，丢了 source、section、page、version、score。

修法：工具返回结构化 evidence。最终回答必须引用 evidence id；没有来源的文本不能进入答案。

坑 5：跨工具权限不一致

现象：向量库做了租户过滤，工单工具却返回了别的团队数据。

根因：Agentic RAG 把多个数据源组合起来，权限模型分散在各工具里，容易漏。

修法：每个工具强制接收 auth context，并在工具内部做 ACL；trace 记录每次工具调用的 tenant/user/source filter。

与相似概念的区别

概念	决策权在哪里	适合什么问题	主要风险
Naive RAG	固定 pipeline	单跳事实问答	漏召回、缺自检
Advanced RAG	控制流规则 + LLM 改写/评分	召回失败、自检、纠错	分支复杂
GraphRAG	图索引和查询路由	全局主题、多跳关系	构图成本高
Agentic RAG	Agent 动态选择工具和步骤	多源、多步、需要规划	不稳定、难评估、成本高
ReAct Agent	Thought/Action/Observation 循环	通用工具使用	容易循环和幻觉动作

Agentic RAG 可以包含 GraphRAG 工具，也可以包含 Advanced RAG 的 query rewrite 和 evidence grading。它更像控制层：决定用哪些检索算法、按什么顺序用、什么时候停止。

面试题深度解析

Q1: Agentic RAG 和普通 RAG 最大区别是什么？

• 30 秒版本：普通 RAG 是固定检索生成流水线；Agentic RAG 把检索源包装成工具，由 Agent 根据问题和中间证据决定查哪里、查几次、是否重写或拒答。
• 追问 1：这带来什么收益？能处理多源、多跳和不确定问题，比如先查 API 文档，再查服务目录，再查工单。
• 追问 2：代价是什么？延迟、成本、可重复性和评估复杂度都上升，所以要有 max steps、trace 和工具权限控制。

Q2: 什么场景值得上 Agentic RAG？

• 30 秒版本：问题需要多源数据、动态分解、补查、或工具选择时值得上。单库 FAQ 和简单政策问答不需要。
• 追问 1：怎么判断？看历史失败 case：如果错误来自"查错库/少查一步/需要跟进问题"，Agentic RAG 有价值；如果错误来自 chunking 或 embedding，先修基础 RAG。
• 追问 2：怎么灰度？先用 query classifier 只让复杂问题进入 Agentic RAG，对比普通 RAG 的答案准确率、延迟和成本。

Q3: Agentic RAG 怎么避免无限检索？

• 30 秒版本：硬上限 + 证据评分 + 停止原因。max steps、timeout、max tool calls 必须写死，不能只靠模型说"我完成了"。
• 追问 1：证据评分怎么做？检查子问题是否都有来源、是否命中必要字段、引用是否来自允许数据源。
• 追问 2：证据不足怎么办？拒答、请求澄清、或走受控补救分支。不要继续无限重写 query。

Q4: Agentic RAG 怎么评估？

• 30 秒版本：评估最终答案还不够，还要评估轨迹：工具选择是否正确、参数是否正确、证据是否支持答案、是否有多余调用。
• 追问 1：指标有哪些？tool selection accuracy、evidence recall、answer accuracy、citation precision、average tool calls、p95 latency、cost per query。
• 追问 2：怎么 debug？保存每步 trace，并把失败归因到 planner、retriever、evaluator、answerer。没有分段归因，Agentic RAG 会变得很难诊断。

延伸阅读

• 文档：LangChain Retrieval: Agentic RAG — 官方对 semantic search、RAG、Agentic RAG 的对比很清楚。
• 文档：LangGraph Agentic RAG — 看 retrieval tool、grade documents、rewrite、generate 如何拆成图节点。
• 文档：LlamaIndex Agentic Strategies — 看如何把 query engine 作为 agent tool。
• 文档：LlamaIndex Agents — 理解 LlamaIndex 里 RAG pipeline 和 agent/tool 的关系。
• 论文：ReAct — Agentic RAG 的行动循环可以从 ReAct 理解。
• 论文：Toolformer — 理解模型何时调用工具、如何使用工具结果。
• 论文：Self-RAG — 检索、生成、自我批判的训练化路线，适合和 Agentic RAG 对比。
• 论文：Reflexion — 语言反思如何帮助 agent 从失败反馈中改进下一步。