《AI Agent 开发入门》3.3 抽成 Agent 类：状态、步骤与回调

3.3 抽成 Agent 类：状态、步骤与回调

前两节我们写出了一个既会跑又会停的 run()。它是纯函数：给同样的输入，出同样的结果。作为教学骨架它无可挑剔，但一旦要在真实项目里反复用，就会露出三个别扭之处——每次都得重传一整套配置、历史状态带不走、执行过程只能等结束才一次性拿到。这一节把这些别扭收进一个对象里。

本节把 minagent 推进到「有身份的运行时」：新增 minagent/agent.py，用 Agent 类持有 model / registry / system / max_steps 与 history，提供 on_step / on_tool_call 两个回调钩子，把「配置、状态、过程」三件事各归其位。

3.3.1 函数式的三个别扭之处

把 run() 用在真实项目里，第一反应会是这样：

result = run(model, registry, "第一个问题", system=SYSTEM, max_steps=8)
result = run(model, registry, "第二个问题", system=SYSTEM, max_steps=8)
result = run(model, registry, "第三个问题", system=SYSTEM, max_steps=8)

三个别扭立刻浮现：

这三个问题，正好对应一个类需要提供的三样东西：构造期持有的配置、跨调用存活的状态、运行中触发的回调。把 run() 搬进 Agent 类，本质上就是把这三样东西显式地摆到台面上。

值得强调的是，抽成类不是为了「面向对象」而面向对象。如果只是把 run() 塞进一个 class 里、参数原样传递，那是换汤不换药。真正的收益来自「生命周期」：配置在构造时确定一次，状态在多次调用间流转，回调在运行中随时插入。这三件事都是函数式写法天然做不好的——函数没有「构造期」，也没有「跨调用的自我」。所以判断要不要抽类，看的是「有没有需要跨调用存活的东西」，而不是代码行数。

3.3.2 Agent 类的构造与字段

先看骨架：

# minagent/agent.py
class Agent:
    def __init__(self, model: Model, registry: ToolRegistry, system: str = "",
                 max_steps: int = 8) -> None:
        self.model = model
        self.registry = registry
        self.system = system
        self.max_steps = max_steps
        self.history: list[Message] = []
        self.on_step: StepHook | None = None
        self.on_tool_call: ToolHook | None = None

七个字段分三类：

• 配置（model / registry / system / max_steps）：构造时确定，之后基本不变。它们决定「这个 Agent 是谁、能做什么、跑多久」。
• 状态（history）：跨多次 run() 存活的消息列表，这是「记忆」的雏形，第 8 章会把它扩展成真正的记忆模块。
• 回调（on_step / on_tool_call）：默认 None，调用方按需挂上，用于日志与流式输出。

注意 on_step 与 on_tool_call 是实例属性而非构造参数。这样做的好处是调用方可以事后自由挂载：

agent = Agent(model, registry, system="你是一个计算助手。")
agent.on_step = lambda step: print(step.detail)   # 想加就加
agent.on_step = None                              # 想撤就撤

如果把它们塞进构造参数，就失去了这种灵活性——而回调恰恰是最需要「按场景增删」的东西。

配置类的字段则相反，适合放进构造参数：它们在一个 Agent 的整个生命周期里几乎不变，构造时定好最省心。这种「配置进构造、回调做属性」的分法，在后面几章会一直沿用。

3.3.3 run() 与 history 状态

Agent.run() 与函数式版本的最大区别，在于消息列表的来路和去处：

def run(self, user_input: str) -> RunResult:
    messages = list(self.history)
    if self.system and not messages:
        messages.append(Message(role="system", content=self.system))
    messages.append(Message(role="user", content=user_input))

    steps: list[Step] = []
    for i in range(self.max_steps):
        resp = self.model.complete(messages, self.registry.schemas())
        msg = resp.message
        messages.append(msg)

        if not msg.tool_calls:
            step = Step(index=i, kind="final", detail=msg.content)
            steps.append(step)
            self._emit(step)
            self.history = messages
            return RunResult(messages=messages, steps=steps, final=msg.content)

        for tc in msg.tool_calls:
            result = self.registry.call(tc)
            messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
            if self.on_tool_call is not None:
                self.on_tool_call(tc, result)
            step = Step(index=i, kind="tool_call", detail=f"{tc.name} -> {result}")
            steps.append(step)
            self._emit(step)

    step = Step(index=self.max_steps, kind="error", detail="达到 max_steps 仍未收敛")
    steps.append(step)
    self._emit(step)
    self.history = messages
    return RunResult(messages=messages, steps=steps, final="")

