pi-ai@earendil-works/pi-ai

要点Key points

• 每个 API 都有一个 *.lazy.ts：lazyApi(() => import("./x.ts"))，provider 只引用这个包装。Every API has a *.lazy.ts: lazyApi(() => import("./x.ts")); providers only reference that wrapper.
• /compat 保留旧的全局 API（stream、complete、getModel），已弃用。/compat keeps the old global API (stream, complete, getModel) and is deprecated.
• bin: pi-ai 提供 CLI（OAuth 登录、模型列表）。bin: pi-ai ships a CLI (OAuth login, model listing).

工作原理How it works

1. 应用导入 builtinModels() 或若干 xxxProvider()。The app imports builtinModels() or a few xxxProvider() factories.
2. createProvider({ api: anthropicMessagesApi() }) 只保存懒包装。createProvider({ api: anthropicMessagesApi() }) stores only the lazy wrapper.
3. 第一次 stream 时 import() 实现模块（及其 SDK），之后由模块缓存去重。The first stream call import()s the implementation (and its SDK); the module cache dedupes later loads.

要点Key points

• 一个 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).
• 缺少对应 API 时 dispatch() 返回一个以 error 结束的流，而不是抛出。With no matching API, dispatch() returns a stream that ends in error instead of throwing.
• getModels() 只返回对话模型；图像与分类器模型通过 getAllModels() 查到。getModels() returns chat models only; image and classifier models are found via getAllModels().

工作原理How it works

1. createModels({ providers }) 构造 ModelsImpl，以 id 为键注册 provider。createModels({ providers }) builds a ModelsImpl keyed by provider id.
2. refresh() 对每个有 fetchModels 的 provider 调用 refreshModels(ctx)。refresh() calls refreshModels(ctx) on every provider with fetchModels.
3. 动态模型按 (type, id) 覆盖基线目录，getModels() 始终同步返回合并结果。Dynamic models override baseline entries by (type, id); getModels() always returns the merge synchronously.

要点Key points

• 认证解析顺序：调用方的 apiKey 覆盖 → 已存凭据（OAuth 会先刷新）→ 环境变量。Auth resolution order: caller apiKey override → stored credential (OAuth refreshed first) → environment.
• auth.baseUrl 会覆盖模型的 baseUrl（如企业网关）；transformHeaders 最后执行。auth.baseUrl overrides the model's baseUrl (e.g. enterprise gateways); transformHeaders runs last.
• forwardStream() 把内层事件逐个转发，最后用内层的 result() 结束外层流。forwardStream() forwards inner events one by one and ends the outer stream with the inner result().

工作原理How it works

1. normalizeContext(context) 同步生成 TranscriptContext。normalizeContext(context) builds a TranscriptContext synchronously.
2. lazyStream(model, setup) 立即返回外层 AssistantMessageEventStream。lazyStream(model, setup) immediately returns the outer AssistantMessageEventStream.
3. setup()：requireChatProvider → applyAuth → provider.streamSimple → dispatch → lazyApi 导入 → api.streamSimple。setup(): requireChatProvider → applyAuth → provider.streamSimple → dispatch → lazyApi import → api.streamSimple.
4. 成功则转发；失败则 push 一个 { type: "error" } 并 end(message)。On success, forward; on failure, push { type: "error" } and end(message).

要点Key points

• 每个 provider 一个文件，用 createProvider() 创建，并在 providers/all.ts 中汇总。One file per provider, created with createProvider() and collected in providers/all.ts.
• API 实现位于 api/，经 *.lazy.ts 中的 lazyApi() 包装：模块在第一次流式调用时才 import。API implementations live in api/, wrapped by lazyApi() in *.lazy.ts: a module is imported on its first stream call.
• KnownApi 列出 10 种对话 API（另有 1 种图像 API、4 种分类器 API）。KnownApi lists 10 chat APIs (plus 1 image API and 4 classifier APIs).
• 例：groq、deepseek 走 openai-completions；xai 走 openai-responses；openrouter 和 github-copilot 同时使用多种 API。e.g. groq and deepseek use openai-completions; xai uses openai-responses; openrouter and github-copilot use several APIs.

工作原理How it works

1. 每个 API 导出 stream 与 streamSimple，返回 AssistantMessageEventStream。Every API exports stream and streamSimple, returning an AssistantMessageEventStream.
2. 事件共 12 种：start、text_*、thinking_*、toolcall_*（各 start / delta / end）、done、error。There are 12 event types: start, text_*, thinking_*, toolcall_* (start / delta / end each), done, error.
3. StreamFn 约定不抛异常：认证、模块加载或请求失败都编码为流中的 error 事件（见 lazyStream()）。The StreamFn contract forbids throwing: auth, module-load and request failures become an error event in the stream (see lazyStream()).

