大模型 API 聚合平台：统一接口、路由与故障切换

多模型 API 网关怎么设计？ 以稳定的内部请求模型连接多个供应商，但不要把“统一 OpenAI 兼容接口”误解为所有能力完全相同。控制面发布租户策略、模型能力目录和路由快照；数据面在固定截止时间内鉴权、规范化、选路、调用、计量并返回可追踪结果。

设计要点： 请求链路固定为 client → tenant identity/policy → model catalog/capability → normalizer → router → provider adapter → provider。路由优先级必须是“合规/地域硬约束 → 能力 → 质量门槛 → SLA/健康 → 成本 → 受限 fallback”，后面的优化项不能推翻前面的硬条件。

多模型 API 网关与大模型 API 聚合平台的适用边界

形态	核心职责	适用边界
普通反向代理	TLS、转发、基础限流	单一协议、少量上游、不理解模型能力
AI API 网关	鉴权、规范化、模型路由、故障隔离、计量与观测	团队能维护适配器、目录与策略版本
大模型 API 聚合平台	网关加账号、控制台、账单、支持和组织治理	额外承担用户、结算、数据与运营责任

大模型 API 聚合平台 GitHub 项目可用于发现实现思路，但仓库界面不等于经过业务验收的平台。自建范围和控制/数据面拆分可继续看自建 AI API 中转站架构。

架构组件与完整请求链路

1. client 发送内部模型 ID、截止时间和能力要求。
2. tenant identity/policy 完成 API 网关鉴权、项目状态、配额、预算与数据策略检查。
3. model catalog/capability 给出版本化的模型 ID、地域、上下文、工具、流式和结构化输出能力。
4. normalizer 把统一接口转换为内部语义，拒绝无法安全映射的字段。
5. router 依硬约束和健康快照生成带原因的候选顺序。
6. provider adapter 转换认证、请求、SSE 事件、错误和用量字段。
7. provider 执行请求；返回路径记录每次尝试并规范化响应。

控制面负责租户、密钥、模型目录、策略和价格版本的审批、发布与回滚；数据面原子加载快照，避免每次请求依赖复杂后台查询。所有尝试共享用户请求 ID，但拥有独立 attempt ID。

多模型 API 路由的选型优先级与受限 fallback

1. 合规/地域硬约束： 先按租户批准的供应商、处理地域和数据规则过滤。
2. 能力： 上下文、模态、工具、结构化输出和流式能力必须满足。
3. 质量门槛： 业务评测集低于阈值的模型不进入候选。
4. SLA/健康： 排除熔断、容量不足或剩余截止时间不够的路径。
5. 成本： 只在合格候选中比较预期成本。
6. 受限 fallback： 仍满足全部硬约束且任务允许替换时才切换。

模型故障切换不能把请求送到未经批准的地域，也不能用缺少工具能力的模型“勉强成功”。流式响应已经向客户端发出内容后，通常不能透明换模型拼接；重试、熔断与半开探测的组合见API 中转站故障切换指南。

多模型统一接口：流式与协议兼容测试

测试面	统一层要求	失败应如何表现
字段	保留共同语义，供应商特有能力进入明确扩展	不支持时显式错误，不静默丢弃
流式	映射开始、增量、工具、用量、结束与取消事件	断流保留已发送状态和 attempt ID
工具/结构化输出	目录声明能力，适配器验证 schema 与参数	能力不足时在路由前拒绝
错误	归一认证、限流、超时、供应商失败，同时保留原始代码	明确 retryable 与否及剩余截止时间
计量	分别记录每次尝试与用户请求汇总	用量缺失进入待对账，不猜测填充

统一 OpenAI 兼容接口应通过真实请求证明，包括普通响应、SSE、取消、工具调用、未知字段、429、5xx 和用量，而不是只比较 URL 与 JSON 外形。

API 网关鉴权、限流熔断、计量与可观测性

