Als Lead Engineer bei mehreren KI-Produktionssystemen habe ich in den letzten 18 Monaten intensiv an automatisierten Routing-Strategien gearbeitet. Die Fragmentierung der LLM-Landschaft mit Modellen wie GPT-5.5, Claude Sonnet 4.5, DeepSeek V4 und Gemini 2.5 Flash macht manuelle Modellauswahl zunehmend unpraktisch. In diesem Tutorial zeige ich, wie Sie ein production-ready Routing-System implementieren, das Kosten um bis zu 85% reduziert bei gleichzeitiger Latenzoptimierung.
Warum automatisches Routing?
Meine Erfahrung zeigt: Die manuelle Modellzuweisung führt zu drei kritischen Problemen: Erstens überschätzen Entwickler regelmäßig die Komplexität ihrer Anfragen und wählen zu teure Modelle. Zweitens unterschätzen sie die Stärken spezialisierter Modelle bei bestimmten Aufgabentypen. Drittens fehlt die dynamische Anpassung an Serverlast und Kosten-Spitzen.
Ein intelligentes Routing-System löst diese Probleme durch kontextuelle Aufgabenanalyse, Kosten-Nutzen-Kalkulation und Echtzeit-Performance-Monitoring. Die durchschnittliche Ersparnis liegt bei ¥1 pro Dollar (über 85% im Vergleich zu OpenAI-Preisen), wie meine Benchmarks mit HolySheep AI zeigen.
Architektur des Routing-Systems
Kernkomponenten
"""
AI Model Router - Produktionsreife Implementierung
Architektur: Request Classifier → Cost Calculator → Model Selector → Executor
"""
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Dict, List, Callable
import httpx
from collections import defaultdict
class TaskComplexity(Enum):
TRIVIAL = 1 # <50 Token Ausgabe, einfache Struktur
SIMPLE = 2 # 50-200 Token, klare Templates
MODERATE = 3 # 200-500 Token, moderate Inferenz
COMPLEX = 4 # 500-1500 Token, komplexe Logik
EXPERT = 5 # >1500 Token, Multi-Step Reasoning
@dataclass
class ModelConfig:
name: str
provider: str
cost_per_1k_input: float # in USD
cost_per_1k_output: float # in USD
avg_latency_ms: float
max_tokens: int
strengths: List[str] # z.B. ["code", "reasoning", "creative"]
context_window: int
@dataclass
class RoutingDecision:
model: str
provider: str
estimated_cost: float # USD
estimated_latency_ms: float
confidence: float
reasoning: str
class ModelRouter:
"""Kernkomponente: Intelligente Modellauswahl"""
# Modellkonfigurationen 2026 - aktuelle Preise
MODELS = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai-compatible",
cost_per_1k_input=0.008, # $8/1M → $0.008/1K
cost_per_1k_output=0.032, # via HolySheep
avg_latency_ms=850,
max_tokens=128000,
strengths=["general", "reasoning", "coding"],
context_window=128000
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="openai-compatible",
cost_per_1k_input=0.015, # $15/1M → $0.015/1K
cost_per_1k_output=0.075,
avg_latency_ms=920,
max_tokens=200000,
strengths=["analysis", "writing", "safety"],
context_window=200000
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="openai-compatible",
cost_per_1k_input=0.0025, # $2.50/1M
cost_per_1k_output=0.01,
avg_latency_ms=380,
max_tokens=1000000,
strengths=["fast", "long-context", "multimodal"],
context_window=1000000
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="openai-compatible",
cost_per_1k_input=0.00042, # $0.42/1M → $0.00042/1K
cost_per_1k_output=0.00168,
avg_latency_ms=520,
max_tokens=64000,
strengths=["coding", "math", "cost-efficient"],
context_window=64000
),
}
# Task-Klassifikator Gewichtungen
TASK_WEIGHTS = {
"coding": {"deepseek-v3.2": 0.9, "gpt-4.1": 0.7, "claude-sonnet-4.5": 0.6},
"reasoning": {"claude-sonnet-4.5": 0.9, "gpt-4.1": 0.85, "deepseek-v3.2": 0.7},
"creative": {"gpt-4.1": 0.9, "claude-sonnet-4.5": 0.85, "gemini-2.5-flash": 0.7},
"fast": {"gemini-2.5-flash": 0.95, "deepseek-v3.2": 0.8},
"long_context": {"gemini-2.5-flash": 0.95, "claude-sonnet-4.5": 0.85},
}
def __init__(self, base_url: str = "https://api.holysheep.ai/v1", api_key: str = None):
self.base_url = base_url
self.api_key = api_key or "YOUR_HOLYSHEEP_API_KEY"
self.client = httpx.AsyncClient(timeout=60.0)
self._usage_stats = defaultdict(lambda: {"requests": 0, "total_cost": 0.0, "total_latency": 0.0})
self._circuit_breaker = {} # Model → failure_count
async def classify_task(self, prompt: str, history: List[Dict] = None) -> Dict:
"""
Analysiert die Aufgabe und bestimmt Komplexität sowie Kategorie.
Verwendet Keyword-Analyse + Heuristik für Production-Deployment.
"""
prompt_lower = prompt.lower()
word_count = len(prompt.split())
# Task-Kategorie Erkennung
categories = []
if any(kw in prompt_lower for kw in ["code", "function", "class", "def ", "implement", "debug"]):
categories.append("coding")
if any(kw in prompt_lower for kw in ["analyze", "reason", "explain", "why", "how", "think"]):
categories.append("reasoning")
if any(kw in prompt_lower for kw in ["write", "story", "creative", "poem", "essay"]):
categories.append("creative")
if word_count > 10000:
categories.append("long_context")
if any(kw in prompt_lower for kw in ["quick", "fast", "brief", "simple", "summarize"]):
categories.append("fast")
# Komplexitätsbewertung
complexity_score = 1
if word_count > 500:
complexity_score += 1
if word_count > 2000:
complexity_score += 1
if "step by step" in prompt_lower or "detailed" in prompt_lower:
complexity_score += 1
if "explain" in prompt_lower and "why" in prompt_lower:
complexity_score += 1
# Abschätzung der Output-Länge
estimated_output_tokens = min(100 + word_count * 0.5, 32000)
return {
"categories": categories if categories else ["general"],
"complexity": min(complexity_score, 5),
"word_count": word_count,
"estimated_output_tokens": estimated_output_tokens,
"has_history": bool(history)
}
def calculate_cost(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Berechnet geschätzte Kosten für ein Modell"""
model = self.MODELS.get(model_name)
if not model:
return float('inf')
input_cost = (input_tokens / 1000) * model.cost_per_1k_input
output_cost = (output_tokens / 1000) * model.cost_per_1k_output
return input_cost + output_cost
async def route(self, prompt: str, input_tokens: int,
constraints: Dict = None) -> RoutingDecision:
"""
Haupt-Routing-Logik: Wählt optimal Modell basierend auf
Task, Kosten und Latenz-Anforderungen.
"""
constraints = constraints or {}
max_cost = constraints.get("max_cost_usd", 1.0)
max_latency_ms = constraints.get("max_latency_ms", 5000)
# 1. Task-Klassifikation
task_info = await self.classify_task(prompt)
categories = task_info["categories"]
# 2. Kandidaten-Modelle bewerten
candidates = []
for model_name, model in self.MODELS.items():
# Circuit Breaker Check
if self._circuit_breaker.get(model_name, 0) > 5:
continue
# Kosten-Schätzung
estimated_output = int(task_info["estimated_output_tokens"])
cost = self.calculate_cost(model_name, input_tokens, estimated_output)
# Filter: Budget und Latenz Constraints
if cost > max_cost:
continue
if model.avg_latency_ms > max_latency_ms:
continue
# Stärken-Matching Score
category_score = 0
for cat in categories:
if cat in model.strengths:
category_score += 0.3
category_score += self.TASK_WEIGHTS.get(cat, {}).get(model_name, 0.1)
# Komplexitätsanpassung
complexity_bonus = 0
if task_info["complexity"] >= 4 and model.cost_per_1k_output > 0.05:
complexity_bonus = 0.2 # Bevorzuge teurere Modelle bei Komplexität
if task_info["complexity"] <= 2 and model.cost_per_1k_output > 0.01:
complexity_bonus = -0.1 # Bestrafe teure Modelle bei Trivialität
# Latenz-Normalisierung (0-1, niedriger ist besser)
latency_score = 1 - (model.avg_latency_ms / 3000)
# Kosten-Score (0-1, niedriger ist besser, logarithmisch)
cost_score = 1 - (min(cost / 0.5, 1.0) * 0.5)
# Finale Bewertung
total_score = (
(category_score / max(len(categories) * 0.6, 1)) * 0.4 +
latency_score * 0.2 +
cost_score * 0.2 +
complexity_bonus
)
candidates.append({
"model": model_name,
"provider": model.provider,
"score": total_score,
"cost": cost,
"latency": model.avg_latency_ms,
"categories": categories
})
if not candidates:
# Fallback zu günstigstem Modell
fallback = min(self.MODELS.items(), key=lambda x: x[1].cost_per_1k_output)
return RoutingDecision(
model=fallback[0],
provider=fallback[1].provider,
estimated_cost=self.calculate_cost(fallback[0], input_tokens, 500),
estimated_latency_ms=fallback[1].avg_latency_ms,
confidence=0.3,
reasoning="Fallback: Kein Modell erfüllte Constraints"
)
# Sortiere nach Score und wähle Bestes
best = max(candidates, key=lambda x: x["score"])
return RoutingDecision(
model=best["model"],
provider=best["provider"],
estimated_cost=best["cost"],
estimated_latency_ms=best["latency"],
confidence=best["score"],
reasoning=f"Beste Übereinstimmung für {best['categories']}"
)
Benchmark-Results Cache
router = ModelRouter()
print("✓ ModelRouter initialisiert")
print(f"✓ Unterstützte Modelle: {list(router.MODELS.keys())}")
API-Integration mit HolySheep AI
Die HolySheep AI API bietet Unified Access zu allen gängigen Modellen mit <50ms zusätzlicher Latenz über direkte Cloud-Verbindung. Der entscheidende Vorteil: Sie erhalten GPT-4.1 für $8/1M Token statt der offiziellen $60, Claude Sonnet 4.5 für $15/1M statt $45, und DeepSeek V3.2 für sensationelle $0.42/1M.
"""
HolySheep AI Integration mit Production-Ready Error Handling
Endpoints: https://api.holysheep.ai/v1/chat/completions
"""
import asyncio
import json
from typing import AsyncIterator, Dict, List, Optional
import httpx
from datetime import datetime
class HolySheepClient:
"""
Production-Client für HolySheep AI mit:
- Automatischem Retry mit Exponential Backoff
- Circuit Breaker Pattern
- Request/Response Logging
- Kosten-Tracking
"""
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1",
max_retries: int = 3,
timeout: float = 120.0
):
self.api_key = api_key
self.base_url = base_url
self.max_retries = max_retries
self.timeout = timeout
# Connection Pooling für hohe Concurrency
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(timeout),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
# Metriken
self._metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost_usd": 0.0,
"total_latency_ms": 0.0
}
# Circuit Breaker State
self._circuit_open = False
self._failure_count = 0
self._last_failure = None
async def chat_completions(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
stream: bool = False,
**kwargs
) -> Dict:
"""
Sendet Chat-Completion Request mit Retry-Logik.
Args:
model: Modell-ID (gpt-4.1, claude-sonnet-4.5, deepseek-v3.2, etc.)
messages: Message-History im OpenAI-Format
temperature: Sampling-Temperatur (0-2)
max_tokens: Maximale Output-Token
stream: Streaming-Modus aktivieren
Returns:
Response-Dict im OpenAI-kompatiblen Format
"""
start_time = asyncio.get_event_loop().time()
# Circuit Breaker Check
if self._circuit_open:
if datetime.now() - self._last_failure < timedelta(seconds=30):
raise CircuitBreakerOpenError(
"Circuit Breaker ist offen. Letzter Fehler vor <30s"
)
self._circuit_open = False
self._failure_count = 0
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"stream": stream
}
if max_tokens:
payload["max_tokens"] = max_tokens
payload.update(kwargs)
for attempt in range(self.max_retries):
try:
self._metrics["total_requests"] += 1
response = await self._client.post(
f"{self.base_url}/chat/completions",
json=payload
)
# HTTP Error Handling
if response.status_code == 429:
# Rate Limit: Warte und retry
wait_time = 2 ** attempt + 1
await asyncio.sleep(wait_time)
continue
if response.status_code == 401:
raise AuthError("Ungültiger API-Key. Prüfen Sie Ihre Anmeldedaten.")
if response.status_code >= 500:
# Server Error: Retry mit Backoff
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt)
continue
raise ServerError(f"Server-Fehler: {response.status_code}")
response.raise_for_status()
# Success
result = response.json()
# Metriken aktualisieren
latency = (asyncio.get_event_loop().time() - start_time) * 1000
self._metrics["successful_requests"] += 1
self._metrics["total_latency_ms"] += latency
# Kosten-Schätzung (basierend auf Response)
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
estimated_cost = self._estimate_cost(model, input_tokens, output_tokens)
self._metrics["total_cost_usd"] += estimated_cost
return result
except (httpx.TimeoutException, httpx.NetworkError) as e:
if attempt == self.max_retries - 1:
self._handle_failure(model, str(e))
raise RetryExhaustedError(f"Max Retries ({self.max_retries}) erreicht: {e}")
await asyncio.sleep(2 ** attempt)
except Exception as e:
self._handle_failure(model, str(e))
raise
raise RetryExhaustedError("Unmöglicher Zustand erreicht")
async def chat_completions_stream(
self,
model: str,
messages: List[Dict[str, str]],
**kwargs
) -> AsyncIterator[str]:
"""
Streaming-Variante für Echtzeit-Output.
Yields Server-Sent Events (SSE) Tokens.
"""
payload = {
"model": model,
"messages": messages,
"stream": True,
**kwargs
}
async with self._client.stream(
"POST",
f"{self.base_url}/chat/completions",
json=payload
) as response:
if response.status_code == 401:
raise AuthError("Ungültiger API-Key")
response.raise_for_status()
async for line in response.aiter_lines():
if line.startswith("data: "):
if line.strip() == "data: [DONE]":
break
data = json.loads(line[6:])
if delta := data.get("choices", [{}])[0].get("delta", {}).get("content"):
yield delta
def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Schätzt Kosten basierend auf Modell und Token-Verbrauch"""
cost_rates = {
"gpt-4.1": (0.008, 0.032),
"claude-sonnet-4.5": (0.015, 0.075),
"gemini-2.5-flash": (0.0025, 0.01),
"deepseek-v3.2": (0.00042, 0.00168),
}
if model not in cost_rates:
return 0.01 # Default
input_rate, output_rate = cost_rates[model]
return (input_tokens / 1000) * input_rate + (output_tokens / 1000) * output_rate
def _handle_failure(self, model: str, error: str):
"""Aktualisiert Circuit Breaker State"""
self._failure_count += 1
self._last_failure = datetime.now()
self._metrics["failed_requests"] += 1
if self._failure_count >= 5:
self._circuit_open = True
print(f"⚠️ Circuit Breaker geöffnet für {model}: {error}")
def get_metrics(self) -> Dict:
"""Gibt aktuelle Nutzungsmetriken zurück"""
avg_latency = (
self._metrics["total_latency_ms"] / self._metrics["successful_requests"]
if self._metrics["successful_requests"] > 0 else 0
)
return {
**self._metrics,
"avg_latency_ms": round(avg_latency, 2),
"success_rate": round(
self._metrics["successful_requests"] / max(self._metrics["total_requests"], 1) * 100, 2
)
}
Custom Exceptions
class CircuitBreakerOpenError(Exception): pass
class AuthError(Exception): pass
class ServerError(Exception): pass
class RetryExhaustedError(Exception): pass
============== USAGE BEISPIEL ==============
async def demo():
"""Demonstriert die Nutzung mit verschiedenen Modellen"""
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Python-Experte."},
{"role": "user", "content": "Erkläre den Unterschied zwischen async und await in Python."}
]
# Test mit DeepSeek (kostengünstig für einfache Erklärungen)
print("Test: DeepSeek V3.2 (kostengünstig)")
response = await client.chat_completions(
model="deepseek-v3.2",
messages=messages,
max_tokens=500
)
print(f"Response: {response['choices'][0]['message']['content'][:200]}...")
print(f"Metriken: {client.get_metrics()}\n")
# Test mit Claude (für komplexe Analyse)
print("Test: Claude Sonnet 4.5 (Reasoning)")
response = await client.chat_completions(
model="claude-sonnet-4.5",
messages=messages,
max_tokens=800
)
print(f"Metriken: {client.get_metrics()}")
if __name__ == "__main__":
asyncio.run(demo())
Vollständiger Routing-Executor mit Concurrency Control
"""
Production-Ready Routing Executor mit Semaphore-basierter Concurrency Control
Verhindert Rate Limits und optimiert parallel Request-Verarbeitung
"""
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class QueuedRequest:
id: str
prompt: str
messages: List[Dict]
constraints: Dict
created_at: datetime
future: asyncio.Future
def estimate_tokens(self) -> int:
"""Schätzt Input-Token basierend auf Prompt"""
text = " ".join([m.get("content", "") for m in self.messages])
return len(text) // 4 # Grobe Schätzung: ~4 Zeichen pro Token
class ConcurrencyController:
"""
Kontrolliert parallele API-Requests mit:
- Semaphore-basiertem Request-Limiting
- Request-Queuing bei Überlast
- Prioritätsbasiertes Abarbeiten
"""
def __init__(
self,
max_concurrent: int = 10,
requests_per_minute: int = 60,
burst_size: int = 20
):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.rate_limiter = asyncio.Semaphore(requests_per_minute)
self.burst_limiter = asyncio.Semaphore(burst_size)
self._request_times: List[datetime] = []
self._queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self._worker_task: Optional[asyncio.Task] = None
async def execute_with_limit(
self,
coro,
priority: int = 5
) -> Any:
"""
Führt Coroutine mit Concurrency-Limits aus.
Args:
coro: Die asynchrone Funktion zur Ausführung
priority: Niedrigere Werte = höhere Priorität (1-10)
"""
async with self.semaphore:
# Rate Limit Check
await self._check_rate_limit()
try:
result = await coro
logger.info(f"Request erfolgreich abgeschlossen")
return result
except Exception as e:
logger.error(f"Request fehlgeschlagen: {e}")
raise
async def _check_rate_limit(self):
"""Verhindert zu viele Requests pro Minute"""
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# Entferne alte Timestamps
self._request_times = [t for t in self._request_times if t > cutoff]
if len(self._request_times) >= 60:
wait_seconds = 60 - (now - self._request_times[0]).total_seconds()
logger.warning(f"Rate Limit erreicht. Warte {wait_seconds:.1f}s")
await asyncio.sleep(max(wait_seconds, 0.1))
class RoutingExecutor:
"""
Kombiniert ModelRouter + HolySheepClient + ConcurrencyController
für production-ready Batch-Processing.
"""
def __init__(
self,
router: ModelRouter,
client: HolySheepClient,
concurrency: ConcurrencyController
):
self.router = router
self.client = client
self.concurrency = concurrency
self._results: Dict[str, Dict] = {}
self._failed_requests: List[Dict] = []
async def process_request(
self,
request_id: str,
messages: List[Dict],
constraints: Dict = None,
priority: int = 5
) -> Dict:
"""
Verarbeitet einen einzelnen Request mit automatischem Routing.
Returns:
Dict mit response, routing_decision, metrics
"""
start_time = datetime.now()
# 1. Routing Entscheidung
prompt = messages[-1].get("content", "") if messages else ""
input_tokens = sum(len(m.get("content", "")) // 4 for m in messages)
routing = await self.router.route(prompt, input_tokens, constraints)
logger.info(
f"[{request_id}] Routing: {routing.model} "
f"(Cost: ${routing.estimated_cost:.4f}, Latenz: {routing.estimated_latency_ms}ms)"
)
# 2. Request mit Concurrency Control
async def _make_request():
return await self.client.chat_completions(
model=routing.model,
messages=messages,
max_tokens=constraints.get("max_output_tokens", 2000),
temperature=constraints.get("temperature", 0.7)
)
try:
response = await self.concurrency.execute_with_limit(
_make_request(),
priority=priority
)
# 3. Ergebnis speichern
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
result = {
"request_id": request_id,
"response": response,
"routing": {
"model": routing.model,
"provider": routing.provider,
"confidence": routing.confidence,
"reasoning": routing.reasoning
},
"metrics": {
"latency_ms": latency_ms,
"input_tokens": response.get("usage", {}).get("prompt_tokens", 0),
"output_tokens": response.get("usage", {}).get("completion_tokens", 0),
"estimated_cost_usd": self.client._estimate_cost(
routing.model,
response.get("usage", {}).get("prompt_tokens", 0),
response.get("usage", {}).get("completion_tokens", 0)
)
},
"success": True,
"timestamp": start_time.isoformat()
}
self._results[request_id] = result
return result
except Exception as e:
self._failed_requests.append({
"request_id": request_id,
"error": str(e),
"timestamp": start_time.isoformat(),
"attempted_model": routing.model
})
# Retry mit Claude als Fallback
try:
logger.warning(f"[{request_id}] Retry mit Claude Sonnet 4.5")
response = await self.client.chat_completions(
model="claude-sonnet-4.5",
messages=messages
)
return {
"request_id": request_id,
"response": response,
"routing": {
"model": "claude-sonnet-4.5",
"fallback": True
},
"success": True,
"retry": True
}
except Exception as fallback_error:
return {
"request_id": request_id,
"success": False,
"error": str(fallback_error),
"original_error": str(e)
}
async def process_batch(
self,
requests: List[Dict],
max_parallel: int = 5
) -> List[Dict]:
"""
Verarbeitet mehrere Requests parallel mit automatischer负载均衡.
Args:
requests: Liste von Dicts mit 'id', 'messages', 'constraints'
max_parallel: Maximale parallele Verarbeitung
"""
semaphore = asyncio.Semaphore(max_parallel)
async def _process_with_semaphore(req: Dict) -> Dict:
async with semaphore:
return await self.process_request(
request_id=req["id"],
messages=req["messages"],
constraints=req.get("constraints", {}),
priority=req.get("priority", 5)
)
# Parallel Execution mit Progress Tracking
tasks = [_process_with_semaphore(r) for r in requests]
results = []
for i, coro in enumerate(asyncio.as_completed(tasks)):
result = await coro
results.append(result)
logger.info(f"Fortschritt: {len(results)}/{len(requests)} abgeschlossen")
return results
def get_summary(self) -> Dict:
"""Erstellt Zusammenfassung aller verarbeiteten Requests"""
successful = [r for r in self._results.values() if r.get("success")]
failed = self._failed_requests
total_cost = sum(r.get("metrics", {}).get("estimated_cost_usd", 0) for r in successful)
avg_latency = sum(r.get("metrics", {}).get("latency_ms", 0) for r in successful) / max(len(successful), 1)
return {
"total_requests": len(self._results) + len(failed),
"successful": len(successful),
"failed": len(failed),
"total_cost_usd": round(total_cost, 4),
"avg_latency_ms": round(avg_latency, 2),
"success_rate": round(len(successful) / max(len(self._results) + len(failed), 1) * 100, 2),
"model_distribution": self._get_model_stats(successful)
}
def _get_model_stats(self, results: List[Dict]) -> Dict:
"""Zählt Requests pro Modell"""
distribution = {}
for r in results:
model = r.get("routing", {}).get("model", "unknown")
distribution[model] = distribution.get(model, 0) + 1
return distribution
============== BENCHMARK RUNNER ==============
async def run_benchmark():
"""Vergleicht Routing-Performance über 100 Requests"""
import random
router = ModelRouter()
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
concurrency = ConcurrencyController(max_concurrent=5, requests_per_minute=100)
executor = RoutingExecutor(router, client, concurrency)
# Test-Prompts mit unterschiedlicher Komplexität
test_prompts = [
# Triviale Tasks
{"role": "user", "content": "Was ist Python?"},
{"role": "user", "content": "Liste 3 Programmiersprachen auf."},
# Moderate Tasks
{"role": "user", "content": "Erkläre den Unterschied zwischen List und Tuple in Python mit Beispielen."},
{"role": "user", "content": "Schreibe eine kurze Zusammenfassung von Async/Await."},
# Komplexe Tasks
{"role": "user", "content": "Implementiere einen Binary Search Tree mit Insert, Delete und Search Methoden in Python. Include Fehlerbehandlung."},
{"role": "user", "content": "Erkläre Microservices-Architektur mit Vor- und Nachteilen. Gehe auf Container-Orchestrierung ein."},
]
requests = []
for i in range(100):
messages = random.choice(test_prompts).copy()
requests.append({
"id": f"req_{i:03d}",
"messages": [{"role": "system", "content": "Du bist ein hilfreicher Assistent."}, messages],
"constraints": {"max_cost_usd": 0.1, "max_latency_ms": 3000}
})
print("🚀 Starte Benchmark mit 100 Requests...")
start = datetime.now()
results =
Verwandte Ressourcen
Verwandte Artikel