v1.1.0 Pi v1.1.0 源码分析 Analysis of Pi v1.1.0 · 2026-10-07
pi-ai pi-ai 统一的 LLM API：模型目录、认证、42 个 provider、事件流 Unified LLM API with automatic model discovery and provider configuration
pi-ai Infra
← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言 ← → navigate · F fullscreen · O overview · T theme · L language
01 · 入口与打包 01 · Entry points and bundling pi-ai
入口与打包 Entry points and bundling pi-ai entry points and bundling 懒加载链 Lazy-loading chain 15 个 *.lazy.ts 包装 15 *.lazy.ts wrappers @earendil-works/pi-ai 无副作用 side-effect free TypeBox 立即加载 eager TypeBox …/pi-ai/models 最小 smallest …/providers/<id> SDK 首次请求时加载 SDK on first request …/providers/all 目录全部打包 all catalogs bundled SDK 仍懒加载 SDKs still lazy …/api/<id> SDK 立即加载 eager SDK …/oauth · /bun-oauth 按需 on demand …/utils/* 按文件 per file …/pi-ai/compat 已弃用 deprecated providers/anthropic.ts createProvider({ api }) anthropic-messages.lazy.ts lazyApi(() => import(…)) anthropic-messages.ts stream · streamSimple @anthropic-ai/sdk 首次请求才加载 loaded on first request 导入路径 IMPORT PATH 得到什么 WHAT YOU GET 加载成本 LOAD COST 核心：类型、Type、createModels、createProvider、EventStream、工具函数 core: types, Type, createModels, createProvider, EventStream, helpers 仅 Models 运行时（不含 TypeBox 校验） Models runtime only (no TypeBox validation) 一个 provider：模型目录 + 认证 + 懒加载的 API one provider: catalog + auth + lazily-loaded API builtinModels() · builtinProviders()（42 个） builtinModels() · builtinProviders() (42) 直接调用某个 API 实现（stream / streamSimple） one API implementation directly (stream / streamSimple) OAuth 登录辅助（Node / Bun） OAuth login helpers (Node / Bun) event-stream · validation · transcript · overflow … 旧的全局 API（stream / complete / getModel）超集 old global API superset (stream / complete / getModel) 每个 API 都有一个 *.lazy.ts：lazyApi(() => import("./x.ts"))，provider 只引用这个包装。 Every API has a *.lazy.ts: lazyApi(() => import("./x.ts")); providers only reference that wrapper.
02 · Models 与 Provider 02 · Models and providers pi-ai
Models 与 Provider Models and providers The Models collection and providers refresh() · 每个 provider 的 refreshModels(ctx) refresh() · each provider’s refreshModels(ctx) builtinProviders() providers/all · 42 个 providers/all · 42 providers createProvider({…}) id · api | { [api]: … } · models fetchModels? · images? · classifiers? · auth? createModels({ providers }) 或 builtinModels() or builtinModels() «implements Models» ModelsImpl − providers: Map<id, Provider> − credentials: CredentialStore (InMemory) − store?: ModelsStore · authContext + getModel(provider, id) · getModels() + getAvailableModels() · checkAuth() + refresh({ allowNetwork, signal }) + login(id, interaction) · logout(id) + stream · complete · streamSimple · completeSimple + streamDeferred · cancelDeferred + generateImages · classify (never reject) «interface» Provider<TApi> + id · name · baseUrl? · headers? + auth?: ApiKeyAuth | OAuthAuth … + getModels() · getAllModels() + refreshModels?(ctx) + stream · streamSimple + fetchDeferred? · cancelDeferred? + generateImages? · classify? dispatch(model, run) 单一 api，或按 model.api 查表 single api, or look up by model.api 缺失 → error 流（不抛出） missing → error stream (no throw) 读取 ModelsStore read ModelsStore ctx.stored.models 先发布缓存 publish cached publish({ update }) allowNetwork？ allowNetwork? 否 / 已中止 → 结束 no / aborted → stop fetchModels(ctx) 网络获取目录 fetch the catalog 合并后发布 merge & publish 按 id 覆盖基线 override baseline by id 每个调用 each calls Provider[] new * stream 时 on stream getModels() 永远同步：返回 基线目录 ∪ 动态目录；refresh 只改变动态部分 getModels() stays synchronous: baseline ∪ dynamic models; refresh only swaps the dynamic part 一个 provider 可以按 model.api 分派到多个 API（如 Fireworks 同时用 OpenAI 与 Anthropic 兼容 API）。 One provider can dispatch to several APIs by model.api (e.g. Fireworks uses both OpenAI- and Anthropic-compatible APIs).
03 · 一次 streamSimple 调用 03 · One streamSimple call pi-ai
一次 streamSimple 调用 One streamSimple call models.streamSimple() end to end setup()（异步，在外层流背后） setup() (async, behind the outer stream) models.streamSimple( model, context, options) normalizeContext(context) → TranscriptContext（同步） → TranscriptContext (sync) lazyStream(model, setup) 立即返回外层流 returns the outer stream now 调用方开始迭代 caller iterates for await … · result() requireChatProvider 未知 provider → 抛出 unknown provider → throw applyAuth(model, opts) resolveProviderAuth() provider.streamSimple dispatch(model.api) lazyApi → import() 首次加载 API 模块 first load of the API module api.streamSimple() SDK 请求 → 内层流 SDK request → inner stream forwardStream(outer, inner) 逐个 push 事件 → end(result) push each event → end(result) 任一步骤抛出 any step throws → push({ type: "error", reason: "error" }) done / error 消息 done / error message 写入 durationMs（1.1.0） gets durationMs (1.1.0) setup() resolveProviderAuth order: 1 options.apiKey override 2 stored credential (oauth refresh | api_key) 3 ambient env (ANTHROPIC_API_KEY …) 之后合并 headers → transformHeaders → env； then merge headers → transformHeaders → env; auth.baseUrl 存在时覆盖 model.baseUrl auth.baseUrl overrides model.baseUrl catch 约定：流式 API 从不 reject —— 所有失败都编码成流里的最终 error 事件， contract: streaming APIs never reject — every failure is encoded as the final error event outer.result() 同样解析为带 stopReason 与 errorMessage 的 AssistantMessage outer.result() resolves to an AssistantMessage carrying stopReason + errorMessage 1 2 3 4 5 6 7 认证解析顺序：调用方的 apiKey 覆盖 → 已存凭据（OAuth 会先刷新）→ 环境变量。 Auth resolution order: caller apiKey override → stored credential (OAuth refreshed first) → environment.
04 · 事件语法与 EventStream 04 · Event grammar and EventStream pi-ai
事件语法与 EventStream Event grammar and EventStream AssistantMessageEvent grammar (12 types) 内容块 × n（按 contentIndex） content blocks × n (by contentIndex) EventStream<T, R> utils/event-stream.ts start partial text_start text_delta × n · delta text_end 完整块 full block thinking_start thinking_delta × n · delta thinking_end 完整块 full block toolcall_start toolcall_delta × n · delta toolcall_end 完整块 full block done stop | length | toolUse | deferred error error | aborted pending stop length toolUse error aborted deferred 生产者 producer push(event) · end(result?) queue: FifoQueue<T> waiting: FifoQueue<resolve> 消费者 consumer for await (const e of s) result(): Promise<R> done / error 时 resolve resolves on done / error reason 每个事件都带 partial（当前的 AssistantMessage）；toolcall_delta 的参数由 parseStreamingJson 增量解析 every event carries partial (the AssistantMessage so far); toolcall_delta args are parsed incrementally by parseStreamingJson StopReason pending 只出现在 partial 上；deferred = 异步任务，稍后 streamDeferred(handle) pending only on partials; deferred = async job, fetch later with streamDeferred(handle) 无人等待 no waiter 直接交付 hand off 排队等待 park isComplete(e) 为真后 done = true，后续 push 被忽略；AssistantMessageEventStream 在 done/error/end 时用 performance.now() 写入 durationMs once isComplete(e) is true, done = true and later pushes are dropped; AssistantMessageEventStream stamps durationMs (performance.now) on done / error / end FifoQueue 用两个数组实现摊还 O(1) 出队 FifoQueue uses two arrays for amortised O(1) dequeue 7 种 StopReason：pending 只出现在 partial 上；deferred 表示异步任务，之后用 streamDeferred(handle) 取结果。 Seven StopReasons: pending appears only on partials; deferred means an async job, fetched later with streamDeferred(handle).
05 · 跨模型消息转换 05 · Cross-model message transform pi-ai
跨模型消息转换 Cross-model message transform transformMessages: cross-model handoff Message[] 来自任意 provider 的历史 history from any provider transformMessages(messages, model, normalizeToolCallId?) api/transform-messages.ts 目标 API 的请求 target API request convertMessages(…) 用户消息 user message thinking 块 thinking blocks 工具调用 id tool-call ids 失败的 assistant failed assistant 孤立工具调用 orphaned tool calls system 消息 system messages 模型不支持图像 → 图像换成文字占位 model lacks image input → images become a text placeholder 同一模型：保留签名；换模型：转成文本，空的丢弃；redacted 跨模型直接丢弃 same model: keep signature; other model: become text, empty ones dropped; redacted dropped cross-model normalizeToolCallId：如 OpenAI 450+ 字符 id → Anthropic ^[a-zA-Z0-9_-]+$ ≤ 64；toolResult 同步改写 normalizeToolCallId: e.g. 450+ char OpenAI ids → Anthropic ^[a-zA-Z0-9_-]+$ ≤ 64; toolResults remapped too stopReason 为 error / aborted 的消息整条跳过（不可回放） messages with stopReason error / aborted are skipped (not replayable) 没有结果的调用 → 合成 isError 的 toolResult（"No result provided"） calls without a result → synthesize an isError toolResult ("No result provided") 在工具调用与其结果之间出现时延后，保证结果紧跟调用 held back while tool results are pending, so results stay adjacent to their calls 按顺序应用于整个 applied in order to the 历史；每次请求都会执行 whole history, every request 结果：在 Claude 中途切到 GPT（或反之）时，同一份会话仍能被新 provider 接受 result: switch from Claude to GPT (or back) mid-session and the same history is still accepted by the new provider 1 2 3 4 5 6 thinking 签名只在同一模型之间有效；跨模型时 thinking 转成普通文本，redacted thinking 直接丢弃。 Thinking signatures are only valid on the same model; across models thinking becomes plain text and redacted thinking is dropped.
06 · 一个 API 实现：anthropic-messages 06 · One API implementation: anthropic-messages pi-ai
一个 API 实现：anthropic-messages One API implementation: anthropic-messages anthropic-messages: SSE → events mapStopReason() streamSimple() 的 thinking 配置 thinking in streamSimple() message_start start content_block_start · text text_start content_block_start · thinking thinking_start … · redacted_thinking thinking_start content_block_start · tool_use toolcall_start content_block_delta text_delta thinking_delta toolcall_delta content_block_stop *_end message_delta message_stop done stop length toolUse error Anthropic SSE ANTHROPIC SSE 处理 HANDLING 发出的事件 EMITTED 记录 id · usage record id · usage 新建 text 块 new text block 新建 thinking 块 new thinking block "[Reasoning redacted]" + 签名 "[Reasoning redacted]" + signature 新建 toolCall，partialJson = "" new toolCall, partialJson = "" text / thinking / input_json / signature 删除 partialJson，完成块 delete partialJson, finish block stop_reason → mapStopReason · usage calculateCost(model, usage) end_turn · pause_turn · stop_sequence → max_tokens → tool_use → refusal · sensitive → !reasoning thinkingEnabled: false 自适应模型 adaptive models thinking: adaptive + effort（由 reasoning 映射） thinking: adaptive + effort (mapped from reasoning) 其余模型 other models adjustMaxTokensForThinking → clampMaxTokensToContext budget min(budget, maxTokens − 1024) 工具参数用 parseStreamingJson 增量解析，UI 可以在参数流式到达时就显示。 Tool arguments are parsed incrementally with parseStreamingJson, so a UI can show them while they stream.
07 · 思考等级、预算与成本 07 · Thinking levels, budgets and cost pi-ai
思考等级、预算与成本 Thinking levels, budgets and cost Thinking levels, token budgets and cost maxTokens 的推导（token 型思考 API） deriving maxTokens (token-budget thinking APIs) «interface» Usage + input · output · cacheRead · cacheWrite + cacheWrite1h? (Anthropic only) + reasoning? (subset of output) + totalTokens + cost: { input · output · cacheRead cacheWrite · total } (USD) ModelThinkingLevel（getSupportedThinkingLevels / clampThinkingLevel） ModelThinkingLevel (getSupportedThinkingLevels / clampThinkingLevel) off 不思考 no thinking minimal 1,024 tok low 2,048 tok medium 8,192 tok high 16,384 tok xhigh 需 thinkingLevelMap 声明 only if thinkingLevelMap has it max 需 thinkingLevelMap 声明 only if thinkingLevelMap has it 预算 = DEFAULT_THINKING_BUDGETS（可被 options.thinkingBudgets 覆盖）；不支持的等级被 clamp 到最近的可用等级 budget = DEFAULT_THINKING_BUDGETS (override with options.thinkingBudgets); unsupported levels clamp to the nearest available one 输入（估算） input (estimated) 安全余量 safety 思考预算 thinking budget 回答 ≥ 1024 answer ≥ 1024 空闲 free contextWindow clampMaxTokensToContext：maxTokens ≤ contextWindow − 估算输入 − 4096；思考预算 ≤ maxTokens − MIN_ANSWER_TOKENS clampMaxTokensToContext: maxTokens ≤ contextWindow − estimated input − 4096; thinking budget ≤ maxTokens − MIN_ANSWER_TOKENS models.ts calculateCost(model, usage): inputTokens = input + cacheRead + cacheWrite rates = highest tier with inputTokens > tier.inputTokensAbove cost.input = rates.input / 1e6 × input cost.cacheWrite = (rates.cacheWrite × short + rates.input × 2 × cacheWrite1h) / 1e6 cost.total = input + output + cacheRead + cacheWrite isContextOverflow(message, contextWindow)（utils/overflow.ts）识别各家“上下文过长”错误，供压缩重试 isContextOverflow(message, contextWindow) (utils/overflow.ts) recognises each provider’s context-overflow error for compact-and-retry provider-retry.ts：retryProviderRequest 对 429 / 5xx 重试，受 maxRetryDelayMs 约束 provider-retry.ts: retryProviderRequest retries 429 / 5xx, bounded by maxRetryDelayMs xhigh 与 max 只有在模型的 thinkingLevelMap 明确声明时才可用；null 表示该等级被禁用。xhigh and max exist only if the model's thinkingLevelMap declares them; null disables a level.
08 · 图像与分类器 08 · Images and classifiers pi-ai
图像与分类器 Images and classifiers One-shot APIs: images and classifiers generateImages(model, context, opts) → Promise<AssistantImages> classify(model, context, opts) → Promise<ClassifierResult> Models 不 reject：失败 = 结果里的 never reject: failure = stopReason "error" stopReason "error" in result openrouter provider openrouter-images ImageApi AssistantImages output[] · usage? · stopReason typesafe-system-one cloudflare-workers-ai-system-one llama-cpp-classify openai-decisions ImageModel（model.type = "image"）不出现在 getModels()，用 getAllModels() 查找 ImageModels (type "image") are not in getModels(); find them via getAllModels() 无 generateImages 的 provider → 返回 error 结果 providers without generateImages → error result typesafe · opencode · openrouter · vercel-ai-gateway cloudflare-workers-ai 无内置 provider · 通过 createProvider 注册 no built-in provider · register via createProvider openai ClassifierContext { state: JSON, images?（1.1.0）, questions: { [k]: choice | score | bool } } { state: JSON, images? (1.1.0), questions: { [k]: choice | score | bool } } ClassifierResult { answers: { choice·probabilities | score | probability }, usage?, stopReason } 分类器 API 有 4 种；1.1.0 新增 openai-decisions，以及 ClassifierContext.images。 There are four classifier APIs; 1.1.0 adds openai-decisions and ClassifierContext.images.
扩展点 Extension points pi-coding-agent
pi-ai 的扩展点 Where to plug into pi-ai 1
createProvider() createProvider()
注册自定义 provider：任意 baseUrl、自己的目录与认证。 Register a custom provider: any baseUrl, your own catalog and auth.
2
ProviderStreams ProviderStreams
实现 stream / streamSimple 即可接入一种新的线协议。 Implement stream / streamSimple to add a new wire protocol.
3
Model.compat Model.compat
为 OpenAI / Anthropic 兼容服务声明差异（字段名、能力开关）。 Declare quirks of OpenAI / Anthropic-compatible servers (field names, capability flags).
4
ApiKeyAuth · OAuth auth ApiKeyAuth · OAuth auth
自定义登录与凭据解析；CredentialStore 可替换为持久化存储。 Custom login and credential resolution; swap CredentialStore for persistent storage.
5
ModelsStore ModelsStore
持久化动态目录，启动时先发布缓存。 Persist dynamic catalogs and publish the cache at start-up.