Bei der Entwicklung produktionsreifer KI-Anwendungen steht jeder Entwickler vor einer fundamentalen Herausforderung: Wie orchestriert man mehrere KI-Modelle effizient, kostengünstig und Ausfallsicher? In diesem praxisorientierten Tutorial zeige ich Ihnen, wie Sie eine professionelle API负载均衡-Architektur implementieren, die mehrere Modelle nahtlos verwaltet, automatische Failover-Mechanismen bietet und dabei bis zu 85% Kosten einspart.
Als langjähriger Backend-Entwickler habe ich unzählige Stunden mit der Integration verschiedener KI-APIs verbracht. Die Erfahrung hat mir gezeigt: Ohne eine robuste Load-Balancing-Strategie wird Ihr System entweder zu teuer, zu langsam oder zu anfällig für Ausfälle. HolySheep AI bietet hier eine elegante All-in-One-Lösung, die ich in diesem Artikel detailliert vorstelle.
为什么需要多模型API负载均衡？
Moderne KI-Anwendungen benötigen verschiedene Modelltypen für unterschiedliche Aufgaben: komplexe Reasoning-Aufgaben erfordern leistungsstarke Modelle wie Claude Sonnet 4.5, während schnelle Inferenzen mit Gemini 2.5 Flash oder DeepSeek V3.2 kosteneffizienter sind. Eine intelligente 智能路由-Strategie verteilt Anfragen basierend auf Komplexität, Latenzanforderungen und Kostenstruktur.
Die Kernvorteile einer solchen Architektur umfassen:
- Kostenoptimierung: Automatische Weiterleitung einfacher Anfragen an günstigere Modelle (DeepSeek V3.2 kostet nur $0.42/MTok)
- 高可用性: Automatischer Failover bei Modell- oder Anbieterstörungen
- 性能提升: Latenz-Optimierung durch geografisch verteilte Endpunkte
- 灵活性: Nahtloser Modellwechsel ohne Änderung der Client-Applikation
HolySheep AI: Ihre zentrale Anlaufstelle für Multi-Modell-APIs
Jetzt registrieren und von einem revolutionären Wechselkursvorteil profitieren: ¥1 entspricht $1, was über 85% Ersparnis gegenüber regulären US-Preisen bedeutet. Das bedeutet konkret:
- GPT-4.1: $8/MTok statt der üblichen $60
- Claude Sonnet 4.5: $15/MTok statt $100
- Gemini 2.5 Flash: $2.50/MTok statt $17.50
- DeepSeek V3.2: $0.42/MTok — der günstigste Einstiegspreis überhaupt
Zusätzlich bietet HolySheep AI Zahlung via WeChat und Alipay, eine benutzerfreundliche Console mit Echtzeit-Nutzungsstatistiken und weniger als 50ms Latenz für optimale Performance. Neuanmeldungen erhalten kostenlose Credits zum Testen.
负载均衡核心概念与策略
Bevor wir in die Implementierung einsteigen, definieren wir die fundamentalen Routing-Strategien:
轮询负载均衡 (Round-Robin)
Die einfachste Methode: Jede Anfrage geht der Reihe nach an ein anderes Modell. Ideal für homogene Workloads mit ähnlicher Komplexität.
加权轮询 (Weighted Round-Robin)
Modelle erhalten Gewichtungen basierend auf Kapazität und Kosten. Günstigere Modelle wie DeepSeek V3.2 erhalten höhere Gewichte für einfache Aufgaben.
基于成本的智能路由 (Cost-Based Routing)
Die fortschrittlichste Strategie analysiert die Anfragekomplexität und wählt das kosteneffizienteste Modell. Komplexe Reasoning-Aufgaben → Claude Sonnet 4.5, einfache Extraktionen → DeepSeek V3.2.
健康检查与故障转移 (Health-Check & Failover)
Kontinuierliche Überwachung der Modellverfügbarkeit mit automatischer Weiterleitung bei Ausfällen. Kritisch für Produktionssysteme.
实战：Python实现智能负载均衡
Im folgenden Code-Beispiel implementiere ich eine vollständige Load-Balancing-Architektur mit HolySheep AI als zentralem Proxy. Die Implementierung nutzt moderne async/await-Patterns für maximale Performance.
#!/usr/bin/env python3
"""
Multi-Model API Load Balancer mit intelligentem Routing
Verwendet HolySheep AI als zentralen API-Endpunkt
"""
import asyncio
import hashlib
import time
from typing import Optional, Dict, List, Callable
from dataclasses import dataclass, field
from enum import Enum
import httpx
class RoutingStrategy(Enum):
ROUND_ROBIN = "round_robin"
WEIGHTED = "weighted"
COST_BASED = "cost_based"
LATENCY_BASED = "latency_based"
@dataclass
class ModelConfig:
name: str
provider: str
cost_per_1k_tokens: float # in USD
max_tokens: int
typical_latency_ms: float
weight: int = 1
is_healthy: bool = True
consecutive_failures: int = 0
@dataclass
class LoadBalancerConfig:
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
strategy: RoutingStrategy = RoutingStrategy.COST_BASED
max_retries: int = 3
timeout_seconds: float = 30.0
health_check_interval: int = 60
failover_threshold: int = 3
class MultiModelLoadBalancer:
"""
Intelligenter Load Balancer für Multi-Model API Routing
Features:
- Kostenbasiertes Routing
- Automatischer Failover
- Latenz-Monitoring
- Health Checks
"""
def __init__(self, config: LoadBalancerConfig):
self.config = config
self.models: Dict[str, ModelConfig] = {}
self.current_index: int = 0
self.request_counts: Dict[str, int] = {}
self.latency_history: Dict[str, List[float]] = {}
self.total_requests: int = 0
self.failed_requests: int = 0
# HolySheep unterstützte Modelle initialisieren
self._initialize_models()
def _initialize_models(self):
"""Initialisiert verfügbare Modelle mit Kosten und Gewichtungen"""
self.models = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai",
cost_per_1k_tokens=8.0, # $8/MTok bei HolySheep
max_tokens=128000,
typical_latency_ms=850,
weight=3
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
cost_per_1k_tokens=15.0, # $15/MTok bei HolySheep
max_tokens=200000,
typical_latency_ms=920,
weight=2
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="google",
cost_per_1k_tokens=2.5, # $2.50/MTok bei HolySheep
max_tokens=1000000,
typical_latency_ms=380,
weight=5
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
cost_per_1k_tokens=0.42, # $0.42/MTok bei HolySheep
max_tokens=64000,
typical_latency_ms=320,
weight=8
)
}
# Statistik-Tracker initialisieren
for model_name in self.models:
self.request_counts[model_name] = 0
self.latency_history[model_name] = []
def _analyze_complexity(self, prompt: str) -> str:
"""
Analysiert die Anfragekomplexität für kosteneffizientes Routing
Returns: 'simple', 'moderate', oder 'complex'
"""
complexity_indicators = {
'simple': ['list', 'extract', 'summarize', 'translate', 'what is', 'define'],
'moderate': ['compare', 'analyze', 'explain', 'describe', 'why', 'how'],
'complex': ['reason', 'prove', 'design', 'architect', 'complex', 'multi-step']
}
prompt_lower = prompt.lower()
scores = {'simple': 0, 'moderate': 0, 'complex': 0}
for level, keywords in complexity_indicators.items():
for keyword in keywords:
if keyword in prompt_lower:
scores[level] += 1
max_level = max(scores, key=scores.get)
return max_level if scores[max_level] > 0 else 'simple'
def _select_model(self, prompt: str, force_model: Optional[str] = None) -> str:
"""
Wählt basierend auf der Routing-Strategie das optimale Modell
"""
if force_model and force_model in self.models:
return force_model
strategy = self.config.strategy
if strategy == RoutingStrategy.COST_BASED:
complexity = self._analyze_complexity(prompt)
# Complexity-basierte Modellauswahl
if complexity == 'complex':
# Komplexe Reasoning-Aufgaben → Leistungsstarke Modelle
candidates = ['claude-sonnet-4.5', 'gpt-4.1']
elif complexity == 'moderate':
# Moderate Aufgaben → Ausgewogenes Verhältnis
candidates = ['gemini-2.5-flash', 'claude-sonnet-4.5']
else:
# Einfache Aufgaben → Günstige, schnelle Modelle
candidates = ['deepseek-v3.2', 'gemini-2.5-flash']
# Wähle das verfügbare Modell mit den wenigsten Requests
available = [m for m in candidates
if self.models[m].is_healthy]
if not available:
# Fallback zu irgendeinem gesunden Modell
available = [m for m in self.models if self.models[m].is_healthy]
if not available:
raise Exception("Keine gesunden Modelle verfügbar")
return min(available, key=lambda m: self.request_counts[m])
elif strategy == RoutingStrategy.ROUND_ROBIN:
available = [m for m in self.models if self.models[m].is_healthy]
selected = available[self.current_index % len(available)]
self.current_index += 1
return selected
elif strategy == RoutingStrategy.WEIGHTED:
# Weighted Random Selection
available = [m for m in self.models if self.models[m].is_healthy]
weights = [self.models[m].weight for m in available]
total = sum(weights)
import random
r = random.uniform(0, total)
cumsum = 0
for i, w in enumerate(weights):
cumsum += w
if cumsum >= r:
return available[i]
return available[-1]
elif strategy == RoutingStrategy.LATENCY_BASED:
# Niedrigste Latenz basierend auf History
available = [m for m in self.models if self.models[m].is_healthy
and self.latency_history.get(m)]
if not available:
return 'gemini-2.5-flash' # Fallback
return min(available,
key=lambda m: sum(self.latency_history[m]) / len(self.latency_history[m]))
# Default
return 'gemini-2.5-flash'
async def _make_request(self, model: str, prompt: str,
system_prompt: str = "You are a helpful assistant.") -> Dict:
"""Führt einen API-Request durch und misst Latenz"""
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"max_tokens": 2048,
"temperature": 0.7
}
start_time = time.perf_counter()
async with httpx.AsyncClient(timeout=self.config.timeout_seconds) as client:
response = await client.post(
f"{self.config.base_url}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status_code == 200:
data = response.json()
# Latenz-Statistik aktualisieren
self.latency_history[model].append(latency_ms)
if len(self.latency_history[model]) > 100:
self.latency_history[model].pop(0)
# Request-Zähler aktualisieren
self.request_counts[model] += 1
self.total_requests += 1
# Modell als gesund markieren
self.models[model].is_healthy = True
self.models[model].consecutive_failures = 0
return {
"success": True,
"model": model,
"latency_ms": round(latency_ms, 2),
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {})
}
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
async def _handle_failover(self, original_model: str, prompt: str,
system_prompt: str, retries: int) -> Dict:
"""Behandelt Failover bei Modellfehlern"""
self.models[original_model].consecutive_failures += 1
# Markiere Modell als ungesund nach Schwellenwert
if (self.models[original_model].consecutive_failures
>= self.config.failover_threshold):
self.models[original_model].is_healthy = False
print(f"⚠️ Modell {original_model} als ungesund markiert")
# Wähle alternatives Modell
healthy_models = [m for m in self.models if self.models[m].is_healthy]
if not healthy_models:
raise Exception("Alle Modelle sind ausgefallen")
# Versuche anderes Modell
for alternative in healthy_models:
if alternative != original_model:
try:
return await self._make_request(alternative, prompt, system_prompt)
except Exception as e:
print(f"❌ Failover zu {alternative} fehlgeschlagen: {e}")
continue
raise Exception("Alle Failover-Versuche fehlgeschlagen")
async def chat(self, prompt: str,
system_prompt: str = "You are a helpful assistant.",
force_model: Optional[str] = None) -> Dict:
"""
Hauptmethode: Sendet Chat-Request mit automatischem Routing
"""
model = self._select_model(prompt, force_model)
for attempt in range(self.config.max_retries):
try:
return await self._make_request(model, prompt, system_prompt)
except Exception as e:
print(f"⚠️ Attempt {attempt + 1} fehlgeschlagen: {e}")
if attempt < self.config.max_retries - 1:
result = await self._handle_failover(model, prompt,
system_prompt, attempt)
if result:
return result
self.failed_requests += 1
raise Exception(f"Request nach {self.config.max_retries} Versuchen fehlgeschlagen")
def get_statistics(self) -> Dict:
"""Liefert aktuelle Load-Balancer-Statistiken"""
avg_latencies = {}
for model, history in self.latency_history.items():
if history:
avg_latencies[model] = round(sum(history) / len(history), 2)
return {
"total_requests": self.total_requests,
"failed_requests": self.failed_requests,
"success_rate": round(
(self.total_requests - self.failed_requests) / max(self.total_requests, 1) * 100, 2
),
"requests_by_model": self.request_counts.copy(),
"average_latencies_ms": avg_latencies,
"healthy_models": [m for m in self.models if self.models[m].is_healthy],
"unhealthy_models": [m for m in self.models if not self.models[m].is_healthy]
}
async def health_check(self):
"""Führt periodischen Health-Check durch"""
print("🔍 Führe Health-Check durch...")
for model_name, model in self.models.items():
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{self.config.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.config.api_key}"},
json={
"model": model_name,
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 10
}
)
if response.status_code == 200:
if not model.is_healthy:
print(f"✅ Modell {model_name} wiederhergestellt")
model.is_healthy = True
model.consecutive_failures = 0
else:
model.consecutive_failures += 1
except Exception as e:
model.consecutive_failures += 1
if model.consecutive_failures >= self.config.failover_threshold:
model.is_healthy = False
Beispiel-Nutzung
async def main():
config = LoadBalancerConfig(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
strategy=RoutingStrategy.COST_BASED,
max_retries=3
)
balancer = MultiModelLoadBalancer(config)
# Test-Anfragen mit verschiedenen Komplexitäten
test_prompts = [
("Einfach", "Was ist Python?"),
("Moderat", "Erkläre den Unterschied zwischen Python und JavaScript"),
("Komplex", "Entwirf eine Microservice-Architektur für eine E-Commerce-Plattform mit Load Balancing und Caching-Strategien")
]
print("🚀 Starte Load Balancer Tests...\n")
for complexity, prompt in test_prompts:
print(f"📝 [{complexity}] Anfrage: {prompt[:50]}...")
try:
result = await balancer.chat(prompt)
print(f" ✅ Modell: {result['model']}")
print(f" ⚡ Latenz: {result['latency_ms']}ms")
print(f" 📊 Antwort: {result['content'][:100]}...\n")
except Exception as e:
print(f" ❌ Fehler: {e}\n")
# Statistiken ausgeben
print("📈 Load Balancer Statistiken:")
stats = balancer.get_statistics()
for key, value in stats.items():
print(f" {key}: {value}")
if __name__ == "__main__":
asyncio.run(main())Dieses Load-Balancer-System bietet folgende Kernfeatures:
- Automatische Komplexitätsanalyse: Erkennt die Anfragekomplexität und wählt das kosteneffizienteste Modell
- Mehrere Routing-Strategien: Round-Robin, Weighted, Cost-Based und Latency-Based
- Intelligenter Failover: Automatische Weiterleitung bei Modellfehlern
- Echtzeit-Statistiken: Verfolgung von Latenz, Erfolgsrate und Modellnutzung
生产环境配置示例
Für produktive Deployments empfehle ich die Verwendung von Redis für zustandsbehaftetes Routing und Kubernetes für horizontale Skalierung. Hier ein erweitertes Beispiel mit erweiterten Features:
#!/usr/bin/env python3
"""
Produktionsreifes Load Balancer Setup mit Redis und Monitoring
Integration: HolySheep AI API
"""
import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import redis.asyncio as redis
import httpx
from dataclasses import dataclass, field
from collections import defaultdict
import aiofiles
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RequestMetrics:
"""Metriken für eine einzelne Anfrage"""
request_id: str
model: str
prompt_length: int
response_length: int
latency_ms: float
tokens_used: int
cost_usd: float
timestamp: datetime
success: bool
error_message: Optional[str] = None
@dataclass
class ModelMetrics:
"""Aggregierte Metriken pro Modell"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
total_tokens: int = 0
total_cost_usd: float = 0.0
avg_latency_ms: float = 0.0
min_latency_ms: float = float('inf')
max_latency_ms: float = 0.0
last_request: Optional[datetime] = None
consecutive_failures: int = 0
is_healthy: bool = True
class ProductionLoadBalancer:
"""
Produktionsreifer Load Balancer mit:
- Redis-basierter Zustandsverwaltung
- Detailliertem Monitoring
- Kostenverfolgung
- Rate Limiting
- Request-Queuing
"""
# HolySheep API Preise (USD pro 1M Tokens)
MODEL_PRICES = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.5, "output": 2.5},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
# Latenz-Garantien von HolySheep
LATENCY_SLA = {
"deepseek-v3.2": {"p50": 320, "p95": 450, "p99": 600},
"gemini-2.5-flash": {"p50": 380, "p95": 520, "p99": 700},
"gpt-4.1": {"p50": 850, "p95": 1200, "p99": 1800},
"claude-sonnet-4.5": {"p50": 920, "p95": 1400, "p99": 2200}
}
def __init__(self, redis_url: str, api_key: str, base_url: str =
"https://api.holysheep.ai/v1"):
self.redis = redis.from_url(redis_url)
self.api_key = api_key
self.base_url = base_url
self.metrics: Dict[str, ModelMetrics] = defaultdict(
lambda: ModelMetrics()
)
self.request_queue = asyncio.Queue(maxsize=1000)
self.shutdown_event = asyncio.Event()
async def _estimate_cost(self, model: str, prompt_tokens: int,
completion_tokens: int) -> float:
"""Berechnet geschätzte Kosten basierend auf HolySheep-Preisen"""
prices = self.MODEL_PRICES.get(model, {"input": 0, "output": 0})
input_cost = (prompt_tokens / 1_000_000) * prices["input"]
output_cost = (completion_tokens / 1_000_000) * prices["output"]
return round(input_cost + output_cost, 6) # 6 Dezimalstellen für Cent-Genauigkeit
async def _record_metrics(self, metrics: RequestMetrics):
"""Speichert Metriken in Redis und lokal"""
model_key = f"metrics:{metrics.model}"
model_metrics = self.metrics[metrics.model]
# Lokale Metriken aktualisieren
model_metrics.total_requests += 1
model_metrics.total_latency_ms += metrics.latency_ms
model_metrics.total_tokens += metrics.tokens_used
model_metrics.total_cost_usd += metrics.cost_usd
model_metrics.last_request = metrics.timestamp
if metrics.success:
model_metrics.successful_requests += 1
model_metrics.consecutive_failures = 0
else:
model_metrics.failed_requests += 1
model_metrics.consecutive_failures += 1
if model_metrics.consecutive_failures >= 5:
model_metrics.is_healthy = False
logger.warning(f"⚠️ Modell {metrics.model} als ungesund markiert")
model_metrics.avg_latency_ms = (
model_metrics.total_latency_ms / model_metrics.total_requests
)
model_metrics.min_latency_ms = min(
model_metrics.min_latency_ms, metrics.latency_ms
)
model_metrics.max_latency_ms = max(
model_metrics.max_latency_ms, metrics.latency_ms
)
# Redis-Storage für persistente Metriken
await self.redis.hincrby(model_key, "total_requests", 1)
await self.redis.hincrbyfloat(model_key, "total_cost_usd", metrics.cost_usd)
await self.redis.hincrbyfloat(model_key, "total_latency_ms", metrics.latency_ms)
if metrics.success:
await self.redis.hincrby(model_key, "successful_requests", 1)
else:
await self.redis.hincrby(model_key, "failed_requests", 1)
# Request-Historie (Rolling Window)
request_data = json.dumps({
"request_id": metrics.request_id,
"latency_ms": metrics.latency_ms,
"cost_usd": metrics.cost_usd,
"timestamp": metrics.timestamp.isoformat(),
"success": metrics.success
})
await self.redis.lpush(f"history:{metrics.model}", request_data)
await self.redis.ltrim(f"history:{metrics.model}", 0, 999) # Letzte 1000
async def _smart_route(self, prompt: str, preferred_latency: Optional[int] = None,
max_cost_per_1k: Optional[float] = None) -> str:
"""
Intelligente Routinge-Entscheidung basierend auf:
1. Modell-Gesundheit
2. Latenz-Anforderungen
3. Kostenbudget
4. Aktueller Load
"""
# Hole verfügbare Modelle aus Redis
available_models = []
for model in self.MODEL_PRICES:
# Prüfe Gesundheitsstatus
health_key = f"metrics:{model}"
consecutive_failures = await self.redis.hget(
health_key, "consecutive_failures"
) or 0
if consecutive_failures and int(consecutive_failures) >= 5:
continue
# Prüfe Kosten-Budget
if max_cost_per_1k:
if self.MODEL_PRICES[model]["input"] > max_cost_per_1k:
continue
# Prüfe Latenz-Anforderungen
if preferred_latency:
p99_latency = self.LATENCY_SLA[model]["p99"]
if p99_latency > preferred_latency:
continue
# Hole aktuelle Load
load = await self.redis.get(f"load:{model}") or 0
available_models.append((model, int(load)))
if not available_models:
raise Exception("Keine geeigneten Modelle verfügbar")
# Wähle Modell mit niedrigster Load
selected = min(available_models, key=lambda x: x[1])[0]
# Incrementiere Load
await self.redis.incr(f"load:{selected}")
return selected
async def _decrement_load(self, model: str):
"""Dekrementiert Load-Counter nach Request"""
current = await self.redis.get(f"load:{model}") or 1
await self.redis.set(f"load:{model}", max(0, int(current) - 1))
async def chat_completion(self, prompt: str,
system_prompt: str = "Du bist ein hilfreicher Assistent.",
preferred_latency: Optional[int] = None,
max_cost_per_1k: Optional[float] = None,
request_id: Optional[str] = None) -> Dict[str, Any]:
"""
Führt einen Chat-Completion-Request mit vollem Monitoring durch
"""
import uuid
request_id = request_id or str(uuid.uuid4())
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
# Intelligente Modellauswahl
model = await self._smart_route(prompt, preferred_latency, max_cost_per_1k)
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"max_tokens": 4096,
"temperature": 0.7
}
start_time = datetime.now()
try:
async with httpx.AsyncClient(
timeout=httpx.Timeout(60.0, connect=10.0)
) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
cost = await self._estimate_cost(
model, prompt_tokens, completion_tokens
)
# Metriken aufzeichnen
metrics = RequestMetrics(
request_id=request_id,
model=model,
prompt_length=len(prompt),
response_length=len(data["choices"][0]["message"]["content"]),
latency_ms=latency_ms,
tokens_used=total_tokens,
cost_usd=cost,
timestamp=start_time,
success=True
)
await self._record_metrics(metrics)
return {
"success": True,
"request_id": request_id,
"model": model,
"latency_ms": round(latency_ms, 2),
"content": data["choices"][0]["message"]["content"],
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens
},
"cost_usd": cost,
"timestamp": start_time.isoformat()
}
else:
raise Exception(f"API Error: {response.status_code}")
except Exception as e:
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
# Fehler-Metriken aufzeichnen
metrics = RequestMetrics(
request_id=request_id,
model=model,
prompt_length=len(prompt),
response_length=0,
latency_ms=latency_ms,
tokens_used=0,
cost_usd=0.0,
timestamp=start_time,
success=False,
error_message=str(e)
)
await self._record_metrics(metrics)
# Versuche Failover
logger.warning(f"Request {request_id} fehlgeschlagen, versuche Failover...")
for _ in range(3):
try:
new_model = await self._smart_route(prompt, preferred_latency, max_cost_per_1k)
if new_model != model:
payload["model"] = new_model
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
data = response.json()
return {
"success": True,
"request_id": request_id,
"model": new_model,
"failover": True,
"original_model": model,
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {})
}
except:
continue
raise Exception(f"Request fehlgeschlagen: {e}")
finally:
await self._decrement_load(model)
async def get_dashboard_stats(self) -> Dict[str, Any]:
"""Liefert Statistiken für Dashboard"""
total_cost = sum(m.total_cost_usd for m in self.metrics.values())
total_requests = sum(m.total_requests for m in self.metrics.values())
total_successful = sum(m.successful_requests for m in self.metrics.values())
return {
"summary": {
"total_requests": total_requests,
"successful_requests": total_successful,
"failed_requests": total_requests - total_successful,
"success_rate": round(
total_successful / max(total_requests, 1