架构（一）：pi-ai 的兼容层

pi-ai 是整个仓库里唯一一个「不对外暴露产品形态」的包。没有 CLI，没有配置文件，没有交互界面。它的全部产出是一个 TypeScript 模块：你 import 它，给它一个模型和一段上下文，它给你一个事件流。

这种没有形态的包通常最难读——你看不到它长什么样，只能从调用点反推。但 pi-ai 恰好是全仓库设计意图最清晰的一个包，因为它要回答的问题非常具体：

42 个供应商、10 种聊天 API 协议、1 种图像协议、4 种分类协议，怎么用一套内部表示统一起来？

而它的答案是——把差异声明出来，而不是写成一堆 if。这是本文的主线。理解了这条，其余部分基本顺理成章。

17.1 定位与依赖

@earendil-works/pi-ai，自述是 "Unified LLM API with provider collections, automatic auth resolution, token and cost tracking, and simple context persistence and hand-off to other models mid-session"。

最后半句值得单独拎出来：hand-off to other models mid-session——会话进行到一半换一个模型继续。这件事能成立，靠的正是下一节要讲的 TranscriptContext 设计。

依赖：十个，其中一个是纯类型

{
  "dependencies": {
    "@anthropic-ai/sdk": "0.129.0",
    "@aws-sdk/client-bedrock-runtime": "3.1127.0",
    "@earendil-works/pi-telemetry": "1.3.0",
    "@google/genai": "2.21.0",
    "@smithy/node-http-handler": "4.12.1",
    "http-proxy-agent": "9.1.0",
    "https-proxy-agent": "9.1.0",
    "openai": "7.19.0",
    "partial-json": "0.1.7",
    "typebox": "1.3.27"
  }
}

这张表本身就是兼容层设计的证据：

最后一条是全仓库唯一的 Pi 内部依赖，用法是 packages/ai/src/types.ts 第一行的 import type { TelemetryContext } from "@earendil-works/pi-telemetry"。纯类型导入，运行时零耦合。

**三个官方 SDK 覆盖 15 种协议里最麻烦的 7 种。**剩下 8 种——Mistral conversations、OpenRouter images、四种 classifier API、以及 Pi 自己的 pi-messages——全部是手写适配器。这个比例本身就是结论：手写可行，官方 SDK 覆盖的地方不要手写。

三种模型类型

export interface ModelTypeMap {
  chat: Model<Api>;
  image: ImageModel<ImageApi>;
  classifier: ClassifierModel<ClassifierApi>;
}

chat 是默认类型——Model 里的 type 字段是可选的，所以判别模型类型要用 isModelType()，不要直接比 type === "chat"。

三类模型的接口是刻意不等价的：Model（chat）带 reasoning、contextWindow、maxTokens，用 stream()；ImageModel 带 output，用 generateImages()；ClassifierModel 只有 contextWindow，用 classify()。ClassifierModel 没有 maxTokens——分类是一次前向返回答案，不需要生成长度，这类模型根本不接受 max_tokens。

协议枚举：10 + 1 + 4

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 KnownImageApi = "openrouter-images";

export type KnownClassifierApi =
  | "typesafe-system-one" | "cloudflare-workers-ai-system-one"
  | "llama-cpp-classify" | "openai-decisions";

但这三个类型都不是封闭的联合：export type Api = KnownApi | (string & {})。(string & {}) 这个 TypeScript 小技巧保留了已知值的自动补全，同时允许任何字符串——你的自定义网关完全可以写 "api": "my-gateway"，pi-ai 不拦它，只是不会给它生成适配器。KnownProvider 同样处理：42 项字面量联合，配 export type ProviderId = KnownProvider | string。

逐个数一遍能看出 Pi 对「同一家公司的不同计费渠道」的建模粒度：moonshotai 和 moonshotai-cn 各算一个，qwen-token-plan 有三个变体，xiaomi-token-plan 有四个。供应商标识的是「一套凭据 + 一个网关地址」，不是「一家公司」。

17.2 兼容层：把差异声明出来，而不是写成一堆 if

这是本文的核心。先说清问题，再给答案。

问题的真实形状

「支持 42 个供应商」听起来像「写 42 个适配器」，那只是最表层的问题。真正难的是同一供应商的不同 API 家族之间有差异，同一族内部不同模型之间还有差异：

• OpenAI 收 max_completion_tokens，但 DeepSeek、Together、NVIDIA 这些「OpenAI 兼容」端点只认 max_tokens
• Anthropic 支持对话中途插 system message，很多 OpenAI 兼容实现会直接报错
• 思考预算字段名有 thinking_token_budget（vLLM）、thinking_budget（Qwen / DashScope / SGLang）、thinking_budget_tokens（llama.cpp）三种
• 表达「思考」的格式有十几种：reasoning_effort、reasoning.effort、thinking.type、顶层 enable_thinking、chat_template_kwargs……

用 if/else 处理这些，你要写的是一棵按 provider 名字或 URL 做字符串判断的分支树。这类代码的致命问题是没法被校验，也没法被穷举——谁能保证 42 个供应商 × 10 种协议的所有组合都被覆盖了？

答案：TypeBox 声明 + 两层解析

Pi 的做法是：为每个 API 家族写一份 TypeBox schema，把这个家族的兼容性差异变成带默认值的字段。

packages/ai/src/providers/compat-schema.ts 有 644 行，导出五份 schema，由这张表分发：

export const CompatSchemasByApi = {
  "openai-completions": OpenAICompletionsCompatSchema,
  "openai-responses": OpenAIResponsesCompatSchema,
  "azure-openai-responses": OpenAIResponsesCompatSchema,
  "openai-codex-responses": OpenAIResponsesCompatSchema,
  "anthropic-messages": AnthropicMessagesCompatSchema,
  "bedrock-converse-stream": BedrockCompatSchema,
  "mistral-conversations": MistralConversationsCompatSchema,
} as const;

7 个键覆盖 10 种聊天协议——三种 OpenAI Responses 系共用同一份。剩下三种——google-generative-ai、google-vertex、pi-messages——根本没有 compat schema，实测这三个 API 模块里 compat 出现次数为 0。

