1.3 Model 与 Provider

pi 要"统一多家厂商"，核心是两件事：用统一的 Model 描述一个模型，用统一的 Provider 描述一个厂商。这一节讲清楚这两个抽象，以及它们如何被组织进 Models 集合。

Model：一个模型 = 一份声明

packages/ai/src/types.ts:779 定义了 Model<TApi>，它描述"某个厂商的某个模型长什么样"：

export interface Model<TApi extends Api> {
	id: string;              // 模型 id，如 "gpt-4o"
	name: string;            // 显示名
	api: TApi;               // 使用的协议（见下）
	provider: ProviderId;    // 所属厂商
	baseUrl: string;         // API 地址
	reasoning: boolean;      // 是否支持推理
	thinkingLevelMap?: ThinkingLevelMap;  // 思考档位映射
	input: ("text" | "image")[];          // 支持文本还是图片
	cost: ModelCost;         // 计价
	contextWindow: number;   // 上下文窗口
	maxTokens: number;       // 最大输出
}

注意 Model 是纯数据，不含任何"怎么调用"的逻辑。怎么调用由 Provider 决定。

api 字段：协议维度

api: TApi 是区分"用哪套协议调用"的关键。pi 定义了一组已知协议：

type KnownApi =
	| "openai-responses" | "openai-completions" | "openai-codex-responses"
	| "anthropic-messages" | "google-generative-ai" | "google-vertex"
	| "bedrock-converse-stream" | "mistral-conversations"
	| "azure-openai-responses" | "openrouter" | "faux" | ...;

一个厂商可能支持多种协议，一个协议也可能被多家厂商复用。例如很多国产模型都走 openai-completions 协议。

Provider：一个厂商 = 一批能力

packages/ai/src/models.ts:97 定义了 Provider：

export interface Provider<TApi extends Api> {
	id: string;
	name: string;
	baseUrl?: string;
	headers?: ProviderHeaders;
	auth: ProviderAuth;                       // 认证方式
	getModels(): readonly Model<TApi>[];      // 该厂商的模型列表
	filterModels?(models, credential);        // 可按凭证过滤可用模型
	stream(model, context, options);          // 流式调用
	streamSimple(model, context, options);    // 简化版流式调用
	fetchDeferred?(model, handle, options);   // 异步任务取结果
	cancelDeferred?(...);
	refreshModels?(context);                  // 动态刷新模型目录
}

关键点：

auth 是必填的。pi 认为每个厂商都有认证语义，即便只是环境变量或本地免鉴权。
stream() 是核心能力：给定一个 Model + Context，返回 AssistantMessageEventStream。
getModels() 返回该厂商的模型目录。静态厂商（如 openai）返回固定目录；动态厂商（如 openrouter）在 refreshModels() 后更新。

Model / Provider / Models 三者的关系

Model 描述"是什么"（数据）。
Provider 描述"怎么调用 + 有哪些模型"（行为 + 目录）。
Models 是前两者的 运行时集合，负责按 provider 路由请求、解析认证。

export interface Models {
	getProvider(id): Provider | undefined;
	getModels(provider?): readonly Model<Api>[];
	getModel(provider, id): Model<Api> | undefined;
	getAuth(providerId | model): Promise<AuthResult | undefined>; // 解析认证
	stream(model, context, options);   // 委托给 model.provider 的 stream
	complete(model, context, options); // 流完并返回最终消息
	streamSimple(...); completeSimple(...);
	login(providerId, type, interaction); logout(providerId);
}

Models 用 Map<provider, Provider> 存所有厂商，setProvider() / deleteProvider() 动态维护。ModelRuntime（coding-agent 里）就是 Models 的一个具体实现。

深入：`stream` 与 `streamSimple` 的三层调用链

stream / streamSimple 这两个名字在三个层级都会出现，职责各不同。看懂这一整条调用链，就理解了"一次模型请求如何被路由、认证、执行"。

① Models 集合层（路由+认证） models.stream / models.streamSimple
         ↓ 委托
② Provider 层（厂商分发）   provider.stream / provider.streamSimple
         ↓ 按 model.api 分发
③ API 实现层（发 HTTP）    streams.stream / streams.streamSimple

先分清 `stream` 与 `streamSimple` 的核心区别

	`stream`	`streamSimple`
