18.1 端到端
前 17 章，我们一节一节地给 minagent 加零件：循环、工具、记忆、检索、编排、可观测性、评测。但零件齐了不等于机器能转——它们分散在各章的示例里，从没在同一次运行里握过手。本节要做的事只有一件：把它们装到一台机器上，按下开关，看着它从头跑到尾。
本节把 minagent 推进到「综合交付」：写一个
demo_e2e.py，用假模型驱动的 Agent 走完「用户提问 → 检索（第 9/10 章）→ 工具调用（第 4 章）→ 多步循环（第 3/5 章）→ 带引用的回答（10.3）」，同时挂上 span 追踪（15.1）、成本核算（15.3）与评测（16.2），并真跑贴出完整输出。这是全卷的收束演示。
18.1.1 一条通路要串起哪些零件
先明确这次运行要回答的问题：「Python 里 CPU 密集和 I/O 密集的并发方案该怎么选？」 它天然需要检索——答案在资料库里，不在模型参数里。围绕这个问题，本次运行会用到下面这些模块：
| 环节 | 来源 | 在本次运行里的角色 |
|---|---|---|
Message / ToolCall | 第 1 章 | 对话的原子结构 |
FakeModel | 第 1.3 节 | 按脚本驱动整个流程 |
Tool / ToolRegistry | 第 4 章 | 把 search / read 暴露给模型 |
run() 循环 | 第 3 章 | 「调用→执行→回填」直到收敛 |
search / read | 第 10.2 节 | 多跳检索的手 |
| 引用校验 | 第 10.3 节 | 核对回答里的来源是否可信 |
Tracer | 第 15.1 节 | 把这次运行拍成 span 树 |
CostTracker | 第 15.3 节 | 算清这次运行花了多少 |
Case / Scorer | 第 16.2 节 | 给这条通路打分，纳入回归网 |
一张表列完，你会发现一个事实：它们没有一个是新的。端到端的价值不在于引入新能力，而在于证明这些能力可以共存、可以互相喂数据——比如追踪里的 run_id 要能对上成本表里的 run_id，评测断言的 final 要真的带着检索来的引用。
18.1.2 装配：把零件接起来
装配脚本分两段。第一段是「接线」——定义脚本、注册表，以及一个把循环、追踪、成本合成一处的 traced_run：
"""18.1 端到端：把检索、工具、循环、追踪、成本、评测串成一条通路。"""
from __future__ import annotations
import time, uuid
from minagent import (
PRICES, Case, CostTracker, FakeModel, INDEX, Message, ModelResponse, RunResult,
Scorer, Step, Tool, ToolCall, ToolRegistry, Tracer, evaluate, read, search_tool, validate,
)
QUESTION = "Python 里 CPU 密集和 I/O 密集的并发方案该怎么选？"
SYSTEM = ("你是严谨的助手。回答每个论断时在句末标注来源 [doc_id#chunk_id]，"
"只能引用 search/read 返回过的块。")
FINAL = ("CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程。[python-concurrency-performance#1]\n"
"I/O 密集型适合 asyncio，事件循环在单线程内调度协程。[python-async-advanced#0]")
def turn(tool_calls, text="", inp=0, out=0):
"""构造一轮带 usage 的响应；有工具调用即 stop_reason=tool_use。"""
return ModelResponse(
message=Message(role="assistant", content=text, tool_calls=list(tool_calls)),
stop_reason="tool_use" if tool_calls else "end_turn",
usage={"input_tokens": inp, "output_tokens": out})
def e2e_script():
return [
turn([ToolCall("t1", "search", {"query": "GIL CPU 密集 多线程", "k": 2})], inp=120, out=18),
turn([ToolCall("t2", "read", {"doc_id": "python-concurrency-performance", "chunk_id": 1})], inp=210, out=14),
turn([ToolCall("t3", "search", {"query": "asyncio I/O 密集 事件循环", "k": 2})], inp=320, out=20),
turn([ToolCall("t4", "read", {"doc_id": "python-async-advanced", "chunk_id": 0})], inp=430, out=12),
turn([], text=FINAL, inp=560, out=64),
]
def build_registry():
reg = ToolRegistry()
reg.add(Tool("search", "在本地文档库按关键词与哈希向量混合检索，返回命中块。",
{"type": "object",
"properties": {"query": {"type": "string"}, "k": {"type": "integer"}},
"required": ["query"]}, search_tool))
reg.add(Tool("read", "按 doc_id 与 chunk_id 取回某个块的完整文本。",
{"type": "object",
"properties": {"doc_id": {"type": "string"}, "chunk_id": {"type": "integer"}},
"required": ["doc_id", "chunk_id"]}, read))
return reg
def traced_run(model, registry, user_input, tracer, tracker,
model_name="claude-sonnet", max_steps=8):
"""把 3.1 的循环、15.1 的 span 树、15.3 的成本核算合成一次运行。"""
run_id = uuid.uuid4().hex[:8]
with tracer.span("run", input=user_input) as root:
messages = [Message(role="system", content=SYSTEM),
Message(role="user", content=user_input)]
steps = []
for i in range(max_steps):
with tracer.span("model.complete", step=i) as ms:
time.sleep(0.003) # 模拟一次网络往返，让 span 树有真实形状
resp = model.complete(messages, registry.schemas())
ms.attributes["stop_reason"] = resp.stop_reason
messages.append(resp.message)
label = resp.message.tool_calls[0].name if resp.message.tool_calls else "final"
tracker.record(run_id, "sess-1", model_name,
resp.usage.get("input_tokens", 0),
resp.usage.get("output_tokens", 0), tool=label)
if not resp.message.tool_calls:
steps.append(Step(index=i, kind="final", detail=resp.message.content))
root.attributes["final"] = resp.message.content
return RunResult(messages=messages, steps=steps, final=resp.message.content), run_id
for tc in resp.message.tool_calls:
with tracer.span(f"tool.{tc.name}", step=i) as ts:
result = registry.call(tc)
ts.attributes["ok"] = "error" not in result
messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
steps.append(Step(index=i, kind="tool_call",
detail=f"{tc.name}({tc.arguments}) -> {result}"))
return RunResult(messages=messages, steps=steps, final=""), run_id
traced_run 是这一节的核心。它和 3.1 的 run() 骨架完全一致，只是在每个模型调用外裹了一层 model.complete span、每个工具调用外裹了一层 tool.* span，并把每轮的 usage 顺手记进 CostTracker。业务逻辑一行没改，观测与核算都是横切进去的——这正是 15.1 用 contextvar 维护父子关系换来的收益。
第二段是「开机」：跑一次、打印结果，再把它当作一条评测用例喂给第 16 章的评测器：
def build_cases():
return [
Case("e2e_qa", QUESTION, script=e2e_script(),
expected_tools=["search", "read"],
expected_answer_contains=["多进程", "asyncio"],
expected_citations=["python-concurrency-performance#1", "python-async-advanced#0"],
system=SYSTEM),
Case("no_retrieval", "Python 怎么并发？",
script=[turn([], text="多线程就够了。", inp=90, out=10)],
expected_tools=["search"], expected_answer_contains=["多进程"],
expected_citations=["python-concurrency-performance#1"], system=SYSTEM),
]
def main():
registry = build_registry()
tracer, tracker = Tracer(), CostTracker(PRICES)
result, run_id = traced_run(FakeModel(e2e_script()), registry, QUESTION, tracer, tracker)
print("=== 最终回答 ==="); print(result.final); print()
print("=== 步骤轨迹 ===")
for s in result.steps:
print(f"[{s.index}] {s.kind:<9} {s.detail}")
print()
print("=== 引用校验 ===")
rows = validate(result.final, INDEX)
for claim, doc_id, cid, verdict, ratio in rows:
print(f"{verdict:<16} {doc_id}#{cid} overlap={ratio} | {claim}")
valid = sum(1 for r in rows if r[3] == "有效")
print(f"有效引用: {valid} 无效引用: {len(rows) - valid} 共 {len(rows)} 条"); print()
print("=== span 树 ==="); print(tracer.render()); print()
print("=== 成本核算 ===")
t = tracker.total()
print(f"run_id={run_id} input={t['input']} output={t['output']} cost=${t['cost']:.6f}")
for tool, slot in tracker.by("tool").items():
print(f" {tool:<8} input={slot['input']:<5} output={slot['output']:<4} cost=${slot['cost']:.6f}")
print()
print("=== 评测报告 ===")
report = evaluate(build_cases(), registry, Scorer())
print(f"{'id':<16}{'工具':>6}{'答案':>6}{'引用':>6}{'步数':>6}{'成本(元)':>11}{'得分':>7} 结果")
for s in report.scores:
print(f"{s.case_id:<16}{s.tool_hit:>6.2f}{s.answer_rate:>6.2f}{s.citation_rate:>6.2f}"
f"{s.steps:>6}{s.cost:>11.5f}{s.score:>7.2f} {'PASS' if s.ok else 'FAIL'}")
passed = sum(s.ok for s in report.scores)
print(f"汇总：通过 {passed}/{len(report.scores)}（{report.pass_rate * 100:.1f}%）"
f"｜平均得分 {report.avg_score:.2f}｜平均步数 {report.avg_steps:.1f}"
f"｜平均成本 {report.avg_cost:.5f} 元")
if __name__ == "__main__":
main()
脚本里那个 turn() 小助手值得单独说一句：它把脚本元素写成显式的 ModelResponse（1.3 节的第三种形态），因为只有这种形态才能带上 usage——而成本核算依赖 usage。这恰好演示了「逃生舱」的用途：平时用 str 和 ToolCall 列表 就够，需要注入自定义字段时再退到 ModelResponse。
18.1.3 真跑：一次带检索的问答
用 venv 里的 Python 3.14.6 跑 demo_e2e.py。先看最终回答与步骤轨迹：
=== 最终回答 ===
CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程。[python-concurrency-performance#1]
I/O 密集型适合 asyncio，事件循环在单线程内调度协程。[python-async-advanced#0]
=== 步骤轨迹 ===
[0] tool_call search({'query': 'GIL CPU 密集 多线程', 'k': 2}) -> [{"doc_id": "python-concurrency-performance", "chunk_id": 1, "score": 6.929, "snippet": "CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程 multiprocessing 或原生扩展。"}, {"doc_id": "python-concurrency-performance", "chunk_id": 2, "score": 6.002, "snippet": "I/O 密集型任务在等待期间会释放 GIL，因此多线程仍有收益。"}]
[1] tool_call read({'doc_id': 'python-concurrency-performance', 'chunk_id': 1}) -> CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程 multiprocessing 或原生扩展。
[2] tool_call search({'query': 'asyncio I/O 密集 事件循环', 'k': 2}) -> [{"doc_id": "python-async-advanced", "chunk_id": 0, "score": 7.176, "snippet": "asyncio 适合 I/O 密集型任务，事件循环在单线程内调度大量协程。"}, {"doc_id": "python-memory-model", "chunk_id": 0, "score": 2.035, "snippet": "Python 的引用计数在对象引用归零时立即回收，循环引用由分代 GC 处理。"}]
[3] tool_call read({'doc_id': 'python-async-advanced', 'chunk_id': 0}) -> asyncio 适合 I/O 密集型任务，事件循环在单线程内调度大量协程。
[4] final CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程。[python-concurrency-performance#1]
I/O 密集型适合 asyncio，事件循环在单线程内调度协程。[python-async-advanced#0]
五步收敛，一条完整的多跳检索链：先搜「GIL / CPU 密集」拿到坐标 #1，read 全文，再搜「asyncio / I/O 密集」拿到坐标 #0，read 全文，最后综合成带引用的回答。第 2 步的第二个候选命中了 python-memory-model#0（讲引用计数的），与问题无关——这是混合检索的正常噪声，模型（脚本）没有采信它，只 read 了真正相关的 python-async-advanced#0。
18.1.4 引用校验：回答可信吗
有了带引用的回答，还要用 10.3 的校验函数核对来源。真跑：
=== 引用校验 ===
有效 python-concurrency-performance#1 overlap=1.0 | CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程。
有效 python-async-advanced#0 overlap=0.833 | I/O 密集型适合 asyncio，事件循环在单线程内调度协程。
有效引用: 2 无效引用: 0 共 2 条
两条引用都通过了「存在性 + 支持度」双层校验：坐标都指向真实存在的块，且论断里的关键词分别有 100% 和 83.3% 出现在被引块中。overlap=1.0 那条几乎是原文摘录，0.833 那条是概括转述——这正是校验函数该有的宽容度：概括不该被判为无效。
18.1.5 span 树与成本：这次运行花了什么
同一份 traced_run 还顺手产出了 span 树与成本表：
=== span 树 ===
- run [21.28 ms] {'input': 'Python 里 CPU 密集和 I/O 密集的并发方案该怎么选？', 'final': '...'}
- model.complete [3.78 ms] {'step': 0, 'stop_reason': 'tool_use'}
- tool.search [0.70 ms] {'step': 0, 'ok': True}
- model.complete [3.36 ms] {'step': 1, 'stop_reason': 'tool_use'}
- tool.read [0.02 ms] {'step': 1, 'ok': True}
- model.complete [4.27 ms] {'step': 2, 'stop_reason': 'tool_use'}
- tool.search [0.64 ms] {'step': 2, 'ok': True}
- model.complete [3.74 ms] {'step': 3, 'stop_reason': 'tool_use'}
- tool.read [0.01 ms] {'step': 3, 'ok': True}
- model.complete [4.33 ms] {'step': 4, 'stop_reason': 'end_turn'}
=== 成本核算 ===
run_id=f4a60923 input=1640 output=128 cost=$0.006840
search input=440 output=38 cost=$0.001890
read input=640 output=26 cost=$0.002310
final input=560 output=64 cost=$0.002640
span 树一眼读出两件事。其一，时间几乎全花在模型上：五次 model.complete 合计约 19.5 ms，占 run 总耗时的九成以上，两次 tool.search 不到 1.5 ms。真实场景里这个比例会更极端。其二，五轮 stop_reason 依次是四个 tool_use 加一个 end_turn——这串值就是「Agent 为什么停下来」的完整轨迹。
成本表则暴露了更隐蔽的事实：input token 逐轮上涨（120 → 210 → 320 → 430 → 560），因为每轮 prompt 都带着全部历史。按 claude-sonnet 的示例单价，这次运行花了 0.00684 美元，其中「为决定用 read 而花的钱」（0.00231）比「search」（0.00189）还高——要降本，该优化的不是检索工具本身，而是把 read 的决策成本压下来。这就是 15.3 三维核算的意义：它把优化靶子照了出来。
18.1.6 评测报告：把它纳入回归网
最后，把这次端到端运行固化成一条评测用例，再配一条「注定失败」的用例（模型不检索、直接作答），让分数有区分度：
=== 评测报告 ===
id 工具 答案 引用 步数 成本(元) 得分 结果
e2e_qa 1.00 1.00 1.00 5 0.01285 1.00 PASS
no_retrieval 0.00 0.00 0.00 1 0.00100 0.00 FAIL
汇总：通过 1/2（50.0%）｜平均得分 0.50｜平均步数 3.0｜平均成本 0.00693 元
e2e_qa 满分通过：工具命中、答案命中、引用可回溯三项全中。no_retrieval 三项全零——它没调 search，答案里既没有关键点「多进程」，也没有引用，三项指标一起归零。这条失败用例的存在，正是报告有价值的证明：一个全是 100% 的报告什么也发现不了，而 50% 的通过率让任何一次改动的净效果都变得可读。
至此，一条从前到后的通路完整跑通：提问 → 检索 → 工具 → 循环 → 带引用回答 → 追踪 → 成本 → 评测。第 1 到 17 章的每一个模块，都在这一次运行里出现了，并且互相喂了数据。
真实模型下会怎样：本节的多跳检索路径是脚本写死的——四个 ToolCall 是我们预先排好的，FakeModel 只是照着念。真实 Claude 拿到 search / read 的 schema 后，会自己决定先搜什么、要不要 read、什么时候收尾：它可能一次并发发出两个 search（4.3 节），可能读到一半就认为够了直接作答，也可能多绕两轮。此外，真实运行里 span 树的 model.complete 会是几百毫秒到数秒量级，成本表的 token 数会是几千到几万——本节这些数字小，是因为假模型把「模型怎么想」和「网络有多慢」都省掉了。但请记住：换掉后端，demo_e2e.py 的接线一行不用改，变的只是模型怎么想，不变的正是我们要交付的这台机器。想回看 TS 客户端如何消费同一份协议，见 17.3 TS 落点一：同一协议的 TS 客户端
。
阅读导航：上一节：17.3 TS 落点一：同一协议的 TS 客户端 · 下一节：18.2 真实 API 冒烟（可选） 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。