这不是遗漏。原因是它们没有「兼容性选择」这个自由度：Google 的协议要么用要么不用，没有中间形态；pi-messages 是 Pi 自己的协议。只有「声称兼容别人但实际各有微差」的协议才需要一层声明式开关。

这是本节最值得学的一点：兼容层不是无条件的，它是「给不完美的兼容声明一个补差机制」。

五份 schema 的规模

27 个字段全部 optional 且带默认值。一份 27 行的 schema 文件，就是 OpenAI 兼容生态的全部分歧点清单。

挑两个有代表性的：

supportsMidConvoSystemMessages: optionalCompatBoolean({
  description:
    "Whether the exact model accepts system or developer messages after the conversation has started. " +
    "When false, later system messages are folded into the leading system message. " +
    "The generated catalog enables this for verified models.",
  default: false,
}),

maxTokensField: Type.Optional(
  Type.Union([Type.Literal("max_completion_tokens"), Type.Literal("max_tokens")], {
    description: "Which field to use for max tokens. Default: auto-detected from URL.",
  }),
),

注意描述里的 "the exact model"——这个开关的粒度是模型而不是 API。同一供应商同一 API 下，强模型支持中途 system message、小模型不支持，所以它挂在模型目录里而不是 provider 里。

三个枚举型开关

差异不全是布尔，有些是字段名或格式名，这类用 Type.Union 声明。thinkingFormat 是最大的一个，11 个取值：

thinkingFormat: Type.Optional(Type.Union([
  Type.Literal("openai"),             // reasoning_effort
  Type.Literal("openrouter"),         // reasoning.effort
  Type.Literal("deepseek"),           // thinking.type + reasoning_effort
  Type.Literal("together"),           // reasoning.enabled + reasoning_effort
  Type.Literal("baseten"),            // chat_template_args + reasoning_effort
  Type.Literal("zai"),                // thinking.type
  Type.Literal("qwen"),               // 顶层 enable_thinking
  Type.Literal("chat-template"),      // chat_template_kwargs
  Type.Literal("qwen-chat-template"), // chat_template_kwargs.enable_thinking + preserve_thinking
  Type.Literal("string-thinking"),    // 顶层 thinking
  Type.Literal("ant-ling"),           // reasoning.effort（仅当映射出的 effort 非 null）
], { description: "Format for reasoning or thinking parameters. ..." })),

**11 种思考参数的表达方式，被压缩成了一个字符串字面量。**适配器里对应的是一串 else if (compat.thinkingFormat === "…")，但每个分支做的事完全确定，没有隐含的 provider 判断。

另外两个：thinkingTokenBudgetField 三个取值（上面提到的三个字段名），sessionAffinityFormat 三个取值（openai / openai-nosession / openrouter）。

两层解析：先探测，再覆盖

schema 只是声明，真正的取值来自两层。

第一层：按 provider 和 baseUrl 自动探测。 packages/ai/src/api/openai-completions.ts 的 detectCompat()：

function detectCompat(model: Model<"openai-completions">): ResolvedOpenAICompletionsCompat {
  const provider = model.provider;
  const baseUrl = model.baseUrl;

  const isZai = provider === "zai" || provider === "zai-coding-cn"
    || baseUrl.includes("api.z.ai") || baseUrl.includes("open.bigmodel.cn");
  const isTogether = provider === "together" || baseUrl.includes("api.together.ai") || baseUrl.includes("api.together.xyz");
  ...
  const isNonStandard = isNvidia || isCerebras || provider === "xai" || ... ;

  return {
    supportsStore: !isNonStandard,
    supportsDeveloperRole: isOpenRouterDeveloperRoleModel || (!isNonStandard && !isOpenRouter),
    maxTokensField: useMaxTokens ? "max_tokens" : "max_completion_tokens",
    requiresReasoningContentOnAssistantMessages: isDeepSeek,
    thinkingFormat: isDeepSeek ? "deepseek" : isZai ? "zai" : isTogether ? "together"
      : isAntLing ? "ant-ling" : isOpenRouter ? "openrouter" : "openai",
    sendSessionAffinityHeaders: isOpenRouter,
    sessionAffinityFormat: isOpenRouter ? "openrouter" : "openai",
    // OpenAI compatibility alone does not imply strict JSON-schema tool support.
    supportsStrictMode: false,
    supportsMidConvoSystemMessages: false,
    supportsMidConvoToolAdditions: false,
    ...
  };
}

先承认一件事：detectCompat 本身确实是 if/else，它做的是字符串匹配。但它的位置让它可接受，有三个性质：它是兜底默认值而不是真相（每个模型目录项都可以覆盖）；它的输出是一个完整对象而不是分支散落（调用方拿到的永远是「27 个字段都有值」的 ResolvedOpenAICompletionsCompat，不存在「某个字段忘了判断」的可能）；末尾是显式的保守默认（supportsStrictMode: false 上面写着注释 "OpenAI compatibility alone does not imply strict JSON-schema tool support"——这是个值得学的态度：宣称兼容不等于功能齐全，默认往保守走）。

第二层：模型目录项的显式覆盖。 getCompat() 逐字段 ?? 合并：

function getCompat(model: Model<"openai-completions">): ResolvedOpenAICompletionsCompat {
  const detected = detectCompat(model);
  if (!model.compat) return detected;
  return {
    supportsStore: model.compat.supportsStore ?? detected.supportsStore,
    supportsDeveloperRole: model.compat.supportsDeveloperRole ?? detected.supportsDeveloperRole,
    ...
  };
}

逐字段 ?? 意味着覆盖是稀疏的——模型只需要声明它和探测结果不一样的那些字段。这对第三方接入的价值是直接的：你的自建网关只需要写差异，不用写全貌。

{
  "id": "my-model",
  "api": "openai-completions",
  "provider": "my-gateway",
  "baseUrl": "http://localhost:8000/v1",
  "compat": {
    "maxTokensField": "max_tokens",
    "thinkingFormat": "qwen",
    "supportsMidConvoSystemMessages": true
  }
}

