LLM 网关的优雅降级设计:5 层 fallback chain 工程实践

一个 demo 现场,主模型返回 502。前端转着圈,台下投资人盯着屏幕。这种事,靠重试解决不了——它需要的是完整的降级链路

线上接 LLM 的服务,最容易被忽视的不是 prompt,不是上下文管理,是故障传播路径

模型厂商的 SLA 都写得漂亮,“99.5% 可用”。但只要把 OpenAI、Anthropic、Google、DeepSeek 几家任意 30 天的事故公告拉出来对一遍,会发现一个朴素事实:单一上游的真实可用率,在某些时段连 95% 都打不到。把一个 LLM 应用的稳定性押在任何一家厂商上,等于把生死线交给别人。

本文按生产经验,把 LLM 网关的降级能力拆成 5 层,从最基础的指数退避,到最少人讨论的语义路由,逐层给出可直接落地的代码踩坑边界


整体架构

降级不是一个开关,是一条链。一次请求按顺序穿过 5 层,任何一层兜住就返回,全部失败才向调用方报错。

层级 名称 触发条件 响应时间影响 实现复杂度
L1 指数退避重试 5xx / 网络抖动 +200ms ~ 2s
L2 同 provider 模型回退 主模型 429 / 限流 +50ms
L3 跨 provider 模型回退 主 provider 整体不可用 +100ms ~ 800ms
L4 能力路由 模型不支持请求的功能 0(路由前判断)
L5 语义路由 任务类型与模型不匹配 0(路由前判断)

下面逐层拆。代码用 OpenAI Python SDK 演示,所有 endpoint 都假定走 OpenAI 兼容接口(自建网关或 TheRouter 这类多 provider 网关都适配)。


L1:指数退避 + jitter

最基础也最容易写错的一层。生产代码里见过的两种典型错误:

  1. 固定 sleeptime.sleep(1) 然后立刻重试。100 个客户端同步重试 = 雪崩。
  2. 无 jittersleep(2 ** retry)。看起来"指数退避"了,但所有重试在同一时刻撞过去。

正确写法:

import time
import random
from openai import OpenAI, APIError, RateLimitError, APIConnectionError

RETRYABLE = (APIConnectionError, APIError)  # 5xx 和网络错
NON_RETRYABLE = (RateLimitError,)  # 429 走 L2,不在 L1 重试

def with_retry(call, max_attempts=4, base=0.5, cap=8.0):
    for attempt in range(max_attempts):
        try:
            return call()
        except NON_RETRYABLE:
            raise  # 立刻抛,让上层走 L2
        except RETRYABLE as e:
            if attempt == max_attempts - 1:
                raise
            # 指数退避 + 全 jitter
            delay = min(cap, base * (2 ** attempt))
            sleep_for = random.uniform(0, delay)
            time.sleep(sleep_for)

关键点:

  • 区分可重试与不可重试:429 在 L1 重试是浪费配额,应该立刻往下走。
  • 全 jitter 比"等比 jitter"在并发场景下方差更小,AWS 那篇经典文章里有数学证明。
  • 重试上限给 4 次:第 5 次的成功率边际几乎为零,反而拉长 P99。

L2:同 provider 内的模型回退

主模型返回 429,往同 provider 的备用模型切换。延迟极低(同一连接、同一 endpoint),客户端无感知。

SAME_PROVIDER_FALLBACKS = {
    "claude-opus-4-7": ["claude-sonnet-4-6", "claude-haiku-4-5"],
    "gpt-5.2": ["gpt-5.1", "gpt-4o"],
    "deepseek-v4-flash": ["deepseek-v3.2"],
}

def call_with_l2_fallback(client, model, messages, **kw):
    candidates = [model] + SAME_PROVIDER_FALLBACKS.get(model, [])
    last_err = None
    for m in candidates:
        try:
            return with_retry(lambda: client.chat.completions.create(
                model=m, messages=messages, **kw
            ))
        except RateLimitError as e:
            last_err = e
            continue
    raise last_err

