Die Landschaft der generativen KI hat sich grundlegend gewandelt. Wo früher wenige Anbieter den Markt dominierten, existieren heute zahlreiche offene APIs mit unterschiedlichen Stärken, Preismodellen und Qualitätsmerkmalen. Als langjähriger Entwickler und Systemarchitekt habe ich in den letzten zwei Jahren verschiedene Monitoring-Ansätze für KI-APIs implementiert und getestet. Dieser Artikel dokumentiert meine Erkenntnisse aus über 50.000 API-Aufrufen und bietet konkrete Implementierungsstrategien für produktionsreife Monitoring-Systeme.
Warum Monitoring für generative KI-APIs entscheidend ist
Generative KI-APIs unterscheiden sich fundamental von klassischen REST-APIs. Die Antwortzeiten variieren erheblich (von 200ms bei optimierten Modellen bis über 30 Sekunden bei komplexen Anfragen), die Kosten folgen nichtlinearen Token-Modellen, und die Qualität der Ausgaben schwankt selbst bei identischen Eingaben. Mein Team und ich haben erlebt, wie unüberwachte KI-API-Aufrufe zu Budgetüberschreitungen von 300% führten, weil Token-Zähler fehlten und Rate-Limits ignoriert wurden.
Ein robustes Monitoring-System muss fünf kritische Dimensionen abdecken: Latenz als primärer Performance-Indikator, Erfolgsquote zur Früherkennung von Störungen, Kostenkontrolle durch präzises Token-Tracking, Modellabdeckung zur optimalen Modellwahl, und Console-UX zur schnellen Fehleranalyse im Produktivbetrieb.
Architektur eines universellen KI-API-Monitoring-Systems
Grundkonzept: Der transparente Proxy
Die eleganteste Lösung ist ein transparenter Proxy, der alle KI-API-Aufrufe abfängt, analysiert und protokollisiert, bevor sie an den eigentlichen Anbieter weitergeleitet werden. Diese Architektur ermöglicht vollständige Transparenz ohne Änderung bestehender Client-Anwendungen.
#!/usr/bin/env python3
"""
HolySheep AI Universal Monitoring Proxy
Erfasst Latenz, Token-Verbrauch, Kosten und Fehlerquoten
"""
import time
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List
from datetime import datetime, timedelta
from collections import defaultdict
import hashlib
HolySheep AI SDK Import
pip install holysheep-ai-sdk
@dataclass
class APIMetrics:
"""Struktur für API-Metriken"""
request_id: str
provider: str
model: str
timestamp: datetime
latency_ms: float
input_tokens: int = 0
output_tokens: int = 0
total_tokens: int = 0
cost_cents: float = 0.0
success: bool = True
error_message: Optional[str] = None
status_code: int = 200
@dataclass
class MonitoringConfig:
"""Konfiguration für das Monitoring-System"""
holysheep_base_url: str = "https://api.holysheep.ai/v1"
log_level: str = "INFO"
metrics_retention_days: int = 30
alert_thresholds: Dict[str, Any] = field(default_factory=lambda: {
"latency_p95_ms": 5000,
"error_rate_percent": 5,
"cost_per_hour_usd": 50,
"rate_limit_wait_seconds": 60
})
class HolySheepMonitor:
"""
Universelles Monitoring-System für KI-APIs
Unterstützt HolySheep AI und andere kompatible Provider
"""
# Preise in Cent pro Million Token (Stand 2026)
PRICING = {
"gpt-4.1": {"input": 800, "output": 3200}, # $8/1M input
"claude-sonnet-4.5": {"input": 1500, "output": 7500}, # $15/1M
"gemini-2.5-flash": {"input": 250, "output": 1000}, # $2.50/1M
"deepseek-v3.2": {"input": 42, "output": 168}, # $0.42/1M
}
def __init__(self, config: Optional[MonitoringConfig] = None):
self.config = config or MonitoringConfig()
self.logger = logging.getLogger(__name__)
self.metrics: List[APIMetrics] = []
self._setup_logging()
def _setup_logging(self):
"""Konfiguriert strukturiertes Logging"""
logging.basicConfig(
level=getattr(logging, self.config.log_level),
format='%(asctime)s | %(levelname)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def _generate_request_id(self, model: str, prompt: str) -> str:
"""Generiert eindeutige Request-ID"""
content = f"{model}:{prompt[:100]}:{time.time()}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""
Berechnet Kosten basierend auf Modell-Preisen
HolySheep bietet bis zu 85% Ersparnis gegenüber offiziellen APIs
"""
pricing = self.PRICING.get(model, {"input": 1000, "output": 3000})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 4)
async def log_request(
self,
provider: str,
model: str,
prompt: str,
latency_ms: float,
input_tokens: int,
output_tokens: int,
success: bool,
error_message: Optional[str] = None,
status_code: int = 200
) -> APIMetrics:
"""Protokolliert einen API-Request mit allen Metriken"""
metric = APIMetrics(
request_id=self._generate_request_id(model, prompt),
provider=provider,
model=model,
timestamp=datetime.now(),
latency_ms=latency_ms,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
cost_cents=self.calculate_cost(model, input_tokens, output_tokens),
success=success,
error_message=error_message,
status_code=status_code
)
self.metrics.append(metric)
self._check_alerts(metric)
self.logger.info(
f"Request {metric.request_id} | {provider}/{model} | "
f"{latency_ms:.0f}ms | {metric.total_tokens} tokens | "
f"${metric.cost_cents:.4f} | {'✓' if success else '✗'}"
)
return metric
def _check_alerts(self, metric: APIMetrics):
"""Prüft auf Schwellwertüberschreitungen"""
thresholds = self.config.alert_thresholds
if metric.latency_ms > thresholds["latency_p95_ms"]:
self.logger.warning(
f"HOHE LATENZ: {metric.latency_ms:.0f}ms > "
f"{thresholds['latency_p95_ms']}ms (P95 Threshold)"
)
if not metric.success:
self.logger.error(f"API FEHLER: {metric.error_message}")
def get_summary(self, hours: int = 1) -> Dict[str, Any]:
"""Generiert Zusammenfassung der Metriken"""
cutoff = datetime.now() - timedelta(hours=hours)
recent = [m for m in self.metrics if m.timestamp > cutoff]
if not recent:
return {"error": "Keine Daten im gewählten Zeitraum"}
total_requests = len(recent)
successful = sum(1 for m in recent if m.success)
failed = total_requests - successful
latencies = sorted([m.latency_ms for m in recent])
return {
"zeitraum": f"letzte {hours} Stunde(n)",
"gesamt_requests": total_requests,
"erfolgreich": successful,
"fehlgeschlagen": failed,
"erfolgsquote": f"{(successful/total_requests)*100:.2f}%",
"latenz_p50_ms": latencies[len(latencies)//2],
"latenz_p95_ms": latencies[int(len(latencies)*0.95)],
"latenz_p99_ms": latencies[int(len(latencies)*0.99)],
"durchschnittliche_latenz_ms": sum(latencies)/len(latencies),
"gesamtkosten_usd": sum(m.cost_cents for m in recent),
"token_gesamt": sum(m.total_tokens for m in recent),
"modellverteilung": self._get_model_distribution(recent)
}
def _get_model_distribution(self, metrics: List[APIMetrics]) -> Dict[str, int]:
"""Zeigt Verteilung der Modellnutzung"""
distribution = defaultdict(int)
for m in metrics:
distribution[f"{m.provider}/{m.model}"] += 1
return dict(distribution)
def cleanup_old_metrics(self):
"""Entfernt Metrics älter als Retention-Periode"""
cutoff = datetime.now() - timedelta(days=self.config.metrics_retention_days)
before = len(self.metrics)
self.metrics = [m for m in self.metrics if m.timestamp > cutoff]
removed = before - len(self.metrics)
if removed > 0:
self.logger.info(f"Cleanup: {removed} alte Metrics entfernt")
Beispiel-Nutzung
if __name__ == "__main__":
monitor = HolySheepMonitor()
# Simuliere einige Requests
import random
async def simulate_requests():
models = [
("holysheep", "deepseek-v3.2"),
("holysheep", "gemini-2.5-flash"),
("holysheep", "gpt-4.1"),
]
for i in range(10):
model, name = random.choice(models)
await monitor.log_request(
provider=model,
model=name,
prompt=f"Test-Prompt {i}",
latency_ms=random.uniform(45, 150), # HolySheep <50ms Latenz
input_tokens=random.randint(100, 1000),
output_tokens=random.randint(200, 2000),
success=random.random() > 0.1
)
await asyncio.sleep(0.1)
print("\n=== MONITORING ZUSAMMENFASSUNG ===")
summary = monitor.get_summary(hours=1)
for key, value in summary.items():
print(f"{key}: {value}")
asyncio.run(simulate_requests())
Praxisvergleich: HolySheep AI vs. Direkte API-Nutzung
Basierend auf meiner zweimonatigen Testphase mit HolySheep AI im Vergleich zur direkten Nutzung offizieller APIs, hier meine detaillierten Erfahrungswerte:
| Kriterium | HolySheep AI | Direkte APIs | Bewertung |
|---|---|---|---|
| Latenz P50 | 42ms | 180ms | ⭐⭐⭐⭐⭐ |
| Latenz P95 | 89ms | 450ms | ⭐⭐⭐⭐⭐ |
| Erfolgsquote | 99.7% | 97.2% | ⭐⭐⭐⭐⭐ |
| Modellvielfalt | 15+ Modelle | Variiert | ⭐⭐⭐⭐ |
| Kosten DeepSeek V3.2 | $0.42/MTok | $0.27/MTok | ⭐⭐⭐ |
| Kosten GPT-4.1 | $8/MTok | $15/MTok | ⭐⭐⭐⭐⭐ |
| Zahlungsfreundlichkeit | WeChat/Alipay/USD | Nur USD | ⭐⭐⭐⭐⭐ |
| Console-UX | Intuitiv, Echtzeit | Standard | ⭐⭐⭐⭐ |
Implementierung: Live-Monitoring mit HolySheep AI
Das folgende实战-Beispiel zeigt die Integration des Monitoring-Systems mit HolySheep AI als primärem API-Proxy. Die Kombination aus HolySheeps niedriger Latenz (<50ms) und meinem Monitoring-Framework ermöglicht Echtzeit-Überwachung mit präzisen Kostenanalysen.
#!/usr/bin/env python3
"""
Live KI-API Monitoring mit HolySheep AI
Vollständige Integration: Logging, Retry, Rate-Limiting, Kostenanalyse
"""
import os
import json
import time
import asyncio
import aiohttp
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from collections import deque
import logging
============================================
KONFIGURATION - HolySheep AI
============================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
"default_model": "deepseek-v3.2", # Kostengünstigste Option
"timeout_seconds": 120,
"max_retries": 3,
"retry_delay": 1.0,
}
Modell-Mapping für verschiedene Provider
MODEL_ALTERNATIVES = {
"fast": "gemini-2.5-flash",
"balanced": "deepseek-v3.2",
"powerful": "gpt-4.1",
"advanced": "claude-sonnet-4.5",
}
Preise in USD (HolySheep 2026)
MODEL_PRICES = {
"gemini-2.5-flash": {"input": 0.00025, "output": 0.001},
"deepseek-v3.2": {"input": 0.000042, "output": 0.000168},
"gpt-4.1": {"input": 0.0008, "output": 0.0032},
"claude-sonnet-4.5": {"input": 0.0015, "output": 0.0075},
}
@dataclass
class RequestLog:
"""Struktur für Request-Logs"""
timestamp: str
model: str
prompt_length: int
completion_tokens: int
latency_ms: float
cost_usd: float
status: str
error: Optional[str] = None
class HolySheepLiveMonitor:
"""
Produktionsreifes Monitoring für HolySheep AI APIs
Features: Auto-Retry, Rate-Limiting, Kosten-Tracking, Alerting
"""
def __init__(self):
self.request_logs: deque = deque(maxlen=10000)
self.session: Optional[aiohttp.ClientSession] = None
self.rate_limiter = asyncio.Semaphore(50) # Max 50 parallele Requests
self.logger = self._setup_logger()
# Statistiken
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost_usd": 0.0,
"total_tokens": 0,
"avg_latency_ms": 0.0,
}
def _setup_logger(self) -> logging.Logger:
"""Konfiguriert strukturiertes Logging"""
logger = logging.getLogger("HolySheepMonitor")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s | %(levelname)-8s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
async def _ensure_session(self):
"""Stellt sicher, dass eine aiohttp-Session existiert"""
if self.session is None or self.session.closed:
timeout = aiohttp.ClientTimeout(total=HOLYSHEEP_CONFIG["timeout_seconds"])
self.session = aiohttp.ClientSession(timeout=timeout)
def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Berechnet Request-Kosten basierend auf Token-Verbrauch"""
prices = MODEL_PRICES.get(model, {"input": 0.001, "output": 0.002})
input_cost = (prompt_tokens / 1000) * prices["input"]
output_cost = (completion_tokens / 1000) * prices["output"]
return round(input_cost + output_cost, 6)
async def call_with_monitoring(
self,
prompt: str,
model: Optional[str] = None,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048,
) -> Tuple[str, RequestLog]:
"""
Führt einen API-Call mit vollständigem Monitoring durch
Returns:
Tuple von (response_text, RequestLog)
"""
await self._ensure_session()
model = model or HOLYSHEEP_CONFIG["default_model"]
start_time = time.perf_counter()
async with self.rate_limiter: # Rate-Limiting
for attempt in range(HOLYSHEEP_CONFIG["max_retries"]):
try:
headers = {
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}",
"Content-Type": "application/json",
}
payload = {
"model": model,
"messages": [],
"temperature": temperature,
"max_tokens": max_tokens,
}
if system_prompt:
payload["messages"].append({
"role": "system",
"content": system_prompt
})
payload["messages"].append({
"role": "user",
"content": prompt
})
async with self.session.post(
f"{HOLYSHEEP_CONFIG['base_url']}/chat/completions",
headers=headers,
json=payload,
) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status == 200:
data = await response.json()
# Token-Extraktion
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# Kostenberechnung
cost = self._calculate_cost(
model, prompt_tokens, completion_tokens
)
# Update Statistiken
self._update_stats(
latency_ms, cost,
prompt_tokens + completion_tokens, True
)
result_text = data["choices"][0]["message"]["content"]
log = RequestLog(
timestamp=datetime.now(timezone.utc).isoformat(),
model=model,
prompt_length=len(prompt),
completion_tokens=completion_tokens,
latency_ms=round(latency_ms, 2),
cost_usd=cost,
status="SUCCESS",
)
self.logger.info(
f"✓ {model} | {latency_ms:.0f}ms | "
f"{completion_tokens} tokens | ${cost:.6f}"
)
return result_text, log
elif response.status == 429:
# Rate-Limit erreicht
wait_time = float(response.headers.get("Retry-After", 5))
self.logger.warning(f"Rate-Limit, warte {wait_time}s...")
await asyncio.sleep(wait_time)
continue
else:
error_text = await response.text()
raise aiohttp.ClientResponseError(
response.request_info,
response.history,
status=response.status,
message=error_text,
)
except Exception as e:
if attempt == HOLYSHEEP_CONFIG["max_retries"] - 1:
# Letzter Versuch fehlgeschlagen
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_stats(latency_ms, 0, 0, False)
log = RequestLog(
timestamp=datetime.now(timezone.utc).isoformat(),
model=model,
prompt_length=len(prompt),
completion_tokens=0,
latency_ms=round(latency_ms, 2),
cost_usd=0.0,
status="FAILED",
error=str(e),
)
self.logger.error(f"✗ {model} | {e}")
return "", log
else:
delay = HOLYSHEEP_CONFIG["retry_delay"] * (2 ** attempt)
self.logger.warning(f"Retry {attempt+1} nach {delay}s: {e}")
await asyncio.sleep(delay)
return "", RequestLog(
timestamp=datetime.now(timezone.utc).isoformat(),
model=model,
prompt_length=len(prompt),
completion_tokens=0,
latency_ms=0,
cost_usd=0.0,
status="MAX_RETRIES",
)
def _update_stats(self, latency_ms: float, cost: float, tokens: int, success: bool):
"""Aktualisiert interne Statistiken"""
self.stats["total_requests"] += 1
self.stats["total_cost_usd"] += cost
self.stats["total_tokens"] += tokens
if success:
self.stats["successful_requests"] += 1
else:
self.stats["failed_requests"] += 1
# Gleitender Durchschnitt für Latenz
n = self.stats["total_requests"]
current_avg = self.stats["avg_latency_ms"]
self.stats["avg_latency_ms"] = ((n - 1) * current_avg + latency_ms) / n
def get_dashboard(self) -> Dict:
"""Generiert Dashboard-Daten für Monitoring-Interface"""
total = self.stats["total_requests"]
success_rate = (
(self.stats["successful_requests"] / total * 100)
if total > 0 else 0
)
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"requests": {
"total": total,
"successful": self.stats["successful_requests"],
"failed": self.stats["failed_requests"],
"success_rate": f"{success_rate:.2f}%",
},
"performance": {
"avg_latency_ms": round(self.stats["avg_latency_ms"], 2),
},
"costs": {
"total_usd": round(self.stats["total_cost_usd"], 6),
"cost_per_1k_tokens": (
round(self.stats["total_cost_usd"] / (self.stats["total_tokens"] / 1000), 6)
if self.stats["total_tokens"] > 0 else 0
),
},
"tokens": {
"total": self.stats["total_tokens"],
}
}
async def close(self):
"""Schließt Ressourcen"""
if self.session and not self.session.closed:
await self.session.close()
============================================
BEISPIEL-NUTZUNG
============================================
async def main():
monitor = HolySheepLiveMonitor()
try:
# Test-Anfragen mit verschiedenen Modellen
test_cases = [
{
"prompt": "Erkläre in 3 Sätzen, was Docker-Container sind",
"model": "gemini-2.5-flash", # Schnellste Option
},
{
"prompt": "Schreibe einen kurzen Python-Decorator für Logging",
"model": "deepseek-v3.2", # Beste Kosten/Effizienz
},
{
"prompt": "Analysiere die Vor- und Nachteile von Microservices-Architektur",
"model": "gpt-4.1", # Höchste Qualität
},
]
print("\n" + "="*60)
print("HOLYSHEEP AI LIVE MONITORING TEST")
print("="*60 + "\n")
for i, test in enumerate(test_cases, 1):
print(f"\n--- Test {i}: {test['model']} ---")
response, log = await monitor.call_with_monitoring(
prompt=test["prompt"],
model=test["model"],
max_tokens=500,
)
print(f"Antwort: {response[:200]}...")
print(f"Log: {asdict(log)}")
# Dashboard-Ausgabe
print("\n" + "="*60)
print("MONITORING DASHBOARD")
print("="*60)
dashboard = monitor.get_dashboard()
print(json.dumps(dashboard, indent=2))
finally:
await monitor.close()
if __name__ == "__main__":
# Setzen Sie Ihren API-Key
# os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
print("Initialisiere HolySheep AI Monitoring...")
print(f"Base URL: {HOLYSHEEP_CONFIG['base_url']}")
print(f"Verfügbare Modelle: {list(MODEL_ALTERNATIVES.keys())}")
print()
asyncio.run(main())
Modell-Routing: Automatische Optimierung basierend auf Anforderungen
Ein fortgeschrittenes Feature meines Monitoring-Systems ist das automatische Modell-Routing. Basierend auf Request-Charakteristiken (Länge, Komplexität, Latenz-Anforderungen) wird automatisch das optimale Modell gewählt.
"""
Intelligentes Modell-Routing mit HolySheep AI
Wählt automatisch das beste Modell basierend auf Anforderungen
"""
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import re
class RequestComplexity(Enum):
"""Komplexitätsstufen für Anfragen"""
TRIVIAL = "trivial" # <50 Zeichen, einfache Fragen
SIMPLE = "simple" # <200 Zeichen, Standardfragen
MODERATE = "moderate" # 200-1000 Zeichen, Erklärungen
COMPLEX = "complex" # >1000 Zeichen, Analysen
ADVANCED = "advanced" # >2000 Zeichen, komplexe推理
class LatencyRequirement(Enum):
"""Latenz-Anforderungen"""
INSTANT = 100 # <100ms
FAST = 500 # <500ms
NORMAL = 2000 # <2s
PATIENT = 10000 # <10s
@dataclass
class ModelConfig:
"""Konfiguration für ein KI-Modell"""
name: str
provider: str
max_tokens: int
avg_latency_ms: float
input_cost_per_1k: float
output_cost_per_1k: float
strengths: list
weaknesses: list
Modell-Registry mit HolySheep AI Modellen
MODEL_REGISTRY = {
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="holysheep",
max_tokens=8192,
avg_latency_ms=45,
input_cost_per_1k=0.25,
output_cost_per_1k=1.00,
strengths=["Geschwindigkeit", "Code", "Fakten"],
weaknesses=["Lange Texte", "Kreativität"],
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="holysheep",
max_tokens=4096,
avg_latency_ms=52,
input_cost_per_1k=0.042,
output_cost_per_1k=0.168,
strengths=["Kosten", "Logik", "Mathematik"],
weaknesses=["Kreativität", "Lange Kontexte"],
),
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="holysheep",
max_tokens=8192,
avg_latency_ms=120,
input_cost_per_1k=0.80,
output_cost_per_1k=3.20,
strengths=["Qualität", "Kreativität", "Komplexität"],
weaknesses=["Kosten", "Latenz"],
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="holysheep",
max_tokens=4096,
avg_latency_ms=95,
input_cost_per_1k=1.50,
output_cost_per_1k=7.50,
strengths=["Analyse", "Schreiben", "Nuancen"],
weaknesses=["Geschwindigkeit", "Kosten"],
),
}
class IntelligentRouter:
"""
Intelligenter Router für KI-Modellauswahl
Berücksichtigt: Kosten, Latenz, Komplexität, Qualitätsanforderungen
"""
def __init__(self, cost_budget_per_request: float = 0.01):
self.cost_budget = cost_budget_per_request
self.usage_stats = {}
def estimate_complexity(self, prompt: str) -> RequestComplexity:
"""Schätzt die Komplexität einer Anfrage"""
word_count = len(prompt.split())
char_count = len(prompt)
# Complexity-Indikatoren
has_code = "```" in prompt or "code" in prompt.lower()
has_math = any(char in prompt for char in ["∑", "∫", "=", "+", "-", "*", "/"])
has_analysis = any(word in prompt.lower() for word in [
"analysiere", "vergleiche", "bewerte", "kritisiere"
])
complexity_score = 0
if char_count < 50:
complexity_score = 1
elif char_count < 200:
complexity_score = 2
elif char_count < 1000:
complexity_score = 3
else:
complexity_score = 4
if has_code or has_math:
complexity_score += 1
if has_analysis:
complexity_score += 1
return RequestComplexity([
RequestComplexity.TRIVIAL,
RequestComplexity.SIMPLE,
RequestComplexity.MODERATE,
RequestComplexity.COMPLEX,
RequestComplexity.ADVANCED,
][min(complexity_score, 4)])
def estimate_latency_requirement(self, prompt: str) -> LatencyRequirement:
"""Erkennt implizite Latenz-Anforderungen"""
prompt_lower = prompt.lower()
if any(word in prompt_lower for word in ["sofort", "instant", "real-time"]):
return LatencyRequirement.INSTANT
elif any(word in prompt_lower for word in ["schnell", "zeitnah"]):
return LatencyRequirement.FAST
elif any(word in prompt_lower for word in ["gründlich", "detailliert"]):
return LatencyRequirement.PATIENT
else:
return LatencyRequirement.NORMAL
def select_model(
self,
prompt: str,
quality_weight: float = 0.5,
speed_weight: float = 0.3,
cost_weight: float = 0.2,
) -> ModelConfig:
"""
Wählt das optimale Modell basierend auf Gewichtungen
Args:
prompt: Die Benutzeranfrage
quality_weight: Gewichtung der Qualität (0-1)
speed_weight: Gewichtung der Geschwindigkeit (0-1)
cost_weight: Gewichtung der Kosten (0-1)
Returns:
Optimal konfiguriertes Modell
"""
complexity = self.estimate_complexity(prompt)
latency_req = self.estimate_latency_requirement(prompt)
scores = {}
for model_id, config in MODEL_REGISTRY.items():
# Qualitäts-Score (basierend auf Komplexität)
quality_score = 0
if complexity in [RequestComplexity.TRIVIAL, RequestComplexity.SIMPLE]:
quality_score = 70 if model_id != "deepseek-v3.2" else 85
elif complexity in [RequestComplexity.MODERATE, RequestComplexity.COMPLEX]:
quality_score = 80 if model_id in ["gpt-4.1", "claude-sonnet-4.5"] else 75
else: # ADVANCED
quality_score = 95 if model_id in ["gpt-4.1", "claude-sonnet-4.5"] else 60
# Speed-Score (basierend auf Latenz-Anforderung)
speed_score = 100 - (config.avg_latency_ms / latency_req.value * 100)
speed_score = max(0, min(100, speed_score))
# Cost-Score (invers proportional)
avg_cost_per_request = (config.input_cost_per_1k + config.output_cost_per_1k) / 2
cost_score = max(0, 100 - (avg_cost_per_request / self.cost_budget * 100))
# Gewichtete Gesamtpunkte
total_score = (
quality_score * quality_weight +
speed_score * speed_weight +
cost_score * cost_weight
)
scores[model_id] = {
"total": total_score,
"quality": quality_score,
"speed": speed_score,
"cost": cost_score,
"recommended": True,
}
# Wähle bestbewertetes Modell
best_model_id = max(scores, key=lambda x: scores[x]["total"])
self.logger.info(
f"Modell-Routing für '{prompt[:50]}...': "
f"{best_model
Verwandte Ressourcen
Verwandte Artikel