《AI Agent 开发入门》1.2 Agent 的最小定义与消息协议

1.2 Agent 的最小定义与消息协议

上一节把仓库骨架立了起来，但包里还是空的。这一节我们往里放第一个真正重要的东西：类型。很多人一上来就写「调用大模型的代码」，结果很快陷入泥潭——因为消息格式没定清楚，工具调用的结构今天这样明天那样，测试根本没法写。所以正确顺序是先把数据结构钉死，再写逻辑。

本节把 minagent 推进到「有了消息协议」：定义 Role、ToolCall、Message、ModelResponse 四个类型，给 Message 加上 to_wire() / from_wire() 双向序列化，并用离线断言脚本证明「序列化 → 反序列化 → 回放」往返一致。到本节结束，模型与运行时之间就有了共同的「语言」。

1.2.1 Agent 到底是什么：四个部件

抛开所有框架的宣传词，一个 Agent 运行时最小只需要四样东西：

这四个部件里，模型和工具都是「可替换的」——今天用假模型，明天换成真实供应商；今天只有加法工具，明天接入数据库。而让它们能互相拼接的粘合剂，正是消息。所以本章的重心是消息，而不是循环。

一句话概括：Agent = 模型 + 工具 + 循环 + 状态，其中「状态」就是一条不断增长的消息序列。理解了消息，就理解了 Agent 的骨架。

反过来说，只要消息结构稳定，其余三个部件都可以独立替换。这正是我们要把消息协议放在第 1 章、把循环推迟到第 3 章的原因：先固定接口，再实现行为，改动成本最低。

1.2.2 Role：谁在说话

每条消息都有说话人。我们用一个字符串字面量类型 Role 表示：

from typing import Literal

Role = Literal["system", "user", "assistant", "tool"]

四个角色各自的语义：

• system：系统指令，通常整段对话只有一条，放在最前面，定义 Agent 的身份与约束。
• user：人类或上层程序输入。
• assistant：模型输出，可能是文本，也可能夹带工具调用。
• tool：工具执行结果回灌给模型，必须带上它回应的是哪次调用。

为什么用 Literal 而不是 enum？因为它在类型检查期提供约束，运行时就是一个普通字符串，序列化成 JSON 时零转换成本。这对「消息要频繁过网络」的场景很关键。

1.2.3 ToolCall 与 Message：两个 dataclass

工具调用是 Agent 区别于「普通聊天」的核心。模型不直接执行函数，而是声明「我想调用哪个工具、参数是什么」，由运行时去执行：

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, Any]

id 是这次调用的唯一标识。它存在的唯一理由，是让后面回灌的 tool 消息能精确对应到是哪一次调用——模型可能在一轮里同时发起多个调用，没有 id 就无从配对。

消息本体：

@dataclass
class Message:
    role: Role
    content: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_call_id: str | None = None

四个字段，恰好覆盖四种消息形态：

注意 field(default_factory=list)：dataclass 里给可变类型设默认值必须用 default_factory，直接写 tool_calls: list = [] 会让所有实例共享同一个列表——这是 Python 里最经典的坑之一。

为什么不直接用 dict？三个理由：

1. 有类型：msg.tool_calls[0].name 比 msg["tool_calls"][0]["name"] 少一层引号，写错字段名时类型检查器能报出来。
2. 有相等语义：dataclass 自动生成 __eq__，Message(...) == Message(...) 逐字段比较，这正是 1.2.6 那个断言能写出来的前提。
3. 有默认值：content 默认空串、tool_calls 默认空列表，构造时可以只填关心的字段。

那为什么不用 pydantic？因为卷一的核心引擎坚持零第三方依赖。pydantic 在参数校验（第 4 章）时确实更省事，但它在明确标注「可选」的章节才会登场；消息类型这种最底层、最常被 import 的东西，用标准库 dataclasses 最稳。

1.2.4 ModelResponse：一次模型调用的完整结果