鉴权在上游调用前验证租户、项目、模型权限和预算；限流同时覆盖入口、租户、密钥和上游，并使用有界队列。熔断器只改变候选健康状态，不能放宽地域、能力或质量门槛。一次请求如发生重试，应把每个 provider attempt 的状态、延迟、输入/输出/缓存用量和错误分别落账，再生成可对账汇总。

API 网关可观测性至少包含按租户、模型、供应商、地域拆分的吞吐、错误、延迟、饱和度与 fallback 比例；trace 为鉴权、排队、规范化、路由和每次上游调用建 span。日志默认排除 Authorization、Cookie、正文和工具参数，排障字段设计见API 中转站可观测性指南。

多模型 API 网关成本公式

expected_request_cost = Σ(route_probability_i × provider_cost_i) + retry_probability × retry_cost + gateway_overhead。route_probability_i 是合格候选 i 被选择的概率；provider_cost_i 是按该请求输入、输出、缓存等变量估算的供应商成本；retry_probability 是需额外尝试的概率；retry_cost 是一次重试的期望变量；gateway_overhead 包含网关计算、存储、网络和观测开销。

示例变量只填关系：交互请求可能有 route_probability_primary 与 route_probability_fallback，并按健康历史估计 retry_probability；所有单价和概率都从团队当前合同、实际流量与测试中读取，不在路由规则里硬编码文章金额。

可复现：本地多模型 API 网关路由与受限 fallback

下面的配置和标准库 fixture 只监听 127.0.0.1:18081，不连接真实供应商。它用请求头注入 primary 的 429/5xx，验证 fallback 只能在已批准地域和能力范围内发生，并覆盖统一接口与 SSE。

1. 写入路由配置并复制 .work

{
  "listen": "127.0.0.1:18081",
  "allowed_regions": ["cn"],
  "routes": {
    "lab-text": [
      {"name":"primary","region":"cn","capabilities":["chat","stream"],"healthy":true},
      {"name":"fallback","region":"cn","capabilities":["chat","stream"],"healthy":true}
    ]
  }
}

cp route.baseline.json route.work.json
export LAB_GATEWAY_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(24))')"
python local_gateway.py

2. 启动本地网关

#!/usr/bin/env python3
import json, os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
CFG = json.load(open("route.work.json", encoding="utf-8"))
KEY = os.environ["LAB_GATEWAY_KEY"]

class Handler(BaseHTTPRequestHandler):
    def out(self, status, value, route=None, ctype="application/json; charset=utf-8"):
        body = value.encode() if isinstance(value, str) else json.dumps(value, separators=(",", ":")).encode()
        self.send_response(status); self.send_header("content-type", ctype)
        if route: self.send_header("x-selected-route", route)
        self.send_header("content-length", str(len(body))); self.end_headers(); self.wfile.write(body)
    def do_POST(self):
        if self.path != "/v1/chat/completions": return self.out(404,{"error":{"type":"not_found"}})
        if self.headers.get("authorization") != f"Bearer {KEY}": return self.out(401,{"error":{"type":"invalid_api_key"}})
        try:
            n=int(self.headers.get("content-length","0")); req=json.loads(self.rfile.read(n))
        except Exception: return self.out(400,{"error":{"type":"invalid_json"}})
        if set(req)-{"model","messages","stream"}: return self.out(400,{"error":{"type":"unknown_field"}})
        model=req.get("model"); region=self.headers.get("x-lab-region","cn")
        if region not in CFG["allowed_regions"]: return self.out(403,{"error":{"type":"policy_no_route","region":region}})
        need="stream" if req.get("stream") is True else "chat"
        candidates=[x for x in CFG["routes"].get(model,[]) if x["region"]==region and need in x["capabilities"] and x["healthy"]]
        if not candidates: return self.out(422,{"error":{"type":"capability_no_route"}})
        simulated=int(self.headers.get("x-lab-primary-status","200"))
        allow_fallback=self.headers.get("x-lab-allow-fallback","true").lower()=="true"
        selected=candidates[0]
        if simulated in (429,500,502,503,504):
            if not allow_fallback or len(candidates)<2:
                return self.out(simulated,{"error":{"type":"upstream_failure","retryable":simulated in (429,500,502,503,504)}},selected["name"])
            selected=candidates[1]
        elif simulated != 200:
            return self.out(simulated,{"error":{"type":"non_retryable"}},selected["name"])
        if req.get("stream") is True:
            body='data: {"route":"%s","delta":"pong"}\n\ndata: [DONE]\n\n' % selected["name"]
            return self.out(200,body,selected["name"],"text/event-stream")
        return self.out(200,{"model":model,"route":selected["name"],"content":"pong","attempts":1 if selected["name"]=="primary" else 2},selected["name"])
    def log_message(self, fmt, *args): pass

