2.2 真实供应商适配与离线替身
上一节我们立好了 Model 协议，也证明了假模型能塞进去。但假模型终究不会真的回答问题，这一节要回答一个很实际的问题：怎么把一家真实供应商接进这个协议，同时又不在没有密钥、没有额度的情况下把读者挡在门外。答案分两半——一个「翻译层」适配器，一个「录像机」离线替身。
本节把 minagent 推进到「能接真实后端，也能离线复现」：写出
AnthropicModel适配器完成双向映射，再引入RecordedModel用存盘的响应 JSON 重放。到本节结束，同一份循环代码既能打真实 API，也能在断网时跑通。
2.2.1 适配器的职责：两侧翻译
适配器是协议与某家 SDK 之间的翻译官。它只做两件事：
| 方向 | 输入 | 输出 | 难点 |
|---|---|---|---|
| 入参映射 | list[Message] + tools | SDK 的请求参数 | system 位置、工具调用的 block 结构 |
| 出参映射 | SDK 的响应对象 | ModelResponse | 内容块拼接、停止原因归一化 |
它不做的事同样重要：不重试、不加缓存、不改业务逻辑。这些留到后面的包装器章节。保持「一层只干一件事」，适配器才能被单独测。
2.2.2 入参映射：Message 列表变成 SDK 参数
Anthropic 的 Messages API 有两个特点：system 提示是顶层参数，不放在消息数组里；工具调用与工具结果都是消息里的 content block。所以映射时要把 minagent 的四种角色拆开处理：
# minagent/anthropic_adapter.py
from __future__ import annotations
import os
from anthropic import Anthropic
from anthropic.types import Message as AnthropicMessage
from minagent.types import Message, ModelResponse, ToolCall
def to_anthropic(messages: list[Message]) -> tuple[str, list[dict]]:
"""把 minagent 的 Message 列表翻译成 Anthropic 的 (system, messages)。"""
system_parts: list[str] = []
out: list[dict] = []
for m in messages:
if m.role == "system":
system_parts.append(m.content)
elif m.role == "user":
out.append({"role": "user", "content": m.content})
elif m.role == "assistant":
blocks: list[dict] = []
if m.content:
blocks.append({"type": "text", "text": m.content})
for tc in m.tool_calls:
blocks.append({"type": "tool_use", "id": tc.id,
"name": tc.name, "input": tc.arguments})
out.append({"role": "assistant", "content": blocks})
elif m.role == "tool":
out.append({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": m.tool_call_id,
"content": m.content}]})
return "\n".join(system_parts), out
几个容易写错的点：
- system 要抽出来拼接，不能当成普通消息塞进
messages，否则 SDK 会报错。 - 助手消息若既有文本又有工具调用，要按顺序放进同一个 content 数组（先
text再tool_use）。 - 工具结果在 Anthropic 里是
role: "user"的消息，且必须用tool_use_id回指是哪次调用的结果。这正是我们Message.tool_call_id字段存在的理由。
2.2.3 出参映射：SDK 响应折回 ModelResponse
反向映射要把 SDK 的 content block 列表「拍平」：文本块拼成一个字符串，工具调用块收集成 ToolCall 列表。
# minagent/anthropic_adapter.py（续）
def from_anthropic(resp: AnthropicMessage) -> ModelResponse:
"""把 SDK 的 Message 折叠回统一的 ModelResponse。"""
texts: list[str] = []
tool_calls: list[ToolCall] = []
for block in resp.content:
if block.type == "text":
texts.append(block.text)
elif block.type == "tool_use":
tool_calls.append(ToolCall(id=block.id, name=block.name,
arguments=dict(block.input)))
return ModelResponse(
message=Message(role="assistant", content="".join(texts),
tool_calls=tool_calls),
stop_reason=resp.stop_reason or "end_turn",
usage={"input_tokens": resp.usage.input_tokens,
"output_tokens": resp.usage.output_tokens},
)
Anthropic 的 stop_reason 取值（end_turn / tool_use / max_tokens）恰好与我们的三个常量一致，所以这里直接透传。换一家供应商时，这里就得写一张映射表——这正是适配器该吸收的差异。
2.2.4 组装 AnthropicModel
把两侧映射包进实现协议的对象里：
# minagent/anthropic_adapter.py（续）
class AnthropicModel:
"""真实供应商适配器。本节示例默认不真调；只在显式构造 client 时才联网。"""
def __init__(self, model: str, max_tokens: int = 1024,
client: Anthropic | None = None) -> None:
self.model = model
self.max_tokens = max_tokens
self._client = client or Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
system, msgs = to_anthropic(messages)
resp = self._client.messages.create(
model=self.model,
max_tokens=self.max_tokens,
system=system,
messages=msgs,
tools=tools or [],
)
return from_anthropic(resp)
三条纪律必须写清楚：
- 密钥永远从
os.environ["ANTHROPIC_API_KEY"]读，绝不硬编码，也不要把真实密钥写进任何文件或正文。 - 本节示例默认不真调。只有你显式构造了
client并且真的调用complete，它才会发网络请求。 - 模型 id 不要凭记忆写。用
client.models.list()或 Models API 查当前可用的 id，再填进model=。书里统一用示例 idclaude-opus-5-5只是占位。
2.2.5 离线验证映射：不发一个请求
好消息是：入参和出参映射都可以离线验证，因为出参映射接收的是「一个长得像 SDK 响应的对象」，而不是网络返回。用 AnthropicMessage.model_validate(raw) 就能从一段 JSON 造出这个对象：
# run22a.py
import sys, json
sys.path.insert(0, "/tmp/ab_work")
from minagent.types import Message, ToolCall
from minagent.anthropic_adapter import to_anthropic, from_anthropic
from anthropic.types import Message as AM
# 入参映射
msgs = [
Message(role="system", content="你是严谨的助手"),
Message(role="user", content="北京现在几点？"),
Message(role="assistant", content="", tool_calls=[ToolCall("t1", "get_time", {"tz": "Asia/Shanghai"})]),
Message(role="tool", content="10:00", tool_call_id="t1"),
]
system, out = to_anthropic(msgs)
print("system =", repr(system))
print("messages =", json.dumps(out, ensure_ascii=False))
# 出参映射：用一段「录制」的原始响应
raw = {
"id": "msg_01ABC", "type": "message", "role": "assistant",
"model": "claude-opus-5-5",
"content": [
{"type": "text", "text": "现在是北京时间 10:00。"},
{"type": "tool_use", "id": "t2", "name": "add", "input": {"a": 1, "b": 2}},
],
"stop_reason": "tool_use", "stop_sequence": None,
"usage": {"input_tokens": 52, "output_tokens": 18},
}
resp = from_anthropic(AM.model_validate(raw))
print("content =", repr(resp.message.content))
print("tool_calls =", resp.message.tool_calls)
print("stop_reason =", resp.stop_reason, "usage =", resp.usage)
真跑的输出（无需密钥、无需网络）：
system = '你是严谨的助手'
messages = [{"role": "user", "content": "北京现在几点？"}, {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "get_time", "input": {"tz": "Asia/Shanghai"}}]}, {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "10:00"}]}]
content = '现在是北京时间 10:00。'
tool_calls = [ToolCall(id='t2', name='add', arguments={'a': 1, 'b': 2})]
stop_reason = tool_use usage = {'input_tokens': 52, 'output_tokens': 18}
注意工具结果被翻译成了 role: "user" + tool_result block，工具调用被折成了 ToolCall(id='t2', ...)。映射对不对，一眼可见，这就是可离线验证的价值。
2.2.6 录制夹具：一次真实、多次重放
出参映射既然能喂 JSON，就可以把「一次真实响应」存成文件，之后反复重放。这套做法叫录制夹具（recording fixture），它把「真实」与「可复现」解耦：
| 时机 | 做什么 | 需要密钥/网络吗 |
|---|---|---|
| 录制 | 真调一次，把响应 JSON 存盘 | 需要（一次性） |
| 重放 | 读 JSON，走同一套 from_anthropic | 不需要 |
| 回归 | 每次 CI 跑重放，断言轨迹不变 | 不需要 |
关键点：重放走的仍是生产代码里的 from_anthropic。所以录制的 JSON 一旦更新（供应商改了字段），测试会立刻失败——它同时是回归测试和「协议漂移」警报器。
2.2.7 夹具 JSON 的结构
夹具就是一个 JSON 文件，外层是元信息，frames 是逐轮响应：
{
"model": "claude-opus-5-5",
"recorded_at": "2026-09-24T10:00:00+08:00",
"frames": [
{
"id": "msg_01AAA", "type": "message", "role": "assistant",
"model": "claude-opus-5-5",
"content": [
{"type": "text", "text": "我先查一下时间。"},
{"type": "tool_use", "id": "toolu_1", "name": "get_time",
"input": {"tz": "Asia/Shanghai"}}
],
"stop_reason": "tool_use", "stop_sequence": null,
"usage": {"input_tokens": 41, "output_tokens": 22}
},
{
"id": "msg_01BBB", "type": "message", "role": "assistant",
"model": "claude-opus-5-5",
"content": [{"type": "text", "text": "现在是北京时间 10:00。"}],
"stop_reason": "end_turn", "stop_sequence": null,
"usage": {"input_tokens": 73, "output_tokens": 11}
}
]
}
frames 数组的顺序就是调用的先后顺序。recorded_at 不是装饰——它让你知道这份录像有多旧，供应商若改了响应形状，你能迅速定位是不是录像过期。
2.2.8 RecordedModel：离线重放
RecordedModel 实现同一份 Model 协议，逐帧重放：
# minagent/recorded.py
from __future__ import annotations
import json
from pathlib import Path
from anthropic.types import Message as AnthropicMessage
from minagent.anthropic_adapter import from_anthropic
from minagent.types import Message, ModelResponse
class RecordedModel:
"""离线重放：读取录制好的响应 JSON，逐帧经适配器映射回 ModelResponse。
零凭据、零网络，用来做确定性回归测试。
"""
def __init__(self, fixture_path: str | Path) -> None:
data = json.loads(Path(fixture_path).read_text(encoding="utf-8"))
self.model = data["model"]
self.recorded_at = data["recorded_at"]
self._frames = data["frames"]
self._i = 0
def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
if self._i >= len(self._frames):
raise RuntimeError("录像已放完：真实对话比录制时更长")
raw = self._frames[self._i]
self._i += 1
return from_anthropic(AnthropicMessage.model_validate(raw))
真跑（读 2.2.7 的夹具，逐轮打印）：
模型: claude-opus-5-5 录制于 2026-09-24T10:00:00+08:00
[1] stop=tool_use text='我先查一下时间。' tools=['get_time'] usage={'input_tokens': 41, 'output_tokens': 22}
[2] stop=end_turn text='现在是北京时间 10:00。' tools=[] usage={'input_tokens': 73, 'output_tokens': 11}
它和 AnthropicModel 实现的是同一个协议，所以 run(model=...) 里可以无缝互换——真实路径与离线路径共用一份循环代码，这正是 2.1 节协议设计的回报。
2.2.9 关于 OpenAI 的对照
很多读者会问「那 OpenAI 呢」。本节只讲接口差异，不给可运行代码，原因很直接：本机 api.openai.com 不可达，把它写成主路径会跑不通。差异主要是三处：
- system 的位置：OpenAI 把 system 当成
messages里的一条（role: "system"），不像 Anthropic 放在顶层参数。映射时不必单独抽取。 - 工具调用形态：OpenAI 的工具调用挂在
message.tool_calls字段上，参数是字符串化的 JSON（需要json.loads），而 Anthropic 是 content block 里的结构化input。 - 停止原因：OpenAI 用
finish_reason（stop/tool_calls/length），与我们的三常量要做一张映射表。
换供应商时，要改的只有适配器，不改协议、不改循环——这就是把供应商隔离在协议之后的全部意义。
真实模型下会怎样
真实调用还要处理一堆本节故意省略的事：限流返回 429 时要退避重试、请求超时要重发、usage 里会有缓存命中与思考 token、长响应要改流式。但这些都不改变协议——它们要么属于第 14 章的可靠性包装器，要么属于第 7 章的流式。本节要建立的信念是：先把形状统一，再逐层加策略。真实与离线共用同一份 from_anthropic，就是你敢在 CI 里断言「线上逻辑没坏」的底气。
阅读导航：上一节：2.1 Model 协议设计 · 下一节：2.3 Token 预算与截断策略 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。