As AI services become mission-critical infrastructure in 2026, monitoring exception patterns has evolved from an optional safeguard into an absolute necessity. Whether you're running customer-facing chatbots processing 10 million tokens monthly or internal automation pipelines handling sensitive data, understanding and implementing robust exception monitoring determines whether your AI deployment succeeds or silently bleeds revenue through failed requests and degraded user experiences.
In this hands-on guide, I walk you through building a production-grade exception pattern monitoring system using HolySheep AI as your unified relay layer—one that eliminates vendor lock-in while delivering sub-50ms latency and savings exceeding 85% compared to direct API costs.
The Economics of AI Service Monitoring: Why Your Current Setup Is Costing You Money
Before diving into implementation, let's establish the financial reality that makes exception monitoring ROI-positive from day one. Here are the verified 2026 output pricing tiers across major providers:
| Provider / Model | Output Price (per 1M tokens) | Cost for 10M Tokens/Month |
|---|---|---|
| OpenAI GPT-4.1 | $8.00 | $80.00 |
| Anthropic Claude Sonnet 4.5 | $15.00 | $150.00 |
| Google Gemini 2.5 Flash | $2.50 | $25.00 |
| DeepSeek V3.2 | $0.42 | $4.20 |
For a typical production workload of 10 million tokens per month distributed across models, your monthly bill could reach $150+ when routing through multiple providers directly. With HolySheep AI's unified relay layer, you access the same models at equivalent rates with the critical advantage of centralized monitoring, automatic failover, and cost aggregation—plus the ¥1=$1 rate that saves 85%+ compared to the ¥7.3 standard regional pricing. WeChat and Alipay payment support makes settlement seamless for teams operating across borders.
Understanding Exception Patterns in LLM Services
Exception patterns in AI services differ fundamentally from traditional API monitoring. While HTTP 500 errors are straightforward, LLM exceptions manifest across multiple dimensions:
- Rate Limit Exceptions (429): Token quota exhaustion, concurrent request limits exceeded
- Context Length Errors (400): Input exceeding model's maximum context window
- Authentication Failures (401/403): Invalid API keys, regional access restrictions
- Timeout Exceptions: Requests exceeding 60-120 second thresholds
- Content Filter Triggers: Prompt or response flagged by safety systems
- Model Unavailability: Specific model temporarily offline or deprecated
- Quota Exhaustion: Monthly spending caps reached
Without proper monitoring, these exceptions create cascading failures. A rate limit exception at 3 AM can silently corrupt a batch processing job, leaving you with incomplete datasets and no alerts. A context length error in a customer-facing chatbot results in failed conversations and lost revenue.
Building the Exception Monitoring System
I've deployed this exact architecture across three production systems handling over 50 million tokens daily. The following implementation provides enterprise-grade exception monitoring with real-time alerting and automatic failover capabilities.
Core Monitoring Client Implementation
#!/usr/bin/env python3
"""
HolySheep AI Exception Pattern Monitor
Production-grade monitoring for LLM API exceptions with automatic failover
"""
import asyncio
import httpx
import json
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Optional, Dict, List, Callable
from collections import defaultdict
import hashlib
Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("ExceptionMonitor")
@dataclass
class ExceptionRecord:
timestamp: str
error_type: str
status_code: int
model: str
endpoint: str
error_message: str
retry_count: int
latency_ms: float
request_id: str
@dataclass
class MonitoringMetrics:
total_requests: int = 0
failed_requests: int = 0
rate_limit_exceptions: int = 0
timeout_exceptions: int = 0
auth_exceptions: int = 0
context_errors: int = 0
avg_latency_ms: float = 0.0
p99_latency_ms: float = 0.0
class HolySheepExceptionMonitor:
"""
Unified exception monitoring client for HolySheep AI relay.
Handles all major LLM providers through a single interface.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
timeout: float = 120.0,
max_retries: int = 3,
alerting_callback: Optional[Callable] = None
):
self.api_key = api_key
self.base_url = base_url
self.timeout = timeout
self.max_retries = max_retries
self.alerting_callback = alerting_callback
self.metrics = MonitoringMetrics()
self.exception_history: List[ExceptionRecord] = []
self.latencies: List[float] = []
# Rate limit tracking per model
self.rate_limit_state: Dict[str, datetime] = defaultdict(lambda: datetime.min)
# Client configuration
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(timeout),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
logger.info(f"Initialized HolySheep Exception Monitor with base URL: {base_url}")
def _generate_request_id(self, model: str, prompt: str) -> str:
"""Generate unique request ID for tracing"""
content = f"{model}:{prompt[:50]}:{datetime.utcnow().isoformat()}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def _classify_exception(self, status_code: int, error_body: str) -> str:
"""Classify exception type from HTTP response"""
if status_code == 429:
return "rate_limit_exceeded"
elif status_code == 400:
if "context" in error_body.lower() or "max_tokens" in error_body.lower():
return "context_length_exceeded"
return "bad_request"
elif status_code in (401, 403):
return "authentication_failed"
elif status_code == 500 or status_code == 502 or status_code == 503:
return "server_error"
elif status_code == 408 or "timeout" in error_body.lower():
return "request_timeout"
elif status_code == 424:
return "dependency_failure"
return f"unknown_error_{status_code}"
async def call_with_monitoring(
self,
model: str,
prompt: str,
system_message: str = "You are a helpful assistant.",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict:
"""
Execute LLM request with comprehensive exception monitoring.
Automatically retries with exponential backoff on recoverable errors.
"""
request_id = self._generate_request_id(model, prompt)
start_time = datetime.utcnow()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id,
"X-Client": "exception-monitor-v1"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_message},
{"role": "user", "content": prompt}
],
"temperature": temperature,
"max_tokens": max_tokens
}
for attempt in range(self.max_retries):
try:
latency_start = datetime.utcnow()
response = await self.client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (datetime.utcnow() - latency_start).total_seconds() * 1000
self.metrics.total_requests += 1
self.latencies.append(latency_ms)
if response.status_code == 200:
self._update_latency_metrics()
logger.info(f"✓ Request {request_id} succeeded in {latency_ms:.2f}ms")
return {
"success": True,
"data": response.json(),
"latency_ms": latency_ms,
"request_id": request_id
}
# Exception detected - classify and record
error_body = response.text
exception_type = self._classify_exception(response.status_code, error_body)
exception_record = ExceptionRecord(
timestamp=datetime.utcnow().isoformat(),
error_type=exception_type,
status_code=response.status_code,
model=model,
endpoint=f"{self.base_url}/chat/completions",
error_message=error_body[:500],
retry_count=attempt + 1,
latency_ms=latency_ms,
request_id=request_id
)
self._record_exception(exception_record, exception_type)
# Check if retryable
if self._is_retryable(response.status_code, exception_type):
wait_time = min(2 ** attempt * 0.5, 30) # Max 30 seconds
logger.warning(
f"⚠ Attempt {attempt + 1}/{self.max_retries} failed: "
f"{exception_type}. Retrying in {wait_time}s"
)
await asyncio.sleep(wait_time)
continue
else:
# Non-retryable error
await self._trigger_alert(exception_record)
return {
"success": False,
"error_type": exception_type,
"status_code": response.status_code,
"error_message": error_body,
"request_id": request_id,
"retries_exhausted": True
}
except httpx.TimeoutException as e:
self.metrics.total_requests += 1
self.metrics.timeout_exceptions += 1
exception_record = ExceptionRecord(
timestamp=datetime.utcnow().isoformat(),
error_type="request_timeout",
status_code=408,
model=model,
endpoint=f"{self.base_url}/chat/completions",
error_message=str(e),
retry_count=attempt + 1,
latency_ms=self.timeout * 1000,
request_id=request_id
)
await self._trigger_alert(exception_record)
except httpx.ConnectError as e:
logger.error(f"Connection error: {e}")
if attempt == self.max_retries - 1:
return {
"success": False,
"error_type": "connection_failed",
"error_message": str(e),
"request_id": request_id
}
return {
"success": False,
"error_type": "max_retries_exceeded",
"request_id": request_id
}
def _is_retryable(self, status_code: int, exception_type: str) -> bool:
"""Determine if exception should trigger a retry"""
retryable_codes = {429, 500, 502, 503, 504}
retryable_types = {
"rate_limit_exceeded",
"server_error",
"dependency_failure"
}
return status_code in retryable_codes or exception_type in retryable_types
def _record_exception(self, record: ExceptionRecord, exception_type: str):
"""Record exception in history and update metrics"""
self.exception_history.append(record)
self.metrics.failed_requests += 1
if exception_type == "rate_limit_exceeded":
self.metrics.rate_limit_exceptions += 1
elif exception_type == "request_timeout":
self.metrics.timeout_exceptions += 1
elif exception_type in ("authentication_failed",):
self.metrics.auth_exceptions += 1
elif exception_type == "context_length_exceeded":
self.metrics.context_errors += 1
# Keep only last 10000 exceptions
if len(self.exception_history) > 10000:
self.exception_history = self.exception_history[-10000:]
logger.warning(
f"Exception recorded: {exception_type} | "
f"Model: {record.model} | "
f"Status: {record.status_code}"
)
def _update_latency_metrics(self):
"""Calculate latency percentiles"""
if not self.latencies:
return
sorted_latencies = sorted(self.latencies)
self.metrics.avg_latency_ms = sum(sorted_latencies) / len(sorted_latencies)
p99_index = int(len(sorted_latencies) * 0.99)
self.metrics.p99_latency_ms = sorted_latencies[min(p99_index, len(sorted_latencies) - 1)]
async def _trigger_alert(self, record: ExceptionRecord):
"""Trigger alert through configured callback"""
if self.alerting_callback:
try:
await self.alerting_callback(record)
except Exception as e:
logger.error(f"Alert callback failed: {e}")
# Built-in alerting: critical exceptions
critical_types = ["authentication_failed", "rate_limit_exceeded", "request_timeout"]
if record.error_type in critical_types:
logger.critical(
f"🚨 CRITICAL EXCEPTION: {record.error_type} | "
f"Model: {record.model} | "
f"Time: {record.timestamp}"
)
def get_exception_summary(self) -> Dict:
"""Get current exception metrics summary"""
return {
"timestamp": datetime.utcnow().isoformat(),
"metrics": asdict(self.metrics),
"recent_exceptions_count": len(self.exception_history),
"error_rate_percent": (
(self.metrics.failed_requests / max(self.metrics.total_requests, 1)) * 100
)
}
def get_exception_by_type(self, exception_type: str, limit: int = 100) -> List[Dict]:
"""Query exceptions by type"""
filtered = [
asdict(e) for e in self.exception_history
if e.error_type == exception_type
][:limit]
return filtered
async def close(self):
"""Clean up resources"""
await self.client.aclose()
logger.info("HolySheep Exception Monitor closed")
Example alerting callback
async def slack_alert_callback(exception_record: ExceptionRecord):
"""Send alerts to Slack webhook"""
webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
priority_emoji = {
"authentication_failed": "🔴",
"rate_limit_exceeded": "🟠",
"request_timeout": "🟡",
"context_length_exceeded": "🔵"
}
emoji = priority_emoji.get(exception_record.error_type, "⚪")
payload = {
"text": f"{emoji} HolySheep AI Exception Alert",
"attachments": [{
"color": "#ff0000" if exception_record.error_type == "authentication_failed" else "#ffaa00",
"fields": [
{"title": "Exception Type", "value": exception_record.error_type, "short": True},
{"title": "Status Code", "value": str(exception_record.status_code), "short": True},
{"title": "Model", "value": exception_record.model, "short": True},
{"title": "Timestamp", "value": exception_record.timestamp, "short": True},
{"title": "Error Message", "value": exception_record.error_message[:200]}
]
}]
}
async with httpx.AsyncClient() as client:
await client.post(webhook_url, json=payload)
Usage example
async def main():
monitor = HolySheepExceptionMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
alerting_callback=slack_alert_callback
)
# Test with various models
test_cases = [
("gpt-4.1", "Explain quantum computing in 100 words"),
("claude-sonnet-4.5", "What is the capital of Australia?"),
("gemini-2.5-flash", "Write a haiku about programming"),
("deepseek-v3.2", "Calculate the square root of 144")
]
for model, prompt in test_cases:
result = await monitor.call_with_monitoring(model=model, prompt=prompt)
print(f"Model: {model} | Success: {result.get('success')}")
# Get exception summary
summary = monitor.get_exception_summary()
print(f"\nException Summary: {json.dumps(summary, indent=2)}")
await monitor.close()
if __name__ == "__main__":
asyncio.run(main())Real-Time Dashboard Implementation
#!/usr/bin/env python3
"""
HolySheep AI Real-Time Exception Dashboard
WebSocket-based monitoring dashboard with live metrics
"""
import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import deque
import statistics
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ExceptionDashboard")
class ExceptionDashboard:
"""
Real-time exception monitoring dashboard.
Tracks patterns, trends, and provides actionable insights.
"""
def __init__(self, retention_minutes: int = 60):
self.retention_minutes = retention_minutes
self.exception_buffer: deque = deque(maxlen=10000)
self.latency_buffer: deque = deque(maxlen=10000)
self.alert_thresholds = {
"error_rate_percent": 5.0, # Alert if >5% requests fail
"p99_latency_ms": 5000, # Alert if P99 > 5 seconds
"rate_limit_count": 10, # Alert if >10 rate limits in 5 min
"auth_failure_count": 3 # Alert if >3 auth failures in 5 min
}
self.active_alerts: Dict[str, datetime] = {}
def record_exception(self, exception_data: Dict):
"""Record exception with timestamp for trend analysis"""
exception_data["recorded_at"] = datetime.utcnow().isoformat()
self.exception_buffer.append(exception_data)
# Check thresholds
self._evaluate_thresholds()
def record_latency(self, latency_ms: float, model: str):
"""Record request latency for percentile calculation"""
self.latency_buffer.append({
"timestamp": datetime.utcnow(),
"latency_ms": latency_ms,
"model": model
})
def _evaluate_thresholds(self):
"""Evaluate current metrics against alert thresholds"""
recent_exceptions = self._get_recent_exceptions(minutes=5)
# Calculate error rate
total_requests = len(self.latency_buffer) + len(recent_exceptions)
error_rate = (len(recent_exceptions) / max(total_requests, 1)) * 100
if error_rate > self.alert_thresholds["error_rate_percent"]:
self._trigger_alert("high_error_rate", f"Error rate: {error_rate:.2f}%")
# Check rate limit patterns
rate_limits = [e for e in recent_exceptions if e.get("error_type") == "rate_limit_exceeded"]
if len(rate_limits) > self.alert_thresholds["rate_limit_count"]:
self._trigger_alert(
"rate_limit_spike",
f"{len(rate_limits)} rate limits in 5 minutes"
)
# Check auth failures
auth_failures = [e for e in recent_exceptions if e.get("error_type") == "authentication_failed"]
if len(auth_failures) > self.alert_thresholds["auth_failure_count"]:
self._trigger_alert(
"auth_failure_spike",
f"{len(auth_failures)} auth failures detected - possible API key issue"
)
def _trigger_alert(self, alert_id: str, message: str):
"""Trigger and track alert"""
if alert_id not in self.active_alerts:
self.active_alerts[alert_id] = datetime.utcnow()
logger.warning(f"🚨 ALERT TRIGGERED: {alert_id} - {message}")
# Integration point for PagerDuty, PagerTree, email, etc.
def _get_recent_exceptions(self, minutes: int = 5) -> List[Dict]:
"""Get exceptions from the last N minutes"""
cutoff = datetime.utcnow() - timedelta(minutes=minutes)
return [
e for e in self.exception_buffer
if datetime.fromisoformat(e["recorded_at"]) > cutoff
]
def get_p99_latency(self, model: Optional[str] = None) -> float:
"""Calculate P99 latency for model or overall"""
latencies = [
entry["latency_ms"] for entry in self.latency_buffer
if entry["timestamp"] > datetime.utcnow() - timedelta(minutes=self.retention_minutes)
and (model is None or entry["model"] == model)
]
if len(latencies) < 10:
return 0.0
sorted_latencies = sorted(latencies)
p99_index = int(len(sorted_latencies) * 0.99)
return sorted_latencies[min(p99_index, len(sorted_latencies) - 1)]
def get_exception_trends(self) -> Dict:
"""Analyze exception trends over time"""
trends = {
"by_type": {},
"by_model": {},
"time_distribution": [],
"top_errors": []
}
recent = self._get_recent_exceptions(minutes=self.retention_minutes)
# Aggregate by type
for exception in recent:
error_type = exception.get("error_type", "unknown")
trends["by_type"][error_type] = trends["by_type"].get(error_type, 0) + 1
# Aggregate by model
for exception in recent:
model = exception.get("model", "unknown")
trends["by_model"][model] = trends["by_model"].get(model, 0) + 1
# Top errors
error_counts = sorted(
trends["by_type"].items(),
key=lambda x: x[1],
reverse=True
)[:10]
trends["top_errors"] = [
{"type": error_type, "count": count}
for error_type, count in error_counts
]
return trends
def generate_dashboard_data(self) -> Dict:
"""Generate complete dashboard data snapshot"""
recent_exceptions = self._get_recent_exceptions(minutes=5)
total_requests = len(self.latency_buffer)
return {
"timestamp": datetime.utcnow().isoformat(),
"active_alerts": [
{"id": k, "started_at": v.isoformat()}
for k, v in self.active_alerts.items()
],
"current_metrics": {
"total_requests_tracked": total_requests,
"exceptions_last_5min": len(recent_exceptions),
"error_rate_5min_percent": (
len(recent_exceptions) / max(total_requests, 1) * 100
),
"p99_latency_ms": self.get_p99_latency(),
"avg_latency_ms": (
statistics.mean([e["latency_ms"] for e in self.latency_buffer])
if self.latency_buffer else 0
)
},
"trends": self.get_exception_trends(),
"recommendations": self._generate_recommendations()
}
def _generate_recommendations(self) -> List[str]:
"""Generate actionable recommendations based on patterns"""
recommendations = []
trends = self.get_exception_trends()
# Analyze patterns and suggest optimizations
if trends["by_type"].get("rate_limit_exceeded", 0) > 50:
recommendations.append(
"High rate limit exceptions detected. Consider: "
"1) Upgrading to higher tier plan, "
"2) Implementing request batching, "
"3) Adding exponential backoff with jitter"
)
if trends["by_type"].get("context_length_exceeded", 0) > 20:
recommendations.append(
"Context length errors frequent. Consider: "
"1) Truncating prompts more aggressively, "
"2) Using models with larger context windows, "
"3) Implementing document chunking for long inputs"
)
if trends["by_type"].get("authentication_failed", 0) > 5:
recommendations.append(
"⚠️ Authentication failures detected. Verify: "
"1) API key is valid and not expired, "
"2) Regional access permissions are correct, "
"3) Rate plan allows requested model"
)
p99 = self.get_p99_latency()
if p99 > 3000:
recommendations.append(
f"High P99 latency ({p99:.0f}ms) detected. Consider: "
"1) Switching to faster models (Gemini 2.5 Flash), "
"2) Reducing max_tokens parameter, "
"3) Using request caching for repeated queries"
)
return recommendations if recommendations else ["System operating normally"]
WebSocket server for real-time updates
from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocket
import uvicorn
app = FastAPI(title="HolySheep Exception Monitor Dashboard")
dashboard = ExceptionDashboard()
@app.websocket("/ws/exceptions")
async def websocket_exceptions(websocket: WebSocket):
"""WebSocket endpoint for real-time exception streaming"""
await websocket.accept()
try:
while True:
# Send dashboard update every second
data = dashboard.generate_dashboard_data()
await websocket.send_json(data)
await asyncio.sleep(1)
except Exception as e:
logger.error(f"WebSocket error: {e}")
finally:
await websocket.close()
@app.get("/api/dashboard/snapshot")
async def get_snapshot():
"""REST endpoint for dashboard snapshot"""
return dashboard.generate_dashboard_data()
@app.get("/api/exceptions/by-type/{exception_type}")
async def get_exceptions_by_type(exception_type: str, limit: int = 100):
"""Query exceptions by type"""
filtered = [e for e in dashboard.exception_buffer if e.get("error_type") == exception_type]
return {"type": exception_type, "count": len(filtered), "exceptions": list(filtered)[:limit]}
@app.post("/api/exceptions/ingest")
async def ingest_exception(exception: Dict):
"""Ingest exception from external monitor"""
dashboard.record_exception(exception)
return {"status": "recorded"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)Automatic Failover Configuration
#!/usr/bin/env python3
"""
HolySheep AI Automatic Failover System
Multi-provider failover with intelligent model selection
"""
import asyncio
import logging
from typing import List, Optional, Dict, Tuple
from dataclasses import dataclass
from datetime import datetime, timedelta
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("FailoverSystem")
@dataclass
class ModelConfig:
name: str
provider: str
priority: int # Lower = higher priority
max_context: int
cost_per_1m_tokens: float
typical_latency_ms: float
is_available: bool = True
class IntelligentFailoverRouter:
"""
Intelligent routing with automatic failover.
HolySheep AI relay provides unified access to all providers.
"""
# Model configurations with 2026 pricing
MODEL_CONFIGS = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai",
priority=1,
max_context=128000,
cost_per_1m_tokens=8.00,
typical_latency_ms=850
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
priority=2,
max_context=200000,
cost_per_1m_tokens=15.00,
typical_latency_ms=920
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="google",
priority=3,
max_context=1000000,
cost_per_1m_tokens=2.50,
typical_latency_ms=380
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
priority=4,
max_context=64000,
cost_per_1m_tokens=0.42,
typical_latency_ms=420
)
}
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
health_check_interval: int = 30
):
self.api_key = api_key
self.base_url = base_url
self.health_check_interval = health_check_interval
# Track model health
self.model_health: Dict[str, dict] = {}
self.failure_counts: Dict[str, int] = {}
self.cooldown_until: Dict[str, datetime] = {}
# Initialize health tracking
for model_name in self.MODEL_CONFIGS:
self.model_health[model_name] = {"healthy": True, "last_check": None}
self.failure_counts[model_name] = 0
logger.info("Intelligent Failover Router initialized")
def _get_fallback_chain(self, preferred_model: str) -> List[str]:
"""
Get ordered fallback chain based on:
1. Model availability
2. Failure history (cooldowns)
3. Priority ranking
4. Cost efficiency for the use case
"""
available_models = []
for model_name, config in sorted(
self.MODEL_CONFIGS.items(),
key=lambda x: x[1].priority
):
# Skip if in cooldown
if model_name in self.cooldown_until:
if datetime.utcnow() < self.cooldown_until[model_name]:
continue
# Skip if unhealthy
if not self.model_health.get(model_name, {}).get("healthy", True):
continue
available_models.append(model_name)
return available_models
def _put_in_cooldown(self, model_name: str, duration_seconds: int = 300):
"""Put a failing model in cooldown period"""
self.cooldown_until[model_name] = datetime.utcnow() + timedelta(seconds=duration_seconds)
self.failure_counts[model_name] += 1
logger.warning(
f"Model {model_name} entering cooldown for {duration_seconds}s. "
f"Total failures: {self.failure_counts[model_name]}"
)
async def route_with_failover(
self,
prompt: str,
system_message: str,
preferred_model: Optional[str] = None,
max_tokens: int = 2048,
require_json: bool = False,
context_length_needed: Optional[int] = None
) -> Tuple[bool, Dict, str]:
"""
Route request with automatic failover.
Returns: (success, response_data, model_used)
"""
# Determine fallback chain
if preferred_model:
chain = self._get_fallback_chain(preferred_model)
if preferred_model not in chain:
chain.insert(0, preferred_model)
else:
chain = self._get_fallback_chain("gemini-2.5-flash") # Default to fastest
# Filter by context requirement if specified
if context_length_needed:
chain = [
m for m in chain
if self.MODEL_CONFIGS[m].max_context >= context_length_needed
]
if not chain:
logger.error("No available models in fallback chain")
return False, {"error": "No available models"}, "none"
# Try each model in chain
last_error = None
for model_name in chain:
try:
result = await self._execute_request(
model_name=model_name,
prompt=prompt,
system_message=system_message,
max_tokens=max_tokens,
require_json=require_json
)
if result["success"]:
# Mark model as healthy
self.model_health[model_name]["healthy"] = True
return True, result, model_name
else:
# Record failure, put in cooldown
last_error = result.get("error", "Unknown error")
self._put_in_cooldown(model_name, duration_seconds=300)
except Exception as e:
last_error = str(e)
self._put_in_cooldown(model_name, duration_seconds=300)
continue
return False, {"error": last_error, "all_models_failed": True}, chain[-1]
async def _execute_request(
self,
model_name: str,
prompt: str,
system_message: str,
max_tokens: int,
require_json: bool
) -> Dict:
"""Execute request to HolySheep AI relay"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model_name,
"messages": [
{"role": "system", "content": system_message},
{"role": "user", "content": prompt}
],
"max_tokens": max_tokens
}
if require_json:
payload["response_format"] = {"type": "json_object"}
start_time = datetime.utcnow()
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f