模型调用：在课程协议和 Provider API 之间转换

#你将得到什么

第 04 章写出的 ScriptedModel 接收 AgentContext，返回 ModelEvent。真实模型也要接到 这个位置上。不过，它在进程外面：消息要先写成 HTTP 请求，回复则以 SSE 字节流的 形式回来。

这一章只跟随一条固定主线：用户要求“读取 README.md”，Provider 用 id: "call-1"、index: 0 提出 read，参数分成 '{"path":' 与 '"README.md"}' 两段返回。发送方向从课程消息进入 fetch；返回方向则沿同一个 id 和 参数，依次经过 SSE payload、ProviderChunk、ModelEvent，最后形成 assistant 中的 tool call。

call-1 是 Chapter 05 聚焦测试使用的 Provider fixture id；序章离线轨迹写作 call_1， 两者不是同一次运行。跨章保持不变的是 read("README.md") 这项动作，以及同一轮内 id 从 Provider 输出一直进入 assistant 的配对关系。

AgentContext("读取 README.md") → ProviderRequest → HTTP
AssistantMessage(call-1) ← ModelEvent ← ProviderChunk(index 0) ← SSE payload

这里会出现两个新名字。adapter 负责课程协议与 Provider 语义之间的转换； transport 负责 HTTP、SSE 和外部 JSON。写完以后，ScriptedModel 与真实模型都能 通过同一个 Model.stream() 交给上层使用。

本章只讨论模型边界的转换。工具参数怎样验证、工具怎样执行，留到第 06 章。实现会 修改 packages/pi-course/src/types.ts 和 packages/pi-course/src/provider-adapter.ts。

#先建立全景

先看发送方向。课程内部的一条 user message 是一个带 content block 的对象：

{
  role: "user",
  content: [{ type: "text", text: "读取 README.md" }],
  timestamp: 1,
}

OpenAI-compatible API 接收的同一条消息更扁平：

{ role: "user", content: "读取 README.md" }

toProviderMessages() 读取前一个对象，写出后一个对象。system prompt、assistant 的 工具调用和 tool result 也在这里转换。完成转换后，它们和模型名一起组成 ProviderRequest。

返回方向多两层。Provider 返回主线参数的第一段时，一条 SSE data 是：

data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"read","arguments":"{\"path\":"}}]},"finish_reason":null}]}

transport 先从字节流中取出完整的 data:，解析 JSON，并检查课程实际用到的字段。 上面的 payload 会变成一个较小的对象，id、index 和参数分片都保持原值：

{
  type: "tool",
  index: 0,
  id: "call-1",
  name: "read",
  argumentsDelta: '{"path":',
}

下一条 SSE payload 仍指向 index: 0，只携带第二段参数 '"README.md"}'；transport 据此产出第二个 tool ProviderChunk。adapter 不再接触 choices[0].delta.tool_calls 之类的外部字段。它按 index 找回 call-1 的 buffer， 把两段参数连成 {"path":"README.md"}，再发出 toolcall_delta、toolcall_end 和 done。最终 assistant 中仍是 id: "call-1" 的 read 调用。

AgentContext
    │  toProviderMessages()
    ▼
ProviderRequest
    │  transport.stream()
    ▼
raw SSE(call-1/index 0) ── transport ──→ ProviderChunk(index 0)
ProviderChunk(index 0) ── adapter ──→ ModelEvent(contentIndex 0)
                                      └─→ AssistantMessage(call-1)

ProviderChunk 是两段代码之间的接缝：

export type ProviderChunk =
  | { type: "text"; delta: string }
  | {
      type: "tool";
      index: number;
      id?: string;
      name?: string;
      argumentsDelta?: string;
    }
  | {
      type: "finish";
      reason: "stop" | "length" | "tool_calls";
      usage?: Partial<Usage>;
    };

真实 transport 和测试使用的 fixedTransport() 都实现同一个接口：

export interface ProviderTransport {
  stream(
    request: ProviderRequest,
    options: { signal?: AbortSignal },
  ): AsyncIterable<ProviderChunk>;
}

因此 adapter 可以先在离线数组上写完。网络解析接好以后，它处理的仍是同样三个 chunk 分支。