模型每被调用一次，返回的不只是「一条消息」，还有停止原因和用量：

@dataclass
class ModelResponse:
    message: Message
    stop_reason: str = "end_turn"      # "end_turn" | "tool_use" | "max_tokens"
    usage: dict[str, int] = field(default_factory=dict)

• stop_reason 是循环的转向灯：tool_use 表示「我要调工具，你先去执行」，end_turn 表示「我说完了」，max_tokens 表示「被截断了」。第 3 章的停止条件完全依赖它。
• usage 记录 token 消耗，第 2 章的预算控制与第 15 章的成本核算都要读它。

把 message 与 stop_reason 分开，是刻意的设计：「模型说了什么」和「模型为什么停」是两件事。早期实现常把它们混在一起，导致循环逻辑到处打补丁。

1.2.5 to_wire：把对象变成线格式

dataclass 是我们内部的表示，模型供应商要的是 JSON。两者之间需要一个转换层，我们叫它 wire 格式（线格式）。给 Message 加一个 to_wire() 方法：

    def to_wire(self) -> dict[str, Any]:
        if self.tool_calls:
            content: Any = [
                {"type": "tool_use", "id": tc.id, "name": tc.name, "input": tc.arguments}
                for tc in self.tool_calls
            ]
        else:
            content = self.content
        wire: dict[str, Any] = {"role": self.role, "content": content}
        if self.tool_call_id is not None:
            wire["tool_call_id"] = self.tool_call_id
        return wire

关键决策：当有工具调用时，content 从字符串升级为「块列表」。这是主流 API 的共同约定——一个消息的内容可以是纯文本，也可以是由 tool_use、text 等块组成的数组。我们只实现必需的两种块，保持最小。

反向转换交给 from_wire()：

    @classmethod
    def from_wire(cls, wire: dict[str, Any]) -> "Message":
        content = wire.get("content", "")
        tool_calls: list[ToolCall] = []
        text = ""
        if isinstance(content, list):
            for block in content:
                if block.get("type") == "tool_use":
                    tool_calls.append(
                        ToolCall(id=block["id"], name=block["name"],
                                 arguments=dict(block.get("input", {})))
                    )
                elif block.get("type") == "text":
                    text += block.get("text", "")
        else:
            text = content
        return cls(role=wire["role"], content=text, tool_calls=tool_calls,
                   tool_call_id=wire.get("tool_call_id"))

to_wire 与 from_wire 必须成对存在。理由很实际：第 7 章要把会话存盘续跑，第 17 章要把消息通过网络发给 TS 客户端——这两个场景都需要「把内存对象写出去、再原样读回来」。没有 from_wire，消息就只能存不能取。

1.2.6 往返一致性实测

光看代码不够，我们来跑一段离线断言脚本，证明往返不丢信息。它构造一段五条消息的对话（覆盖纯文本、工具调用、工具回灌三种形态），序列化成 JSON，再反序列化回来：

import json

from minagent.types import Message, ToolCall

conversation = [
    Message(role="system", content="你是一个只做加法的助手。"),
    Message(role="user", content="1 + 2 等于多少？"),
    Message(role="assistant", content="",
            tool_calls=[ToolCall(id="call_1", name="add", arguments={"a": 1, "b": 2})]),
    Message(role="tool", content="3", tool_call_id="call_1"),
    Message(role="assistant", content="结果是 3。"),
]

wire = [m.to_wire() for m in conversation]
payload = json.dumps(wire, ensure_ascii=False, sort_keys=True)
reloaded = [Message.from_wire(w) for w in json.loads(payload)]
replayed = [m.to_wire() for m in reloaded]

assert replayed == wire, "往返后 wire 结构不一致"
assert reloaded == conversation, "往返后对象不一致"
print(payload)
print("roundtrip OK:", len(wire), "messages")

本机实测输出（两行）：