选项类型	`ApiStreamOptions<TApi>`（按 API 强类型）	`SimpleStreamOptions`（通用简化）
适用场景	明确知道是哪个 API，想用尽能力	不关心具体 API，只想要"最简流式调用"
关键	每个 API 有专属选项	正好匹配 `Agent` 的 `StreamFn` 形状

关键点：@pi/agent-core 的 StreamFn 契约就是按 streamSimple 的形状定义的。所以 coding-agent 用 setDefaultStreamFn(streamSimple) 注入默认实现（sdk.ts:36）。

① Models 层：`lazyStream` + 认证

models.ts:690 的 streamSimple：

streamSimple(model, context, options): AssistantMessageEventStream {
	return lazyStream(model, async () => {              // ① 懒包装：同步返回
		const provider = this.requireProvider(model);    // ② 找到厂商
		const { requestModel, requestOptions } = await this.applyAuth(model, options); // ③ 认证
		return provider.streamSimple(requestModel, context, requestOptions);           // ④ 委托给 Provider
	});
}

stream（models.ts:667）结构完全一样，只是选项类型不同、最后调 provider.stream。

精妙之处：lazyStream（api/lazy.ts:46）让 streamSimple 同步返回一个事件流，同时把异步的认证解析放到后台：

export function lazyStream(model, setup): AssistantMessageEventStream {
	const outer = new AssistantMessageEventStream();
	setup()                                    // setup 是异步的（含认证）
		.then((inner) => forwardStream(outer, inner))       // 成功：把内部流转发到 outer
		.catch((error) => {
			const message = createSetupErrorMessage(model, error); // 失败：编码成 error 事件
			outer.push({ type: "error", reason: "error", error: message });
			outer.end(message);
		});
	return outer;                              // 立刻返回！
}

它保证了"同步返回 + 异步准备"：调用方拿到流的那一刻，认证可能还没完成，但没关系 —— 认证失败会以 error 事件结束流，而绝不 throw。这正好符合 StreamFunction 契约（见 1.4）。

applyAuth 做了什么（models.ts:636）：

async applyAuth(model, options) {
	this.requireProvider(model);
	const resolution = await this.getAuth(model, { apiKey, env, signal });   // 解析认证
	if (!resolution) throw new ModelsError("auth", `Provider is not configured`);
	const apiKey = options?.apiKey ?? resolution.auth.apiKey;   // 显式 apiKey 优先
	const headers = mergeHeaders(resolution.auth.headers, options?.headers); // 合并请求头
	if (options?.transformHeaders) headers = await options.transformHeaders(headers); // 允许改写
	const requestModel = resolution.auth.baseUrl ? { ...model, baseUrl: ... } : model; // 可能重写
	return { requestModel, requestOptions: { ...providerOptions, apiKey, headers, env } };
}

解析认证后，把 apiKey、headers、可能重写的 baseUrl 塞进 requestModel/requestOptions，再交给 Provider。

② Provider 层：按 API 分发

createProvider 生成的 provider.stream / provider.streamSimple（models.ts:829-831）只是一个 dispatch：

stream: (model, context, options) => dispatch(model, (streams) => streams.stream(model, context, options)),
streamSimple: (model, context, options) => dispatch(model, (streams) => streams.streamSimple(model, context, options)),

dispatch 根据 model.api 找到对应的 API 实现（ProviderStreams），调用它的 stream/streamSimple。一个厂商可能支持多种 API（如同时支持 openai-completions 和 openai-responses），这里就按 model.api 路由。

③ API 实现层：真正的 HTTP 调用

这一层才真正发请求。例如 Faux Provider（faux.ts）：

streamSimple: (streamModel, context, streamOptions) => stream(streamModel, context, streamOptions),

streamSimple 常常就是直接转调 stream（因为 HTTP 调用本身是同一套）。真正的 stream 负责：构造请求体 → 发 HTTP → 解析响应 → 把结果切成事件推给 AssistantMessageEventStream。

端到端总结

Agent 调 streamSimple(model, context, options)
  → ① Models.streamSimple：lazyStream 同步返回，后台 applyAuth 解析认证
  → ② provider.streamSimple：dispatch 按 model.api 找到 API 实现
  → ③ 真正发 HTTP，流式事件一路 forward 回 Agent 的事件流
  → 认证/网络失败 → 以 error 事件结束，绝不 throw

