跳到主要内容

创建日期:2026-09-07 | 最近更新:2026-09-07 事实核对基于 @earendil-works/pi-ai 0.85.1(2026-09-05 发布)的 README、源码与本机实测;示例无需 API Key(用 fauxProvider)。

pi-ai:统一几十家 LLM 的模型连接层

一句话:pi-ai 是 pi 的地基——把「跟大模型说话」这件事里所有跟厂商相关的脏活(模型目录、鉴权、流式、工具调用、thinking、换手)收敛成一个干净接口。它不管「要不要再调一轮工具」,那是 agent-core(篇 2)的事。

1. 它替你解决了什么

直接写官方 SDK 时,你很快就会撞上这些墙:

  • 每家模型目录、定价、上下文长度都不同,还经常变;
  • OpenAI 的流式是 chunk,Anthropic 是 SSE,Gemini 又不一样,事件形状各不相同;
  • 工具调用参数的 schema、格式各有各的脾气;
  • thinking/reasoning 的开关和参数三家各写各的;
  • 「把 A 家模型的多轮上下文搬去 B 家」基本要靠手搓转换。

pi-ai 的做法是把它们全部抹平成一套厂商无关的模型(Model)+ 上下文(Context)+ 事件(event)模型。它的取舍很鲜明:

  • 目录是数据,不是代码。 内置各厂商模型目录由脚本从 models.dev 抓取生成为静态数据,模型对象只是可 JSON 序列化的普通对象。所以「查询有哪些模型」「某个模型支不支持图片/推理」「这个模型多少钱」都是同步、离线、类型完整的。
  • 只收录支持工具调用的模型。 因为 agent 工作流里工具是刚需——这决定了 pi-ai 不是「通用聊天 SDK」,而是「为 agent 而生」的 SDK。
  • 鉴权归 provider 管。 Key 从哪来(env / 存好的凭据 / OAuth)是每个 provider 自己的事,调模型的人不用关心。

2. 三个核心对象:provider / model / Models

对象是什么关键点
provider一个厂商的「运行时单元」拥有自己的模型目录、鉴权(env/OAuth/凭据)、流式行为;内部共享几种 wire 协议实现(anthropic-messagesopenai-responsesopenai-completionsgoogle-generative-ai…)
model一个具体的模型,纯数据字段如 id / name / api / provider / baseUrl / reasoning / input[] / output[] / cost{} / contextWindow / maxTokens;可序列化、可随便换
Models(collection)装 provider 的容器models.getModel(providerId, modelId) 同步查询;models.stream/complete(…) 把请求路由到拥有该模型的 provider;models.refresh() 刷新动态模型列表

最常用的三个查询:

import { createModels } from '@earendil-works/pi-ai';
import { anthropicProvider } from '@earendil-works/pi-ai/providers/anthropic';
import { openaiProvider } from '@earendil-works/pi-ai/providers/openai';

const models = createModels();
models.setProvider(anthropicProvider());
models.setProvider(openaiProvider());

const m = models.getModel('anthropic', 'claude-sonnet-4-6');
console.log(m.contextWindow, m.input.includes('image'), m.reasoning); // 上下文 / 是否支持图片 / 是否支持推理

「调哪个模型」是一个运行时值(甚至可以来自配置文件/用户输入),代码不做任何字符串分支——这是整层设计最重要的气质。

从哪引入:树摇与静态目录

import { builtinModels } from '@earendil-works/pi-ai/providers/all'; // 全部内置 provider,图省事
import { getBuiltinModel } from '@earendil-works/pi-ai/providers/all'; // 纯静态目录查询,类型自动补全

import { createModels } from '@earendil-works/pi-ai'; // 空 collection,手动 setProvider
  • 每个厂商一个子路径入口providers/anthropicproviders/openai…),只拉自己的目录与懒加载的 SDK wrapper → 配代码分割后,某家 SDK 直到第一次真正调它才加载。
  • 老版本暴露的全局 APIgetModel() / stream() / registerApiProvider()…)原样搬到了 @earendil-works/pi-ai/compat,新代码别用它,按上面的 collection 写法迁移即可。

3. 鉴权:谁提供、从哪来、谁能覆盖

pi-ai 的哲学:你(应用)不跟 Key 打交道,Key 解析是 provider 内部的事。

调用 models.stream/complete 时,鉴权由拥有该模型的 provider 解析并合并进请求。解析优先级从低到高大致是:

默认值 → provider 内部默认(env)→ 存好的凭据(CredentialStore)→ 你显式传的 apiKey
  • env 变量:每家一组,如 OPENAI_API_KEY / ANTHROPIC_API_KEY / GEMINI_API_KEY / DEEPSEEK_API_KEY / MOONSHOT_API_KEY / MINIMAX_API_KEY / GROQ_API_KEY / XAI_API_KEY / OPENROUTER_API_KEY… 完整表见 README。
  • 存好的凭据createModels({ credentials: myStore }) 注入一个 CredentialStore(内置内存实现,可换成文件/DB 实现),契约就四个操作:read / list / modify / delete存了凭据就「拥有」该 provider——env 只在没存凭据时才被看,OAuth token 过期也不静默回退 env。
  • OAuth 型 provider:Anthropic(Claude 订阅)、OpenAI Codex(ChatGPT 订阅)、GitHub Copilot、OpenRouter 走 models.login('anthropic', 'oauth', { prompt, notify }),之后自动刷新。CLI 一行完成:npx @earendil-works/pi-ai login(结果存当前目录 auth.json)。
  • 显式覆盖:任何请求都能直接给 apiKey,它赢过一切:
await models.complete(model, context, { apiKey: 'sk-explicit' });
  • 排查用:不发请求也能看解析结果:await models.getAuth(model) → 返回 { auth, source }source 会告诉你是 ANTHROPIC_API_KEYOAuth 还是 stored credential。
  • 统一改头transformHeaders 在鉴权与 model.headers 合并完之后、发给 provider 前跑一次,适合注入 X-Request-ID 之类。
  • 按请求隔离env: { ... } 选项可把某次请求的厂商配置(Cloudflare 账号、Azure 端点、代理等)限定到这一次,不让进程级 env 串味。

4. 工具:TypeBox schema + 边流边解析 + 显式校验

定义

工具参数用 TypeBox schema(pi-ai 重新导出了 TypeStringEnum):

import { Type, StringEnum } from '@earendil-works/pi-ai';

const weatherTool = {
name: 'get_weather',
description: '查询某城市天气',
parameters: Type.Object({
location: Type.String({ description: '城市名' }),
units: StringEnum(['celsius', 'fahrenheit'], { default: 'celsius' }), // 兼容 Google:别用 Type.Enum
}),
};

:TypeBox 的 Type.Enum 会生成 anyOf/const 模式,Google 不支持;给 Google 兼容的 API 用 helper StringEnum

边流边解析(agent UI 的利器)

模型在流式生成工具参数时,pi-ai 用 partial-json 对参数做增量解析——参数还没流完,你就能拿到「目前解析出来的部分」。这让 UI 可以实时显示「正在写入 /path/foo…」,不用等完整 JSON。

实测的流式轨迹(fauxProvider,本机跑):

[start]
[thinking_start]
[thinking_delta] "先想一想用什么工具。"
[thinking_end]
[toolcall_start] contentIndex=1
[toolcall_delta] name=get_weather argsSoFar={"location":"Shanghai"}
[toolcall_delta] name=get_weather argsSoFar={"location":"Shanghai"}
[toolcall_end] get_weather({"location":"Shanghai"})
[done] reason=toolUse

处理 toolcall_delta 时三条铁律(README 原话的转述):

  1. event.partial.content[event.contentIndex] 里才是正在流的那个 toolCall;
  2. 参数可能不完整——字符串可能断在词中间、数组可能没流完,取值前必须判存在;保底是空对象 {},不会是 undefined
  3. toolcall_end 里的 event.toolCall.arguments 才是完整参数(但还没过 schema 校验)。

执行前显式校验

自己写循环时,用 validateToolCall(tools, toolCall)真正执行前校验参数;抛错就把错误当 isError: truetoolResult 回填,让模型自己纠错重试——这是 agent 健壮性的标准姿势。

5. 事件模型:一整套细粒度事件

流式接口 models.stream(model, context) 吐出统一的事件流(非流式用 models.complete)。成功是 start → 各种 delta* → done,中途挂了是 start → updates → error

事件含义关键字段
start流开始partial:当前消息骨架
text_start / text_delta / text_end正文开始 / 增量 / 结束deltacontentIndex
thinking_start / thinking_delta / thinking_end思考内容流deltacontentIndex
toolcall_start / toolcall_delta / toolcall_end工具调用:开始 / 参数增量流 / 结束toolCall(完整但未校验)、contentIndex
done结束reason: stop / length / toolUse
error出错reason: error / aborted

两个坑要提前知道:

  • 各 block 事件不是连续的。同一 chunk 里可能既有文本又有 thinking 又有工具增量,pi 会交错发事件(text_start, text_delta, toolcall_start, text_delta, toolcall_delta…)。所以永远用 contentIndex 把 delta/end 挂回它所属的 block,别假设一个 block 的 start→delta→end 中间不被别的 block 打断。
  • toolcall_deltapartial共享的实时对象(不是事件时刻的快照),别把它当历史状态存起来;要看某个时刻就用对应事件自己带上来的数据。