工程要点:

  • 回退顺序按能力等级:opus → sonnet → haiku,能力下降但响应不空。
  • 不要无脑回退到不同代际:把 GPT-5.2 回退到 GPT-3.5 是灾难,输出质量差太远,调用方拿到的等于错答案。
  • 单次调用最多走 3 个候选,再多就该跳到 L3。

L3:跨 provider 回退

整个 provider 挂了——这种事比想象中频繁。OpenAI 2024 全年至少有 6 次 30 分钟以上的全球性中断。L3 就是这一层的护身符。

PROVIDERS = [
    {"name": "anthropic", "client": anthropic_client, "model": "claude-sonnet-4-6"},
    {"name": "openai",    "client": openai_client,    "model": "gpt-5.1"},
    {"name": "deepseek",  "client": deepseek_client,  "model": "deepseek-v3.2"},
    {"name": "zhipu",     "client": zhipu_client,     "model": "glm-4.7"},
]

def call_with_l3_fallback(messages, **kw):
    last_err = None
    for p in PROVIDERS:
        try:
            return call_with_l2_fallback(
                p["client"], p["model"], messages, **kw
            )
        except (APIError, APIConnectionError) as e:
            last_err = e
            continue
    raise last_err

跨 provider 的真正难点不是写循环,而是让请求和响应在不同 provider 间保持等价。三个常见兼容性陷阱:

字段 / 能力 OpenAI Anthropic DeepSeek Gemini
max_tokens 含义 输出上限 输出上限 输出上限 输入+输出
系统消息 role:system 顶层 system 字段 role:system systemInstruction
response_format=json_object ⚠️ 仅部分模型
tool_choice="required" ⚠️ 静默忽略
流式 usage 字段 ✅ 末包返回 ✅ delta 累计 ❌ 不返回

写网关层一定要做参数翻译,否则 L3 触发时表面上"成功"了,实际响应少字段、JSON 没强制、工具没强制调用——客户端拿到一份悄悄出错的数据,比直接 502 还危险。


L4:能力路由

L4 解决一类被低估的故障:模型说支持,实际不支持

举个真实例子:某国产模型文档写着支持 function calling,传 tools 参数也不报错。但实际返回里 tool_calls 永远是 null——等于把 tool calling 静默吞了。这种 bug 在 5xx 之前就该被路由层挡掉。

实现思路是建一张"能力矩阵",请求进来先按需求过滤候选模型:

CAPABILITIES = {
    "gpt-5.2":          {"tools": True,  "vision": True,  "json_strict": True,  "stream_usage": True},
    "claude-opus-4-7":  {"tools": True,  "vision": True,  "json_strict": False, "stream_usage": True},
    "deepseek-v4-flash":{"tools": True,  "vision": False, "json_strict": True,  "stream_usage": False},
    "qwen-max":         {"tools": False, "vision": True,  "json_strict": False, "stream_usage": False},
    "glm-4.7":          {"tools": True,  "vision": True,  "json_strict": True,  "stream_usage": False},
}

def detect_required_capabilities(req):
    caps = []
    if req.get("tools"):
        caps.append("tools")
    if any(isinstance(c, dict) and c.get("type") == "image_url"
           for m in req["messages"] for c in (m.get("content") or [])):
        caps.append("vision")
    if req.get("response_format", {}).get("type") == "json_object":
        caps.append("json_strict")
    return caps

def filter_by_capability(providers, required):
    return [p for p in providers
            if all(CAPABILITIES.get(p["model"], {}).get(c) for c in required)]

请求路由前先调用 filter_by_capability(PROVIDERS, detect_required_capabilities(req)),从候选里剔除不合格的模型。这一步几乎无延迟,但能避免 L1-L3 整套链路在错误的模型上空转。

能力矩阵的维护:不要相信 provider 文档,写探针定期验证。每天凌晨跑一遍 capability probe,对每个模型发一个最小测试请求验证 tools / vision / json_strict 是否真的工作,结果落库。这是网关层的"健康检查"。


