如何给 AI 工作流设置异常重试和降级方案？

学完这篇，你能给一条 AI 工作流加上「出错自动重试 + 重试失败后降级」的完整保护，让接口抖动、限流、超时不再直接变成用户看到的报错。

第一步：先分清哪些错误值得重试

这一步要搞清楚错误分类，避免把「重试也没用」的错误硬重试 5 次，白白烧钱又拖慢响应。

把错误分成三类：

1. 可重试：HTTP 429（限流）、500/502/503/504、连接超时、读取超时、openai.error.APIConnectionError、RateLimitError。这类通常是临时的，隔一会儿再来就行。
2. 不可重试：400（参数错误，如 context_length_exceeded）、401/403（密钥错误、无权限）、404（模型名写错）。重试 100 次结果一样，直接失败并告警。
3. 可降级：模型不存在、余额不足、单次请求超过最大上下文。这类不重试，直接走备用方案。

注意：429 和 503 要区别对待。429 是「你太快了」，退避时间要长一点；503 是「服务端挂了」，可以稍快重试，但连续失败 3 次就应放弃。

第二步：用指数退避写重试逻辑

这一步给可重试错误加上带抖动的指数退避，做完后请求会在 1s、2s、4s 左右依次重试，而不是死循环式猛打。

以 Python 为例，不引入额外库的核心写法：

import time, random

def call_with_retry(fn, max_retries=3, base=1.0, cap=20.0,
                    retry_on=(429, 500, 502, 503, 504)):
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except Exception as e:
            status = getattr(e, "status_code", None) or \
                     getattr(getattr(e, "response", None), "status_code", None)
            if status not in retry_on or attempt == max_retries:
                raise
            delay = min(cap, base * (2 ** attempt))
            delay = delay * (0.5 + random.random())   # 抖动，防止同时重试
            time.sleep(delay)

关键参数：max_retries=3（总共最多 4 次尝试）、base=1.0、cap=20.0 封顶、抖动系数取 0.5~1.5 之间。如果用 LangChain，可以在 ChatOpenAI(...) 里直接传 max_retries=3；用 Node.js 的 openai SDK，构造时传 maxRetries: 3 即可。

注意：整条工作流的超时必须大于「所有重试耗时之和」，否则重试还没跑完就被上游掐断。建议给整条链路设 60~90 秒预算。

第三步：设计降级链路

这一步给「重试也救不回来」的情况准备备胎，做完后即使主模型挂了，用户仍能拿到一个可接受的结果。

按优先级排一条降级链，通常 3 层就够：

1. 换模型：主模型 gpt-4o → 备用 gpt-4o-mini（同一厂商，改模型名即可，SDK 无需更换）。
2. 换供应商：OpenAI 挂了 → 切到 Azure OpenAI 或 Claude。这需要你在代码里抽象一层 LLMClient 接口，把 chat() 方法做成统一入口，换实现只改配置。
3. 降功能：还不通就返回缓存结果（Redis 里存的上次相同 prompt 的输出，TTL 设 1 小时），或返回固定兜底话术：「当前请求繁忙，请稍后重试」。

伪代码结构：

def robust_call(prompt):
    # 第一层：主模型 + 重试
    try:
        return call_with_retry(lambda: primary.chat(prompt))
    except Exception:
        pass
    # 第二层：备用模型
    try:
        return call_with_retry(lambda: fallback.chat(prompt), max_retries=1)
    except Exception:
        pass
    # 第三层：缓存兜底
    cached = cache.get(prompt)
    if cached:
        return cached
    return {"content": "服务繁忙，请稍后重试", "degraded": True}

注意返回值里带一个 degraded: true 标记，方便上游做提示或监控统计。

注意：降级链每一层都要设独立的短超时（如 10 秒），否则第一层卡满 60 秒才轮到第二层，用户早就走了。

第四步：加监控，别让降级悄悄发生

这一步让重试和降级的次数变成可见指标。在工作流里埋三个计数器：retry_count、fallback_count、degraded_count，打到 Prometheus 或直接写日志。

简单做法：每次触发重试或降级时，logging.warning("fallback to %s, reason=%s", model, err)。当 fallback_count 在 5 分钟内超过 10 次，发告警——这通常说明主模型或网络真出问题了，而不是偶发抖动。

注意：不要只记录成功日志。没有失败日志的重试机制等于没有。

小结

• 先分错误类型：可重试（429/5xx/超时）、不可重试（4xx）、可降级（模型不可用/超上下文）。
• 重试用指数退避 + 抖动，次数 2~3 次封顶，别死循环。
• 降级链按「换模型 → 换供应商 → 缓存/兜底话术」三层排，每层设独立短超时。
• 整条链路总超时要大于所有重试之和。
• 每次重试和降级都要记日志、打指标，超阈值告警。