要点Key points

• 7 种 StopReason：pending 只出现在 partial 上；deferred 表示异步任务，之后用 streamDeferred(handle) 取结果。Seven StopReasons: pending appears only on partials; deferred means an async job, fetched later with streamDeferred(handle).
• 1.1.0：AssistantMessageEventStream 自动为最终消息写入 durationMs（单调时钟，从流创建开始计时）。1.1.0: AssistantMessageEventStream stamps durationMs on the final message (monotonic clock from stream creation).
• 终止后 push 被忽略——迟到的事件不会污染已完成的消息。After termination, push is ignored — late events cannot corrupt a finished message.

工作原理How it works

1. API 实现创建 AssistantMessageEventStream，边解析 SSE 边 push。API implementations create an AssistantMessageEventStream and push while parsing SSE.
2. 有等待中的消费者就直接交付，否则进入 FIFO 队列。If a consumer is waiting, hand off directly; otherwise enqueue FIFO.
3. done / error 时 isComplete 为真，result() 得到最终 AssistantMessage。On done / error, isComplete is true and result() yields the final AssistantMessage.

要点Key points

• thinking 签名只在同一模型之间有效；跨模型时 thinking 转成普通文本，redacted thinking 直接丢弃。Thinking signatures are only valid on the same model; across models thinking becomes plain text and redacted thinking is dropped.
• 工具调用 id 的映射同时应用到对应的 toolResult，保证配对关系不被破坏。The tool-call id map is applied to matching toolResults too, so pairs stay intact.
• 没有结果的工具调用会得到合成的 isError 结果——多数 API 要求每个调用都有结果。Tool calls without a result get a synthesized isError result — most APIs require one per call.

工作原理How it works

1. 各 API 的 convertMessages() 先调用 transformMessages(messages, model, normalizeToolCallId)。Each API's convertMessages() first calls transformMessages(messages, model, normalizeToolCallId).
2. 然后才把统一消息映射成该 API 的请求格式。Only then are unified messages mapped to that API's request format.

要点Key points

• 工具参数用 parseStreamingJson 增量解析，UI 可以在参数流式到达时就显示。Tool arguments are parsed incrementally with parseStreamingJson, so a UI can show them while they stream.
• refusal 与 sensitive 被映射为 error，pause_turn 视为正常 stop。refusal and sensitive map to error; pause_turn is treated as a normal stop.
• 请求经过 retryProviderRequest，并提供 onPayload / onResponse 钩子用于调试与改写。Requests go through retryProviderRequest and expose onPayload / onResponse hooks for debugging and rewriting.

工作原理How it works

1. message_start → start；每个 content_block_start → *_start；delta → *_delta；stop → *_end。message_start → start; each content_block_start → *_start; deltas → *_delta; stop → *_end.
2. message_delta 带 stop_reason 与 usage；结束时 calculateCost() 并 push done。message_delta carries stop_reason and usage; at the end calculateCost() and push done.

要点Key points

• xhigh 与 max 只有在模型的 thinkingLevelMap 明确声明时才可用；null 表示该等级被禁用。xhigh and max exist only if the model's thinkingLevelMap declares them; null disables a level.
• 分层定价：输入 token（含缓存）超过某层阈值时整次请求按该层计价。Tiered pricing: when input tokens (cache included) exceed a tier threshold, the whole request uses that tier.
• isContextOverflow() 识别各家的“上下文过长”错误，coding-agent 据此压缩后重试。isContextOverflow() recognises each provider's context-overflow error; coding-agent compacts and retries on it.

工作原理How it works

1. clampThinkingLevel(model, level) 找到最近的受支持等级。clampThinkingLevel(model, level) finds the nearest supported level.
2. adjustMaxTokensForThinking + clampMaxTokensToContext 计算最终 maxTokens 与思考预算。adjustMaxTokensForThinking + clampMaxTokensToContext compute the final maxTokens and thinking budget.
3. 响应结束时 calculateCost(model, usage) 填充 usage.cost。At the end of a response calculateCost(model, usage) fills usage.cost.

要点Key points

• 分类器 API 有 4 种；1.1.0 新增 openai-decisions，以及 ClassifierContext.images。There are four classifier APIs; 1.1.0 adds openai-decisions and ClassifierContext.images.
• 问题有三种：choice（带概率分布）、score、bool，每个答案都带置信度或概率。Questions come in three kinds: choice (with a probability distribution), score and bool; each answer carries confidence or probability.

工作原理How it works

1. createProvider({ images, classifiers }) 按 API 名注册实现。createProvider({ images, classifiers }) registers implementations by API name.
2. models.classify(model, context) 走与流相同的认证路径，然后调用 provider。models.classify(model, context) follows the same auth path as streams, then calls the provider.