第31章 接通第一根管道——ModelClient

本章导读： 你会写 src/types.ts 的 ModelClient 接口和 ModelEvent 类型，再在 src/model-client.ts 里把 Anthropic SDK 的事件翻译成内部事件。核心是一条纪律：业务代码永远不直接碰 SDK 的事件命名。

范围定了，小a打开编辑器，第一件事却是对着 types.ts 发呆。他问老z：“模型接入怎么写？直接调 SDK 不就行了？”

“行，但你会后悔的。”老z说，“先隔离模型协议。src/types.ts 的 ModelClient.stream() 产出内部 ModelEvent；src/model-client.ts 的 AnthropicModelClient 才依赖官方 SDK。业务代码永远不读 SDK 的事件命名。”

小a：“那中间这层，到底隔离了什么？”

先看内部消息契约

老z在 types.ts 里敲出一段：

// src/types.ts，节选
export type ModelEvent =
  | { type: "text"; text: string }
  | { type: "tool_call"; call: ToolCall }
  | { type: "done" };

“文本、工具调用和完成被拆开，是因为流中它们到达的时间不同。”老z说，“done 不携带‘回答一定有效’的结论；循环还要检查是否累计文本或工具调用。这个小契约使业务代码不需读取 SDK 的事件命名。”

小a追问：“那为什么不干脆设计成一个 message 事件，把文本、调用、完成一次带齐？”

“因为到达时间不同，下游能消费的时机不同。”老z说，“你把它合成一个事件，要么等整条流收完再发——那 stream() 就不配叫流式了；要么发半成品，让循环去猜哪一段能执行。三种事件对应三种‘时间上可独立消费’的事实：text 一到就能追加显示；tool_call 一到就代表一个完整、可执行的调用；done 一到代表协议层收尾。事件的粒度，由‘下游最早能做什么’决定，不由 provider 的事件长什么样决定。”

设计问题	本产物的选择	放弃的替代方案	放弃的理由
事件粒度	三种原子事件	一条聚合 message	聚合意味着等收齐，或发半成品
参数完整性	tool_call 只发完整 input	逐段增量参数	下游无从判断何时可执行
done 语义	只代表流收尾	代表“回答有效”	有效性要循环自行检查
业务可见协议	内部三事件	SDK 原始事件名	换厂商时业务层零改动

老z随手画了一张映射示意：

Anthropic SDK 原始事件                    内部 ModelEvent
─────────────────────────────            ──────────────
content_block_delta (text_delta) ───────► { type:"text" }
content_block_start (tool_use)
content_block_delta (input_json_delta)     （攒着，不产事件）
content_block_stop                 ──────► { type:"tool_call" }
message_delta / message_stop       ──────► { type:"done" }

“注意中间那块：input_json_delta 在映射表里没有对应行——它不产内部事件，只更新内部状态。SDK 事件不是一一对应到内部事件，而是一段状态机的输入。”

消息映射不是类型断言

小a盯着 toolInput 函数看了半天，问：“工具输入在 TypeScript 里是 unknown，为什么映射时还要检查？类型不都标好了吗？”

“因为模型和网络都能给出任意 JSON。”老z打开 model-client.ts：

function toolInput(input: unknown): Record<string, unknown> {
  if (input === null || typeof input !== "object" || Array.isArray(input)) {
    throw new Error("tool input must be an object");
  }
  return input as Record<string, unknown>;
}

“第一行保留不可信类型；条件排除三类 JSON 非对象；最后的断言只发生在运行时检查之后。”老z说，“这仍不验证字段是否符合某个工具 schema——字段验证属于工具执行层，不是这里。”

小a：“那这层到底防什么、防到什么程度？”

“防‘结构性的不可能’，不防‘业务上的不合法’。”老z说，“工具输入必须是 JSON 对象——null、数组、字符串在结构上就不可能是工具参数。这里只拒结构，不拒业务：{ path: 42 } 这种字段类型错的输入，这一层会放行，留给 execute 里的 stringField 去拒。两层防线，一层管‘它是个对象吗’，一层管‘字段对吗’。 对照真实代码能看见层次：model-client.ts 的 toolInput 只做结构检查；tools.ts 的 read.execute 第一行是 const path = stringField(input, 'path')。每一层只答自己该答的问题，错位就变成重复校验或漏校验。 还有一个隐藏好处：toolInput 抛的 tool input must be an object 不满足 toModelEvents 里 startsWith("Anthropic") 的判断，会被包成 tool_use block N returned non-object input 的协议错误——错误消息带上 block index，定位时才知道是哪一段流出了问题。”

流式工具参数的状态

小a：“那流式输出呢？工具参数是分好几段到的，我怎么知道什么时候凑齐了？”

“这就到了最绕的地方——流式工具参数的状态。”老z指着 AnthropicModelClient.stream()：