L5:语义路由

最高一层,也是工程量最大的一层。前面 4 层都是被动响应故障,L5 是主动按任务匹配模型

举几个明显的场景:

  • 用户问"今天北京天气"——根本不该路由到 opus 级模型,便宜的 haiku / flash 完全够。
  • 用户贴了一段 1000 行代码让你重构——必须路由到能装下、且代码能力强的模型(claude-opus / gpt-5.2)。
  • 用户上传一张电路图问元器件——路由到 vision 能力强且对工业图理解好的模型。

最朴素的实现:

def route_by_task(req):
    text = " ".join(m.get("content", "") for m in req["messages"]
                    if isinstance(m.get("content"), str))
    has_code = "```" in text or "def " in text or "function " in text
    has_image = any(isinstance(c, dict) and c.get("type") == "image_url"
                    for m in req["messages"] for c in (m.get("content") or []))
    token_estimate = len(text) // 3  # 粗略估算

    if has_image:
        return "claude-opus-4-7"  # vision 任务
    if has_code and token_estimate > 2000:
        return "gpt-5.2"  # 长代码
    if token_estimate < 200:
        return "claude-haiku-4-5"  # 短问答省钱
    return "claude-sonnet-4-6"  # 默认

关键词路由能解决 60% 场景,剩下 40% 需要更精细的方法:

  1. 小模型分类器:用一个 100ms 内出结果的 classifier(比如 BGE 系列 embedding + 简单 KNN),把请求归类到 [简单问答 / 代码 / 推理 / 创作 / 多模态] 五类。
  2. 历史性能反馈:记录每个模型在每类任务上的用户反馈(重新生成率、点赞率),用 epsilon-greedy 或 Thompson sampling 动态调整。
  3. 成本约束:把"在 P95 质量下选最便宜模型"作为优化目标,不是"质量最高"。

L5 的 ROI 往往超出预期。某团队在生产环境做了语义路由后,整体 token 成本降了 47%,用户感知质量没下降——便宜模型在简单任务上的输出,跟昂贵模型几乎没差异。


五层之外:必备的可观测性

降级链路再优雅,没有指标就是黑盒。最少要采集这 4 组:

指标 用途
fallback_triggered_total{layer="L1|L2|L3|L4|L5"} 看每层触发频率,定位最脆弱的上游
fallback_latency_seconds 直方图 P95/P99 延迟,判断降级是否影响用户
fallback_success_rate{from_model, to_model} 哪些回退路径靠谱
provider_capability_drift{model, capability} 探针发现的能力变化,提醒调整路由

接 Prometheus + Grafana 即可。重点是让降级不再是"出事后才知道"的暗操作,而是有度量、可改进的工程系统。


生产环境上线检查清单

最后给一份在多个团队验证过的 pre-launch checklist:

  • L1 重试有 jitter,且区分可重试/不可重试错误码
  • L2 同 provider 候选按能力等级排序,不跨代际回退
  • L3 跨 provider 做了参数翻译(max_tokens / system message / json_format)
  • L3 触发时记录 from→to 路径,便于事后对账
  • L4 能力矩阵有自动化探针,至少每天跑一次
  • L5 路由决策可解释(能输出"为什么选这个模型")
  • 5 层各自有独立 metric,能在 Grafana 上看趋势
  • 降级触发时通过 trace ID 串联日志,方便定位
  • 重要客户走"VIP 链路",固定主备模型,不参与全局路由
  • 有"全部 provider 不可用"的 graceful 错误响应(返回缓存 / 模板回复 / 排队)

笔者目前的多模型项目走的是统一网关方案,把 L1-L4 都封装在网关层,业务方只调一个 OpenAI 兼容 endpoint,后端透明做完整降级。有类似需求的可以试试 TheRouter(therouter.ai),也欢迎对照本文的 5 层设计做自建。

TheRouter — 多 provider LLM API 统一网关,一个 key 调 30+ 模型

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