添加新模型

本指南说明如何向 HybridInference 网关添加 LLM 模型与 provider,同时保持客户端看到的 OpenAI 兼容 API 接口不变。

两种需求共用这一份指南。根据你的情况,选择其中一条路径:

  1. 复用已有的 provider adapter——只需改注册表 YAML 和环境变量。

  2. 接入新的 provider——先添加或接上 adapter,再改注册表 YAML 和环境变量。

如果模型是你自己用 vLLM、SGLang 或 Ollama 提供服务的,见添加新的本地模型——本地服务器的路由注册由那一篇负责。

模型注册表在哪里

本仓库里没有 config/models.yaml。后端在启动时于 apps/backend/serving/config/distribution.pyresolve_config_path("models"))解析注册表路径,优先级如下:

  1. MODELS_CONFIG_PATH(旧别名 MODELS_CONFIG)。显式给出的路径始终优先。

  2. distribution manifest 的 paths.models——仅当 DISTRIBUTION_CONFIG_PATH 指向一份 manifest DISTRIBUTION_CONFIG_MODE=active 时才使用。manifest 里的相对路径按 manifest 自身所在目录解析。DISTRIBUTION_CONFIG_MODE 默认是 dark:manifest 会被加载、校验并写进日志以供比对,但不会真正生效。

  3. config/examples/models.openrouter.yaml——仓库自带的参考注册表;全新检出在没有配置其他来源时就用它对外提供服务。

路由配置走同一条解析链(ROUTING_CONFIG_PATH / paths.routing,默认为 config/examples/routing.minimal.yaml)。

所以自建部署有两种形态可选:

  • 指向一个文件。注册表放在任意位置,设置 MODELS_CONFIG_PATH=/path/to/models.yaml 即可。README.md 里的快速开始就是这么做的。

  • 发布一个 overlay。创建 distributions/<name>/,里面放一份 distribution.yaml manifest 和一份 config/models.yaml,然后设置 DISTRIBUTION_CONFIG_PATH=distributions/<name>/distribution.yamlDISTRIBUTION_CONFIG_MODE=activedistributions/example/ 是一个可以直接复制的可用 overlay;config/examples/distribution.example.yaml 是一份带注释的 manifest。

本指南中所说的「你的模型注册表」,指的就是上述解析过程最终选中的那个文件。

如果解析出的路径上没有注册表,后端会记录一条错误日志,/v1/models 返回空列表,并把每个模型都报为 not found。

快速开始:为已有 provider 添加模型

如果该 provider 已经有 adapter,只需要改配置。

  1. 在模型注册表中添加一条模型条目

models:
  - id: your-model-id
    name: Your Model Display Name
    provider: existing_provider  # e.g. "gemini", "deepseek"
    provider_model_id: "actual-provider-model-id"
    base_url: ${PROVIDER_BASE_URL}
    api_key: ${PROVIDER_API_KEY}
    quantization: "bf16"
    input_modalities: ["text"]
    output_modalities: ["text"]
    context_length: 8192
    max_output_length: 4096
    supports_tools: true
    supports_structured_output: true
    supported_params: [temperature, top_p, max_tokens, stop]
    pricing:
      prompt: "0"
      completion: "0"
      image: "0"
      request: "0"
      input_cache_reads: "0"
      input_cache_writes: "0"
    route:
      - kind: existing_provider
        weight: 1.0

这条路由从模型那里继承 base_urlapi_key,所以它只需要写出不同的部分。当第二条路由指向别处时,再在那条路由上重写它们。

  1. 在仓库根目录的 .env设置环境变量(后端的配置加载器读的就是这个文件):

PROVIDER_BASE_URL=https://api.provider.example/v1
PROVIDER_API_KEY=your-api-key
  1. 重启后端以加载新模型。

  2. 验证(见第 5 步:通过网关验证)。

