大话 Agent

简体中文

English

简体中文

English

外观

Part II · 源码篇 / Ch 03

源码篇第3章 先开票，后办事——pi-ai ​

老Z递出空白流句柄，小A看着琥珀色光流中的零散事件汇聚成完整消息。

本章导读： 源码路线第 ② 站（核心层）。仍在 pi-ai 包，但深入 api/lazy.ts 和 api/transform-messages.ts。这一站要拆开那个让小A发愣的 AssistantMessageEventStream：厂商零散的增量事件，是怎么被收拢成循环能直接消费的统一形态的。

源码篇第2章走完，小A终于让模型“出发”了。可 Provider 交回来的东西又一次让他愣住：不是一段完整回答，而是一个 AssistantMessageEventStream。他盯着这个流，问老Z：“我到底该怎么接？等它全部结束，还是一个一个拿？”

“都不对。”老Z说，“源码篇第2章 目录与户口——pi-ai已经完成‘模型对象到 Provider’的分派，但 Provider 交回的不是一个完成的字符串，而是 AssistantMessageEventStream。你现在要核对三件事：异步设置为什么不会阻塞入口；一个真实 adapter 如何把厂商流变成统一事件；取消和异常最终以什么状态向上报告。”

本章证据固定在上游 pi 标签 v1.0.0（提交 a13d35a742c6ef8462812a28fbe1d8c8b7431c32）、@earendil-works/pi-ai 1.0.0。本章挑 packages/ai/src/api/openai-responses.ts 作为一个真实 adapter 来读，这不等于声称所有 Provider 或 API 都采用它的字段和失败策略；也不覆盖所有厂商协议、模型定价规则和 WebSocket adapter，更不把通用 retry 工具的存在写成所有调用都会重试。

返回流之前，先把设置留到稍后 ​

“ModelsImpl.stream 与 streamSimple 都调用 lazyStream，函数立刻把一个可迭代的流交到调用者手里，认证解析、动态模块加载、Provider 调用都在后台进行。”老Z说，“所以拿到流不等于‘请求一定已经开始’，它更不是一个普通的 Promise<AssistantMessage>。”

ts
// packages/ai/src/api/lazy.ts，lazyStream（v1.0.0，节选）
export function lazyStream(model, setup): AssistantMessageEventStream {
	const outer = new AssistantMessageEventStream();
	setup()
		.then((inner) => forwardStream(outer, inner))
		.catch((error) => {
			const message = createSetupErrorMessage(model, error);
			outer.push({ type: "error", reason: "error", error: message });
			outer.end(message);
		});
	return outer;
}

小A：“为什么要这样绕一圈？直接等 setup 完成再返回不行吗？”

“逐段看。”老Z指着代码说，“第一行构造 outer，调用点因此同步拿到同一类型的输出，不必区分‘尚在认证’和‘已在接收网络事件’。第二段不 await setup()，而是用 promise continuation 把 forwardStream 接上去，这正是延迟设置：ModelsImpl 里的 requireProvider、applyAuth 和 provider.stream* 都被放进传入的 async 回调。不过‘延迟’是相对同步返回说的，它不承诺等到第一次 for await 才执行，setup() 在函数体里就调用了。”

“第三段把 setup rejection 归一成一个 AssistantMessage：空 content、零 usage、stopReason: "error"、错误文字，随后 push error 事件并 end(message)。于是未知 Provider 或认证失败对消费者呈现为流内错误，而不是调用 Models.stream 时的同步 throw。”

比喻：先开票，后办事

lazyStream 像先给你一张取餐号（outer stream），后厨（setup）慢慢做。做砸了，也是通过“取餐口”告诉你“这单做不了”，而不是在你下单那一刻就摔盘子。

forwardStream 逐个 push 内流事件，迭代结束后若内流带有 .result() 就用它结束 outer。这一规则保留了 adapter 给出的最终消息；别从这里推断任何事件一定包含 text delta。

阶段产物正常下一步设置失败时
Models.stream* 返回outer stream调用者订阅／迭代尚无同步异常
setup()inner async iterableforwardStream.catch 创建 error message
inner eventsstart、delta、done／error 等原样推到 outeradapter 决定事件内容
inner 完结result() 或 undefinedouter.end由 adapter 的终止行为决定

lazy API 是第二层延迟，不是第二种事件协议 ​

小A顺着问：“那 Provider 内部的适配器模块呢？也是懒加载？”

“对，但那是第二层延迟。”老Z打开 lazy.ts：

lazyApi(load, capabilities?) 返回一个 ProviderStreams 对象。其两个流方法都再以 lazyStream 包住 await load() 后的模块方法；v1.0.0 新增的 capabilities 声明这个 API 是否还支持 deferred 响应的取回与取消，本章不展开：

