Rate limits are the most predictable failure mode in LLM production, yet most teams treat them as an afterthought. Multi-provider routing rate limits solutions work because they acknowledge a simple reality: no single provider stays healthy 100% of the time. This guide walks through building a routing layer that detects limit errors, fails over cleanly, and preserves your application’s semantics — without inventing a custom SDK for every model vendor.
Understand the failure modes before you route
Rate limits manifest differently across providers. OpenAI returns 429 with a retry-after header. Anthropic uses 429 with a JSON body containing retry_after. Google Vertex AI returns 429 with gRPC status codes. Some providers throttle silently by increasing latency. Others return 503 when capacity is exhausted.
Your routing logic needs to normalize these signals. Start by classifying every upstream response into three buckets: success, retryable failure, and non-retryable failure. Only retryable failures should trigger a provider switch.
# Normalized error classification
from enum import Enum
from dataclasses import dataclass
class ErrorClass(Enum):
SUCCESS = "success"
RETRYABLE = "retryable" # rate limit, timeout, 5xx
NON_RETRYABLE = "non_retryable" # 4xx except 429, auth errors, invalid request
@dataclass
class ProviderResponse:
error_class: ErrorClass
retry_after_ms: int | None = None
provider: str = ""
model: str = ""
usage: dict | None = None
def classify_response(status: int, headers: dict, body: dict, provider: str) -> ProviderResponse:
if 200 <= status < 300:
return ProviderResponse(ErrorClass.SUCCESS, provider=provider)
if status == 429:
retry_after = parse_retry_after(headers, body, provider)
return ProviderResponse(ErrorClass.RETRYABLE, retry_after_ms=retry_after, provider=provider)
if 500 <= status < 600:
return ProviderResponse(ErrorClass.RETRYABLE, provider=provider)
return ProviderResponse(ErrorClass.NON_RETRYABLE, provider=provider)Build a provider-agnostic request contract
Before you can route, you need a request format that every provider accepts. The OpenAI chat completions schema has become the de facto standard, but providers diverge on parameters: max_tokens vs max_output_tokens, stop vs stop_sequences, tool calling formats, and system prompt handling.
Define a canonical request in your gateway and translate per provider. Keep the translation layer thin — it’s a maintenance burden that grows with every new model.
# Canonical request your application sends
@dataclass
class ChatRequest:
messages: list[dict]
model: str # logical name, e.g. "gpt-4o-class"
temperature: float = 0.7
max_tokens: int | None = None
stop: list[str] | None = None
tools: list[dict] | None = None
stream: bool = False
metadata: dict | None = None # routing hints, priority, cost ceiling
# Provider-specific translation
def translate_to_openai(req: ChatRequest) -> dict:
payload = {
"model": resolve_model_id(req.model, "openai"),
"messages": req.messages,
"temperature": req.temperature,
"stream": req.stream,
}
if req.max_tokens:
payload["max_tokens"] = req.max_tokens
if req.stop:
payload["stop"] = req.stop
if req.tools:
payload["tools"] = req.tools
return payload
def translate_to_anthropic(req: ChatRequest) -> dict:
payload = {
"model": resolve_model_id(req.model, "anthropic"),
"messages": req.messages,
"temperature": req.temperature,
"stream": req.stream,
}
if req.max_tokens:
payload["max_tokens"] = req.max_tokens
if req.stop:
payload["stop_sequences"] = req.stop
# Anthropic expects system prompt in top-level field
system_msgs = [m for m in req.messages if m["role"] == "system"]
if system_msgs:
payload["system"] = system_msgs[0]["content"]
payload["messages"] = [m for m in req.messages if m["role"] != "system"]
return payloadImplement ordered fallback with health awareness
A naive round-robin fails under load because it sends traffic to already-degraded providers. Instead, maintain a priority-ordered provider list per logical model, and skip providers that recently returned rate limits.
import time
from collections import deque
from threading import Lock
class ProviderHealth:
def __init__(self, window_seconds: int = 60):
self.window = window_seconds
self.errors: deque[tuple[float, str]] = deque() # (timestamp, error_type)
self.lock = Lock()
def record_error(self, error_type: str):
with self.lock:
now = time.time()
self.errors.append((now, error_type))
self._prune(now)
def record_success(self):
with self.lock:
self._prune(time.time())
def is_healthy(self, max_rate_limit_errors: int = 3) -> bool:
with self.lock:
now = time.time()
self._prune(now)
rate_limit_count = sum(1 for _, et in self.errors if et == "rate_limit")
return rate_limit_count < max_rate_limit_errors
def _prune(self, now: float):
cutoff = now - self.window
while self.errors and self.errors[0][0] < cutoff:
self.errors.popleft()
class ModelRouter:
def __init__(self):
# Logical model -> ordered list of (provider, model_id)
self.routes: dict[str, list[tuple[str, str]]] = {
"gpt-4o-class": [
("openai", "gpt-4o"),
("azure", "gpt-4o"),
("anthropic", "claude-3-5-sonnet-20241022"),
],
"claude-sonnet-class": [
("anthropic", "claude-3-5-sonnet-20241022"),
("vertex", "claude-3-5-sonnet-v2"),
],
}
self.health: dict[str, ProviderHealth] = {}
def get_healthy_providers(self, logical_model: str) -> list[tuple[str, str]]:
candidates = self.routes.get(logical_model, [])
healthy = []
for provider, model_id in candidates:
key = f"{provider}:{model_id}"
if key not in self.health:
self.health[key] = ProviderHealth()
if self.health[key].is_healthy():
healthy.append((provider, model_id))
return healthyHandle streaming fallback without breaking clients
Streaming responses complicate fallback. If provider A streams 50 tokens then hits a rate limit, you cannot seamlessly switch to provider B — the client has already received partial output. You have three options:
- Fail fast on stream: Treat any stream error as non-retryable. Return the error to the client. Simple but wastes generated tokens.
- Buffer then switch: Buffer the entire stream in memory, and only send to client after completion. If rate limited, retry on next provider and send complete response. Adds latency equal to full generation time.
- Accept partial failure: Document that streaming requests may return partial output on fallback. Clients must handle incomplete responses.
Option 2 is the most common for production systems. Implement it with a bounded buffer and configurable timeout.
async def stream_with_fallback(
self,
request: ChatRequest,
providers: list[tuple[str, str]],
buffer_limit: int = 100_000, # chars
) -> AsyncGenerator[str, None]:
for provider, model_id in providers:
buffer = []
try:
async for chunk in self._stream_provider(provider, model_id, request):
buffer.append(chunk)
if sum(len(c) for c in buffer) > buffer_limit:
raise BufferError("Stream buffer exceeded limit")
# Success — yield buffered content
for chunk in buffer:
yield chunk
self.health[f"{provider}:{model_id}"].record_success()
return
except RateLimitError as e:
self.health[f"{provider}:{model_id}"].record_error("rate_limit")
# Wait before trying next provider
if e.retry_after_ms:
await asyncio.sleep(e.retry_after_ms / 1000)
continue
except Exception as e:
self.health[f"{provider}:{model_id}"].record_error("other")
raise
raise AllProvidersExhaustedError("No healthy providers available")Respect provider cache-control hints
Providers increasingly return cache-control headers indicating prompt prefix caching eligibility. OpenAI returns x-cache-status: hit|miss and x-remaining-tokens. Anthropic returns cache-control headers with max-age. When routing, prefer providers that signal a cache hit — it reduces latency and cost, and avoids consuming rate limit quota on cacheable work.
def select_provider_with_cache_preference(
self,
logical_model: str,
request: ChatRequest,
) -> list[tuple[str, str]]:
candidates = self.get_healthy_providers(logical_model)
# If request has cacheable prefix (system prompt + few-shot examples),
# prefer providers that recently returned cache hits
if self._has_cacheable_prefix(request):
scored = []
for provider, model_id in candidates:
key = f"{provider}:{model_id}"
cache_hit_rate = self._get_recent_cache_hit_rate(key)
scored.append((cache_hit_rate, provider, model_id))
scored.sort(reverse=True)
return [(p, m) for _, p, m in scored]
return candidatesImplement client-side routing directives
Your gateway should honor explicit routing hints from clients. A routing field in the request metadata lets callers pin to a provider, exclude providers, or set cost/latency ceilings. This is essential for workloads with strict compliance or latency requirements.
@dataclass
class RoutingDirective:
provider: str | None = None # pin to specific provider
exclude: list[str] | None = None # providers to avoid
max_cost_per_1k: float | None = None # USD
max_latency_ms: int | None = None
require_cache: bool = False # only use if cache hit likely
def apply_routing_directive(
self,
candidates: list[tuple[str, str]],
directive: RoutingDirective,
) -> list[tuple[str, str]]:
filtered = candidates
if directive.provider:
filtered = [(p, m) for p, m in filtered if p == directive.provider]
if directive.exclude:
filtered = [(p, m) for p, m in filtered if p not in directive.exclude]
if directive.max_cost_per_1k:
filtered = [
(p, m) for p, m in filtered
if self.estimate_cost(p, m) <= directive.max_cost_per_1k
]
return filteredMeter usage per provider for quota management
Rate limits are quotas over time windows. Without per-provider metering, you’ll discover quota exhaustion only when requests start failing. Track token consumption per provider and model, and proactively deprioritize providers approaching their limits.
import threading
from collections import defaultdict
class QuotaTracker:
def __init__(self, window_seconds: int = 60):
self.window = window_seconds
# provider -> model -> deque of (timestamp, tokens)
self.usage: dict[str, dict[str, deque[tuple[float, int]]]] = defaultdict(lambda: defaultdict(deque))
self.lock = threading.Lock()
# Configured limits per provider/model (tokens per window)
self.limits: dict[str, dict[str, int]] = {}
def record_usage(self, provider: str, model: str, tokens: int):
with self.lock:
now = time.time()
self.usage[provider][model].append((now, tokens))
self._prune(provider, model, now)
def get_utilization(self, provider: str, model: str) -> float:
with self.lock:
now = time.time()
self._prune(provider, model, now)
total = sum(t for _, t in self.usage[provider][model])
limit = self.limits.get(provider, {}).get(model, float('inf'))
return total / limit if limit > 0 else 0.0
def is_near_limit(self, provider: str, model: str, threshold: float = 0.8) -> bool:
return self.get_utilization(provider, model) >= threshold
def _prune(self, provider: str, model: str, now: float):
cutoff = now - self.window
dq = self.usage[provider][model]
while dq and dq[0][0] < cutoff:
dq.popleft()Common pitfalls and tradeoffs
Pitfall: Silent semantic drift. Different models produce different outputs for the same prompt. Falling back from GPT-4o to Claude Sonnet changes the response distribution. If your application depends on specific formatting, tool calling behavior, or refusal style, fallback breaks correctness. Mitigate by defining model equivalence classes with validated behavioral parity, not just capability parity.
Pitfall: Cascading failures. When one provider hits rate limits, traffic shifts to the next. That provider then hits its limits. The cascade continues until all providers are exhausted. Mitigate with circuit breakers: after N consecutive fallbacks, stop routing to that logical model and return a deterministic error. Let upstream callers implement their own retry with backoff.
Pitfall: Cost surprise. Fallback providers often have different pricing. A request routed from a $5/1M token model to a $15/1M token model increases cost 3x. Enforce cost ceilings in routing directives, and alert when fallback causes cost spikes.
Tradeoff: Latency vs. availability. Adding health checks, translation, and buffering adds 10-50ms overhead per request. For latency-sensitive paths, maintain a “fast path” that skips routing logic for a single preferred provider, and only engages multi-provider routing on explicit failure.
Tradeoff: Complexity vs. control. Building this yourself gives full control but requires maintaining provider SDKs, error mappings, and model catalogs. A managed gateway like n4n.ai handles provider normalization, fallback, and metering behind one OpenAI-compatible endpoint — you send one request format and get consistent behavior across 240+ models. The tradeoff is less visibility into per-provider internals.
Test your routing under load
Unit tests for routing logic are necessary but insufficient. You need integration tests that simulate real provider behavior:
# Test: fallback triggers on rate limit
async def test_fallback_on_rate_limit():
router = ModelRouter()
router.routes["test-model"] = [("provider_a", "model-a"), ("provider_b", "model-b")]
# Mock provider_a to return 429, provider_b to succeed
with mock_provider("provider_a", status=429, retry_after=100), \
mock_provider("provider_b", status=200, response={"content": "ok"}):
result = await router.complete(ChatRequest(model="test-model", messages=[...]))
assert result.content == "ok"
assert router.health["provider_a:model-a"].is_healthy() == False
# Test: cache hit preference
async def test_prefers_cache_hit():
router = ModelRouter()
router.routes["test-model"] = [("provider_a", "model-a"), ("provider_b", "model-b")]
router._record_cache_hit("provider_a:model-a", 0.9)
router._record_cache_hit("provider_b:model-b", 0.1)
request = ChatRequest(model="test-model", messages=[{"role": "system", "content": "x"*1000}])
providers = router.select_provider_with_cache_preference("test-model", request)
assert providers[0] == ("provider_a", "model-a")Run chaos tests: kill a provider mid-stream, simulate quota exhaustion, verify circuit breaker activation. The routing layer is infrastructure — treat it like one.
Start simple, evolve deliberately
You don’t need all of this on day one. Start with:
- A static provider priority list per model
- Basic 429 detection and single retry
- Per-provider error logging
Add health tracking, cache awareness, quota metering, and client directives as pain points emerge. The goal is not a perfect routing engine — it’s a system that degrades gracefully when the inevitable rate limit hits.