关于 aliases:如果想让客户端也能用第二个名字调用这个模型——比如 OpenRouter 风格的厂商 slug,或者你的推理运行时使用的原始模型路径——把这些名字列进 aliases。它们解析到同一组路由。

添加新的 provider

第 1 步:先判断到底需不需要 adapter

多数新 provider 都提供 OpenAI 风格的 /chat/completions 端点。对这类 provider,不要写 adapter 类。注册一个 provider profile,并把这个 kind 加进 apps/backend/serving/servers/registry.py 中的 OpenAI 兼容分发元组(_make_adapter):

_make_adapter 是一条很长的 if kind ... / elif kind ... 链,分派依据是路由的 kind。先在这条链上为你的 profile 加一个分支,再把这个 kind 加进同一个函数里更下面的 OpenAI-compat 元组:

# ...among the per-kind arms of _make_adapter:
elif kind == "your_provider":
    cfg = {**cfg, "provider_profile": "your_provider"}

# ...further down in the same function:
if kind in (
    "vllm",
    "sglang",
    "chutes",
    "featherless",
    "ollama",
    "cliproxy",
    "openai_compat",
    "staging",
    "deepseek",
    "kimi",
    "minimax",
    "your_provider",  # <-- add it here
):
    return OpenAICompatAdapter(model_cfg)

deepseekkimiminimax 现在就是这样接入的:apps/backend/serving/adapters/profiles.py 里每个 provider 一份的 profile 承载用量指标或路径上的差异,其余全部交给 OpenAICompatAdapterzaikimi_coding 用的是同一套 profile 机制,但它们额外绑定了一个 coding 工具身份,所以 _make_adapter 会在走到那个元组之前,先把它们短路到 CodingIdentityAdapterOpenAICompatAdapter 的一个很薄的子类)。

只有当 provider 说的确实不是 OpenAI 的 wire 格式时,才写专门的 adapter——例如 Gemini 的 generateContent、Anthropic Messages API、OpenRouter 用来锁定 sub-provider 的 body 字段。geminiclaudeanthropicopenrouter 都属于这一类,各自有独立的分发分支:

if kind == "your_provider":
    return YourProviderAdapter(model_cfg)

第 2 步:编写 adapter(仅限自定义协议)

apps/backend/serving/adapters/ 下新建一个文件,例如 apps/backend/serving/adapters/your_provider.py。后端的包根目录是 apps/backend,因此 import 写成 serving.… / routing.…

import json
from collections.abc import AsyncGenerator
from typing import Any

from serving.stream import done_sentinel, make_final_usage_chunk
from serving.utils.tokens import estimate_prompt_tokens, estimate_text_tokens
from .base import BaseAdapter, UsageInfo


