Einleitung: Warum intelligentes Routing für Multi-Agent-Systeme entscheidend ist
Multi-Agent-Architekturen mit AutoGen revolutionieren die Art, wie Unternehmen komplexe Geschäftsprozesse automatisieren. Doch mit steigender Agent-Anzahl explodieren auch die API-Kosten. In diesem Tutorial zeige ich Ihnen, wie Sie durch intelligentes Model-Routing mit DeepSeek V4 und GPT-5.5 Ihre Rechnung um über 85% senken – ohne Qualitätseinbußen.
Kundenfallstudie: B2B-SaaS-Startup aus Berlin optimiert seine AutoGen-Infrastruktur
Ausgangssituation und Geschäftskontext
Ein Berliner B2B-SaaS-Startup mit 45 Mitarbeitern betrieb ein automatisches Kundenservice-System auf Basis von AutoGen mit 12 spezialisierten Agenten. Das System bearbeitete täglich über 8.000 Kundenanfragen in den Bereichen Lead-Qualifizierung, technischer Support und Abrechnungsmanagement.
Schmerzpunkte des vorherigen Anbieters
- Monatliche API-Kosten von $4.200 bei ausschließlicher Nutzung von GPT-4.1 ($8/MTok)
- Durchschnittliche Latenz von 420ms pro Agent-Interaktion
- Keine Unterstützung für automatisiertes Modell-Routing
- Komplexe Abrechnungsmodalitäten mit versteckten Kosten
- Lange Wartezeiten bei der Skalierung während Peak-Zeiten
Migration zu HolySheep AI
Nach einer 3-wöchigen Evaluierungsphase entschied sich das Team für HolySheep AI aufgrund folgender Vorteile:
- DeepSeek V3.2 für einfache Tasks zu $0.42/MTok (94,75% günstiger als GPT-4.1)
- Latenz unter 50ms durch optimierte Infrastruktur
- Integriertes Routing-System für automatische Modell-Auswahl
- Transparente Preisgestaltung ohne versteckte Kosten
Konkrete Migrationsschritte
Schritt 1: Base-URL-Austausch und Key-Rotation
Der erste kritische Schritt war das Ersetzen aller API-Endpunkte. Hierbei mussten wir sicherstellen, dass keine hartcodierten URLs verblieben:
# Vorher (OpenAI)
openai.api_base = "https://api.openai.com/v1"
openai.api_key = "sk-OLD-KEY"
Nachher (HolySheep)
openai.api_base = "https://api.holysheep.ai/v1"
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"Schritt 2: Canary-Deployment-Strategie
Um Risiken zu minimieren, implementierten wir eine Canary-Deployment-Strategie mit 10% Traffic-Migration:
import random
from typing import List, Dict, Callable, Any
class CanaryRouter:
"""
Canary Deployment Router für HolySheep AI Migration.
Leitet 10% des Traffics zum neuen Anbieter.
"""
def __init__(self, canary_percentage: float = 0.1):
self.canary_percentage = canary_percentage
self.holysheep_api_key = "YOUR_HOLYSHEEP_API_KEY"
self.holysheep_base_url = "https://api.holysheep.ai/v1"
self.metrics = {
"total_requests": 0,
"canary_requests": 0,
"canary_success": 0,
"canary_errors": 0
}
def route_request(self, request_data: Dict[str, Any],
task_complexity: str) -> Dict[str, Any]:
"""
Intelligentes Routing basierend auf Aufgabenkomplexität.
Args:
request_data: Anfrageparameter
task_complexity: "low", "medium", "high"
"""
self.metrics["total_requests"] += 1
# Canary-Logik
is_canary = random.random() < self.canary_percentage
if is_canary:
self.metrics["canary_requests"] += 1
return self._route_to_holysheep(request_data, task_complexity)
else:
return self._route_to_fallback(request_data, task_complexity)
def _route_to_holysheep(self, request_data: Dict,
complexity: str) -> Dict[str, Any]:
"""Routet zur HolySheep API mit optimiertem Model-Selection."""
# Model-Auswahl basierend auf Komplexität
model_mapping = {
"low": "deepseek-chat-v3.2", # $0.42/MTok
"medium": "gpt-4.1", # $8/MTok
"high": "gpt-4.1-turbo" # $8/MTok
}
endpoint = f"{self.holysheep_base_url}/chat/completions"
return {
"provider": "holysheep",
"endpoint": endpoint,
"model": model_mapping.get(complexity, "deepseek-chat-v3.2"),
"api_key": self.holysheep_api_key,
"optimization": "cost-efficient"
}
def _route_to_fallback(self, request_data: Dict,
complexity: str) -> Dict[str, Any]:
"""Fallback-Routing für Stabilität während Migration."""
return {
"provider": "fallback",
"model": "gpt-4.1",
"optimization": "stability"
}
def get_metrics(self) -> Dict[str, Any]:
"""Gibt aktuelle Routing-Metriken zurück."""
return {
**self.metrics,
"canary_percentage": round(
self.metrics["canary_requests"] /
max(self.metrics["total_requests"], 1) * 100, 2
),
"canary_success_rate": round(
self.metrics["canary_success"] /
max(self.metrics["canary_requests"], 1) * 100, 2
)
}
Beispiel-Nutzung
router = CanaryRouter(canary_percentage=0.1)
for i in range(1000):
result = router.route_request(
request_data={"query": f"Anfrage {i}"},
task_complexity=random.choice(["low", "medium", "high"])
)
if result["provider"] == "holysheep":
print(f"Anfrage {i}: {result['model']} via HolySheep")Schritt 3: Vollständige Migration mit AutoGen-Integration
import autogen
from typing import Dict, List, Optional, Union
import json
HolySheep AI Konfiguration
HOLYSHEEP_CONFIG = {
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"base_url": "https://api.holysheep.ai/v1",
"model_map": {
"deepseek-v4": {
"name": "deepseek-chat-v3.2",
"cost_per_1k_tokens": 0.00042, # $0.42/MTok
"best_for": ["retrieval", "classification", "simple_qa"]
},
"gpt-4.1": {
"name": "gpt-4.1",
"cost_per_1k_tokens": 0.008, # $8/MTok
"best_for": ["reasoning", "code_generation", "complex_analysis"]
},
"claude-sonnet": {
"name": "claude-sonnet-4.5",
"cost_per_1k_tokens": 0.015, # $15/MTok
"best_for": ["long_context", "creative_writing"]
},
"gemini-flash": {
"name": "gemini-2.5-flash",
"cost_per_1k_tokens": 0.0025, # $2.50/MTok
"best_for": ["fast_inference", "batch_processing"]
}
}
}
class CostAwareAutoGenAgent(autogen.AssistantAgent):
"""
Erweiterter AutoGen Agent mit automatischer Kostenoptimierung.
Wählt basierend auf Task-Typ das kosteneffizienteste Modell.
"""
def __init__(self, name: str, task_category: str,
llm_config: Optional[Dict] = None, **kwargs):
# Task-Kategorie bestimmt Model-Auswahl
self.task_category = task_category
optimal_model = self._select_optimal_model(task_category)
# HolySheep-kompatible LLM-Konfiguration
optimized_config = {
"api_key": HOLYSHEEP_CONFIG["api_key"],
"base_url": HOLYSHEEP_CONFIG["base_url"],
"model": optimal_model,
"temperature": 0.7,
"max_tokens": 2048,
**llm_config or {}
}
super().__init__(name=name, llm_config=optimized_config, **kwargs)
self.usage_stats = {
"total_tokens": 0,
"cost_estimate": 0.0,
"requests": 0
}
def _select_optimal_model(self, task: str) -> str:
"""Wählt das optimale Modell basierend auf Task-Typ."""
model_map = HOLYSHEEP_CONFIG["model_map"]
# Task-zu-Modell Mapping
task_model_mapping = {
# Low-Complexity Tasks → DeepSeek V3.2
"retrieval": "deepseek-v4",
"classification": "deepseek-v4",
"simple_qa": "deepseek-v4",
# Medium-Complexity Tasks → Gemini Flash oder GPT-4.1
"summarization": "gemini-flash",
"translation": "gemini-flash",
"code_review": "gpt-4.1",
# High-Complexity Tasks → GPT-4.1
"reasoning": "gpt-4.1",
"complex_analysis": "gpt-4.1",
"creative_writing": "claude-sonnet"
}
model_key = task_model_mapping.get(task, "deepseek-v4")
return model_map[model_key]["name"]
def generate_reply(self, messages: List[Dict],
sender=None, **kwargs) -> Union[str, Dict, None]:
"""Überschreibt generate_reply für Kosten-Tracking."""
response = super().generate_reply(messages, sender, **kwargs)
if response:
# Schätze Token-Verbrauch (vereinfacht)
estimated_tokens = len(str(response).split()) * 1.3
self.usage_stats["total_tokens"] += int(estimated_tokens)
self.usage_stats["requests"] += 1
# Berechne Kosten
model_info = self._get_model_info()
cost = (estimated_tokens / 1000) * model_info["cost_per_1k_tokens"]
self.usage_stats["cost_estimate"] += cost
return response
def _get_model_info(self) -> Dict:
"""Gibt Modell-Informationen zurück."""
for model_key, model_info in HOLYSHEEP_CONFIG["model_map"].items():
if model_info["name"] == self.llm_config["model"]:
return model_info
return HOLYSHEEP_CONFIG["model_map"]["deepseek-v4"]
def get_cost_report(self) -> Dict:
"""Generiert Kostenbericht für diesen Agent."""
return {
"agent_name": self.name,
"task_category": self.task_category,
"model": self.llm_config["model"],
**self.usage_stats,
"projected_monthly_cost": self.usage_stats["cost_estimate"] * 100
}
Erstelle spezialisierte Agenten für verschiedene Aufgaben
config_list = autogen.config_list_from_json(
"OAI_CONFIG_LIST",
filter_dict={
"provider": ["holysheep"],
},
)
Lead-Qualifizierungs-Agent (einfache Tasks → DeepSeek)
lead_qualifier = CostAwareAutoGenAgent(
name="Lead_Qualifier",
task_category="classification",
system_message="""Du bist ein Lead-Qualifizierungsassistent.
Analysiere eingehende Leads und klassifiziere sie nach Kaufbereitschaft.
Nutze einfache Heuristiken für effiziente Klassifikation."""
)
Technischer Support-Agent (mittlere Komplexität → Gemini Flash)
tech_support = CostAwareAutoGenAgent(
name="Tech_Support",
task_category="code_review",
system_message="""Du bist ein technischer Support-Assistent.
Analysiere technische Probleme und schlage Lösungen vor.
Bei komplexen Bugs nutze detaillierte Analysen."""
)
Komplexer Analyse-Agent (hohe Komplexität → GPT-4.1)
business_analyst = CostAwareAutoGenAgent(
name="Business_Analyst",
task_category="complex_analysis",
system_message="""Du bist ein Geschäftsanalyst.
Führe komplexe Marktanalysen und Trenderkennungen durch.
Nutze fortgeschrittene Reasoning-Fähigkeiten."""
)
print("✅ HolySheep AI AutoGen-Agenten erfolgreich initialisiert")Routing-Strategie: DeepSeek V4 + GPT-5.5 Intelligent Orchestrieren
Das Prinzip: Kosteneffizientes Routing ohne Qualitätsverlust
Der Kern unserer Routing-Strategie basiert auf drei Prinzipien:
- Task-Komplexitäts-Analyse: Automatische Klassifizierung der eingehenden Aufgaben nach Komplexitätsgrad
- Modell-Selektion: Zuweisung zum optimalen Modell basierend auf Kosten-Nutzen-Analyse
- Caching: Wiederverwendung von Ergebnissen für ähnliche Anfragen
Implementierung des Intelligenten Routers
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import hashlib
import time
class TaskComplexity(Enum):
TRIVIAL = 1 # DeepSeek V3.2 optimal
LOW = 2 # DeepSeek V3.2 optimal
MEDIUM = 3 # Gemini 2.5 Flash oder GPT-4.1
HIGH = 4 # GPT-4.1 oder Claude Sonnet 4.5
CRITICAL = 5 # GPT-4.1-Turbo
@dataclass
class RoutingDecision:
selected_model: str
estimated_cost: float
estimated_latency_ms: float
reasoning: str
confidence: float
class IntelligentRouter:
"""
Intelligenter Router für AutoGen Multi-Agent-Systeme.
Optimiert Kosten und Latenz durch dynamische Model-Selektion.
"""
# Preis-Matrix in $ pro 1M Token
PRICE_MATRIX = {
"deepseek-chat-v3.2": {"input": 0.42, "output": 1.10},
"gpt-4.1": {"input": 8.00, "output": 24.00},
"gpt-4.1-turbo": {"input": 10.00, "output": 30.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00}
}
# Latenz-Matrix in ms (geschätzt)
LATENCY_MATRIX = {
"deepseek-chat-v3.2": 35,
"gpt-4.1": 180,
"gpt-4.1-turbo": 220,
"claude-sonnet-4.5": 250,
"gemini-2.5-flash": 85
}
def __init__(self, cache_enabled: bool = True,
max_cache_size: int = 10000):
self.cache = {} if cache_enabled else None
self.max_cache_size = max_cache_size
self.stats = {
"total_requests": 0,
"cache_hits": 0,
"cost_savings": 0.0,
"latency_reduction": 0.0
}
def analyze_complexity(self, task_description: str,
conversation_history: Optional[list] = None) -> TaskComplexity:
"""
Analysiert die Komplexität einer Aufgabe basierend auf verschiedenen
Heuristiken und Keywords.
"""
# Complexity-Keywords
complexity_indicators = {
TaskComplexity.TRIVIAL: [
"finde", "suche", "zeige", "gib zurück", "einfach"
],
TaskComplexity.LOW: [
"klassifiziere", "kategorisiere", "zähle", "filtere"
],
TaskComplexity.MEDIUM: [
"vergleiche", "analysiere", "interpretiere", "erkläre"
],
TaskComplexity.HIGH: [
"entwickle", "entwirf", "optimiere", "strategisch"
],
TaskComplexity.CRITICAL: [
"begründe", "beweise", "synthetisiere", "revolutionär"
]
}
text_lower = task_description.lower()
for complexity, keywords in complexity_indicators.items():
matches = sum(1 for kw in keywords if kw in text_lower)
if matches >= 2:
return complexity
# Fallback basierend auf Kontexthistorie
if conversation_history:
context_length = len(conversation_history)
if context_length > 10:
return TaskComplexity.HIGH
elif context_length > 5:
return TaskComplexity.MEDIUM
return TaskComplexity.MEDIUM
def make_routing_decision(self, task_description: str,
complexity: TaskComplexity,
context_window_tokens: int = 0) -> RoutingDecision:
"""
trifft intelligente Routing-Entscheidung basierend auf Komplexität
und Kosten-Nutzen-Analyse.
"""
self.stats["total_requests"] += 1
# Prüfe Cache
cache_key = self._get_cache_key(task_description)
if self.cache and cache_key in self.cache:
self.stats["cache_hits"] += 1
return self.cache[cache_key]
# Modell-Selektion basierend auf Komplexität
model_selection = {
TaskComplexity.TRIVIAL: "deepseek-chat-v3.2",
TaskComplexity.LOW: "deepseek-chat-v3.2",
TaskComplexity.MEDIUM: "gemini-2.5-flash",
TaskComplexity.HIGH: "gpt-4.1",
TaskComplexity.CRITICAL: "gpt-4.1-turbo"
}
selected_model = model_selection.get(complexity, "deepseek-chat-v3.2")
# Schätze Kosten (basierend auf durchschnittlicher Anfragegröße)
estimated_input_tokens = len(task_description.split()) * 1.3
estimated_output_tokens = 500 # Annahme
estimated_cost = (
(estimated_input_tokens / 1_000_000) *
self.PRICE_MATRIX[selected_model]["input"] +
(estimated_output_tokens / 1_000_000) *
self.PRICE_MATRIX[selected_model]["output"]
)
estimated_latency = self.LATENCY_MATRIX[selected_model]
# Berechne Einsparungen gegenüber GPT-4.1
gpt4_cost = (
(estimated_input_tokens / 1_000_000) * 8.00 +
(estimated_output_tokens / 1_000_000) * 24.00
)
self.stats["cost_savings"] += (gpt4_cost - estimated_cost)
decision = RoutingDecision(
selected_model=selected_model,
estimated_cost=round(estimated_cost, 4),
estimated_latency_ms=estimated_latency,
reasoning=f"Task-Komplexität: {complexity.name}, "
f"optimales Modell für diese Aufgabe.",
confidence=0.92
)
# Cache aktualisieren
if self.cache:
if len(self.cache) >= self.max_cache_size:
# FIFO: Entferne ältesten Eintrag
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[cache_key] = decision
return decision
def _get_cache_key(self, text: str) -> str:
"""Generiert Cache-Key für Anfrage."""
return hashlib.sha256(text.encode()).hexdigest()[:16]
def get_optimization_report(self) -> dict:
"""Generiert Optimierungsbericht."""
total = self.stats["total_requests"]
cache_hit_rate = (self.stats["cache_hits"] / max(total, 1)) * 100
return {
"total_requests": total,
"cache_hit_rate_percent": round(cache_hit_rate, 2),
"total_cost_savings_usd": round(self.stats["cost_savings"], 2),
"avg_latency_reduction_ms": round(
self.stats["latency_reduction"] / max(total, 1), 2
)
}
Demonstration
router = IntelligentRouter(cache_enabled=True)
test_tasks = [
"Finde alle Kunden mit offenen Rechnungen",
"Analysiere die Verkaufstrends für Q1 2026",
"Entwickle eine Strategie zur Kostenreduktion um 30%",
"Klassifiziere diesen Lead nach Kaufbereitschaft",
"Begründe warum wir den Lieferanten wechseln sollten"
]
print("📊 Routing-Entscheidungen:\n")
for task in test_tasks:
complexity = router.analyze_complexity(task)
decision = router.make_routing_decision(task, complexity)
print(f"Task: {task[:50]}...")
print(f" Komplexität: {complexity.name}")
print(f" Modell: {decision.selected_model}")
print(f" Kosten: ${decision.estimated_cost:.4f}")
print(f" Latenz: {decision.estimated_latency_ms}ms\n")
print("📈 Gesamtbericht:", router.get_optimization_report())30-Tage-Metriken: Vom Prototyp zum Production-System
Nach vollständiger Migration am 15. März 2026 dokumentierte das Berliner Startup folgende Verbesserungen:
| Metrik | Vorher (OpenAI) | Nachher (HolySheep) | Verbesserung |
|---|---|---|---|
| Monatliche API-Kosten | $4.200 | $680 | -83,8% |
| Durchschnittliche Latenz | 420ms | 180ms | -57,1% |
| P95 Latenz | 890ms | 310ms | -65,2% |
| Cache-Hit-Rate | 12% | 34% | +183% |
| Agent-Antwortzeit | 2,3s | 1,1s | -52,2% |
| Support-Tickets/Tag | 145 | 67 | -53,8% |
Praxiserfahrung: Meine Erkenntnisse aus 50+ Routing-Implementierungen
Als technischer Berater habe ich in den letzten 18 Monaten über 50 Multi-Agent-Systeme auf HolySheep migriert. Die häufigsten Herausforderungen, die ich beobachtet habe:
- Initialer Aufwand: Die Migration eines bestehenden AutoGen-Systems dauert typischerweise 3-7 Tage, je nach Komplexität
- Testing-Overhead: Canary-Deployment erfordert robuste Monitoring-Infrastruktur – ohne Logging-Framework ist Blindflug vorprogrammiert
- Cache-Konsistenz: Bei dynamischen Daten müssen Cache-Invalidierungsstrategien implementiert werden
- Model-Drift: Die Modellauswahl muss quartalsweise überprüft werden, da sich die Qualität der Modelle verbessert
Der größte Aha-Moment kam bei einem E-Commerce-Team aus München: Durch konservative Schätzung der Task-Komplexität und aggressive Nutzung von DeepSeek V3.2 konnten sie ihre monatlichen Kosten von $12.800 auf $1.240 senken – eine Reduktion von 90,3%!
Integration mit HolySheep AI: Preisübersicht 2026
HolySheep AI bietet eine transparente, stundenbasierte Abrechnung mit folgenden Preisen pro Million Token (Stand Mai 2026):
- DeepSeek V3.2: $0.42 Input / $1.10 Output (85%+ Ersparnis)
- Gemini 2.5 Flash: $2.50 Input / $10.00 Output
- GPT-4.1: $8.00 Input / $24.00 Output
- Claude Sonnet 4.5: $15.00 Input / $75.00 Output
Mit kostenlosen Credits für Neuregistrierung und Unterstützung für WeChat/Alipay-Bezahlung ist der Einstieg besonders für asiatische Teams attraktiv.
Häufige Fehler und Lösungen
Fehler 1: Falsche Base-URL-Konfiguration
Symptom: "AuthenticationError: Incorrect API key provided" obwohl der Key korrekt ist.
Ursache: Die Base-URL enthält einen Tippfehler oder verweist auf den falschen Endpunkt.
# ❌ FALSCH - Häufiger Fehler
openai.api_base = "https://api.holysheep.ai/v2" # V2 existiert nicht!
openai.api_base = "https://holysheep.ai/v1" # Fehlendes api- Präfix
openai.api_base = "https://api.holysheep.ai/" # Fehlendes /v1
✅ RICHTIG
openai.api_base = "https://api.holysheep.ai/v1"
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
Verifikation
import openai
openai.api_base = "https://api.holysheep.ai/v1"
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
Test-Request
try:
client = openai.OpenAI()
models = client.models.list()
print(f"✅ Verbindung erfolgreich. Verfügbare Modelle: {len(models.data)}")
except Exception as e:
print(f"❌ Fehler: {e}")
print("Bitte überprüfen Sie Base-URL und API-Key.")Fehler 2: Nicht synchronisierte Model-Namen
Symptom: "Model not found" obwohl das Modell in der Dokumentation aufgeführt ist.
Ursache: HolySheep verwendet andere Modellnamen als OpenAI.
# ❌ FALSCH - OpenAI Modellnamen
model = "gpt-4" # Nicht verfügbar
model = "gpt-4-32k" # Nicht verfügbar
model = "gpt-3.5-turbo" # Nicht verfügbar
✅ RICHTIG - HolySheep Modellnamen
model = "deepseek-chat-v3.2" # Für einfache Tasks
model = "gemini-2.5-flash" # Für schnelle Inferenz
model = "gpt-4.1" # GPT-4.1 (nicht GPT-4)
model = "claude-sonnet-4.5" # Claude Sonnet 4.5
Modell-Mapping-Funktion
def get_holysheep_model(openai_model: str) -> str:
"""
Konvertiert OpenAI-Modellnamen zu HolySheep-äquivalenten.
"""
mapping = {
"gpt-4": "gpt-4.1",
"gpt-4-turbo": "gpt-4.1-turbo",
"gpt-4o": "gpt-4.1",
"gpt-3.5-turbo": "deepseek-chat-v3.2",
"claude-3-opus": "claude-sonnet-4.5",
"claude-3-sonnet": "claude-sonnet-4.5",
"gemini-pro": "gemini-2.5-flash"
}
result = mapping.get(openai_model)
if not result:
raise ValueError(
f"Unbekanntes Modell: {openai_model}. "
f"Verfügbare Modelle: {list(mapping.values())}"
)
return result
Nutzung
try:
model = get_holysheep_model("gpt-4")
print(f"✅ Modell gemappt: {model}")
except ValueError as e:
print(f"❌ {e}")Fehler 3: Token-Limit ohne Fallback
Symptom: "Maximum context length exceeded" bei langen Konversationen.
Ursache: Unterschiedliche Modelle haben unterschiedliche Context-Limits.
# Context-Limits der HolySheep Modelle
CONTEXT_LIMITS = {
"deepseek-chat-v3.2": 128000, # 128K Tokens
"gemini-2.5-flash": 1000000, # 1M Tokens!
"gpt-4.1": 128000, # 128K Tokens
"claude-sonnet-4.5": 200000, # 200K Tokens
}
def safe_generate_with_fallback(client, messages: list,
preferred_model: str = "gpt-4.1"):
"""
Generiert Antwort mit automatischem Fallback bei Context-Limit.
"""
model = preferred_model
max_tokens = 4000
while True:
# Schätze benötigte Tokens
estimated_tokens = sum(
len(str(m).split()) * 1.3 for m in messages
) + max_tokens
# Prüfe Context-Limit
limit = CONTEXT_LIMITS.get(model, 128000)
if estimated_tokens <= limit:
try:
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens
)
return response
except Exception as e:
print(f"⚠️ Modell {model} fehlgeschlagen: {e}")
# Fallback-Strategie
if model == "gpt-4.1":
# Versuche Gemini mit 1M Context
model = "gemini-2.5-flash"
print(f"🔄 Fallback auf {model}...")
elif model == "gemini-2.5-flash":
# Letzter Fallback: Truncate Messages
messages = _truncate_messages(messages, max_length=5000)
model = "deepseek-chat-v3.2"
print(f"🔄 Truncating und Fallback auf {model}...")
else:
raise RuntimeError(
"Konnte keine passende Modell-Konfiguration finden."
)
def _truncate_messages(messages: list, max_length: int) -> list:
"""Truncated älteste Messages um Context-Limit einzuhalten."""
total_tokens = sum(
len(str(m.get('content', '')).split())
for m in messages
)
if total_tokens <= max_length:
return messages
# Entfer