LLM 调用失败兜底设计：让应用在模型故障时依然可用

大模型 API 不是 100% 可靠的基础设施：供应商偶发宕机、限流（429）、超时（网络或模型过载）是生产环境的常态。没有兜底设计的应用，在高峰期或供应商故障时会直接对用户报错。合理的多层兜底让你的应用在模型故障时仍能提供降级服务。

如果你只上线过一个纯调用 OpenAI SDK 的小 Demo，可能觉得这套东西是过度设计。但只要你的日活过了几千、或者产品用在客服/工单这种不能中断的场景，下面这个场景你迟早会遇到：晚高峰用户量涨起来，某个供应商那边突然限流或者小范围过载，你的应用后端瞬间堆起一堆挂起的请求，前端表现为转圈圈转半天最后报”服务异常”。这时候值班的人第一反应是去查自己的代码有没有 bug，结果排查半天发现是上游模型那边的问题——但用户不管这些，他们只知道你的产品挂了。多层兜底要解决的就是这类问题：把”上游偶发故障”和”用户能不能继续用”这两件事解耦开。

失败分类与应对策略

错误类型	HTTP 码	原因	推荐处理
限流	429	超过 RPM/TPM 限额	指数退避重试（≤3次）
服务不可用	503	供应商故障/过载	切换备用模型
超时	408 / 网络超时	网络或模型推理慢	先重试，仍失败则降级
认证失败	401	Key 过期或无效	立即报警，不重试
内容拒绝	400	违规内容过滤	返回用户友好提示，不重试
上下文超长	400	超出模型窗口	压缩上下文后重试

这张表不是拿来背的，关键是你得先能”分清楚是哪一种”，否则重试策略全都乱套。实际排查时你看到的往往不是干净的状态码，而是 SDK 包了一层的异常文本，比如 OpenAI SDK 抛出的 openai.RateLimitError: Error code: 429 - {'error': {'message': 'Rate limit reached for requests', 'type': 'requests', 'code': 'rate_limit_exceeded'}}，这种一看 type 就知道是配额问题，该退避重试；但如果 message 里写的是 You exceeded your current quota, please check your plan and billing details，同样是 429，根因却是余额或套餐用完了，重试多少次都没用，得走告警而不是走重试逻辑，很多人一上来图省事把所有 429 都塞进同一个重试分支，结果余额耗尽时还在傻等退避，用户等了几十秒最后还是失败。同理，403 有时候是权限问题（Key 没开通某个模型），有时候是区域限制（IP 被判定在受限地区），这两种也不能用同一套处理逻辑，得看清楚 error 里的 code 字段再分流处理。

第一层：指数退避重试

import asyncio
import random
from openai import RateLimitError, APIStatusError

async def call_with_retry(
    client, messages: list, model: str = "gpt-4o", max_retries: int = 3
) -> str:
    for attempt in range(max_retries):
        try:
            response = await client.chat.completions.create(
                model=model, messages=messages, timeout=30
            )
            return response.choices[0].message.content

        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            # 指数退避 + 随机抖动，避免惊群效应
            wait = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(wait)

        except APIStatusError as e:
            if e.status_code in (400, 401):  # 不可重试的错误
                raise
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

关键细节：随机抖动（Jitter） 防止多个请求在同一时刻同时重试冲垮刚恢复的服务。

为什么退避时间要用 2 ** attempt 而不是固定等 3 秒？固定等待有个问题：如果这次故障是供应商那边真的在做扩容或者重启服务，短时间内根本不会恢复，你固定等 3 秒重试，等于是在故障期间持续给对方增加压力，还可能因为大量客户端同时都是”等 3 秒再打”而造成第二波拥堵；指数增长则是让等待时间随失败次数拉长（1 秒、2 秒、4 秒……），给对方留出恢复空间，同时叠加 random.uniform(0, 1) 的抖动，让不同请求的重试时间点错开，这就是业界常说的”退避加抖动”（backoff with jitter），AWS 的架构博客里专门讨论过这个模式能显著降低雪崩概率。

还有一个很多人踩过的坑：不是所有请求都能安全重试。如果你的调用链里模型输出会触发一个有副作用的动作（比如函数调用去扣库存、发短信、写数据库），那么”超时后重试”就有可能变成”重复执行”——原始请求可能已经在服务端处理完了，只是响应包没传回来，你这边判定超时又发了一遍，副作用就执行了两次。稳妥的做法是给每次业务请求带上幂等键（idempotency key），或者干脆把有副作用的步骤和”生成回答”的步骤拆开，只对纯读取/纯生成的调用做自动重试。另外流式（stream）请求重试也要小心：如果已经吐出去一半的 token 给前端了，这时候连接断了，你不能简单地”重新调用整个请求”再把结果拼接回去，那样用户会看到前半段重复出现，正确做法是记录已经消费到的位置，重试时明确丢弃旧的部分输出，或者直接对用户提示”回答中断，正在重新生成”再整体刷新。

