《AI Agent 开发入门》1.3 可编程假模型 FakeModel

本节写下全书的测试基石 FakeModel:一个按脚本列表吐响应的假模型,脚本元素可以是文本、ToolCall 列表或 ModelResponse,它记录每次调用入参、脚本耗尽即抛 RuntimeError。随后用一个极简循环跑通「add 到 mul 再文本收敛」,打印完整轨迹并断言两次运行逐字节一致。

1.3 可编程假模型 FakeModel

1.2 定义了消息,但还没有任何东西会「回话」。真实的模型调用又慢、又贵、又不确定——同一段提示词,两次返回可能一字不差,也可能天差地别。用它来开发 Agent,等于把调试建立在流沙上。所以本书走一条更硬的路:先造一个完全可控的假模型,让循环逻辑在没有网络、没有费用、没有随机性的环境里跑通并断言。

本节把 minagent 推进到「第一轮假模型跑通」:实现 FakeModel,用一个极简循环跑出「add → mul → 文本收敛」三步轨迹,并断言两次运行逐字节一致。到本节结束,第 1 章的三块基石(环境、消息、假模型)全部到位。

1.3.1 为什么需要一个假模型

先说清楚:FakeModel 不是「假装在调模型」,它是测试夹具 + 教学骨架。它的职责是把「模型会怎么回话」这件事变成一份可以提前写死的脚本。

对比一下两种开发方式:

维度真实模型FakeModel
响应速度数百毫秒到数十秒微秒级
费用按 token 计费零
确定性不保证逐字节可复现
可注入边界情况难(要构造提示词碰运气)易(脚本里直接写)
适用场景冒烟、演示开发、单元测试、评测

真实模型有一个致命问题:你没法可靠地复现一个 bug。今天失败的那次对话,明天可能就通过了,你根本不知道是哪次循环的哪一步出了问题。FakeModel 把这个问题彻底解决——脚本写死,轨迹就写死,断言才立得住。

更重要的是,真实供应商适配器实现的是同一个 Model 协议(第 2 章)。换句话说,FakeModel 与真实模型是可互换的:用假模型跑通、跑对,换成真模型时循环逻辑一行不用改。

1.3.2 脚本的三种元素

FakeModel 的核心是一份脚本列表,每个元素是三种形态之一:

元素形态含义对应 ModelResponse
str文本回复,表示「我说完了」stop_reason="end_turn"
list[ToolCall]本轮发起工具调用stop_reason="tool_use"
ModelResponse原样返回,可注入错误或自定义 stop_reason由你构造

第三种形态是「逃生舱」:需要模拟 max_tokens 截断、模拟一次异常响应、或注入带自定义 usage 的回复时,直接构造 ModelResponse 塞进脚本即可。前两种覆盖 90% 的日常用例,第三种覆盖剩下的边界情况。

这份脚本按顺序消费:每被调用一次,取出一项。脚本长度天然等于「这个 Agent 最多能转几轮」——写短了会提前报错(见 1.3.7),写长了则末尾元素永远用不上。

1.3.3 FakeModel 的实现

完整实现不到 30 行,放在 minagent/fake.py:

from __future__ import annotations

from typing import Any

from .types import Message, ModelResponse, ToolCall


class FakeModel:
    """脚本化模型:按脚本顺序返回响应,用于确定性测试与教学。"""

    def __init__(self, script: list[Any]) -> None:
        self.script = list(script)
        self.calls: list[tuple[list[Message], list[dict]]] = []
        self._cursor = 0

    def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
        self.calls.append((list(messages), list(tools)))
        if self._cursor >= len(self.script):
            raise RuntimeError("FakeModel 脚本耗尽:Agent 没有收敛")
        item = self.script[self._cursor]
        self._cursor += 1
        if isinstance(item, str):
            return ModelResponse(message=Message(role="assistant", content=item))
        if isinstance(item, ModelResponse):
            return item
        return ModelResponse(
            message=Message(role="assistant", tool_calls=list(item)),
            stop_reason="tool_use",
        )

逐点拆解:

  • complete 的签名与 Model 协议一致:(messages, tools) -> ModelResponse。正因如此,它才能和真实适配器互换。
  • self.calls 记录每次入参:每调一次追加一个二元组 (消息列表, 工具 schema 列表),即模型当时看到的完整输入。第 16 章的评测要断言「模型第 3 次调用时看到了几条消息、拿到了哪些工具」,第 6 章要断言「裁剪后的历史是不是真变短了」,全靠它。注意这里做了 list(...) 浅拷贝——否则后续对 messages 的追加会污染历史记录。
  • isinstance 的顺序有讲究:先判 str,再判 ModelResponse,最后兜底当作 list[ToolCall]。因为 str 也是可迭代的,若不先判它,list[ToolCall] 分支会把字符串拆成字符。
  • 耗尽即抛错:脚本用完还在被调用,说明 Agent 没有收敛——要么循环停止条件写错了,要么脚本本身写少了。这是一个必须响亮的失败信号,绝不能静默返回空响应。