{
  type: "tool",
  index: 0,
  id: "call-1",
  name: "read",
  argumentsDelta: '{"path":',
}
{ type: "tool", index: 0, argumentsDelta: '"README.md"}' }

#把消息写成 Provider 请求

模型请求还需要知道本轮可以调用哪些工具。ToolDefinition 保存工具名称、说明和 交给模型的 JSON Schema：

export interface ToolDefinition {
  name: string;
  description: string;
  parameters: Record<string, unknown>;
}

export interface AgentContext {
  systemPrompt?: string;
  messages: AgentMessage[];
  tools?: ToolDefinition[];
}

这里的 schema 只会进入模型请求。第 06 章会为它配上运行时验证器。

看一份稍完整的 context：

const context: AgentContext = {
  systemPrompt: "回答要简洁。",
  messages: [
    userMessage("读取 README.md"),
    assistantMessage([
      {
        type: "toolCall",
        id: "call-1",
        name: "read",
        arguments: { path: "README.md" },
        rawArguments: '{"path":"README.md"}',
      },
    ], "toolUse"),
    {
      role: "toolResult",
      toolCallId: "call-1",
      toolName: "read",
      content: [text("# Project")],
      isError: false,
      timestamp: 1,
    },
  ],
};

toProviderMessages(context) 写出的数组是：

[
  { role: "system", content: "回答要简洁。" },
  { role: "user", content: "读取 README.md" },
  {
    role: "assistant",
    content: null,
    tool_calls: [{
      id: "call-1",
      type: "function",
      function: {
        name: "read",
        arguments: '{"path":"README.md"}',
      },
    }],
  },
  {
    role: "tool",
    tool_call_id: "call-1",
    name: "read",
    content: "# Project",
  },
]

四种 role 在这个例子里都出现了。system prompt 变成第一条 system message；user 的 text block 合并成字符串；assistant 的工具调用进入 tool_calls；课程中的 toolResult 则写成 Provider 的 role: "tool"，并继续使用原来的 call id。

工具参数优先采用 rawArguments：

arguments:
  call.rawArguments ?? JSON.stringify(call.arguments ?? {})

rawArguments 保存 Provider 原先给出的字符序列。重新序列化一个已经解析过的对象， 可能改变空格或字段顺序，也无法表示被截断的原文。只有没有原文时，才从 arguments 生成 JSON。

toProviderTools() 做另一项转换，把每个 ToolDefinition 包进 Provider 所需的 { type: "function", function: ... }。没有工具时返回 undefined，请求体也就不会 带一个空数组。这两个转换都是纯函数；传入的 context 在调用后保持原样。

npm run build -w @pi/course
node --test --test-name-pattern="出站转换" \
  packages/pi-course/dist/test/05-*.test.js

#从 ProviderChunk 读回模型回复

发送方向处理的是完整对象。返回方向是一段会逐渐长出来的回复。先把主线的两个参数分片 直接交给离线 fixedTransport()：

[
  {
    type: "tool",
    index: 0,
    id: "call-1",
    name: "read",
    argumentsDelta: '{"path":',
  },
  { type: "tool", index: 0, argumentsDelta: '"README.md"}' },
  { type: "finish", reason: "tool_calls", usage: { output: 2 } },
]

第一个 chunk 让 adapter 建立 toolBuffers.get(0)：其中保存 id: "call-1"、 name: "read"、contentIndex: 0 和参数原文 '{"path":'。第二个 chunk 仍用 Provider index 0 找回这个 buffer，并把参数追加成 '{"path":"README.md"}'。对应事件是：

ProviderChunk #1
  → toolcall_delta(contentIndex 0, delta '{"path":')
  → partial: toolCall(call-1, rawArguments '{"path":')

ProviderChunk #2
  → toolcall_delta(contentIndex 0, delta '"README.md"}')
  → partial: toolCall(call-1, rawArguments '{"path":"README.md"}')

finish(tool_calls)
  → JSON.parse('{"path":"README.md"}')
  → toolcall_end(contentIndex 0, call-1)
  → done(toolUse)

最终 assistant 仍携带从 SSE 开始的同一个 id 和参数：