[{"content": "你是一个只做加法的助手。", "role": "system"}, {"content": "1 + 2 等于多少？", "role": "user"}, {"content": [{"id": "call_1", "input": {"a": 1, "b": 2}, "name": "add", "type": "tool_use"}], "role": "assistant"}, {"content": "3", "role": "tool", "tool_call_id": "call_1"}, {"content": "结果是 3。", "role": "assistant"}]
roundtrip OK: 5 messages

两个断言都通过，说明：

• 对象级往返一致：reloaded == conversation，dataclass 自动生成的 __eq__ 逐字段比较，工具调用、tool_call_id 都没丢。
• 结构级往返一致：replayed == wire，再序列化一遍得到的 JSON 与第一次完全相同——这正是「存盘续跑」与「跨语言传输」所依赖的稳定性。

注意 sort_keys=True：它让字典键有序，输出才可逐字节复现。这在写确定性测试时是标配（第 16 章会大量用到）。如果去掉它，== 仍成立，但打印结果会随实现变化，不利于比对。

还有一个容易忽略的细节：那条助手消息 content=""、只带 tool_calls。往返后 from_wire 从块列表里解析出工具调用，content 仍然是空串——空串是合法值，不能和「缺失」混为一谈。如果哪天改成 content=None 表示缺失，from_wire 里 wire.get("content", "") 的默认值就要相应调整，否则 None 会被 isinstance(content, list) 判为假，走进字符串分支。这类「空值语义」在协议设计里必须一次定清，后补的代价很高。

1.2.7 真实模型下会怎样

假模型（1.3）与真实供应商共用一个 Message 结构，但 to_wire() 的具体形状要跟着 API 走。真实 API 的差异主要有三处：

• 块的种类更多：真实响应里 content 块可能是 text / tool_use / thinking / image 等，我们的最小实现只覆盖前两种。工程化时要在 from_wire 里为未知块保留原始数据，而不是静默丢弃。
• 字段名与嵌套更深：不同供应商对「工具调用」「工具结果」的字段命名不一致，适配器（第 2 章）负责把差异吸收在这一层，Message 本身保持中立。
• token 用量更细：真实 usage 常带缓存命中、缓存写入等细分字段，dict[str, int] 的宽松类型正好容纳。

一句话：协议中立、适配器吸收差异。这也是为什么我们坚持先定 Message 再谈供应商——只要内部表示稳定，外部接谁都不慌。第 2 章会用一个可选的真实供应商适配器把这条边界演示清楚，而在此之前，1.3 的假模型已经足够让循环跑起来。

小结

• Agent 最小定义 = 模型 + 工具 + 循环 + 状态，其中状态就是一条增长的消息序列。

• Role 用 Literal 而非 enum：类型期有约束，运行时是普通字符串。

• ToolCall.id 是配对工具调用与回灌结果的唯一凭据，一轮可发多个调用。

• Message 的可变字段默认值必须用 field(default_factory=...)。

• ModelResponse 把「说了什么」与「为什么停」分开，stop_reason 驱动循环。

• to_wire() / from_wire() 成对出现，为存盘续跑与跨语言传输打基础。

• 往返一致性用 assert 实测：对象级与结构级都要相等，sort_keys=True 保证可复现。

• 用 dataclass 而非 dict 或 pydantic：有类型、有相等语义、有默认值，且守住零依赖底线。

• 空串与缺失是两种语义，协议里一次定清，别让 from_wire 的默认值埋雷。

• 协议先行、行为后置：先定 Message 再谈循环，改动成本最低。

消息协议立住了。下一节 1.3 可编程假模型 FakeModel 会给这套消息配上一个「完全可预测」的模型后端，让 Agent 循环第一次真正转起来，并且每次运行的结果逐字节一致。想回看环境搭建，翻 1.1 环境与仓库骨架 。

阅读导航：上一节：1.1 环境与仓库骨架 · 下一节：1.3 可编程假模型 FakeModel 。