自定义提供商
扩展可以通过 pi.registerProvider() 注册自定义模型提供商。这支持:
- Proxies 代理 — 通过公司代理或 API 网关路由请求
- Custom endpoints 自定义端点 — 使用自托管或私有模型部署
- OAuth/SSO — 为企业提供商添加认证流程
- Custom APIs 自定义 API — 为非标准 LLM API 实现流式处理
Example Extensions 示例扩展
查看以下完整提供商示例:
Table of Contents 目录
- Example Extensions 示例扩展
- Quick Reference 快速参考
- Override Existing Provider 覆盖现有提供商
- Register New Provider 注册新提供商
- Unregister Provider 注销提供商
- OAuth Support OAuth 支持
- Custom Streaming API 自定义流式 API
- Context Overflow Errors 上下文溢出错误
- Testing Your Implementation 测试你的实现
- Config Reference 配置参考
- Model Definition Reference 模型定义参考
Quick Reference 快速参考
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
export default function (pi: ExtensionAPI) {
// 覆盖现有提供商的 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 命令中均可用。
Override Existing Provider 覆盖现有提供商
最简单的使用场景:通过代理重定向现有提供商。
// 所有 Anthropic 请求现在通过你的代理
pi.registerProvider("anthropic", {
baseUrl: "https://proxy.example.com"
});
// 为 OpenAI 请求添加自定义请求头
pi.registerProvider("openai", {
headers: {
"X-Custom-Header": "value"
}
});
// 同时设置 baseUrl 和 headers
pi.registerProvider("google", {
baseUrl: "https://ai-gateway.corp.com/google",
headers: {
"X-Corp-Auth": "$CORP_AUTH_TOKEN" // 环境变量或字面值
}
});当仅提供 baseUrl 和/或 headers(不提供 models)时,该提供商的所有现有模型将保留并使用新的端点。
Register New Provider 注册新提供商
要添加一个全新的提供商,需在所需配置的同时指定 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} 会插值环境变量;$$ 会输出字面量 $;$! 会输出字面量 !。
Unregister Provider 注销提供商
使用 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 key 回退、OAuth 提供商注册以及自定义流处理器注册。任何被覆盖的内置模型或提供商行为都会被恢复。
在初始扩展加载阶段之后进行的调用会立即生效,因此无需执行 /reload。
API Types 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 SDK Conversations/Chat 流式处理 |
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 级别是选用的,需要非空的映射条目,且可能被不支持的级别隔开:
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, // tool result 需要 name 字段
thinkingFormat: "qwen", // 顶层 enable_thinking: true
cacheControlFormat: "anthropic" // Anthropic 风格的 cache_control 标记
}
}]使用 openrouter 处理 OpenRouter 风格的 reasoning: { effort } 控制。使用 together 处理 Together 风格的 reasoning: { enabled } 控制;启用 supportsReasoningEffort 时,还会发送 reasoning_effort。使用 qwen-chat-template 用于读取 chat_template_kwargs.enable_thinking 且需要 preserve_thinking 的本地 Qwen 兼容服务器。
使用 cacheControlFormat: "anthropic" 用于那些通过系统提示词、最后一个工具定义和最后一个用户/助手文本内容上的 cache_control 暴露 Anthropic 风格提示缓存的 OpenAI 兼容提供商。
对于使用 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标志。
Auth Header 认证请求头
如果你的提供商期望 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: [...]
});每次请求时都会解析该 key。显式的请求 Authorization 请求头优先级高于生成的值。
OAuth Support 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: "选择登录方式:",
options: [
{ id: "browser", label: "浏览器 OAuth" },
{ id: "device", label: "设备码" }
]
});
if (!method) throw new Error("登录已取消");
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: "请输入 SSO 码:" });
}
// 兑换令牌(你的实现)
const tokens = await exchangeCodeForTokens(code);
return {
refresh: tokens.refreshToken,
access: tokens.accessToken,
expires: Date.now() + tokens.expiresIn * 1000
};
},
async refreshToken(credentials: OAuthCredentials): Promise<OAuthCredentials> {
const tokens = await refreshAccessToken(credentials.refresh);
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
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
凭据持久化存储在 ~/.pi/agent/auth.json 中:
interface OAuthCredentials {
refresh: string; // 刷新令牌(用于 refreshToken())
access: string; // 访问令牌(由 getApiKey() 返回)
expires: number; // 过期时间戳(毫秒)
}Custom Streaming API 自定义流式 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
Stream Pattern 流式模式
所有提供商遵循相同的模式:
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: "stop",
timestamp: Date.now(),
};
try {
// 推送 start 事件
stream.push({ type: "start", partial: output });
// 发起 API 请求并处理响应...
// 随着内容的到达推送内容事件...
// 推送 done 事件
stream.push({
type: "done",
reason: output.stopReason as "stop" | "length" | "toolUse",
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;
}Event Types 事件类型
按以下顺序通过 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 包含在内。
Content Blocks 内容块
在内容到达时向 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 });Tool Calls 工具调用
工具调用需要累积 JSON 并解析:
// 开始工具调用
output.content.push({
type: "toolCall",
id: toolCallId,
name: toolName,
arguments: {}
});
stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
// 累积 JSON
let 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
});Usage and Cost 用量与成本
根据 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);Context Overflow Errors 上下文溢出错误
当请求超出模型的上下文窗口时,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 = /你的提供商的溢出短语/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时跳过,以使处理程序幂等。
Registration 注册
注册你的流函数:
pi.registerProvider("my-provider", {
baseUrl: "https://api.example.com",
apiKey: "$MY_API_KEY",
api: "my-custom-api",
models: [...],
streamSimple: streamMyProvider
});Testing Your Implementation 测试你的实现
使用与内置提供商相同的测试套件测试你的提供商。复制并适配来自 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 | 提供商之间上下文交接 |
使用你的提供商/模型组合运行测试以验证兼容性。
Config Reference 配置参考
interface ProviderConfig {
/** 在 UI(如 /login)中显示的提供商名称。 */
name?: string;
/** API 端点 URL。定义模型时必需。 */
baseUrl?: string;
/** API key 字面值、环境变量插值($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 key 添加 Authorization: Bearer 请求头。 */
authHeader?: boolean;
/** 要注册的模型。如果提供,将替换该提供商的所有现有模型。 */
models?: ProviderModelConfig[];
/** 支持 /login 的 OAuth 提供商。 */
oauth?: {
name: string;
login(callbacks: OAuthLoginCallbacks): Promise<OAuthCredentials>;
refreshToken(credentials: OAuthCredentials): Promise<OAuthCredentials>;
getApiKey(credentials: OAuthCredentials): string;
};
}Model Definition Reference 模型定义参考
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;
maxTokensField?: "max_completion_tokens" | "max_tokens";
requiresToolResultName?: boolean;
requiresAssistantAfterToolResult?: boolean;
requiresThinkingAsText?: boolean;
requiresReasoningContentOnAssistantMessages?: boolean;
thinkingFormat?: "openai" | "openrouter" | "deepseek" | "together" | "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 }>;
cacheControlFormat?: "anthropic";
sessionAffinityFormat?: "openai" | "openai-nosession" | "openrouter";
sendSessionAffinityHeaders?: boolean;
// anthropic-messages
supportsEagerToolInputStreaming?: boolean;
supportsLongCacheRetention?: boolean;
sendSessionAffinityHeaders?: boolean;
supportsCacheControlOnTools?: boolean;
forceAdaptiveThinking?: boolean;
allowEmptySignature?: boolean;
};
}openrouter 发送 reasoning: { effort }。deepseek 在启用时发送 thinking: { type: "enabled" | "disabled" } 和 reasoning_effort。together 发送 reasoning: { enabled },并且在启用 supportsReasoningEffort 时也发送 reasoning_effort。qwen 用于 DashScope 风格的顶层 enable_thinking。使用 qwen-chat-template 用于读取 chat_template_kwargs.enable_thinking 且需要 preserve_thinking 的本地 Qwen 兼容服务器。使用 chat-template 用于可配置的 chat_template_kwargs,例如运行在 vLLM 之上的 DeepSeek V3.x,使用 chatTemplateKwargs: { "thinking": { "$var": "thinking.enabled" } }。
cacheControlFormat: "anthropic" 将 Anthropic 风格的 cache_control 标记应用于系统提示词、最后一个工具定义以及最后一个用户/助手文本内容。