{
  role: "assistant",
  content: [{
    type: "toolCall",
    id: "call-1",
    name: "read",
    arguments: { path: "README.md" },
    rawArguments: '{"path":"README.md"}',
  }],
  stopReason: "toolUse",
  // provider、model、usage、timestamp 省略
}

这里 Provider index 与 canonical contentIndex 都是 0，只是因为主线只有一个 content block；两者的含义并不相同。前者把网络中的后续分片送回同一个工具，后者指出 tool call 位于 assistant content 的哪个槽位。

#局部 fixture：只看文本 partial

主线闭合以后，可以用一个纯文本 fixture 单独观察 partial 快照：

[
  { type: "text", delta: "正在" },
  { type: "text", delta: "读取" },
  { type: "finish", reason: "stop", usage: { output: 2 } },
]

adapter 依次发出 start、两个 text_delta 和 done。第一个 delta 之后，partial 中 的文本是“正在”；第二个 delta 之后，它变成“正在读取”。每个事件保存的是该时刻 已经形成的完整 partial，而非最后一个字符串片段。这个 fixture 只隔离文本分支，不替换 call-1/index 0 的章节主线。

#诊断 fixture：用 index 4/2 分开两种顺序

另一个 fixture 故意让文本与两个工具交错，并使用不连续的 Provider index：

tool index 4   {"path":
text           先读
tool index 2   {"query":
tool index 4   "README.md"}
text           ，再查
tool index 2   "pi"}
finish         tool_calls

它用于诊断两种 index 是否被混淆，不是新的业务主线。canonical content 仍按照内容 第一次出现的先后排列：

content[0] = tool index 4
content[1] = text
content[2] = tool index 2

第一次遇到工具 4 时，adapter 为它建立一个 ToolBuffer，把当时的 content.length 记为 contentIndex。以后再收到 index 4，只更新这个 buffer 和 content[0]。工具 2 同样保留自己的 buffer，但它第一次出现得更晚，所以占据 content[2]。

一次 stream() 需要保存这些值：

这些值属于一次请求，写在 stream() 内。每次 delta 先更新对应槽位，再用 structuredClone(content) 留下事件快照。后来的字符不会反过来改变已经发出的 partial。

finish 决定工具参数怎样收尾。tool_calls 表示 Provider 已经完成工具调用，此时 adapter 解析完整 JSON，并发出 toolcall_end。length 表示输出被截断；参数原文会 留下来，stop reason 也保持 length，后续 Agent Loop 不会执行它。

transport 抛出的异常会沿用当前 content 生成 error message。取消对应 aborted，其他异常对应 error。终态事件与 stream.result() 使用同一条最终消息。

npm run build -w @pi/course
node --test --test-name-pattern="normalized transport" \
  packages/pi-course/dist/test/05-*.test.js

到这里，adapter 已经可以完全离线运行。余下工作是让网络响应产生同样的 ProviderChunk。

#从 SSE 字节流读出 ProviderChunk

fetch 返回 ReadableStream<Uint8Array>。一次 reader.read() 只表示“这次读到了 一些字节”，它不对应一条 SSE event。主线中的第一条 data 可以在 call-1 中间被切成 两个网络 chunk。下面用引号表示每次解码出的字符串；chunk #1 末尾没有换行：

chunk #1
'data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call-'

chunk #2
'1","type":"function","function":{"name":"read","arguments":"{\"path\":"}}]},"finish_reason":null}]}\n\n'

网络边界只是 chunk #1 | chunk #2，不会在 JSON 数据中加入字符。transport 维护一个 字符串 buffer，新字节由带 { stream: true } 的 TextDecoder 解码后依次 append：

append chunk #1 后
buffer = 'data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call-'
         没有 \n，暂时取不出一行

append chunk #2 后
buffer = 'data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"read","arguments":"{\"path\":"}}]},"finish_reason":null}]}\n\n'
         第一个 \n 结束 data 行；第二个 \n 形成空行并结束这条 SSE event

这里的 \n\n 是 SSE 协议本来就有的“行结束 + 空行”，与两次 reader.read() 的切分 位置无关。空行出现后，此前收集的 data: 行才构成一条完整 SSE payload。

