When building production AI applications, a single API provider can become a bottleneck. Imagine your customer service chatbot suddenly returning errors because your primary AI model provider is experiencing downtime—that's lost revenue and frustrated users. This is precisely why implementing API load balancing with intelligent routing and automatic failover has become essential for modern AI-powered applications.
In this comprehensive guide, I'll walk you through building a robust multi-model API routing system from scratch. Whether you're a complete beginner with no prior API experience or an intermediate developer looking to optimize your infrastructure, you'll find actionable steps, real code examples, and lessons learned from real-world deployments. Sign up here to get started with competitive pricing and free credits.
What is API Load Balancing and Why Do You Need It?
Think of API load balancing like a traffic controller at a busy airport. Instead of all planes (requests) lining up at one runway (API provider), the controller distributes them across multiple runways based on current conditions, capacity, and priority. This ensures no single runway gets overwhelmed, flights (requests) reach their destination even if one runway closes, and the overall system operates efficiently.
For AI APIs specifically, load balancing becomes critical because:
- Cost Optimization: Different models have vastly different pricing. DeepSeek V3.2 costs $0.42 per million tokens while Claude Sonnet 4.5 runs $15 per million tokens. Smart routing can route simple queries to cheaper models.
- Reliability: Even top-tier providers experience outages. Automatic failover ensures your application keeps running.
- Performance: HolySheep AI delivers less than 50ms latency, but geographic distribution and provider diversity can improve response times further.
- Rate Limit Management: Bypass individual provider rate limits by distributing requests.
Understanding the Core Concepts
The Three Pillars of API Load Balancing
1. Routing involves directing each incoming request to the most appropriate backend based on criteria like model capability, current load, cost, or latency requirements. Think of it as an intelligent traffic light system.
2. Health Checking continuously monitors each API endpoint to determine availability. When a provider shows signs of trouble (slow responses, increased errors), the system marks it as unhealthy and routes traffic elsewhere.
3. Failover automatically redirects requests from failed or degraded endpoints to healthy alternatives. This happens seamlessly without user intervention.
Load Balancing Strategies
Different strategies suit different use cases. Round Robin distributes requests evenly across all healthy backends—simple but not always optimal. Weighted Round Robin assigns different traffic percentages to backends based on capacity or cost—ideal for using premium models for 20% of traffic while routing 80% to cost-effective alternatives. Least Connections sends new requests to the backend with the fewest active connections—excellent for requests with varying processing times. Response Time Weighted routes more traffic to faster-responding backends—optimizes for user experience.
Building Your First Load Balancer: Step-by-Step
Prerequisites
Before we dive into code, ensure you have Python 3.8+ installed, an API key from HolySheep AI (which supports WeChat and Alipay payments with rates starting at ¥1=$1, saving 85%+ compared to ¥7.3 alternatives), and the requests library installed via pip install requests.
Step 1: Creating a Basic Health Monitor
Health monitoring is the foundation of any load balancing system. Without it, you risk sending requests to failing endpoints.
# health_monitor.py
import time
import asyncio
from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class HealthStatus:
"""Stores health information for a single backend endpoint."""
endpoint: str
is_healthy: bool = True
response_time_ms: float = 0.0
consecutive_failures: int = 0
last_success: datetime = None
last_failure: datetime = None
total_requests: int = 0
successful_requests: int = 0
class HealthMonitor:
"""
Monitors the health of multiple API endpoints.
Implements circuit breaker pattern to prevent cascading failures.
"""
def __init__(self, failure_threshold: int = 3, recovery_timeout: int = 30):
"""
Initialize health monitor.
Args:
failure_threshold: Number of consecutive failures before marking unhealthy
recovery_timeout: Seconds to wait before retrying a failed endpoint
"""
self.endpoints: Dict[str, HealthStatus] = {}
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
def register_endpoint(self, name: str, url: str):
"""Register a new backend endpoint for monitoring."""
self.endpoints[name] = HealthStatus(endpoint=url)
print(f"✓ Registered endpoint: {name} -> {url}")
async def check_health(self, name: str) -> HealthStatus:
"""Perform health check on a single endpoint."""
status = self.endpoints.get(name)
if not status:
return None
start_time = time.time()
try:
# Simple health check - try to get a response
import requests
response = requests.get(
status.endpoint.rstrip('/') + '/health',
timeout=5
)
status.response_time_ms = (time.time() - start_time) * 1000
status.consecutive_failures = 0
status.last_success = datetime.now()
status.successful_requests += 1
status.total_requests += 1
status.is_healthy = True
except Exception as e:
status.consecutive_failures += 1
status.last_failure = datetime.now()
status.total_requests += 1
if status.consecutive_failures >= self.failure_threshold:
status.is_healthy = False
print(f"⚠ {name} marked unhealthy after {status.consecutive_failures} failures")
return status
async def check_all(self):
"""Run health checks on all registered endpoints concurrently."""
tasks = [self.check_health(name) for name in self.endpoints]
await asyncio.gather(*tasks)
def get_healthy_endpoints(self) -> List[str]:
"""Return list of currently healthy endpoint names."""
return [
name for name, status in self.endpoints.items()
if status.is_healthy
]
def should_attempt_recovery(self, name: str) -> bool:
"""Check if enough time has passed to attempt recovering an unhealthy endpoint."""
status = self.endpoints.get(name)
if not status or status.is_healthy:
return False
if status.last_failure:
elapsed = (datetime.now() - status.last_failure).total_seconds()
return elapsed >= self.recovery_timeout
return True
Example usage
if __name__ == "__main__":
monitor = HealthMonitor(failure_threshold=3, recovery_timeout=30)
monitor.register_endpoint("holysheep-primary", "https://api.holysheep.ai/v1")
monitor.register_endpoint("holysheep-backup", "https://api.holysheep.ai/v1")
monitor.register_endpoint("competitor-api", "https://api.competitor.ai/v1")
print(f"Healthy endpoints: {monitor.get_healthy_endpoints()}")Step 2: Implementing Intelligent Request Routing
Now we need a router that intelligently distributes requests based on multiple factors including model availability, current load, response times, and cost considerations.
# intelligent_router.py
import asyncio
import hashlib
import random
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from health_monitor import HealthMonitor, HealthStatus
class RoutingStrategy(Enum):
"""Available routing strategies."""
ROUND_ROBIN = "round_robin"
WEIGHTED = "weighted"
LEAST_LATENCY = "least_latency"
LEAST_LOAD = "least_load"
COST_OPTIMIZED = "cost_optimized"
INTELLIGENT = "intelligent"
@dataclass
class ModelInfo:
"""Information about a specific AI model."""
name: str
provider: str
endpoint: str
cost_per_million_tokens: float # USD
average_latency_ms: float
max_tokens: int
capabilities: List[str] = field(default_factory=list)
current_load: int = 0
@dataclass
class RouteResult:
"""Result of routing decision."""
model: ModelInfo
endpoint: str
strategy_used: str
fallback_used: bool = False
class IntelligentRouter:
"""
Routes API requests to optimal backends based on strategy and conditions.
"""
def __init__(self, health_monitor: HealthMonitor):
self.health_monitor = health_monitor
self.models: Dict[str, ModelInfo] = {}
self.round_robin_counters: Dict[str, int] = {}
self.request_log: List[Tuple[str, str, float]] = [] # (model, timestamp, duration)
def register_model(self, model: ModelInfo):
"""Register a new model for routing."""
self.models[model.name] = model
self.round_robin_counters[model.name] = 0
print(f"✓ Registered model: {model.name} (${model.cost_per_million_tokens}/MTok)")
def register_default_models(self):
"""Register commonly used models with accurate 2026 pricing."""
models = [
# HolySheep AI models (cost-effective options)
ModelInfo(
name="gpt-4.1",
provider="holysheep",
endpoint="https://api.holysheep.ai/v1/chat/completions",
cost_per_million_tokens=8.0, # $8/MTok
average_latency_ms=45,
max_tokens=128000,
capabilities=["reasoning", "coding", "analysis", "creative"]
),
ModelInfo(
name="claude-sonnet-4.5",
provider="holysheep",
endpoint="https://api.holysheep.ai/v1/chat/completions",
cost_per_million_tokens=15.0, # $15/MTok
average_latency_ms=52,
max_tokens=200000,
capabilities=["reasoning", "writing", "analysis", "long-context"]
),
ModelInfo(
name="gemini-2.5-flash",
provider="holysheep",
endpoint="https://api.holysheep.ai/v1/chat/completions",
cost_per_million_tokens=2.50, # $2.50/MTok
average_latency_ms=38,
max_tokens=1000000,
capabilities=["fast", "multimodal", "cost-efficient"]
),
ModelInfo(
name="deepseek-v3.2",
provider="holysheep",
endpoint="https://api.holysheep.ai/v1/chat/completions",
cost_per_million_tokens=0.42, # $0.42/MTok
average_latency_ms=42,
max_tokens=64000,
capabilities=["coding", "math", "reasoning", "budget-friendly"]
),
]
for model in models:
self.register_model(model)
print(f"📊 Registered {len(models)} models with optimized routing")
def _get_healthy_models(self) -> List[ModelInfo]:
"""Filter to only healthy models that are available."""
healthy = []
provider_health = self.health_monitor.get_healthy_endpoints()
for model in self.models.values():
# Check if the model's provider is healthy
provider_key = f"{model.provider}-primary"
if model.provider == "holysheep" and "holysheep-primary" in provider_health:
healthy.append(model)
elif model.provider in provider_health:
healthy.append(model)
return healthy
def route(self,
strategy: RoutingStrategy = RoutingStrategy.INTELLIGENT,
required_capability: Optional[str] = None,
max_cost_per_1k: Optional[float] = None) -> RouteResult:
"""
Route a request to the optimal model based on strategy.
Args:
strategy: Routing strategy to use
required_capability: Filter models by required capability
max_cost_per_1k: Maximum acceptable cost per 1000 tokens
Returns:
RouteResult with chosen model and endpoint
"""
candidates = self._get_healthy_models()
# Filter by capability if specified
if required_capability:
candidates = [m for m in candidates if required_capability in m.capabilities]
# Filter by cost if specified
if max_cost_per_1k is not None:
candidates = [m for m in candidates if m.cost_per_million_tokens <= max_cost_per_1k * 1000]
if not candidates:
raise Exception("No healthy models available for routing")
if strategy == RoutingStrategy.ROUND_ROBIN:
return self._round_robin_route(candidates)
elif strategy == RoutingStrategy.WEIGHTED:
return self._weighted_route(candidates)
elif strategy == RoutingStrategy.LEAST_LATENCY:
return self._least_latency_route(candidates)
elif strategy == RoutingStrategy.COST_OPTIMIZED:
return self._cost_optimized_route(candidates)
else: # INTELLIGENT
return self._intelligent_route(candidates)
def _round_robin_route(self, candidates: List[ModelInfo]) -> RouteResult:
"""Simple round-robin routing."""
# Rotate through models
model_name = list(self.models.keys())[0] # Simplified for demo
model = self.models[model_name]
return RouteResult(model=model, endpoint=model.endpoint, strategy_used="round_robin")
def _least_latency_route(self, candidates: List[ModelInfo]) -> RouteResult:
"""Route to the fastest responding model."""
fastest = min(candidates, key=lambda m: m.average_latency_ms)
return RouteResult(model=fastest, endpoint=fastest.endpoint, strategy_used="least_latency")
def _cost_optimized_route(self, candidates: List[ModelInfo]) -> RouteResult:
"""Route to the most cost-effective model."""
cheapest = min(candidates, key=lambda m: m.cost_per_million_tokens)
return RouteResult(model=cheapest, endpoint=cheapest.endpoint, strategy_used="cost_optimized")
def _weighted_route(self, candidates: List[ModelInfo]) -> RouteResult:
"""Route based on weighted probability (lower cost = higher probability)."""
# Inverse cost weighting: cheaper models get more traffic
weights = []
for model in candidates:
# Weight is inverse of cost (cheaper = higher weight)
weight = 100 / model.cost_per_million_tokens
weights.append(weight)
# Normalize weights
total = sum(weights)
normalized = [w / total for w in weights]
# Weighted random selection
selected = random.choices(candidates, weights=normalized, k=1)[0]
return RouteResult(model=selected, endpoint=selected.endpoint, strategy_used="weighted")
def _intelligent_route(self, candidates: List[ModelInfo]) -> RouteResult:
"""
Intelligent routing considers multiple factors:
1. Cost (40% weight)
2. Latency (30% weight)
3. Current load (30% weight)
"""
scores = []
# Normalize factors
max_cost = max(m.cost_per_million_tokens for m in candidates)
min_cost = min(m.cost_per_million_tokens for m in candidates)
max_latency = max(m.average_latency_ms for m in candidates)
min_latency = min(m.average_latency_ms for m in candidates)
max_load = max(m.current_load for m in candidates) if candidates else 1
for model in candidates:
# Cost score: lower is better (invert so higher = better)
cost_score = (1 - (model.cost_per_million_tokens - min_cost) / (max_cost - min_cost + 0.01)) * 100
# Latency score: lower is better
latency_score = (1 - (model.average_latency_ms - min_latency) / (max_latency - min_latency + 0.01)) * 100
# Load score: lower load is better
load_score = (1 - model.current_load / (max_load + 1)) * 100
# Weighted final score
final_score = cost_score * 0.4 + latency_score * 0.3 + load_score * 0.3
scores.append(final_score)
best_idx = scores.index(max(scores))
best_model = candidates[best_idx]
return RouteResult(
model=best_model,
endpoint=best_model.endpoint,
strategy_used="intelligent",
fallback_used=False
)
Example usage
if __name__ == "__main__":
# Initialize components
health = HealthMonitor()
router = IntelligentRouter(health)
# Register default models
router.register_default_models()
# Test different routing strategies
print("\n--- Routing Test Results ---")
for strategy in [
RoutingStrategy.LEAST_LATENCY,
RoutingStrategy.COST_OPTIMIZED,
RoutingStrategy.INTELLIGENT
]:
result = router.route(strategy=strategy)
print(f"{strategy.value}: {result.model.name} (${result.model.cost_per_million_tokens}/MTok)")Step 3: Implementing Automatic Failover
The most critical component for production reliability is automatic failover. When a primary endpoint fails, requests should seamlessly route to healthy alternatives without user-facing errors.
# failover_manager.py
import asyncio
import time
from typing import Dict, List, Optional, Callable, Any
from dataclasses import dataclass
from datetime import datetime
from intelligent_router import IntelligentRouter, RoutingStrategy, RouteResult
@dataclass
class FailoverConfig:
"""Configuration for failover behavior."""
max_retries: int = 3
retry_delay_seconds: float = 1.0
exponential_backoff: bool = True
circuit_breaker_threshold: int = 5
circuit_breaker_timeout: int = 60
class CircuitBreaker:
"""
Implements circuit breaker pattern to prevent cascading failures.
States: CLOSED (normal) -> OPEN (failing) -> HALF_OPEN (testing)
"""
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failures = 0
self.last_failure_time: Optional[datetime] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_success(self):
"""Reset failure count on successful request."""
self.failures = 0
self.state = "CLOSED"
def record_failure(self):
"""Record a failure and potentially open the circuit."""
self.failures += 1
self.last_failure_time = datetime.now()
if self.failures >= self.failure_threshold:
self.state = "OPEN"
print(f"⚡ Circuit breaker OPENED after {self.failures} failures")
def can_attempt(self) -> bool:
"""Check if a request attempt is allowed."""
if self.state == "CLOSED":
return True
if self.state == "OPEN":
if self.last_failure_time:
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
if elapsed >= self.timeout:
self.state = "HALF_OPEN"
print("🔄 Circuit breaker transitioning to HALF_OPEN")
return True
return False
# HALF_OPEN: allow limited attempts
return True
class FailoverManager:
"""
Manages automatic failover with retry logic and circuit breakers.
"""
def __init__(
self,
router: IntelligentRouter,
config: Optional[FailoverConfig] = None
):
self.router = router
self.config = config or FailoverConfig()
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
self.request_history: List[Dict] = []
self.fallback_chain: List[str] = []
def _init_circuit_breakers(self):
"""Initialize circuit breakers for each model."""
for model_name in self.router.models.keys():
self.circuit_breakers[model_name] = CircuitBreaker(
failure_threshold=self.config.circuit_breaker_threshold,
timeout=self.config.circuit_breaker_timeout
)
async def call_with_failover(
self,
prompt: str,
fallback_chain: Optional[List[str]] = None,
on_failure: Optional[Callable] = None,
**kwargs
) -> Dict[str, Any]:
"""
Make an API call with automatic failover.
Args:
prompt: The text prompt to send to the model
fallback_chain: Priority list of model names to try (e.g., ["deepseek-v3.2", "gemini-2.5-flash"])
on_failure: Callback function called when a failure occurs
**kwargs: Additional arguments passed to the API
Returns:
Response dictionary from successful endpoint
Raises:
Exception: If all endpoints fail
"""
self._init_circuit_breakers()
# Determine the fallback chain
if fallback_chain is None:
fallback_chain = list(self.router.models.keys())
self.fallback_chain = fallback_chain
last_error = None
for attempt in range(self.config.max_retries):
for model_name in fallback_chain:
model = self.router.models.get(model_name)
if not model:
continue
cb = self.circuit_breakers.get(model_name, CircuitBreaker())
if not cb.can_attempt():
print(f"⏳ Skipping {model_name} (circuit breaker: {cb.state})")
continue
try:
print(f"📤 Attempting {model_name} (attempt {attempt + 1})")
# Make the actual API call
response = await self._make_api_call(
endpoint=model.endpoint,
model=model.name,
prompt=prompt,
**kwargs
)
# Success!
cb.record_success()
self._log_request(model_name, "success", attempt + 1)
return {
"success": True,
"model_used": model.name,
"response": response,
"failover_count": attempt
}
except Exception as e:
last_error = e
cb.record_failure()
self._log_request(model_name, "failed", attempt + 1, str(e))
if on_failure:
on_failure(model_name, str(e))
print(f"❌ {model_name} failed: {str(e)}")
# Apply delay before next retry
if attempt < self.config.max_retries - 1:
delay = self._calculate_delay(attempt)
print(f"⏱ Waiting {delay}s before retry...")
await asyncio.sleep(delay)
# All attempts exhausted
raise Exception(
f"All {len(fallback_chain)} models failed after {self.config.max_retries} retries. "
f"Last error: {last_error}"
)
def _calculate_delay(self, attempt: int) -> float:
"""Calculate retry delay with optional exponential backoff."""
if self.config.exponential_backoff:
return self.config.retry_delay_seconds * (2 ** attempt)
return self.config.retry_delay_seconds
async def _make_api_call(
self,
endpoint: str,
model: str,
prompt: str,
**kwargs
) -> Dict[str, Any]:
"""
Make the actual API call to the endpoint.
Replace with your actual HTTP client implementation.
"""
import aiohttp
import json
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
async with aiohttp.ClientSession() as session:
async with session.post(
endpoint,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"API error {response.status}: {error_text}")
return await response.json()
def _log_request(self, model: str, status: str, attempt: int, error: str = None):
"""Log request details for monitoring."""
self.request_history.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"status": status,
"attempt": attempt,
"error": error
})
def get_failover_stats(self) -> Dict[str, Any]:
"""Get statistics about failover operations."""
total = len(self.request_history)
successful = sum(1 for r in self.request_history if r["status"] == "success")
failed = total - successful
model_stats = {}
for record in self.request_history:
model = record["model"]
if model not in model_stats:
model_stats[model] = {"success": 0, "failed": 0}
model_stats[model][record["status"]] += 1
return {
"total_requests": total,
"successful": successful,
"failed": failed,
"success_rate": successful / total if total > 0 else 0,
"model_stats": model_stats,
"circuit_breaker_states": {
name: cb.state for name, cb in self.circuit_breakers.items()
}
}
Example usage
async def main():
from health_monitor import HealthMonitor
from intelligent_router import IntelligentRouter
# Initialize system
health = HealthMonitor()
router = IntelligentRouter(health)
router.register_default_models()
failover = FailoverManager(
router,
config=FailoverConfig(
max_retries=3,
retry_delay_seconds=1.0,
exponential_backoff=True
)
)
# Define fallback chain: try cheapest first, escalate if needed
fallback_chain = [
"deepseek-v3.2", # $0.42/MTok - cheapest
"gemini-2.5-flash", # $2.50/MTok - fast and affordable
"gpt-4.1", # $8/MTok - premium option
]
try:
result = await failover.call_with_failover(
prompt="Explain quantum computing in simple terms",
fallback_chain=fallback_chain,
temperature=0.7,
max_tokens=500
)
print(f"\n✅ Success!")
print(f" Model used: {result['model_used']}")
print(f" Failover count: {result['failover_count']}")
print(f" Response: {result['response']}")
except Exception as e:
print(f"\n❌ All endpoints failed: {e}")
# Print statistics
stats = failover.get_failover_stats()
print(f"\n📊 Failover Statistics:")
print(f" Total requests: {stats['total_requests']}")
print(f" Success rate: {stats['success_rate']:.1%}")
print(f" Circuit breakers: {stats['circuit_breaker_states']}")
if __name__ == "__main__":
asyncio.run(main())Complete Production-Ready Implementation
Now let's put everything together into a production-ready load balancer that you can deploy in real applications.
# production_load_balancer.py
"""
Production-Ready API Load Balancer for Multi-Model AI APIs
Supports HolySheep AI and other providers with intelligent routing.
"""
import asyncio
import time
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import hashlib
Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RequestPriority(Enum):
"""Request priority levels for routing decisions."""
CRITICAL = 1
HIGH = 2
NORMAL = 3
BUDGET = 4
@dataclass
class APIRequest:
"""Represents an incoming API request."""
id: str
prompt: str
priority: RequestPriority = RequestPriority.NORMAL
max_cost_per_1k: Optional[float] = None
required_capabilities: List[str] = field(default_factory=list)
metadata: Dict = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class APIResponse:
"""Represents an API response."""
request_id: str
success: bool
model_used: str
response_data: Optional[Dict] = None
error: Optional[str] = None
latency_ms: float = 0.0
cost_used: float = 0.0
failover_attempts: int = 0
class ProductionLoadBalancer:
"""
Production-ready load balancer with:
- Intelligent routing based on request characteristics
- Automatic failover with circuit breakers
- Cost tracking and budget management
- Request queuing and prioritization
- Comprehensive monitoring
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Component initialization
self.health_monitor = self._setup_health_monitor()
self.router = self._setup_router()
self.failover_manager = self._setup_failover()
# Metrics and tracking
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost": 0.0,
"total_tokens": 0,
"avg_latency_ms": 0.0,
"model_usage": {},
"failover_count": 0
}
# Request queue
self.request_queue: asyncio.Queue = asyncio.Queue()
self.processing = False
def _setup_health_monitor(self):
"""Initialize health monitoring."""
from health_monitor import HealthMonitor
monitor = HealthMonitor(failure_threshold=3, recovery_timeout=30)
monitor.register_endpoint("holysheep-primary", self.base_url)
return monitor
def _setup_router(self):
"""Initialize intelligent routing."""
from intelligent_router import IntelligentRouter, RoutingStrategy
router = IntelligentRouter(self.health_monitor)
# Register models with accurate 2026 pricing
router.register_model(self._create_model(
name="deepseek-v3.2",
provider="holysheep",
cost=0.42,
latency=42,
capabilities=["coding", "math", "reasoning", "budget-friendly"]
))
router.register_model(self._create_model(
name="gemini-2.5-flash",
provider="holysheep",
cost=2.50,
latency=38,
capabilities=["fast", "multimodal", "cost-efficient"]
))
router.register_model(self._create_model(
name="gpt-4.1",
provider="holysheep",
cost=8.0,
latency=45,
capabilities=["reasoning", "coding", "analysis", "creative"]
))
router.register_model(self._create_model(
name="claude-sonnet-4.5",
provider="holysheep",
cost=15.0,
latency=52,
capabilities=["reasoning", "writing", "analysis", "long-context"]
))
return router
def _create_model(self, name: str, provider: str, cost: float, latency: float, capabilities: List[str]):
"""Helper to create model info."""
from intelligent_router import ModelInfo
return ModelInfo(
name=name,
provider=provider,
endpoint=f"{self.base_url}/chat/completions",
cost_per_million_tokens=cost,
average_latency_ms=latency,
max_tokens=128000,
capabilities=capabilities
)
def _setup_failover(self):
"""Initialize failover management."""
from failover_manager import FailoverManager, FailoverConfig
return FailoverManager(
self.router,
config=FailoverConfig(
max_retries=3,
retry_delay_seconds=1.0,
exponential_backoff=True
)
)
def _determine_fallback_chain(self, request: APIRequest) -> List[str]:
"""Determine optimal fallback chain based on request priority."""
if request.priority == RequestPriority.BUDGET:
# Budget requests: cheapest first
return ["deepseek-v3.2", "gemini-2.5-flash"]
elif request.priority == RequestPriority.NORMAL:
# Normal requests: balance cost and quality
return ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"]
elif request.priority == RequestPriority.HIGH:
# High priority: quality focused
return ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"]
else: # CRITICAL
# Critical: best available, regardless of cost
return ["gpt-4.1", "claude-sonnet-4.5"]
async def process_request(self, request: APIRequest) -> APIResponse:
"""Process a single API request with full load balancing."""
start_time = time.time()
self.metrics["total_requests"] += 1
# Determine routing strategy based on request
if request.priority == RequestPriority.CRITICAL:
strategy = RoutingStrategy.LEAST_LATENCY
elif request.priority == RequestPriority.BUDGET:
strategy = RoutingStrategy.COST_OPTIMIZED
else:
strategy = RoutingStrategy.INTELLIGENT
fallback_chain = self._determine_fallback_chain(request)
try:
# Route to best available model
route_result = self.router.route(
strategy=strategy,
max_cost_per_1k=request.max_cost_per_1k
)