Retrieval Loop：查到够为止

旅游助手只搜一次，如果 query 不好，答案就完了。用户问"杭州茶文化轻松步行路线"，但 query 变成"杭州景点"，检索回来的是泛泛的热门景点列表。

Retrieval Loop 不是完整的 Agent。它在 RAG 上加一个小循环：提 query → 检索 → 判断证据够不够 → 够就回答，不够就改 query 再查。

一句话

Retrieval Loop 把"一次检索就回答"变成"query → 检索 → 判断 → 回答或重查"，让简单 RAG 能修复弱检索。

它修什么失败

问题	表面看起来	实际风险
只检索一次	快	query 不好等于答案不好
没有证据充分性检查	有 context	模型可能在弱证据上硬答
检索结果没记录	有引用	难以复现为什么用了这些文档

它引入什么复杂度

谁	负责什么
模型	提 query 和判断 done/not done
检索器	返回片段和 doc id
Python	控制轮次、存命中、写 trace

它比 Agentic RAG 窄：循环里唯一的动作就是检索。

和其他模式的关系

flowchart LR
  RAG["RAG（一次检索）"] -->|"一次不够"| RL["Retrieval Loop"]
  RL -->|"需要多种工具"| ARAG["Agentic RAG"]
  RL -.->|"长报告按章节查"| STM["STORM"]

RAG：Retrieval Loop 是 RAG 的升级版。如果一次检索够用，不需要 loop。
Agentic RAG：如果除了检索还需要其他工具（天气、订票、计算），就超出了 Retrieval Loop 的范围。
STORM：长报告按章节分别检索，是 Retrieval Loop 的结构化应用。

完整实现：旅游信息多轮检索

"""retrieval_loop_travel.py — 查到够为止"""
from __future__ import annotations

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


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

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

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 = _tokenize(query)
        if not terms:
            return []
        scored = []
        for doc in self._docs:
            doc_tokens = _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]


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


# ── Retrieval Loop 核心逻辑 ─────────────────────────────

@dataclass(frozen=True)
class RetrievalResult:
    answer: str
    evidence: list[SearchResult]

@dataclass(frozen=True)
class Decision:
    done: bool
    answer: str


def retrieval_loop(
    model: MockLLM,
    *,
    question: str,
    index: SimpleSearchIndex,
    max_rounds: int = 3,
    top_k: int = 3,
) -> RetrievalResult:
    """
    query → 检索 → 判断 → 回答或重查。
    """
    evidence: list[SearchResult] = []
    seen: set[str] = set()
    notes = ""

    for r in range(max_rounds):
        # 模型提 query
        raw_query = model.complete([
            Message(role="system", content="提出下一个检索 query。只返回 JSON。"),
            Message(role="user", content=f"问题：\n{question}\n\n已有笔记：\n{notes}"),
        ])
        query = json.loads(raw_query)["query"]
        print(f"[轮次 {r + 1}] query: {query}")

        # 检索
        results = index.search(query, k=top_k)
        for res in results:
            if res.doc.doc_id not in seen:
                seen.add(res.doc.doc_id)
                evidence.append(res)
        notes = "\n".join(
            f"- ({res.doc.doc_id}, score={res.score}) {res.doc.text[:80]}"
            for res in evidence
        )
        print(f"  命中 {len(results)} 条，累计证据 {len(evidence)} 条")

        # 判断够不够
        raw_decision = model.complete([
            Message(role="system", content="判断证据是否充分。只返回 JSON。"),
            Message(role="user", content=f"问题：\n{question}\n\n证据：\n{notes}"),
        ])
        dec = json.loads(raw_decision)
        print(f"  done={dec['done']}")

        if dec["done"]:
            return RetrievalResult(answer=dec["answer"], evidence=evidence)

    # 到上限，用最后的证据硬答
    fallback = model.complete([
        Message(role="system", content="用已有证据回答。"),
        Message(role="user", content=f"问题：\n{question}\n\n证据：\n{notes}"),
    ])
    return RetrievalResult(answer=fallback, evidence=evidence)


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