stream 与 streamSimple 的差别只在"选项的类型化程度"：stream 用 API 专属的 ApiStreamOptions<TApi>（适合明确知道 API 时用尽能力），streamSimple 用通用 SimpleStreamOptions（适合 Agent 这种"不关心具体 API"的调用方）。其余路由、认证、懒加载、错误编码逻辑完全一致。

伪代码：跟着数据走一遍

还是觉得抽象？那就把它当成一个具体的函数调用故事。假设 coding-agent 的 Agent 想用 Claude，调用 streamSimple(claude模型, {消息历史}, {})：

text

# ============ ① Models 层（路由 + 认证）============
function streamSimple(model, context, options):
    # 关键：立刻还一个"空的事件流"，马上 return —— 调用方不用等
    outer = new 事件流()

    # 实际工作放到后台异步做
    异步执行:
        # 找到这个模型属于哪个厂商
        provider = 查表(model.provider)          # 例：Anthropic 厂商
        如果 provider 不存在: 抛错"未知厂商"

        # 解析认证（去环境变量 / 凭证库取 key）
        auth = 解析认证(model, options)
        如果该厂商没配置认证: 抛错"未配置"

        apiKey    = options.apiKey 或 auth.apiKey
        headers   = 合并(auth.headers, options.headers)
        请求参数  = { apiKey, headers, 其他选项 }

        # 把球传给 ② Providers 层
        inner = provider.streamSimple(model, context, 请求参数)

        # 把 inner 里的事件一个个搬进 outer（这就是"转发"）
        for 事件 in inner:
            outer.push(事件)
        outer.end(inner 的最终结果)

    # 万一上面任何一步出错（没厂商/没认证/网络断）
    出错时:
        outer.push({ error 事件 })      # 编码成错误事件
        outer.end()

    return outer                        # 立刻返回，调用方继续干别的事

# ============ ② Provider 层（按 API 分发）============
function provider.streamSimple(model, context, options):
    # 一个厂商可能支持多套协议，按 model.api 挑对应的"协议实现"
    api实现 = 按api分发(model.api)       # 例：anthropic-messages
    return api实现.streamSimple(model, context, options)

# ============ ③ API 实现层（真正发 HTTP）============
function api实现.streamSimple(model, context, options):
    # 通常就是直接转调 stream（HTTP 调用是同一套）
    return api实现.stream(model, context, options)

function api实现.stream(model, context, options):
    # 真正的网络请求
    请求体 = 把消息历史转成该协议的 JSON 格式
    resp = HTTP 发送(model.baseUrl, headers=options.headers, body=请求体)

    stream = new 事件流()
    for 一段增量 in resp 的流式返回:
        stream.push({ text_delta, 增量的文本 })   # 边收边推
    stream.push({ done, 完整消息 })               # 收完了
    stream.end()
    return stream

读法提示：从上往下看，就像一个"委托链" —— ① 找厂商、认证后，把球交给 ②；② 按协议挑好实现，把球交给 ③；③ 真正打电话（HTTP），然后把结果一路传回 ① 的 outer，最终回到 Agent 手里。中间任何环节出错，都在 outer 里以 error 事件收尾，绝不往外抛异常。

compat 里的旧版全局 streamSimple

coding-agent 用的 streamSimple 其实来自 compat.ts:275（兼容旧全局 API 的入口）。它比 Models.streamSimple 多一步 withEnvApiKey：若没显式给 apiKey，就从环境变量注入（env-api-keys.ts）。新代码用 createModels() 则不需要。

面试角度：为什么这样分层

Q1：为什么 Model 是"纯数据"，而 Provider 才是"行为"？ 因为"模型是什么"（id、成本、上下文窗口）和"怎么调用"（认证、HTTP、协议）是两个变化维度。把数据和行为分离，模型目录可以静态生成、可缓存、可对比（modelsAreEqual），而调用逻辑可以独立演进、可替换实现。这也是分层架构的最小体现。

Q2：为什么 Model 还要带 api 字段？ 因为一个"怎么调用"其实有两层：厂商（provider）和协议（api）。一个厂商可能支持多套协议（如 OpenAI 同时有 openai-completions 和 openai-responses），一个协议也可能被多家厂商复用。api 字段就是"用哪套协议"的开关，让 Provider 能按 model.api 分发到正确的协议实现。