6. thinking / reasoning:一套参数,三家厂商

pi-ai 提供两套接口:

  • 简化接口 models.streamSimple/completeSimple(model, context, { reasoning }) —— 把三家差异收敛成一个枚举 'minimal' | 'low' | 'medium' | 'high' | 'xhigh' | 'max',跨厂商一致(xhigh/max 是否可用要查 getSupportedThinkingLevels(model))。
  • 完整接口 models.stream/complete —— 用 hasApi() 把模型收窄到具体协议后,能拿到该 API 的全部原生选项:
import { hasApi } from '@earendil-works/pi-ai';

const m = models.getModel('openai', 'gpt-5-mini');
if (hasApi(m, 'openai-responses')) {
await models.complete(m, context, { reasoningEffort: 'medium' });
}
// anthropic-messages → { thinkingEnabled: true, thinkingBudgetTokens: 8192 }
// google-generative-ai → { thinking: { enabled: true, budgetTokens: 8192 } }

模型元数据里有 reasoning 字段可查是否支持推理;把推理选项传给不支持的模型会被静默忽略,不会炸。

7. 跨厂商换手 & 上下文序列化(这个设计很值钱)

同一套 Context 可以在不同厂商的模型间搬来搬去,库自动做兼容转换(README 规则):

消息类型换手时的处理
user / toolResult原样通过
同厂商/同协议的 assistant原样保留
跨厂商的 assistantthinking block 转成带 <thinking> 标签的文本
toolCall / 普通文本原样保留

典型玩法:先用便宜的模型打草稿,中途切成更强的模型做精修;或者 A 厂商挂了直接切 B 家续聊,上下文不断。

Context 本身就是普通 JSON(系统提示、消息数组、工具定义都序列化得动);Model 也是纯数据。整段对话 JSON.stringify 存库、JSON.parse 恢复、换模型继续——三步就实现了「断点续聊」。

8. 错误处理:流内不抛,aborted 能续聊

  • models.stream() 一旦把流交给你,请求失败不再 throw,而是发 error 事件、最终消息带上 stopReason: 'error'errorMessage
  • abort:传 signal,中断后 stopReason === 'aborted'usage 会带部分 token。把这条 aborted 的 assistant 消息 push 回上下文,再补一句「请继续」就能续上——不用重开整个对话;
  • 例外:streamSimple 这类直接打 API 的调用在缺鉴权时会同步 throw(collection 走 provider 鉴权则不会)。

9. 无 Key 开发:fauxProvider(强烈建议用起来)

import { createModels, fauxProvider, fauxAssistantMessage, fauxThinking, fauxToolCall, fauxText } from '@earendil-works/pi-ai';

const faux = fauxProvider(); // 响应按队列消费,不联网
const models = createModels();
models.setProvider(faux.provider);
const model = faux.getModel();

faux.setResponses([
fauxAssistantMessage([fauxThinking('想想再答。'), fauxToolCall('get_weather', { location: 'Shanghai' })], { stopReason: 'toolUse' }),
]);
  • fauxAssistantMessage + fauxText / fauxThinking / fauxToolCall 拼脚本化响应;
  • 队列空了会返回一条 errorMessage: "No more faux responses queued" 的 assistant error——正好用来测你的「模型乱来」兜底;
  • sessionId + cacheRetention 时连 prompt cache 读写都会模拟;
  • 想测「不同模型切换」,fauxProvider({ models: [/* 多个 id */] }) 造多模型假 provider。

写 agent 代码时先用它把逻辑钉死,再接真厂商——这套习惯能省掉大量调试 SDK 的时间。

10. 图像:也是公民

0.85 在聊天之外另起了一套 ImagesModelsbuiltinImagesModels() / imagesModels.generateImages(model, { input })),支持图像输入({ type: 'image', data, mimeType })与生成(目前生成端主要是 OpenRouter 的 Gemini 图像模型)。别把图像生成塞进 chat/stream API——它是一次性接口,失败不 reject,而是返回 stopReason: 'error' 的结果对象。

11. 在 pi-ai 上写 agent 的取舍小结

  • 别自己维护「哪家模型叫什么、支不支持工具」——getModels() / getModel() 的目录就是答案,还带类型补全;
  • 工具 schema 用 TypeBox,别用 Type.Enum(Google);执行前 validateToolCall,失败回填 isError: true 让模型自己救;
  • 认事件不认字符串:拿 contentIndex 对齐 block,别假设事件连续;
  • completeSimple({ reasoning }) 就别手搓各家 thinking 参数,需要厂商特有能力再用 hasApi() 收窄;
  • 代码先对着 fauxProvider 写,再接真 Key;
  • Context / Model 天生可序列化,持久化、换手、续聊都是顺手的事。

关联

参考