构建多模型故障转移路由：打造零停机AI工作流

每一个上线的AI产品都藏着同一个隐患：它只依赖一家模型供应商。而所有供应商都会宕机、触发限流，或在不声不响间改变延迟表现。当这些问题在凌晨两点发生时，你所谓的"自主运行"工作流就彻底停摆了。

故障转移路由（failover router）是你能添加的性价比最高的可靠性升级。本文梳理了一套可直接落地的架构，既有 n8n 可视化实现，也有纯 Python 版本。

你真正要绕开的四种故障模式

故障转移并不是简单的"挂了就换另一家"。实际存在四种截然不同的失败情形，每一种都需要不同的应对策略：

故障类型	症状	正确应对
硬性宕机	5xx / 连接被拒	立即切换
限流	429	切换 + 冷却计时
响应过慢	返回 200 但耗时 30s+	超时即切换，不要死等
静默垃圾	200 但内容残缺/为空	校验 → 重试，再切换

最后一种才是真正的杀手。供应商完全可能返回一个格式正确的 HTTP 200，但响应内容是你的解析器根本无法使用的。如果你不做校验，就会"成功地"失败——系统认为请求完成了，实际上拿到的是废料。

"静默垃圾"（silent garbage）在分布式系统中也被称为 silent failure 或 semantic failure，是最难通过基础监控发现的一类问题。传统的 HTTP 健康检查、进程存活探针乃至 APM 工具，通常只关注状态码和延迟，不解析响应语义。因此，一个持续返回 200 OK 但内容为空 JSON、截断文本或重复乱码的供应商，在基础设施监控层面会被判定为"完全健康"。这也是为什么在 LLM 调用链中，应用层校验（application-level validation）必须作为路由逻辑的一部分，而不能外包给基础设施层。

核心思路：优先级列表 + 健康闸门

不要把单一模型写死在代码里。定义一个有序列表，并为每次请求设定预算（超时时间）：

PROVIDERS = [
    {"name": "claude",  "call": call_claude,  "timeout": 25},
    {"name": "gpt4o",   "call": call_gpt,     "timeout": 25},
    {"name": "deepseek","call": call_deepseek,"timeout": 30},
]

然后遍历这个列表，遇到任何一类失败——超时、429、5xx 或输出无效——都前进到下一个供应商。同时记录冷却状态，让一个被限流的供应商在接下来的 N 秒内被跳过，而不是被反复猛敲：

import time

_cooldown: dict[str, float] = {}

def route(prompt: str, validate=None) -> str:
    last_err = None
    for p in PROVIDERS:
        if _cooldown.get(p["name"], 0) > time.time():
            continue  # 仍在冷却中
        try:
            out = p["call"](prompt, timeout=p["timeout"])
            if validate and not validate(out):
                raise ValueError("invalid output")
            return out
        except RateLimited:
            _cooldown[p["name"]] = time.time() + 60
            last_err = "rate_limited"
        except Exception as e:
            last_err = str(e)
    raise RuntimeError(f"all providers failed: {last_err}")

整个模式就这么简单，归结为三个要点：有序列表、每次请求的超时、限流后的冷却。

冷却（cooldown）机制本质上是一种指数退避（exponential backoff）的简化版本。标准的指数退避策略会在每次重试失败后将等待时间翻倍（如 1s、2s、4s、8s……），并加入随机抖动（jitter）以避免多个客户端同时重试导致的"惊群效应"（thundering herd）。上文代码使用了固定 60 秒冷却，适合快速实现；生产系统中可进一步改为：首次限流冷却 30s，第二次 60s，第三次 120s，同时记录连续失败次数。此外，冷却状态存在内存字典中意味着多进程或重启后状态丢失，团队协作场景下应将其持久化到 Redis 等共享存储。

校验器才是真正的关键

大多数"故障转移"实现只捕获异常，这正是它们仍然会输出垃圾的原因。给路由器加上一个校验器（validator），它就从单纯的"可用性闸门"升级为"正确性闸门"：

import json

def valid_json_with_keys(*keys):
    def check(s: str) -> bool:
        try:
            obj = json.loads(s)
        except Exception:
            return False
        return all(k in obj for k in keys)
    return check

# 用法: route(prompt, validate=valid_json_with_keys("summary", "score"))

这样，一个返回散文而非 JSON 的供应商会被自动降级。可用性与正确性被同时纳入了判断逻辑，这正是生产级系统和演示 Demo 的分水岭。

在 n8n 中如何实现

如果你用的是 n8n，架构思路完全一致，只是改为可视化搭建：

1. 为供应商 A 配置 HTTP Request 节点，并设置一个紧凑的超时时间——不要用默认值，必须显式设定。
2. 用 IF 节点判断状态码，再配一个 Code 节点解析并校验响应体。
3. 在失败分支上，连接到供应商 B 的 HTTP Request 节点，依此类推。
4. 用一小块 static data / Redis 为每个供应商维护计数器，让限流分支能够设置冷却时间。
5. 最后设置一个**"全部失败"的 webhook**，推送到 Telegram 或 Slack，让你在整条链路耗尽时能第一时间察觉。

n8n 里的一个经典陷阱：很多人把后备方案放进并行分支而非串行分支，结果每次都同时调用两家供应商，成本直接翻倍。故障转移必须是串行的。

n8n 的 HTTP Request 节点默认超时为 300 秒，这对故障转移场景极为危险——一个挂起的供应商会占用整条工作流长达 5 分钟。正确做法是在每个节点的"Options"中将超时显式设置为 20-30 秒。同时，n8n 的 static data 是工作流级别的持久化存储，重启后依然保留，适合存储冷却截止时间戳；但它不支持跨工作流共享状态，若同一套路由逻辑被多个工作流复用，需改用 Redis 节点或外部数据库统一管理供应商健康状态，否则各工作流的冷却计数器相互独立，无法协同限流保护。

容易踩的坑

• 不要对每个错误都触发故障转移。 一个格式错误的 prompt 会在全部三家供应商上都失败。先校验你的输入，否则会因为一次坏请求而把整条链路白白烧穿。
• 归一化输出。 不同供应商返回的数据结构各异。你的校验器应该针对一个归一化后的对象运行，而不是各家原始响应。
• 记录每次请求由哪家供应商服务。 没有这个日志，你无法区分一个健康的"主供应商 90% 命中率"和一个悄然退化的系统。
• 把超时设在调用方超时之下。 如果工作流允许 60s，而每家供应商各占 25s，三次失败就是 75s，预算早已超支。要为总耗时设上限，而不只是单次调用。
• 免费额度一样会限流。 一个免费的前沿模型适合作为供应商 C，而非供应商 A。

为什么值得做

单供应商工作流本质上是一个披着生产力外衣的单点故障。加上大约 30 行路由代码，就能把"AI 挂了"从一次凌晨两点的紧急告警，降级为一行日志。这就是 Demo 与真正产品之间的差距。

原作者还将这套方案封装成了生产版本——一个三级级联路由（Claude → GPT-4o → DeepSeek^广告），带冷却状态、输出校验、BANT 线索分级以及 Telegram 运维告警，做成了可直接导入的 n8n 工作流。同时提供了功能一致的 Python 版本（仅依赖标准库，内置本地故障转移链）。对于团队而言，"静默垃圾"这类难以察觉的失败往往是最需要持续调优的环节。