class YourProviderAdapter(BaseAdapter):
    """Adapter for YourProvider API.

    This adapter translates OpenAI-compatible requests to YourProvider's
    API format and normalizes responses back to OpenAI format.
    """

    async def chat_completion(
        self, messages: list[dict[str, Any]], **params
    ) -> dict[str, Any]:
        """Execute a non-streaming chat completion request.

        Args:
            messages: List of chat messages in OpenAI format.
            **params: Additional parameters (temperature, max_tokens, etc.).

        Returns:
            OpenAI-compatible response dictionary.
        """
        # Validate and clamp parameters against this model's declared support.
        validated_params = self.validate_params(params)

        # Build the provider-specific request payload.
        payload = {
            "model": self.config.provider_model_id or self.config.id,
            "messages": messages,
            **validated_params,
        }

        if params.get("tools"):
            payload["tools"] = params["tools"]

        if params.get("response_format", {}).get("type") == "json_object":
            payload["response_format"] = {"type": "json_object"}

        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.config.api_key}",
        }

        data = await self.http.json_post_with_retry(
            f"{self.config.base_url}/chat/completions",
            json=payload,
            headers=headers,
        )

        usage = UsageInfo(
            prompt_tokens=data.get("usage", {}).get("prompt_tokens", 0),
            completion_tokens=data.get("usage", {}).get("completion_tokens", 0),
            total_tokens=data.get("usage", {}).get("total_tokens", 0),
        )

        # Fall back to estimation when the provider reports no usage.
        if usage.total_tokens == 0:
            content = data["choices"][0]["message"].get("content", "")
            prompt_tokens = estimate_prompt_tokens(messages)
            completion_tokens = estimate_text_tokens(content)
            usage = UsageInfo(
                prompt_tokens=int(prompt_tokens),
                completion_tokens=int(completion_tokens),
                total_tokens=int(prompt_tokens + completion_tokens),
            )

        tool_calls = None
        if "tool_calls" in data["choices"][0]["message"]:
            tool_calls = data["choices"][0]["message"]["tool_calls"]

        return self.format_response(
            content=data["choices"][0]["message"].get("content", ""),
            model=self.config.id,
            usage=usage,
            tool_calls=tool_calls,
            finish_reason=data["choices"][0].get("finish_reason", "stop"),
        )

    async def stream_chat_completion(
        self, messages: list[dict[str, Any]], **params
    ) -> AsyncGenerator[str, None]:
        """Execute a streaming chat completion request.

        Args:
            messages: List of chat messages in OpenAI format.
            **params: Additional parameters.

        Yields:
            Server-sent event formatted strings.
        """
        validated_params = self.validate_params(params)

        payload = {
            "model": self.config.provider_model_id or self.config.id,
            "messages": messages,
            "stream": True,
            **validated_params,
        }

        if params.get("tools"):
            payload["tools"] = params["tools"]

        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.config.api_key}",
        }

        total_content = ""
        prompt_tokens = 0

        async for line in self.http.stream_post(
            f"{self.config.base_url}/chat/completions",
            json=payload,
            headers=headers,
        ):
            if not line.startswith("data: "):
                continue

            if line == "data: [DONE]":
                # Emit the final usage chunk with the shared helper.
                yield make_final_usage_chunk(
                    model=self.config.id,
                    messages=messages,
                    total_content=total_content,
                    prompt_tokens_override=prompt_tokens or None,
                    finish_reason="stop",
                )
                yield done_sentinel()
                break

            try:
                chunk_data = json.loads(line[6:])

                if "usage" in chunk_data:
                    prompt_tokens = chunk_data["usage"].get("prompt_tokens", prompt_tokens)

                if chunk_data["choices"][0]["delta"].get("content"):
                    content = chunk_data["choices"][0]["delta"]["content"]
                    total_content += content
                    yield self.format_stream_chunk(content, self.config.id)
            except json.JSONDecodeError:
                continue

apps/backend/serving/adapters/__init__.py 里导出它:

from .your_provider import YourProviderAdapter

__all__ = [
    # ... existing exports
    "YourProviderAdapter",
]

并在 apps/backend/serving/servers/registry.py 顶部导入它:

from serving.adapters import (
    # ... existing imports
    YourProviderAdapter,
)

第 3 步:把模型加进注册表

models:
  - id: your-model-id
    name: Your Model Name
    provider: your_provider
    provider_model_id: "actual-model-id"
    base_url: ${YOUR_PROVIDER_BASE_URL}
    api_key: ${YOUR_PROVIDER_API_KEY}
    quantization: "bf16"
    input_modalities: ["text"]
    output_modalities: ["text"]
    context_length: 8192
    max_output_length: 4096
    supports_tools: true
    supports_structured_output: true
    supported_params: [temperature, top_p, max_tokens, stop]
    aliases: []  # Optional alternative names
    pricing:
      prompt: "0"
      completion: "0"
      image: "0"
      request: "0"
      input_cache_reads: "0"
      input_cache_writes: "0"
    route:
      - kind: your_provider
        weight: 1.0
        base_url: ${YOUR_PROVIDER_BASE_URL}
        api_key: ${YOUR_PROVIDER_API_KEY}

