第 11 篇结尾预告的收官项目来了：一个本地文档问答助手。把一个文件夹的 md 与 txt 文档喂给它，启动时切块编号建库；你用中文提问，它先调检索工具找相关片段，再作答并标注出处；多轮追问不必重复检索，上一轮的结果还在历史里；知识库前缀挂上缓存断点，整场会话的输入大头按一折计费。全部代码约一百行，前 11 篇的零件在这里各就各位。
本文是《Anthropic SDK 从入门到实战》第 12 篇收官实战，环境沿用第 1 篇装好的 venv 与 2.3.0 版 SDK。全文按"架构 → 预处理 → 检索工具 → 主循环 → 缓存断点 → 运行实录 → 对照与扩展 → 总结"组织，每个模块都回扣对应篇目。
1、需求与整体架构
先把需求钉死。助手要具备五条能力：
- 加载本地 md 与 txt 文档，切块并记录出处
- 回答前先检索，答案必须标注来源
- 支持多轮追问，历史滚雪球不重查
- 知识库前缀走缓存，成本逐轮可查
- 回答流式输出，逐字上屏
运行形态刻意保持朴素：一个脚本加一个 docs 目录，终端里一问一答，没有数据库也没有常驻服务。启动分两步——预处理跑一次产出索引，之后每问只花检索与生成的钱；文档更新就重跑预处理，问答代码一行不动。智能环节全部交给模型：何时检索、用什么关键词、怎么把碎片组织成连贯回答、引用哪条出处，都由它临场决定，你的代码只提供工具与规矩。
第一个设计决策是正文不塞 system。第 7 篇的实战把十万 token 手册整个放进 system 吃缓存，对固定的小知识库是最优解；但文档一多，全量前缀既贵又不精准。本项目换成分工：目录（每篇文档的块数与开头摘要）进 system 当稳定前缀，正文切块留在本地，模型按需调检索工具去取——缓存管"目录与规则不变"，工具管"正文按需取"，两个机制互不抢活。第二个简化是检索本身：不引向量库，关键词计分排序即可命中，零依赖跑得动，向量化的升级路径放在第 7 节。
整体架构与数据流如下：
解读：左框启动时跑一次，产出块列表与 system 目录；右框每一问转一圈，检索与否由模型自行决定。连接两框的那条线是本项目的成本关键——目录挂了缓存断点，左框的产出在整场会话里只按一折重复计费，文档换一批也只需重跑一次预处理。
2、文档预处理与切块
预处理要产出两样东西：带出处的块列表，以及给 system 用的目录。实现如下：
import os
from dataclasses import dataclass
@dataclass
class Chunk:
source: str # 出处：文件名
index: int # 文件内块序号
text: str # 块正文，约 600 字
def load_chunks(folder, size=600):
chunks = []
for name in sorted(os.listdir(folder)):
if not name.endswith((".md", ".txt")):
continue # PDF 先抽出文本存为 txt 再入库
paras = open(os.path.join(folder, name),
encoding="utf-8").read().split("\n\n")
buf, idx = "", 0
for para in paras: # 小段攒批，够 size 落一块
if buf and len(buf) + len(para) > size:
chunks.append(Chunk(name, idx, buf))
idx, buf = idx + 1, ""
buf += para + "\n\n"
if buf.strip():
chunks.append(Chunk(name, idx, buf))
return chunks
解读：块是检索与引用的最小单位，source 与 index 两个字段就是日后答案里的出处坐标；块列表交给第 3 节的检索工具，目录则留给第 5 节的 system，同一份数据两处消费。切块按空行分段、攒够 600 字左右落一块——块太大检索不精准，太小上下文破碎，500 到 800 字是文档问答的常用区间。一个已知局限：代码块内部的空行也会被当作段落边界，含大量代码的文档按标题（## 行）切分更稳，改的只是 paras 的来源。PDF 的处理刻意放在管线之外：用 pdfplumber 之类抽成纯文本存 txt，问答链路一行不用改，预处理与问答解耦，后续换抽取器互不影响。
3、检索工具定义与出处
工具定义沿用第 4 篇的三件套，schema 如下：
TOOLS = [{
"name": "search_docs",
"description": "按关键词检索本地文档库，返回片段与出处。",
"input_schema": {
"type": "object",
"properties": {"keywords": {
"type": "array", "items": {"type": "string"},
"description": "查询关键词，2 到 4 个"}},
"required": ["keywords"],
},
}]
解读：description 写清"什么时候该用我"，是模型挑工具的主要依据（第 4 篇）；keywords 用数组并建议 2 到 4 个——单个词检索太散，太多则容易一条都命中不了。执行侧是纯 Python 的计分检索：
def search_docs(keywords, chunks, top_k=3):
"""关键词计分：命中一次记一分，返回前 top_k 块。"""
scored = [(sum(c.text.count(k) for k in keywords), c)
for c in chunks]
scored = [s for s in scored if s[0] > 0]
scored.sort(key=lambda s: -s[0])
return scored[:top_k]
def execute_tool(block, chunks):
if block.name != "search_docs":
return "错误：未知工具"
hits = search_docs(block.input["keywords"], chunks)
if not hits:
return "没有检索到相关内容，请换关键词再试。"
return "\n\n".join(
f"【出处】{c.source} 第 {c.index + 1} 块\n{c.text}"
for _, c in hits)
解读：出处直接拼进 tool_result 文本——模型看到的结果自带【出处】标记，引用不靠事后补，第 5 节的 system 规则再把它约定成最终答案的固定格式。检索不到也返回可读文本而非抛异常，模型有机会换词重试，是第 4 篇 is_error 的温和版。计分检索看似简陋，对几千块以内的文档库够用：提问里的关键词与块内词面重合，覆盖了大多数查询意图；真正解决"换个说法就查不到"要靠向量检索，属于第 7 节的升级项。top_k 取 3 是精度与上下文占用的折中，检索结果太多会稀释有效信息。参数与回传的完整规矩见官方工具使用文档。
4、主问答循环与流式
主循环把第 4 篇的 while 骨架与第 2 篇的流式上下文管理器合体，client 沿用第 1 篇的环境变量初始化：
def ask(question, messages, meter):
messages.append({"role": "user", "content": question})
while True:
with client.messages.stream( # 流式管理器：逐字上屏
model="claude-sonnet-5-5", max_tokens=1024,
system=SYSTEM, tools=TOOLS, messages=messages,
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
resp = stream.get_final_message() # 落回完整消息
meter(resp.usage) # 每次请求都记账，含检索轮
messages.append({"role": "assistant",
"content": resp.content})
if resp.stop_reason != "tool_use": # 第 4 篇的出口
return resp
results = []
for b in resp.content:
if b.type == "tool_use":
print(f" [检索关键词：{b.input['keywords']}]")
results.append({"type": "tool_result",
"tool_use_id": b.id,
"content": execute_tool(b, chunks)})
messages.append({"role": "user", "content": results})
解读：get_final_message() 把增量装配回完整对象——content 块数组、stop_reason、usage 一应俱全，工具循环拿它判断出口，与非流式写法无缝衔接（第 2 篇），事件语义见官方流式文档。一个容易疑惑的点：工具轮里 text_stream 常常只流出半句"我查一下"甚至什么都不出——那一轮的主角是 tool_use 块，增量装配照样完整，不必期待正文上屏。assistant 回复整条入史，text 与 tool_use 同居其中（第 5 篇）；tool_result 装进 user 消息数组、tool_use_id 一一对应（第 4 篇）。照旧两句：max_tokens 必填，采样参数已在 2026-10-21 API 版本移除，参数全集见官方 Messages API 参考。生产加固交给第 11 篇那套：SDK 默认 max_retries=2 已扛住 429 与瞬时 5xx，外层只需按异常层级兜住 APIError、记下 request-id 即可。
5、知识库前缀与缓存断点
system 的构造把目录、规则与缓存断点打包在一起：
def build_system(chunks):
"""目录与规则进 system：稳定前缀，正文留给检索工具。"""
files = {}
for c in chunks:
files.setdefault(c.source, []).append(c)
catalog = "\n".join(
f"- {name}：共 {len(cs)} 块，开头「{cs[0].text[:30]}…」"
for name, cs in files.items())
rules = ("你是文档问答助手。作答前先用 search_docs 检索，"
"引用一律标注【文件名 第 N 块】，检索不到就如实说明，"
"不要编造文档内容。")
return [{"type": "text",
"text": f"文档库目录：\n{catalog}\n\n{rules}",
"cache_control": {"type": "ephemeral"}}] # 第 7 篇断点
解读：第 7 篇的布局纪律在这里兑现——稳定内容在前、断点钉在稳定段尾部。渲染顺序是 tools、system、messages，断点放在 system 最后一块，会把工具定义连同目录规则一起缓存；反过来说，TOOLS 与目录在会话中必须一个字都不改，改了前缀缓存全部作废。两个门槛也要记着：可缓存前缀最少 512 token，目录太短会静默不缓存；缓存默认 5 分钟过期，每次读取刷新计时器，连续追问天然保活。历史是断点之后唯一增长的部分，多轮会话连历史也想吃缓存，可用第 7 篇提过的 SDK 顶层 cache_control 让断点随对话后移；本项目提问短、历史轻，一个 system 断点已覆盖输入大头。细则见官方提示缓存文档。
提醒：重建索引或改写规则请安排在会话之间做。会话中途换目录，等于当场击穿全部缓存，下一次提问按全价重写前缀。
6、会话统计与运行实录
会话驱动把第 9 篇的四项计费公式挂到每次请求上：
PRICE = (2.00, 10.00, 0.20, 2.50) # 输入、输出、缓存读、缓存写
session = {"usd": 0.0, "read": 0}
def meter(usage):
p_in, p_out, p_read, p_write = PRICE
session["usd"] += (usage.input_tokens * p_in
+ usage.cache_read_input_tokens * p_read
+ usage.cache_creation_input_tokens * p_write
+ usage.output_tokens * p_out) / 1e6
session["read"] += usage.cache_read_input_tokens
chunks = load_chunks("docs")
SYSTEM, messages = build_system(chunks), []
for q in ("deploy.py 上传前会做哪些校验？",
"MD5 对不上时具体会发生什么？"):
print(f"\n你：{q}")
before = session["read"]
ask(q, messages, meter) # 历史滚雪球，追问可直答
print(f"[本轮缓存读 {session['read'] - before} token，"
f"累计 {session['usd']:.4f} 美元]")
一次典型运行的两轮问答实录如下：
你：deploy.py 上传前会做哪些校验？
[检索关键词：['deploy.py', '校验', 'MD5']]
deploy.py 上传前依次做三步校验：本地组装 standalone 包并剔除其中
的 .env、上传后逐文件比对 MD5、解压前校验包完整性。任何一步
不过，部署都会中止并打印失败文件清单。
出处：【运维手册.md 第 3 块】
[本轮缓存读 1530 token，累计 0.0070 美元]
你：MD5 对不上时具体会发生什么？
MD5 比对不一致时脚本立即中止，不会解压覆盖线上目录，并逐行列出
哈希不一致的文件，提示重新构建后再部署。
出处：【运维手册.md 第 3 块】
[本轮缓存读 1954 token，累计 0.0091 美元]
解读：三处行为值得看。第一问触发一次检索，检索往返加最终作答共两次请求，都进了账；第二问没有再出现检索行——上一轮的 tool_result 连同出处仍在历史里，模型直接引用作答，这正是多轮追问的价值。缓存读从 1530 涨到 1954，目录加旧历史整段按一折计费，只有新增问答付全价；两问合计不到一美分，文档量再翻几倍，这个成本结构也不变。对照不开缓存的口径更直观：目录与历史每轮全价重发，同样两问账目约翻四倍——第 7 篇那张对照表在文档场景的复现。session 里的两个数字顺手打进日志，就是第 11 篇可观测三件套里"usage 落账"的最小形态。若两问间隔超过 5 分钟，缓存过期，下一问会多付一次写入溢价，属预期行为。
7、知识点对照与扩展方向
十二篇的知识点在本项目里的落点对照如下：
| 篇目 | 核心知识点 | 本项目落点 |
|---|---|---|
| 第 1 篇 首次调用 | system 独立、max_tokens 必填 | 每次请求的固定参数 |
| 第 2 篇 流式输出 | stream 上下文管理器 | 回答逐字上屏 |
| 第 3 篇 结构化输出 | parse 与 Pydantic 模型 | 练习的置信度字段 |
| 第 4 篇 工具调用 | tool_use 与 tool_result 循环 | search_docs 往来 |
| 第 5 篇 上下文管理 | 历史滚雪球与裁剪 | 多轮追问入史 |
| 第 6 篇 扩展思考 | adaptive 与 effort 档位 | 扩展方向之一 |
| 第 7 篇 提示缓存 | cache_control 断点 | 目录稳定前缀 |
| 第 8 篇 Batch 批处理 | 异步排队五折 | 扩展方向之一 |
| 第 9 篇 选型与成本 | 四项计费公式 | meter 逐请求落账 |
| 第 10 篇 MCP 接入 | connector 与 stdio 两路线 | fetch 扩展方向 |
| 第 11 篇 错误处理 | max_retries 与异常层级 | 生产加固清单 |
| 第 12 篇 本篇 | 全部零件组装 | 文档问答助手 |
表里三行的落点只写着"扩展方向"，它们是下一轮迭代的事，各一句指篇：
- 批量问答进 Batch：问题清单打包提交，token 一律五折，Haiku 更省（第 8 篇）
- 外部资料上 MCP：挂官方 fetch server，检索不到时抓网页补充（第 10 篇）
- 疑难升 effort：提到 high 重试一轮，代价与缓存击穿账见第 6、7 篇
三个方向都不动主循环，只在请求参数与服务进程上做加法——这正是分层架构的回报。
8、总结
收官项目把 12 篇零件装成一台整机：预处理切块记出处，目录进 system 挂缓存断点，正文留在检索工具后面按需取；主循环是工具 while、流式管理器与逐请求记账的三合体；行为上拿到三样东西——答案带出处、追问直答不重查、输入大头一折计费。约一百行代码，没有一个新 API，全部是前 11 篇讲过的零件换了种组合方式。能独立把这台整机搭起来，就说明十二篇的知识点不只是看过，而是真的在手上了。
留三个练习：其一，把最终作答改成 client.messages.parse()，定义 answer、sources、confidence 三字段的 Pydantic 模型，给答案加置信度字段，低于阈值提示人工复核——第 3 篇的 parse 在这里接上；其二，给会话历史加上第 5 篇的滑动窗口裁剪，验证 tool_use 与 tool_result 的配对不被截断点拆开；其三，把二十个常见问题用 Batch 跑一遍 FAQ 预生成，与在线问答的账单对照。对于手头攒着一堆文档想直接问答的开发者，建议今天就照第 2 节把文档切块跑起来——零件全是现成的，组装不过是下午的事。系列到此收官，十二篇从第一次调用走到完整项目，这条主线值得从头再翻一遍。