ts
// packages/ai/src/api/lazy.ts，lazyApi（v1.0.0，节选）
const api: ProviderStreams = {
	stream: (model, context, options) =>
		lazyStream(model, async () => (await load()).stream(model, context, options)),
	streamSimple: (model, context, options) =>
		lazyStream(model, async () => (await load()).streamSimple(model, context, options)),
};

“外层 ModelsImpl 的 lazy stream 覆盖的是‘Provider／认证设置’，这里的 lazy stream 覆盖的是‘适配器模块加载’。两层产出的都是同一种 AssistantMessageEventStream，所以会出现嵌套转发。模块 import cache 是否去重由宿主运行时负责，源码注释确实提到了 host import cache，但没为任意运行环境证明过性能结果。”

“收益是未使用的 adapter 不必在根入口加载；代价是模块加载错误推迟到第一次请求，排错时还得把 auth setup、import setup 和网络请求三层分开看。”

真实 adapter 的一次正常事件流 ​

小A提议：“挑个真实 adapter，看它正常怎么走？”

“看 openai-responses.ts。”老Z说，“它的 stream 在函数开始就创建 AssistantMessageEventStream，再用 resolveTranscript 按 compat 处理 transcript：模型不接受对话中途的 system message 时，把所有 system message 合并成开头一条。然后启动一个 async IIFE。它先建一个 stopReason: "pending" 的 output，成功后才推 start、让 processResponsesStream 更新它，再推 done。工具清单也不再从 context.tools 读，而是用 getDeclaredTools 从 system message 里回放。”

ts
// packages/ai/src/api/openai-responses.ts，stream（v1.0.0，节选）
const { data: openaiStream, response } = await retryProviderRequest(
	() => client.responses.create(params, requestOptions).withResponse(),
	{ maxRetries: options?.maxRetries, maxRetryDelayMs: options?.maxRetryDelayMs, signal: options?.signal },
);
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
stream.push({ type: "start", partial: output });
await processResponsesStream(openaiStream, output, stream, model, { /* 节选 */ });
stream.push({ type: "done", reason: output.stopReason, message: output });
stream.end();

“注意这段的顺序。”老Z提醒，“请求经过 retryProviderRequest，但 client 配置 maxRetries: 0 关闭 SDK 自带重试；helper 再按自己的 options.maxRetries 控制另一层重试。两者不是同一个参数。重不重试，仍由 options.maxRetries、信号和 helper 的可重试判定决定，别把代码里有这一层读成‘任何错误必重试’。”

“onResponse 在拿到 HTTP response 之后、推 start 之前触发，提供 status 和转换成 record 的 headers。随后的 processResponsesStream 才是上游事件到统一 delta 的实际转换位置；本章不逐个展开它的 OpenAI event case，那些 case 是这一个 adapter 的细节，不是跨 Provider 的契约。”

一个正常序列可读作：

text
buildParams / createClient
  → retryProviderRequest 发起 Responses 请求
  → 收到 HTTP response，调用 onResponse
  → push start(partial output)
  → processResponsesStream 连续修改 output 并 push text/thinking/tool-call 事件
  → output 得到非 pending、非 aborted/error 的 stopReason
  → push done(message)，end()

“对 Agent 来说，最关键的不是上游 event 叫什么名字，而是它先看到一个可更新的 partial，最后拿到完整的 AssistantMessage。下一章的 streamAssistantResponse 能把 start 写入上下文、把 delta 转成 message_update，靠的就是这一点。”

transform 与 compat：统一形状不等于抹平差异 ​

小A把两个名字摆在一起：“前面 applyAuth 里有 transformHeaders，这里又有 getCompat，它们是同一回事吗？”

“不是。它们解决的层次不同。”老Z说，“ModelsImpl.applyAuth 中的 transformHeaders 是请求发送前的 Models 专用变换；openai-responses.ts 又会针对模型的 compat 建立默认兼容配置。”

ts
// packages/ai/src/api/openai-responses.ts，getCompat（v1.0.0，节选）
function getCompat(model: Model<"openai-responses">): Required<OpenAIResponsesCompat> {
	return {
		supportsDeveloperRole: model.compat?.supportsDeveloperRole ?? true,
		supportsMidConvoSystemMessages: model.compat?.supportsMidConvoSystemMessages ?? false,
		sessionAffinityFormat: model.compat?.sessionAffinityFormat ?? detectSessionAffinityFormat(model),
		supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
		// 其余能力开关已删节
	};
}

