Every AI application that survives contact with production learns the same lessons: providers rate-limit you at 2 a.m., someone pastes a 400k-token document into your prompt, and a "cheap" batch job quietly burns your monthly budget in an afternoon. The fix isn't vigilance — it's a self-healing pipeline that anticipates these failure modes and recovers without human intervention.
In this tutorial, you'll build one in about 150 lines of Python. It handles the four classic failure modes:
- Rate limits / transient errors -> exponential backoff with jitter
- Provider outages -> automatic fallback to a backup model
- Context overflow -> automatic input shortening and retry
- Budget overruns -> a hard cost cap enforced in code, not on a dashboard
Step 0: The Setup
We'll use the OpenAI-compatible API format, which works across virtually every provider. Grab a key from Qubax, where models are priced on an open market (usually well below retail), and set:
export AI_API_KEY="your-key-here"
export AI_BASE_URL="https://api.qubax.ai/v1"import os, time, random, requests
BASE_URL = os.environ["AI_BASE_URL"]
API_KEY = os.environ["AI_API_KEY"]
def chat(messages, model, max_tokens=1024, timeout=90):
r = requests.post(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": model, "messages": messages, "max_tokens": max_tokens},
timeout=timeout,
)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]Step 1: Exponential Backoff with Jitter
The first shield. Retry only errors that are worth retrying (429 rate limit, 5xx server errors, timeouts), and never hammer a struggling server:
RETRYABLE = (429, 500, 502, 503, 504)
def with_backoff(fn, max_attempts=5, base_delay=1.0):
for attempt in range(max_attempts):
try:
return fn()
except requests.HTTPError as e:
code = e.response.status_code if e.response is not None else 0
if code not in RETRYABLE or attempt == max_attempts - 1:
raise
except (requests.Timeout, requests.ConnectionError):
if attempt == max_attempts - 1:
raise
# exponential backoff + jitter: 1s, 2s, 4s, 8s... +/-50%
delay = base_delay * (2 ** attempt) * random.uniform(0.5, 1.5)
print(f"[retry] attempt {attempt+1} failed, sleeping {delay:.1f}s")
time.sleep(delay)The jitter matters. Without it, a thousand clients that all got rate-limited retry simultaneously and re-trigger the limit. Randomized delays spread them out.
Step 2: Model Fallback Chains
When one model is down (or suddenly slow), fall back to a cheaper backup instead of failing. Define chains from strongest to cheapest:
FALLBACK_CHAINS = {
"quality": ["gpt-5.6-sol", "claude-sonnet-5", "glm-5.2"],
"cheap": ["glm-5.3-flash", "deepseek-v4-flash", "gpt-5-nano"],
"reasoning":["gpt-6-astra", "glm-5.2", "deepseek-v4-pro"],
}
def chat_with_fallback(messages, chain_name="quality", **kwargs):
errors = []
for model in FALLBACK_CHAINS[chain_name]:
try:
return with_backoff(lambda: chat(messages, model, **kwargs)), model
except Exception as e:
errors.append(f"{model}: {e}")
print(f"[fallback] {model} failed, trying next")
raise RuntimeError("All models failed: " + "; ".join(errors))Now a GLM outage degrades you to DeepSeek pricing instead of a downtime page. Note the pattern: backoff inside fallback — each model gets its own retry budget before you give up on it.
Step 3: Handle Context Overflow Gracefully
Different models accept different context sizes. Rather than failing when a user pastes a novel, shorten the middle (the least important part for most tasks) and retry:
def tokens_estimate(text):
return len(text) // 3 # ~3 chars/token, good enough for guarding
MAX_CONTEXT = {"gpt-5.6-sol": 350_000, "glm-5.2": 120_000, "glm-5.3-flash": 120_000}
def shorten_middle(text, max_tokens):
if tokens_estimate(text) <= max_tokens:
return text
half = (max_tokens * 3) // 2
return text[:half] + "\n\n[... middle omitted ...]\n\n" + text[-half:]
def safe_chat(messages, model, **kwargs):
budget = MAX_CONTEXT.get(model, 100_000) - 2_000 # leave room for output
fixed = [
{**m, "content": shorten_middle(m["content"], budget)} if m["role"] == "user"
else m
for m in messages
]
return chat(messages=fixed, model=model, **kwargs)A 400k-token paste into a 120k model now becomes a successful (if lossy) response instead of a 500 error.
Step 4: Hard Cost Caps — the Shield That Saves Money
The most important one. Track actual token usage from the API response and enforce a budget in code. On Qubax, prices are shown per million tokens on every model page, so you can wire in exact numbers:
PRICES = { # USD per 1M tokens: (input, output)
"gpt-6-astra": (1.05, 4.21),
"glm-5.2": (0.049, 0.196),
"glm-5.3-flash": (0.0077, 0.031),
}
class BudgetExceeded(Exception): pass
class CostGuard:
def __init__(self, monthly_usd=10.0):
self.limit = monthly_usd
self.spent = 0.0
def track(self, model, usage):
p_in, p_out = PRICES[model]
cost = (usage["prompt_tokens"] * p_in +
usage["completion_tokens"] * p_out) / 1_000_000
self.spent += cost
if self.spent > self.limit:
raise BudgetExceeded(
f"Budget ${self.limit:.2f} exceeded (spent ${self.spent:.4f})"
)
return costPro tip: make the cap part of your fallback logic — when the quality chain exceeds budget, degrade to the cheap chain instead of erroring:
guard = CostGuard(monthly_usd=10.0)
def resilient_chat(messages, chain="quality"):
try:
result, model = chat_with_fallback(messages, chain)
guard.track(model, last_usage) # usage comes from the API response
return result
except BudgetExceeded:
print("[budget] downgrading to cheap chain")
result, model = chat_with_fallback(messages, "cheap")
guard.track(model, last_usage)
return result
# Survives: rate limits (backoff), outages (fallback), huge inputs (shorten), overspend (cap)
print(resilient_chat([{"role": "user", "content": "Explain ADFGVX ciphers in 3 sentences"}]))Production Hardening Checklist
- Persist cost state — the in-memory
guardresets on restart; storespentin Redis or Postgres - Log every fallback — if 40% of traffic is hitting backup models, you should know
- Set per-request timeouts — a hung stream is worse than a fast failure
- Alert on retry storms — a spike in 429s often means your concurrency limit changed
- Prefer routers — if this is getting complex, a model router over an aggregator like Qubax gives you fallback + price competition in one endpoint; see the docs for the OpenAI-compatible base URL
FAQ
Why exponential backoff with jitter instead of fixed retries?
Fixed intervals synchronize all failing clients into retry waves (the "thundering herd"), which re-triggers rate limits. Exponential spacing gives the server room to recover; jitter desynchronizes clients.
Should I fall back on every error?
No. Fall back on infrastructure errors (5xx, timeouts) and persistent 429s. Don't fall back on 400-level errors like invalid API keys or malformed requests — those fail identically on the next model.
How accurate is the cost tracking?
Exact, if you use the usage object from the API response — it reports actual billed tokens. The tokens_estimate heuristic is only used for pre-flight shortening, not billing.
Where can I check model prices for my PRICES table?
Every model on Qubax's marketplace lists its live per-million-token price, updated as compute providers compete on the open market — usually well below first-party retail.
