10.2 索引与检索工具
10.1 把文档切成了 217 个块，但块躺在一个列表里，模型看不见也摸不着。要让 Agent 用上它们，得把"检索"这件事包成工具——模型像调用任何其他工具一样，发出一个 search 请求，拿回几段候选，再决定要不要 read 全文。
本节把 minagent 推进到
minagent/rag/tools.py：把块列表建成索引，包出search与read两个工具，注册进第 4 章的ToolRegistry，并跑通一条 search → read → search → read → 综合的多跳检索轨迹。
10.2.1 检索作为工具，而不是检索算法
先划清边界。第 9 章已经讲过本地嵌入、numpy 向量索引、混合检索——那些是检索算法。本节不重复它们，而是问一个不同的问题：怎么把检索能力交给模型，让它自己决定搜什么、读什么、读几段？
这两件事是正交的：
| 关注点 | 第 9 章（算法） | 本章（工具） |
|---|---|---|
| 核心问题 | 怎么算相关度 | 怎么让模型驱动检索 |
| 输入 | query 向量 | 模型发出的 ToolCall |
| 输出 | 排序好的块 | 模型可读的 JSON 字符串 |
| 谁决定检索次数 | 代码写死 | 模型，按需多跳 |
| 失败模式 | 召回不准 | 模型搜错、读偏、循环不停 |
把检索做成工具，最大的收益是多跳。单轮 RAG 是"一搜一答"，问题稍复杂就答不好；工具化的检索让模型能"先搜 A、读到线索后再搜 B"，像人查资料一样逐步逼近答案。本节的重点全在这个"让模型自主检索"的闭环上。
10.2.2 建索引：块列表 → 可检索
10.1 的产物是 list[(doc_id, chunk_id, text)]。检索之前，要先把它变成"能打分的东西"。本节的打分很朴素，只用到两个信号：
- 关键词分数：query 的词在块里出现了几次（词频）。
- 哈希向量相似度：把 query 和块各自哈希成一个定长向量，算余弦相似度。
先定义分词。中文没有空格，用字符级 bigram 是廉价而稳的近似；英文则按单词切：
import re
def tokens(text: str) -> list[str]:
text = text.lower()
words = [w for w in re.findall(r"[a-z0-9_]+", text) if len(w) >= 2]
cjk = re.findall(r"[一-鿿]", text)
bigrams = ["".join(p) for p in zip(cjk, cjk[1:])]
return words + bigrams
tokens("GIL 全局解释器锁") 会得到 ["gil", "全局", "局解", "解释", "释器", "器锁"]——英文单词加上中文 bigram。bigram 的好处是不需要词典，任何中文串都能切成可比较的单元。
哈希向量把每个 token 映射到固定维度的桶里，是一种不依赖嵌入模型的"穷人的向量"：
import hashlib, math
DIM = 256
def vec(text: str) -> list[float]:
v = [0.0] * DIM
for t in tokens(text):
h = int(hashlib.md5(t.encode()).hexdigest(), 16) % DIM
v[h] += 1.0
return v
def cosine(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
return dot / (na * nb) if na and nb else 0.0
哈希向量的核心是"同一个 token 永远落进同一个桶"，于是词形相近的两段文本会在向量空间里靠近。它抓不住语义（“汽车"和"轿车"不会靠近），但足够让本节演示"混合打分"这件事，且完全确定、零依赖。真正要做语义检索，把 vec() 换成第 9 章的嵌入即可，工具层不用改一行。
本节选关键词 + 哈希向量这条最朴素的路，不是因为它最好，而是因为它完全确定——同样的 query 永远返回同样的排序，第 16 章的轨迹断言才写得出来。三条路的对照：
| 方案 | 依赖 | 适合 |
|---|---|---|
| 本节：关键词 + 哈希向量 | 零 | 教学、可断言 |
| 第 9 章：本地嵌入 + numpy | numpy | 语义检索 |
| 生产：向量数据库（pgvector 等） | 重 | 大规模、持久化 |
升级检索算法时，工具层的接口（search/read 的签名与返回结构）一行都不用改——这正是把"算法"和"工具"分开的价值。
10.2.3 search(query)：混合打分
有了分词和向量，search 就是把两个分数加起来排序：
CHUNKS: list[tuple[str, int, str]] = load_corpus("content/posts/python")
_VECS = [(d, c, vec(t)) for d, c, t in CHUNKS]
def kw_score(qtoks: list[str], text: str) -> float:
tt = tokens(text)
return float(sum(tt.count(t) for t in qtoks))
def search(query: str, k: int = 3) -> list[dict]:
qtoks = tokens(query)
qv = vec(query)
scored = []
for (doc_id, cid, text), (_, _, cv) in zip(CHUNKS, _VECS):
score = kw_score(qtoks, text) + 5.0 * cosine(qv, cv)
if score > 0:
scored.append((score, doc_id, cid, text))
scored.sort(key=lambda x: (-x[0], x[1], x[2]))
out = []
for score, doc_id, cid, text in scored[:k]:
out.append({"doc_id": doc_id, "chunk_id": cid,
"score": round(score, 3),
"snippet": text[:60].replace("\n", " ")})
return out
两个设计点值得说明。第一，关键词分数与向量相似度直接相加，用 5.0 把余弦（值域 0~1）放大到与词频同一量级——这是个粗糙但够用的权重。第二，sort 的 key 是 (-score, doc_id, chunk_id)：先按分数降序，分数相同时按 doc_id、chunk_id 升序。加上后两个键是为了让结果完全确定——否则同分块的相对顺序取决于字典遍历顺序，跑两次可能不一样，第 16 章的断言就无从写起。
search 返回的不是块全文，而是 snippet（前 60 字）。这是刻意的：搜索只负责"指路”，不负责"载货"。一屏里塞三段全文，既浪费上下文，又逼模型在一堆长文本里找重点。真要看全文，用下一个工具。
对 10.1 的 6 篇语料真跑一次 search("GIL 是什么", k=3)，实测返回：
[{'doc_id': 'python-concurrency-performance', 'chunk_id': 0, 'score': 9.441,
'snippet': '> 本文假设 Python 3.11+。所有代码可直接复制运行。 --- ## 1. GIL（全局解释器锁）的真相 '},
{'doc_id': 'python-concurrency-performance', 'chunk_id': 3, 'score': 6.145,
'snippet': '0.85s Multi-thread: 1.05s # 更慢！因为线程切换开销 + GIL 竞争 ``` ### '},
{'doc_id': 'python-concurrency-performance', 'chunk_id': 1, 'score': 3.778,
'snippet': '*例外：** 执行 C 扩展（如 NumPy 矩阵运算）时，C 代码可以释放 GIL。 ### 1.2 GIL 对性能'}]
排名第 0 的正是 GIL 定义那块，第 1、2 名也都来自同一篇文章——分别讲"GIL 竞争让多线程更慢"和"GIL 对性能的影响"。前三名同属一篇，说明关键词 GIL 在这篇文章里密度最高，混合打分把整个文档推了上来。这也暴露一个真实问题：同一文档霸榜会让结果缺乏多样性，工程上常用"每文档最多取 N 块"来抑制。
10.2.4 read(doc_id, chunk_id)：取回完整块
read 是 search 的配套：拿 doc_id 和 chunk_id 把完整块取回来。
def read(doc_id: str, chunk_id: int) -> str:
for d, c, t in CHUNKS:
if d == doc_id and c == chunk_id:
return t
return json.dumps({"error": f"no chunk {doc_id}#{chunk_id}"},
ensure_ascii=False)
找不到时返回 {"error": ...} 而不是抛异常，这是第 4 章就立下的约定：工具出错要变成模型可读的反馈。模型看到一个 no chunk ... 的错误，会知道刚才的 chunk_id 编错了，可以重新 search 一次——而不是让整个循环崩掉。
search 与 read 的分工，本质是"索引"与"原文"的分工：
| 工具 | 返回 | 用途 | 上下文成本 |
|---|---|---|---|
search | 若干块的 60 字片段 + 坐标 | 定位候选 | 低 |
read | 一个块的完整 400 字 | 精读证据 | 高 |
模型先 search 拿到坐标，再对少数几个坐标 read。这套"两段式"是 RAG 工具的标准形态——让廉价的搜索承担试错，让昂贵的读取只花在值得的地方。
10.2.5 注册进 ToolRegistry
把两个函数包成 Tool，加进第 4 章的注册表：
import json
from minagent.tools import Tool, ToolRegistry
def search_tool(query: str, k: int = 3) -> str:
return json.dumps(search(query, k), ensure_ascii=False)
registry = ToolRegistry()
registry.add(Tool(
name="search",
description="在本地文档库中按关键词与语义混合检索，返回命中块（含 doc_id、chunk_id、片段）。",
parameters={"type": "object",
"properties": {"query": {"type": "string"},
"k": {"type": "integer"}},
"required": ["query"], "additionalProperties": False},
func=search_tool))
registry.add(Tool(
name="read",
description="按 doc_id 与 chunk_id 取回某个块的完整文本。",
parameters={"type": "object",
"properties": {"doc_id": {"type": "string"},
"chunk_id": {"type": "integer"}},
"required": ["doc_id", "chunk_id"], "additionalProperties": False},
func=read))
注意 search_tool 多包了一层 json.dumps：ToolRegistry.call 对非字符串结果会自动 json.dumps，但这里显式序列化，是为了保证 ensure_ascii=False——中文 doc_id 不被转义成 \uXXXX，模型读起来更省 token。两个工具的 description 都写清了"什么时候用它"，这正是第 4 章强调的：工具描述是提示词的一部分。
注册完，registry.schemas() 就把这两条喂给模型。从此模型的"手"里多了检索能力。
10.2.6 真跑一条多跳轨迹
光看代码看不出多跳的价值，得真跑一条。用假模型脚本化一次两跳检索：先搜 GIL、读一块，再搜 asyncio、读一块，最后综合。
script = [
[ToolCall("t1", "search", {"query": "GIL 全局解释器锁", "k": 2})],
[ToolCall("t2", "read", {"doc_id": "python-concurrency-performance",
"chunk_id": 0})],
[ToolCall("t3", "search", {"query": "asyncio 事件循环 并发", "k": 2})],
[ToolCall("t4", "read", {"doc_id": "python-concurrency-performance",
"chunk_id": 9})],
"综合两段材料：GIL 让 CPython 同一时刻只跑一个线程，"
"因此 CPU 密集用多进程；I/O 密集用 asyncio 事件循环。",
]
messages, steps, final = run(FakeModel(script), registry,
"Python 里的并发有哪些方案？GIL 有什么影响？")
真跑（Python 3.14.6）后的轨迹：
[0] tool_call: search({'query': 'GIL 全局解释器锁', 'k': 2}) -> [{"doc_id": "python-concurrency-performance", ...
[1] tool_call: read({'doc_id': 'python-concurrency-performance', 'chunk_id': 0}) -> > 本文假设 Python 3.11+...
[2] tool_call: search({'query': 'asyncio 事件循环 并发', 'k': 2}) -> [{"doc_id": "python-concurrency-performance", ...
[3] tool_call: read({'doc_id': 'python-concurrency-performance', 'chunk_id': 9}) -> o.CancelledError: ...
[4] final: 综合两段材料：GIL 让 CPython 同一时刻只跑一个线程，因此 CPU 密集用多进程；I/O 密集用 asyncio 事件循环。
第一跳 search("GIL 全局解释器锁") 命中了 python-concurrency-performance#0——那正是文章开头讲 GIL 定义的那一块。read 把它完整取回后，模型（脚本）据此发出第二跳 search("asyncio 事件循环 并发")，命中 #9（asyncio 最佳实践那块），再 read，最后综合。
轨迹本身就是证据。把每一步的入参和出参打印出来，就能验证：命中块对不对、read 的坐标是不是来自上一步的 search、模型有没有在拿到材料后才收尾。这正是卷一坚持用假模型的原因——真实模型会把这些决策藏在一次响应里，而假模型把决策摊开成可断言的步骤。
多跳的价值也在这条轨迹里：如果只有单轮 RAG，模型只能拿第一跳的 GIL 材料作答，答不出 asyncio 那半句。有了 search/read 工具，它能把一个问题拆成"先搞清 GIL，再搞清 asyncio"两步，每步都取到真正相关的块。
一个容易忽略的细节：两次 search 返回的候选里，#0 和 #9 都来自同一篇文章。真实检索常常这样——同一文档的多个块命中同一问题。所以 read 的坐标不能只记 chunk_id，必须带 doc_id，否则"第 0 块"这种坐标会在不同文档间串味。
10.2.7 检索预算：别让模型无限搜下去
工具化检索把"搜几次"的控制权交给了模型，好处是灵活，风险是可能停不下来。一个没有约束的 Agent 可能：反复搜同一个词、read 一堆不相关的块、在"再搜一次"里打转。两种兜底手段：
- 步数上限：第 3 章的
run(..., max_steps=8)已经给了硬上限，超过就强制收尾。 - 检索预算：给
search/read各记一个调用计数，超过阈值就把工具从schemas()里摘掉，逼模型用现有材料作答。
| 约束 | 位置 | 效果 |
|---|---|---|
max_steps | 循环层 | 硬性终止，可能截断 |
| 工具计数 | 工具层 | 精确控制检索量 |
| 结果去重 | search 内 | 避免重复块占额度 |
卷一只做 max_steps 这一条，因为它最简单、最确定。检索预算和去重留到卷二的生产化加固。到这里，10.2 给了 minagent 一双"会查资料的手"；但这双手查回来的东西可不可信，还没人验——那是 10.3 的事。
真实模型下会怎样：本节的多跳是脚本写死的——假模型不会自己决定"先搜什么、再搜什么"，那四个 ToolCall 是我们预先排好的。真实 Claude 拿到 search/read 两个工具的 schema 后，会自己规划检索路径：先搜一个宽泛的词，看到片段后判断"还缺什么"，再发一个更精确的 search，或直接 read 命中的块。它甚至可能一轮里并发发出多个 search（第 4.3 节的并行调用）。但这带来两个真实问题：一是检索次数不可控，模型可能反复搜同一个词空转，需要设一个 max_steps 或"检索预算"兜底；二是命中质量依赖工具描述，description 写得含糊时，模型会搜错词或读了不相关的块。这些在卷一只做"能跑通、能断言"，策略化的部分留到卷二。
阅读导航：上一节：10.1 文档加载与切分 · 下一节：10.3 带引用的回答 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。