1.3.4 一个极简循环

有了假模型,还需要一个「问模型 → 执行工具 → 再问」的循环。第 3 章会把它打磨成正式的 run(),这里先用最小版本让第 1 章闭环:

from minagent.fake import FakeModel
from minagent.types import Message, ToolCall

TOOLS = {"add": lambda a, b: a + b, "mul": lambda a, b: a * b}


def dispatch(tc: ToolCall) -> str:
    fn = TOOLS[tc.name]
    return str(fn(**tc.arguments))


def run(model: FakeModel, user_input: str, system: str = "", max_steps: int = 8):
    messages = [Message(role="system", content=system)] if system else []
    messages.append(Message(role="user", content=user_input))
    trace = []
    for step in range(max_steps):
        resp = model.complete(messages, [])
        messages.append(resp.message)
        if resp.message.tool_calls:
            for tc in resp.message.tool_calls:
                out = dispatch(tc)
                trace.append(f"[{step}] {tc.name}{tc.arguments} -> {out}")
                messages.append(Message(role="tool", content=out, tool_call_id=tc.id))
            continue
        trace.append(f"[{step}] final: {resp.message.content}")
        return resp.message.content, trace
    raise RuntimeError("超过 max_steps")

这段代码刻意「简陋」:工具表是一个 dict、没有 schema、没有参数校验、没有错误处理。这些缺失正是后续章节的抓手——第 3 章抽 Agent 类,第 4 章换成 ToolRegistry。第 1 章只需要证明循环能转。

两个关键点先记住:每次工具执行后,结果都以 tool 角色的消息回灌(带 tool_call_id),而不是塞回 assistant 消息里;continue 让循环回到顶部再问一次模型,直到某次响应不再带工具调用,才走 final 分支。

1.3.5 跑通 add → mul → 收敛

现在写一个脚本,让假模型依次发出「加法调用」「乘法调用」「文本收尾」三步:

from minagent.fake import FakeModel
from minagent.types import ToolCall


def build_model() -> FakeModel:
    return FakeModel([
        [ToolCall(id="c1", name="add", arguments={"a": 2, "b": 3})],
        [ToolCall(id="c2", name="mul", arguments={"a": 5, "b": 4})],
        "2 + 3 = 5,再乘 4 得 20。",
    ])


first_final, first_trace = run(build_model(), "先算 2+3,再把结果乘 4")
second_final, second_trace = run(build_model(), "先算 2+3,再把结果乘 4")

注意第二轮的乘法参数是 {"a": 5, "b": 4}——5 是第一轮加法得到的结果。在真实 Agent 里,这个 5 是模型看到工具返回后自己填进去的;在假模型里,我们直接把它写死在脚本里。假模型省掉的是「模型怎么想」,保留的是「循环怎么转」,而后者才是本章要验证的。

本机实测轨迹:

[0] add{'a': 2, 'b': 3} -> 5
[1] mul{'a': 5, 'b': 4} -> 20
[2] final: 2 + 3 = 5,再乘 4 得 20。
final: 2 + 3 = 5,再乘 4 得 20。
deterministic OK: True

三行轨迹分别对应 step=0/1/2。第 0、1 步走工具分支,第 2 步走 final 分支——因为第 3 个脚本元素是 str,不带 tool_calls,循环就此收敛。这就是最简单的停止条件:模型不再要求调用工具。

1.3.6 确定性:两次运行逐字节一致

细看 1.3.5 的代码,我们把 run 跑了两遍,然后断言两次结果相同:

assert first_trace == second_trace
assert first_final == second_final
assert first_trace[-1] == "[2] final: 2 + 3 = 5,再乘 4 得 20。"

三个断言的意义层层递进:

  1. 轨迹相等:两次运行走了完全相同的步骤序列,证明循环没有隐藏的随机性。
  2. 终值相等:最终回答一字不差。
  3. 末步形态正确:最后一步是 final 而非 tool_call,证明循环确实收敛了,而不是撞到 max_steps 才停。

本机输出末尾的 deterministic OK: True 是把两次轨迹各序列化成 JSON 后比较得到的:

import json
json.dumps(first_trace, ensure_ascii=False) == json.dumps(second_trace, ensure_ascii=False)

