月均200万次模型调用的团队，如何做到一家AI厂商挂了业务无感
上周二凌晨 2:47，我被 PagerDuty 炸醒了。
团队的 AI 服务挂了三次。告警信息出奇一致——“上游模型响应超时”。先是我 OpenAI 的 GPT-4 抽风，刚切过去，Claude 那边又返回 529。我躺在床上盯着告警列表往上翻，突然意识到一个很尴尬的事实：我们接了五家 AI 厂商，但路由逻辑全写死在代码里。哪个挂了就得手动切，跟接水管似的，拧开这个关上那个。
这不是我们一家的问题。
上个月跟几个做 AI 应用的哥们喝酒，发现大家都在被同样的破事折磨。有个做电商客服的团队，双十一当天 OpenAI 挂了 40 分钟——注意，是 2024 年的双十一，那会儿 GPT-4 Turbo 刚大规模推送没多久，稳定性特别飘。他们愣是手动改了 DNS 才切到备用厂商。我问他们损失了多少订单，对方闷了一口酒，说“不敢算”。
所以今天想聊聊，怎么用 API Gateway 的思路，把一堆 AI 厂商的接口统一管起来，顺带把健康检查和自动切换的问题也解决了。这玩意儿我前后踩了三个月的坑，从去年 10 月折腾到今年 1 月，希望你能少走点弯路。
为什么你需要一个 AI API Gateway？
先看几个数据点，你感受一下：
- 我们团队现在接了 5 个厂商（OpenAI、Anthropic、Google AI、阿里百炼、DeepSeek），平均每天处理大概 200 万次模型调用。峰值的时候能到 350 万。
- 根据我们内部的监控数据，单一厂商的月可用性大概在 99.5% 左右。听着还行对吧？但如果你同时用三家，**至少一家出问题的概率会飙升到每个月 3-4 次**。这不是数学题，这是我们的真实统计。
- 2024 年 11 月 OpenAI 那次大规模故障持续了 3 小时 12 分钟，期间我们的自动切换机制帮我们兜住了大概 12 万次请求。如果没有这套东西，那天晚上我大概得在机房过夜。
说白了，单一依赖任何一家 AI 厂商都是在赌博。但问题来了——每家厂商的 API 格式、认证方式、错误码都不一样。OpenAI 返回的是 choices[0].message.content，Anthropic 是 content[0].text，Google AI 又有一套自己的结构。你怎么把它们揉到一起，让业务侧无感？
核心架构：统一路由层
先看整体架构。我用的是最常见的 Gateway 模式，没什么黑魔法：
客户端请求
│
▼
┌─────────────────┐
│ 统一 API 接口 │ ← 对外暴露标准 OpenAI 格式
└─────────────────┘
│
▼
┌─────────────────┐
│ 路由决策引擎 │ ← 根据模型名、负载、健康状态选厂商
└─────────────────┘
│
├──────► OpenAI Adapter
├──────► Anthropic Adapter
├──────► Google AI Adapter
├──────► 阿里百炼 Adapter
└──────► DeepSeek Adapter关键设计原则就一条：对外统一，对内适配。
翻译成人话就是：你的业务代码永远只调一个接口，Gateway 负责把请求翻译成各家厂商的格式，再把响应翻译回来。业务侧不需要知道底层到底用的是 GPT-4 还是 DeepSeek，它只看到一个标准的 OpenAI 格式响应。
等等，这里我要更正一下。我刚才说“标准 OpenAI 格式”，其实不太准确。OpenAI 自己也在改格式，2024 年他们就改了三次。我说的“标准”是指我们内部约定的一套格式规范，以 OpenAI 2024 年 6 月那个版本为基准，然后所有厂商的响应都往这个格式上转。
我最早犯的错误是试图让所有厂商的响应格式完全一致。结果适配层写得巨复杂，每个字段都要映射、转换、补默认值，代码跟意大利面一样。后来学乖了——只保证核心字段统一（content、finish_reason、token_usage），其他厂商特有的字段原样放在一个 metadata 字段里，业务侧按需取用。
省了不知道多少适配代码。
路由策略：不只是轮询那么简单
路由这块是我踩坑最多的地方。最开始我用了最简单的轮询，心想反正模型能力都差不多，均匀分配就行了。
天真了。
问题一：成本差异巨大。 GPT-4 的价格是 DeepSeek 的 20 多倍，你让它们平均分配流量？财务看了想打人。我们 12 月的账单出来的时候，CTO 专门找我聊了十分钟。
问题二：模型能力不对齐。 有次我把一个需要 JSON 结构化输出的请求路由到了某个厂商——具体哪个我就不点名了——结果它返回的 JSON 格式各种飘，字段名一会儿驼峰一会儿下划线，下游解析全挂了。那次故障处理了四个小时。
所以现在的路由策略分了三层：
第一层：模型映射。 维护一个模型能力矩阵，标明每个厂商支持哪些模型、各自的特长是什么。大概长这样：
model_mapping:
gpt-4:
providers:
- name: openai
model: gpt-4-0125-preview
priority: 1
cost_weight: 1.0
- name: deepseek
model: deepseek-chat
priority: 2
cost_weight: 0.05 # 成本是 GPT-4 的 1/20
capabilities:
- json_mode
- function_calling
- vision第二层：成本感知路由。 对于非关键任务，优先走便宜的厂商。我们设置了一个 cost_tolerance 参数，业务侧可以指定这个请求对成本的敏感度。打摘要、生成初稿这些，走低价通道；核心对话、结构化输出这些，走高价通道。
第三层：健康状态路由。 最后一道防线。如果一个厂商挂了或者延迟飙升，自动把它从候选列表里踢掉。
实际代码大概长这样（简化版，去掉了公司内部的配置加载逻辑）：
def route_request(request, model_name):
candidates = get_candidates(model_name)
# 过滤掉不健康的厂商
healthy = [c for c in candidates if health_checker.is_healthy(c.provider)]
if not healthy:
raise AllProvidersDownError(f"All providers for {model_name} are unhealthy")
# 按优先级和成本排序
sorted_candidates = sorted(
healthy,
key=lambda c: (c.priority, c.cost_weight * request.cost_tolerance)
)
return sorted_candidates[0]健康检查机制：别只 Ping 一下
健康检查是整个系统里最容易被低估的部分。
我最开始的做法特别 naive——每隔 30 秒调一次 /v1/models 接口，能返回 200 就算健康。觉得自己很聪明。
然后就被现实教育了。
2024 年 12 月有一次，OpenAI 的 API 能正常返回模型列表，但实际调用 chat completions 的时候疯狂超时。我们的健康检查完全没发现，流量还是往里打，错误率飙到 40%。等我发现的时候，已经过去了 23 分钟。
现在的健康检查分三个层次：
L1 - 连通性检查（每 10 秒）： 调一次轻量接口，确认服务可达。这个层面只看 HTTP 状态码和响应时间。用的是 /v1/models，超时设的 2 秒。
L2 - 功能检查（每 60 秒）： 发一个真实的推理请求，用固定的 prompt——“回复 OK，不要有任何其他内容”——验证模型确实在工作。这个能发现“接口通但模型挂了”的情况。那次 OpenAI 事故就是这种。
L3 - 质量检查（每 5 分钟）： 发一组标准测试用例，检查响应质量是否下降。主要是防止模型降级或者输出异常。这个想法来自我们的一次事故——某厂商悄悄把模型版本换了，API 格式没变，但输出质量明显下降，用户投诉说“AI 变傻了”。我们花了两天才定位到问题。
嗯...这个 L3 检查其实不太好实现。怎么定义“质量下降”本身就挺主观的。我们现在用的是对比上一个版本的输出相似度，超过阈值就告警，但误报率还挺高的。这块我还在调。
健康状态的判断也不是简单的“健康/不健康”二元状态。我用了一个滑动窗口的打分机制：
class HealthScorer:
def __init__(self, window_size=60, threshold=0.8):
self.window_size = window_size # 60秒窗口
self.threshold = threshold
self.checks = [] # 存储最近60秒的检查结果
def record_check(self, success, latency_ms):
self.checks.append({
'success': success,
'latency': latency_ms,
'timestamp': time.time()
})
# 清理过期记录
self.checks = [c for c in self.checks
if time.time() - c['timestamp'] < self.window_size]
def get_score(self):
if not self.checks:
return 0
recent = self.checks[-10:] # 最近10次检查
success_rate = sum(1 for c in recent if c['success']) / len(recent)
avg_latency = sum(c['latency'] for c in recent if c['success']) / max(1, len(recent))
# 延迟超过5秒扣分
latency_penalty = max(0, (avg_latency - 5000) / 5000)
return success_rate * (1 - latency_penalty * 0.3)
def is_healthy(self):
return self.get_score() >= self.threshold这个打分机制的好处是，偶尔一次超时不会直接把厂商标记为不健康，但持续的问题会快速拉低分数。我们设的阈值是 0.8，低于这个值就触发切换。
实际踩坑记录
说几个让我印象深刻的坑。都是真金白银换来的教训。
坑一：速率限制的乒乓效应。
2024 年 11 月有一次，我们把 OpenAI 的流量切到 DeepSeek，结果 DeepSeek 那边有速率限制（每分钟 1000 次），瞬间被打爆，返回 429。然后健康检查发现它“不健康”，又把流量切回 OpenAI——但 OpenAI 当时也在限流。两个厂商来回切，形成了“乒乓效应”。那次故障持续了 17 分钟，两个厂商的 429 响应加起来有 8000 多次。
解决方案：在健康检查里加入 429 的特殊处理。收到 429 不直接标记不健康，而是根据 Retry-After 头降级处理，同时触发告警让人工介入。这个逻辑救了我们至少三次。
坑二：流式响应的适配地狱。
各家厂商的 SSE 格式都不一样。OpenAI 是 data: {"choices":[{"delta":{"content":"hello"}}]}，Anthropic 是 data: {"type":"content_block_delta","delta":{"text":"hello"}}。我写适配器的时候，光流式响应就调了三天。
中间还发现一个很恶心的问题：Google AI 的流式响应在连接断开后不会主动关闭，导致我们的连接池被占满。最后加了 30 秒的强制超时才解决。
建议直接用现成的方案。LiteLLM 的流式处理比我自己写的稳多了，我是后来才把这块重构了。
坑三：成本统计对不齐。
统一网关之后，财务问我“上个月 AI 花了多少钱”，我按各厂商的账单加了一下，发现比网关统计的多出 15%。排查了整整两天，发现是有些请求在网关层超时了，但厂商那边已经消耗了 token——等于我们付了钱但没拿到结果。
现在我们在网关层做了严格的超时控制和幂等处理。超时时间设的 30 秒，比厂商默认的 60 秒短，尽量避免“付费但无响应”的情况。
要不要自己造轮子？
这个问题我被问了好多次。我的建议是：
- **如果你只用 2-3 家厂商，业务量不大**，直接用 LiteLLM 或者 One API 这种开源方案就够了。别自己写。真的。
- **如果你需要深度定制路由策略、成本控制、审计日志**，可以考虑自研。但要做好投入 2-3 个人月的心理准备，我说的是至少。
- **如果你是大厂，日调用量过千万**，那基本得自研了。开源方案在那个量级下性能和稳定性都够呛——我不是说它们不好，是那个量级确实不是它们设计目标。
我们团队现在是自研 + 部分借鉴 LiteLLM 的适配器代码，算是折中方案。核心路由和健康检查是自己写的，厂商适配层参考了 LiteLLM 的实现，省了不少时间。
后续想折腾的方向
目前这套系统还有一些我想优化但还没动手的地方：
1. 智能预热：检测到某个厂商健康度下降时，提前给备用厂商发 warm-up 请求，减少切换时的冷启动延迟。我们现在切换后第一次调用平均延迟会比正常高 200-300ms，就是因为连接池还是冷的。
2. 成本预测：根据历史数据预测下个月的成本，提前做预算控制。财务问过我能不能给个预估，我现在只能拍脑袋说“大概跟这个月差不多”。
3. 多区域部署：同一个厂商在不同区域部署多个接入点。我们现在 OpenAI 只走了美国节点，延迟大概 180ms。如果能加上欧洲和新加坡的节点，理论上能压到 60-80ms。不过这个优先级不高，因为当前延迟业务侧还能接受。
如果你也在折腾类似的东西，或者有更好的实践，欢迎在评论区聊聊。我特别想知道你们是怎么做成本控制的——这块我感觉还有很大的优化空间。我们现在只是一个简单的优先级+权重，感觉不够精细。有没有用强化学习做动态路由的？效果怎么样？
标签： #AI网关 #API网关 #多厂商路由 #健康检查 #LLMOps #成本优化
读者评论 3