Q3：为什么要有 Models 集合层统一做"路由 + 认证"，而不让调用方直接调 provider.stream？ 因为认证解析是有状态的、复杂的（可能要 OAuth 刷新、从凭证库取 key、合并请求头）。如果每个调用方都自己处理，逻辑会重复且容易出错。Models 把它收敛成"给我一个 model 和 options，我负责认证好再调"，调用方只关心业务。

Q4：为什么 stream 和 streamSimple 要分两个方法？Agent 为什么用 streamSimple？stream 用 API 专属的强类型选项（ApiStreamOptions<TApi>），适合"明确知道是哪个 API"的高级调用；streamSimple 用通用简化选项，适合"不关心具体 API"的调用方。Agent 恰好不关心 API，所以 StreamFn 契约按 streamSimple 形状定义，Agent 用 setDefaultStreamFn(streamSimple) 注入默认实现 —— 这样 Agent 与厂商彻底解耦（见 1.4）。

Q5：为什么 lazyStream 要"同步返回 + 异步准备"，并且错误编码进流？ 因为认证解析是异步的，但调用方不想等。lazyStream 立刻返回一个空流，后台异步解析认证，失败时以 error 事件结束流而非 throw。这既保证了"同步返回"的便捷，又遵守了"绝不 throw"的契约（见 1.4）。

createProvider：如何注册一个厂商

pi 提供 createProvider() 工厂，把"模型 + 认证 + 流式实现"打包成一个 Provider：

const provider = createProvider({
	id: "my-provider",
	auth: { apiKey: { name: "MyKey", resolve: async () => ({ auth: {...} }) } },
	models: [myModel],
	api: {
		stream: (model, context, options) => { /* 真正的 HTTP 调用 */ },
		streamSimple: (model, context, options) => { ... },
	},
});

之后把它塞进 Models：

const models = createModels();
models.setProvider(provider);

一个直观的例子：Faux Provider

pi 自带一个 Faux（假）Provider（packages/ai/src/providers/faux.ts），零网络、纯脚本化返回，专门用于测试和教学。它的 stream 不真调模型，而是把预设好的 AssistantMessage 按 token 切片一点点推给事件流：

const faux = fauxProvider();
faux.setResponses([fauxAssistantMessage("你好，我是假的模型")]);
models.setProvider(faux.provider);

const stream = models.streamSimple(faux.getModel(), { messages: [...] });
for await (const e of stream) { /* 会看到 text_delta 逐段拼出这句话 */ }

它是我们第一个 Demo 的基础（见 Demo：最小 Faux Provider）。

小结

Model = 模型的纯数据声明（含 api 协议维度）。
Provider = 厂商的行为能力（认证 + 模型目录 + 流式调用）。
Models = 运行时集合，负责路由与认证，是上层（agent、coding-agent）的入口。
createProvider + setProvider 就可以注册一个新厂商。
stream / streamSimple 走三层调用链：Models（认证+懒加载）→ Provider（按 API 分发）→ API 实现（发 HTTP）；两者差别只在选项的类型化程度。

真实源码位置

Model：packages/ai/src/types.ts:779
Provider：packages/ai/src/models.ts:97
Models：packages/ai/src/models.ts:156
Models.stream / streamSimple：packages/ai/src/models.ts:667,690
lazyStream：packages/ai/src/api/lazy.ts:46
applyAuth：packages/ai/src/models.ts:636
旧版全局 streamSimple：packages/ai/src/compat.ts:275
createProvider：packages/ai/src/models.ts:762
Faux Provider：packages/ai/src/providers/faux.ts

下一步：1.4 StreamFunction 契约 —— 模型与 Agent 之间的边界。

1.3 Model 与 Provider ​

Model：一个模型 = 一份声明 ​

api 字段：协议维度 ​

Provider：一个厂商 = 一批能力 ​

Model / Provider / Models 三者的关系 ​

深入：stream 与 streamSimple 的三层调用链 ​

先分清 stream 与 streamSimple 的核心区别 ​

① Models 层：lazyStream + 认证 ​

② Provider 层：按 API 分发 ​

③ API 实现层：真正的 HTTP 调用 ​

端到端总结 ​

伪代码：跟着数据走一遍 ​

面试角度：为什么这样分层 ​

createProvider：如何注册一个厂商 ​

一个直观的例子：Faux Provider ​

小结 ​