第二层：模型降级链

主模型失败时自动切换备用模型：

MODEL_FALLBACK_CHAIN = [
    {"model": "gpt-4o",            "client": openai_client},
    {"model": "claude-3-5-sonnet", "client": anthropic_client},
    {"model": "deepseek-chat",     "client": deepseek_client},
    {"model": "gpt-4o-mini",       "client": openai_client},   # 最后兜底用轻量模型
]

async def call_with_fallback(messages: list) -> tuple[str, str]:
    """返回 (回答内容, 实际使用的模型名)"""
    last_error = None
    for config in MODEL_FALLBACK_CHAIN:
        try:
            result = await call_with_retry(
                config["client"], messages, config["model"]
            )
            return result, config["model"]
        except Exception as e:
            last_error = e
            # 记录降级事件，用于监控和告警
            logger.warning("model_fallback", model=config["model"], error=str(e))
            continue

    raise RuntimeError(f"所有模型均不可用: {last_error}")

降级链设计建议：按能力从强到弱排列；不同供应商交替放置（避免同一供应商连续失败）；轻量模型（gpt-4o-mini、deepseek-chat）放最后作为最低保障。

降级顺序不是拍脑袋定的，得结合你实际业务对”质量”和”延迟”的容忍度来排。举个判断思路：如果你的场景是客服问答，用户更在意”有没有回应”而不是”回答多精妙”，那降级链可以更激进地往轻量模型靠；如果是代码生成或者复杂数据分析，降级到弱模型基本等于给用户一个错误答案，这种场景宁可多等几秒重试同一个强模型，也不要轻易降级，这时候降级链里第二顺位该放”同一模型的另一个区域/另一个供应商代理”而不是能力更弱的模型。实践中可以按下面的维度给候选模型打分再排序：

维度	说明
首 token 延迟	流式场景下用户对”卡顿感”最敏感的指标
单价	每 1K token 的成本，降级链末位通常选性价比最高的
能力对齐度	与主模型输出风格/结构的接近程度，避免降级后格式突变
供应商独立性	是否与主模型同一云/同一区域，决定同故障域概率

这四个维度里，“能力对齐度”最容易被忽略：如果你的主模型习惯输出 JSON 结构化结果，降级模型却经常在 JSON 外面加一堆解释性文字，前端解析就会崩，所以降级链上的每个模型都建议单独跑一遍你的输出格式测试用例，而不是只测过一次主模型就上线。

第三层：熔断器（Circuit Breaker）

防止向已知故障的服务继续发送请求：

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"      # 正常，允许请求
    OPEN = "open"          # 熔断，拒绝请求
    HALF_OPEN = "half_open"  # 探测恢复

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = CircuitState.CLOSED
        self.opened_at = None

    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.opened_at > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise RuntimeError("Circuit breaker OPEN, skipping call")

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = CircuitState.CLOSED

    def _on_failure(self):
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN
            self.opened_at = time.time()

熔断器和重试是两回事：重试解决的是”单次请求偶发失败”，熔断器解决的是”这个服务现在整体就是不行，别再浪费时间和配额去试了”。二者配合的顺序应该是——单次调用先走重试逻辑，重试彻底失败之后才计入熔断器的失败计数，而不是每次重试的每一次尝试都单独计一次失败，不然 failure_threshold=5 可能一次业务请求内部的 3 次重试就快把额度用完了，熔断器过早跳闸。

上面这版 CircuitBreaker 是同步简化实现，真正接入异步高并发系统时有个容易被忽略的坑：half_open 状态下的并发探测。假设熔断器进入 HALF_OPEN 后，同一时刻有 50 个协程都在调用这个服务，它们会同时判断”现在是 half_open，试一下”，于是 50 个探测请求同时打过去——如果这个服务本来就是被打崩的，你这波”探测”直接把它又打死了。工程上更稳妥的做法是在 half_open 状态下限制并发探测数（比如只放行 1 个请求，其余的直接走降级），等这一个探测成功之后再逐步放开更多流量，这也是很多网关产品（如 Envoy、Sentinel）里”半开限流”的实现思路，自己写的时候记得用锁或信号量把这个”只放一个”的逻辑显式做出来，不能只靠状态判断，否则并发场景下形同虚设。

参数怎么调有没有经验值？failure_threshold 建议从 5 起步，太小容易被瞬时抖动误判熔断；recovery_timeout 建议从 30~60 秒起步，太短起不到”给对方恢复时间”的作用，太长又会让本来已经恢复的服务被你晾在一边空等。上线后最好把这两个参数做成可配置项，跑一周观察熔断触发和恢复的实际间隔，再回头微调，不要一次定死。