为什么「逐字节一致」这么重要?因为它是第 16 章评测的地基。评测的本质是「给定输入,断言输出轨迹」——如果同一份脚本两次跑出的轨迹都可能不同,任何断言都无从谈起。假模型把这层不确定性彻底消除,让 Agent 的测试第一次变成像测试纯函数一样的事。

1.3.7 脚本耗尽即报错

最后验证一下「不收敛」的行为。让脚本只有一项,却让它带工具调用——循环必然需要第二次调用模型,而此时脚本已空:

from minagent.fake import FakeModel
from minagent.types import Message

m = FakeModel(["done"])
r = m.complete([Message(role="user", content="hi")], [])
print("first:", r.message.content, "| stop:", r.stop_reason)
seen_msgs, seen_tools = m.calls[0]
print("calls recorded:", len(m.calls), "| n_messages:", len(seen_msgs), "| tools:", seen_tools)
try:
    m.complete([], [])
except RuntimeError as e:
    print("RuntimeError:", e)

本机实测:

first: done | stop: end_turn
calls recorded: 1 | n_messages: 1 | tools: []
RuntimeError: FakeModel 脚本耗尽:Agent 没有收敛

三个信号值得注意:

  • calls 确实被记录了:调用一次后长度为 1,元组里是 1 条消息与一个空的 tools(本节的极简循环还没传工具 schema)。
  • stop_reason 默认是 end_turn:文本回复天然表示「结束」。
  • 耗尽抛的是 RuntimeError,而不是返回空响应。这一点很关键:空响应会被循环误当成「模型说了空话」从而正常收敛,把真正的 bug 掩盖掉;抛异常才能让问题立刻暴露。

在实际测试里,你应该故意写一个「脚本不够长」的用例,断言它抛出 RuntimeError——这验证了「循环确实会因为没收敛而失败」,而不是碰巧跑对了。

1.3.8 真实模型下会怎样

把 FakeModel 换成真实适配器,run 的代码一行不用改,但行为差异要心里有数:

  • 不再确定:同样的输入,真实模型可能在第 2 轮就收敛,也可能多转两轮;工具参数(比如那个 5)由模型现场推理得出,不再写死。
  • 会真的失败:网络超时、限流、上下文超长都会发生。complete 不再保证返回 ModelResponse,所以真实适配器要负责把异常转成可识别的错误响应,交给第 14 章的重试逻辑。
  • 有 token 成本:usage 不再是空的 {},而是真实的输入/输出 token 数,第 2 章的预算控制由此生效。
  • calls 记录依然有用:调试真实 Agent 时,先看「每一步模型拿到几条消息、可选哪些工具」,往往一眼就能看出历史是不是漏灌、schema 是不是没传下去,这个习惯从假模型延续到真模型。

一句话:假模型负责「证明逻辑对」,真实模型负责「证明逻辑在现实中也能用」。第 2 章先补齐 Model 协议与可选适配器,把这条换装路径铺平。

小结

  • FakeModel 是测试夹具 + 教学骨架,不是「假装在调模型」;它与真实适配器实现同一 Model 协议、可互换。
  • 脚本三种元素:str(收敛)、list[ToolCall](发起调用)、ModelResponse(注入边界情况)。
  • self.calls 每次调用追加一个 (消息列表, 工具 schema 列表) 元组,是第 16 章断言「模型看到了什么」、第 6 章断言「裁剪是否生效」的数据来源;记录时要浅拷贝。
  • 循环每次执行工具后,用 tool 角色消息 + tool_call_id 回灌,再回到顶部重新问模型。
  • 最简单的停止条件是「模型不再要求调用工具」;stop_reason 是更完备的判据。
  • 两次运行逐字节一致,是评测可写的前提;假模型把不确定性降到零。
  • 脚本耗尽必须 raise RuntimeError,绝不能静默返回空响应——那会掩盖不收敛的 bug。

第 1 章到此收尾:环境立住、消息定死、假模型跑通,minagent 从空壳长出了第一块肌肉。下一节 2.1 Model 协议设计 会把「模型」这件事抽象成一个正式协议,并规划 token 预算与真实供应商适配的边界。想回看消息结构,翻 1.2 Agent 的最小定义与消息协议 。

阅读导航:上一节:1.2 Agent 的最小定义与消息协议 · 下一节:2.1 Model 协议设计 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「agent」更多文章

  1. 《AI Agent 高级编程》目录
  2. 《AI Agent 高级编程》11.3 全景回顾:从零到框架的完整路径与展望
  3. 《AI Agent 高级编程》11.2 真实 API 冒烟:假模型 vs 真模型量化差距