有三处细节值得停一下：

• 开头 messages = list(self.history)：拷贝一份旧历史，而不是直接改 self.history。这样即使在中间抛异常，self.history 也保持上一轮的干净状态，不会留下半截脏数据。
• system 只在历史为空时注入：第二轮进来时 self.history 已经有内容，if not messages 为假，于是不会重复塞系统提示。
• 结尾 self.history = messages：把这一轮累积的完整列表写回实例，下一次 run() 就自动带上上下文。这就是「多轮对话」的实现——不需要额外机制，只是把列表存下来再用。

history 的增长是可以精确对账的。第一轮两步收敛后是 7 条（system + user + 两次 assistant/tool 对 + 最终 assistant）；第二轮只多了一条 user 和一条 assistant，变成 9 条，且没有重复的 system：

这条曲线也暴露了 history 的天真之处：它只增不减。对话越长，每轮喂给模型的 token 就越多，成本随轮数线性上涨，最终必然撞上上下文窗口。这正是第 2 章「Token 预算与截断策略」和第 6 章「上下文压缩」要解决的问题——history 是它们共同的输入。

3.3.4 回调钩子：on_step 与 on_tool_call

两个回调的分工不同：

统一由一个私有方法触发，省去到处判空：

def _emit(self, step: Step) -> None:
    if self.on_step is not None:
        self.on_step(step)

为什么 on_tool_call 要单独存在，而不复用 on_step？因为 Step 是给用户看的摘要（add -> 5），而工具调用的入参在 Step.detail 里被拼成了字符串，拿回来解析很别扭。on_tool_call 直接给出原始 ToolCall 对象，埋点与审计需要结构化数据时更方便。一个面向展示，一个面向程序，分工清晰。

写回调时有一条铁律：回调里的异常不能拖垮循环。上面直接 self.on_step(step) 的写法在真实项目里应当再包一层 try/except，把回调自己的报错吞掉并记日志——因为「打印日志失败」绝不该导致「整个 Agent 崩掉」。本节为了代码简洁没有包，但这是生产化的必做项，第 15 章讲结构化日志时会补上。回调是「观察者」，观察者不该反过来影响被观察者，这是设计模式里的老规矩。

3.3.5 跑一遍：带回调的轨迹

把回调挂上，跑一个两步收敛的算数任务：

agent = Agent(model, registry(), system="你是一个计算助手。", max_steps=8)
agent.on_step = lambda s: print(f"[step {s.index}] {s.kind:<9} {s.detail}")
agent.on_tool_call = lambda tc, r: print(f"  -> 工具 {tc.name} 返回 {r}")

result = agent.run("请计算 (2+3)*4")
print("最终回答:", result.final)
print("history 长度:", len(agent.history))
agent.run("再加 10 是多少")
print("第二轮 history 长度:", len(agent.history))

模型脚本里备了两轮的内容：第一轮走 add → multiply → 文本答案，第二轮直接给文本。输出：

  -> 工具 add 返回 5
[step 0] tool_call add -> 5
  -> 工具 multiply 返回 20
[step 1] tool_call multiply -> 20
[step 2] final     计算结果是 20。
最终回答: 计算结果是 20。
history 长度: 7
[step 0] final     在 20 的基础上再加 10 是 30。
第二轮 history 长度: 9

两处细节印证了前面的设计。其一，on_tool_call 在 on_step 之前触发——因为工具执行完先回调工具钩子，再产生对应的 Step，所以打印顺序是「工具返回」在前、「step」在后。其二，第二轮 history 从 7 涨到 9：多了一条 user 和一条 assistant，而且第二轮没有重新注入 system（否则会变成 10）。这说明 history 真的在跨调用累积，多轮对话已经跑通。

如果把这个 Agent 接到命令行上，on_step 就能实时打印「正在调用 add……」，用户不必等到整轮结束才看到反馈。这正是回调模式最朴素也最有用的价值。

注意回调是在 Agent 实例上挂的，所以同一个 Agent 连续跑多轮时，回调会一直有效；而 steps 是每次 run() 各自独立的列表，不会跨轮累积。这个「回调长驻、轨迹短命」的差别，设计日志系统时要记在心里。

