Single Provider (fragile):
 App → OpenAI → ❌ Down = App Down

Multi-Provider (resilient):
 App → Load Balancer → OpenAI (primary)
 → Claude (fallback)
 → Gemini (fallback)
 → DeepSeek (fallback)
 = Always available ✅

from openai import OpenAI
import time

class AIFallbackClient:
 def __init__(self):
 self.providers = [
 {
 "name": "OpenAI",
 "client": OpenAI(api_key="sk-openai-key"),
 "model": "gpt-4o",
 "healthy": True,
 "last_error": None
 },
 {
 "name": "Anthropic (via OpenAI SDK)",
 "client": OpenAI(
 api_key="sk-anthropic-key",
 base_url="https://api.anthropic.com/v1/"
 ),
 "model": "claude-sonnet-4-20250514",
 "healthy": True,
 "last_error": None
 },
 {
 "name": "DeepSeek",
 "client": OpenAI(
 api_key="sk-deepseek-key",
 base_url="https://api.deepseek.com/v1"
 ),
 "model": "deepseek-chat",
 "healthy": True,
 "last_error": None
 }
 ]
 
 def chat(self, messages, **kwargs):
 errors = []
 
 for provider in self.providers:
 if not provider["healthy"]:
 # Check if enough time has passed to retry
 if time.time() - provider["last_error"] < 60:
 continue # Skip unhealthy providers for 60s
 provider["healthy"] = True # Reset after cooldown
 
 try:
 response = provider["client"].chat.completions.create(
 model=provider["model"],
 messages=messages,
 timeout=30,
 **kwargs
 )
 return response
 except Exception as e:
 provider["healthy"] = False
 provider["last_error"] = time.time()
 errors.append(f"{provider['name']}: {str(e)}")
 continue
 
 raise Exception(f"All providers failed: {'; '.join(errors)}")

# Usage
client = AIFallbackClient()
response = client.chat([
 {"role": "user", "content": "Hello, world!"}
])

import random
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ProviderStats:
 name: str
 weight: float # 0-1, higher = more traffic
 avg_latency: float = 0.0
 error_count: int = 0
 success_count: int = 0
 last_error_time: float = 0.0
 circuit_open: bool = False
 
 @property
 def error_rate(self):
 total = self.error_count + self.success_count
 return self.error_count / total if total > 0 else 0

class WeightedLoadBalancer:
 def __init__(self, providers: list):
 self.providers = providers
 self.stats = {
 p["name"]: ProviderStats(name=p["name"], weight=p.get("weight", 1.0))
 for p in providers
 }
 
 def select_provider(self):
 """Select a provider using weighted random selection."""
 available = [
 (p, self.stats[p["name"]]) 
 for p in self.providers 
 if not self.stats[p["name"]].circuit_open
 ]
 
 if not available:
 # All circuits open — reset the one with oldest error
 oldest = min(self.stats.values(), key=lambda s: s.last_error_time)
 oldest.circuit_open = False
 return next(p for p in self.providers if p["name"] == oldest.name)
 
 # Weighted random selection
 total_weight = sum(s.weight for _, s in available)
 r = random.uniform(0, total_weight)
 cumulative = 0
 
 for provider, stats in available:
 cumulative += stats.weight
 if r <= cumulative:
 return provider
 
 return available[-1][0]
 
 def record_success(self, name: str, latency: float):
 stats = self.stats[name]
 stats.success_count += 1
 stats.avg_latency = (stats.avg_latency * 0.9) + (latency * 0.1)
 # Increase weight for well-performing providers
 stats.weight = min(stats.weight * 1.05, 2.0)
 
 def record_failure(self, name: str):
 stats = self.stats[name]
 stats.error_count += 1
 stats.last_error_time = time.time()
 stats.weight = max(stats.weight * 0.5, 0.1) # Reduce weight
 
 if stats.error_rate > 0.5: # >50% error rate
 stats.circuit_open = True

from openai import OpenAI

# Crazyrouter handles load balancing, failover, and rate limits
# across 300+ models automatically
client = OpenAI(
 api_key="YOUR_CRAZYROUTER_KEY",
 base_url="https://crazyrouter.com/v1"
)

# Just specify the model — Crazyrouter handles the rest
response = client.chat.completions.create(
 model="gpt-4o", # Automatically fails over if OpenAI is down
 messages=[{"role": "user", "content": "Hello!"}]
)