三行覆盖，剩下 24 个字段自动继承探测结果。而且五份 schema 都标了 additionalProperties: true，自定义扩展项也能塞进去。

模型目录是生成的，不是手写的

packages/ai/src/models.generated.ts 的头两行说明了性质：// This file is auto-generated by scripts/generate-models.ts。生成器有 3646 行，输入来自多个来源（models-dev-reasoning-options、ai-gateway-pricing、openrouter-catalog、radius-config）。

这件事的意义不只是省事：**它是 supportsMidConvoSystemMessages 这类开关的真相来源。**那些 schema 描述里写着 "The generated catalog enables this for verified models"——「已验证」背后是一次性的批量核对，而不是每次运行时的启发式判断。

每个供应商的 .models.ts 从一份 JSON 数据文件读入，然后统一导出三个形状：

export const ANTHROPIC_MODELS: ChatModelCatalog<typeof values, "anthropic"> =
  flattenChatModelCatalog("anthropic", values);
export const ANTHROPIC_IMAGE_MODELS: ImageModelCatalog<typeof values, "anthropic"> =
  flattenImageModelCatalog("anthropic", values);
export const ANTHROPIC_CLASSIFIER_MODELS: ClassifierModelCatalog<typeof values, "anthropic"> =
  flattenClassifierModelCatalog("anthropic", values);

注意：**每个供应商都有三个导出，哪怕后两个是空的。**统一了形状，调用方就不需要按供应商分支。

这里有一个必须说清楚的边界：这些 JSON 数据文件不在版本库里，需要跑 npm run generate-models 生成。所以「pi-ai 内置了多少个模型」在没生成过产物的 checkout 里无法回答，只能回答「有多少个供应商有模型」。任何声称具体模型数量的说法都得先跑一次生成。

三个失效点

一、探测层不认你。 detectCompat 靠 provider id 和 baseUrl 子串。自建端点跑在某个已知供应商域名背后（反代）会命中错误分支，解法是显式覆盖。

二、默认 false 是安全的，但会损失能力。 supportsMidConvoSystemMessages 标 true 的是核对过的，标 false 可能只是没核对过。一个其实支持中途 system message 的模型会走「折叠重放」路径，效果是工具变更时 prompt 缓存失效。

三、compat 决定的是「请求怎么发」，覆盖不了「响应里多出来的字段」。 结构性差异只能靠类型上的可选字段承载，比如 ToolCall.thoughtSignature（Google）和 ToolCall.namespace（OpenAI Responses）。

17.3 TranscriptContext：系统提示住在对话里

这是 pi-ai 最容易被误解、也最漂亮的一个设计。

常规做法 vs Pi 的做法

绝大多数 LLM SDK 的接口是 create({ model, system, messages, tools })——系统提示和工具声明是请求对象上的独立字段，和对话历史是平级的三个东西。生态里所有的 prompt 模板系统、所有的工具注册机制，都围绕这个形状长出来。

Pi 把它们全部塞进对话历史的第一条消息：

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

declare const transcriptContextBrand: unique symbol;

export type TranscriptContext = {
  messages: Message[];
  readonly [transcriptContextBrand]: true;
};

TranscriptContext 上只有 messages 一个字段——systemPrompt 和 tools 根本不存在。源码注释把意图说得很直接：

Only normalizeContext() produces this type, so a raw Context cannot reach provider code by accident.

品牌类型。 transcriptContextBrand 是一个 unique symbol，只能由 normalizeContext() 产生。这不是类型洁癖，是一条真实防线——而这条防线由适配器接口强制执行：

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

所有 15 个 API 适配器的 stream 签名里，context 的类型都是 TranscriptContext。 写适配器的人想拿到原始 Context 是拿不到的——类型系统直接不给。

归一化只有一次

export function normalizeContext(context: Context): TranscriptContext {
  const initialMessage = createInitialSystemMessage(context.systemPrompt, context.tools);
  const messages = initialMessage ? [initialMessage, ...context.messages] : context.messages;
  return { messages } as TranscriptContext;
}

createInitialSystemMessage 在两者都为空时返回 undefined，所以 messages: [] 的空上下文不会变成「一条空 system 消息」——空转录保持为空。

