奇 奇连 AI
学习路线图课程工具企业服务关于加入会员

开源编程 Agent pi 的统一模型接口：一套类型怎么盖住十种协议

2026-07-29

本文基于 pi v0.82.1（仓库 commit 027a584，2026-07-28）梳理，该项目仍在快速迭代，具体行为以官方文档 https://pi.dev/docs/latest 与仓库最新代码为准。

pi 的统一接口没有去抽象「请求和响应」，它抽象的是「消息和事件」——能被压进 Message 数组、能被还原成 AssistantMessageEvent 流的东西才进公共层，剩下的差异一律留在各家 API 模块里各写各的。 这个取舍决定了它后面所有的设计形状：公共层薄得出奇，只有几个 interface；而 packages/ai/src/api/ 下的实现文件动辄几百上千行手写适配（anthropic-messages.ts 一千三百多行），跨家能共用的是 transform-messages.ts、simple-options.ts、deferred-tools.ts、constrained-sampling.ts 这类窄口径工具模块（其中 constrained-sampling.ts 被 Anthropic、Google、Mistral、Bedrock、OpenAI 五个族的实现一起引用，是复用面最宽的一个），再加上同族内部的 openai-responses-shared.ts、google-shared.ts、openai-prompt-cache.ts，协议本身的编解码仍是各写各的。

站内已有两篇讲通用方法论的文章：API 接入方式对比讲的是你该选官方 SDK、聚合网关还是自己封一层，多模型 fallback 设计讲的是切换与兜底的策略。本篇不重复那些判断，只做一件事——把一个真实项目的抽象层拆开，看它在哪里划线、在哪里留缝、为此付出了什么。仓库是 https://github.com/earendil-works/pi ，MIT 许可证，截至 2026 年 7 月在 GitHub 上约 8 万 star。

一、它要解决的问题：十种 API 形态，一个 Agent 循环

先看规模。packages/ai/src/types.ts 里的 KnownApi 列了十个值：

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

export type Api = KnownApi | (string & {});

注意最后那行 (string & {})。这是个惯用写法：既保留十个已知值的自动补全，又允许传入任意字符串，也就是自定义 API 不需要改这个包。同一个文件里的 KnownProvider 列得更长——从 anthropic、openai、google 到 deepseek、zai、moonshotai、minimax、groq、openrouter 等等三十多个。API 和 provider 是两个正交维度：一个 provider 可以走多种 API，多个 provider 也可以共用同一种 API。

这十种协议的差别不是语法糖级别的。Anthropic Messages 是 content block 数组加 SSE 增量事件；OpenAI Responses 是 item 序列加加密的 reasoning 内容；Bedrock 走 Smithy 中间件和 SigV4 签名；Google 有自己的 thought signature。而上层的 Agent 循环只想要一件事：给我一串消息，还我一串事件，中途别抛异常。

二、公共层只抽三样东西

翻遍 types.ts，真正的公共抽象只有三块。

第一块是消息模型。 Message 是三个成员的联合：UserMessage、AssistantMessage、ToolResultMessage。内容块也只有四种：TextContent、ThinkingContent、ImageContent、ToolCall。上层拿到的 Context 更简单：

export interface Context {
	systemPrompt?: string;
	messages: Message[];
	tools?: Tool[];
}

没有 role 字符串的自由发挥，没有 provider 专属字段的透传口子。想塞私货只能塞进那几个可选的 signature 字段里，后面会讲。

第二块是事件协议。 AssistantMessageEvent 是个十二个成员的联合类型，形状高度对称：start，然后 text_start / text_delta / text_end 三件套，thinking 和 toolcall 各有一套同构的三件套，最后以 done 或 error 收尾。前十个事件都带一个 partial: AssistantMessage，也就是当前累积出来的半成品消息；两个终止事件换了字段名，done 带 message、error 带 error，装的都是最终那条完整消息。这意味着渲染层永远不需要自己拼状态——任何时刻拿最新事件里那份消息就是全量视图。

