自定义提供者

扩展可以通过 pi.registerProvider() 注册自定义模型提供者。这可以实现以下功能：

• 代理——通过企业代理或 API 网关路由请求
• 自定义端点——使用自托管或私有模型部署
• OAuth/SSO——为企业提供者添加身份验证流程
• 自定义 API——为非标准 LLM API 实现流式传输

示例扩展

请参阅这些完整的提供者示例：

• examples/extensions/custom-provider-anthropic/
• examples/extensions/custom-provider-gitlab-duo/

快速参考

扩展可以注册完整的 pi-ai Provider，也可以使用传统的提供者配置形式。当需要自定义身份验证、过滤、刷新或流式传输行为时，优先使用完整的 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()
  }));

  // 传统提供者配置形式：
  // 覆盖现有提供者的 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 中都可用。

覆盖现有提供者

最简单的用例：通过代理重定向现有的提供者。

// 所有 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）时，该提供者的所有现有模型都会保留，并使用新的端点。

注册新提供者

要添加一个全新的提供者，除了所需的配置外，还需要指定 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,           // 美元/百万 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, ...) 注册的提供者：

// 注册
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 类型

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,   // 工具结果需要 name 字段
    thinkingFormat: "qwen",        // 顶层 enable_thinking: true
    cacheControlFormat: "anthropic" // Anthropic 风格的 cache_control 标记
  }
}]

使用 openrouter 来实现 OpenRouter 风格的 reasoning: { effort } 控制。使用 together 来实现 Together 风格的 reasoning: { enabled } 控制；当启用了 supportsReasoningEffort 时，它还会发送 reasoning_effort。对于本地的 Qwen 兼容服务器，使用 qwen-chat-template，这些服务器会读取 chat_template_kwargs.enable_thinking 并且需要 preserve_thinking。 对于暴露 Anthropic 风格提示缓存的 OpenAI 兼容提供者，使用 cacheControlFormat: "anthropic"，该缓存通过系统提示、最后一个工具定义以及最后一个用户、助手或工具结果文本内容上的 cache_control 实现。

对于使用 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 支持

添加与 /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:" });
      }

      // 交换Token（你的实现）
      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;

  // 提示用户输入（用于手动输入Token）
  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;   // 刷新Token（用于 refreshToken()）
  access: string;    // 访问Token（由 getApiKey() 返回）
  expires: number;   // 过期时间戳（毫秒）
}

自定义流式 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 {
      // 推送开始事件
      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");
      }

      // 推送完成事件
      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() 推送事件：

1. { type: "start", partial: output } —— 流开始
2. 内容事件（可重复，为每个块跟踪 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 } —— 工具调用结束
3. { 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 });

// 累积 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
});

使用量和成本

根据 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 为自动压缩而跟踪助手消息之抢跑，因此 pi 检查的是重写后的 errorMessage。有了这个机制，pi 将：

1. 从 errorMessage 检测溢出。
2. 从实时上下文中丢弃失败的助手消息。
3. 运行压缩。
4. 重试请求一次。

谨慎地进行重写保护：

• 将其范围限定在你的提供者（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/ 复制并调整这些测试文件：

使用你的提供者/模型对运行测试以验证兼容性。

配置参考

interface ProviderConfig {
  /** 在 UI 中（例如 /login）显示的提供者名称。 */
  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): Promise<OAuthCredentials>;
    getApiKey(credentials: OAuthCredentials): string;
  };
}

模型定义参考

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;
    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" | "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;
    supportsStrictTools?: boolean;
  };
}

openrouter 发送 reasoning: { effort }。deepseek 发送 thinking: { type: "enabled" | "disabled" }，并在启用时发送 reasoning_effort。together 发送 reasoning: { enabled }，并在启用了 supportsReasoningEffort 时也发送 reasoning_effort。qwen 用于 DashScope 风格的顶层 enable_thinking。对于本地的 Qwen 兼容服务器，使用 qwen-chat-template，这些服务器会读取 chat_template_kwargs.enable_thinking 并需要 preserve_thinking。使用 chat-template 用于可配置的 chat_template_kwargs，例如通过 vLLM 的 DeepSeek V3.x 使用 chatTemplateKwargs: { "thinking": { "$var": "thinking.enabled" } }。 cacheControlFormat: "anthropic" 将 Anthropic 风格的 cache_control 标记应用于系统提示、最后一个工具定义以及最后一个用户、助手或工具结果文本内容。

• 原文链接： github.com/badlogic/pi-m...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～