客製化Providers
Extensions可以透過pi.registerProvider()註冊自訂模型提供者。這使得:
- 代理 - 透過公司代理或 API 網關路由請求
- 自訂端點 - 使用自架或私人模型部署
- OAuth/SSO - 為企業提供者新增身分驗證流程
- 自訂 APIs - 為非標準 LLM APIs 實現串流傳輸
範例Extensions
請參閱這些完整的提供者範例:
目錄
- Example Extensions
- Quick Reference
- Override Existing Provider
- Register New Provider
- Unregister Provider
- OAuth Support
- Custom Streaming API
- Context Overflow Errors
- Testing Your Implementation
- Config Reference
- Model Definition Reference
快速參考
Extensions 可以註冊完整的 pi-ai 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()
}));
// Legacy provider-config form:
// Override baseUrl for existing provider
pi.registerProvider("anthropic", {
baseUrl: "https://proxy.example.com"
});
// Register new provider with models
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 期間可用。
覆蓋現有提供者
最簡單的用例:透過代理程式重定向現有提供者。
// All Anthropic requests now go through your proxy
pi.registerProvider("anthropic", {
baseUrl: "https://proxy.example.com"
});
// Add custom headers to OpenAI requests
pi.registerProvider("openai", {
headers: {
"X-Custom-Header": "value"
}
});
// Both baseUrl and headers
pi.registerProvider("google", {
baseUrl: "https://ai-gateway.corp.com/google",
headers: {
"X-Corp-Auth": "$CORP_AUTH_TOKEN" // env var or literal
}
});當僅提供 baseUrl 和/或 headers(無 models)時,該提供者的所有現有模型都將與新端點一起保留。
註冊新提供者
若要新增全新的提供程序,請指定 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", // env var reference
api: "openai-completions", // which streaming API to use
models: [
{
id: "my-llm-large",
name: "My LLM Large",
reasoning: true, // supports extended thinking
input: ["text", "image"],
cost: {
input: 3.0, // $/million tokens
output: 15.0,
cacheRead: 0.3,
cacheWrite: 3.75
},
contextWindow: 200000,
maxTokens: 16384
}
]
});當提供 models 時,它會替換該提供者的所有現有模型。
apiKey 和自訂標頭值使用與 models.json 相同的設定值語法:開頭的 !command 會對整個值執行命令,$ENV_VAR 和 ${ENV_VAR} 會插入環境變數,$ 會輸出字面 $,$! 會輸出字面 !。
取消註冊提供者
使用 pi.unregisterProvider(name) 刪除先前透過 pi.registerProvider(name,...) 註冊的提供者:
// Register
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
}
]
});
// Later, remove it
pi.unregisterProvider("my-llm");取消註冊會刪除該提供者的動態模型、API key 後備、OAuth 提供者註冊和自訂流程處理程序註冊。任何被覆蓋的內建模型或提供者行為都會被恢復。
初始擴充載入階段之後進行的呼叫會立即套用,因此不需要 /reload。
API 類型
api字段決定使用哪種流實現:
| API | 用於 |
|---|---|
anthropic-messages |
人擇克勞德 API 及其兼容者 |
openai-completions |
OpenAI 聊天完成 API 和相容版本 |
openai-responses |
OpenAI 回應 API |
azure-openai-responses |
Azure OpenAI 回應 API |
openai-codex-responses |
OpenAI Codex 回覆 API |
mistral-conversations |
本地米斯特拉爾聊天完成串流 |
google-generative-ai |
谷歌生成人工智慧API |
google-vertex |
Google Vertex AI API |
bedrock-converse-stream |
亞馬遜 Bedrock 匡威 API |
大多數與 OpenAI 相容的供應商都使用 openai-completions。使用模型級別 thinkingLevelMap 來實現特定於模型的思維級別,使用 compat 來實現提供者的怪癖。 xhigh 和 max 等級是可選的,需要非空映射條目,並且可能被不支援的孔分隔:
models: [{
id: "custom-model",
// ...
reasoning: true,
thinkingLevelMap: { // map pi levels to provider values; null hides unsupported levels
minimal: null,
low: null,
medium: null,
high: "default",
xhigh: null,
max: "max"
},
compat: {
supportsDeveloperRole: false, // use "system" instead of "developer"
supportsReasoningEffort: true,
maxTokensField: "max_tokens", // instead of "max_completion_tokens"
requiresToolResultName: true, // tool results need name field
thinkingFormat: "qwen", // top-level enable_thinking: true
cacheControlFormat: "anthropic" // Anthropic-style cache_control markers
}
}]將 openrouter 用於 OpenRouter 樣式 reasoning: { effort } 控制。將 together 用於 Together 樣式 reasoning: { enabled } 控制;對於supportsReasoningEffort,它也發送reasoning_effort。對於讀取 chat_template_kwargs.enable_thinking 並需要 preserve_thinking 的本機 Qwen 相容伺服器,請使用 qwen-chat-template。
將 cacheControlFormat: "anthropic" 用於與 OpenAI 相容的提供程序,透過 cache_control 在系統提示、最後一個工具定義以及最後一個使用者、助手或工具結果文字內容上公開人類風格的提示快取。
對於使用api: "anthropic-messages"的人類相容提供者,在其上游模型需要自適應思維的模型或提供者上設定compat.forceAdaptiveThinking: true(thinking.type: "adaptive"加output_config.effort)。內建自適應克勞德模型會自動設定此功能。僅針對發出空思維簽名並期望重播時 signature: "" 的提供者設定 compat.allowEmptySignature: true。
遷移注意:米斯特拉爾從
openai-completions移至mistral-conversations。 對原生 Mistral 模型使用mistral-conversations。 如果您有意透過openai-completions路由 Mistral 相容/自訂端點,請根據需要明確設定compat標誌。
驗證頭
如果您的提供者期望 Authorization: Bearer <key> 但不使用標準 API,請設定 authHeader: true:
pi.registerProvider("custom-api", {
baseUrl: "https://api.example.com",
apiKey: "$MY_API_KEY",
authHeader: true, // adds Authorization: Bearer header
api: "openai-completions",
models: [...]
});每個請求都會解析密鑰。明確請求 Authorization 標頭優先於產生的值。
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:" });
}
// Exchange for tokens (your implementation)
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進行身份驗證。
OAuth登入回調
callbacks 物件為提供者擁有的流程提供 UI 中立的互動:
interface OAuthLoginCallbacks {
// Open URL in browser (for OAuth redirects)
onAuth(params: { url: string }): void;
// Show device code (for device authorization flow)
onDeviceCode(params: {
userCode: string;
verificationUri: string;
intervalSeconds?: number;
expiresInSeconds?: number;
}): void;
// Show transient progress
onProgress?(message: string): void;
// Prompt user for input (for manual token entry)
onPrompt(params: { message: string }): Promise<string>;
// Show an interactive selector, e.g. to choose browser OAuth vs device code
onSelect(params: {
message: string;
options: { id: string; label: string }[];
}): Promise<string | undefined>;
}OAuth憑證
憑證保存在 ~/.pi/agent/auth.json 中:
interface OAuthCredentials {
refresh: string; // Refresh token (for refreshToken())
access: string; // Access token (returned by getApiKey())
expires: number; // Expiration timestamp in milliseconds
}自訂串流媒體API
對於具有非標準API的供應商,實作streamSimple。在編寫自己的提供程序之前,請先研究現有的提供者實作:
參考實作:
- anthropic.ts - 人為訊息 API
- mistral.ts - 米斯特拉爾對話 API
- openai-completions.ts - OpenAI 聊天完成
- openai-responses.ts - OpenAI 回應 API
- google.ts - 谷歌生成式人工智慧
- amazon-bedrock.ts - AWS 基岩
流模式
所有提供者都遵循相同的模式:
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 () => {
// Initialize output message
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 {
// Push start event
stream.push({ type: "start", partial: output });
// Make API request and process response...
// Push content events as they arrive and set stopReason from the terminal event.
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");
}
// Push done event
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:
// Text block
output.content.push({ type: "text", text: "" });
stream.push({ type: "text_start", contentIndex: output.content.length - 1, partial: output });
// As text arrives
const block = output.content[contentIndex];
if (block.type === "text") {
block.text += delta;
stream.push({ type: "text_delta", contentIndex, delta, partial: output });
}
// When block completes
stream.push({ type: "text_end", contentIndex, content: block.text, partial: output });工具調用
工具呼叫需要累加JSON並解析:
// Start tool call
output.content.push({
type: "toolCall",
id: toolCallId,
name: toolName,
arguments: {}
});
stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
// Accumulate JSON
let partialJson = "";
partialJson += jsonDelta;
try {
block.arguments = JSON.parse(partialJson);
} catch {}
stream.push({ type: "toolcall_delta", contentIndex, delta: jsonDelta, partial: output });
// Complete
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);上下文溢出錯誤
當請求超出模型的上下文視窗時,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
});測試您的實施
根據內建提供者使用的相同測試套件來測試您的提供者。從 packages/ai/test/ 複製並調整這些測試檔案:
| 測試 | 目的 |
|---|---|
stream.test.ts |
基本串流、文字輸出 |
tokens.test.ts |
令牌計數和使用 |
abort.test.ts |
Abort訊號處理 |
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 |
總代幣計算 |
cross-provider-handoff.test.ts |
提供者之間的上下文切換 |
使用您的提供者/模型對執行測試以驗證相容性。
配置參考
interface ProviderConfig {
/** Display name for the provider in UI such as /login. */
name?: string;
/** API endpoint URL. Required when defining models. */
baseUrl?: string;
/** API key literal, env interpolation ($ENV_VAR or ${ENV_VAR}), or !command. Required when defining models (unless oauth). */
apiKey?: string;
/** API type for streaming. Required at provider or model level when defining models. */
api?: Api;
/** Custom streaming implementation for non-standard APIs. */
streamSimple?: (
model: Model<Api>,
context: Context,
options?: SimpleStreamOptions
) => AssistantMessageEventStream;
/** Custom headers to include in requests. Values use the same resolution syntax as apiKey. */
headers?: Record<string, string>;
/** If true, adds Authorization: Bearer header with the resolved API key. */
authHeader?: boolean;
/** Models to register. If provided, replaces all existing models for this provider. */
models?: ProviderModelConfig[];
/** OAuth provider for /login support. */
oauth?: {
name: string;
login(callbacks: OAuthLoginCallbacks): Promise<OAuthCredentials>;
refreshToken(credentials: OAuthCredentials, signal: AbortSignal): Promise<OAuthCredentials>;
getApiKey(credentials: OAuthCredentials): string;
};
}模型定義參考
interface ProviderModelConfig {
/** Model ID (e.g., "claude-sonnet-4-20250514"). */
id: string;
/** Display name (e.g., "Claude 4 Sonnet"). */
name: string;
/** API type override for this specific model. */
api?: Api;
/** API endpoint URL override for this specific model. */
baseUrl?: string;
/** Whether the model supports extended thinking. */
reasoning: boolean;
/** Maps pi thinking levels to provider/model-specific values; null marks a level unsupported. */
thinkingLevelMap?: Partial<Record<"off" | "minimal" | "low" | "medium" | "high" | "xhigh" | "max", string | null>>;
/** Supported input types. */
input: ("text" | "image")[];
/** Cost per million tokens (for usage tracking). */
cost: {
input: number;
output: number;
cacheRead: number;
cacheWrite: number;
};
/** Maximum context window size in tokens. */
contextWindow: number;
/** Maximum output tokens. */
maxTokens: number;
/** Custom headers for this specific model. */
headers?: Record<string, string>;
/** Compatibility settings for the selected API. */
compat?: {
// openai-completions
supportsStore?: boolean;
supportsDeveloperRole?: boolean;
supportsReasoningEffort?: boolean;
supportsUsageInStreaming?: boolean;
supportsFinishReason?: boolean;
supportsStrictMode?: boolean;
supportsOpenAIGrammarTools?: boolean; // openai-completions/openai-responses; false falls back to normal function tools
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。當supportsReasoningEffort啟用時,together會發送reasoning: { enabled },也會發送reasoning_effort。 qwen 適用於 DashScope 樣式的頂級 enable_thinking。對於讀取 chat_template_kwargs.enable_thinking 且需要 preserve_thinking 的本機 Qwen 相容伺服器,請使用 qwen-chat-template。使用 chat-template 來配置 chat_template_kwargs,例如 vLLM 後面的 DeepSeek V3.x 帶有 chatTemplateKwargs: { "thinking": { "$var": "thinking.enabled" } }。當提供者期望切換值低於 chat_template_args 並可選擇支援頂級 reasoning_effort 時,請使用 thinkingFormat: "baseten" 和 chatTemplateArgs。
cacheControlFormat: "anthropic" 將人類風格的 cache_control 標記應用於系統提示、最後一個工具定義以及最後一個使用者、助手或工具結果文字內容。