Module 5: OpenRouter - Introduction
Fallback Strategies with OpenRouter
Overview
You'll implement robust fallback for high availability: if one model fails, it automatically tries another.
Time: 20 minutes
Difficulty: Medium
🎯 Objectives
- ✅ Fallback chain (multiple models)
- ✅ Retry with exponential backoff
- ✅ Model health check
- ✅ Failure logging
🔄 Basic Fallback
from openai import OpenAI
import os
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY")
)
def chat_with_fallback(prompt: str) -> str:
"""Fallback chain: if one fails, try the next."""
models = [
"openai/gpt-3.5-turbo", # Primary
"anthropic/claude-3-haiku", # Fallback 1
"mistralai/mixtral-8x7b-instruct" # Fallback 2
]
for model in models:
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
timeout=10.0
)
print(f"✅ Success with: {model}")
return response.choices[0].message.content
except Exception as e:
print(f"❌ {model} failed: {e}")
continue
return "Error: All models failed"
# Test
print(chat_with_fallback("Hi"))⚡ Fallback with Exponential Backoff
import time
def chat_with_smart_fallback(prompt: str, max_retries: int = 3) -> str:
"""Fallback + retry with backoff."""
models = [
"openai/gpt-3.5-turbo",
"anthropic/claude-3-haiku",
"mistralai/mixtral-8x7b-instruct"
]
for model in models:
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
print(f"✅ {model} (attempt {attempt + 1})")
return response.choices[0].message.content
except Exception as e:
wait = 2 ** attempt # 1s, 2s, 4s
print(f"⚠️ {model} attempt {attempt + 1} failed, waiting {wait}s...")
if attempt < max_retries - 1:
time.sleep(wait)
print(f"❌ {model} failed after {max_retries} attempts")
return "Error: All models exhausted"🎯 Smart Fallback (Cheap → Expensive)
def smart_fallback(prompt: str) -> dict:
"""Fallback chain ordered by cost."""
# Ordered by price (cheap → expensive)
models = [
{"name": "mistralai/mixtral-8x7b-instruct", "cost": 0.24},
{"name": "anthropic/claude-3-haiku", "cost": 0.25},
{"name": "openai/gpt-3.5-turbo", "cost": 0.50},
{"name": "openai/gpt-4-turbo", "cost": 10.00}
]
for model_info in models:
model = model_info["name"]
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
return {
"response": response.choices[0].message.content,
"model_used": model,
"cost_per_million": model_info["cost"],
"tokens": response.usage.total_tokens
}
except Exception as e:
print(f"⚠️ {model} (${model_info['cost']}/1M) failed")
continue
return {"error": "All models failed"}
# Test
result = smart_fallback("Hi")
print(f"Response: {result['response']}")
print(f"Model: {result['model_used']}")📊 Fallback with Logging
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FallbackChat:
"""Chat with fallback and full logging."""
def __init__(self):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY")
)
self.models = [
"openai/gpt-3.5-turbo",
"anthropic/claude-3-haiku",
"mistralai/mixtral-8x7b-instruct"
]
def chat(self, prompt: str) -> str:
"""Chat with robust fallback."""
for i, model in enumerate(self.models):
try:
logger.info(f"Trying model {i+1}/{len(self.models)}: {model}")
response = self.client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
logger.info(f"✅ Success with {model}")
return response.choices[0].message.content
except Exception as e:
logger.error(f"❌ {model} failed: {type(e).__name__} - {e}")
if i == len(self.models) - 1:
logger.critical("All models exhausted!")
raise
return None
# Usage
bot = FallbackChat()
try:
response = bot.chat("Hi")
print(response)
except Exception as e:
print(f"Fatal error: {e}")🔍 Model Health Check
def check_model_health(model: str) -> bool:
"""Check whether a model is available."""
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": "test"}],
max_tokens=5,
timeout=5.0
)
return True
except Exception:
return False
def get_healthy_models() -> list:
"""Return the available models."""
all_models = [
"openai/gpt-3.5-turbo",
"anthropic/claude-3-haiku",
"mistralai/mixtral-8x7b-instruct",
"google/gemini-pro"
]
healthy = []
for model in all_models:
if check_model_health(model):
healthy.append(model)
print(f"✅ {model}")
else:
print(f"❌ {model}")
return healthy
# Pre-check before using
healthy_models = get_healthy_models()
# Use only healthy models📊 Metrics and Monitoring
class MonitoredFallbackChat:
"""Chat with metrics."""
def __init__(self):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY")
)
self.stats = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"by_model": {}
}
def chat(self, prompt: str, models: list) -> str:
"""Chat with metrics tracking."""
self.stats["total_requests"] += 1
for model in models:
try:
response = self.client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
# Track success
self.stats["successful"] += 1
if model not in self.stats["by_model"]:
self.stats["by_model"][model] = {"success": 0, "failed": 0}
self.stats["by_model"][model]["success"] += 1
return response.choices[0].message.content
except Exception as e:
# Track failure
if model not in self.stats["by_model"]:
self.stats["by_model"][model] = {"success": 0, "failed": 0}
self.stats["by_model"][model]["failed"] += 1
self.stats["failed"] += 1
return None
def print_stats(self):
"""Show the statistics."""
print("\n=== Metrics ===")
print(f"Total requests: {self.stats['total_requests']}")
print(f"Successful: {self.stats['successful']}")
print(f"Failed: {self.stats['failed']}")
print("\nBy model:")
for model, stats in self.stats["by_model"].items():
total = stats["success"] + stats["failed"]
success_rate = (stats["success"] / total * 100) if total > 0 else 0
print(f" {model}: {success_rate:.1f}% success ({stats['success']}/{total})")
# Usage
bot = MonitoredFallbackChat()
models = ["openai/gpt-3.5-turbo", "anthropic/claude-3-haiku"]
for _ in range(10):
bot.chat("Test query", models)
bot.print_stats()✅ Summary
Fallback strategies:
- Chain (try multiple models)
- Exponential backoff (retry with delays)
- Cheap-to-expensive ordering
- Health checks pre-flight
- Comprehensive logging
- Metrics tracking
High availability: 99.9%+ with 3+ models
Next: 06-model-selection-auto.md
You'll implement intelligent model auto-selection based on query type.
Time: 25 min