自定义模型提供方
扩展可以通过 pi.registerProvider() 注册自定义模型提供方。这可以实现:
- 代理 - 通过企业代理或 API 网关路由请求
- 自定义端点 - 使用自托管或私有模型部署
- OAuth/SSO - 为企业模型提供方添加身份验证流程
- 自定义 API - 为非标准 LLM API 实现流式传输
参见以下完整的模型提供方示例:
扩展可以注册一个完整的 pi-ai Provider,也可以使用旧的 provider-config(模型提供方配置)形式。当需要自定义身份验证、过滤、刷新或流式行为时,优先使用完整的 Provider。Pi 在已注册的原生模型提供方之上叠加 models.json 覆盖。
import { createProvider, openAICompletionsApi } from "@earendil-works/pi-ai";import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
export default function (pi: ExtensionAPI) { pi.registerProvider(createProvider({ id: "native-local", name: "Native Local", baseUrl: "http://localhost:8080/v1", auth: { apiKey: { name: "Local server API key", async login(interaction) { return { type: "api_key", key: await interaction.prompt({ type: "secret", message: "API key" }) }; }, async resolve({ credential }) { return credential?.key ? { auth: { apiKey: credential.key }, source: "stored API key" } : undefined; } } }, models: [], api: openAICompletionsApi() }));
// 旧的 provider-config 形式: // 覆盖现有模型提供方的 baseUrl pi.registerProvider("anthropic", { baseUrl: "https://proxy.example.com" });
// 注册带模型的新模型提供方 pi.registerProvider("my-provider", { name: "My Provider", baseUrl: "https://api.example.com", apiKey: "$MY_API_KEY", api: "openai-completions", models: [ { id: "my-model", name: "My Model", reasoning: false, input: ["text", "image"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 128000, maxTokens: 4096 } ] });}扩展工厂函数也可以是 async。对于动态模型发现,请在工厂函数中获取并注册模型,而不是在 session_start 中。Pi 会在启动继续之前等待工厂函数,因此该模型提供方在交互式启动时以及 pi --list-models 中都可用。
覆盖现有模型提供方
Section titled “覆盖现有模型提供方”最简单的用例:将现有模型提供方重定向到代理。
// 所有 Anthropic 请求现在都经过你的代理pi.registerProvider("anthropic", { baseUrl: "https://proxy.example.com"});
// 为 OpenAI 请求添加自定义请求头pi.registerProvider("openai", { headers: { "X-Custom-Header": "value" }});
// 同时设置 baseUrl 和请求头pi.registerProvider("google", { baseUrl: "https://ai-gateway.corp.com/google", headers: { "X-Corp-Auth": "$CORP_AUTH_TOKEN" // 环境变量或字面量 }});当只提供 baseUrl 和/或 headers(不提供 models)时,该模型提供方的所有现有模型都会保留,并使用新端点。
注册新模型提供方
Section titled “注册新模型提供方”要添加一个全新的模型提供方,请指定 models 以及必要的配置。
如果模型列表来自远程端点,请使用异步扩展工厂函数:
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
export default async function (pi: ExtensionAPI) { const response = await fetch("http://localhost:1234/v1/models"); const payload = (await response.json()) as { data: Array<{ id: string; name?: string; context_window?: number; max_tokens?: number; }>; };
pi.registerProvider("local-openai", { baseUrl: "http://localhost:1234/v1", apiKey: "$LOCAL_OPENAI_API_KEY", api: "openai-completions", models: payload.data.map((model) => ({ id: model.id, name: model.name ?? model.id, reasoning: false, input: ["text"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: model.context_window ?? 128000, maxTokens: model.max_tokens ?? 4096, })), });}这样会在启动完成之前注册获取到的模型。
pi.registerProvider("my-llm", { baseUrl: "https://api.my-llm.com/v1", apiKey: "$MY_LLM_API_KEY", // 环境变量引用 api: "openai-completions", // 使用哪个流式 API models: [ { id: "my-llm-large", name: "My LLM Large", reasoning: true, // 支持扩展思考 input: ["text", "image"], cost: { input: 3.0, // 美元/百万 token output: 15.0, cacheRead: 0.3, cacheWrite: 3.75 }, contextWindow: 200000, maxTokens: 16384 } ]});当提供 models 时,它会替换该模型提供方的所有现有模型。
apiKey 和自定义请求头值使用与 models.json 相同的配置值语法:开头的 !command 会为整个值执行一条命令,$ENV_VAR 和 ${ENV_VAR} 插值环境变量,$$ 输出字面量 $,$! 输出字面量 !。
注销模型提供方
Section titled “注销模型提供方”使用 pi.unregisterProvider(name) 移除之前通过 pi.registerProvider(name, ...) 注册的模型提供方:
// 注册pi.registerProvider("my-llm", { baseUrl: "https://api.my-llm.com/v1", apiKey: "$MY_LLM_API_KEY", api: "openai-completions", models: [ { id: "my-llm-large", name: "My LLM Large", reasoning: true, input: ["text", "image"], cost: { input: 3.0, output: 15.0, cacheRead: 0.3, cacheWrite: 3.75 }, contextWindow: 200000, maxTokens: 16384 } ]});
// 稍后移除pi.unregisterProvider("my-llm");注销会移除该模型提供方的动态模型、API 密钥回退、OAuth 模型提供方注册以及自定义流处理器注册。任何被覆盖的内置模型或模型提供方行为都会恢复。
在初始扩展加载阶段之后进行的调用会立即生效,因此无需 /reload。
API 类型
Section titled “API 类型”api 字段决定使用哪种流式实现:
| API | 用途 |
|---|---|
anthropic-messages |
Anthropic Claude API 及兼容实现 |
openai-completions |
OpenAI Chat Completions API 及兼容实现 |
openai-responses |
OpenAI Responses API |
azure-openai-responses |
Azure OpenAI Responses API |
openai-codex-responses |
OpenAI Codex Responses API |
mistral-conversations |
原生 Mistral Chat Completions 流式传输 |
google-generative-ai |
Google Generative AI API |
google-vertex |
Google Vertex AI API |
bedrock-converse-stream |
Amazon Bedrock Converse API |
大多数兼容 OpenAI 的模型提供方都可以使用 openai-completions。使用模型级别的 thinkingLevelMap 来设置特定模型的思考级别,使用 compat 来处理模型提供方的特殊行为。xhigh 和 max 级别是选择加入的,需要非 null 的映射条目,并且可能被不支持的空洞隔开:
models: [{ id: "custom-model", // ... reasoning: true, thinkingLevelMap: { // 将 pi 的级别映射到模型提供方值;null 隐藏不支持的级别 minimal: null, low: null, medium: null, high: "default", xhigh: null, max: "max" }, compat: { supportsDeveloperRole: false, // 使用 "system" 而不是 "developer" supportsReasoningEffort: true, maxTokensField: "max_tokens", // 而不是 "max_completion_tokens" requiresToolResultName: true, // 工具结果需要 name 字段 thinkingFormat: "qwen", // 顶层 enable_thinking: true cacheControlFormat: "anthropic" // Anthropic 风格的 cache_control 标记 }}]对于 OpenRouter 风格的 reasoning: { effort } 控制,使用 openrouter。对于 Together 风格的 reasoning: { enabled } 控制,使用 together;配合 supportsReasoningEffort 时,它还会发送 reasoning_effort。对于读取 chat_template_kwargs.enable_thinking 且需要 preserve_thinking 的本地 Qwen 兼容服务器,使用 qwen-chat-template。
当兼容 OpenAI 的模型提供方通过系统提示词、最后一个工具定义以及最后一条 user、assistant 或 tool-result 文本内容上的 cache_control 暴露 Anthropic 风格的提示词缓存时,使用 cacheControlFormat: "anthropic"。
对于使用 api: "anthropic-messages" 的 Anthropic 兼容模型提供方,如果上游模型需要自适应思考(thinking.type: "adaptive" 加上 output_config.effort),请在模型或模型提供方上设置 compat.forceAdaptiveThinking: true。内置的自适应 Claude 模型会自动设置此项。仅当模型提供方会发出空的思考签名并期望在重放时使用 signature: "",才设置 compat.allowEmptySignature: true。
迁移说明:Mistral 从
openai-completions迁移到了mistral-conversations。 对于原生 Mistral 模型,使用mistral-conversations。 如果你有意将 Mistral 兼容/自定义端点路由到openai-completions,请根据需要显式设置compat标志。
如果你的模型提供方期望 Authorization: Bearer <key>,但不使用标准 API,请设置 authHeader: true:
pi.registerProvider("custom-api", { baseUrl: "https://api.example.com", apiKey: "$MY_API_KEY", authHeader: true, // 添加 Authorization: Bearer 请求头 api: "openai-completions", models: [...]});密钥会在每个请求时解析。显式设置的请求 Authorization 请求头优先于生成的值。
OAuth 支持
Section titled “OAuth 支持”添加与 /login 集成的 OAuth/SSO 身份验证:
import type { OAuthCredentials, OAuthLoginCallbacks } from "@earendil-works/pi-ai";
pi.registerProvider("corporate-ai", { baseUrl: "https://ai.corp.com/v1", api: "openai-responses", models: [...], oauth: { name: "Corporate AI (SSO)",
async login(callbacks: OAuthLoginCallbacks): Promise<OAuthCredentials> { const method = await callbacks.onSelect({ message: "Select login method:", options: [ { id: "browser", label: "Browser OAuth" }, { id: "device", label: "Device code" } ] }); if (!method) throw new Error("Login cancelled");
let code: string; if (method === "device") { callbacks.onDeviceCode({ userCode: "ABCD-1234", verificationUri: "https://sso.corp.com/device", intervalSeconds: 5, expiresInSeconds: 900 }); code = await pollDeviceCodeUntilComplete(); } else { callbacks.onAuth({ url: "https://sso.corp.com/authorize?..." }); code = await callbacks.onPrompt({ message: "Enter SSO code:" }); }
// 交换令牌(你的实现) const tokens = await exchangeCodeForTokens(code);
return { refresh: tokens.refreshToken, access: tokens.accessToken, expires: Date.now() + tokens.expiresIn * 1000 }; },
async refreshToken(credentials: OAuthCredentials, signal: AbortSignal): Promise<OAuthCredentials> { const tokens = await refreshAccessToken(credentials.refresh, signal); return { refresh: tokens.refreshToken ?? credentials.refresh, access: tokens.accessToken, expires: Date.now() + tokens.expiresIn * 1000 }; },
getApiKey(credentials: OAuthCredentials): string { return credentials.access; } }});注册后,用户可以通过 /login corporate-ai 进行身份验证。
OAuthLoginCallbacks
Section titled “OAuthLoginCallbacks”callbacks 对象为模型提供方拥有的流程提供与 UI 无关的交互:
interface OAuthLoginCallbacks { // 在浏览器中打开 URL(用于 OAuth 重定向) onAuth(params: { url: string }): void;
// 显示设备代码(用于设备授权流程) onDeviceCode(params: { userCode: string; verificationUri: string; intervalSeconds?: number; expiresInSeconds?: number; }): void;
// 显示临时进度 onProgress?(message: string): void;
// 提示用户输入(用于手动输入令牌) onPrompt(params: { message: string }): Promise<string>;
// 显示交互式选择器,例如选择浏览器 OAuth 或设备代码 onSelect(params: { message: string; options: { id: string; label: string }[]; }): Promise<string | undefined>;}OAuthCredentials
Section titled “OAuthCredentials”凭据持久化在 ~/.pi/agent/auth.json 中:
interface OAuthCredentials { refresh: string; // 刷新令牌(用于 refreshToken()) access: string; // 访问令牌(由 getApiKey() 返回) expires: number; // 过期时间戳(毫秒)}自定义流式 API
Section titled “自定义流式 API”对于 API 不标准的模型提供方,请实现 streamSimple。在编写自己的实现之前,先研究现有的模型提供方实现:
参考实现:
- anthropic.ts - Anthropic Messages API
- mistral.ts - Mistral Conversations API
- openai-completions.ts - OpenAI Chat Completions
- openai-responses.ts - OpenAI Responses API
- google.ts - Google Generative AI
- amazon-bedrock.ts - AWS Bedrock
所有模型提供方都遵循相同的模式:
import { type AssistantMessage, type AssistantMessageEventStream, type Context, type Model, type SimpleStreamOptions, calculateCost, createAssistantMessageEventStream,} from "@earendil-works/pi-ai";
function streamMyProvider( model: Model<any>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream { const stream = createAssistantMessageEventStream();
(async () => { // 初始化输出消息 const output: AssistantMessage = { role: "assistant", content: [], api: model.api, provider: model.provider, model: model.id, usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "pending", timestamp: Date.now(), };
try { // 推送 start 事件 stream.push({ type: "start", partial: output });
// 发出 API 请求并处理响应... // 随着内容到达推送内容事件,并从终止事件设置 stopReason。 if (output.stopReason === "pending") { throw new Error("Provider stream ended without a stop reason"); } if (output.stopReason === "error" || output.stopReason === "aborted") { throw new Error(output.errorMessage || "An unknown error occurred"); }
// 推送 done 事件 stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { output.stopReason = options?.signal?.aborted ? "aborted" : "error"; output.errorMessage = error instanceof Error ? error.message : String(error); stream.push({ type: "error", reason: output.stopReason, error: output }); stream.end(); } })();
return stream;}按以下顺序通过 stream.push() 推送事件:
-
{ type: "start", partial: output }- 流已开始 -
内容事件(可重复,为每个块跟踪
contentIndex):{ type: "text_start", contentIndex, partial }- 文本块开始{ type: "text_delta", contentIndex, delta, partial }- 文本片段{ type: "text_end", contentIndex, content, partial }- 文本块结束{ type: "thinking_start", contentIndex, partial }- 思考开始{ type: "thinking_delta", contentIndex, delta, partial }- 思考片段{ type: "thinking_end", contentIndex, content, partial }- 思考结束{ type: "toolcall_start", contentIndex, partial }- 工具调用开始{ type: "toolcall_delta", contentIndex, delta, partial }- 工具调用 JSON 片段{ type: "toolcall_end", contentIndex, toolCall, partial }- 工具调用结束
-
{ type: "done", reason, message }或{ type: "error", reason, error }- 流已结束
每个事件中的 partial 字段包含当前的 AssistantMessage 状态。收到数据时更新 output.content,然后将 output 作为 partial 传入。
随着内容块的到达,将其添加到 output.content:
// 文本块output.content.push({ type: "text", text: "" });stream.push({ type: "text_start", contentIndex: output.content.length - 1, partial: output });
// 文本到达时const block = output.content[contentIndex];if (block.type === "text") { block.text += delta; stream.push({ type: "text_delta", contentIndex, delta, partial: output });}
// 块完成时stream.push({ type: "text_end", contentIndex, content: block.text, partial: output });工具调用需要累积 JSON 并解析:
// 开始工具调用output.content.push({ type: "toolCall", id: toolCallId, name: toolName, arguments: {}});stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
// 累积 JSONlet partialJson = "";partialJson += jsonDelta;try { block.arguments = JSON.parse(partialJson);} catch {}stream.push({ type: "toolcall_delta", contentIndex, delta: jsonDelta, partial: output });
// 完成stream.push({ type: "toolcall_end", contentIndex, toolCall: { type: "toolCall", id, name, arguments: block.arguments }, partial: output});从 API 响应更新用量并计算成本:
output.usage.input = response.usage.input_tokens;output.usage.output = response.usage.output_tokens;output.usage.cacheRead = response.usage.cache_read_tokens ?? 0;output.usage.cacheWrite = response.usage.cache_write_tokens ?? 0;output.usage.totalTokens = output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;calculateCost(model, output.usage);上下文溢出错误
Section titled “上下文溢出错误”当请求超过模型的上下文窗口时,pi 可以通过压缩对话并重试来自动恢复。只有在 pi 识别出该失败为溢出时,这种恢复才会触发。
检测在最终确定的助手消息上运行:
stopReason === "error"errorMessage匹配 pi 已知的溢出模式之一(参见packages/ai/src/utils/overflow.ts)
如果模型提供方返回的溢出错误带有一条 pi 无法识别的消息,请在与注册该模型提供方相同的扩展中规范化该错误。使用 message_end 处理器重写助手消息,使其 errorMessage 以 pi 能识别的短语开头。通用的回退 context_length_exceeded 是最安全的选择。
const MY_PROVIDER_OVERFLOW_PATTERN = /your provider's overflow phrase/i;
export default function (pi: ExtensionAPI) { pi.registerProvider("my-provider", { /* ... */ });
pi.on("message_end", (event, ctx) => { const message = event.message; if (message.role !== "assistant") return; if (message.stopReason !== "error") return; if ( message.provider !== "my-provider" && ctx.model?.provider !== "my-provider" ) return;
const errorMessage = message.errorMessage ?? ""; if (errorMessage.includes("context_length_exceeded")) return; if (!MY_PROVIDER_OVERFLOW_PATTERN.test(errorMessage)) return;
return { message: { ...message, errorMessage: `context_length_exceeded: ${errorMessage}`, }, }; });}message_end 在 pi 跟踪助手消息以进行自动压缩之前运行,因此重写后的 errorMessage 就是 pi 检查的内容。有了这些,pi 将:
- 从
errorMessage检测溢出。 - 从实时上下文中丢弃失败的助手消息。
- 运行压缩。
- 重试请求一次。
请谨慎守卫重写:
- 将其限定在你的模型提供方(
message.provider和ctx.model?.provider)内,这样其他模型提供方无关的错误不会被改动。 - 匹配特定于模型提供方的模式,而不是 pi 的通用溢出模式。重写限流或节流错误(
rate limit、too many requests)会错误地触发压缩,而不是走 pi 正常的带退避的重试路径。 - 当
errorMessage已包含context_length_exceeded时跳过,使处理器保持幂等。
注册你的流函数:
pi.registerProvider("my-provider", { baseUrl: "https://api.example.com", apiKey: "$MY_API_KEY", api: "my-custom-api", models: [...], streamSimple: streamMyProvider});测试你的实现
Section titled “测试你的实现”针对内置模型提供方使用的同一套测试套件来测试你的模型提供方。复制并调整 packages/ai/test/ 中的这些测试文件:
| 测试 | 目的 |
|---|---|
stream.test.ts |
基本流式、文本输出 |
tokens.test.ts |
token 计数和用量 |
abort.test.ts |
AbortSignal 处理 |
empty.test.ts |
空/最小响应 |
context-overflow.test.ts |
上下文窗口限制 |
image-limits.test.ts |
图像输入处理 |
unicode-surrogate.test.ts |
Unicode 边界情况 |
tool-call-without-result.test.ts |
工具调用边界情况 |
image-tool-result.test.ts |
工具结果中的图像 |
total-tokens.test.ts |
总 token 计算 |
cross-provider-handoff.test.ts |
模型提供方之间的上下文交接 |
使用你的模型提供方/模型组合运行测试以验证兼容性。
interface ProviderConfig { /** 在 /login 等 UI 中显示的模型提供方名称。 */ name?: string;
/** API 端点 URL。定义模型时为必填。 */ baseUrl?: string;
/** API 密钥字面量、环境变量插值($ENV_VAR 或 ${ENV_VAR})或 !command。定义模型时为必填(除非使用 oauth)。 */ apiKey?: string;
/** 用于流式的 API 类型。定义模型时,在模型提供方或模型级别为必填。 */ api?: Api;
/** 用于非标准 API 的自定义流式实现。 */ streamSimple?: ( model: Model<Api>, context: Context, options?: SimpleStreamOptions ) => AssistantMessageEventStream;
/** 要包含在请求中的自定义请求头。值与 apiKey 使用相同的解析语法。 */ headers?: Record<string, string>;
/** 如果为 true,则使用解析后的 API 密钥添加 Authorization: Bearer 请求头。 */ authHeader?: boolean;
/** 要注册的模型。如果提供,则替换该模型提供方的所有现有模型。 */ models?: ProviderModelConfig[];
/** 用于 /login 支持的 OAuth 模型提供方。 */ oauth?: { name: string; login(callbacks: OAuthLoginCallbacks): Promise<OAuthCredentials>; refreshToken(credentials: OAuthCredentials, signal: AbortSignal): Promise<OAuthCredentials>; getApiKey(credentials: OAuthCredentials): string; };}模型定义参考
Section titled “模型定义参考”interface ProviderModelConfig { /** 模型 ID(例如 "claude-sonnet-4-20250514")。 */ id: string;
/** 显示名称(例如 "Claude 4 Sonnet")。 */ name: string;
/** 该特定模型的 API 类型覆盖。 */ api?: Api;
/** 该特定模型的 API 端点 URL 覆盖。 */ baseUrl?: string;
/** 模型是否支持扩展思考。 */ reasoning: boolean;
/** 将 pi 的思考级别映射到模型提供方/模型特定的值;null 表示不支持该级别。 */ thinkingLevelMap?: Partial<Record<"off" | "minimal" | "low" | "medium" | "high" | "xhigh" | "max", string | null>>;
/** 支持的输入类型。 */ input: ("text" | "image")[];
/** 每百万 token 的成本(用于用量跟踪)。 */ cost: { input: number; output: number; cacheRead: number; cacheWrite: number; };
/** 最大上下文窗口大小(token 数)。 */ contextWindow: number;
/** 最大输出 token 数。 */ maxTokens: number;
/** 该特定模型的自定义请求头。 */ headers?: Record<string, string>;
/** 所选 API 的兼容性设置。 */ compat?: { // openai-completions supportsStore?: boolean; supportsDeveloperRole?: boolean; supportsReasoningEffort?: boolean; supportsUsageInStreaming?: boolean; supportsFinishReason?: boolean; supportsStrictMode?: boolean; supportsOpenAIGrammarTools?: boolean; // openai-completions/openai-responses;false 时回退到普通函数工具 maxTokensField?: "max_completion_tokens" | "max_tokens"; requiresToolResultName?: boolean; requiresAssistantAfterToolResult?: boolean; requiresThinkingAsText?: boolean; requiresReasoningContentOnAssistantMessages?: boolean; thinkingFormat?: "openai" | "openrouter" | "deepseek" | "together" | "baseten" | "zai" | "qwen" | "chat-template" | "qwen-chat-template" | "string-thinking" | "ant-ling"; chatTemplateKwargs?: Record<string, string | number | boolean | null | { "$var": "thinking.enabled" | "thinking.effort"; omitWhenOff?: boolean }>; chatTemplateArgs?: Record<string, string | number | boolean | null | { "$var": "thinking.enabled" | "thinking.effort"; omitWhenOff?: boolean }>; cacheControlFormat?: "anthropic"; sessionAffinityFormat?: "openai" | "openai-nosession" | "openrouter"; sendSessionAffinityHeaders?: boolean;
// anthropic-messages supportsEagerToolInputStreaming?: boolean; supportsLongCacheRetention?: boolean; sendSessionAffinityHeaders?: boolean; supportsCacheControlOnTools?: boolean; forceAdaptiveThinking?: boolean; allowEmptySignature?: boolean; supportsStrictTools?: boolean; };}openrouter 发送 reasoning: { effort }。deepseek 发送 thinking: { type: "enabled" | "disabled" },并在启用时发送 reasoning_effort。together 发送 reasoning: { enabled },并在 supportsReasoningEffort 启用时也发送 reasoning_effort。qwen 用于 DashScope 风格的顶层 enable_thinking。对于读取 chat_template_kwargs.enable_thinking 且需要 preserve_thinking 的本地 Qwen 兼容服务器,使用 qwen-chat-template。对于可配置的 chat_template_kwargs 使用 chat-template,例如 vLLM 后的 DeepSeek V3.x,配合 chatTemplateKwargs: { "thinking": { "$var": "thinking.enabled" } }。当模型提供方期望在 chat_template_args 下使用切换值,并可选择支持顶层 reasoning_effort 时,将 thinkingFormat: "baseten" 与 chatTemplateArgs 一起使用。
cacheControlFormat: "anthropic" 会将 Anthropic 风格的 cache_control 标记应用到系统提示词、最后一个工具定义以及最后一条 user、assistant 或 tool-result 文本内容上。