stream/complete:LLM 调用的门面入口

源码版本v0.73.1

stream.ts 是 @mariozechner/pi-ai 对外的最薄一层——四个函数 stream / complete / streamSimple / completeSimple,加上一个内部 resolveApiProvider。它不做 SSE 解析、不构造请求体、不管重试,只负责把 (model, context, options) 三元组按 model.api 路由到对应 provider。可以把它理解成「调用方与 provider 实现之间的派发所」:调用方只看到统一的返回类型 AssistantMessageEventStream,看不到底下是 Anthropic SSE 还是 OpenAI Responses 还是 Bedrock Converse。

职责

这一层做四件事:

1. 派发:stream 按 model.api 找到注册表里的 provider,转交 (model, context, options)。见 packages/ai/src/stream.ts:25-32。
2. 同步返回流:无论底层 provider 是不是异步初始化(懒加载),stream 都同步返回一个 AssistantMessageEventStream,事件晚到时再 push。见 packages/ai/src/stream.ts:30-31。
3. 简化版入口:streamSimple / completeSimple 接收 SimpleStreamOptions(只暴露 reasoning、thinkingBudgets 等 UI 常用字段),由 provider 自己把 simple options 翻译成完整 options。见 packages/ai/src/stream.ts:43-50。
4. 缺 provider 报错:resolveApiProvider 在没注册时抛 No API provider registered for api: <api>,不在调用点静默失败。见 packages/ai/src/stream.ts:17-23。

设计动机

为什么要有这么薄的一层?因为调用方关心的两件事——「拿哪个模型调」和「流式还是拿最终结果」——跟具体 provider 解耦。stream.ts 把这两个维度收成四个函数,下游 pi-coding-agent 的 AgentSession、CLI 工具、扩展都用同一套入口,不必直接 import 具体 provider 模块。这样 provider 实现可以懒加载、可以替换、可以由扩展注册新的 api,调用方代码完全不变。

complete 不另写一套非流式逻辑,而是复用 stream 再 s.result()——流式基础设施与一次性返回共用一条管线,provider 只需实现流式版本。

关键文件

• packages/ai/src/stream.ts:1-3 — 顶部 import "./providers/register-builtins.js" 触发 9 家内置 provider 自注册;getApiProvider 从注册表取。
• packages/ai/src/stream.ts:17-23 — resolveApiProvider,缺 provider 时抛错。
• packages/ai/src/stream.ts:25-32 — stream 全选项版。
• packages/ai/src/stream.ts:34-41 — complete 等价于 stream(...).result()。
• packages/ai/src/stream.ts:43-50 — streamSimple 走 provider.streamSimple。
• packages/ai/src/stream.ts:52-59 — completeSimple 等价于 streamSimple(...).result()。

stream 的全部实现就是 resolve + 转交,没有任何业务逻辑:

// packages/ai/src/stream.ts:25-32
export function stream<TApi extends Api>(
	model: Model<TApi>,
	context: Context,
	options?: ProviderStreamOptions,
): AssistantMessageEventStream {
	const provider = resolveApiProvider(model.api);
	return provider.stream(model, context, options as StreamOptions);
}

complete 复用 stream 再取 result(),不重复派发逻辑:

// packages/ai/src/stream.ts:34-41
export async function complete<TApi extends Api>(
	model: Model<TApi>,
	context: Context,
	options?: ProviderStreamOptions,
): Promise<AssistantMessage> {
	const s = stream(model, context, options);
	return s.result();
}

数据流

stream(model, ctx, opts) 的派发链:

边界与失败

• Provider 缺失:resolveApiProvider 直接抛错,不返回空流,见 packages/ai/src/stream.ts:19-21。扩展动态注册的 provider 没注册时,调用方第一时间知道。
• 懒加载 provider:具体 provider 模块用 createLazyStream 包过,stream 同步返回 outer stream,模块加载完后再 forwardStream 到 inner,见 packages/ai/src/providers/register-builtins.ts:159-178。stream 本身不感知懒加载。
• 类型断言:options as StreamOptions 把 ProviderStreamOptions(含扩展自定义字段)强转到 provider 期望的类型,provider 内部按需读字段。
• options 可选:三个函数都允许 options? 缺省,provider 用默认值兜底。

小结

stream.ts 是 60 行的派发门面,把 (model, context, options) 路由到注册表里的 provider。stream / complete 是全选项版,streamSimple / completeSimple 是 UI 友好子集版。注册表本身的结构看 Provider 注册表,9 家内置 provider 的注册看 Provider 抽象与内置,最终返回的 AssistantMessageEventStream 看 EventStream 异步迭代器。