“transformHeaders 的输入是已经合并的 header record，适合宿主加追踪或路由信息；它不会改 adapter 的 message 结构。getCompat 从 model.compat 和 provider／baseUrl 推断能力开关，随后被 resolveTranscript、buildParams、grammar tool 输入等路径使用。同一个 API 名下仍可以有 capability 差异，但默认值是 true 不代表远端一定接受这个能力。”

机制发生位置作用对象失败或限制
transformHeadersModelsImpl.applyAuth已合并 header抛错成为 lazy setup error
onPayloadadapter 组装 params 后provider payload返回值可替换 params；hook 失败进入 catch
getCompatadapter 参数构造前Model capability 标志默认值是本地选择，不是服务端证明
processResponsesStream请求建立后上游事件→统一事件具体字段只适用于该 adapter

小A指着表里 getCompat 一行问：“那个 sessionAffinityFormat 默认走 detectSessionAffinityFormat，它到底看什么？”

“两处判断。”老Z说，“一处是 model.provider === "openrouter" 或 baseUrl 里含 openrouter.ai，任中其一就选 openrouter，否则默认 openai。这是本地启发式，不是服务端声明。之后 createClient 建会话头时会用到它：有 sessionId 且是 openrouter，就发 x-session-id；是 openai 分支则发 session_id 并附 x-client-request-id。所以 compat 标志确实改变了发送的 header，至于默认值，只代表本地这一侧的选型。”

“那这个分支算错误路径吗？”小A追问。

“更值得记的是 getClientApiKey 的位置差异。stream 里它在 async IIFE 的 try 内，缺 key 抛错会进 catch、变成 error event；streamSimple 则在构造 stream 之前就同步调用它。所以直接调 adapter 的 streamSimple 时，缺 key 是同步 throw，不走流内错误；只有经由 Models 集合的 async setup 闭包，它才被归一成 error event。这也应了上一节那句话：绕过 ModelsImpl，契约要你自己满足。”

abort、错误与不完整流不能被吞掉 ​

小A把担心的情形一口气列了出来：“abort、没结束就断、报错，流中途出问题各怎么处理？”

“真实 adapter 的尾部有三道检查：请求后若 signal.aborted 就抛错；上游流结束仍是 pending 就报‘没有 stop reason’；若 output 已是 aborted／error 也转为错误路径，并带上 output 已有的 errorMessage。一个没有完整终止信息的流，在这里不会被误报为成功。”

ts
// packages/ai/src/api/openai-responses.ts，stream 的收尾与 catch（v1.0.0，节选）
if (options?.signal?.aborted) throw new Error("Request was aborted");
if (output.stopReason === "pending") throw new Error("OpenAI Responses stream ended without a stop reason");

// catch 中：
output.stopReason = options?.signal?.aborted ? "aborted" : "error";
const errorMessage = formatProviderError(
	normalizeProviderError(error),
	`${model.provider === "openai" ? "OpenAI" : model.provider} API error`,
);
output.errorMessage = /* 订阅额度用尽时追加 ChatGPT 用量页链接 */ errorMessage;
stream.push({ type: "error", reason: output.stopReason, error: output });
stream.end();

“catch 之前还删除了 content block 上仅用于流式解析的 index、partialJson、customInput。这是一个可确认的清理行为，别把它夸张成‘所有中间数据绝不留存’，其他层的日志与 store 都不在这个函数内。”

路径触发条件事件／最终 stop reason可恢复性边界
正常完成parser 填好有效 stop reasondone，原 stop reason调用者可继续处理工具调用
主动中止AbortSignal 已 abortederror，aborted是否重试由上层决定
请求／解析异常SDK、hook、参数或网络抛错error，errorerrorMessage 经 formatter 规范化
上游无终止原因结束仍 pendingerror，error防止把半截响应当成功
lazy setup 失败auth、Provider 或模块加载拒绝error，error没有进入本 adapter 的正常 start

小A盯着表问：“中止为什么也用 error event，而不是 done？”

“事件类型在这里表达‘非正常完成通道’，reason 才负责区分 aborted 与 error。”老Z说，“所以消费者不能只看 event type。源码篇第4章 双层循环——pi-agent-core里会看到，loop 依 stop reason 停止，而不是把 aborted 当工具调用为空。”

“那 catch 里合成 errorMessage 时会把 HTTP 状态丢掉吗？我能读到什么？”

“它是 formatProviderError(normalizeProviderError(error), 前缀) 合成的，前缀在 provider 是 openai 时为 ‘OpenAI API error’，否则用 provider ID。normalize 先按 SDK 字段形状探测状态：statusCode、status、$metadata.httpStatusCode、$response.statusCode，按这个顺序取第一个数字。body 只接受普通对象或字符串，类实例和流对象都算没有 body，内容截断到 4000 字符。format 只在同时拿到状态和 body 时才拼出 前缀 (状态): body，否则保留 SDK 自己的 error.message，免得用噪音替换有用文本。消息里含 subscription_sharing_usage_limit_exceeded 时，adapter 还会追加一行 ChatGPT 用量页链接。”

