创建日期:2026-09-07 | 最近更新:2026-09-07 事实核对基于
@earendil-works/pi-ai0.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-messages、openai-responses、openai-completions、google-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/anthropic、providers/openai…),只拉自己的目录与懒加载的 SDK wrapper → 配代码分割后,某家 SDK 直到第一次真正调它才加载。 - 老版本暴露的全局 API(
getModel()/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_KEY、OAuth还是 stored credential。 - 统一改头:
transformHeaders在鉴权与model.headers合并完之后、发给 provider 前跑一次,适合注入X-Request-ID之类。 - 按请求隔离:
env: { ... }选项可把某次请求的厂商配置(Cloudflare 账号、Azure 端点、代理等)限定到这一次,不让进程级 env 串味。
4. 工具:TypeBox schema + 边流边解析 + 显式校验
定义
工具参数用 TypeBox schema(pi-ai 重新导出了 Type、StringEnum):
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 用 helperStringEnum。
边流边解析(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 原话的转述):
event.partial.content[event.contentIndex]里才是正在流的那个 toolCall;- 参数可能不完整——字符串可能断在词中间、数组可能没流完,取值前必须判存在;保底是空对象
{},不会是undefined; toolcall_end里的event.toolCall.arguments才是完整参数(但还没过 schema 校验)。
执行前显式校验
自己写循环时,用 validateToolCall(tools, toolCall) 在真正执行前校验参数;抛错就把错误当 isError: true 的 toolResult 回填,让模型自己纠错重试——这是 agent 健壮性的标准姿势。
5. 事件模型:一整套细粒度事件
流式接口 models.stream(model, context) 吐出统一的事件流(非流式用 models.complete)。成功是 start → 各种 delta* → done,中途挂了是 start → updates → error:
| 事件 | 含义 | 关键字段 |
|---|---|---|
start | 流开始 | partial:当前消息骨架 |
text_start / text_delta / text_end | 正文开始 / 增量 / 结束 | delta、contentIndex |
thinking_start / thinking_delta / thinking_end | 思考内容流 | delta、contentIndex |
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_delta的partial是共享的实时对象(不是事件时刻的快照),别把它当历史状态存起来;要看某个时刻就用对应事件自己带上来的数据。
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 | 原样保留 |
跨厂商的 assistant | thinking 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 在聊天之外另起了一套 ImagesModels(builtinImagesModels() / 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天生可序列化,持久化、换手、续聊都是顺手的事。
关联
- 下一篇:pi-agent-core:Agent harness 的多轮循环与事件流——这一层封装了「模型提议 → 宿主执行 → 继续」的机械循环。
- pi 入门:家族地图、安装、5 分钟跑起第一个 Agent。
参考
- README:npmjs.com/package/@earendil-works/pi-ai(本文所有 API 以它为准)
- 事件轨迹、工具流示例均为本机
fauxProvider实测输出 - 许可:MIT