Agentic RAG 进阶：让 Agent 自己决定"查什么、查几次、怎么查"

作者：yumking | 2026 年 9 月 19 日 | 技术标签：Agentic RAG / 多跳检索 / GraphRAG / 查询规划 / 自反思

摘要

第 4 篇搭了 RAG 基础流水线（查一次 → 拼 prompt → 生成），把准确率从 58% 拉到 92%；第 12 篇把检索从纯文本扩到图文多模态。但两篇共享一个隐含假设：一次检索就能拿全证据。真实世界里"对比 A、B 两公司去年利润增长率"要查四次（两家 × 两年），“张三导师的学生都在哪些公司"要沿关系图走三跳——单跳 RAG 在这种问题上要么硬凑要么放弃。本文把检索从"被动的一次性调用"升级为"Agent 主动规划的迭代过程”：查询分解与路由、多跳迭代检索、GraphRAG 让关系可检索、自反思重写查不好就重查。实测多跳问题准确率从 34%（单跳 RAG）提升至 88%，关系类问题从 22% 提升至 91%，代价是平均检索轮数从 1.0 升到 2.7、延迟从 450ms 升到 1.8s——用"多查几轮"换"答得对"。

一、为什么"查一次"不够

1.1 第 4 篇 RAG 的隐含假设

第 4 篇的五阶段流水线，在线查询阶段是这样的：

用户提问 ──④向量检索──► Top-K 相关块 ──⑤LLM 生成──► 答案
                          ↑
                    一次性检索，K 通常取 5-10

这套流程背后藏着一个假设：用户的问题可以被一次检索覆盖。对于"公司年假政策是什么"这类单点事实问题，假设成立——查一次就够。但很多真实问题不是单点的。

1.2 三类"查一次"解决不了的问题

一个真实翻车案例：

用户问："我们上季度收入最高的三个产品线，各自的负责人和汇报对象是谁？"
单跳 RAG：拿"收入最高 三个产品线 负责人 汇报对象"整句去检索
  → 知识库里没有哪篇文档同时写了这些信息
  → 召回一堆提到"收入"的零散片段
  → LLM 硬凑："根据资料，可能是..."（幻觉）
实际需要四跳：
  ① 查上季度各产品线收入  ② 排序取前三
  ③ 查这三个产品线的负责人  ④ 查这些负责人的汇报对象
每跳的输出是下一跳的输入——这是检索的"链式调用"。

1.3 与第 4/12 篇的关系

第 4 篇（RAG 基础）
  → 解决"查文字知识"，本文解决"复杂问题怎么查"
第 12 篇（多模态 RAG）
  → 扩展了检索的"模态宽度"（文+图+表），本文扩展"检索深度"（多跳+迭代）
第 7 篇（任务规划）
  → 本文的查询规划是任务规划在检索层的特化——把"怎么完成任务"具体到"怎么查资料"
第 9 篇（经验学习）
  → 本文的自反思重写会沉淀"哪种重写策略有效"，喂回经验库

第 4 篇是"会查"，第 12 篇是"查得全"，本文是"查得深、查得准、查得聪明"。

二、查询规划：把"查什么"交给 Agent

2.1 从单次检索到检索计划

单跳 RAG 的检索是"无脑的一步"：query 进去，Top-K 出来。Agentic RAG 在这一步前面加一个规划阶段——让 LLM 先看问题，决定要查几次、每次查什么、结果怎么组合。

单跳 RAG：
  query ──► retrieve ──► generate

Agentic RAG：
  query ──► plan ──► [retrieve₁ → retrieve₂ → ... → retrieveₙ] ──► synthesize
              ↑
        规划器决定 n、每次的 sub-query、路由到哪个数据源

规划器的输出不是答案，是一份检索计划——类似第 7 篇的任务分解，但粒度更细，专门针对检索。

2.2 查询分解：复杂问题拆子问题

# query_planner.py —— 查询规划器
from pydantic import BaseModel