调用层次是分层的：Models.stream() / streamSimple() / complete() / completeSimple() 接受 Context 并调一次 normalizeContext()；Provider.stream()、ProviderStreams、以及所有 api/*.ts 只接受 TranscriptContext。

对使用者的影响很具体：把 pi-ai 当库用，Context 那一层是你唯一要接触的；写自己的 provider 时，你面对的是 TranscriptContext，提示和工具要从 context.messages 里读出来。

真正的收益：中途变更

把提示和工具放进历史之后，它们就可以在历史中间被修改。

export interface SystemMessage {
  role: "system";
  content: string | TextContent[];              // 首条=基线提示，之后=追加指令
  sections?: Record<string, string | null>;      // 按名替换或删除提示段（null=删除）
  toolsAdded?: Tool[];                          // 从这里开始可用的工具（完整定义）
  toolsRemoved?: ToolReference[];               // 从这里开始不可用的工具（只有名字）
  timestamp: number;
}

SystemMessage 不只是首条消息，它可以是第 7 条、第 23 条、第 100 条。README 给了完整例子：

const messages: Message[] = [
  { role: "system", content: "You are helpful.", sections: { rules: "<rules>Be brief.</rules>" }, toolsAdded: [readTool], timestamp: 1 },
  { role: "user",   content: "hi", timestamp: 2 },
  { role: "system", content: "", sections: { rules: "<rules>Be thorough.</rules>" }, toolsRemoved: [{ name: "read" }], timestamp: 3 },
];

getCurrentSystemPrompt(messages); // "You are helpful.\n\n<rules>Be thorough.</rules>"
getCurrentTools(messages);        // []

第一条建立基线，第二条是用户输入，第三条把 rules 段从「简短」改成「详细」并撤掉 read 工具。结果是重放出来的当前状态。

重放函数有两个。工具集那个实现很朴素：

export function getCurrentTools(messages: TranscriptMessages): Tool[] {
  const tools = new Map<string, Tool>();
  for (const message of messages) {
    if (!isSystemMessage(message)) continue;
    for (const tool of message.toolsRemoved ?? []) tools.delete(tool.name);
    for (const tool of message.toolsAdded ?? []) tools.set(tool.name, tool);
  }
  return [...tools.values()];
}

注意顺序：先 remove 再 add，所以同一条消息里既移除又新增同名工具时，新增胜出。

中途变更怎么送到各家供应商

两条路，由 supportsMidConvoSystemMessages 决定：

export function resolveTranscript(
  context: TranscriptContext,
  supportsMidConvoSystemMessages: boolean | undefined,
): TranscriptContext {
  return supportsMidConvoSystemMessages ? context : collapseSystemMessages(context);
}

**这是 compat 层最有商业价值的一个开关。**一个长会话里模型中途启用/停用了某些工具，对支持它的模型只是一条追加消息，缓存前缀完好；对不支持的模型，整条首条 system 消息要重写，缓存全废。Anthropic 的 prompt cache 按前缀匹配，一改就是全量重算。

**同一个工具变更，在不同模型上的实际代价差一个数量级。**这就是为什么这个开关值得用 27 项 schema 里的一个字段去精确声明。

工具变更的第二种处理：anchored additions

工具变更还有一个更细的分支。resolveTranscriptTools 处理「工具声明放在哪」：

export function resolveTranscriptTools(messages: TranscriptMessages, supportsToolAdditions: boolean): TranscriptTools {
  const anchorsAdditions = supportsToolAdditions && !hasNonAdditiveToolChanges(messages);
  return {
    requestTools: anchorsAdditions
      ? (getInitialSystemMessage(messages)?.toolsAdded ?? [])
      : getCurrentTools(messages),
    anchorsAdditions,
  };
}

hasNonAdditiveToolChanges 在历史里有移除或同名重定义时返回 true。这两种没法「就地追加」，只能重发完整工具列表。

README 里两家做法的细节值得引：

Anthropic models that also set supportsMidConvoToolChanges send tool changes as native blocks (inline-tools-2026-09-15 beta): the top-level tool list holds the initial tools plus a stable deferred placeholder and never changes, later tools are defined by value in tool_addition blocks..., and removals are tool_removal references, so tool changes do not invalidate the prompt cache.

「顶层工具列表放初始工具 + 一个稳定的 deferred 占位符，然后永远不变」——顶层列表不变所以缓存前缀稳定；新工具以块的形式出现在它生效的位置。

工具声明的比较方式

算工具状态变化要比较两个工具定义，但 Tool 是从 TypeBox schema 生成的，里面有 symbol 键和 undefined 字段：

export function toToolDeclaration(tool: Tool): Tool {
  return {
    name: tool.name,
    description: tool.description,
    parameters: JSON.parse(JSON.stringify(tool.parameters)) as Tool["parameters"],
    ...(tool.constrainedSampling === undefined ? {} : { constrainedSampling: tool.constrainedSampling }),
  };
}

export function declarationsEqual(left: Tool, right: Tool): boolean {
  return JSON.stringify(toToolDeclaration(left)) === JSON.stringify(toToolDeclaration(right));
}

**先 JSON round-trip，再比较序列化结果。**注释解释了三个理由：round-trip 丢掉 typebox 的 symbol 键和 undefined；两个对象用同样的键序构造，所以序列化比较是精确的；以及最重要的一条——avoid a deep-equal dependency in a browser-safe package。JSON round-trip 顺手解决三个问题，零依赖。

而定义变化的处理方式是「一次移除 + 一次添加」——工具描述改了一句话，对 transcript 来说就是「这个工具没了，那个新工具来了」。这会让 hasNonAdditiveToolChanges 判定为 true，走完整工具列表重发——保守但正确。

为什么这套设计值得学

代价是每个适配器都要多写一段「把 system 消息投影到我的原生格式」。收益有四条：会话内换模型成为可能（状态全在历史里，切换供应商不需要重建上下文，只需要换一个能读懂同一种 transcript 的适配器，README 的 "Cross-Provider Handoffs" 讲的正是这件事）；工具和提示的变更有了地址（它们是历史里的具体位置，不是隐式的全局状态）；diff 变得有意义（一次系统提示变更就是历史里多了一条消息，可以被审、被版本控制、被 fork——常规做法里改个系统提示是一次隐式的状态覆盖，你没法回答「这个会话当时用的提示是什么」）；整个上下文就是一段 JSON（模型信息也一样，BaseModel 全是纯数据，所以「这个会话当时用哪个模型」也是一次 JSON.stringify 的事）。

17.4 事件流与懒加载

事件协议：12 个事件

export type AssistantMessageEvent =
  | { type: "start"; partial: AssistantMessage }
  | { type: "text_start";     contentIndex: number; partial: AssistantMessage }
  | { type: "text_delta";     contentIndex: number; delta: string; partial: AssistantMessage }
  | { type: "text_end";       contentIndex: number; content: string; partial: AssistantMessage }
  | { type: "thinking_start"; contentIndex: number; partial: AssistantMessage }
  | { type: "thinking_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
  | { type: "thinking_end";   contentIndex: number; content: string; partial: AssistantMessage }
  | { type: "toolcall_start"; contentIndex: number; partial: AssistantMessage }
  | { type: "toolcall_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
  | { type: "toolcall_end";   contentIndex: number; toolCall: ToolCall; partial: AssistantMessage }
  | { type: "done";  reason: ...; message: AssistantMessage }
  | { type: "error"; reason: ...; error: AssistantMessage };

三个 block 类型 × 三态（start / delta / end），加两个终结事件。 这个形状是通用的——不管底层是 OpenAI 的 SSE、Anthropic 的事件序列还是 Bedrock 的 event stream chunk，暴露出来的都是这 12 种。

partial 不是快照

这是最容易写错、也最需要单独讲的一条。源码注释说它是 "the shared live response-so-far helper, not an event-time snapshot"；README 的措辞更直接：

It is intentionally not an event-time snapshot: providers may mutate the same message and content blocks as generation advances, including while older events wait in the stream queue. Inspect it when handling an event instead of retaining it as historical state.

最后那句是关键：在处理事件时读它，不要把它留下来当历史状态。

机制上，partial 是同一个 AssistantMessage 对象被反复 mutate，事件里带的是它的引用：

// ✅ 正确：当场读
for await (const event of stream) {
  if (event.type === "text_delta") process.stdout.write(event.delta);
  if (event.type === "toolcall_start") render(event.partial);
}

// ❌ 错误：留着当历史
const snapshots = [];
for await (const event of stream) snapshots.push(event.partial);  // 全是同一个引用

最后一行得到的是 N 份指向同一个对象的引用。这不是性能问题，是正确性问题。

还有两个细节。text_start 时文本块是空的，之后靠 text_delta 增长，最后 text_end 给权威值——UI 在 text_start 就把块渲染成「已完成」会闪一下空白。

以及一条容易被忽略的约束——事件可能跨 block 交错：

Consumers must use contentIndex to associate each delta/end event with its block and must not assume that a block's *_start/*_delta/*_end sequence is uninterrupted by events for other blocks.

序列可能是 text_start → text_delta → toolcall_start → text_delta → toolcall_delta，同一个文本块的 delta 之间夹着别块的事件。

EventStream 的实现

容器是一个手写的 AsyncIterable，内部两个 FifoQueue——一个装事件，一个装等待中的消费者。FifoQueue 是双数组实现：incoming 顺序入队，outgoing 逆序出队，dequeue() 时把 incoming 整个倒进 outgoing 再 pop。这是为了避开 JS 数组 shift() 的 O(n)。

它同时提供两条消费路径：for await 逐事件消费，await stream.result() 只等最终结果。后者拿到的是 AssistantMessageEventStream，它在推入终结事件时顺手计时：

#time(message: AssistantMessage): void {
  if (this.done || message.durationMs !== undefined || message.timestamp < this.#startedAt) return;
  message.durationMs = Math.max(0, Math.round(performance.now() - this.#startedAtMonotonic));
}

三个防护条件：已经算过了、尊重外来值、以及「这是后来才取到的结果（比如延迟取回的 deferred），不该算到这次流头上」。用的是 performance.now() 单调时钟，不受系统时间调整影响。

错误也在流里

契约把「什么时候可以抛」划得很清楚：streamSimple() 返回之前（比如缺 auth）可以同步抛出；返回之后的所有失败必须编码进流，终止于 error 事件。原因很简单——返回之后调用方已经在 for await 里了，抛异常没法被 catch 到。

而且 error 事件带的是一个完整的 AssistantMessage，不是 Error 对象：

| { type: "error"; reason: Extract<StopReason, "aborted" | "error">; error: AssistantMessage }

这意味着出错时的部分内容不丢——模型生成了 500 字然后被拒绝，这 500 字在 error.content 里。

StopReason 一共七个值：pending、stop、length、toolUse、error、aborted、deferred。done 只带 stop/length/toolUse/deferred，error 只带 aborted/error——类型层面就保证了 done 事件不会带一个 pending。

15 个懒加载模块

packages/ai/src/api/ 下有 15 个 .lazy.ts 文件（ls packages/ai/src/api/*.lazy.ts | wc -l），覆盖 10 种聊天协议 + 1 种图像协议 + 4 种分类协议中的 4 种。

机制在 api/lazy.ts，核心是这一段：

export function lazyStream(
  model: Model<Api>,
  setup: () => Promise<AsyncIterable<AssistantMessageEvent>>,
): AssistantMessageEventStream {
  const startedAt = Date.now();
  const outer = new AssistantMessageEventStream();

  setup()
    .then((inner) => forwardStream(outer, inner))
    .catch((error) => {
      const message = createSetupErrorMessage(model, error, startedAt);
      outer.push({ type: "error", reason: "error", error: message });
      outer.end(message);
    });

  return outer;
}

这个函数的签名是整节最值得看的地方：lazyStream 同步返回流，异步准备工作藏在后面。

这一点很重要，因为 API 契约要求「拿不到 auth 要同步抛」。如果 lazyStream 是 async，它在类型上就做不到「同步返回流」。于是做了个变通：异步准备工作失败时，把错误变成 error 事件推进流里。

注意 createSetupErrorMessage 造的是结构完整的 AssistantMessage——content: []、usage 全零、stopReason: "error"，不是随手扔个 Error。因为流协议要求终结事件必须是这个形状。

lazyApi 把这层包装成 ProviderStreams，并用 capabilities 控制可选方法挂不挂——不支持 deferred 的 API 就不会暴露 fetchDeferred，这样 ProviderStreams 里的可选成员真正起到「能力探测」的作用。

实际收益：三个官方 SDK 加起来不小，你只用 Anthropic 就不该下载 OpenAI 和 AWS 的 SDK。README 的打包章节给了入口表：

但有一条诚实的提醒：没有代码分割就没有懒加载。

Without code splitting, bundlers fold reachable lazy API implementations into the single bundle.

这不是 pi-ai 的缺陷，是 bundler 的基本行为，但知道自己构建出来是什么很重要。Bedrock 还是个特例：它通过「bundler 不透明」的导入加载 AWS SDK，单文件 bundle 里要用得显式调用 setBedrockProviderModule() 注册。

一个已经标了「临时」的模块

packages/ai/src/compat.ts 是给旧版全局 API 用的过渡层，头注释写着 "This module is deleted with the coding-agent ModelManager migration."，package.json 的 sideEffects 里也只有它和两个 image 相关模块。README 里还有一条禁令：Avoid @earendil-works/pi-ai/compat in new bundled apps。

**判断一个包是不是被认真维护的信号，是看它有没有明确标记的临时层。**明确说「这个会删」比悄悄留着强。

17.5 工具调用与部分 JSON

工具调用参数是流式到达的：先来 {"pa，再来 th": "src/in，最后才是完整的 JSON。中间任何时刻拿到的都是不完整的 JSON。

部分解析：四级降级

utils/json-parse.ts 的 parseStreamingJson 是一条降级链：

export function parseStreamingJson<T = Record<string, unknown>>(partialJson: string | undefined): T {
  if (!partialJson || partialJson.trim() === "") return {} as T;
  try { return parseJsonWithRepair<T>(partialJson); } catch {
    try { const result = partialParse(partialJson); return (result ?? {}) as T; } catch {
      try { const result = partialParse(repairJson(partialJson)); return (result ?? {}) as T; } catch {
        return {} as T;
      }
    }
  }
}

顺序是：**完整 JSON.parse（含修复）→ partial-json 解析 → 修复后再 partial-json 解析 → 空对象。**那个 partial-json 依赖（版本 0.1.7）干的是第二件事。

不管走哪条路，返回值保证是一个对象。所以消费方永远不需要判空，只需要判「字段在不在」。

repairJson 修两种畸形：字符串里的裸控制字符，以及非法转义前的反斜杠。

/**
 * Repairs malformed JSON string literals by:
 * - escaping raw control characters inside strings
 * - doubling backslashes before invalid escape characters
 */

