Agentic RAG：让 Agent 决定怎么搜

基础 RAG 的流程是：收到问题 → 检索一次 → 把结果塞进 prompt → 回答。

旅游规划往往不够：要查天气、搜雨天室内茶馆、估路线时间，而且可能要改 query 重查。你还想知道每条结论的证据从哪来。

Agentic RAG 把检索工具和证据账本放进 Agent 循环。

一句话

Agentic RAG 把一次性 RAG 变成"检索作为 Agent 动作"，让模型决定查什么、什么时候查、证据够不够、哪些证据进账本。

它修什么失败

问题	表面看起来	实际风险
只搜一次	快	query 不好等于答案不好
检索和回答绑死	流程简单	不能根据新证据再搜
没有证据账本	有引用	断言和证据可能对不上

它引入什么复杂度

谁	负责什么
模型	选 `search` 或 `final`
检索工具	返回 doc id + 片段
证据账本（Evidence Ledger）	存真正使用的证据
Python	执行动作、去重、限制轮次、trace

检索到的文本不是可信指令，是待检查的证据。

和其他模式的关系

flowchart LR
  RL["Retrieval Loop"] -->|"检索之外还需要工具"| ARAG["Agentic RAG"]
  ARAG -.->|"事实验证"| COVE["CoVe"]
  ARAG -.->|"长报告"| STM["STORM"]
  ARAG -.->|"和 ReAct 同构"| REACT["ReAct"]

Retrieval Loop：循环里只有检索；Agentic RAG 的 Agent 还能做其他动作。
ReAct：Agentic RAG 本质是"ReAct + 检索工具 + 证据管理 + 停止条件"。
CoVe：搜完回答后，事实断言仍可能需要逐条验证。
STORM：长文章按章节检索和写作，是 Agentic RAG 的结构化扩展。

完整实现：旅游规划的 Agentic RAG