class SubQuery(BaseModel):
    text: str               # 子问题文本
    source: str             # 路由到哪个数据源：vector / graph / sql / web
    depends_on: list[int]   # 依赖哪些前序子问题的结果（-1 表示无依赖）
    extract: str            # 从前序结果里提取什么填进自己的 query

class RetrievalPlan(BaseModel):
    sub_queries: list[SubQuery]
    synthesize: str         # 最终怎么组合子答案

PLANNER_PROMPT = """你是检索规划器。把用户问题分解成若干子检索。
规则：
1. 每个子检索必须能被一次向量/图/SQL 检索回答，不要把多跳塞进一个子检索
2. 用 depends_on 标注依赖：子检索 B 需要 A 的结果时，B.depends_on=[A 的序号]
3. extract 字段说明从前序结果提取什么：如 "上一步返回的三个产品线名称"
4. source 选最合适的源：关系问题选 graph，数值聚合选 sql，一般选 vector
5. synthesize 说明最终怎么组合

用户问题：{question}
可用数据源：{sources}
"""

class QueryPlanner:
    def __init__(self, llm, retrievers: dict):
        self._llm = llm
        self._retrievers = retrievers   # {"vector": vec, "graph": graph, "sql": sql}

    def plan(self, question: str, sources: list[str]) -> RetrievalPlan:
        resp = self._llm.complete(
            PLANNER_PROMPT.format(question=question, sources=sources),
            response_format=RetrievalPlan,
        )
        return resp.parsed

    def execute(self, plan: RetrievalPlan) -> dict:
        """按依赖顺序执行子检索，后一步可引用前一步结果"""
        results = {}
        for i, sq in enumerate(plan.sub_queries):
            # 把前序结果填入当前子问题
            resolved_query = self._fill_dependencies(sq, results)
            retriever = self._retrievers[sq.source]
            results[i] = retriever.search(resolved_query, top_k=8)
        return {"sub_results": results, "plan": plan}

    def _fill_dependencies(self, sq: SubQuery, results: dict) -> str:
        if not sq.depends_on:
            return sq.text
        # 简化：把依赖步的返回文本拼进 query；生产环境用 LLM 按 extract 字段精确提取
        deps = " ".join(
            results[d].summary for d in sq.depends_on if d in results)
        return f"{sq.text} 上下文：{deps}"

规划器的关键不在分解本身，在于依赖标注——depends_on 让子检索可以串成链，而不是并行各查各的。

2.3 查询路由：不同子问题去不同源

单跳 RAG 把所有问题都塞进一个向量库——查关系查不出、查统计查不准。路由的本质是让每种问题去它该去的地方。

三、多跳检索：链式追证据

3.1 迭代检索-生成循环

多跳检索不是"并行查 n 次"，而是"查一次拿到一部分答案，根据这部分答案决定下一步查什么"——每跳的 query 由上一跳的结果动态生成。

跳 1：query₀ = "上季度收入最高的三个产品线"
       → 结果：[企业版, 国际版, 教育版]
跳 2：query₁ = "企业版/国际版/教育版 的负责人"   ← query 由跳 1 结果生成
       → 结果：{企业版: 张三, 国际版: 李四, 教育版: 王五}
跳 3：query₂ = "张三/李四/王五 的汇报对象"       ← query 由跳 2 结果生成
       → 结果：{张三→VP刘, 李四→VP陈, 王五→CEO赵}
合成：收入最高三线 = 企业版/国际版/教育版
      负责人 = 张三/李四/王五，汇报 = VP刘/VP陈/CEO赵

3.2 证据积累与终止判断

多跳循环要解决两个控制问题：什么时候停、怎么避免空转。

# multi_hop_retriever.py —— 多跳迭代检索器
import re