模型在 JSON 字符串里输出裸换行很常见（它把「内容」和「格式」混了），裸控制字符会让 JSON.parse 直接失败。

完成时和流式时是两套逻辑

这是一个很关键的设计区分：

/**
 * Unlike {@link parseStreamingJson}, this **never keeps a partially parsed prefix**:
 * once the call is finished, text that is not a valid JSON object means the model
 * sent malformed arguments. In that case `arguments` is `{}` and `argumentsError`
 * describes the problem, including the text the model sent, so tool validation
 * can report it back to the model.
 */
export function parseToolCallArguments(json: string | undefined): Pick<ToolCall, "arguments" | "argumentsError"> {
  if (!json?.trim()) return { arguments: {} };
  try {
    const parsed = parseJsonWithRepair<ToolCall["arguments"] | null>(json);
    if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
      throw new Error("Tool arguments must be a JSON object");
    }
    return { arguments: parsed };
  } catch (error) {
    return {
      arguments: {},
      argumentsError: `${error instanceof Error ? error.message : String(error)}\n\nReceived arguments:\n${json}`,
    };
  }
}

为什么完成后要变严格？ 流式时宽松是必要的（数据还没到齐），完成后宽松就是纵容——一个拼错的参数被半猜半就地执行，比直接报错糟糕得多。