ThreadingHTTPServer(("127.0.0.1",18081),Handler).serve_forever()

先将上面的 JSON 保存为 route.baseline.json，复制为 route.work.json，再把 Python 保存为 local_gateway.py。环境变量必须在启动进程和测试终端中一致。

3. 本地请求及成功/失败响应

curl -i http://127.0.0.1:18081/v1/chat/completions \
 -H "Authorization: Bearer $LAB_GATEWAY_KEY" -H "Content-Type: application/json" \
 -d '{"model":"lab-text","messages":[{"role":"user","content":"ping"}],"stream":false}'
curl -i http://localhost:18081/v1/chat/completions \
 -H "Authorization: Bearer $LAB_GATEWAY_KEY" -H "Content-Type: application/json" \
 -H "X-Lab-Primary-Status: 429" -d '{"model":"lab-text","messages":[{"role":"user","content":"ping"}]}'
curl -i http://127.0.0.1:18081/v1/chat/completions \
 -H "Authorization: Bearer $LAB_GATEWAY_KEY" -H "Content-Type: application/json" \
 -H "X-Lab-Region: us" -d '{"model":"lab-text","messages":[{"role":"user","content":"ping"}]}'
curl -N http://127.0.0.1:18081/v1/chat/completions \
 -H "Authorization: Bearer $LAB_GATEWAY_KEY" -H "Content-Type: application/json" \
 -d '{"model":"lab-text","messages":[{"role":"user","content":"ping"}],"stream":true}'

成功样例： primary 正常时返回 200、响应头 x-selected-route: primary 与 {"model":"lab-text","route":"primary","content":"pong","attempts":1}；注入 429 或 500 后返回 200、route 为 fallback、attempts 为 2。

失败样例： X-Lab-Region: us 返回 403 {"error":{"type":"policy_no_route","region":"us"}}；设置 X-Lab-Allow-Fallback: false 并注入 429 时返回 429，不能静默切换。

4. 自动验收 429、5xx、流式和受限 fallback

#!/usr/bin/env python3
import json, math, os, time, urllib.error, urllib.request
URL="http://127.0.0.1:18081/v1/chat/completions"; KEY=os.environ["LAB_GATEWAY_KEY"]
def call(extra=None, stream=False):
    payload={"model":"lab-text","messages":[{"role":"user","content":"ping"}],"stream":stream}
    headers={"content-type":"application/json","authorization":f"Bearer {KEY}",**(extra or {})}
    req=urllib.request.Request(URL,data=json.dumps(payload).encode(),method="POST",headers=headers); start=time.perf_counter()
    try:
        with urllib.request.urlopen(req,timeout=2) as r: return r.status,r.headers, r.read().decode(),(time.perf_counter()-start)*1000
    except urllib.error.HTTPError as e: return e.code,e.headers,e.read().decode(),(time.perf_counter()-start)*1000
cases=[]
def check(name, result, predicate):
    cases.append((name, bool(predicate(result)), round(result[3],2)))
def body(result):
    try: return json.loads(result[2])
    except json.JSONDecodeError: return {}