class MultiHopRetriever:
    """迭代检索：每跳根据已有证据生成下一跳 query，直到证据充分或触顶"""

    def __init__(self, llm, vector_store, max_hops: int = 5,
                 evidence_threshold: float = 0.82):
        self._llm = llm
        self._vs = vector_store
        self._max_hops = max_hops
        self._threshold = evidence_threshold

    NEXT_QUERY_PROMPT = """已有问题：{question}
已收集证据：
{evidence}
还有哪些信息缺失才能回答完整问题？如果证据已充分，返回 DONE。
否则返回下一个该查的子问题（一句话，能被一次向量检索回答）。
"""

    def retrieve(self, question: str) -> dict:
        evidence = []
        trace = []
        for hop in range(self._max_hops):
            # ① 决定这一跳查什么
            if hop == 0:
                next_query = question
            else:
                next_query = self._decide_next(question, evidence)
                if next_query == "DONE":
                    trace.append({"hop": hop, "action": "terminate",
                                  "reason": "evidence_sufficient"})
                    break

            # ② 执行检索
            docs = self._vs.search(next_query, top_k=6)
            evidence.extend(docs)
            trace.append({
                "hop": hop, "query": next_query,
                "retrieved": len(docs),
                "cumulative_evidence": len(evidence),
            })

            # ③ 证据充分性自评
            if self._sufficiency(question, evidence) >= self._threshold:
                trace.append({"hop": hop, "action": "terminate",
                              "reason": "threshold_met"})
                break

        return {"evidence": evidence, "hops": len(trace) - 1,
                "trace": trace}

    def _decide_next(self, question: str, evidence: list) -> str:
        ev_text = "\n".join(d.summary for d in evidence[-4:])
        resp = self._llm.complete(
            self.NEXT_QUERY_PROMPT.format(
                question=question, evidence=ev_text))
        return resp.strip()

    def _sufficiency(self, question: str, evidence: list) -> float:
        """让 LLM 自评：当前证据够不够回答问题，返回 0-1"""
        resp = self._llm.complete(
            f"问题：{question}\n证据：{[d.summary for d in evidence]}\n"
            f"证据是否充分回答问题？只回 0-1 的数字。")
        m = re.search(r"([01](?:\.\d+)?)", resp)
        return float(m.group(1)) if m else 0.0

三个设计要点：

① max_hops 硬顶：防止 LLM 永远觉得"还差一点"无限循环
② sufficiency 自评：每跳后让 LLM 打分，够分就停，避免多余检索
③ trace 全程留痕：每跳的 query、召回数、终止原因都记下，
   接第 23 篇可观测性——多跳检索的每跳是一个 span

3.3 一个多跳案例的完整轨迹

问题："我们去年离职率最高的团队，其空缺岗位的招聘进度如何"

跳 0  query="去年各团队离职率"
      retrieved=6  sufficiency=0.4  → 继续
跳 1  query="离职率最高的团队名称"（从跳 0 证据里提炼）
      retrieved=5  sufficiency=0.6  → 继续
跳 2  query="该团队当前空缺岗位"
      retrieved=4  sufficiency=0.85 → 阈值 0.82 达标，停
合成：3 跳，总耗时 1.4s，召回 15 篇，答案有据可查

单跳 RAG 在同样问题上：拿整句检索 → 召回一堆提到"离职率"的片段 → 幻觉。

四、GraphRAG：让关系可检索

4.1 向量检索的盲区：关系

向量检索擅长"语义相似"，但有个结构性盲区——查不了关系：

向量检索能查：
  "K8s 是什么" → 找语义相似的文档 ✓

向量检索查不好：
  "谁向 CEO 汇报" → 汇报关系是结构化边，不是语义相似
  "订单服务依赖哪些下游" → 依赖关系是图边，散落在多篇文档里
  "张三导师的学生" → 三跳关系链，向量空间里没有这种路径

根因：向量空间把文档压成点，点之间只有距离没有边。关系是图的结构，要用图来查。

4.2 知识图谱构建：实体抽取 + 关系抽取

GraphRAG 在向量库旁边再建一个知识图谱——离线索引阶段，让 LLM 从每篇文档里抽实体和关系：

# graph_rag.py —— GraphRAG：向量 + 图混合检索
from dataclasses import dataclass

@dataclass
class Entity:
    name: str
    type: str          # person / team / product / service ...

@dataclass
class Relation:
    src: str           # 实体名
    dst: str
    type: str          # reports_to / depends_on / member_of ...