argumentsError 里带了模型原始发来的文本，这个字段的设计意图是让错误能回传给模型让它自我纠正。模型看到「你的参数不是合法 JSON 加上你实际发的东西」，比看到一个抽象的「参数校验失败」有用得多。

ToolCall 上另有两个供应商特例的逃生舱：thoughtSignature（Google 特有，多轮携带思考上下文签名）和 namespace（OpenAI Responses 的动态工具命名空间）。

一个供应商完全不支持的事

README 里有一条硬限制：

The Google provider does not support function call streaming. Instead, you will receive a single toolcall_delta event with the full arguments.

所以 Google 上跑的会话，toolcall_delta 只会有一个事件，内容是完整参数。**任何依赖「参数逐步到达」做 UI 的代码，在 Google 上会退化成一次性更新。**这是协议层面的能力差异，pi-ai 没法补，只能如实暴露。

校验的责任在你

toolcall_end 的 toolCall 是权威值，但注释写得很清楚："Complete, but not schema-validated"。schema 校验是消费方的循环该做的事，而校验失败要作为工具结果回给模型，而不是崩掉：

for await (const event of s) {
  if (event.type === 'toolcall_end') {
    try {
      const validatedArgs = validateToolCall(tools, event.toolCall);
      const result = await executeMyTool(event.toolCall.name, validatedArgs);
    } catch (error) {
      context.messages.push({
        role: 'toolResult', toolCallId: event.toolCall.id, toolName: event.toolCall.name,
        content: [{ type: 'text', text: error.message }], isError: true, timestamp: Date.now()
      });
    }
  }
}

这是让模型能自我纠正的关键——它看到自己的参数没被接受，可以改一次再试。

17.6 认证解析

只有两种认证类型

export type AuthType = "api_key" | "oauth";

没有第三种。这两种覆盖了全部场景：有静态密钥的走 api_key，走订阅的（Claude Pro/Max、ChatGPT Plus/Pro、GitHub Copilot）走 oauth。

而「AWS profile」「ADC 文件」「环境变量」不属于第三种类型——它们是 api_key 认证的环境来源。源码注释：

Api-key auth: stored key/provider env plus ambient sources (env vars, AWS profiles, ADC files). Ambient-only providers omit login.

Bedrock 用 IAM profile，没有「登录」这个概念——凭据在环境里，所以它的 api_key 认证只有 resolve()，没有 login()。这就是为什么认证类型是两种而不是三种：环境凭据是 api_key 认证的一种配置形态，不是一类新的东西。

而 ProviderAuth 的注释给了一条硬约束：

At least one of apiKey/oauth must be present: even ambient-credential providers and keyless local servers provide apiKey auth whose resolve() reports whether the provider is configured.

「无密钥的本地服务器」（Ollama、LM Studio）也必须有 apiKey 认证，只是它的 resolve() 总是返回「已配置」。每个供应商都有认证语义，这样 Models.getAuth() 的返回类型才统一：要么给一个 AuthResult，要么给 undefined。

ApiKeyAuth 的三个成员里只有 resolve 必需。login 缺失意味着「只有环境凭据，UI 里不显示登录按钮」；check 缺失意味着 Models 改为调 resolve() 来检查。而 resolve 的合并规则是逐字段的——注释写的是 "merging per field (credential.key ?? env("..."), credential.env?.NAME ?? env("..."))"，也就是说你存了一个 key 但没存 provider env，那个 env 还是从环境里取。credential 参数本身是可选的，传 undefined 表示「存储里没有，去环境里找」。

解析顺序：存储优先，没有静默回退

auth/resolve.ts 的主函数注释就是规则：