"""agentic_rag_travel.py — Agent 驱动的检索 + 证据账本"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any


# ── 极简运行时 ──────────────────────────────────────────

@dataclass(frozen=True)
class Message:
    role: str
    content: str
    name: str | None = None

class MockLLM:
    def __init__(self, responses: list[str]):
        self._responses = list(responses)
        self._idx = 0
    def complete(self, messages: list[Message]) -> str:
        if self._idx >= len(self._responses):
            raise RuntimeError("MockLLM 回复用完了")
        out = self._responses[self._idx]
        self._idx += 1
        return out

@dataclass(frozen=True)
class Document:
    doc_id: str
    text: str

@dataclass(frozen=True)
class SearchResult:
    doc: Document
    score: int

class SimpleSearchIndex:
    def __init__(self, docs: list[Document]):
        self._docs = docs
    def search(self, query: str, k: int = 3) -> list[SearchResult]:
        terms = self._tokenize(query)
        if not terms:
            return []
        scored = []
        for doc in self._docs:
            doc_tokens = self._tokenize(doc.text)
            score = sum(1 for t in terms for d in doc_tokens if t == d)
            if score > 0:
                scored.append(SearchResult(doc=doc, score=score))
        scored.sort(key=lambda r: r.score, reverse=True)
        return scored[:k]
    @staticmethod
    def _tokenize(text: str) -> list[str]:
        tokens, buf = [], ""
        for ch in text.lower():
            if ch.isalnum() or '\u4e00' <= ch <= '\u9fff':
                buf += ch
            else:
                if buf:
                    tokens.append(buf)
                    buf = ""
        if buf:
            tokens.append(buf)
        return tokens


# ── 证据账本 ────────────────────────────────────────────

class EvidenceLedger:
    """去重存储检索到的证据。"""
    def __init__(self):
        self._evidence: list[SearchResult] = []
        self._seen: set[str] = set()

    def add(self, results: list[SearchResult]) -> int:
        added = 0
        for r in results:
            if r.doc.doc_id not in self._seen:
                self._seen.add(r.doc.doc_id)
                self._evidence.append(r)
                added += 1
        return added

    def notes(self) -> str:
        return "\n".join(
            f"- [{r.doc.doc_id}] (score={r.score}) {r.doc.text[:80]}"
            for r in self._evidence
        )

    def doc_ids(self) -> list[str]:
        return [r.doc.doc_id for r in self._evidence]


# ── Agentic RAG 核心逻辑 ───────────────────────────────

def agentic_rag(
    model: MockLLM,
    *,
    question: str,
    index: SimpleSearchIndex,
    max_steps: int = 6,
) -> dict[str, Any]:
    """
    Agent 循环：
    - 动作类型 "search"：调用检索工具，结果写入证据账本
    - 动作类型 "final"：返回最终回答 + 引用
    """
    ledger = EvidenceLedger()

    system_prompt = (
        "你是一个 Agentic RAG 系统。\n"
        "- 用 search 工具收集证据，然后再回答。\n"
        "- 优先多次针对性搜索，避免一次模糊搜索。\n"
        "- 回答时引用 doc_id，如 [doc_id]。\n"
        "- 只返回 JSON 动作。\n\n"
        '搜索动作：{"type":"tool","tool":"search","args":{"query":"...","k":3}}\n'
        '最终回答：{"type":"final","answer":"..."}'
    )

    messages: list[Message] = [
        Message(role="system", content=system_prompt),
        Message(role="user", content=question),
    ]

    for step in range(max_steps):
        raw = model.complete(messages)
        action = json.loads(raw)

        if action["type"] == "final":
            print(f"[步骤 {step + 1}] final")
            return {
                "answer": action["answer"],
                "evidence": ledger.doc_ids(),
            }

        if action["type"] == "tool" and action["tool"] == "search":
            query = action["args"]["query"]
            k = action["args"].get("k", 3)
            results = index.search(query, k=k)
            added = ledger.add(results)
            observation = ledger.notes()
            print(f"[步骤 {step + 1}] search('{query}') → 命中 {len(results)}，新增 {added}")

            messages.append(Message(role="assistant", content=raw))
            messages.append(Message(role="tool", name="search", content=observation))

    return {"answer": "[到达步数上限]", "evidence": ledger.doc_ids()}


# ── 运行 ────────────────────────────────────────────────

def main() -> None:
    docs = [
        Document("weather", "杭州明天：上午晴，下午 15:00 后小雨，气温 18-25 度。"),
        Document("tea_museum", "中国茶叶博物馆，室内展馆，9:00-17:00，免费，适合雨天。"),
        Document("tea_village", "龙井村可体验采茶，户外为主，步行友好，约 2 小时。"),
        Document("silk_museum", "中国丝绸博物馆，室内，免费，雨天备选。"),
        Document("hefang", "河坊街小吃街，部分室内，步行 15 分钟从西湖可达。"),
    ]
    index = SimpleSearchIndex(docs)

    model = MockLLM([
        # 步骤 1：先查天气
        '{"type":"tool","tool":"search","args":{"query":"杭州明天天气","k":2}}',
        # 步骤 2：知道下午下雨，查室内茶文化
        '{"type":"tool","tool":"search","args":{"query":"杭州茶文化室内","k":3}}',
        # 步骤 3：够了，给最终回答
        json.dumps({
            "type": "final",
            "answer": (
                "杭州明天上午晴、下午 15:00 后小雨。"
                "建议上午去龙井村采茶（户外，约 2 小时）[tea_village]，"
                "下午转中国茶叶博物馆（室内，免费）[tea_museum]。"
                "如时间充裕，河坊街可作为傍晚小吃站 [hefang]。"
            ),
        }, ensure_ascii=False),
    ])

    result = agentic_rag(model, question="杭州明天茶文化一日游路线", index=index)
    print(f"\n回答：\n{result['answer']}")
    print(f"证据来源：{result['evidence']}")


if __name__ == "__main__":
    main()

运行日志

[步骤 1] search('杭州明天天气') → 命中 1，新增 1
[步骤 2] search('杭州茶文化室内') → 命中 3，新增 3
[步骤 3] final

回答：
杭州明天上午晴、下午 15:00 后小雨。建议上午去龙井村采茶（户外，约 2 小时）[tea_village]，下午转中国茶叶博物馆（室内，免费）[tea_museum]。如时间充裕，河坊街可作为傍晚小吃站 [hefang]。
证据来源：['weather', 'tea_museum', 'tea_village', 'silk_museum']

Agent 先查天气，发现下午下雨，第二步针对性搜"茶文化室内"，然后综合证据给出路线。

调试：trace 里看什么

要看的	怎么看	说明什么
每步动作类型	`action["type"]`	如果一上来就 final，没搜就答 = 问题
search query 的演变	连续几步的 query	应该越来越具体
证据新增数	`added`	连续 0 新增 = 停滞，应该停
最终引用	答案里的 `[doc_id]`	引用不在账本里 = 幻觉
步数	到了多少步 final	太多步说明证据不收敛

工程笔记

成本公式

每步 = 1 次模型调用 + 0 或 1 次检索
总调用 = 步数 × 1（模型）+ 检索次数
最好 = 2（1 次搜索 + 1 次 final）
最坏 = max_steps 次模型调用

context 每步增长（messages 追加），后期步骤的 token 数更大。

常见坑

坑	现象	修法
检索注入	文档内容指示模型忽略规则	把检索到的文本当不可信输入
假引用	断言不匹配引用的文档	跑断言-证据检查（CoVe）
过度搜索	很多轮，没有新证据	加预算和停滞检测
账本脏	重复或矛盾的证据	按 doc_id 去重，加来源/时间标签
context 爆炸	messages 越来越长	摘要或截断早期对话

什么时候用

问题需要多次搜索。
断言必须追溯到 doc id。
检索可能不完整，query 需要修复。
搜索行为需要 trace 和评测。

什么时候别用

一次检索够用。
不需要引用和账本。
来源不可信且没有隔离策略。
成本/延迟不支持开放循环。

读完以后

Agentic RAG 适合通过搜索收敛的知识任务。

如果要写长文章，读 STORM。如果事实断言需要逐条核查，读 CoVe。如果只需要简单的检索循环，退回 Retrieval Loop。