primary=call(); check("primary",primary,lambda x:x[0]==200 and x[1].get("x-selected-route")=="primary" and body(x).get("attempts")==1)
fallback_429=call({"x-lab-primary-status":"429"}); check("429_fallback",fallback_429,lambda x:x[0]==200 and x[1].get("x-selected-route")=="fallback" and body(x).get("attempts")==2)
fallback_5xx=call({"x-lab-primary-status":"500"}); check("5xx_fallback",fallback_5xx,lambda x:x[0]==200 and x[1].get("x-selected-route")=="fallback" and body(x).get("attempts")==2)
r=call({"x-lab-primary-status":"429","x-lab-allow-fallback":"false"}); check("fallback_disabled",r,lambda x:x[0]==429 and x[1].get("x-selected-route")=="primary")
r=call({"x-lab-region":"us"}); check("region_denied",r,lambda x:x[0]==403 and "policy_no_route" in x[2])
r=call(stream=True); check("stream",r,lambda x:x[0]==200 and x[1].get_content_type()=="text/event-stream" and "[DONE]" in x[2])
passed=sum(ok for _,ok,_ in cases); p95=sorted(ms for _,_,ms in cases)[math.ceil(len(cases)*.95)-1]
attempts=[(body(primary).get("attempts"),1),(body(fallback_429).get("attempts"),2),(body(fallback_5xx).get("attempts"),2)]
attempt_rate=sum(recorded==actual for recorded,actual in attempts)/len(attempts)
report={"cases":cases,"pass_rate":passed/len(cases),"fallback_boundary_violations":0 if cases[3][1] and cases[4][1] else 1,"attempt_accounting_rate":attempt_rate,"local_p95_ms":p95}
print(json.dumps(report,ensure_ascii=False)); raise SystemExit(0 if report["pass_rate"]==1 and report["fallback_boundary_violations"]==0 and attempt_rate==1 and p95<250 else 1)

python validate_gateway.py

参数边界、拒绝行为与错误/失败模式

条件	边界	预期行为
身份	Bearer 必须匹配 LAB_GATEWAY_KEY	401，路由和供应商均不执行
模型/能力	模型必须在目录；stream 要求候选声明 stream	422 capability_no_route，不降级能力
地域	只允许配置中的 cn	其他地域 403，fallback 不得绕过
429/500/502/503/504	仅 allow_fallback=true 且存在合格候选	切到 fallback 并记录 2 次尝试
400/401/403 等	视为不可重试	原样失败，不调用 fallback
SSE 已发出	开始输出后不允许透明换模型	保留断流与 attempt，客户端决定续试
配置损坏/候选全不健康	启动校验失败或无候选	拒绝发布/返回明确 no_route，不随机选路

可计算验证指标与阈值

• route_contract_pass_rate = passed / 6，阈值为 1.00。
• fallback_boundary_violations：地域拒绝和禁用 fallback 两项中出现越界的次数，阈值为 0。
• local_p95_ms：六个 localhost 合同请求的 p95，fixture 验收阈值为 <250 ms；真实网关应另减去直连 provider 基线。
• attempt_accounting_rate = recorded_attempts / actual_attempts，阈值为 1.00，429/5xx fallback 样例必须记录 2。

逐步故障排查

1. 确认 route.work.json 可被 JSON 解析，且进程只监听 127.0.0.1:18081。
2. 若 401，比较启动进程与测试终端的环境变量；禁止把密钥打印到日志。
3. 若无路由，按顺序打印策略版本、region、所需 capability、健康快照和每个候选的淘汰原因。
4. 若 429/5xx 没有 fallback，确认状态属于允许集合、fallback 开关为 true、第二候选满足相同地域与能力。
5. 若发生越界 fallback，立即停止流量并检查过滤是在排序前执行，而不是失败后绕开。
6. 若 SSE 卡住，用 curl -N 检查 content-type、空行、[DONE]、缓冲和取消传播。

明确回滚 runbook

# 1) 停止 localhost 网关（前台运行时 Ctrl-C），保持入口无流量
# 2) 原 baseline 不动，用上一次通过验收的配置覆盖 .work
cp route.baseline.json route.work.json
# 3) 重新启动固定版本
python local_gateway.py
# 4) 另一个终端运行 6 项验收；必须 pass_rate=1、越界=0、attempt_accounting_rate=1、p95<250ms
python validate_gateway.py
# 5) 未达阈值则再次停机，不启用 fallback，不扩大地域，并保留配置 diff 与测试报告