A stored credential owns the provider: ambient/env is consulted only when nothing is stored. No silent env fallback after a failed refresh or for a credential type without a matching handler.

第 4 行是那条规则的执行点。为什么重要：如果你登出了某个 OAuth 账号，存储里是空的，这时如果悄悄回退到环境变量里的一个过期 key，请求会以你的名义发出去，而你以为自己是登出状态。宁可明确失败。

OAuth 刷新：双检锁 + 明确的超时边界

const DEFAULT_OAUTH_MINIMUM_VALIDITY_MS = 5 * 60 * 1000;  // 5 分钟
const DEFAULT_OAUTH_REFRESH_TIMEOUT_MS = 15_000;           // 15 秒

**剩余有效期少于 5 分钟就刷新。**不是为了「快过期了」，而是为了给一次请求留余量——一个 30 秒的流式请求不能在中途 token 过期。刷新在凭据存储的锁里做，两个检查都在锁内：

post = await credentials.modify(
  providerId,
  async (current) => {
    if (current?.type !== "oauth") return undefined;   // 中途登出了
    if (!needsRefresh(current)) return undefined;     // 别的进程已经刷过了
    try {
      return await oauth.refresh(current, AbortSignal.timeout(DEFAULT_OAUTH_REFRESH_TIMEOUT_MS));
    } catch (error) {
      throw new ModelsError("oauth", `OAuth refresh failed for ${providerId}`, { cause: error });
    }
  },
  { signal: lockWait.signal },
);

然后是这一整段里最微妙的注释：

signal cancels only the wait for the lock. Once a refresh starts, the provider may already have rotated the refresh token, so the refresh and its persistence ignore signal and are bounded only by a timeout. Otherwise a cancelled caller could discard the only valid refresh token.

翻译一下这个设计：

因为刷新令牌是轮转的——很多 OAuth 服务刷新时会发一个新的并作废旧的。如果刷新进行到一半用户按了 Ctrl+C，持久化被取消，那个新令牌就丢了，旧的已经被作废——账号直接登出。

所以策略是：取消可以取消「还没开始的事」，但不能取消「已经改了世界的事」。需要快速响应的调用方把这个 Promise 和自己的 signal 赛跑——也就是可以在等待侧放弃，但不能在写入侧放弃。

CredentialStore：唯一的写路径

export interface CredentialStore {
  read(providerId: string, options?): Promise<Credential | undefined>;
  list(options?): Promise<readonly CredentialInfo[]>;
  modify(providerId: string, fn: (current) => Promise<Credential | undefined>, options?): Promise<Credential | undefined>;
  delete(providerId: string, options?): Promise<void>;
}

三个设计点：modify 是唯一写路径（读—改—写天然是竞态，做成受锁保护的回调后竞态就不存在了，上面那个 OAuth 刷新就依赖这一点）；fn 能看到当前值，返回 undefined 表示不改（刷新时能判断「是不是已经刷过了」，登出时能发现「已经是空的」）；跨进程互斥——注释写的是 "cross-process too where the backing store supports it (e.g. a file lock)"，文件锁的实现在 app 层（coding-agent 的 AuthStorage），pi-ai 只定义接口。

list() 还有一条安全约束：**Implementations must not execute configured API-key commands while listing.**有些供应商支持「用一条命令取 key」（AWS 的 credential_process），列举账号状态时执行这些命令既是不必要的风险，也是同步阻塞的。

两个 OAuth 成员的分工

export interface OAuthAuth {
  isSubscription?: boolean;
  loginLabel?: string;
  login(interaction, options?): Promise<OAuthCredential>;
  refresh(credential: OAuthCredential, signal: AbortSignal): Promise<OAuthCredential>;
  toAuth(credential: OAuthCredential): Promise<ModelAuth>;
}

注释解释了 refresh / toAuth 的拆分理由：「The refresh/toAuth split lets Models own the locked refresh pattern」。刷新的锁逻辑由 Models 拥有，不是每个 provider 各写一遍——provider 只提供「怎么换」，「什么时候换、换的时候锁着」是框架的事。

toAuth 是 async 的，因为 GitHub Copilot 这类 provider 的 baseUrl 是按凭据的（不同订阅不同的网关地址）。注释点名了这个 case。

请求级的环境覆盖

AuthResolutionOverrides 的 env 字段效果是覆盖认证上下文的环境查询：

function overlayEnvAuthContext(base: AuthContext, env: ProviderEnv): AuthContext {
  return {
    env: async (name) => env[name] || (await base.env(name)),
    fileExists: (path) => base.fileExists(path),
  };
}

**这是「请求级」的，不是进程级的。**同一时刻两个请求可以用不同的环境值调同一个 provider。对多租户场景很关键：两个租户用同一个 provider id 但不同凭据，可以并发跑，不用改进程环境。

17.7 成本与用量

Usage 结构

export interface Usage {
  input: number;
  output: number;
  cacheRead: number;
  cacheWrite: number;
  cacheWrite1h?: number;   // cacheWrite 中用 1h 保留期写入的子集。只有 Anthropic 报这个拆分
  reasoning?: number;     // 推理 token。output 的子集：output 已经包含它们
  totalTokens: number;
  cost: { input: number; output: number; cacheRead: number; cacheWrite: number; total: number };
}

三条设计注释，每条都在防一个具体的坑：

cacheWrite1h 为什么存在？ 因为 Anthropic 对 1 小时保留期的缓存写入收两倍的输入价。要算准成本，必须知道这批写入里有多少走了长保留期。

reasoning 为什么标注是 output 的子集？ 注释明确：output already includes these tokens。怕的是有人拿 output + reasoning 算 token 数，那样会重复计算。它存在只是为了「显示推理占比」这类用途。

cost 为什么嵌在 Usage 里？ 因为成本随 usage 一起产生，绑在一起最省事。

成本计算：两处特殊处理

models.ts 的 calculateCost()：