网络字节块            没有协议含义
    ↓ TextDecoder
文本行                 识别 data:
    ↓ 空行结束一组 data
SSE payload 字符串
    ↓ JSON.parse + 字段检查
ProviderChunk

JSON.parse() 的结果从 unknown 开始。主线走的是 delta.tool_calls；先用一个 主线完成后的纯文本 payload，把逐层收窄路径看短一些：

rawChunk: unknown
  → isRecord(rawChunk)
  → Array.isArray(rawChunk.choices)
  → choices.length === 1
  → choice = choices[0]，isRecord(choice)
  → isRecord(choice.delta)
  → content = choice.delta.content，typeof content === "string"
  → { type: "text", delta: content } satisfies ProviderChunk

例如 {"choices":[{"delta":{"content":"正在读取"}}]} 会沿这条路径得到 { type: "text", delta: "正在读取" }。若外部数据改成 {"choices":[{"delta":{"content":42}}]}，验证停在 typeof content === "string"， 抛出 Invalid provider chunk: choices[0].delta.content must be a string；数字不会进入 ProviderChunk。

其余字段复用同一种模式：tool_calls 先检查数组，再逐项检查 record，其中 index 必须是非负安全整数，id、name、arguments 在出现时必须是字符串； finish_reason 检查 stop/length/tool_calls 联合；usage 先检查 record，再检查三个 非负安全整数。于是主线 payload 通过这些检查后，才成为前文的 call-1/index 0 tool ProviderChunk。普通流只接受一个 choice；空 choices 专门承载尾随 usage。

finish_reason 与 usage 可能来自不同 payload。transport 读到结束原因后先记住它， 继续等待尾随 usage；遇到 [DONE] 或输入结束，再合成唯一的 finish chunk。整个流 没有出现结束原因时，adapter 会用已经积累的 partial 形成 error 终态。

读取前后都检查 abort signal。reader 在 finally 中取消并释放 lock，所以正常结束、 解析异常和取消会走过同一个资源清理位置。

npm run build -w @pi/course
node --test --test-name-pattern="SSE" \
  packages/pi-course/dist/test/05-*.test.js

#发出 HTTP 请求

transport 最后需要写出完整的 fetch 选项。base URL 去掉末尾斜杠后，加上 /chat/completions：

{
  method: "POST",
  headers: {
    Accept: "text/event-stream",
    Authorization: `Bearer ${apiKey}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    ...request,
    stream: true,
    stream_options: { include_usage: true },
  }),
  signal,
}

API key 由 transport 配置持有。请求发出时，它进入 Authorization header；body、 AgentContext 和 canonical message 中都没有这项数据。底层 fetch 的异常文本可能 包含认证信息，因此 transport 在错误离开这一层之前，把 key 替换成 [redacted]。 AbortError 保持原来的名称，adapter 才能识别取消。

课程测试注入离线 fetch，直接记录 URL、header、body 和 signal。它不会调用真实 Provider，也不需要 API key 或网络费用。

npm run build -w @pi/course
node --test --test-name-pattern="fetch transport" \
  packages/pi-course/dist/test/05-*.test.js
node --test packages/pi-course/dist/test/05-*.test.js

#故意把它弄坏

这一节是完成实现后的诊断实验。把工具的 contentIndex 临时改成 Provider 给出的 chunk.index：

contentIndex: chunk.index

测试数据中的工具 index 按 4、2 到达，中间还有文本。改动以后，第一个 toolcall_delta 会指向 content[4]，而当时真正创建的工具槽位是 content[0]。

#本章验收

npm run build -w @pi/course
node --test packages/pi-course/dist/test/05-*.test.js

#可选迁移练习

#小结

一次真实模型调用现在可以从头读到尾。AgentContext 先变成 Provider 的消息和工具 定义；transport 发出 HTTP 请求，从 SSE 中读出经过检查的 ProviderChunk；adapter 再把这些 chunk 累积成 ModelEvent 与最终 AssistantMessage。

第 04 章的离线模型和这一章的真实模型因此共享 Model.stream()。上层代码不需要 读取 Provider 的 URL、header、choices 或 SSE。下一章会继续处理模型已经提出的 工具调用。