done 和 error 的 reason 还被 Extract 收窄过：done 只可能是 stop / length / toolUse，error 只可能是 aborted / error。也就是说「正常收尾」和「异常收尾」在类型层面就分好了岔，调用方判完 event.type 之后，编译器已经把不可能的 stopReason 组合排除掉了，不需要再写防御性分支。

配套的 AssistantMessageEventStream 在 packages/ai/src/utils/event-stream.ts，是个泛型 EventStream<T, R> 的实例化，构造时传入「什么算结束」和「怎么取最终结果」两个函数：

export class AssistantMessageEventStream extends EventStream<AssistantMessageEvent, AssistantMessage> {
	constructor() {
		super(
			(event) => event.type === "done" || event.type === "error",
			(event) => { /* done 取 message，error 取 error */ },
		);
	}
}

它同时是 AsyncIterable（给要逐帧渲染的调用方）和一个 result() Promise（给只要最终结果的调用方）。一个对象服务两种消费姿势，省掉了「流式版」和「非流式版」两套 API。

第三块是模块契约。 ProviderStreams 只有两个方法：

export interface ProviderStreams {
	stream(model: Model<Api>, context: Context, options?: StreamOptions): AssistantMessageEventStream;
	streamSimple(model: Model<Api>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream;
}

注释里写得很直白：「每个 src/api/ 下的模块都恰好导出 stream 和 streamSimple，所以模块本身就满足这个 interface」。这是个挺省事的做法——模块即实现，不需要 class、不需要注册装饰器、不需要工厂返回对象。import * as impl from "./anthropic-messages.ts" 出来的东西直接就能当 ProviderStreams 用。

还有一条容易被忽略但很关键的约定，写在 StreamFunction 的注释里：函数一旦被调用，请求、模型、运行时的失败都必须编码进返回的流里，而不是抛出来；错误终止要产出一个 stopReason 为 "error" 或 "aborted" 且带 errorMessage 的 AssistantMessage。上层因此只有一条错误通路。关于返回结构不稳定该怎么防，可以对照接口返回结构变化的处理。

三、差异往哪里放：三道接缝

公共层这么薄，差异总得有地方去。pi 开了三道口子。

接缝一：逐 API 的选项类型。 基础的 StreamOptions 只放跨家通用的东西——temperature、maxTokens、signal、apiKey、fetch、headers、timeoutMs、maxRetries、cacheRetention、sessionId、onPayload、onResponse 等。各家的私货放在自己的扩展类型里：Anthropic 的 AnthropicOptions 有 thinkingEnabled、thinkingBudgetTokens、effort、thinkingDisplay、interleavedThinking；OpenAI Responses 的 OpenAIResponsesOptions 有 reasoningEffort、reasoningSummary、serviceTier。两者都 extends StreamOptions，但字段完全不同名——因为背后的机制本来就不同名。

把它们缝起来的是一张类型映射表加一个条件类型：

export type ApiStreamOptions<TApi extends Api> = TApi extends keyof ApiOptionsMap
	? ApiOptionsMap[TApi]
	: StreamOptions & Record<string, unknown>;

已知 API 解析到具体的选项类型，自定义 API 字符串退回到通用形状。ApiOptionsMap 里的 import 全是 import type，编译后被擦除，所以这张表不会把十个实现模块都拖进 bundle。

接缝二：compat 兼容位。 这是我觉得最实用的一处设计。Model<TApi> 上有个 compat 字段，它的类型由 TApi 条件推导：openai-completions 得到 OpenAICompletionsCompat，openai-responses 系列得到 OpenAIResponsesCompat，anthropic-messages 得到 AnthropicMessagesCompat，bedrock-converse-stream 得到 BedrockCompat，其余为 never。

为什么需要它？因为「兼容 Anthropic Messages 协议」的服务商一大把，但兼容程度参差。AnthropicMessagesCompat 里的字段几乎每一条都对应一个真实踩过的坑：supportsCacheControlOnTools（有的服务商不接受工具定义上的 cache_control）、allowEmptySignature（有的会发出并接受空签名的 thinking 块）、supportsTemperature（注释写明某些新模型拒绝非默认温度）、sendSessionAffinityHeaders（有的靠会话亲和头把请求路由到同一副本以提高缓存命中）。OpenAICompletionsCompat 更夸张，光 thinkingFormat 一个字段就有十种取值，覆盖 openai、openrouter、deepseek、together、zai、qwen 等各自的 thinking 参数写法。

这些开关在实现里的读法是统一的——一个 getAnthropicCompat()/getCompat() 函数把「未设置」折叠成默认值，返回 Required<...>，之后正文里再也不写 ??：

return {
	supportsEagerToolInputStreaming: model.compat?.supportsEagerToolInputStreaming ?? true,
	supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
	// ...
	supportsToolReferences: model.compat?.supportsToolReferences ?? defaultSupportsToolReferences(model),
};

最后那行还有个细节：默认值不是常量而是一个函数，它按 provider 和模型 id 正则解析出主次版本号来决定。也就是说，兼容位既能被模型目录显式声明，也能由代码按已知规律兜底。

接缝三：streamSimple 这一层降维。 stream 是「你完全懂这家 API，把参数原样给我」；streamSimple 是「我只知道要 low 还是 high，剩下你看着办」。SimpleStreamOptions 在基础选项上只加了 reasoning?: ThinkingLevel 和 thinkingBudgets，而 ThinkingLevel 是 minimal | low | medium | high | xhigh | max 这六档。

两家的落地方式完全不同。Anthropic 那边分岔：模型标了 forceAdaptiveThinking 就把档位映射成 effort 交给模型自己决定思考量；老模型则走预算制，先用 adjustMaxTokensForThinking() 按档位取 token 预算（packages/ai/src/api/simple-options.ts 里的默认表），再用 clampMaxTokensToContext() 按上下文余量夹一次，最后还有一步 Math.min(adjusted.thinkingBudget, Math.max(0, maxTokens - 1024)) 保证至少留 1024 给正文。OpenAI Responses 那边简单得多，clampThinkingLevel() 把档位夹到该模型支持的范围内，然后直接当 reasoning.effort 发出去。同一个「high」，两边走的是两条完全不同的代码路径。分层调度的通用思路可以参考模型分层调度与模型路由策略。

四、组件对照表

组成部分它负责什么仓库位置你什么时候会碰到它
消息与内容块类型Message、Context、Tool、Usage、StopReasonpackages/ai/src/types.ts自己拼历史、做上下文裁剪、存会话文件时
事件协议与流容器AssistantMessageEvent 十二种事件；流同时可迭代可 awaitpackages/ai/src/types.ts、packages/ai/src/utils/event-stream.ts写渲染层、写中途取消、统计 token 时
模块契约ProviderStreams 的 stream / streamSimplepackages/ai/src/types.ts接一个仓库里没有的自定义 API 时
逐 API 选项AnthropicOptions、OpenAIResponsesOptions，经 ApiOptionsMap 汇总packages/ai/src/api/anthropic-messages.ts、packages/ai/src/api/openai-responses.ts要精细控制 thinking、tool_choice、service_tier 时
兼容位AnthropicMessagesCompat 等四组开关packages/ai/src/types.ts接 Anthropic/OpenAI 协议兼容的第三方端点时
跨模型消息归一transformMessages()：降级图片、处理 thinking、规范工具调用 idpackages/ai/src/api/transform-messages.ts中途换模型继续同一段对话时
延迟工具切分splitDeferredTools() 分出 immediate 与 deferredpackages/ai/src/utils/deferred-tools.ts工具数量大、想让模型按需加载工具定义时
懒加载包装lazyApi() / lazyStream()：同步返回流，异步跑 setuppackages/ai/src/api/lazy.ts关心冷启动体积、不想把十个 SDK 都打进包时
Provider 装配createProvider() 按 model.api 分发到实现packages/ai/src/models.ts注册自定义 provider、混合 API 的 provider 时
具体 provider 工厂拼 id、baseUrl、auth、模型目录、api 实现packages/ai/src/providers/anthropic.ts照着抄一个自己的 provider 时

再说两个表里没展开的点。

lazyApi 的写法很省心：anthropic-messages.lazy.ts 整个文件只有四行，导出 anthropicMessagesApi = () => lazyApi(() => import("./anthropic-messages.ts"))。lazyStream 会先同步造好一个空的 AssistantMessageEventStream 返回给调用方，背后再去跑 auth 解析和动态 import，setup 失败就往流里推一个 error 事件。所以「返回流」这个动作永远是同步的，调用方不需要 await 两次。packages/ai/src/index.ts 顶部那段注释也交代了同样的取向：核心入口无副作用，不含生成的模型目录、provider 工厂和 OAuth 实现，那些走 @earendil-works/pi-ai/providers/* 和 @earendil-works/pi-ai/api/* 子路径导出。

createProvider 的 api 字段可以是单个 ProviderStreams，也可以是按 model.api 索引的一张表。分发时找不到实现就返回一个直接报错的流，错误信息是「Provider X has no API implementation for …」——注意仍然是流内错误，不是抛异常，和前面那条约定一致。

五、边界与代价：它明确不管什么

抽象层薄，代价就得有人承担。诚实地列一下。

它不抹平语义，只抹平形状。 同一个 reasoning: "high"，Anthropic 走 effort 或 token 预算，OpenAI 走 reasoning effort，实际思考量、延迟、计费口径都不一样。类型对齐了不等于行为对齐了。跨模型做 A/B 对比时，这一层不会帮你把差异消掉。

跨模型续聊必然有损。 transformMessages() 里有个 isSameModel 判断，同时比 provider、api 和 model id 三项。不是同一个模型时：redacted 的 thinking 块直接丢弃（它是只对原模型有效的加密内容），带签名的 thinking 块降级成纯文本，Google 专属的 thoughtSignature 被删掉，工具调用 id 还要重新规范化——注释里写明 OpenAI Responses 生成的 id 可能有 450 多字符并含 |，而 Anthropic 要求匹配 ^[a-zA-Z0-9_-]+$ 且不超过 64 字符，于是 Anthropic 那边的 normalizeToolCallId 就是一句 id.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 64)。换模型继续，等于主动放弃一部分推理上下文。

它不做重试编排，也不做多模型兜底。 StreamOptions 里确实有 maxRetries 和 maxRetryDelayMs，但那是交给各家 SDK 或 retryProviderRequest 的单次请求级重试；maxRetryDelayMs 的注释说得很清楚——服务端要求的等待超过上限就立刻失败，「让更高层的重试逻辑带着可见性去处理」。也就是说，切模型、降级、熔断这些属于上层职责，这个包不碰。

它不接管上下文管理。 clampMaxTokensToContext() 只做一件粗活：用 estimateContextTokens() 估一下，减掉 4096 的安全余量，把 maxTokens 夹住。真正的压缩、摘要、裁剪不在这一层。

兼容位是穷举，不是推断。 每多一个不听话的服务商，就多一个布尔字段。这套设计的可扩展性来自「加一个开关很便宜」，而不是「自动适配」。你接一个新端点，大概率要自己试出该开哪几个开关。

协议本身的东西它管不了。 例如 ToolResultMessage.addedToolNames 的注释直说：有原生延迟工具加载的服务商拿它当加载点，其他服务商忽略它、照常用 Context.tools。同一份数据在不同 API 下含义强弱不同，这是协议差异，抽象层只能如实转达。

最后一条现实约束：海外模型服务商官方对中国大陆存在区域限制，不支持直连；市面上有第三方中转，本文不做背书也不给具体渠道，各家规则不同且会调整，以官方最新说明为准。这一层和框架设计无关，但会实际影响你能不能把某个 provider 跑起来。

六、上手与避坑清单

别从 stream() 开始读，从 types.ts 开始。 为什么会踩：anthropic-messages.ts 一千三百多行，直接进去会淹在 SSE 解码和 content block 转换里。怎么避：先把 Message、AssistantMessageEvent、ProviderStreams 这三组类型看熟，再回头看实现，你会发现每个实现文件都是同一套骨架——建 output、建 client、buildParams、迭代事件、推 event、收尾。

别指望 catch 住流错误。 为什么会踩：TypeScript 类型上 stream() 返回的是同步值，直觉会想 try/catch。怎么避：按契约，错误在流里，要么在 for await 里判 event.type === "error"，要么 await result() 后看 stopReason。把这条写进你自己的封装约定，否则错误会静默丢失。

接第三方兼容端点时，先假设兼容位不对。 为什么会踩：很多服务商宣称兼容 Anthropic 或 OpenAI 协议，但工具定义上的 cache_control、strict 模式、空签名 thinking 块、usage 字段这些边角处经常不一致，症状是 400 或者字段悄悄丢失。怎么避：照着 AnthropicMessagesCompat / OpenAICompletionsCompat 的字段清单逐条对，把不支持的显式关掉，而不是改实现代码。

跨模型换机之前，先想清楚 thinking 块怎么办。 为什么会踩：换个模型继续同一段会话，历史里的加密推理内容会被丢弃或降级成文本，模型的表现可能突然变化，而日志上看不出任何报错。怎么避：把换模型当成一次上下文重置来对待，必要时在换机点补一段小结。相关权衡见Agent 框架对比。

自定义 API 要走字符串扩展路径，不要改 KnownApi。 为什么会踩：Api 类型故意写成 KnownApi | (string & {}) 就是为了留口子，改联合类型会让你每次升级都得重新打补丁。怎么避：自定义 API 名直接用字符串，选项类型会退化成 StreamOptions & Record<string, unknown>，然后按 createProvider 的 api 映射表挂进去。

注意 usage 字段各家给得不一样。 为什么会踩：Usage 里的 reasoning 字段注释写明是「服务商报了才有」，totalTokens 在 Anthropic 侧是代码自己按四项相加算出来的（因为对方不给 total）。直接拿这些数做成本统计，换一家就对不上。怎么避：先确认目标 provider 到底报了哪些字段，缺的按估算处理并在报表里标出来。

收束

如果你只带走一句话：pi 把「所有模型都长一样」这个幻想砍掉了，只保证「所有模型的输入输出能被同一组类型描述」，中间的差异被显式地放进 compat 开关、逐 API 选项和 transform 函数里，看得见、改得动。

想动手的话，建议按这个顺序读四个文件：packages/ai/src/types.ts 建立词汇表，packages/ai/src/api/lazy.ts 看最小的实现包装（七十多行，最容易读完），packages/ai/src/api/openai-responses.ts 看一个中等复杂度的真实适配，最后再啃 packages/ai/src/api/anthropic-messages.ts。

自检三问：你的封装层里，错误是走异常还是走返回值？换模型时历史消息谁负责归一？服务商的兼容差异是写在配置里还是散在 if 里？这三个问题的答案，基本就决定了你那套抽象能撑多久。

本篇属于一个把开源编程 Agent 项目 pi 逐层拆开讲的系列，整体地图见开源编程 Agent pi 是什么；沿着这条线往下，还可以看 开源编程 Agent pi 的仓库结构导读 和 开源编程 Agent pi 的服务商层。

想系统学会用 AI？报名体系课或加入会员，照着学、照着用。

浏览课程 加入会员