第 4 步:配置环境变量

加进 .env

YOUR_PROVIDER_BASE_URL=https://api.yourprovider.example/v1
YOUR_PROVIDER_API_KEY=your-api-key-here

一条路由的 api_keyapi_keysbase_url 若由 ${VAR} 提供而解析结果为空,这条路由不会被注册。默认情况下整个模型都会被丢弃,后端会记录哪些模型被跳过、哪些变量没设置。给路由标上 optional: true,就只跳过这一条路由,模型的其余部分照常注册。

第 5 步:通过网关验证

启动后端。在一个检出里,最快的启动形式是:

PYTHONPATH=apps/backend \
  MODELS_CONFIG_PATH=/path/to/your/models.yaml \
  uv run uvicorn serving.servers.app:app --port 8080

如果用仓库自带的 Docker Compose,则改用 make build s=backend 重新构建并重启 backend。

GET /v1/models 接受匿名请求——它解析 API key 只是为了决定要不要包含仅管理员可见的条目:

curl -s http://localhost:8080/v1/models | jq

POST /v1/chat/completions 会经过 verify_api_key,因此没有有效的网关 API key 就会返回 401,除非后端以 USER_AUTH_ENABLED=false 运行。请发送你为自己的网关签发的 key(是网关的 key,不是上游 provider 的):

export GATEWAY_API_KEY=<your gateway API key>

curl -s -X POST http://localhost:8080/v1/chat/completions \
  -H "Authorization: Bearer $GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "your-model-id",
    "messages": [{"role": "user", "content": "Hello!"}]
  }' | jq

流式测试:

curl -N -s -X POST http://localhost:8080/v1/chat/completions \
  -H "Authorization: Bearer $GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "your-model-id",
    "messages": [{"role": "user", "content": "Stream test"}],
    "stream": true,
    "max_tokens": 64
  }'

配置参考

模型字段

这些键从模型条目里读出并传给 ModelConfigapps/backend/serving/adapters/base.py)。其中大多数也可以设在单条路由上,此时路由上的值优先;idname 标识的是模型本身,只在模型层级读取。

字段

类型

必填

说明

id

string

唯一的模型标识;客户端发送的就是这个名字

name

string

展示名称

provider

string

provider/adapter 的种类;在没有给出 route: 列表时,它同时是 route[].kind 的默认值。注意这里有个同名陷阱:模型层级的 provider: 选的是 adapter,而路由上的 provider: 只是一个分析标签——见路由字段表

base_url

string

API 端点的 base URL

api_key

string

API 鉴权密钥

provider_model_id

string

