添加新模型
本指南说明如何向 HybridInference 网关添加 LLM 模型与 provider,同时保持客户端看到的 OpenAI 兼容 API 接口不变。
两种需求共用这一份指南。根据你的情况,选择其中一条路径:
复用已有的 provider adapter——只需改注册表 YAML 和环境变量。
接入新的 provider——先添加或接上 adapter,再改注册表 YAML 和环境变量。
如果模型是你自己用 vLLM、SGLang 或 Ollama 提供服务的,见添加新的本地模型——本地服务器的路由注册由那一篇负责。
模型注册表在哪里
本仓库里没有 config/models.yaml。后端在启动时于 apps/backend/serving/config/distribution.py(resolve_config_path("models"))解析注册表路径,优先级如下:
MODELS_CONFIG_PATH(旧别名MODELS_CONFIG)。显式给出的路径始终优先。distribution manifest 的
paths.models——仅当DISTRIBUTION_CONFIG_PATH指向一份 manifest 且DISTRIBUTION_CONFIG_MODE=active时才使用。manifest 里的相对路径按 manifest 自身所在目录解析。DISTRIBUTION_CONFIG_MODE默认是dark:manifest 会被加载、校验并写进日志以供比对,但不会真正生效。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.yamlmanifest 和一份config/models.yaml,然后设置DISTRIBUTION_CONFIG_PATH=distributions/<name>/distribution.yaml和DISTRIBUTION_CONFIG_MODE=active。distributions/example/是一个可以直接复制的可用 overlay;config/examples/distribution.example.yaml是一份带注释的 manifest。
本指南中所说的「你的模型注册表」,指的就是上述解析过程最终选中的那个文件。
如果解析出的路径上没有注册表,后端会记录一条错误日志,/v1/models 返回空列表,并把每个模型都报为 not found。
快速开始:为已有 provider 添加模型
如果该 provider 已经有 adapter,只需要改配置。
在模型注册表中添加一条模型条目:
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_url 和 api_key,所以它只需要写出不同的部分。当第二条路由指向别处时,再在那条路由上重写它们。
在仓库根目录的
.env里设置环境变量(后端的配置加载器读的就是这个文件):
PROVIDER_BASE_URL=https://api.provider.example/v1
PROVIDER_API_KEY=your-api-key
重启后端以加载新模型。
验证(见第 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)
deepseek、kimi 和 minimax 现在就是这样接入的:apps/backend/serving/adapters/profiles.py 里每个 provider 一份的 profile 承载用量指标或路径上的差异,其余全部交给 OpenAICompatAdapter。zai 和 kimi_coding 用的是同一套 profile 机制,但它们额外绑定了一个 coding 工具身份,所以 _make_adapter 会在走到那个元组之前,先把它们短路到 CodingIdentityAdapter(OpenAICompatAdapter 的一个很薄的子类)。
只有当 provider 说的确实不是 OpenAI 的 wire 格式时,才写专门的 adapter——例如 Gemini 的 generateContent、Anthropic Messages API、OpenRouter 用来锁定 sub-provider 的 body 字段。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 步:通过网关验证
启动后端。在一个检出里,最快的启动形式是:
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
}'
配置参考
模型字段
这些键从模型条目里读出并传给 ModelConfig(apps/backend/serving/adapters/base.py)。其中大多数也可以设在单条路由上,此时路由上的值优先;id 和 name 标识的是模型本身,只在模型层级读取。
字段 |
类型 |
必填 |
说明 |
|---|---|---|---|
|
string |
是 |
唯一的模型标识;客户端发送的就是这个名字 |
|
string |
是 |
展示名称 |
|
string |
是 |
provider/adapter 的种类;在没有给出 |
|
string |
是 |
API 端点的 base URL |
|
string |
否 |
API 鉴权密钥 |
|
string |
否 |
provider 侧的模型标识(在 wire 上覆盖 |
|
string |
否 |
|
|
list[string] |
否 |
用于路由的备用名称 |
|
string |
否 |
量化格式(默认: |
|
list[string] |
否 |
输入类型: |
|
list[string] |
否 |
输出类型: |
|
int |
否 |
最大上下文窗口(默认:8192) |
|
int |
否 |
最大输出 token 数(默认:4096); |
|
bool |
否 |
是否支持 function calling(默认:false) |
|
bool |
否 |
是否支持 JSON mode(默认:false) |
|
list[string] |
否 |
允许的参数名(默认: |
|
list[string] |
否 |
这个模型的 |
|
bool |
否 |
模型在共享 GPU 上按需加载(首次请求时启动,空闲时停止)。会在 |
|
string |
否 |
为 |
|
dict |
否 |
合并进 OpenAI 兼容上游请求体的默认字段。核心字段和已校验的客户端参数优先 |
|
bool |
否 |
该端点上的 sglang 服务器是带 |
|
dict |
否 |
供路由策略消费的、每条路由上的自由格式元数据 |
|
dict |
否 |
基础成本信息。 |
|
dict |
否 |
只按 UTC 计的生效时间,加上每天重复的价格时间窗: |
模型条目里有几个键并不是 ModelConfig 的字段,而是在别处被消费:route(见下)、router 和 router_params(按模型选择 router)、admin_only、required_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}
除上面的模型字段之外,路由条目还接受:
字段 |
类型 |
说明 |
|---|---|---|
|
string |
adapter 的 kind(见下)。默认取模型的 |
|
float |
流量的相对占比(默认 1.0)。 |
|
list[string] |
这个端点的 key 池,用来替代 |
|
bool |
当由 |
|
string |
只覆盖分析统计里的标签——它重命名的是仪表盘里的那一行,不会选择 adapter,选 adapter 的是 |
|
string |
RouteWise 的成本类别: |
|
— |
RouteWise 的资源池与预算元数据 |
一条路由用自己的 kind 作为 provider 标签上报——这个值记录在 api_logs.provider 里,所有按 provider 划分的管理视图都以它分组。因此同一个 kind 的两条路由会共用仪表盘上的同一行;用 provider: 才能把它们分开。
支持的 adapter kind
每条路由条目里的 kind 字段决定用哪个后端 adapter。所有标为 OpenAI-compat 的 kind 共用同一份 OpenAICompatAdapter 实现,并自动套用各 provider 专属的 profile。
Kind |
类别 |
备注 |
|---|---|---|
|
OpenAI-compat |
通用的 OpenAI 兼容端点;没有更贴切的 kind 时用它 |
|
OpenAI-compat |
|
|
OpenAI-compat |
本地 vLLM 推理服务器 |
|
OpenAI-compat |
本地 SGLang 推理服务器 |
|
OpenAI-compat |
本地或远程的 Ollama 服务器 |
|
OpenAI-compat |
Chutes.ai 托管推理 |
|
OpenAI-compat |
Featherless.ai 托管推理 |
|
OpenAI-compat |
面向 OpenAI 兼容模型的 CLI 代理端点 |
|
OpenAI-compat |
DeepSeek API(套用 DeepSeek 用量 profile) |
|
OpenAI-compat |
Moonshot/Kimi 按 token 计费的 API(套用 Kimi 用量 profile) |
|
OpenAI-compat |
Kimi coding 套餐端点;分发到 |
|
OpenAI-compat |
Z.AI GLM coding 套餐:chat 路径是在 base URL 后面拼 |
|
OpenAI-compat |
MiniMax API(套用 MiniMax 用量 profile) |
|
自定义 |
OpenRouter 聚合器。用方括号形式 |
|
自定义 |
Google Gemini API(需要转换消息格式) |
|
自定义 |
经 Google Vertex 访问的 Anthropic Claude |
|
自定义 |
直连 Anthropic Messages API 的客户端 |
其他任何 kind 都会在加载注册表时抛出 ValueError: Unknown adapter kind。
混合路由
加权路由在注册时生效。之后,路由配置文件(通过 ROUTING_CONFIG_PATH / manifest 的 paths.routing 解析)可以经 RoutingManager 集中调整权重。见路由。
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"}
限流(按 provider 的限流并不存在)
没有按 provider 的限流器可以开启;本节记录的是它将来该放在哪里。apps/backend/serving/servers/bootstrap.py 并没有配置按 provider 的限流器。那里接线的唯一进程内限流器是 UserConcurrencyLimiter。鉴权流程的静态限流值在 apps/backend/serving/config/settings.py(signup_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.py。vllm和sglang这两个 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_url、api_key、api_keys和provider_model_id生效。
响应格式错误
确保
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中。
另见
添加新的本地模型——注册自建的 vLLM/SGLang/Ollama 服务器
快速开始——一次可运行的部署,从第一个请求到本地服务器
通过 OpenRouter 路由——架构与端点
路由——集中式权重覆盖与策略
配置——环境变量与 YAML 配置