if (event.type === "content_block_delta" && event.delta.type === "input_json_delta") {
  activeTool.json += event.delta.partial_json;
}
if (event.type === "content_block_stop" && activeTool) {
  input = JSON.parse(activeTool.json) as unknown;
}

“正常序列是 tool block start 创建 { index, id, name, startInput, json }，同 index 的多个 delta 追加 JSON，同 index 的 block stop 才解析并产生内部 call。”老z说，“start 已带完整对象且没有 delta 时，直接使用该对象，所以零参数工具的 {} 也有效。任何较早时刻的 delta 字符串都可能不完整，不能拿去执行。”

小a：“那如果流中途出问题呢？”

“解析失败、非对象输入、未知 index 的 JSON delta，或流结束时仍有未关闭的工具 block，都会中断该响应；text block 的 stop 不会关闭工具 block。”

“多个工具同时来呢？”小a追问。

“多个工具 block 可以交错到达，不能用单个‘当前工具’变量关联它们：工具调用按各自 stop 的顺序发出，但参数永远由同一 index 的 start/delta/stop 组成。”

“那一个工具 block 的生命周期，画出来长什么样？”小a问。

老z画了一张状态图：

              content_block_start (tool_use)
                        │ 创建 ActiveTool{id,name,startInput,json:"",received:false}
                        ▼
              ┌───────────────────────┐
              │   未收到 delta 的块     │
              │  （startInput 权威）    │
              └───────────────────────┘
                        │
        input_json_delta（同 index）
                        │ json 追加 + received=true
                        ▼
              ┌───────────────────────┐
              │    已收到 delta 的块    │
              │     （json 权威）       │
              └───────────────────────┘
                        │
                        │ content_block_stop
                        ▼
        JSON.parse(json) → toolInput → yield tool_call → 从 Map 删除