“所以 errorMessage 是显示用的合成字符串，不是结构化错误对象。”老Z总结，“它有前缀，尽力保留状态和 body；消费者要做分支，优先看 stopReason 和事件本身，别去解析这段文本。aborted 与 error 的差别 reason 已经给出，用不着从文本里猜。”

消费方应保存最终消息，而非自行拼接文本 ​

小A想走条捷径：“那我把 text_delta 拼起来，是不是就拿到完整回答了？”

“会丢东西。”老Z说，“adapter 在流里发布 partial event，是为了让消费方及时显示或更新状态。最终的 AssistantMessage 除了文本还有 usage、provider、model、stop reason，以及可能的 tool call content；只把 text_delta 拼成字符串，这些 Agent 下一步要用的信息就全没了。”

消费策略能得到什么丢失或风险
只显示 delta低延迟文本体验tool call、usage、stop reason 不完整
只等待 result完整 message没有流式可见性
同时消费 event 与 resultpartial UI 加最终状态必须避免把 partial 当持久化终值
收到 error 立即忽略可快速返回抹去 aborted 与 provider error 的区别

“lazyStream 的 forwardStream 不重写内层 event，所以 event 序列的细粒度语义仍由 adapter 决定。setup failure 会统一合成为 error event，调用者因此能用同一个订阅面处理配置失败；但这不是‘所有失败都有相同 message 文本’的保证，createSetupErrorMessage 用 error.message 或 String(error) 填 errorMessage，adapter 也可能用自己的 formatter。”

“重试要放在消费语义之外看。retryProviderRequest 覆盖的只是本 adapter 发出 Responses 请求的那段 await，它不包围之后的 Agent 工具执行，也不包围 session append。请求已经开始输出才中断的话，能不能重试，取决于调用者是否接受重复生成。对有外部副作用的 tool result，模型请求的重试规则不能直接搬过去，这是通用工程判断，不是 OpenAI Responses adapter 的自动保护。”

收益、代价与未覆盖范围 ​

基于源码可以推断：把 API adapter 统一为事件流，Agent 能在同一套循环里显示文本、思考和 tool-call 增量；Provider 也仍能通过 typed options、compat 和 payload hook 保留差异。代价是错误来源跨多层，一个通用事件类型也自动抹平不了计费、缓存或取消的远端语义。

本章没有验证真实服务是否响应、重试退避的全部算法、processResponsesStream 的每一条事件映射，也没有比较 Anthropic、Google 或 WebSocket adapter。这些需要分别读各实现和在受控凭据下运行，不能由 OpenAI Responses 的代码替代。

小结 ​

这一章回答的是：Provider 交回的那个 AssistantMessageEventStream，到底该怎么接。它不是一个“等全部结束再读”的完整对象，而是一条有生命周期的增量事件流：lazyStream 让入口立即给出统一流句柄，把认证、模块加载的失败统一转换成 error message；adapter 的正常路径是请求、start、事件转换、done，错误路径用 reason 区分 abort 与 error，并拒绝没有 stop reason 的半截流。

  • 边界：header transform、payload hook 与 compat 分别在不同层解决不同问题，互不替代。消费者不能只看 event type，还要看 reason：aborted 与 error 是两条不同的恢复路径。

记住： 流是有生命周期的，不是一次返回。增量事件、stop reason、半截流拒绝，这三样合起来定义了“流的终止形态”。循环要依据 stop reason 决定下一步，而不是把 aborted 当成“工具调用为空”。

源码走查 ​

  1. 阅读 packages/ai/src/api/lazy.ts 的 createSetupErrorMessage、forwardStream、lazyStream；确认 setup 错误用零 usage 的 assistant message 结束 outer stream。
  2. 在 packages/ai/src/models.ts 搜索 lazyStream(model；分别标出 stream 和 streamSimple 的 setup 闭包，确认它们都在闭包内调用 Provider。
  3. 打开 packages/ai/src/api/openai-responses.ts 的 stream；从 retryProviderRequest 跟到 stream.push({ type: "start"，写下 start 出现在 HTTP response 的哪个阶段。
  4. 搜索 getCompat 与 buildParams；选择一个 compatibility flag，确认它被用于参数构造，而非由根 Models 自动决定。
  5. 在同一 stream 的 catch 中检查 partialJson 删除和 stopReason 赋值；分别触发信号已 aborted 与普通 Error 时，推演 reason 的差异。