provider 侧的模型标识(在 wire 上覆盖 id

model_type

string

"chat"(默认)或 "embedding"。别名 type: 与之等价;embedding 模型绕过加权 router,把自己的多条路由当作有序的回退链使用

aliases

list[string]

用于路由的备用名称

quantization

string

量化格式(默认:"bf16"

input_modalities

list[string]

输入类型:"text""image"

output_modalities

list[string]

输出类型:"text"

context_length

int

最大上下文窗口(默认:8192)

max_output_length

int

最大输出 token 数(默认:4096);max_tokens 会被钳制到这个值

supports_tools

bool

是否支持 function calling(默认:false)

supports_structured_output

bool

是否支持 JSON mode(默认:false)

supported_params

list[string]

允许的参数名(默认:temperaturetop_pmax_tokens

reasoning_efforts

list[string]

这个模型的 reasoning_effort 接受哪些取值。只有当 reasoning_effortsupported_params 里时才有意义;为空表示「不提供」。这里刻意没有默认值——各模型接受的取值集合不同,猜一个只会对外宣称一个上游并不接受的值

on_demand

bool

模型在共享 GPU 上按需加载(首次请求时启动,空闲时停止)。会在 /v1/models 中暴露,且 RouteWise 的后台延迟探测器绝不探测这类端点(默认:false)

processor

string

OpenAICompatAdapter 指定输出处理器,绕过按模型 id 的自动识别。可取值:"default""glm""qwen_coder""think_block"

extra_body

dict

合并进 OpenAI 兼容上游请求体的默认字段。核心字段和已校验的客户端参数优先

priority_scheduling

bool

该端点上的 sglang 服务器是带 --enable-priority-scheduling 启动的;见在 sglang 路由上优先调度 decode。通常按路由设置,而不是按模型设置

route_metadata

dict

供路由策略消费的、每条路由上的自由格式元数据

pricing

dict

基础成本信息。promptcompletioninput_cache_readsinput_cache_writes 的单位是每 100 万 token 美元;request 是每次请求美元,image 是每张图美元

pricing_schedule

dict

只按 UTC 计的生效时间,加上每天重复的价格时间窗:effective_at 之前继续使用 pricing;之后,落在每个左闭右开 [start, end) 窗口之外的时段适用 default,而窗口自带的 pricing 覆盖基础字段

模型条目里有几个键并不是 ModelConfig 的字段,而是在别处被消费:route(见下)、routerrouter_params(按模型选择 router)、admin_onlyrequired_role

路由配置

路由让一个模型拥有多个端点,并按权重分配流量:

route:
  # Local vLLM deployment
  - kind: vllm
    weight: 0.7  # 70% of traffic
    base_url: http://localhost:8000
    provider_model_id: "/models/local-model"

  # Remote API fallback
  - kind: your_provider
    weight: 0.3  # 30% of traffic
    base_url: https://api.provider.example
    api_key: ${API_KEY}

除上面的模型字段之外,路由条目还接受:

字段

类型

说明

kind

string

adapter 的 kind(见下)。默认取模型的 provider

weight

float

流量的相对占比(默认 1.0)。0 表示保留这条路由的配置但不会被选中

api_keys

list[string]

这个端点的 key 池,用来替代 api_key。两者同时设置会报错。在延迟统计和配额记账上,整个池算作一个端点;真正互相独立的资源要拆成不同的路由

optional

bool

当由 ${VAR} 提供的 key 或 base_url 解析为空时,只跳过这一条路由,而不是丢弃整个模型

provider / provider_display_name

string

只覆盖分析统计里的标签——它重命名的是仪表盘里的那一行,不会选择 adapter,选 adapter 的是 kind。见在仪表盘中给路由命名

provider_type

string

RouteWise 的成本类别:on_demandquotaconcurrency

routewise_pool, quota_pool, concurrency_pool, quota_source, quota, concurrency

RouteWise 的资源池与预算元数据

一条路由用自己的 kind 作为 provider 标签上报——这个值记录在 api_logs.provider 里,所有按 provider 划分的管理视图都以它分组。因此同一个 kind 的两条路由会共用仪表盘上的同一行;用 provider: 才能把它们分开。

支持的 adapter kind

每条路由条目里的 kind 字段决定用哪个后端 adapter。所有标为 OpenAI-compat 的 kind 共用同一份 OpenAICompatAdapter 实现,并自动套用各 provider 专属的 profile。

Kind

类别

备注

openai_compat

OpenAI-compat

通用的 OpenAI 兼容端点;没有更贴切的 kind 时用它

staging

OpenAI-compat

openai_compat 的克隆,但有自己的 provider 标签,便于在指标里单独跟踪第二个通用端点

vllm

OpenAI-compat

本地 vLLM 推理服务器

sglang

OpenAI-compat

本地 SGLang 推理服务器

ollama

OpenAI-compat

本地或远程的 Ollama 服务器

chutes

OpenAI-compat

Chutes.ai 托管推理

featherless

OpenAI-compat

Featherless.ai 托管推理

cliproxy

OpenAI-compat

面向 OpenAI 兼容模型的 CLI 代理端点

deepseek

OpenAI-compat

DeepSeek API(套用 DeepSeek 用量 profile)

kimi

OpenAI-compat

Moonshot/Kimi 按 token 计费的 API(套用 Kimi 用量 profile)

kimi_coding

OpenAI-compat

Kimi coding 套餐端点;分发到 CodingIdentityAdapter

zai

OpenAI-compat

Z.AI GLM coding 套餐:chat 路径是在 base URL 后面拼 /chat/completions,而不是默认的 /v1/chat/completions,因为 Z.AI 的 base URL 里已经带了版本段(profiles.default_chat_path)。它分发到 CodingIdentityAdapter——后者会带上 coding 工具的 User-Agent 和一条前置 system 消息

minimax

OpenAI-compat

MiniMax API(套用 MiniMax 用量 profile)

openrouter

自定义

OpenRouter 聚合器。用方括号形式 openrouter[<slug>] 锁定某个 sub-provider

gemini

自定义

Google Gemini API(需要转换消息格式)

claude

自定义

经 Google Vertex 访问的 Anthropic Claude

anthropic

自定义

直连 Anthropic Messages API 的客户端

其他任何 kind 都会在加载注册表时抛出 ValueError: Unknown adapter kind

混合路由

加权路由在注册时生效。之后,路由配置文件(通过 ROUTING_CONFIG_PATH / manifest 的 paths.routing 解析)可以经 RoutingManager 集中调整权重。见路由

BaseAdapter API 参考

所有 adapter 都继承自 BaseAdapterapps/backend/serving/adapters/base.py),并实现:

async def chat_completion(
    self, messages: list[dict[str, Any]], **params
) -> dict[str, Any]:
    """Execute non-streaming chat completion."""

async def stream_chat_completion(
    self, messages: list[dict[str, Any]], **params
) -> AsyncGenerator[str, None]:
    """Execute streaming chat completion."""

基类提供的工具方法:

def validate_params(self, params: dict[str, Any]) -> dict[str, Any]:
    """Validate and clamp parameters to supported ranges."""

def format_response(
    self,
    content: str | None,
    model: str,
    usage: UsageInfo | None = None,
    tool_calls: list[dict] | None = None,
    reasoning_content: str | None = None,
    finish_reason: str = "stop",
) -> dict[str, Any]:
    """Format response in OpenAI-compatible format."""

def format_stream_chunk(
    self,
    content: str,
    model: str,
    finish_reason: str | None = None,
    role: str | None = None,
) -> str:
    """Format an SSE chunk for streaming responses."""

def format_tool_chunk(self, tool_calls: list[dict[str, Any]], model: str) -> str:
    """Format tool calls into an OpenAI-compatible streaming chunk."""

可用的属性:

self.config       # ModelConfig instance
self.http         # AsyncHTTPClient (apps/backend/serving/http.py), shared

进阶功能

多模态支持

对于接受图像输入的模型:

input_modalities: ["text", "image"]

在 adapter 的 chat_completion 里处理图像内容块。单条路由可以声明比模型更窄的 input_modalities,这样纯文本的回退路由就永远不会收到媒体内容。

工具 / function calling

supports_tools: true

把 provider 返回的工具调用解析成 OpenAI 的结构再透传出去:

tool_calls = []
if "function_call" in data:
    tool_calls.append({
        "id": f"call_{int(time.time() * 1000)}",
        "type": "function",
        "function": {
            "name": data["function_call"]["name"],
            "arguments": data["function_call"]["arguments"],
        },
    })

return self.format_response(
    content=content,
    model=self.config.id,
    usage=usage,
    tool_calls=tool_calls,
)

结构化输出(JSON mode)

supports_structured_output: true

处理 response_format 参数:

if params.get("response_format", {}).get("type") == "json_object":
    payload["response_format"] = {"type": "json_object"}

限流(按 provider 的限流并不存在)

没有按 provider 的限流器可以开启;本节记录的是它将来该放在哪里。apps/backend/serving/servers/bootstrap.py 并没有配置按 provider 的限流器。那里接线的唯一进程内限流器是 UserConcurrencyLimiter。鉴权流程的静态限流值在 apps/backend/serving/config/settings.pysignup_rate_limit_*login_rate_limit_* 字段)。若要做按 provider 的令牌桶或配额,应放进 apps/backend/serving/admin/provider_quotas.py(或一个新模块),并通过 apps/backend/serving/servers/deps.py 暴露出来。

示例

  • OpenAI 兼容的 provider。DeepSeek 没有 adapter 文件。apps/backend/serving/servers/registry.py 里的 _make_adapter 设置 provider_profile = "deepseek" 并返回 OpenAICompatAdapter;profile 本身在 apps/backend/serving/adapters/profiles.py

  • 自定义 API 格式。针对非 OpenAI wire 格式的消息转换,见 apps/backend/serving/adapters/gemini.py

  • 本地部署。vLLM 和 SGLang 复用 apps/backend/serving/adapters/openai_compat.pyvllmsglang 这两个 kind 分发到同一个类;本地与远程的行为差异来自 base_url 和路由层,而不是来自专门的 adapter。

排查

模型没有出现在 /v1/models

  • 确认后端实际加载的是哪份注册表。它在启动时会记录 Registered N routes from <path>;如果那个路径上没有注册表,则会记录一条写明路径的错误日志。

  • 检查 models: 下的 YAML 语法和缩进。

  • 检查是否有模型被跳过:一条路由的 key 或 base_url 若由 ${VAR} 提供且解析为空,该模型就会被丢弃,日志里会同时写出模型名和未设置的变量名。

  • 如果用了 aliases,确认规范的 id 只出现一次,且别名没有和别的模型撞名。重复的别名会解析到最后加载的那个模型,并记录一条警告。

鉴权失败

  • 网关返回的 401 表示你的网关 API key 缺失或无效,或者鉴权已开启而你没有发送 Authorization 头。

  • 从路由透出来的 401 表示上游的 key 不对。后端会把它记为 upstream_auth_misconfig,并带上端点 id 和上游自己的错误响应体。

  • 确认 ${ENV_VAR} 展开成功:只有恰好是 ${NAME} 这种形式的取值才会被展开,而且只对 base_urlapi_keyapi_keysprovider_model_id 生效。

响应格式错误

  • 确保 format_response() 返回 OpenAI 兼容的结构。

  • 校验 UsageInfo 的各字段都是整数。

  • 检查 finish_reasonstoplengthcontent_filtertool_calls 之一。

  • 流式场景下,第一个非空内容分片一可用就立刻发出,这样首 token 时延(time-to-first-token)才会被准确记录。

流式问题

  • 确保分片是 SSE 格式:data: {json}\n\n

  • data: [DONE] 之前发送最后那个 usage 分片。

  • 妥善处理 JSON 解析错误。

最佳实践

  1. 错误处理——用 self.http.json_post_with_retry(),并用有价值的信息把 provider 的故障暴露出来。

  2. 用量记账——优先采用 provider 上报的 usage;拿不到再回退到 estimate_prompt_tokens() / estimate_text_tokens()

  3. 流式辅助函数——用 format_stream_chunk()make_final_usage_chunk()done_sentinel() 保证 SSE 输出一致。

  4. 类型安全——写完整的类型标注,并让请求/响应的结构与 apps/backend/serving/schemas.py 保持一致。

  5. 测试——流式和非流式两条路径都要跑到,并用大 prompt 验证 token 的钳制逻辑。

  6. 文档与风格——用英文写 Google 风格的 docstring;不要把 provider 专属的逻辑塞进共享代码。

  7. 密钥——在 YAML 里用 ${ENV_VAR},不要硬编码 key 或端点,取值放在 .env 中。

另见