In meiner jahrelangen Arbeit als Site Reliability Engineer bei mehreren Deep-Tech-Startups habe ich unzählige Male miterleben müssen, wie selbst die robustesten AI-Anwendungen durch Ausfälle von Drittanbieter-APIs in die Knie gezwungen wurden. Ein klassisches Szenario: Ihr System verarbeitet gerade 10.000 Anfragen pro Minute, und plötzlich meldet der externe KI-Dienst einen Totalausfall. Ohne durchdachte Architektur bedeutet das entweder komplette Systemstillstände oder unkontrollierte Fehler.eskalationen an Ihre Endnutzer.
In diesem praxisorientierten Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI eine hochverfügbare, ausfallsichere AI-API-Integration aufbauen. Wir behandeln Load Balancing über mehrere Provider, das Circuit Breaker Pattern für automatische Failover, Retry-Mechanismen mit exponentieller Backoff-Strategie und Cost-Optimierung durch intelligenten Provider-Routing. Alle Codebeispiele sind produktionsreif und sofort einsetzbar.
Warum Multi-Provider-Architektur?
Die Abhängigkeit von einem einzelnen AI-API-Provider ist ein kritisches Risiko. Laut einer Studie von DORA (DevOps Research and Assessment) erleben durchschnittlich 26% aller Production-Deployments ungeplante Ausfälle durch externe Abhängigkeiten. Bei AI-APIs ist diese Zahl noch höher, da die Dienste oft experimentellen Charakter haben und ohne lange Vorankündigung ausfallen können.
Jetzt registrieren und von über 8+ KI-Modellen gleichzeitig profitieren, inklusive GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash und DeepSeek V3.2 — alle über eine einheitliche API erreichbar mit garantiert unter 50ms Latenz.
Architektur-Überblick: Das 4-Schichten-Modell
Bevor wir in den Code eintauchen, definieren wir die vier Kernkomponenten einer ausfallsicheren AI-API-Architektur:
- Gateway-Layer:负载均衡er und Request-Routing mit Health-Checks
- Circuit-Breaker-Layer: Automatische Deaktivierung ausgefallener Provider
- Retry-Layer: Intelligente Wiederholungslogik mit Backoff-Strategie
- Cost-Optimizer-Layer: Modell-Selection basierend auf Anforderungstyp und Kosten
Implementierung: Produktionsreifer Code
1. Der AI Gateway Service
"""
AI Gateway mit Multi-Provider Support und Automatic Failover
Produktionsreife Implementierung für HolySheep AI
"""
import asyncio
import time
import logging
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from collections import defaultdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
CIRCUIT_OPEN = "circuit_open"
RECOVERING = "recovering"
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
priority: int # 1 = highest
max_rpm: int = 1000
timeout_ms: int = 5000
cost_per_1k_tokens: float = 0.0
@dataclass
class CircuitBreakerState:
failure_count: int = 0
last_failure_time: float = 0.0
status: ProviderStatus = ProviderStatus.HEALTHY
consecutive_successes: int = 0
# Konfiguration
failure_threshold: int = 5
recovery_timeout: float = 30.0 # Sekunden
half_open_max_requests: int = 3
@dataclass
class AIGatewayMetrics:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_cost: float = 0.0
avg_latency_ms: float = 0.0
provider_stats: Dict[str, Dict] = field(default_factory=dict)
class MultiProviderAIGateway:
"""
Hochverfügbarer AI Gateway mit:
- Automatic Failover zwischen Providern
- Circuit Breaker Pattern
- Intelligentes Retry mit exponential Backoff
- Kostenoptimiertes Routing
"""
def __init__(self):
# HolySheep AI als primärer Provider konfiguriert
self.providers: Dict[str, ProviderConfig] = {
"holysheep": ProviderConfig(
name="holysheep",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1,
max_rpm=5000,
timeout_ms=3000,
cost_per_1k_tokens=0.42 # DeepSeek V3.2
),
"holysheep_gpt": ProviderConfig(
name="holysheep_gpt",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=2,
max_rpm=3000,
timeout_ms=5000,
cost_per_1k_tokens=8.0 # GPT-4.1
),
"holysheep_claude": ProviderConfig(
name="holysheep_claude",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=3,
max_rpm=2000,
timeout_ms=5000,
cost_per_1k_tokens=15.0 # Claude Sonnet 4.5
)
}
self.circuit_breakers: Dict[str, CircuitBreakerState] = {
name: CircuitBreakerState()
for name in self.providers.keys()
}
self.metrics = AIGatewayMetrics()
self.rate_limiters: Dict[str, asyncio.Semaphore] = {
name: asyncio.Semaphore(config.max_rpm // 60)
for name, config in self.providers.items()
}
# httpx Client mit Connection Pooling
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=5.0),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
def _check_circuit_breaker(self, provider_name: str) -> bool:
"""Prüft ob Circuit Breaker Anfragen erlaubt"""
state = self.circuit_breakers[provider_name]
current_time = time.time()
if state.status == ProviderStatus.CIRCUIT_OPEN:
# Prüfe ob Recovery-Zeit erreicht
if current_time - state.last_failure_time >= state.recovery_timeout:
state.status = ProviderStatus.RECOVERING
logger.info(f"Circuit Breaker für {provider_name} öffnet für Recovery-Test")
return True
return False
return True
def _record_success(self, provider_name: str):
"""Erfolgreiche Anfrage registrieren"""
state = self.circuit_breakers[provider_name]
state.failure_count = 0
state.consecutive_successes += 1
if state.status == ProviderStatus.RECOVERING:
if state.consecutive_successes >= 3:
state.status = ProviderStatus.HEALTHY
state.consecutive_successes = 0
logger.info(f"Provider {provider_name} vollständig recovered")
def _record_failure(self, provider_name: str):
"""Fehlgeschlagene Anfrage registrieren"""
state = self.circuit_breakers[provider_name]
state.failure_count += 1
state.last_failure_time = time.time()
state.consecutive_successes = 0
if state.failure_count >= state.failure_threshold:
state.status = ProviderStatus.CIRCUIT_OPEN
logger.warning(f"Circuit Breaker für {provider_name} geschlossen nach {state.failure_count} Fehlern")
def _get_available_provider(self) -> Optional[str]:
"""Findet nächsten verfügbaren Provider mit Circuit Breaker Check"""
available = []
for name, config in sorted(
self.providers.items(),
key=lambda x: x[1].priority
):
if self._check_circuit_breaker(name):
available.append(name)
return available[0] if available else None
async def complete_text(
self,
prompt: str,
model: str = "deepseek-v3.2",
max_tokens: int = 1000,
temperature: float = 0.7,
context: Optional[List[Dict]] = None
) -> Dict[str, Any]:
"""
Führt AI-Completion mit automatisiertem Failover durch
"""
start_time = time.time()
retry_count = 0
max_retries = 3
while retry_count <= max_retries:
provider_name = self._get_available_provider()
if not provider_name:
# Alle Provider ausgefallen - warte auf Recovery
logger.error("Alle Provider ausgefallen, warte auf Recovery...")
await asyncio.sleep(5)
continue
provider = self.providers[provider_name]
try:
# Rate Limiting
async with self.rate_limiters[provider_name]:
response = await self._make_request(
provider=provider,
model=model,
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
context=context
)
# Erfolg registrieren
self._record_success(provider_name)
# Metrics aktualisieren
latency = (time.time() - start_time) * 1000
self._update_metrics(provider_name, latency, response, max_tokens)
return {
"success": True,
"content": response["choices"][0]["message"]["content"],
"provider": provider_name,
"model": model,
"latency_ms": round(latency, 2),
"cost_usd": self._calculate_cost(model, max_tokens)
}
except httpx.TimeoutException:
logger.warning(f"Timeout bei {provider_name}, Retry {retry_count}/{max_retries}")
self._record_failure(provider_name)
retry_count += 1
except httpx.HTTPStatusError as e:
logger.error(f"HTTP Error {e.response.status_code} bei {provider_name}")
if e.response.status_code >= 500:
self._record_failure(provider_name)
retry_count += 1
else:
raise
except Exception as e:
logger.error(f"Unerwarteter Fehler: {e}")
self._record_failure(provider_name)
retry_count += 1
# Alle Retries exhausted
self.metrics.failed_requests += 1
raise RuntimeError("Alle Provider und Retries exhausted")
async def _make_request(
self,
provider: ProviderConfig,
model: str,
prompt: str,
max_tokens: int,
temperature: float,
context: Optional[List[Dict]]
) -> Dict:
"""Führt den tatsächlichen API-Call durch"""
messages = []
if context:
messages.extend(context)
messages.append({"role": "user", "content": prompt})
request_payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
response = await self.client.post(
f"{provider.base_url}/chat/completions",
json=request_payload,
headers={
"Authorization": f"Bearer {provider.api_key}",
"Content-Type": "application/json"
}
)
response.raise_for_status()
return response.json()
def _update_metrics(
self,
provider_name: str,
latency: float,
response: Dict,
tokens_used: int
):
"""Aktualisiert Metriken für Monitoring"""
self.metrics.total_requests += 1
self.metrics.successful_requests += 1
# Provider-spezifische Stats
if provider_name not in self.metrics.provider_stats:
self.metrics.provider_stats[provider_name] = {
"requests": 0,
"avg_latency": 0,
"success_rate": 100.0
}
stats = self.metrics.provider_stats[provider_name]
stats["requests"] += 1
stats["avg_latency"] = (
(stats["avg_latency"] * (stats["requests"] - 1) + latency)
/ stats["requests"]
)
def _calculate_cost(self, model: str, tokens: int) -> float:
"""Berechnet Kosten basierend auf Modell"""
costs = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
return (costs.get(model, 1.0) * tokens) / 1000
async def health_check_all(self) -> Dict[str, bool]:
"""Führt Health-Checks für alle Provider durch"""
results = {}
for name, config in self.providers.items():
try:
response = await self.client.get(
f"{config.base_url}/models",
headers={"Authorization": f"Bearer {config.api_key}"},
timeout=5.0
)
results[name] = response.status_code == 200
except:
results[name] = False
return results
async def close(self):
"""Räumt Resources auf"""
await self.client.aclose()
Benchmark-Test
async def benchmark_gateway():
"""Misst Performance und Zuverlässigkeit des Gateways"""
gateway = MultiProviderAIGateway()
print("=" * 60)
print("HOLYSHEEP AI GATEWAY BENCHMARK")
print("=" * 60)
# Latenz-Test
latencies = []
for i in range(100):
start = time.time()
try:
result = await gateway.complete_text(
prompt=f"Erkläre kurz: Was ist {i}?",
model="deepseek-v3.2",
max_tokens=50
)
latencies.append((time.time() - start) * 1000)
print(f"Request {i+1}: {result['latency_ms']:.2f}ms via {result['provider']}")
except Exception as e:
print(f"Request {i+1} fehlgeschlagen: {e}")
if latencies:
print(f"\n📊 LATENZ-BENCHMARK:")
print(f" Durchschnitt: {sum(latencies)/len(latencies):.2f}ms")
print(f" Minimum: {min(latencies):.2f}ms")
print(f" Maximum: {max(latencies):.2f}ms")
print(f" P95: {sorted(latencies)[int(len(latencies)*0.95)]:.2f}ms")
# Provider-Status
health = await gateway.health_check_all()
print(f"\n🏥 PROVIDER HEALTH:")
for provider, status in health.items():
state = gateway.circuit_breakers[provider]
print(f" {provider}: {'✓' if status else '✗'} (Circuit: {state.status.value})")
await gateway.close()
if __name__ == "__main__":
asyncio.run(benchmark_gateway())
2. Circuit Breaker mit Exponential Backoff
"""
Erweiterter Circuit Breaker mit Exponential Backoff und Jitter
Implementiert das Bulkhead Pattern für bessere Isolation
"""
import asyncio
import random
from typing import Callable, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Normaler Betrieb
OPEN = "open" # Blockiert Anfragen
HALF_OPEN = "half_open" # Testet Recovery
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5
success_threshold: int = 3
timeout: float = 30.0
half_open_max_calls: int = 3
base_backoff: float = 1.0
max_backoff: float = 60.0
class CircuitBreaker:
"""
Produktionsreifer Circuit Breaker mit:
- Exponential Backoff mit Jitter
- Bulkhead Isolation
- Metriken und Monitoring
"""
def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[datetime] = None
self.half_open_calls = 0
# Metriken
self.total_calls = 0
self.successful_calls = 0
self.failed_calls = 0
self.rejected_calls = 0
# Lock für Thread-Safety
self._lock = asyncio.Lock()
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Führt Funktion mit Circuit Breaker Protection aus"""
async with self._lock:
self.total_calls += 1
# Status-Prüfung
if not self._can_execute():
self.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit Breaker '{self.name}' ist OPEN"
)
# HALF_OPEN Limit prüfen
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.config.half_open_max_calls:
self.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit Breaker '{self.name}' in HALF_OPEN Limit erreicht"
)
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
await self._on_success()
return result
except Exception as e:
await self._on_failure()
raise
def _can_execute(self) -> bool:
"""Prüft ob Anfrage durchgelassen werden darf"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
return True
return False
# HALF_OPEN: Limit wurde oben geprüft
return True
def _should_attempt_reset(self) -> bool:
"""Prüft ob Timeout für Reset erreicht"""
if not self.last_failure_time:
return True
elapsed = datetime.now() - self.last_failure_time
return elapsed >= timedelta(seconds=self.config.timeout)
async def _on_success(self):
"""Behandelt erfolgreichen Aufruf"""
async with self._lock:
self.successful_calls += 1
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._transition_to(CircuitState.CLOSED)
async def _on_failure(self):
"""Behandelt fehlgeschlagenen Aufruf"""
async with self._lock:
self.failed_calls += 1
self.failure_count += 1
self.last_failure_time = datetime.now()
self.success_count = 0
if self.state == CircuitState.CLOSED:
if self.failure_count >= self.config.failure_threshold:
self._transition_to(CircuitState.OPEN)
elif self.state == CircuitState.HALF_OPEN:
# Jeder Fehler in HALF_OPEN öffnet wieder
self._transition_to(CircuitState.OPEN)
def _transition_to(self, new_state: CircuitState):
"""Zustandsübergang mit Logging"""
old_state = self.state
self.state = new_state
if new_state == CircuitState.OPEN:
# Berechne nächsten Reset mit Jitter
self.next_reset = datetime.now() + timedelta(
seconds=self._calculate_backoff()
)
logger.warning(
f"Circuit Breaker '{self.name}' geöffnet. "
f"Reset geplant für {self.next_reset}"
)
elif new_state == CircuitState.HALF_OPEN:
self.half_open_calls = 0
self.success_count = 0
logger.info(f"Circuit Breaker '{self.name}' in HALF_OPEN")
elif new_state == CircuitState.CLOSED:
self.failure_count = 0
self.half_open_calls = 0
logger.info(f"Circuit Breaker '{self.name}' geschlossen")
def _calculate_backoff(self) -> float:
"""Exponential Backoff mit Jitter"""
base = self.config.base_backoff * (2 ** self.failure_count)
capped = min(base, self.config.max_backoff)
# Full Jitter für bessere Verteilung
jitter = random.uniform(0, capped * 0.3)
return capped + jitter
def get_stats(self) -> dict:
"""Gibt aktuelle Statistiken zurück"""
success_rate = (
self.successful_calls / self.total_calls * 100
if self.total_calls > 0 else 0
)
return {
"name": self.name,
"state": self.state.value,
"total_calls": self.total_calls,
"successful_calls": self.successful_calls,
"failed_calls": self.failed_calls,
"rejected_calls": self.rejected_calls,
"success_rate": round(success_rate, 2),
"failure_count": self.failure_count,
"last_failure": self.last_failure_time.isoformat() if self.last_failure_time else None
}
class CircuitBreakerOpenError(Exception):
"""Exception wenn Circuit Breaker offen ist"""
pass
Beispiel: Integration mit Retry
async def resilient_ai_call(
prompt: str,
circuit_breakers: dict,
max_retries: int = 3
):
"""
Vollständig ausfallsichere AI-Anfrage mit:
- Circuit Breaker
- Exponential Backoff Retry
- Automatic Failover
"""
base_delay = 1.0
last_exception = None
for attempt in range(max_retries):
# Versuche jeden verfügbaren Provider
for provider_name, cb in circuit_breakers.items():
if cb.state == CircuitState.OPEN:
continue
try:
# Simulierter API-Call über HolySheep
result = await cb.call(
_call_holysheep_api,
provider_name,
prompt
)
return {
"success": True,
"provider": provider_name,
"attempt": attempt + 1,
"data": result
}
except CircuitBreakerOpenError:
continue
except Exception as e:
logger.error(f"Provider {provider_name} Fehler: {e}")
last_exception = e
continue
# Exponential Backoff vor nächstem Retry
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
logger.info(f"Retry {attempt + 1}/{max_retries} in {delay:.2f}s")
await asyncio.sleep(delay)
raise RuntimeError(
f"Alle Provider ausgefallen nach {max_retries} Versuchen: {last_exception}"
)
async def _call_holysheep_api(provider: str, prompt: str) -> dict:
"""Simuliert API-Call (ersetzt durch echte Implementierung)"""
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500
},
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
timeout=10.0
)
response.raise_for_status()
return response.json()
Usage Example
async def demo():
# Initialisiere Circuit Breaker für jeden Provider
circuit_breakers = {
"holysheep_primary": CircuitBreaker("primary"),
"holysheep_fallback": CircuitBreaker("fallback"),
"holysheep_emergency": CircuitBreaker("emergency")
}
# Führe ausfallsichere Anfrage durch
result = await resilient_ai_call(
prompt="Erkläre mir Hochverfügbarkeit",
circuit_breakers=circuit_breakers
)
print(f"✓ Anfrage erfolgreich über {result['provider']}")
print(f" Versuche: {result['attempt']}")
# Zeige Circuit Breaker Statistiken
for name, cb in circuit_breakers.items():
stats = cb.get_stats()
print(f"\n📊 {name}:")
print(f" State: {stats['state']}")
print(f" Erfolgsrate: {stats['success_rate']}%")
if __name__ == "__main__":
asyncio.run(demo())
3. Cost-Optimierter Model Router
"""
Intelligenter Model Router für Kostenoptimierung
Wählt optimalen Provider basierend auf Task-Typ, Latenz und Kosten
"""
from dataclasses import dataclass
from typing import Optional, List, Dict, Callable
from enum import Enum
import asyncio
class TaskType(Enum):
COMPLETION = "completion" # Texte vervollständigen
CLASSIFICATION = "classification" # Klassifikation/Kategorisierung
EXTRACTION = "extraction" # Information Extraction
REASONING = "reasoning" # Komplexes Reasoning
CREATIVE = "creative" # Kreative Aufgaben
SUMMARIZATION = "summarization" # Zusammenfassungen
CODE = "code" # Code-Generierung
@dataclass
class ModelInfo:
name: str
provider: str
cost_per_1k_input: float
cost_per_1k_output: float
avg_latency_ms: float
max_tokens: int
capabilities: List[str]
def total_cost(self, input_tokens: int, output_tokens: int) -> float:
return (
(input_tokens / 1000) * self.cost_per_1k_input +
(output_tokens / 1000) * self.cost_per_1k_output
)
class CostOptimizerRouter:
"""
Optimiert Model-Auswahl basierend auf:
1. Task-Typ Kompatibilität
2. Kosten
3. Latenz-Anforderungen
4. Verfügbarkeit
"""
# Modell-Katalog mit Preisen (Stand 2026)
MODELS = {
"deepseek-v3.2": ModelInfo(
name="deepseek-v3.2",
provider="holysheep",
cost_per_1k_input=0.14,
cost_per_1k_output=0.28,
avg_latency_ms=45,
max_tokens=64000,
capabilities=["completion", "reasoning", "code", "extraction"]
),
"gemini-2.5-flash": ModelInfo(
name="gemini-2.5-flash",
provider="holysheep",
cost_per_1k_input=0.35,
cost_per_1k_output=2.15,
avg_latency_ms=35,
max_tokens=64000,
capabilities=["completion", "reasoning", "creative", "summarization"]
),
"gpt-4.1": ModelInfo(
name="gpt-4.1",
provider="holysheep",
cost_per_1k_input=2.0,
cost_per_1k_output=6.0,
avg_latency_ms=80,
max_tokens=128000,
capabilities=["completion", "reasoning", "creative", "code", "classification"]
),
"claude-sonnet-4.5": ModelInfo(
name="claude-sonnet-4.5",
provider="holysheep",
cost_per_1k_input=3.0,
cost_per_1k_output=12.0,
avg_latency_ms=90,
max_tokens=200000,
capabilities=["completion", "reasoning", "creative", "code", "classification"]
)
}
# Routing-Strategien
STRATEGIES = {
TaskType.CLASSIFICATION: {
"priority": ["gemini-2.5-flash", "deepseek-v3.2"],
"max_cost_factor": 1.0
},
TaskType.EXTRACTION: {
"priority": ["deepseek-v3.2", "gemini-2.5-flash"],
"max_cost_factor": 1.2
},
TaskType.REASONING: {
"priority": ["deepseek-v3.2", "gpt-4.1", "claude-sonnet-4.5"],
"max_cost_factor": 2.0
},
TaskType.CREATIVE: {
"priority": ["claude-sonnet-4.5", "gpt-4.1", "gemini-2.5-flash"],
"max_cost_factor": 3.0
},
TaskType.SUMMARIZATION: {
"priority": ["gemini-2.5-flash", "deepseek-v3.2"],
"max_cost_factor": 0.8
},
TaskType.CODE: {
"priority": ["deepseek-v3.2", "gpt-4.1", "claude-sonnet-4.5"],
"max_cost_factor": 2.5
},
TaskType.COMPLETION: {
"priority": ["deepseek-v3.2", "gemini-2.5-flash"],
"max_cost_factor": 1.0
}
}
def __init__(self, budget_cap_usd: float = 1000.0):
self.daily_budget = budget_cap_usd
self.daily_spent = 0.0
self.circuit_breakers: Dict[str, asyncio.Event] = {
model: asyncio.Event() for model in self.MODELS
}
# Alle initial verfügbar
for event in self.circuit_breakers.values():
event.set()
def select_model(
self,
task_type: TaskType,
estimated_input_tokens: int,
estimated_output_tokens: int,
priority_latency: bool = False,
priority_cost: bool = True
) -> Optional[ModelInfo]:
"""
Wählt optimalen Model basierend auf Strategie
"""
strategy = self.STRATEGIES.get(task_type, self.STRATEGIES[TaskType.COMPLETION])
priority_order = strategy["priority"]
candidates = []
for model_name in priority_order:
model = self.MODELS.get(model_name)
if not model:
continue
# Prüfe Circuit Breaker
if not self.circuit_breakers[model_name].is_set():
continue
# Prüfe Capability
if task_type.value not in model.capabilities and task_type.value.replace("_", "") not in model.capabilities:
continue
# Berechne Kosten
cost = model.total_cost(estimated_input_tokens, estimated_output_tokens)
# Prüfe Budget
if self.daily_spent + cost > self.daily_budget:
continue
# Kosten-Score (niedriger = besser)
cost_score = cost / 0.01 # Normalisiert
# Latency-Score (niedriger = besser)
latency_score = model.avg_latency_ms / 10
# Finale Score
if priority_cost:
final_score = cost_score * 0.7 + latency_score * 0.3
else:
final_score = latency_score * 0.7 + cost_score * 0.3
candidates.append({
"model": model,
"cost": cost,
"score": final_score
})
if not candidates:
# Fallback: günstigster verfügbarer Model
for model_name, model in self.MODELS.items():
if self.circuit_breakers[model_name].is_set():
return model
return None
# Sortiere nach Score und wähle besten
candidates.sort(key=lambda x: x["score"])
selected = candidates[0]["model"]
# Track Ausgaben
self.daily_spent += candidates[0]["cost"]
return selected
def mark_model_unavailable(self, model_name: str):
"""Markiert Model als nicht verfügbar (Circuit Open)"""
self.circuit_breakers[model_name].clear()
print(f"⚠️ Model {model_name} als unavailable markiert")
def mark_model_available(self, model_name: str):
"""Markiert Model als wieder verfügbar"""
self.circuit_breakers[model_name].set()
print(f"✓ Model {model_name} wieder verfügbar")
def get_budget_status(self) -> Dict:
"""Gibt Budget-Status zurück"""
return {
"daily_budget_usd": self.daily_budget,
"daily_spent_usd": round(self.daily_spent, 4),
"remaining_usd": round(self.daily_budget - self.daily_spent, 4),
"usage_percent": round(self.daily_spent / self.daily_budget * 100, 2)
}
Kostenvergleich Demo
def demo_cost_comparison():
"""Demonstriert Kosteneinsparungen durch intelligent Routing"""
router = CostOptimizerRouter(budget_cap_usd=100.0)
scenarios = [
# (Task, Input-Tokens, Output-Tokens, Häufigkeit/Monat)
(TaskType.CLASSIFICATION, 500, 50, 100000),
(TaskType.SUMM