export function calculateCost(model: AnyModel, usage: Usage): Usage["cost"] {
  const inputTokens = usage.input + usage.cacheRead + usage.cacheWrite;
  let rates: ModelCostRates = model.cost;
  let matchedThreshold = -1;
  for (const tier of model.cost.tiers ?? []) {
    if (inputTokens > tier.inputTokensAbove && tier.inputTokensAbove > matchedThreshold) {
      rates = tier;
      matchedThreshold = tier.inputTokensAbove;
    }
  }

  // Anthropic charges 2x base input for 1h cache writes.
  const longWrite = usage.cacheWrite1h ?? 0;
  const shortWrite = usage.cacheWrite - longWrite;
  usage.cost.input = (rates.input / 1000000) * usage.input;
  usage.cost.output = (rates.output / 1000000) * usage.output;
  usage.cost.cacheRead = (rates.cacheRead / 1000000) * usage.cacheRead;
  usage.cost.cacheWrite = (rates.cacheWrite * shortWrite + rates.input * 2 * longWrite) / 1000000;
  usage.cost.total = usage.cost.input + usage.cost.output + usage.cost.cacheRead + usage.cost.cacheWrite;
  return usage.cost;
}

第一，阶梯定价按「整个请求」结算，不是分段结算。 注释写明 "Request-wide pricing tiers. The highest matching input threshold applies to the full request."。有些供应商确实这样计费——一次请求输入超过 200K，整次按长上下文的费率算，不是前 200K 和后 3K 分开算。判定的 inputTokens 是 input + cacheRead + cacheWrite，全部输入侧加起来——档位看的是这次请求总共带进去多少上下文，不管它们被计为普通输入还是缓存读。

第二，缓存写入的定价是混合的。rates.cacheWrite * shortWrite + rates.input * 2 * longWrite——短保留期按缓存写入价，长保留期按输入价的两倍。这一行是全包唯一硬编码的供应商特例，注释直接点名了 Anthropic。

定价的声明方式

const ModelCostRatesProperties = {
  input:      Type.Number({ description: "Input cost in USD per million tokens." }),
  output:     Type.Number({ description: "Output cost in USD per million tokens." }),
  cacheRead:  Type.Number({ description: "Cache-read cost in USD per million tokens." }),
  cacheWrite: Type.Number({ description: "Cache-write cost in USD per million tokens." }),
};

export const ModelCostTierSchema = Type.Object({
  inputTokensAbove: Type.Number({ description: "Use this tier when total request input exceeds this token count." }),
  ...ModelCostRatesProperties,
});

四个费率，单位是每百万 token 的美元数。除以 1000000 是在 calculateCost 里做的，目录里存整数还是小数点几位都无所谓。阶梯是可选数组，inputTokensAbove 是触发阈值，计算时取阈值最大的匹配项，所以阈值不需要排序。

上下文用量估算

有 provider 报的时候用 provider 报的，没有的时候要估。utils/estimate.ts 做后者：

const CHARS_PER_TOKEN = 3.5;
const ESTIMATED_IMAGE_CHARS = 4800;

**字符数除以 3.5 得到 token 数。**这是个经验常数，对中文偏保守（中文一个字大概 1 个 token，所以 3.5 会低估）。每张图折算成 4800 个字符，约 1370 个 token。

估算的关键设计是它不是从头算全部：

export function estimateContextTokens(context: TranscriptContext | readonly Message[]): ContextUsageEstimate {
  const messages = "messages" in context ? context.messages : context;
  const usageInfo = getLastAssistantUsageInfo(messages);
  if (usageInfo) {
    const usageTokens = calculateContextTokens(usageInfo.usage);
    let trailingTokens = 0;
    for (let i = usageInfo.index + 1; i < messages.length; i++) {
      trailingTokens += estimateMessageTokens(messages[i]);
    }
    return { tokens: usageTokens + trailingTokens, usageTokens, trailingTokens, lastUsageIndex: usageInfo.index };
  }
  let tokens = 0;
  for (const message of messages) tokens += estimateMessageTokens(message);
  return { tokens, usageTokens: 0, trailingTokens: tokens, lastUsageIndex: null };
}

只估算最近一次 usage 之后的消息。 因为 provider 报的数字是权威的，之前的部分已经被计过了，没必要重算。

返回结构把两个数分开——tokens（总计）、usageTokens（provider 报的）、trailingTokens（估算的）、lastUsageIndex（用了哪条消息）。把权威值和估算值分开返回，调用方能知道自己的数字有多可信。压缩策略需要读这里——细节在 上下文与压缩。

calculateContextTokens 把四个数字加起来，reasoning 不在里面：

export function calculateContextTokens(usage: Usage): number {
  return usage.input + usage.output + usage.cacheRead + usage.cacheWrite;
}

因为它是 output 的子集，加了就重复了。这和 Usage 里那条注释是配套的。

其他元数据

同一个 schema 文件里还有几组非成本的元数据：

CACHE_RETENTIONS 有三档而 MODEL_PROMPT_CACHE_RETENTIONS 只有两档，这个不对称是有意的：none 表示「不缓存」，它是请求的意图而不是模型的能力。模型字段 promptCache 描述的是「这两档缓存各能活多久（秒）」，没填就说明该 provider 的缓存行为未知——Pi 不会盲目去热身一个时长不明的缓存。

模型是纯数据

有一个对使用者很关键的性质：Models are plain serializable data too — no functions or implementations attached，所以「这个会话当时用哪个模型」就是一次 JSON.stringify 的事。

BaseModel 里全是字符串、数字和数组，没有函数、没有类实例、没有 provider 引用。所以一个 Model 可以安全地 JSON.stringify 存进数据库，几天后反序列化出来再传给 stream()——它需要的全部信息都在里面。compat 也在里面，因为它本来就是纯 JSON 对象。

这也解释了动态 provider 的机制：OpenRouter、Radius 这类网关通过 refreshModels() 拉目录，ModelsStore 持久化 models 加 lastModified 加 etag，下次用 If-None-Match 问一次服务器，没变就 304。目录数据的形态和静态目录完全一样，消费者分不出区别。

17.8 事实来源

本文所有数字、开关名和枚举值都可在仓库里复算：

完整事实表（含每条断言的推导命令）见专栏仓库的 research/pi-agent/FACTS.md。