Zero-downtime AI: how to build a multi-model failover router (n8n + Python)

#ai #llm #n8n #automation

Every AI product I've shipped has the same hidden dependency: one model provider. And every provider goes down, gets rate-limited, or silently changes its latency profile. When that happens at 2am, your "autonomous" workflow just stops.

A failover router is the cheapest reliability upgrade you can add. Here's the architecture I use, in n8n and in plain Python.

The failure modes you're actually routing around

Failover isn't just "if down, use other." There are four distinct failures, and they need different responses:

Failure	Symptom	Right response
Hard outage	5xx / connection refused	Immediate failover
Rate limit	429	Failover + cooldown timer
Slow	200 but 30s+	Timeout → failover (don't wait)
Silent garbage	200, malformed/empty	Validate → retry, then failover

The last one is the killer. A provider can return a perfectly valid HTTP 200 with a response your parser can't use. If you don't validate, you'll "successfully" fail.

The core idea: a priority list + a health gate

Don't hard-code a single model. Define an ordered list and a per-request budget:

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

Then walk the list, and on any failure class — timeout, 429, 5xx, or invalid output — advance to the next. Track a cooldown so a rate-limited provider is skipped for the next N seconds instead of being hammered:

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}")

That's the whole pattern. Three ideas: ordered list, per-request timeout, cooldown on rate limit.

The validator is the important part

Most "failover" implementations only catch exceptions. That's why they still ship garbage. Give the router a validator and it becomes a correctness gate, not just an uptime gate:

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"))

Now a provider that returns prose instead of JSON gets demoted automatically.

Doing it in n8n

If you're in n8n, the shape is the same but you build it visually:

HTTP Request node for provider A with a tight timeout (don't use the default — set it explicitly).
IF node on the status code + a Code node that parses and validates the body.
On the failure branch, wire to provider B's HTTP Request node. Repeat.
Keep a tiny static data / Redis counter per provider so the rate-limit branch can set a cooldown.
A final "all failed" webhook to Telegram/Slack so you know when the chain is exhausted.

The trap in n8n: people put the fallback in a parallel branch instead of a sequential one, so both providers get called every time and you double your cost. Failover must be sequential.

Gotchas

Don't fail over on every error. A malformed prompt will fail on all three providers. Validate your input first, or you'll burn the whole list on a bad request.
Normalize the output. Different providers return different shapes. Your validator should run against a normalized object, not each provider's raw response.
Log which provider served each request. Without this you can't tell a healthy 90% primary-hit-rate from a silently degraded one.
Set the timeout below your caller's timeout. If your workflow allows 60s and each provider takes 25s, three failures = 75s and you've already blown the budget. Cap total time, not just per-call.
Free tiers still rate-limit. A free frontier model is great as provider C, not provider A.

Why bother

A single-provider workflow is a single point of failure wearing a productivity hat. Adding 30 lines of routing turns "the AI is down" from a 2am page into a log line. That's the difference between a demo and a product.

I packaged a production version of this — a 3-tier cascading router (Claude → GPT-4o → DeepSeek) with cooldown state, output validation, BANT lead triage, and Telegram ops alerts — as ready-to-import n8n workflows:

https://ancuboy.gumroad.com/l/agentflow-os

And a Python version of the same router (stdlib-only, with a local failover chain) lives in the AI Agent Production Kit:

https://ancuboy.gumroad.com/l/production-ai-agent-kit

Both pay-what-you-want. Curious how you handle the "silent garbage" case — that's the one I still tune most.

DEV Community