AI Gateways and Model Routing

# litellm_config.yaml - LiteLLM Proxy configuration
model_list:
 # Primary model with multiple deployments for load balancing
 - model_name: "gpt-4o"
 litellm_params:
 model: "openai/gpt-4o"
 api_key: "os.environ/OPENAI_API_KEY"
 rpm: 500 # Requests per minute limit
 tpm: 200000 # Tokens per minute limit

 - model_name: "gpt-4o"
 litellm_params:
 model: "azure/gpt-4o-eastus"
 api_key: "os.environ/AZURE_API_KEY"
 api_base: "https://my-eastus.openai.azure.com"
 rpm: 300

 # Fallback model (cheaper, faster)
 - model_name: "gpt-4o-mini"
 litellm_params:
 model: "openai/gpt-4o-mini"
 api_key: "os.environ/OPENAI_API_KEY"

 # Alternative provider (Anthropic via OpenAI-compatible API)
 - model_name: "claude-sonnet"
 litellm_params:
 model: "anthropic/claude-sonnet-4-20250514"
 api_key: "os.environ/ANTHROPIC_API_KEY"

# Router settings
router_settings:
 routing_strategy: "latency-based-routing"
 num_retries: 3
 retry_after: 5
 allowed_fails: 2 # Remove deployment after 2 consecutive failures
 cooldown_time: 60 # Re-check failed deployment after 60s

# Budget and rate limiting
general_settings:
 master_key: "sk-gateway-master-key"
 database_url: "postgresql://user:pass@db:5432/litellm"
 max_budget: 1000.0 # Monthly budget in USD
 budget_duration: "30d"

import openai

# Point the standard OpenAI client at the LiteLLM Proxy
client = openai.OpenAI(
 base_url="http://litellm-proxy:4000/v1",
 api_key="sk-team-analytics-key", # Virtual key for cost tracking
)

# This request is transparently routed, load-balanced, and tracked
response = client.chat.completions.create(
 model="gpt-4o", # Logical model name, resolved by the proxy
 messages=[{"role": "user", "content": "Summarize this quarter's revenue."}],
 temperature=0.3,
)

# The proxy handles:
# 1. Routing to the fastest available gpt-4o deployment
# 2. Automatic retry if the first deployment fails
# 3. Token counting and cost attribution to the virtual key
# 4. Rate limiting per the team's allocated budget
print(response.choices[0].message.content)

from litellm import Router

# Configure a router with fallback chains
router = Router(
 model_list=[
 {
 "model_name": "primary-chat",
 "litellm_params": {
 "model": "openai/gpt-4o",
 "api_key": OPENAI_KEY,
 },
 },
 {
 "model_name": "primary-chat",
 "litellm_params": {
 "model": "anthropic/claude-sonnet-4-20250514",
 "api_key": ANTHROPIC_KEY,
 },
 },
 {
 "model_name": "primary-chat",
 "litellm_params": {
 "model": "google/gemini-2.5-pro",
 "api_key": GOOGLE_KEY,
 },
 },
 ],
 # Fallback configuration
 fallbacks=[
 {"primary-chat": ["fallback-chat"]},
 ],
 # Context window overflow handling
 context_window_fallbacks=[
 {"primary-chat": ["long-context-chat"]},
 ],
 routing_strategy="latency-based-routing",
 num_retries=2,
 retry_after=1,
)

# The router automatically handles failover
async def resilient_completion(messages, **kwargs):
 """Make an LLM call with automatic multi-provider failover."""
 try:
 response = await router.acompletion(
 model="primary-chat",
 messages=messages,
 **kwargs,
 )
 return response
 except Exception as e:
 # All providers failed; log and raise
 logger.error(f"All providers exhausted: {e}")
 raise ServiceUnavailableError("LLM service temporarily unavailable")

import asyncio
from dataclasses import dataclass, field
from collections import defaultdict
import time

@dataclass
class RateLimitConfig:
 requests_per_minute: int = 60
 tokens_per_minute: int = 100000
 max_budget_usd: float = 500.0
 budget_period_days: int = 30

