用 n8n 与 Python 实现模型故障切换

我上线过的每一款 AI 产品，都有同一个隐藏依赖：单一模型供应商。而每家供应商都可能宕机、触发限流，或者悄悄改变延迟表现。一旦这种情况发生在凌晨 2 点，你的“自主”工作流就会直接停摆。

故障切换路由器，是你能以最低成本增加的一项可靠性保障。下面是我在 n8n 和纯 Python 中使用的架构。

你真正需要绕开的故障类型

故障切换不只是“挂了就换一家”。这里有四种不同的故障，需要分别应对：

最后一种最致命。供应商可能返回完全有效的 HTTP 200，但响应内容却无法被你的解析器使用。如果不做校验，你就会“成功地”失败。

核心思路：优先级列表 + 健康检查关卡

不要把单一模型硬编码在代码里。定义一个有序列表，并为每次请求设置预算：

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                      # still cooling down
        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}")

整个模式就是这样。三个要点：有序列表、每次请求的超时时间，以及触发限流后的冷却期。

校验器才是关键

大多数“故障切换”实现只捕获异常，所以它们仍然会把垃圾结果交付出去。给路由器加上校验器，它就能把关结果的正确性，而不只是服务的可用性：

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

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

这样一来，如果供应商返回的是普通文本而不是 JSON，路由器就会自动放弃这次结果，切换到下一家。

如果你使用 n8n，结构也是一样的，只是通过可视化方式搭建：

为供应商 A 配置一个 HTTP Request 节点，并设置较短的超时时间（不要使用默认值，要显式设置）。

用 IF 节点检查状态码，再用 Code 节点解析并校验响应体。

将失败分支连接到供应商 B 的 HTTP Request 节点。依此重复。

为每家供应商维护一个小型静态数据 / Redis 计数器，让限流分支能够设置冷却期。

最后配置一个“全部失败”的 webhook，向 Telegram/Slack 发送通知，让你知道整个切换链已经耗尽。

n8n 中的一个常见陷阱是：有人把备用调用放在并行分支，而不是串行分支。这样每次请求都会调用两家供应商，成本直接翻倍。故障切换必须串行执行。

不要遇到任何错误都切换供应商。格式错误的 prompt，在三家供应商那里都会失败。先校验输入，否则一个错误请求就会把整个供应商列表消耗一遍。

统一输出格式。不同供应商返回的数据结构各不相同。校验器应该针对统一格式后的对象运行，而不是直接校验各家供应商的原始响应。

记录每次请求由哪家供应商提供服务。没有这项记录，你就无法分辨：主供应商 90% 的命中率究竟意味着系统健康，还是系统已经悄悄降级。

把超时时间设得比调用方的超时时间更短。如果你的工作流允许执行 60 秒，而每家供应商都耗时 25 秒，那么三次失败就是 75 秒，早已超出预算。要限制总耗时，而不只是每次调用的耗时。

免费套餐同样会限流。免费的前沿模型很适合做供应商 C，但不适合做供应商 A。

只依赖一家供应商的工作流，本质上是披着生产力外衣的单点故障。增加 30 行路由代码，就能把“AI 挂了”从凌晨 2 点把你叫醒的告警，变成一条日志。这就是演示和产品之间的区别。

我把这套方案的生产版本打包成了可直接导入的 n8n 工作流：一个三级级联路由器（Claude → GPT-4o → DeepSeek），包含冷却状态、输出校验、BANT 潜在客户分级，以及 Telegram 运维告警。

同一路由器的 Python 版本则收录在 AI Agent Production Kit 中，仅使用标准库，并带有本地故障切换链。

两者都采用自主定价。很好奇你们是怎么处理“悄悄返回垃圾结果”这种情况的——这仍然是我花最多时间调优的部分。

如果需要采取进一步行动，你可以考虑屏蔽此人，或者举报滥用行为。