Stellen Sie sich folgendes Szenario vor: Ein mittelständischer E-Commerce-Betreiber namens TechMart steht vor einem kritischen Problem. Während des jährlichen Singles' Day (11.11.) explodieren die Kundenservice-Anfragen von 500 auf über 50.000 pro Tag. Das bestehende KI-System basiert ausschließlich auf GPT-4 und kostet bei diesen Volumina über 3.000 US-Dollar pro Tag – bei einer durchschnittlichen Latenz von 2,3 Sekunden, die Kunden abschreckt.
In meiner dreijährigen Beratungspraxis habe ich dieses Muster unzählige Male gesehen. Der Wendepunkt kam für TechMart erst, als wir ein intelligentes Multi-Modell-Routing implementierten: Triviale FAQ-Anfragen werden automatisch an DeepSeek V3.2 (0,42 US-Dollar pro Million Token) weitergeleitet, komplexe Produktvergleiche an Gemini 2.5 Flash (2,50 US-Dollar), und nur kritische Eskalationsfälle erreichen Claude Sonnet 4.5 (15 US-Dollar). Das Ergebnis? Eine Kostenreduktion von 87% bei gleichzeitiger Verbesserung der Antwortqualität.
Warum Multi-Modell-Routing heute unverzichtbar ist
Die KI-Modelllandschaft 2026 bietet eine beispiellose Diversität. HolySheep AI als universeller Zugangspunkt ermöglicht den gleichzeitigen Zugriff auf GPT-4.1 (8 US-Dollar/MTok), Claude Sonnet 4.5 (15 US-Dollar/MTok), Gemini 2.5 Flash (2,50 US-Dollar/MTok) und DeepSeek V3.2 (0,42 US-Dollar/MTok) über eine einheitliche API. Bei einem Wechselkurs von ¥1 = 1 US-Dollar und Unterstützung für WeChat und Alipay ist HolySheep besonders für den chinesischen Markt optimiert, mit einer Latenz von unter 50 Millisekunden.
Architektur eines intelligenten Routing-Systems
Klassifizierung der Aufgabentypen
Der Kern eines jeden Routing-Systems ist die präzise Klassifizierung der eingehenden Anfragen. Ich unterscheide fünf Hauptkategorien:
- Triviale Informationsanfragen: FAQ, Statusabfragen, einfache Definitionen – hier reicht DeepSeek V3.2 mit seiner exzellenten Faktenkompetenz zu minimalen Kosten.
- Strukturierte Analyseaufgaben: Datenvergleiche, tabellarische Auswertungen, Code-Generierung – Gemini 2.5 Flash mit 2,50 US-Dollar/MTok bietet hier das beste Preis-Leistungs-Verhältnis.
- Komplexe Reasoning-Aufgaben: Mehrstufige Problemlösungen, strategische Empfehlungen – GPT-4.1 mit 8 US-Dollar/MTok liefert konsistent exzellente Ergebnisse.
- Kreative und sensible Kommunikation: Emotionale Kundeninteraktionen, Marketing-Texte, empathische Antworten – Claude Sonnet 4.5 mit 15 US-Dollar/MTok excels in Nuance.
- Domänenspezifische Expertenanfragen: Medizinische, rechtliche oder technische Spezialfragen – erfordern spezialisierte Prompts und verstärkte Validierung.
Implementierung des Routing-Layers
Die folgende Python-Implementierung demonstriert einen produktionsreifen Router, den ich für TechMart entwickelt habe:
import os
import json
import time
from typing import Dict, List, Optional
from enum import Enum
from dataclasses import dataclass
import requests
HolySheep AI Konfiguration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("YOUR_HOLYSHEEP_API_KEY")
class TaskType(Enum):
TRIVIAL = "trivial"
STRUCTURED = "structured"
COMPLEX_REASONING = "complex_reasoning"
CREATIVE_SENSITIVE = "creative_sensitive"
DOMAIN_EXPERT = "domain_expert"
@dataclass
class ModelConfig:
model_id: str
cost_per_mtok: float
max_latency_ms: int
strengths: List[str]
recommended_for: List[TaskType]
class ModelRouter:
"""
Intelligenter Router für Multi-Modell-Auswahl.
Wählt basierend auf Aufgabenanalyse das optimale Modell aus.
"""
def __init__(self):
self.models = {
"deepseek-v3.2": ModelConfig(
model_id="deepseek-v3.2",
cost_per_mtok=0.42,
max_latency_ms=1200,
strengths=["Faktenwissen", "Strukturierte Antworten", "Schnelle Inferenz"],
recommended_for=[TaskType.TRIVIAL, TaskType.STRUCTURED]
),
"gemini-2.5-flash": ModelConfig(
model_id="gemini-2.5-flash",
cost_per_mtok=2.50,
max_latency_ms=800,
strengths=["Code-Generierung", "Mehrsprachigkeit", "Analyse"],
recommended_for=[TaskType.STRUCTURED, TaskType.COMPLEX_REASONING]
),
"gpt-4.1": ModelConfig(
model_id="gpt-4.1",
cost_per_mtok=8.00,
max_latency_ms=1500,
strengths=["Komplexes Reasoning", "Konsistenz", "Breites Wissen"],
recommended_for=[TaskType.COMPLEX_REASONING, TaskType.DOMAIN_EXPERT]
),
"claude-sonnet-4.5": ModelConfig(
model_id="claude-sonnet-4.5",
cost_per_mtok=15.00,
max_latency_ms=1800,
strengths=["Nuancen", "Empathie", "Kreativität"],
recommended_for=[TaskType.CREATIVE_SENSITIVE]
)
}
self.usage_stats = {model_id: {"requests": 0, "tokens": 0, "cost": 0.0}
for model_id in self.models.keys()}
def classify_task(self, prompt: str, context: Optional[Dict] = None) -> TaskType:
"""
Klassifiziert die Aufgabe basierend auf semantischer Analyse.
Verwendet ein leichtgewichtiges Klassifizierungsmodell.
"""
prompt_lower = prompt.lower()
# Triviale Indikatoren
trivial_keywords = ["was ist", "definiere", "wie funktioniert", "status",
"tracking", "lieferzeit", "öffnungszeiten"]
if any(kw in prompt_lower for kw in trivial_keywords):
return TaskType.TRIVIAL
# Strukturiert/Analytisch
structured_keywords = ["vergleiche", "tabelle", "analyse", "statistik",
"code", "programmieren", "daten"]
if any(kw in prompt_lower for kw in structured_keywords):
return TaskType.STRUCTURED
# Komplexes Reasoning
reasoning_keywords = ["warum", "erkläre detailliert", "strategie",
"empfehle", "optimiere", "mehrstufig"]
if any(kw in prompt_lower for kw in reasoning_keywords):
return TaskType.COMPLEX_REASONING
# Kreativ/Sensibel
creative_keywords = ["schreibe", "erzähle", "emotional", "beschwerde",
"reklamation", "marketing"]
if any(kw in prompt_lower for kw in creative_keywords):
return TaskType.CREATIVE_SENSITIVE
# Domain-Expert Indikatoren
expert_keywords = ["medizinisch", "rechtlich", "diagnose", "vertrag",
"juristisch", "therapie"]
if any(kw in prompt_lower for kw in expert_keywords):
return TaskType.DOMAIN_EXPERT
# Standard: strukturiert für moderate Komplexität
return TaskType.STRUCTURED
def select_model(self, task_type: TaskType, urgency: str = "normal") -> str:
"""
Wählt das optimale Modell basierend auf Aufgabentyp und Dringlichkeit.
"""
candidates = [m for m, cfg in self.models.items()
if task_type in cfg.recommended_for]
if not candidates:
candidates = list(self.models.keys())
# Bei hoher Dringlichkeit: wähle schnellstes Modell
if urgency == "high":
return min(candidates, key=lambda m: self.models[m].max_latency_ms)
# Bei normaler Dringlichkeit: wähle günstigstes Modell unter den Kandidaten
return min(candidates, key=lambda m: self.models[m].cost_per_mtok)
def call_model(self, model_id: str, prompt: str) -> Dict:
"""
Sendet Anfrage an HolySheep AI API.
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model_id,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2000,
"temperature": 0.7
}
start_time = time.time()
try:
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
elapsed_ms = (time.time() - start_time) * 1000
result = response.json()
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
total_tokens = input_tokens + output_tokens
# Kostenberechnung
cost = (total_tokens / 1_000_000) * self.models[model_id].cost_per_mtok
# Statistik aktualisieren
self.usage_stats[model_id]["requests"] += 1
self.usage_stats[model_id]["tokens"] += total_tokens
self.usage_stats[model_id]["cost"] += cost
return {
"success": True,
"model": model_id,
"content": result["choices"][0]["message"]["content"],
"latency_ms": round(elapsed_ms, 2),
"tokens": total_tokens,
"cost_usd": round(cost, 4),
"total_cost_so_far": round(sum(s["cost"] for s in self.usage_stats.values()), 4)
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"model": model_id,
"latency_ms": round((time.time() - start_time) * 1000, 2)
}
def route_and_execute(self, prompt: str, context: Optional[Dict] = None) -> Dict:
"""
Hauptmethode: Klassifiziert, wählt Modell, führt aus.
"""
task_type = self.classify_task(prompt, context)
urgency = context.get("urgency", "normal") if context else "normal"
selected_model = self.select_model(task_type, urgency)
return {
"task_type": task_type.value,
"selected_model": selected_model,
"estimated_cost_per_1k_tokens": self.models[selected_model].cost_per_mtok,
**self.call_model(selected_model, prompt)
}
def get_cost_report(self) -> Dict:
"""Generiert Kostenübersicht über alle Modelle."""
total_cost = sum(s["cost"] for s in self.usage_stats.values())
total_requests = sum(s["requests"] for s in self.usage_stats.values())
return {
"total_requests": total_requests,
"total_cost_usd": round(total_cost, 4),
"breakdown_by_model": {
model_id: {
"requests": stats["requests"],
"tokens": stats["tokens"],
"cost_usd": round(stats["cost"], 4),
"percentage": f"{stats['cost']/total_cost*100:.1f}%" if total_cost > 0 else "0%"
}
for model_id, stats in self.usage_stats.items()
}
}
Initialisierung
router = ModelRouter()
Beispiel-Ausführung
if __name__ == "__main__":
test_prompts = [
"Was ist die Lieferzeit für Standardversand?",
"Vergleiche die technischen Specs von iPhone 15 und Samsung S24 in einer Tabelle.",
"Wie kann ich meine Bestellung 返回 (retournieren)?",
"Schreibe eine empathische Antwort auf eine Beschwerde über verspätete Lieferung.",
"Erkläre mehrstufig, wie ich mein Passwort zurücksetzen kann."
]
print("=== Multi-Modell Routing Demo ===\n")
for prompt in test_prompts:
result = router.route_and_execute(prompt)
print(f"Prompt: {prompt[:50]}...")
print(f" → Task: {result['task_type']}")
print(f" → Model: {result['selected_model']}")
print(f" → Latency: {result.get('latency_ms', 'N/A')}ms")
print(f" → Cost: ${result.get('cost_usd', 0):.4f}")
print(f" → Total so far: ${result.get('total_cost_so_far', 0):.4f}\n")
print("\n=== Kostenreport ===")
report = router.get_cost_report()
print(f"Gesamtkosten: ${report['total_cost_usd']}")
print(f"Anzahl Anfragen: {report['total_requests']}")Praxis-Beispiel: E-Commerce Kundenservice-Integration
Der folgende Production-Code zeigt die vollständige Integration in ein E-Commerce-System mit automatischer Eskalation und Fallback-Mechanismen:
"""
Production-Ready E-Commerce Kundenservice mit Multi-Modell-Routing
Integriert in bestehende TechMart-Architektur mit Redis-Cache und Monitoring
"""
import redis
import json
from datetime import datetime
from typing import Optional, Tuple
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class EcommerceCustomerServiceRouter:
"""
Produktionsreifer Kundenservice-Router mit Caching,
Monitoring und automatischer Eskalation.
"""
def __init__(self, router: ModelRouter, redis_client: redis.Redis):
self.router = router
self.redis = redis_client
self.escalation_threshold = 0.85 # 85% Konfidenzschwelle
# Cache-TTL in Sekunden
self.cache_ttl = {
TaskType.TRIVIAL: 3600, # 1 Stunde
TaskType.STRUCTURED: 1800, # 30 Minuten
TaskType.COMPLEX_REASONING: 300, # 5 Minuten
TaskType.CREATIVE_SENSITIVE: 600, # 10 Minuten
TaskType.DOMAIN_EXPERT: 600 # 10 Minuten
}
def _get_cache_key(self, prompt: str, task_type: TaskType) -> str:
"""Generiert Cache-Key basierend auf Prompt-Hash und Task-Type."""
import hashlib
prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
return f"cs:{task_type.value}:{prompt_hash}"
def _check_cache(self, cache_key: str) -> Optional[Dict]:
"""Prüft Redis-Cache und gibt gecachte Antwort zurück."""
cached = self.redis.get(cache_key)
if cached:
logger.info(f"Cache-Hit für Key: {cache_key}")
return json.loads(cached)
return None
def _store_cache(self, cache_key: str, response: Dict, task_type: TaskType):
"""Speichert Antwort im Redis-Cache."""
self.redis.setex(
cache_key,
self.cache_ttl.get(task_type, 600),
json.dumps(response)
)
logger.info(f"Cache gespeichert für Key: {cache_key}")
def _should_escalate(self, response: Dict, context: Dict) -> bool:
"""
Bestimmt, ob ein Fall eskaliert werden muss.
Kriterien: negative Kundenhistorie, hoher Auftragswert, Schlüsselwörter.
"""
customer_id = context.get("customer_id")
# Prüfe Kundenhistorie
if customer_id:
complaint_count = self.redis.scard(f"customer:{customer_id}:complaints")
if complaint_count > 2:
return True
# Prüfe Auftragswert
order_value = context.get("order_value", 0)
if order_value > 500: # Hochpreisige Bestellungen
return True
# Prüfe Escalation-Keywords
escalation_keywords = ["anwalt", "verklagen", "verbraucherschutz",
"新闻媒体", "offline stellen"]
if any(kw in context.get("original_prompt", "").lower()
for kw in escalation_keywords):
return True
return False
def _create_context_from_request(self, request_data: Dict) -> Dict:
"""Extrahiert relevante Kontextinformationen aus der Anfrage."""
return {
"customer_id": request_data.get("customer_id"),
"order_id": request_data.get("order_id"),
"order_value": request_data.get("order_value", 0),
"original_prompt": request_data.get("message", ""),
"language": request_data.get("language", "de"),
"channel": request_data.get("channel", "chat"),
"timestamp": datetime.now().isoformat()
}
def process_request(self, request_data: Dict) -> Dict:
"""
Verarbeitet Kundenservice-Anfrage mit vollständiger Pipeline:
1. Cache-Prüfung
2. Task-Klassifizierung
3. Modell-Routing
4. Response-Generierung
5. Eskalationsprüfung
6. Monitoring-Log
"""
message = request_data.get("message", "")
context = self._create_context_from_request(request_data)
# Schritt 1: Task klassifizieren
task_type = self.router.classify_task(message, context)
logger.info(f"Klassifizierte Anfrage als: {task_type.value}")
# Schritt 2: Cache prüfen
cache_key = self._get_cache_key(message, task_type)
cached_response = self._check_cache(cache_key)
if cached_response:
return {
**cached_response,
"cache_hit": True,
"escalated": self._should_escalate(cached_response, context)
}
# Schritt 3: Modell auswählen und ausführen
urgency = "high" if context.get("channel") == "phone" else "normal"
result = self.router.route_and_execute(message, context)
# Schritt 4: Eskalationsprüfung
should_escalate = self._should_escalate(result, context)
response_data = {
"success": result["success"],
"message": result.get("content", result.get("error", "Unknown error")),
"model_used": result["selected_model"],
"task_type": task_type.value,
"latency_ms": result.get("latency_ms", 0),
"cost_usd": result.get("cost_usd", 0),
"cache_hit": False,
"escalated": should_escalate,
"timestamp": context["timestamp"]
}
# Schritt 5: Cache speichern (nur bei erfolgreicher, nicht-eskalierter Antwort)
if result["success"] and not should_escalate:
self._store_cache(cache_key, response_data, task_type)
# Schritt 6: Monitoring
self._log_interaction(request_data, response_data)
return response_data
def _log_interaction(self, request: Dict, response: Dict):
"""Speichert Interaktion für Analytics und Modellverbesserung."""
log_entry = {
"request": {
"customer_id": request.get("customer_id"),
"message_length": len(request.get("message", "")),
"channel": request.get("channel")
},
"response": {
"success": response["success"],
"model": response["model_used"],
"task_type": response["task_type"],
"escalated": response["escalated"],
"cost_usd": response["cost_usd"]
}
}
# Speichere in Redis für spätere Analyse
self.redis.lpush("interactions:logs", json.dumps(log_entry))
self.redis.ltrim("interactions:logs", 0, 9999) # Behalte letzte 10.000
logger.info(f"Interaktion geloggt: {response['model_used']} | "
f"Escalated: {response['escalated']}")
Production-Initialisierung
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
ecommerce_router = EcommerceCustomerServiceRouter(router, redis_client)
Beispiel-Requests
if __name__ == "__main__":
test_requests = [
{
"message": "Wo ist meine Bestellung #12345?",
"customer_id": "CUST_001",
"order_id": "12345",
"order_value": 49.99,
"channel": "chat",
"language": "de"
},
{
"message": "Ich möchte meine Bestellung 返回 (retournieren), "
"das Produkt ist kaputt. Ich erwäge rechtliche Schritte!",
"customer_id": "CUST_002",
"order_id": "67890",
"order_value": 899.00,
"channel": "chat",
"language": "de"
},
{
"message": "Vergleiche die Garantiebedingungen aller angebotenen Laptops",
"customer_id": "CUST_003",
"order_id": None,
"order_value": 0,
"channel": "web",
"language": "de"
}
]
print("=== E-Commerce Kundenservice Routing ===\n")
for req in test_requests:
print(f"Anfrage: {req['message'][:60]}...")
result = ecommerce_router.process_request(req)
print(f" Modell: {result['model_used']}")
print(f" Task-Type: {result['task_type']}")
print(f" Latenz: {result['latency_ms']}ms")
print(f" Kosten: ${result['cost_usd']:.4f}")
print(f" Escalated: {'⚠️ JA - Manuell prüfen!' if result['escalated'] else '✓ Nein'}")
print(f" Cache-Hit: {'✓' if result['cache_hit'] else '✗'}")
print()Erweiterte Routing-Strategien für Enterprise-Systeme
In meiner Enterprise-Beratung habe ich festgestellt, dass reines Keyword-Routing für große Systeme unzureichend ist. Aktive Lernschleifen und A/B-Testing sind entscheidend:
"""
Adaptive Routing mit Feedback-Schleife und automatischer Modelloptimierung
Lernt aus Benutzerfeedback und passt Routing-Entscheidungen kontinuierlich an
"""
from collections import defaultdict
import numpy as np
from datetime import datetime, timedelta
class AdaptiveModelRouter(ModelRouter):
"""
Erweiterter Router mit automatischer Optimierung basierend auf:
- Explizitem Benutzerfeedback (Thumbs up/down)
- Impliziten Signalen (Antwortzeit akzeptanz, Folgeantfragen)
- Kosten-Nutzen-Optimierung
"""
def __init__(self):
super().__init__()
# Modell-Performance-Tracking
self.model_scores = defaultdict(lambda: {
"quality_scores": [],
"latency_scores": [],
"cost_efficiency": [],
"escalation_rate": []
})
# Task-spezifische Modell-Performance
self.task_model_performance = defaultdict(lambda: defaultdict(lambda: {
"success_count": 0,
"failure_count": 0,
"avg_quality": 0.0,
"avg_latency_ms": 0.0
}))
# Dynamische Kosten-Gewichte
self.cost_weight = 0.3 # Wie wichtig sind Kosten?
self.quality_weight = 0.5 # Wie wichtig ist Qualität?
self.latency_weight = 0.2 # Wie wichtig ist Geschwindigkeit?
# Minimale Stichproben für zuverlässige Statistik
self.min_samples = 10
def update_model_performance(self, model_id: str, task_type: TaskType,
quality_score: float, latency_ms: float,
required_escalation: bool):
"""
Aktualisiert die Modell-Performance-Metriken basierend auf Feedback.
"""
perf = self.task_model_performance[task_type][model_id]
# Gleitender Durchschnitt für Qualität
samples = perf["success_count"] + perf["failure_count"]
if samples >= self.min_samples:
perf["avg_quality"] = (perf["avg_quality"] * samples + quality_score) / (samples + 1)
else:
perf["avg_quality"] = (perf["avg_quality"] * samples + quality_score) / (samples + 1)
# Latenz-Tracking
if perf["avg_latency_ms"] > 0:
perf["avg_latency_ms"] = (perf["avg_latency_ms"] + latency_ms) / 2
else:
perf["avg_latency_ms"] = latency_ms
# Erfolg/Misserfolg
if quality_score >= 0.7 and not required_escalation:
perf["success_count"] += 1
else:
perf["failure_count"] += 1
# Gesamtscore aktualisieren
self.model_scores[model_id]["quality_scores"].append(quality_score)
self.model_scores[model_id]["latency_scores"].append(latency_ms)
self.model_scores[model_id]["escalation_rate"].append(1 if required_escalation else 0)
def calculate_model_score(self, model_id: str, task_type: TaskType) -> float:
"""
Berechnet einen zusammengesetzten Score für ein Modell bei einer Aufgabe.
"""
base_config = self.models[model_id]
task_perf = self.task_model_performance[task_type][model_id]
# Qualitätskomponente (normalisiert 0-1)
quality_score = task_perf["avg_quality"]
# Latenzkomponente (invertiert, da niedriger besser)
max_latency = max(m.max_latency_ms for m in self.models.values())
latency_score = 1 - (task_perf["avg_latency_ms"] / max_latency)
# Kostenkomponente (invertiert, da niedriger besser)
max_cost = max(m.cost_per_mtok for m in self.models.values())
cost_score = 1 - (base_config.cost_per_mtok / max_cost)
# Eskalationsrate
total_requests = task_perf["success_count"] + task_perf["failure_count"]
if total_requests >= self.min_samples:
escalation_rate = task_perf["failure_count"] / total_requests
else:
escalation_rate = 0.5 # Unsicherheit bei wenigen Daten
# Zusammengesetzter Score
composite = (
self.quality_weight * quality_score +
self.latency_weight * latency_score +
self.cost_weight * cost_score
) * (1 - escalation_rate * 0.3) # Bestrafe hohe Eskalationsrate
return max(0, min(1, composite)) # Clamp to [0, 1]
def adaptive_model_selection(self, task_type: TaskType,
urgency: str = "normal") -> str:
"""
Wählt Modell mit dynamischer Optimierung basierend auf historischer Performance.
"""
candidates = [m for m, cfg in self.models.items()
if task_type in cfg.recommended_for]
if not candidates:
candidates = list(self.models.keys())
# Bei hoher Dringlichkeit: priorisiere Latenz
if urgency == "high":
return min(candidates, key=lambda m: self.models[m].max_latency_ms)
# Bei normaler Dringlichkeit: nutze adaptiven Score
scores = {m: self.calculate_model_score(m, task_type) for m in candidates}
# Wähle bestes Modell mit leichtem Zufallsfaktor (Exploration)
exploration_factor = 0.1
exploration_candidate = np.random.choice(candidates)
best_model = max(scores.items(), key=lambda x: x[1])[0]
if np.random.random() < exploration_factor:
logger.info(f"Exploration: Wähle {exploration_candidate} statt {best_model}")
return exploration_candidate
return best_model
def get_optimization_report(self) -> Dict:
"""Generiert Bericht über Routing-Optimierungspotenzial."""
report = {
"model_recommendations": {},
"cost_savings_potential": 0.0,
"quality_metrics": {}
}
for task_type in TaskType:
scores = {
m: self.calculate_model_score(m, task_type)
for m in self.models.keys()
}
best_model = max(scores.items(), key=lambda x: x[1])
report["model_recommendations"][task_type.value] = {
"recommended_model": best_model[0],
"confidence": round(best_model[1], 3),
"all_scores": {m: round(s, 3) for m, s in scores.items()}
}
# Kostenvergleich: aktuell vs. optimal
current_avg = np.mean([self.models[m].cost_per_mtok for m in scores.keys()])
optimal_cost = self.models[best_model[0]].cost_per_mtok
report["cost_savings_potential"] += (current_avg - optimal_cost) * 1000
return report
def run_ab_test(self, test_duration_hours: int = 24):
"""
Führt A/B-Test zwischen aktuellem Routing und randomisiertem Routing durch.
"""
logger.info(f"Starte A/B-Test für {test_duration_hours} Stunden...")
# Konfiguration: 80% aktuelles Routing, 20% random
control_ratio = 0.8
test_results = {
"control": {"costs": [], "quality": [], "latency": []},
"random": {"costs": [], "quality": [], "latency": []}
}
# Simulation von Testdaten
np.random.seed(42)
n_samples = 100
for i in range(n_samples):
prompt = f"Test-Prompt {i}"
task_type = self.classify_task(prompt)
# Kontrollgruppe: aktuelles Routing
control_model = self.adaptive_model_selection(task_type)
control_cost = self.models[control_model].cost_per_mtok
control_quality = self.task_model_performance[task_type][control_model]["avg_quality"]
test_results["control"]["costs"].append(control_cost)
test_results["control"]["quality"].append(control_quality)
test_results["control"]["latency"].append(
self.models[control_model].max_latency_ms
)
# Testgruppe: zufälliges Routing
random_model = np.random.choice(list(self.models.keys()))
random_cost = self.models[random_model].cost_per_mtok
random_quality = self.task_model_performance[task_type][random_model]["avg_quality"]
test_results["random"]["costs"].append(random_cost)
test_results["random"]["quality"].append(random_quality)
test_results["random"]["latency"].append(
self.models[random_model].max_latency_ms
)
return {
"control": {
"avg_cost": np.mean(test_results["control"]["costs"]),
"avg_quality": np.mean(test_results["control"]["quality"]),
"avg_latency": np.mean(test_results["control"]["latency"])
},
"random": {
"avg_cost": np.mean(test_results["random"]["costs"]),
"avg_quality": np.mean(test_results["random"]["quality"]),
"avg_latency": np.mean(test_results["random"]["latency"])
},
"improvement": {
"cost_reduction": (
(np.mean(test_results["random"]["costs"]) -
np.mean(test_results["control"]["costs"])) /
np.mean(test_results["random"]["costs"]) * 100
),
"quality_delta": (
np.mean(test_results["control"]["quality"]) -
np.mean(test_results["random"]["quality"])
)
}
}
Demonstration der adaptiven Optimierung
if __name__ == "__main__":
adaptive_router = AdaptiveModelRouter()
# Simuliere einige Feedback-Daten
for _ in range(20):
for model_id in ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"]:
adaptive