class GatewayRateLimiter:
 """Token-aware rate limiter for AI gateway traffic."""

 def __init__(self):
 self.team_configs: dict[str, RateLimitConfig] = {}
 self.request_windows: dict[str, list[float]] = defaultdict(list)
 self.token_windows: dict[str, list[tuple[float, int]]] = defaultdict(list)
 self._lock = asyncio.Lock()

 async def check_rate_limit(
 self, team_id: str, estimated_tokens: int
 ) -> bool:
 """Check if a request is within rate limits. Returns True if allowed."""
 config = self.team_configs.get(team_id, RateLimitConfig())
 now = time.monotonic()
 window_start = now - 60 # 1-minute window

 async with self._lock:
 # Clean old entries
 self.request_windows[team_id] = [
 t for t in self.request_windows[team_id] if t > window_start
 ]
 self.token_windows[team_id] = [
 (t, n) for t, n in self.token_windows[team_id]
 if t > window_start
 ]

 # Check RPM
 if len(self.request_windows[team_id]) >= config.requests_per_minute:
 return False

 # Check TPM
 current_tokens = sum(
 n for _, n in self.token_windows[team_id]
 )
 if current_tokens + estimated_tokens > config.tokens_per_minute:
 return False

 # Record this request
 self.request_windows[team_id].append(now)
 self.token_windows[team_id].append((now, estimated_tokens))
 return True

import numpy as np
from typing import Optional

class SemanticCache:
 """Embedding-based semantic cache for LLM responses."""

 def __init__(self, embedding_model, similarity_threshold=0.95):
 self.embedding_model = embedding_model
 self.threshold = similarity_threshold
 self.cache: list[dict] = [] # In production, use a vector DB

 async def get(self, messages: list[dict]) -> Optional[str]:
 """Look up a semantically similar cached response."""
 query = self._extract_query(messages)
 query_embedding = await self.embedding_model.embed(query)

 best_score = 0.0
 best_response = None

 for entry in self.cache:
 score = self._cosine_similarity(
 query_embedding, entry["embedding"]
 )
 if score > best_score:
 best_score = score
 best_response = entry["response"]

 if best_score >= self.threshold:
 return best_response
 return None

 async def put(self, messages: list[dict], response: str):
 """Cache a response with its embedding."""
 query = self._extract_query(messages)
 embedding = await self.embedding_model.embed(query)
 self.cache.append({
 "query": query,
 "embedding": embedding,
 "response": response,
 "created_at": time.time(),
 })

 def _extract_query(self, messages: list[dict]) -> str:
 """Extract the cacheable query from messages."""
 return messages[-1]["content"]

 def _cosine_similarity(self, a, b) -> float:
 return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

from datetime import datetime, timedelta

class BudgetEnforcer:
 """Enforce spending limits per team with tiered responses."""

 def __init__(self, db):
 self.db = db

 async def check_and_record(
 self, team_id: str, model: str, estimated_cost: float
 ) -> dict:
 """Check budget, record spend, return routing decision."""
 team = await self.db.get_team(team_id)
 period_start = datetime.utcnow() - timedelta(days=team.budget_period_days)
 current_spend = await self.db.get_spend(team_id, since=period_start)

 remaining = team.max_budget - current_spend
 utilization = current_spend / team.max_budget

 # Tiered response based on budget utilization
 if utilization >= 1.0:
 return {
 "action": "reject",
 "reason": f"Budget exhausted (${current_spend:.2f}/${team.max_budget:.2f})",
 }
 elif utilization >= 0.9:
 return {
 "action": "downgrade",
 "target_model": team.budget_fallback_model,
 "reason": f"Budget >90%, routing to {team.budget_fallback_model}",
 }
 elif utilization >= 0.75:
 return {
 "action": "allow",
 "warning": f"Budget at {utilization:.0%}. ${remaining:.2f} remaining.",
 }
 else:
 return {"action": "allow"}

 async def record_spend(self, team_id: str, model: str, cost: float):
 """Record actual spend after a successful response."""
 await self.db.record_transaction(
 team_id=team_id,
 model=model,
 cost=cost,
 timestamp=datetime.utcnow(),
 )