让 AI 功能在生产环境稳住：大模型调用重试、熔断与降级的三步实战法

大模型能力接进业务之后，最怕的不是模型答得不好，而是调用本身不稳定：偶发超时、429 限流、网关 500。很多团队的第一反应是“重试一次”，上线后才发现：重试把限流打得更严重，一个挂掉的依赖拖垮了整条链路，用户看到的还是报错。

让 AI 功能在生产环境稳住，不需要很重的中间件，三步就够：给重试立规矩、上熔断器、准备降级。下面按落地的顺序讲。

第一步：给重试立规矩

重试不是“失败了就再来一次”，先分清哪些错值得重试。

能重试的错：网络超时、429 限流、500/502/503 这类服务端偶发错误。这些错误和请求本身无关，再试一次很可能成功。

不能重试的错：400 参数错误、401/403 鉴权失败、内容审核拒绝。重试多少次结果都一样，只会浪费 Token 和时间。

退避策略要用指数退避加抖动：第一次等 1 秒，第二次等 2 秒，第三次等 4 秒，再叠加一个随机抖动。固定间隔重试最容易造成“惊群效应”——大量失败请求在同一时刻一起重试，把刚恢复的服务再次打垮。

import random, time

def call_with_retry(fn, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except TransientError:  # 只捕获超时、429、5xx 这类瞬时错误
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            time.sleep(delay)

重试次数要有上限：一般 3 次足够。超过 3 次还失败，大概率不是抖动，而是依赖真的出问题了，这时应该交给熔断器和降级，而不是继续重试。

写操作要带幂等键：如果大模型调用之后还会写库、发消息，重试可能导致重复执行。给每次业务请求生成一个幂等键传给下游，重复请求直接返回第一次的结果。

第二步：上熔断器，别让坏掉的依赖拖垮链路

重试解决的是“偶发失败”，熔断器解决的是“依赖持续不可用”。当模型服务连续失败时，继续重试只会堆积等待的线程、拉长整体耗时，最后连不依赖大模型的功能也被拖垮。

熔断器有三个状态：

• 关闭：正常放行请求，同时统计失败率。
• 打开：失败达到阈值后直接拒绝请求，不再真正调用，快速失败。
• 半开：打开一段时间后放少量请求试探，成功就恢复关闭，失败就继续打开。

一组可以直接抄的初始阈值：10 秒内失败率超过 50%，或者连续失败 5 次，就打开熔断器 30 秒；半开状态放 1 个试探请求，成功后再放行。

import time

class CircuitBreaker:
    def __init__(self, fail_threshold=5, open_seconds=30):
        self.failures = 0
        self.threshold = fail_threshold
        self.open_until = 0
        self.open_seconds = open_seconds

    def call(self, fn):
        if time.time() < self.open_until:
            raise CircuitOpenError("熔断器打开，快速失败")
        try:
            result = fn()
            self.failures = 0
            return result
        except TransientError:
            self.failures += 1
            if self.failures >= self.threshold:
                self.open_until = time.time() + self.open_seconds
            raise

熔断器打开期间要记一条日志、打一个监控点，否则排查时你会完全不知道请求为什么被拒绝。

第三步：准备降级，宁可给差一点的答案也不直接报错

当前两步都兜不住时，降级决定了用户看到的是“能用的结果”还是“一行报错”。降级按效果从好到差排个序：

1. 返回缓存答案：同样的问题之前答过，直接返回缓存，几乎零成本。
2. 切换更小、更便宜的模型：主模型不可用时用轻量模型先顶上，答案质量稍降但不断服务。
3. 模板化兜底回复：把关键信息用固定模板组织出来，再明确告诉用户“当前为简化结果，详细分析稍后补上”。
4. 明确提示稍后再试：实在什么都给不出时，也要给用户一个明确的预期，而不是一个晦涩的错误码。

不是所有场景都能降级：涉及资金、写库、下单这类有副作用的操作，宁可明确失败，也不要用降级结果代替真实执行。降级只适用于“读”和“生成建议”类场景。

把三步串成一条调用链

一次完整的调用链路长这样：

1. 入口先查熔断器：打开了就直接走降级，不发起调用。
2. 熔断器关闭时发起调用，内部按第一步的策略重试。
3. 重试耗尽仍失败：更新熔断器状态，然后走降级分支。
4. 全程记录耗时、重试次数、熔断状态和降级原因，方便事后复盘。

def ask_model(prompt):
    if breaker.is_open():
        return fallback_answer(prompt, reason="breaker_open")
    try:
        return call_with_retry(lambda: model.chat(prompt))
    except TransientError:
        return fallback_answer(prompt, reason="retry_exhausted")

这段代码很短，但三层保护都在里面：重试扛住抖动，熔断器防止雪崩，降级保住用户体验。

落地检查清单

上线前按这张清单逐项确认：

小结

AI 功能的稳定性不靠模型本身，靠调用链路的三层保护：重试处理偶发抖动，熔断器阻止故障扩散，降级保住最基本的可用性。先从这三步做起，再根据真实的失败数据调阈值，比一上来就搭重型网关要划算得多。