前面 18 章把 minagent 从 20 行的循环长成了「消息 / 工具 / 记忆 / 编排 / 评测 / 服务」的完整模块。这一页只做一件事:把全书反复使用的类型契约、模块布局和调用片段压成表,方便你在写代码时对号入座,而不必翻回某一章找签名。
本附录所有片段都在本机 Python 3.14.6 的 venv 里实际运行过,输出为真实结果。字段名与 §六 的权威契约逐字一致,任何章节都不得改写它们。
A.1 模块布局
minagent 是一个纯标准库的包。卷一从 minagent/ 目录开始,18 章逐步往里加文件,但核心五个文件从第 1–3 章就固定下来,后续章节只叠加、不改名。
| 文件 | 职责 | 首次出现 |
|---|---|---|
minagent/__init__.py | 汇总导出公开名字,供 from minagent import ... | 第 1 章 |
minagent/types.py | 消息协议:Role / ToolCall / Message / ModelResponse / Model / Step / RunResult | 第 1 章 |
minagent/model.py | FakeModel 假模型,以及可选的真实供应商适配器 | 第 1–2 章 |
minagent/tools.py | Tool 与 ToolRegistry:schema 生成、注册分发、错误回灌 | 第 4 章 |
minagent/loop.py | run() 主循环与停止条件 | 第 3 章 |
minagent/agent.py | Agent 类:把模型、工具、系统提示、步数上限封成对象 | 第 3 章 |
后续章节在核心五文件之外继续加模块,但都不改核心契约:
| 文件 | 章节 | 职责 |
|---|---|---|
minagent/memory.py | 第 8–9 章 | 缓冲 / 摘要记忆、检索注入、写入策略 |
minagent/rag.py | 第 10 章 | 文档加载切分、search / read 工具 |
minagent/planner.py | 第 11 章 | 任务分解 DAG、计划执行器、重规划 |
minagent/graph.py | 第 12 章 | 最小图引擎:节点 / 边 / 条件跳转 |
minagent/multiagent.py | 第 13 章 | supervisor / worker、进程内消息总线 |
minagent/observability.py | 第 15 章 | span 树、结构化日志、成本核算 |
minagent/eval.py | 第 16 章 | 确定性测试、评测集、对比实验 |
minagent/service.py | 第 17 章 | 稳定接口 + FastAPI / SSE 落点 |
推荐的最小导入面是顶层包:
from minagent import (
Message, Role, ToolCall, ModelResponse, Model,
Step, RunResult, Tool, ToolRegistry, FakeModel, run, Agent,
)
A.2 消息与类型契约
types.py 是全书的公理。下面逐个类型列出字段的名字、类型与默认值——字段名是契约的一部分,不得改动。
A.2.1 Role
Role 是一个字符串字面量联合,只有四个合法取值:
| 取值 | 含义 |
|---|---|
"system" | 系统指令,通常放在消息列表首位 |
"user" | 用户输入 |
"assistant" | 模型回复(可能带 tool_calls) |
"tool" | 工具执行结果,回灌给模型 |
A.2.2 ToolCall
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
id | str | 无 | 调用标识,回灌 tool 消息时用 tool_call_id 对应 |
name | str | 无 | 工具名,必须已在 ToolRegistry 注册 |
arguments | dict[str, Any] | 无 | 参数键值对,展开后传给工具函数 |
A.2.3 Message
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
role | Role | 无 | 四选一 |
content | str | "" | 文本内容;工具调用轮可以为空串 |
tool_calls | list[ToolCall] | [] | 仅 assistant 消息可能非空 |
tool_call_id | str | None | None | 仅 tool 消息使用,指向被回灌的调用 |
A.2.4 ModelResponse
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
message | Message | 无 | 模型这一轮的回复 |
stop_reason | str | "end_turn" | 三选一:"end_turn" / "tool_use" / "max_tokens" |
usage | dict[str, int] | {} | 形如 {"input_tokens":..,"output_tokens":..} |
A.2.5 Model 协议
class Model(Protocol):
def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse: ...
任何后端只要实现这一个方法就满足协议:FakeModel 实现它,真实供应商适配器也实现它,两者可互换。注意 tools 是「已序列化的 JSON Schema 列表」,不是 Tool 对象列表——循环会先调 registry.schemas() 转换。
A.2.6 Step 与 RunResult
| 类型 | 字段 | 类型 | 说明 |
|---|---|---|---|
Step | index | int | 第几轮(从 0 起) |
Step | kind | str | "tool_call" / "final" / "error" |
Step | detail | str | 人类可读的一行轨迹 |
RunResult | messages | list[Message] | 完整消息历史,含 system / user / assistant / tool |
RunResult | steps | list[Step] | 轨迹,测试断言的主战场 |
RunResult | final | str | 最后一轮助手文本,即「最终答复」 |
A.3 工具契约
A.3.1 Tool
| 字段 | 类型 | 说明 |
|---|---|---|
name | str | 工具名,注册表的键 |
description | str | 给模型看的自然语言说明 |
parameters | dict[str, Any] | JSON Schema 片段(type / properties / required) |
func | Callable[..., Any] | 实际执行的 Python 函数,按关键字接收 arguments |
schema() | -> dict[str, Any] | 返回 {"name":..,"description":..,"parameters":..} |
A.3.2 ToolRegistry
| 方法 | 签名 | 行为 |
|---|---|---|
add | (tool: Tool) -> None | 按 tool.name 存入字典,重名覆盖 |
schemas | () -> list[dict[str, Any]] | 返回全部工具的 schema 列表,喂给 model.complete |
call | (tc: ToolCall) -> str | 执行工具;出错返回 {"error": ...} JSON,绝不抛异常 |
call 不抛异常是设计决定,不是疏漏:工具报错要作为一条 tool 消息回灌给模型,让它自己决定重试、换工具还是道歉,而不是中断整个循环。附录 B 与附录 D 会分别从测试和排错的角度展开这一点。
A.4 循环与 Agent
A.4.1 run()
def run(model: Model, registry: ToolRegistry, user_input: str,
system: str = "", max_steps: int = 8) -> RunResult: ...
| 参数 | 默认 | 说明 |
|---|---|---|
model | 无 | 任何满足 Model 协议的后端 |
registry | 无 | 已注册好工具的工具表 |
user_input | 无 | 本轮用户输入 |
system | "" | 非空时作为首条 system 消息插入 |
max_steps | 8 | 超过步数仍不收敛则 raise RuntimeError |
A.4.2 Agent
Agent 只是把上面五个参数封成对象,run(user_input) 委托给模块级 run()。它的价值在于让第 7 章的 Session、第 17 章的服务层有稳定的持有物。
| 成员 | 说明 |
|---|---|
Agent(model, registry, system="", max_steps=8) | 构造 |
.run(user_input) -> RunResult | 跑一轮 |
A.5 常见调用片段
A.5.1 注册一个工具并跑通「工具调用 → 文本收敛」
from minagent import FakeModel, Tool, ToolCall, ToolRegistry, run
reg = ToolRegistry()
reg.add(Tool(
name="add",
description="两数相加",
parameters={
"type": "object",
"properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"],
},
func=lambda a, b: a + b,
))
model = FakeModel([
[ToolCall("c1", "add", {"a": 2, "b": 3})], # 第 1 轮:发工具调用
"结果是 5。", # 第 2 轮:给文本,收敛
])
result = run(model, reg, "2+3 等于多少?", system="你是一个计算助手")
assert [s.kind for s in result.steps] == ["tool_call", "final"]
assert result.final == "结果是 5。"
print(result.final) # 结果是 5。
A.5.2 用 model.calls 断言「模型看到了什么」
FakeModel 每次被调用都会往 model.calls 追加一个二元组 (消息列表, 工具 schema 列表)——即模型当时看到的完整输入。这让「循环有没有把工具 schema 传下去」「消息数有没有按预期增长」变成可断言的事实。
msgs, tools = model.calls[0]
assert len(msgs) == 2 # system + user
assert [t["name"] for t in tools] == ["add"]
assert len(model.calls[1][0]) == 4 # + assistant + tool
A.5.3 用 Agent 封装同一段逻辑
from minagent import Agent
agent = Agent(model=FakeModel([[ToolCall("c1", "add", {"a": 2, "b": 3})], "结果是 5。"]),
registry=reg, system="你是一个计算助手", max_steps=4)
result = agent.run("2+3 等于多少?")
assert result.final == "结果是 5。"
A.5.4 构造 ModelResponse 注入自定义 stop_reason 与用量
脚本项除了 str 和 ToolCall 列表,还可以直接放一个 ModelResponse,原样返回:
from minagent import Message, ModelResponse
resp = ModelResponse(
message=Message(role="assistant", content=""),
stop_reason="max_tokens",
usage={"input_tokens": 12, "output_tokens": 0},
)
model = FakeModel([resp])
result = run(model, reg, "随便问一句")
assert result.steps[-1].kind == "final"
assert result.messages[-1].content == ""
这是模拟「模型截断、没有给出完整答复」的最简办法,第 6 章的结构化输出与第 14 章的重试都靠它。
A.5.5 断言「工具报错被回灌而不是中断」
def boom(a, b):
raise ValueError("除零")
reg.add(Tool("div", "两数相除", {"type": "object", "properties": {}}, boom))
model = FakeModel([[ToolCall("c1", "div", {"a": 1, "b": 0})], "无法计算。"])
result = run(model, reg, "1/0")
assert "ValueError" in result.steps[0].detail # 错误进了轨迹
assert result.final == "无法计算。" # 循环继续并收敛
A.5.6 一段可复现的「双跑一致性」断言
确定性是卷一的卖点之一。同一份脚本跑两次,轨迹必须逐字节一致:
script = [[ToolCall("c1", "add", {"a": 2, "b": 3})], "结果是 5。"]
r1 = run(FakeModel(script), reg, "2+3 等于多少?", system="计算助手")
r2 = run(FakeModel(script), reg, "2+3 等于多少?", system="计算助手")
assert [s.detail for s in r1.steps] == [s.detail for s in r2.steps]
assert r1.final == r2.final
A.6 常见误用对照
| 症状 | 误用 | 正解 |
|---|---|---|
| 工具没被调用 | 脚本项写成单个 ToolCall 而非 [ToolCall(...)] | 工具调用轮必须是列表,哪怕只有一个 |
TypeError: 不认识的脚本项 | 脚本里放了 None / dict | 只放 str / ToolCall 列表 / ModelResponse |
RuntimeError: 脚本已耗尽 | 脚本没给出终止的文本回复 | 补一条 str,或检查工具名是否写错导致空转 |
断言 calls[0] 失败 | 忘了 system 会占一条消息 | 传了 system 时首轮消息数从 2 起算 |
参数报 unexpected keyword | parameters 的 properties 键与函数形参名不一致 | 两者逐字对齐,required 里的键必须存在 |
| 工具异常让测试崩了 | 直接调 tool.func(...) 绕过注册表 | 走 registry.call(tc),异常会被转成 {"error": ...} |
FakeModel 与真实适配器的差别只在 complete() 的实现:假模型按脚本吐响应,真实适配器把 messages 与 tools 序列化成 HTTP 请求、再把响应反序列化成 ModelResponse。上层的 run() 循环、ToolRegistry、Step 轨迹一行都不用改——这正是把 Model 抽成协议的意义。
两点必须记住:真实模型的 usage 会有真实 token 数,而假模型默认是空字典;真实模型的 stop_reason 由供应商返回,可能遇到 "max_tokens",假模型只在你显式构造 ModelResponse 时才会给出它。卷一默认走假模型,真实 API 调用只在第 18 章末尾、且需你明确同意才进行。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。