EXTRACT_PROMPT = """从以下文档抽取实体和关系，输出 JSON。
实体类型：person/team/product/service/concept
关系类型：reports_to/depends_on/member_of/owns/uses
只抽文档明确陈述的，不要推断。
文档：{doc}
"""

class GraphRAG:
    """向量召回 + 图遍历的混合检索"""

    def __init__(self, llm, vector_store, graph_store):
        self._llm = llm
        self._vs = vector_store
        self._gs = graph_store       # 网络/Neo4j 等

    def index_doc(self, doc: str) -> None:
        """离线索引：文档同时进向量库和图"""
        self._vs.add(doc)                         # 向量索引
        entities, relations = self._extract(doc)  # LLM 抽实体关系
        for e in entities:
            self._gs.add_entity(e)
        for r in relations:
            self._gs.add_relation(r)

    def _extract(self, doc: str) -> tuple[list, list]:
        resp = self._llm.complete(
            EXTRACT_PROMPT.format(doc=doc))
        return resp.entities, resp.relations

    def search(self, query: str, top_k: int = 5,
               max_hops: int = 2) -> dict:
        """混合检索：向量召回种子 → 图扩展相关实体"""
        # ① 向量检索召回种子文档
        seed_docs = self._vs.search(query, top_k=top_k)
        seed_entities = self._extract_entities_from_query(query)

        # ② 图遍历扩展：从种子实体出发走 max_hops 跳，收相关实体
        expanded = set()
        for ent in seed_entities:
            for hop in range(1, max_hops + 1):
                neighbors = self._gs.neighbors(ent, hop=hop)
                expanded.update(neighbors)

        # ③ 用扩展到的实体再回向量库拉相关文档
        extra_docs = []
        for ent in list(expanded)[:10]:
            extra_docs.extend(
                self._vs.search(ent.name, top_k=2))

        return {
            "seed_docs": seed_docs,           # 直接语义相关
            "graph_entities": list(expanded), # 图扩展到的实体
            "extra_docs": extra_docs,         # 因关系而相关的文档
        }

    def _extract_entities_from_query(self, query: str) -> list[Entity]:
        """从 query 里识别已知实体名，作为图遍历起点"""
        resp = self._llm.complete(
            f"识别这句话里的实体名（只回名字，逗号分隔）：{query}")
        names = [n.strip() for n in resp.split(",") if n.strip()]
        return [self._gs.find_entity(n) for n in names
                if self._gs.find_entity(n)]

GraphRAG 的精髓在第二步：向量检索找到种子，图遍历沿关系扩展，再回向量库拉文档。三个步骤形成一个"召回-扩展-再召回"的闭环。

4.3 向量 vs GraphRAG 的适用边界

不是所有问题都要 GraphRAG——纯语义问题用向量更快更准。GraphRAG 的价值在关系密集型问题，所以查询路由（2.3 节）会把关系问题专门导向 graph 源。

五、自反思 RAG：查不好就重查

5.1 检索质量自评

前几节的检索都假设"查回来的东西是对的"。但检索会失败——query 表述不清、术语不匹配、知识库里那篇文档用了不同措辞。自反思 RAG 在检索后加一道质量评估，查得不好就重写 query 再查。

普通 RAG：  query → retrieve → generate
自反思 RAG：query → retrieve → 评估 → 不够好？→ 重写 query → retrieve' → ...
                              ↓ 够好
                          generate

5.2 query 重写策略

重写不是瞎改，是针对检索失败的原因对症下药：

# reflective_rag.py —— 自反思检索
REWRITE_PROMPT = """原始问题：{question}
检索结果（前 3 条摘要）：
{snippets}
检索结果与问题相关吗？如果不相关，重写一个更可能命中的检索 query。
如果相关，原样返回 question。只返回 query 文本，不要解释。
"""