第四层：静态兜底响应

当所有模型均不可用时，返回有意义的静态回复而非空白报错：

STATIC_FALLBACKS = {
    "greeting": "你好！我是 AI 助手。当前服务繁忙，请稍后再试。",
    "error":    "抱歉，AI 服务暂时不可用，请 5 分钟后重试，或联系客服。",
    "search":   "搜索功能暂时不可用，您可以直接访问 [知识库链接]。",
}

def get_fallback_response(intent: str = "error") -> str:
    return STATIC_FALLBACKS.get(intent, STATIC_FALLBACKS["error"])

这一层最容易被做成”敷衍用户”，这里提醒一句：能诚实告知服务降级，就不要装作一切正常。比如你的产品是智能客服，全部模型都挂了的时候，与其硬凑一句听起来正常的话让用户误以为得到了真实回答，不如明确告诉对方”当前 AI 服务繁忙，已为您转接人工/请稍后重试”，前者短期看起来体验没中断，长期看会因为答非所问砸掉用户对产品的信任。静态兜底文案也建议按功能重要性分级维护：核心付费功能（比如生成正式文档）失败时要给出更明确的补救路径（转人工、保留草稿、稍后自动重试并通知），非核心的辅助功能（比如智能推荐一句文案）失败时直接隐藏这个功能模块或者返回空，不必打扰用户。

进阶：给每一层都埋监控指标

四层兜底如果没有监控，你根本不知道哪一层在实际生产中真正起作用、参数是不是调得合理。建议至少埋这几个指标：重试次数分布（按 attempt 次数分桶，看多少请求是重试 1 次就成功、多少是拖到第 3 次）、降级链命中率（按模型分别统计被降级命中的次数和占比，占比持续走高说明主模型不稳定，该找供应商反馈或者干脆换主力模型）、熔断器状态切换次数和平均持续时长（跳闸太频繁说明阈值设低了，或者上游确实不稳定该换供应商）、静态兜底触发率（这个数字理论上应该趋近于零，一旦持续不为零就是全链路都在扛不住，得升级为运维事件而不是继续静默兜底）。把这几个指标接到你现有的监控面板（Grafana、云监控都行），比每次靠人工翻日志排查靠谱得多，也能在参数调优时给你确凿的数据支撑，而不是凭感觉改。

常见问题

重试多少次合适？超时设多久？ API 调用建议最多重试 3 次。超时建议：非流式请求 30s，流式请求首 token 超时 10s（之后每 5s 无数据则中断）。超时设太长会拖住请求队列，导致级联故障。

降级到轻量模型后，回答质量下降用户会察觉吗？ 一般场景（问答、摘要）差异不明显；复杂推理（代码、数学）差异明显。解决方案：降级时在回复末尾加提示”当前使用精简模式，回答可能不够详细”，管理用户预期，同时在日志中标记降级事件方便复盘。

如何快速知道哪个供应商在故障？ 订阅各供应商状态页（OpenAI: status.openai.com、Anthropic: status.anthropic.com）的 RSS 或 webhook；在自己的监控中加成功率仪表盘，连续 5 次失败即触发告警。

四层兜底叠在一起，总耗时会不会失控，用户到底要等多久？ 这是实际压测时最容易翻车的地方：如果每一层都各自设置了自己的重试和超时，极端情况下一次请求可能要经历”3 次重试 × 30 秒超时 + 切到第二个模型再重试 + 熔断判断”，加起来用户能等出去 1 分钟以上，这在任何产品里都是不可接受的。正确做法是给整条链路设一个”总超时预算”（比如端到端 15 秒），每一层在拿到调用结果时都要检查剩余预算还有多少，预算耗尽就直接跳到静态兜底，不再往下一层继续尝试，而不是让每一层的超时设置互相独立、无限叠加。

这套兜底逻辑要不要自己每个项目都重写一遍？ 如果你只有一两个应用，自己维护这套重试 + 降级链 + 熔断器的代码成本还能接受；但如果你同时在跑多个产品线，每个都要单独接入并维护多家供应商的 Key、限额和故障状态，运维成本会指数级上升。这也是为什么很多团队后来会把这层收敛到一个统一的网关层去做，而不是让每个业务应用各自实现一遍。

---

← 返回 应用模式总览：从 Prompt 到 Agent | 应用模式专题

相关阅读：应用可观测与日志 · 多 Agent 协作：任务拆解与编排

多供应商兜底配置繁琐？力达云聚合 API 内置多模型自动 failover，一个接口后面挂多家供应商，无需在应用层自己维护降级链。