3.3.6 为什么是回调而不是返回列表

有人会问：既然 RunResult.steps 里已经有完整轨迹了，为什么还要回调？答案是时机。steps 要等 run() 返回才能拿到，回调是执行中实时触发的。

这个差别在两类场景里是决定性的：

• 长任务：一次跑 30 秒，用户盯着空屏幕。有回调就能实时打印「第 3 步：正在查询……」，体验天差地别。
• 流式输出：第 7 章要把 Agent 接到 SSE 上，每产生一个事件就推给前端。这时的数据源正是 on_step 这类钩子，而不是事后的 steps 列表。

选择「回调」而不是「继承一个基类再重写方法」，也是刻意的。回调是组合，一个 Agent 实例可以同时挂多个关心不同事情的钩子；继承是耦合，一个类只能继承一条父链。在 Agent 这种需要「日志、监控、UI、测试」多方同时旁听的场景里，组合远比继承合适。

换句话说，steps 是事后记录，回调是事中通知。两者互补，缺一不可——本节的 Agent 同时提供了这两条通道，正是为了给第 7 章的流式打好地基。回调模式还有一个隐性好处：它是解耦的。Agent 不需要知道调用方是要打印、要写日志还是推 SSE，只管在正确的时机 emit 一下，具体做什么由调用方决定。

函数式 run() 与 Agent 类的取舍，可以总结成一张表：

函数式版本并非被淘汰——它依然是 Agent.run() 内部的核心逻辑，只是外面套了一层状态与钩子。当你要写单元测试时，直接调函数式 run() 往往比构造一个 Agent 更省事；当你要跑真实业务时，Agent 类的状态与回调才显出价值。第 16 章讲确定性测试时，两者都会用到。

3.3.7 本节小结

把这三节的演进连起来看：3.1 让循环「能跑」，3.2 让它「会停」，3.3 让它「有身份」。至此，一个最小可用的 Agent 运行时已经成形。

• Agent 类把三样东西显式化：构造期持有的配置、跨调用存活的状态（history）、运行中触发的回调。
• run() 开头拷贝 history、结尾写回 history，两行代码就实现了多轮对话；system 靠 if not messages 保证只注入一次。
• on_step 面向展示（每个 Step 触发一次），on_tool_call 面向程序（拿到原始 ToolCall 与结果），分工互补。
• steps 是事后记录，回调是事中通知；后者是流式输出与进度反馈的基础。
• 函数式 run() 并未废弃，它是 Agent.run() 的内核，也是单元测试最省事的入口。

3.3.8 真实模型下会怎样

用假模型时，回调触发的时机是确定的：脚本说几步就几步，on_step 被调用几次完全可预测。真实模型下，回调的次数和节奏都是动态的——模型可能一步就给出答案（回调只触发一次 final），也可能连着调五个工具（触发五次 tool_call 加一次 final）。调用方不能假设步数，只能把回调写成「来一次处理一次」的形态。

这也是为什么回调签名里只给 Step 而不给「总步数」或「百分比」——循环事先并不知道要跑几步，进度条只能显示「已完成 N 步」，无法显示百分比。真实产品里通常改用「当前在做什么」的语义化提示，而不是一个假精确的百分比。

还有一点：真实模型的回调可能携带流式片段。假模型的 Step.detail 是一次性拼好的完整字符串，而真实适配器可以在模型逐 token 吐字时，把一个 final 拆成多个增量事件推给 on_step。本节的回调只定义了「一个 Step 一次」的最小语义，第 7 章会在它之上加一个更细的事件模型。

也要提醒一句：history 在真实多轮场景下会迅速膨胀。用户每问一句，历史就多出「用户消息 + 模型回复 + 所有工具往返」一长串。假模型的脚本只有几步，看不出压力；真实对话跑上十几轮，光历史就可能占满上下文窗口。这就是为什么「有状态」既是 Agent 类的价值，也是它必须被管理的原因。

到这里，minagent 已经有了「会跑、会停、有身份」的运行时。但它还没有真正的工具系统——现在每个例子都得手搭一个 ToolRegistry。下一章 4.1 工具即函数：schema 生成、注册表与分发 会把「工具」做成 minagent 的一等公民。想回看止损线是怎么加的，见 3.2 停止条件与循环上限 。

阅读导航：上一节：3.2 停止条件与循环上限 · 下一节：4.1 工具即函数：schema 生成、注册表与分发 。