import OpenAI from 'openai';

const client = new OpenAI({
 apiKey: process.env.CRAZYROUTER_API_KEY,
 baseURL: 'https://crazyrouter.com/v1'
});

// Same code works for any model — failover is automatic
async function reliableChat(messages) {
 try {
 return await client.chat.completions.create({
 model: 'gpt-4o',
 messages
 });
 } catch (error) {
 // Even this manual fallback is rarely needed with Crazyrouter
 console.warn('Primary model failed, trying fallback');
 return await client.chat.completions.create({
 model: 'claude-sonnet-4-20250514',
 messages
 });
 }
}

import time
from enum import Enum

class CircuitState(Enum):
 CLOSED = "closed" # Normal operation
 OPEN = "open" # Blocking requests
 HALF_OPEN = "half_open" # Testing with limited requests

class CircuitBreaker:
 def __init__(self, failure_threshold=5, recovery_timeout=60):
 self.failure_threshold = failure_threshold
 self.recovery_timeout = recovery_timeout
 self.state = CircuitState.CLOSED
 self.failure_count = 0
 self.last_failure_time = 0
 self.success_count = 0
 
 def can_execute(self) -> bool:
 if self.state == CircuitState.CLOSED:
 return True
 elif self.state == CircuitState.OPEN:
 if time.time() - self.last_failure_time > self.recovery_timeout:
 self.state = CircuitState.HALF_OPEN
 return True
 return False
 elif self.state == CircuitState.HALF_OPEN:
 return True
 
 def record_success(self):
 if self.state == CircuitState.HALF_OPEN:
 self.success_count += 1
 if self.success_count >= 3: # 3 successful requests to close
 self.state = CircuitState.CLOSED
 self.failure_count = 0
 self.success_count = 0
 
 def record_failure(self):
 self.failure_count += 1
 self.last_failure_time = time.time()
 
 if self.failure_count >= self.failure_threshold:
 self.state = CircuitState.OPEN
 
 if self.state == CircuitState.HALF_OPEN:
 self.state = CircuitState.OPEN

# Usage with multiple providers
breakers = {
 "openai": CircuitBreaker(),
 "anthropic": CircuitBreaker(),
 "deepseek": CircuitBreaker()
}

def route_request(messages, requirements):
 """Route to the optimal model based on request needs."""
 
 total_tokens = estimate_tokens(messages)
 
 if requirements.get("reasoning"):
 # Complex reasoning tasks
 return "deepseek-r2" if total_tokens < 64000 else "gemini-2.5-pro"
 
 elif requirements.get("vision"):
 # Image understanding
 return "gpt-4o" if total_tokens < 128000 else "gemini-2.5-flash"
 
 elif requirements.get("long_context") and total_tokens > 200000:
 # Very long context
 return "gemini-2.5-pro" # 1M context window
 
 elif requirements.get("speed"):
 # Latency-sensitive
 return "gpt-4o-mini"
 
 elif requirements.get("cost_sensitive"):
 # Budget-friendly
 return "deepseek-chat"
 
 else:
 # Default: best quality-price ratio
 return "claude-sonnet-4-20250514"

import logging
from datetime import datetime

class AIMetrics:
 def __init__(self):
 self.requests = []
 
 def log_request(self, provider, model, latency, tokens, success, error=None):
 self.requests.append({
 "timestamp": datetime.utcnow().isoformat(),
 "provider": provider,
 "model": model,
 "latency_ms": latency,
 "input_tokens": tokens.get("input", 0),
 "output_tokens": tokens.get("output", 0),
 "success": success,
 "error": str(error) if error else None,
 "cost": self.calculate_cost(model, tokens)
 })
 
 def get_provider_health(self):
 """Get health status of each provider (last 100 requests)."""
 recent = self.requests[-100:]
 providers = set(r["provider"] for r in recent)
 
 health = {}
 for provider in providers:
 provider_requests = [r for r in recent if r["provider"] == provider]
 success_rate = sum(1 for r in provider_requests if r["success"]) / len(provider_requests)
 avg_latency = sum(r["latency_ms"] for r in provider_requests) / len(provider_requests)
 health[provider] = {
 "success_rate": f"{success_rate:.1%}",
 "avg_latency_ms": f"{avg_latency:.0f}",
 "total_requests": len(provider_requests)
 }
 
 return health