在原型阶段，业务代码通常直接调用某个模型服务：读取一段环境变量中的密钥，拼接请求体，等待响应，然后把结果返回给前端。这样的代码可以很快跑通，但一旦进入多人协作或生产环境，问题会集中出现：模型供应商更换时需要修改业务代码；不同接口的字段略有差异；请求超时会占用工作线程；日志可能意外记录密钥或用户输入；Docker 容器虽然能启动，却没有健康检查、非 root 用户和稳定的配置边界。
多模型网关的价值不在于简单增加一个转发地址，而在于把“业务请求”和“模型供应商协议”分开。业务侧只依赖一份内部契约，适配层负责模型名映射、认证头、超时、错误分类和审计字段。本文用一个小型 FastAPI 服务演示这条边界，并通过 Dockerfile 固定运行环境。
本文采用的是 OpenAI 兼容请求形式作为示例。兼容并不等于所有服务的行为完全一致，实际使用时仍需依据目标服务当前文档核对路径、请求字段、流式协议、模型名称和数据处理条款。
适配层的基本原理
一次请求可以拆成四层：
- 业务契约：接收
model、messages、temperature等受控字段。 - 路由决策：根据配置把逻辑模型名映射到目标模型和上游地址。
- 传输执行：添加认证头，设置连接与读取超时，处理 HTTP 状态码。
- 审计输出：记录请求 ID、逻辑模型、耗时和结果状态，不记录完整密钥与敏感正文。
这里要特别区分“重试”和“换模型”。网络连接失败、部分 5xx 错误有时适合有限重试；参数错误、认证失败和内容策略拒绝不应盲目重试。模型切换也应该通过明确的路由规则完成，而不是在异常后无限尝试多个供应商，否则会放大费用、延迟和合规风险。
下面的实现只保留最小功能：单路由、超时、健康检查和脱敏日志。它适合说明边界，不代表一个完整的生产网关。
实现一个最小适配服务
项目目录如下：
model-proxy/
├── app.py
├── requirements.txt
├── Dockerfile
└── .dockerignore
requirements.txt：
fastapi==0.116.1
uvicorn[standard]==0.35.0
httpx==0.28.1
版本号只是示例，部署前应结合团队的 Python 版本和安全扫描结果确认可用组合，并提交锁定文件或构建摘要。
app.py：
import os
import time
import uuid
from typing import Any
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
app = FastAPI()
UPSTREAM_URL = os.environ["UPSTREAM_URL"].rstrip("/")
UPSTREAM_API_KEY = os.environ["UPSTREAM_API_KEY"]
UPSTREAM_MODEL = os.getenv("UPSTREAM_MODEL", "default-model")
TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "60"))
class ChatRequest(BaseModel):
messages: list[dict[str, Any]] = Field(min_length=1)
model: str = "default"
temperature: float | None = Field(default=None, ge=0, le=2)
@app.get("/healthz")
async def healthz() -> dict[str, str]:
return {
"status": "ok"}
@app.post("/v1/chat/completions")
async def chat(request: ChatRequest) -> dict[str, Any]:
request_id = str(uuid.uuid4())
payload: dict[str, Any] = {
"model": UPSTREAM_MODEL,
"messages": request.messages,
}
if request.temperature is not None:
payload["temperature"] = request.temperature
started = time.monotonic()
headers = {
"Authorization": f"Bearer {UPSTREAM_API_KEY}",
"Content-Type": "application/json",
"X-Request-ID": request_id,
}
try:
async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS) as client:
response = await client.post(
f"{UPSTREAM_URL}/v1/chat/completions",
json=payload,
headers=headers,
)
except httpx.TimeoutException as exc:
raise HTTPException(status_code=504, detail="upstream timeout") from exc
except httpx.RequestError as exc:
raise HTTPException(status_code=502, detail="upstream unavailable") from exc
elapsed_ms = int((time.monotonic() - started) * 1000)
print(f"request_id={request_id} status={response.status_code} elapsed_ms={elapsed_ms}")
if response.status_code >= 400:
# 生产环境应按状态码和上游错误结构分类，并限制错误正文长度。
raise HTTPException(status_code=502, detail="upstream rejected request")
return response.json()
代码中的上游地址、密钥和模型名都来自环境变量。应用日志只输出请求 ID、状态码和耗时，不输出 Authorization、完整请求消息或上游响应。实际项目还应使用结构化日志，并根据数据分类要求决定是否记录消息摘要。
如果上游接口的认证方式、路径或响应结构不同，可以新增一个适配器函数，而不是让业务层到处判断供应商名称。例如：
def build_upstream_payload(request: ChatRequest) -> dict[str, Any]:
return {
"model": UPSTREAM_MODEL,
"messages": request.messages,
}
当适配器数量增加时，可按供应商建立模块，并让路由配置只负责选择适配器。这样修改某一家服务的字段时，影响范围更容易审查。
编写可审计的 Dockerfile
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt \
&& useradd --create-home --uid 10001 appuser
COPY app.py .
RUN chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/healthz')"
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
这个 Dockerfile 有几个关键点。基础镜像使用 slim 变体可以减少不必要的软件包，但不应把“镜像更小”直接等同于“绝对安全”；仍要进行漏洞扫描并定期更新。依赖安装与源代码复制分开，可以利用构建缓存。应用进程使用非 root 用户运行，降低服务被利用后的权限范围。HEALTHCHECK 只检查进程是否能响应，不代表上游模型服务可用，因此还需要在监控系统中区分本地存活、上游连通和业务成功率。
.dockerignore 至少应排除密钥和开发文件：
.git
.venv
__pycache__
*.pyc
.env
.env.*
构建与启动：
docker build --tag model-proxy:local .
docker run --rm -p 8000:8000 \
-e UPSTREAM_URL="https://api.example.invalid" \
-e UPSTREAM_API_KEY="$UPSTREAM_API_KEY" \
-e UPSTREAM_MODEL="your-model-name" \
-e UPSTREAM_TIMEOUT_SECONDS="60" \
model-proxy:local
不要把密钥写入 Dockerfile、镜像层、Shell 历史记录或仓库。生产环境应由编排平台的 Secret、云密钥服务或 CI/CD 的受保护变量注入，并限制读取权限。
如何接入第三方模型接口
当团队需要统一接入多个模型服务时，可以把 UPSTREAM_URL 和 UPSTREAM_MODEL 放到部署配置中，把业务请求固定在内部 /v1/chat/completions 契约上。以 HaerAPI 为例，是否能够直接使用该适配方式，取决于其当前是否提供相应的兼容接口、认证方式和模型能力，不能仅依据“兼容”字样推断全部字段都可用；应先用非敏感测试请求核对文档和返回结构。
接入时建议建立一张内部核对表：请求地址和路径、必填请求字段、模型列表、上下文限制、流式响应格式、错误码、超时建议、数据留存规则、地域和合规要求。对于涉及个人信息、源代码、合同或内部文档的请求，先完成数据分级和脱敏，再决定是否发送到外部服务。
常见问题
为什么健康检查通过，但请求仍然失败？
/healthz 只证明本地进程工作。上游地址错误、DNS 失败、证书问题、密钥失效、模型名称不可用，都可能只在真实请求时暴露。应分别监控进程存活、上游连接成功率、HTTP 状态码、首字节延迟和完整请求耗时。
是否应该给所有失败请求自动重试？
不应该。超时和部分临时性 5xx 可以在有限次数、指数退避和总截止时间内重试；认证失败、参数错误和明确的业务拒绝应立即返回。对非幂等操作尤其要谨慎，避免一次业务动作被执行多次。
为什么不在网关里直接记录完整请求和响应？
模型请求往往包含用户输入、源代码或业务文档。完整记录会增加泄露面，也可能违反组织的数据保留要求。更稳妥的做法是记录请求 ID、逻辑模型、状态、耗时、令牌用量字段（若上游提供且允许记录）和经过规则化处理的错误类别；调试样本应经过授权、脱敏和保留期限控制。
多模型路由应该按什么条件切换？
先定义可验证的规则，例如任务类型、预算上限、延迟目标或人工指定的模型档位。不要只依据一次异常随机切换。路由规则应可配置、可审计，并在变更前用固定样例验证输出格式、工具调用和安全约束。
总结
模型接入的工程难点不只是发出一次 HTTP 请求，而是建立稳定、可替换和可追踪的边界。业务契约隔离供应商差异，超时与错误分类控制故障传播，脱敏日志降低数据暴露风险，Dockerfile 则把依赖、用户权限和健康检查固化为可审查的构建规则。
在实际落地时，先用单一上游完成契约和监控，再增加模型路由；先明确哪些数据允许外发，再评估服务选型。任何第三方 API 的接入都应以当前文档、实际验证结果和组织合规要求为准。