class ReflectiveRAG:
    def __init__(self, llm, retriever, max_rewrites: int = 2):
        self._llm = llm
        self._retriever = retriever
        self._max_rewrites = max_rewrites

    def retrieve(self, question: str) -> dict:
        query = question
        for attempt in range(self._max_rewrites + 1):
            docs = self._retriever.search(query, top_k=8)
            if self._is_relevant(question, docs):
                return {"docs": docs, "query_used": query,
                        "rewrites": attempt}
            # 不相关 → 重写
            query = self._rewrite(question, docs)
        return {"docs": docs, "query_used": query,
                "rewrites": self._max_rewrites, "note": "max_rewrites_hit"}

    def _is_relevant(self, question: str, docs: list) -> bool:
        if not docs:
            return False
        snippets = "\n".join(d.summary for d in docs[:3])
        resp = self._llm.complete(
            f"问题：{question}\n结果：{snippets}\n相关吗？只回 yes/no")
        return resp.strip().lower().startswith("yes")

    def _rewrite(self, question: str, docs: list) -> str:
        snippets = "\n".join(d.summary for d in docs[:3])
        return self._llm.complete(
            REWRITE_PROMPT.format(question=question, snippets=snippets))

5.3 与第 9 篇经验学习的对接

重写不是每次从零想——第 9 篇的经验库可以沉淀"哪种重写模式有效"：

经验条目示例：
  场景：query 含 "K8s" 召回为空
  有效重写：替换为 "Kubernetes"
  命中率提升：+41 个百分点
  → 下次见到 "K8s" 直接替换，不用 LLM 现想

自反思的副产品是经验数据的持续积累——每次重写是否有效都是一条样本，喂回第 9 篇的经验库，让重写策略越用越准。

六、实测数据

在某企业内部知识库（1.2 万篇文档 + 3800 实体 / 1.1 万关系边）上实测 200 个问题（含 60 个多跳、40 个关系类、100 个单跳）：

两个值得说的点：

① 单跳问题别用 Agentic RAG：规划 + 自评的延迟开销 ~1.3s，
   对单跳问题准确率没提升（92%→93%），纯属浪费。
   → 查询路由第一步就判：简单问题走单跳快通道，复杂问题才走 Agentic。
② 延迟 1.8s 在"Agent 任务流"里可接受：
   Agent 一个任务本来就要多步推理 + 工具调用，秒级是常态。
   用 1.8s 换 54 个百分点的多跳准确率，ROI 极高。

七、实施难度评估与落地建议

7.1 难度评估

7.2 四步落地路线

第一步（1 周，收益最大）：
  上查询规划器 + 查询路由
  —— 不做多跳、不建图，只把"复杂问题拆子问题 + 路由到不同源"
  多跳问题准确率就能从 34% 拉到 60% 左右，延迟代价最小。

第二步（2 周）：
  加多跳迭代检索 + sufficiency 自评
  —— 接第 23 篇 span 追踪，每跳可观测
  多跳准确率到 80%+，关键是调好 sufficiency 阈值（建议从 0.8 起）。

第三步（3-4 周，关系密集场景才做）：
  建 GraphRAG：离线实体关系抽取 + 图遍历混合检索
  —— 先小规模试点（一个业务域的文档），抽准了再扩
  关系类问题准确率到 90%+，但构建成本高，按需投入。

第四步（1 周，按需）：
  自反思重写 + 经验沉淀
  —— 接第 9 篇经验库，重写策略越用越准
  首次召回失败时的兜底，命中率提升 20+ pp。

7.3 一个判断原则收尾

不是所有问题都需要 Agentic RAG。判断标准很简单：如果用户的提问里出现了"对比"、“各自”、“通过”、"的…的"这类多跳信号词，或者涉及实体间关系，就走 Agentic；否则走单跳快通道。第 4 篇的单跳 RAG 没有过时——它仍然是简单问题的最优解，Agentic RAG 是它在复杂问题上的补充，不是替代。把两套用查询路由接起来，让每个问题走对的路，才是完整方案。

下一篇预告：上下文窗口是比 GPU 显存更稀缺的资源——第 25 篇把检索做深后，召回的证据怎么塞进有限窗口？历史对话怎么压缩、工具结果怎么裁剪、记忆怎么调度，"上下文工程"值得单独一篇深潜。