编写 provider adapter
本页面向要给网关接入新 provider 的贡献者。如果只是想用网关已经支持的 provider 再提供一个模型，完全不用写代码；见添加新模型。
大多数 provider 也不需要新写 adapter。如果 API 兼容 OpenAI，一个 profile 加上 registry.py 里的几行改动就够了，见第 1 步。只有协议格式自成一套的 provider，才需要专门的 adapter。
第 1 步：先判断到底需不需要 adapter
大多数新 provider 都提供 OpenAI 风格的 /chat/completions 端点。对这类 provider，不要写 adapter 类。先注册一个 provider profile，再在 apps/backend/serving/servers/registry.py 的 _make_adapter 里改两处：一是加一个选中该 profile 的分支，二是在同一函数靠后的 OpenAI 兼容元组里加上这个 kind：
# ...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",
"zai",
"kimi",
"minimax",
"your_provider", # <-- add it here
):
return OpenAICompatAdapter(model_cfg)
还要把这个 kind 加进同一模块的 RESERVED_PROVIDER_LABELS，防止路由把它拿去当自己的标签；如果可分派的 kind 没有加进这个集合，单元测试会失败。
deepseek、zai、kimi 和 minimax 目前都是这样接入的：每个 provider 在 apps/backend/serving/adapters/profiles.py 里有一份 profile，记录用量指标或路径上的特殊处理，其余都交给 OpenAICompatAdapter。
只有 provider 用的确实不是 OpenAI 的协议格式时，才需要写专门的 adapter，例如 Gemini 的 generateContent、Anthropic Messages API，以及 OpenRouter 用来锁定 sub-provider 的请求体字段。gemini、claude、anthropic 和 openrouter 都是这种情况，各自有独立的分派分支：
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_key、api_keys 或 base_url 引用的 ${VAR} 解析为空，这条路由就不会注册。默认情况下整个模型都会被丢弃，后端日志会写明跳过了哪些模型、哪些变量没设置。给路由加上 optional: true，就只跳过这一条路由，模型的其他路由照常注册。
第 5 步：验证
用包含新 kind 的注册表启动网关，再按通过网关验证里的方法检查。
部署自带的 adapter
部署可以注册自己的 adapter 工厂，不用改内置的分派逻辑。做法是在一个模块里提供一个同步、无参数的 register() 函数：
from serving.adapters import ModelConfig, OpenAICompatAdapter
from serving.servers.registry import register_adapter_factory
def make_example_adapter(cfg):
return OpenAICompatAdapter(ModelConfig(**cfg))
def register():
register_adapter_factory("example_service", make_example_adapter)
把 BACKEND_EXTENSIONS 设为这个模块的导入名，并在模型注册表里使用 kind: example_service。工厂接收路由的配置字典，返回一个 adapter。它在内置 provider 默认值生效之前运行，所以需要的 profile 或路径默认值都得自己提供。重复注册会失败；要替换内置 kind，必须调用 register_adapter_factory(kind, factory, override=True)，并且会记一条日志。启动和部署方面的要求见后端扩展。
BaseAdapter API 参考
所有 adapter 都继承自 BaseAdapter（apps/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"}
限制并发请求
adapter 不需要自己做限流。网关已经限制了每个 provider key 同时在途的请求数，provider 返回 429 时还会调低这个上限；见出站并发。新 adapter 只要像内置 adapter 那样，把对上游的调用包在从 apps/backend/serving/adapters/upstream_limiter.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 协议格式转换消息的做法，可以参考
apps/backend/serving/adapters/gemini.py。本地部署。vLLM 和 SGLang 复用
apps/backend/serving/adapters/openai_compat.py。vllm和sglang这两个 kind 分发到同一个类；本地与远程的行为差异来自base_url和路由层，而不是来自专门的 adapter。
故障排查
响应格式错误
确保
format_response()返回 OpenAI 兼容的结构。校验
UsageInfo的各字段都是整数。检查
finish_reason是stop、length、content_filter、tool_calls之一。流式场景下，第一个非空内容分片一可用就立刻发出，这样首 token 时延（time-to-first-token）才会被准确记录。
流式问题
确保分片是 SSE 格式：
data: {json}\n\n。在
data: [DONE]之前发送最后那个 usage 分片。妥善处理 JSON 解析错误。
最佳实践
错误处理——使用
self.http.json_post_with_retry()；provider 出错时，要给出有用的错误信息。用量记账——优先采用 provider 上报的 usage；拿不到再回退到
estimate_prompt_tokens()/estimate_text_tokens()。流式辅助函数——用
format_stream_chunk()、make_final_usage_chunk()和done_sentinel()保证 SSE 输出一致。类型安全——写完整的类型标注，并让请求/响应的结构与
apps/backend/serving/schemas.py保持一致。测试——流式和非流式两条路径都要测到，并用大 prompt 验证 token 数会被正确截断。
文档与风格——用英文写 Google 风格的 docstring；不要把 provider 专属的逻辑塞进共享代码。
密钥——在 YAML 里用
${ENV_VAR}，不要硬编码 key 或端点，取值放在.env中。