“两条规则守住它：只有 block stop 才允许解析和产出，delta 永远只追加不解析。 为什么不每次 delta 都试一次 JSON.parse？因为 partial_json 常常是还没闭合的半截——{"path":"a" 这种在任何时刻都解析不了，逐段试解析只会吐出一堆噪音错误，而真正的坏 JSON 要等 stop 时才能确认。”

小a：“那交错到达的两个工具呢？”

同一流里两个工具 block 交错到达（model-client.test.ts 覆盖的场景）

start(1) → start(3) → delta(3) → delta(1) → stop(1) → stop(3)
   │          │          │          │           │          │
activeTools: {1}        {1,3}      {1,3}      {1,3}       {3}       {}
                                                                     │
产出事件:                                         tool_call(1)  tool_call(3)  done

“关键在最后一行：index 1 和 index 3 各自攒自己的 JSON，互不干扰。如果用一个‘当前工具’变量，第二个 start 会覆盖第一个，第一个工具的参数就丢了。 这也是为什么 toModelEvents 要用 Map<index, ActiveTool> 而不是一个变量——我们下一节细看。”

toModelEvents 是一台小型状态机

小a：“那真实的 toModelEvents 到底怎么处理这些？我照着概念写个 activeTool 变量够吗？”

“不够。”老z打开真实的 model-client.ts，指着中间那段：“它维护的不是一个变量，而是一个 Map<index, ActiveTool>，因为多个工具 block 可能交错到达：”

// examples/mini-agent/src/model-client.ts，节选
const activeTools = new Map<number, ActiveTool>();
let stopReason: Anthropic.StopReason | null | undefined;
let toolCallCount = 0;
let receivedMessageStop = false;

“每个开着的 block 都按它的 index 存着。你看完整的 ActiveTool 字段：”

type ActiveTool = {
  id: string;
  name: string;
  startInput: unknown;
  json: string;
  receivedInputJsonDelta: boolean;
};

小a盯着 receivedInputJsonDelta 问：“这个布尔是干嘛的？”

“这是最容易漏的细节。”老z说，“content_block_start 事件里的 input 可能已经带了完整参数（比如零参数工具 {}），也可能只是开头。receivedInputJsonDelta 记录‘这次有没有收到过增量 JSON’——**只有收到过，才需要用拼接的 json 重新 JSON.parse；否则直接采用 startInput。**你看这段：”

// examples/mini-agent/src/model-client.ts，节选
let input: unknown = activeTool.startInput;
if (activeTool.receivedInputJsonDelta) {
  try {
    input = JSON.parse(activeTool.json) as unknown;
  } catch {
    throw protocolError(
      `tool_use block ${event.index} had invalid input_json_delta JSON`,
    );
  }
}

“这也解释了为什么零参数工具的 {} 有效——它可能根本没有 delta，直接走 startInput。而一旦有 delta，json 就是权威，拼不完整或解析失败都直接抛错。”

小a：“那流中途断了呢？”

老z把错误路径数了一遍：“input_json_delta 到达时没有对应的开着的 block、重复的 tool_use start、block stop 时 JSON 解析失败、流结束时还有未关闭的 block、没收到 message_stop、stop reason 不支持或与实际调用数矛盾——每一种都 throw protocolError，不会产出 done。”

// examples/mini-agent/src/model-client.ts，节选
if (activeTools.size > 0) {
  throw protocolError(
    `stream ended with unclosed tool_use block(s): ${[...activeTools.keys()].join(", ")}`,
  );
}
if (!receivedMessageStop) {
  throw protocolError("stream ended before message_stop");
}

“小a注意到 protocolError 统一带 Anthropic stream protocol error: 前缀——让调用方能一眼分辨是协议问题还是业务问题。这就是 fail loud：流状态不完整，宁可整个响应失败，也不把半截参数当完整调用交出去。”

老z把这张状态机的错误路径收成一张表，全部带 Anthropic stream protocol error: 前缀：

触发	错误消息（前缀略）	为什么不是静默跳过
input_json_delta 无对应开着的 block	input_json_delta arrived without an open tool_use block at index N	顺序已损坏，静默会串流
重复 tool_use start	duplicate tool_use start for block N	同一 index 两个 start，参数归属不清
block stop 时 JSON 解析失败	tool_use block N had invalid input_json_delta JSON	半截参数不能当完整调用
流结束仍有未关闭 block	stream ended with unclosed tool_use block(s): …	缺结果的调用让模型悬空
没收到 message_stop	stream ended before message_stop	无法确认本轮完整结束
缺 stop reason	message_delta did not provide a stop_reason	无法判断该不该继续循环

“对照 model-client.test.ts，上面每一行都有离线测试盯着。协议错误不靠人工观察，靠测试红。”

小a：“为什么坚持用 Map 而不是单个变量？我写个 activeTool 不够吗？”

方案	关联依据	交错到达时
单个 activeTool 变量	“当前正在拼的工具”	第二个 start 覆盖第一个，参数丢失
Map<index, ActiveTool>	provider 给的 content block index	各 index 独立累积，互不干扰

“单变量不是每次都错——只有交错时才错，这正是它诱人的地方。但交错是 provider 允许的正常情况，associates simultaneous tool JSON with their content-block indexes 这个测试专门盯着它。状态机存在的理由，往往不是主路径，而是那条你不在意的交叉路。”

停止原因是提交前提

小a在日志里看到 stop_reason 字段，随口说：“这个不就是个展示信息吗？”

“不是。”老z的语气严肃起来，“message_delta.delta.stop_reason 不是展示信息。此适配器目前只接受两种完整语义：**

• end_turn：必须没有已完成的工具调用；
• tool_use：必须至少有一个已完成的工具调用。

“max_tokens、pause_turn、refusal、stop_sequence、缺失的 stop reason，以及上述两种 reason 与实际工具调用不一致，都会抛出带 provider 协议上下文的错误，不会产出 done。”老z说，“这避免截断文本被下游循环误当成完整回答。message_stop 或迭代器自然结束时若工具 block 尚未关闭，也同样报错。”

老z把 stop reason 的语义列成表：

stop_reason	允许的条件	语义	循环的动作
end_turn	无已完成工具调用	模型回答完整	提交文本，返回 complete
tool_use	至少一个已完成调用	模型请求工具	分派工具，下一轮继续
max_tokens	不支持	输出被截断	抛错，不产 done
pause_turn	不支持	模型保留轮次	抛错，不产 done
refusal	不支持	模型拒绝	抛错，不产 done
stop_sequence	不支持	自定义停止词	抛错，不产 done
缺失	不支持	协议不完整	抛错，不产 done

小a：“max_tokens 在真实产品里很常见吧？一刀切拒绝是不是太保守了？”

“因为 max_tokens 意味着输出被切断——文本说到一半没了，或者工具参数拼到一半。把它当完整回答提交是误导；把它当工具请求分派，参数可能根本不可执行。当前产物宁可整轮失败，也不在‘截断’上假装‘完整’。 将来要支持，得先设计‘半截文本要不要保留、要不要自动重试、重试几次’这些语义——那是新需求，不是这一章的默认值。”

正常、中止与证据边界

小a：“那取消呢？用户点了停止，我该怎么办？”

“中止路径有两个 checkpoint。”老z打开 agent.ts：

// examples/mini-agent/src/agent.ts，节选
for await (const event of options.client.stream(request)) {
  if (aborted(options.signal)) {
    return { messages, stopReason: "aborted" };
  }
  // 收集本轮事件
}
if (aborted(options.signal)) {
  return { messages, stopReason: "aborted" };
}

“第一个判断防止在已知取消后继续吸收事件；第二个判断位于流完成与 assistant commit 之间。它们不提供任意时刻的强制抢占：若底层异步迭代器永不返回且不响应 signal，宿主仍无法越过正在等待的 for await。”

小a：“所以取消是……协作式的？”

“对。AbortSignal 会传给 SDK，但取消仍是协作式的：客户端若不及时响应 signal，循环只能在事件返回或流结束后的 checkpoint 观察取消。已经收到的部分文本不会被提交为完整 assistant。”

老z把整条链路的验证边界列了一张表：

事实	本轮如何验证	未验证
纯映射拒绝 null input	fake 单元测试	SDK 服务端接收
工具按 index 累积、start input 和 block 闭合	构造的离线 SDK 事件单元测试	真实 Provider 流顺序
不支持/缺失 stop reason 不会给出 done	构造 max_tokens、缺 reason 等离线测试	Provider 对异常响应的实际投递
fake 回路工作	npm test	真实 API、费用、限流
流结束竞态不提交 assistant	非协作 fake stream 回归测试	SDK 的取消延迟上限

“记住这张表的第二列和第三列——‘本轮如何验证’是承诺，‘未验证’是诚实。”

“那取消在循环里到底落在哪些点？”小a问。

老z把 runAgent 一个 turn 的检查点画成图：

runAgent 单个 turn 的取消检查点（agent.ts）

turn 开始
  │  ① aborted?（本轮循环前）──────────是──► return aborted
  ▼
for await (client.stream(...))
  │  ② aborted?（每个事件后）────────────是──► return aborted
  ▼
流结束
  │  ③ aborted?（assistant 提交前）───────是──► return aborted（竞态不提交）
  ▼
提交 assistant（text=="" && calls==0 时抛错）
  ▼
for call of calls
  │  ④ aborted?（每个工具执行前）──────────是──► 记录 aborted before execution
  │  ⑤ aborted?（审批之后）
  ▼
tool.execute(call.input, signal)
  │  ⑥ aborted?（工具执行后）──────────────是──► 记录后续未执行
  ▼
下一个 call / 下一轮

“注意 ③ 这道闸：流已经自然结束、但 signal 恰好在最后一段事件到达时被置位——不检查这一下，半截文本就会被当完整 assistant 提交。cancellation.test.ts 的 abort after a non-cooperative stream completes prevents assistant commit 就是专门为这道闸写的。每个检查点都是一次‘宁可少提交，不可多提交’的取舍。”

小结

接通模型流的关键，不是“发请求收响应”，而是在厂商协议和内部事件之间留一道适配缝。ModelClient 只暴露一个 stream() 方法，接收 system、messages、tools 和可选 signal，产出三种内部事件：text、tool_call、done。循环只消费这三种事件，永远不必知道背后是 Anthropic、OpenAI 还是别的什么——适配层的全部价值就在这一层薄薄的隔离上。事件的粒度不是随便定的：text 一到就能显示，tool_call 一到就代表一个完整可执行的调用，done 只承诺协议收尾、不承诺回答有效——后者要循环自己检查文本和调用数。

toModelEvents 是一台小状态机，它的存在理由是那几条交叉路。工具参数按 provider 的 content block index 用 Map 累积，receivedInputJsonDelta 区分“参数跟着 start 一次给全”和“靠 delta 拼出来”两种来源；只有 block stop 才解析并产出，任何时刻的半截 JSON 都不能拿去执行。协议错误统一带 Anthropic stream protocol error: 前缀，宁可靠测试红、宁可整轮失败，也不把半截参数当完整调用交出去。stop reason 同样严格：end_turn 和 tool_use 两种完整语义之外，max_tokens、refusal 都直接抛错——截断的文本不该被当成完整回答。

取消是协作式的，不是任意抢占。AbortSignal 会传给 SDK，但循环只在几个固定检查点上观察取消：每轮流开始前、流进行中、流结束与 assistant 提交之间、每个工具执行前后。流结束后的那道闸尤其重要——signal 恰好在最后一段事件时到达，不做检查就会把半截文本提交成完整 assistant。每个检查点都是一次取舍：宁可少提交，不可多提交，signal 的传播路径因此可预测，调试时能定位中断发生在哪一道闸。

但这道缝不会自动适配所有厂商。“adapter 能跑”只证明当前这家厂商的事件格式被正确翻译了，换一家就要重写消息角色转换、partial JSON 拼接、停止语义映射——每一项都得用 fake stream 配合真实服务的受控测试来验证。当前产物的边界要诚实：真实 Anthropic 端到端流尚未验证，超时、重试、用量统计也没实现。把“adapter 接好了”当成“全厂商可用”，是这一章最容易踩的坑。

阶段验收

运行 npm test 和 npm run typecheck；不要为此设置真实凭据。