def main() -> None:
    docs = [
        Document("tea_museum", "中国茶叶博物馆，龙井路 88 号，9:00-17:00，周一闭馆，免费。"),
        Document("tea_village", "龙井村产龙井茶，可体验采茶，步行友好，约 2 小时游览。"),
        Document("silk_museum", "中国丝绸博物馆，玉皇山路 73 号，室内展馆，免费，适合雨天。"),
        Document("west_lake", "西湖全天开放，免费，环湖步行约 3 小时。"),
    ]
    index = SimpleSearchIndex(docs)

    model = MockLLM([
        # 轮次 1：query 太宽泛
        '{"query":"杭州景点"}',
        # 轮次 1：不够，缺茶文化和步行信息
        '{"done": false, "answer": ""}',
        # 轮次 2：更具体的 query
        '{"query":"杭州茶文化步行"}',
        # 轮次 2：够了
        json.dumps({
            "done": True,
            "answer": (
                "推荐路线：上午龙井村采茶体验（步行友好，约 2 小时）→ "
                "中午茶叶博物馆（免费，9:00-17:00）。"
                "如下雨可转丝绸博物馆。[tea_village] [tea_museum] [silk_museum]"
            )
        }, ensure_ascii=False),
    ])

    result = retrieval_loop(model, question="杭州茶文化轻松步行路线", index=index, max_rounds=3)
    print(f"\n回答：\n{result.answer}")
    print(f"证据来源：{[r.doc.doc_id for r in result.evidence]}")


if __name__ == "__main__":
    main()

运行日志

[轮次 1] query: 杭州景点
  命中 3 条，累计证据 3 条
  done=False
[轮次 2] query: 杭州茶文化步行
  命中 2 条，累计证据 4 条
  done=True

回答：
推荐路线：上午龙井村采茶体验（步行友好，约 2 小时）→ 中午茶叶博物馆（免费，9:00-17:00）。如下雨可转丝绸博物馆。[tea_village] [tea_museum] [silk_museum]
证据来源：['tea_museum', 'west_lake', 'tea_village', 'silk_museum']

第一轮 query 太宽泛，模型判断证据不够，第二轮换了更具体的 query，拿到茶文化相关的文档后回答。

调试：trace 里看什么

要看的	怎么看	说明什么
每轮 query	trace 里的 query 字段	query 没变说明模型没学会改写
每轮新增证据	`added` 数	连续两轮 0 新增 = 停滞
`done` 判断	第几轮 done=true	太早 done 说明充分性检查太松
最终证据列表	evidence 的 doc_id	如果答案引用了不在列表里的来源 = 幻觉

工程笔记

成本公式

每轮 = 1（提 query）+ 1 次检索 + 1（判断）= 2 次模型调用 + 1 次检索
总调用 = 轮次 × 2
最好 = 2（第 1 轮就够）
最坏 = max_rounds × 2 + 1（fallback）

常见坑

坑	现象	修法
query 原地打转	多轮 query 几乎一样	加停滞检测（连续 0 新增就停）
假引用	引用了不相关的文档	强制答案只从命中文档生成
context 膨胀	每轮所有结果都追加	只保留 top-k 和摘要
query 太宽泛	只拿到通用文档	要求模型输出具体的 query 字段
永远不 done	充分性检查太严	加轮次上限 + fallback

什么时候用

一次检索经常不够。
query 需要改写。
你想 trace 每轮检索。
完整的 Agent loop 太重。

什么时候别用

一次检索就够。
检索质量差，多查只加噪音。
需要多种工具，那是 Agentic RAG 或 ReAct。
没有轮次上限。

读完以后

Retrieval Loop 是 RAG 里的小循环。

如果检索只是多种工具之一，读 Agentic RAG。如果要写长报告按章节检索，读 STORM。如果一次检索就够，停在 RAG。