部署与运维检查表

• 版本化模型目录、租户策略、供应商适配器和价格表，支持审批与一键回滚。
• 以密钥管理系统保存上游凭据；管理面与数据面分离鉴权并审计变更。
• 压测普通/SSE 流量、队列上限、限流、熔断、半开、取消和优雅停机。
• 演练供应商 429/5xx/断流、目录错误、策略回滚、用量缺失与账单重算。
• 配置 SLO、容量告警、fallback 比例告警、备份恢复和适配器升级责任人。

多供应商数据与合规风险排查

先把租户允许的供应商、地域和数据类别变成机器可执行硬约束，再核对各云/模型的最新官方控制说明。AWS 的模型与区域可用性见 Bedrock region compatibility；Azure 的数据处理说明见 Azure OpenAI data, privacy, and security；OpenAI 的端点数据控制见 OpenAI data controls；Vertex AI 的安全控制见 Vertex AI security controls。

排查面	问题	验证记录
地域/供应商	请求和 fallback 是否始终落在租户批准范围？	策略版本、路由原因、provider/region 与拒绝日志
数据流	正文、附件、工具参数进入哪些日志、缓存、追踪和支持流程？	字段级数据流、保留/删除测试、脱敏样本
访问控制	调用面、管理面、值班与供应商密钥权限是否分离？	角色矩阵、短期授权和变更审计
变更	官方控制、模型地域或默认行为变化后谁复核？	文档复核日期、策略变更单与回归结果

官方页面描述的是各平台当前控制边界，不能自动变成团队的法律结论；需要结合具体租户、地域、数据与专业意见评估。

网关实现与限流的官方资料入口

• LiteLLM 官方仓库
• Apache APISIX AI Proxy 官方文档
• Envoy Gateway 限流概念
• Kong AI Gateway 官方文档

具体模型的接入示例见 DeepSeek API 中转站接入 与 Gemini API 中转接入；自建网关的源码比较见 开源 API 网关选型。

多模型 API 网关常见问题

多模型 API 网关怎么设计请求链路？

固定为 client → tenant identity/policy → model catalog/capability → normalizer → router → provider adapter → provider；返回路径再完成流式事件、错误和用量规范化，并记录请求与每次尝试。

多模型 API 路由应该只按成本选择吗？

不应该。先满足合规和地域硬约束、能力、质量门槛及 SLA/健康，再比较成本；fallback 也只能进入仍满足全部硬约束的候选。

多模型统一接口能否隐藏所有供应商差异？

不能。统一层可规范共同语义，但工具、结构化输出、流式事件和错误仍可能不同；无法映射的字段应显式拒绝或放入版本化扩展，并用回归测试验证。

模型故障切换为什么必须受限？

不同模型可能处于不同地域，且能力、质量和数据控制不同。切换前必须重新检查原始策略；流式内容已经发出后，也不能把另一模型的结果静默拼接。

API 网关可观测性至少记录什么？

至少记录请求 ID、attempt ID、租户、内部模型、供应商、地域、路由原因、策略版本、状态、延迟、用量和错误分类；敏感正文与凭据默认不进入日志或追踪。

多模型 API 网关如何验证流式响应？

用本地 SSE fixture 固定事件序列，检查首事件时延、分片顺序、工具调用 ID、终止标记、客户端取消和用量事件。对每个 Provider 适配器单独回归，不能用非流式成功推断流式兼容。

多模型 API 网关出现 429 或 5xx 怎么排查？

先用根 request ID 关联租户限流、路由决策、全部上游 attempt 和熔断状态；区分网关 429 与上游 429，再核对重试次数、Retry-After、区域约束和备用路由健康，禁止无限重试。

多模型 API 网关如何安全回滚路由配置？

停止继续灰度，保留失败请求与指标快照，把工作配置恢复为上一已验证副本并重启固定版本；随后重跑鉴权、429、5xx、流式和受限 fallback 用例，指标恢复基线后再逐步放量。