Einleitung: Warum AI Agent Failover kritisch ist
Stellen Sie sich folgendes Szenario vor: Es ist der größte Shopping-Tag des Jahres — 11. November 2026, 23:58 Uhr. Ihr E-Commerce-KI-Chatbot verarbeitet gerade 15.000 Anfragen pro Minute. Plötzlich meldet Ihr Monitoring: OpenAI API Error 429 — Rate Limit Exceeded. In den nächsten 30 Sekunden bricht Ihr System zusammen, und Sie verlieren potenziell 50.000 € an verpassten Verkäufen.
Als Engineering Lead bei einem mittelständischen E-Commerce-Unternehmen habe ich genau dieses Szenario dreimal erlebt — einmal während eines Produkt-Launches, einmal während einer Marketing-Kampagne und einmal während eines viralen Social-Media-Posts. Der Schaden war enorm: negative Kundenerlebnisse, Support-Tickets explodierten, und unser NPS sank um 15 Punkte.
Die Lösung ist ein robustes Multi-Provider-Failover-System, das ich in diesem Tutorial vollständig implementieren werde. Mit HolySheep AI als zentralem Aggregator können Sie redundant gegen Ausfälle absichern und gleichzeitig bis zu 85% der API-Kosten sparen.
Das Problem: Single-Provider-Abhängigkeit
Die meisten AI-Agenten-Implementierungen haben eine gefährliche Architektur: Sie nutzen einen einzigen API-Provider. Das ist riskant, weil:
- API-Ausfälle: OpenAI hatte 2025 durchschnittlich 3 ungeplante Ausfälle pro Monat
- Rate Limits: Enterprise-Kontingente werden schnell erschöpft bei Lastspitzen
- Geografische Latenz: Ein Provider in einer Region kann langsam sein
- Kostenexplosion: Ohne Lastverteilung zahlen Sie bei Spitzenauslastung Premium-Preise
Die Lösung: HolySheep Multi-Provider Architecture
HolySheep fungiert als intelligenter Router zwischen Ihrem AI Agent und mehreren Providern: OpenAI GPT-4.1, Anthropic Claude Sonnet 4.5, Google Gemini 2.5 Flash und DeepSeek V3.2. Mit <50ms zusätzlicher Latenz und automatischer Failover-Logik wird Ihr System ausfallsicher.
Architektur-Übersicht
┌─────────────────────────────────────────────────────────────────┐
│ AI Agent (Your Application) │
└───────────────────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ HolySheep Gateway (Load Balancer) │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ Health Probe Manager (Alle 5 Sekunden) │ │
│ │ ├── GPT-4.1: Status 200, Latency 45ms ✓ │ │
│ │ ├── Claude Sonnet 4.5: Status 200, Latency 38ms ✓ │ │
│ │ ├── Gemini 2.5 Flash: Status 503, Latency Timeout ✗ │ │
│ │ └── DeepSeek V3.2: Status 200, Latency 32ms ✓ │ │
│ └───────────────────────────────────────────────────────────┘ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ Failover Strategy Engine │ │
│ │ 1. Primary: DeepSeek V3.2 (günstigster, schnellster) │ │
│ │ 2. Secondary: Claude Sonnet 4.5 (beste Qualität) │ │
│ │ 3. Tertiary: GPT-4.1 (Backup) │ │
│ └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
┌───────────────────────┼───────────────────────┐
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ DeepSeek V3.2│ │ Claude Sonnet │ │ GPT-4.1 │
│ $0.42/MTok │ │ 4.5 $15 │ │ $8/MTok │
└───────────────┘ └───────────────┘ └───────────────┘Health Probe Implementation
Der Health Probe ist das Herzstück des Failover-Systems. Er überwacht kontinuierlich die Erreichbarkeit und Performance aller konfigurierten Provider.
import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
UNKNOWN = "unknown"
@dataclass
class ProviderHealth:
name: str
base_url: str
status: ProviderStatus = ProviderStatus.UNKNOWN
latency_ms: float = 0.0
error_count: int = 0
success_count: int = 0
last_check: float = 0.0
consecutive_failures: int = 0
class HolySheepHealthProbe:
"""
Multi-Provider Health Probe für HolySheep AI Gateway
Überwacht kontinuierlich alle Provider und berechnet Health Scores
"""
def __init__(self, api_key: str, check_interval: int = 5):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.check_interval = check_interval
self.providers: Dict[str, ProviderHealth] = {}
self.health_threshold_ms = 2000 # 2 Sekunden Timeout
self.failure_threshold = 3 # 3 Fehler = Provider markiert als unhealthy
async def initialize_providers(self):
"""Initialisiert alle verfügbaren Provider mit HolySheep"""
provider_configs = [
{"name": "deepseek", "model": "deepseek-v3.2", "priority": 1},
{"name": "claude", "model": "claude-sonnet-4.5", "priority": 2},
{"name": "gpt4", "model": "gpt-4.1", "priority": 3},
{"name": "gemini", "model": "gemini-2.5-flash", "priority": 4},
]
for config in provider_configs:
self.providers[config["name"]] = ProviderHealth(
name=config["name"],
base_url=f"{self.base_url}/chat/completions"
)
async def _probe_provider(
self,
session: aiohttp.ClientSession,
provider: ProviderHealth,
model: str
) -> tuple[bool, float]:
"""
Sendet einen Health-Check-Request an einen Provider
Gibt (success, latency_ms) zurück
"""
start_time = time.perf_counter()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 5, # Minimal für Speed-Test
"temperature": 0.1
}
try:
async with session.post(
provider.base_url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.health_threshold_ms / 1000)
) as response:
latency = (time.perf_counter() - start_time) * 1000
if response.status == 200:
return True, latency
elif response.status in [429, 500, 502, 503, 504]:
# Rate Limit oder Server Error - vorübergehendes Problem
return False, latency
else:
return False, latency
except asyncio.TimeoutError:
latency = (time.perf_counter() - start_time) * 1000
return False, latency
except Exception as e:
latency = (time.perf_counter() - start_time) * 1000
return False, latency
async def check_all_providers(self) -> Dict[str, ProviderHealth]:
"""
Führt Health-Checks für alle Provider parallel aus
"""
async with aiohttp.ClientSession() as session:
tasks = []
model_map = {
"deepseek": "deepseek-v3.2",
"claude": "claude-sonnet-4.5",
"gpt4": "gpt-4.1",
"gemini": "gemini-2.5-flash"
}
for name, provider in self.providers.items():
tasks.append(self._probe_provider(session, provider, model_map[name]))
results = await asyncio.gather(*tasks, return_exceptions=True)
for (name, provider), result in zip(self.providers.items(), results):
if isinstance(result, Exception):
provider.consecutive_failures += 1
provider.status = ProviderStatus.UNHEALTHY
else:
success, latency = result
provider.latency_ms = latency
provider.last_check = time.time()
if success:
provider.success_count += 1
provider.consecutive_failures = 0
# Latenz-basiertes Health-Ranking
if latency < 100:
provider.status = ProviderStatus.HEALTHY
elif latency < 500:
provider.status = ProviderStatus.DEGRADED
else:
provider.status = ProviderStatus.DEGRADED
else:
provider.consecutive_failures += 1
if provider.consecutive_failures >= self.failure_threshold:
provider.status = ProviderStatus.UNHEALTHY
provider.error_count += 1
return self.providers
def get_best_provider(self) -> Optional[ProviderHealth]:
"""
Gibt den besten verfügbaren Provider basierend auf Health und Latenz zurück
"""
available = [
p for p in self.providers.values()
if p.status != ProviderStatus.UNHEALTHY
]
if not available:
return None
# Sortiere nach: 1. Status, 2. Latenz
available.sort(key=lambda x: (x.status.value, x.latency_ms))
return available[0]
async def start_monitoring(self):
"""Startet kontinuierliches Monitoring im Hintergrund"""
await self.initialize_providers()
while True:
await self.check_all_providers()
await asyncio.sleep(self.check_interval)
Beispiel: Monitoring starten
async def main():
probe = HolySheepHealthProbe(
api_key="YOUR_HOLYSHEEP_API_KEY",
check_interval=5
)
await probe.initialize_providers()
# Einmaliger Health-Check
results = await probe.check_all_providers()
print("=== HolySheep Provider Health Status ===")
for name, health in results.items():
print(f"{name}: {health.status.value} | Latenz: {health.latency_ms:.1f}ms")
best = probe.get_best_provider()
print(f"\nEmpfohlener Provider: {best.name if best else 'KEINER'}")
if __name__ == "__main__":
asyncio.run(main())Automatic Failover Strategie mit Retry-Logic
Der Failover-Mechanismus muss intelligent sein: Er sollte nicht bei jedem kleinen Fehler springen, aber schnell genug reagieren, um den Benutzer nicht zu frustrieren.
import asyncio
import aiohttp
import time
import logging
from typing import List, Optional, Callable, Any
from dataclasses import dataclass
from enum import Enum
from holy_health_probe import HolySheepHealthProbe, ProviderHealth, ProviderStatus
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FailoverStrategy(Enum):
FAILOVER_ON_ERROR = "failover_on_error" # Sofort wechseln bei Fehler
FAILOVER_AFTER_RETRY = "failover_after_retry" # Erst retries, dann wechseln
CIRCUIT_BREAKER = "circuit_breaker" # Progressive Backoff-Strategie
@dataclass
class RequestContext:
model: str
messages: List[dict]
temperature: float = 0.7
max_tokens: int = 2048
timeout: float = 30.0
@dataclass
class FailoverConfig:
max_retries: int = 3
base_delay_ms: int = 100
max_delay_ms: int = 5000
exponential_base: float = 2.0
jitter: float = 0.1
strategy: FailoverStrategy = FailoverStrategy.FAILOVER_AFTER_RETRY
class HolySheepFailoverClient:
"""
HolySheep AI Client mit automatischem Failover
Nutzt Health Probes für intelligente Provider-Auswahl
"""
def __init__(
self,
api_key: str,
health_probe: HolySheepHealthProbe,
config: Optional[FailoverConfig] = None
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.health_probe = health_probe
self.config = config or FailoverConfig()
self.request_count = 0
self.failover_count = 0
async def _calculate_delay(self, attempt: int) -> float:
"""Berechnet Delay mit Exponential Backoff und Jitter"""
delay = self.config.base_delay_ms * (self.config.exponential_base ** attempt)
delay = min(delay, self.config.max_delay_ms)
# Jitter hinzufügen um Thundering Herd zu vermeiden
import random
jitter_range = delay * self.config.jitter
delay += random.uniform(-jitter_range, jitter_range)
return delay / 1000 # Zurück in Sekunden
async def _call_provider(
self,
session: aiohttp.ClientSession,
provider: ProviderHealth,
context: RequestContext
) -> dict:
"""Führt einen API-Call zum angegebenen Provider durch"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": context.model,
"messages": context.messages,
"temperature": context.temperature,
"max_tokens": context.max_tokens
}
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=context.timeout)
) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=429,
message="Rate Limit Exceeded"
)
elif response.status >= 500:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=f"Server Error: {response.status}"
)
else:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=f"Client Error: {response.status}"
)
def _get_ordered_providers(self) -> List[ProviderHealth]:
"""Gibt Provider in Failover-Reihenfolge zurück"""
providers = list(self.health_probe.providers.values())
# Filtere unhealthy Provider
available = [p for p in providers if p.status != ProviderStatus.UNHEALTHY]
# Sortiere nach Latenz (schnellster zuerst)
available.sort(key=lambda x: x.latency_ms)
return available
async def chat_completion(
self,
model: str,
messages: List[dict],
temperature: float = 0.7,
max_tokens: int = 2048,
on_failover: Optional[Callable[[str, str], None]] = None
) -> dict:
"""
Führt einen Chat-Completion-Request mit automatischem Failover aus
Args:
model: Modell-ID (z.B. "deepseek-v3.2", "claude-sonnet-4.5")
messages: Chat-Nachrichten
temperature: Sampling-Temperatur
max_tokens: Maximale Token
on_failover: Callback-Funktion für Failover-Events
Returns:
API-Response als Dictionary
"""
context = RequestContext(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
providers = self._get_ordered_providers()
if not providers:
raise RuntimeError("Keine verfügbaren Provider!")
last_error = None
provider_tried = []
for attempt in range(self.config.max_retries):
# Wähle Provider (Round-Robin innerhalb der verfügbaren Provider)
provider_index = attempt % len(providers)
provider = providers[provider_index]
if provider.name in provider_tried and len(provider_tried) < len(providers):
# Probiere nächsten Provider im Failover-Fall
continue
logger.info(
f"Anfrage #{self.request_count} | "
f"Versuch {attempt + 1}/{self.config.max_retries} | "
f"Provider: {provider.name} ({provider.latency_ms:.0f}ms)"
)
async with aiohttp.ClientSession() as session:
try:
self.request_count += 1
result = await self._call_provider(session, provider, context)
if attempt > 0:
logger.info(f"✓ Failover erfolgreich zu {provider.name}")
if on_failover:
on_failover(provider_tried[-1], provider.name)
return result
except aiohttp.ClientResponseError as e:
last_error = e
logger.warning(
f"✗ {provider.name} fehlgeschlagen: {e.status} {e.message}"
)
provider_tried.append(provider.name)
# Bei Rate Limit oder Server Error: Failover sofort
if e.status in [429, 500, 502, 503, 504]:
if on_failover:
self.failover_count += 1
continue
# Exponential Backoff für andere Fehler
if attempt < self.config.max_retries - 1:
delay = await self._calculate_delay(attempt)
logger.info(f"Warte {delay*1000:.0f}ms vor Retry...")
await asyncio.sleep(delay)
except asyncio.TimeoutError:
last_error = Exception("Timeout")
logger.warning(f"✗ {provider.name} Timeout nach {context.timeout}s")
provider_tried.append(provider.name)
continue
except Exception as e:
last_error = e
logger.error(f"✗ Unerwarteter Fehler: {e}")
provider_tried.append(provider.name)
continue
# Alle Provider und Retries erschöpft
raise RuntimeError(
f"Alle Provider fehlgeschlagen nach {len(provider_tried)} Versuchen. "
f"Letzter Fehler: {last_error}"
)
async def main():
"""Beispiel: HolySheep Failover-Client verwenden"""
# Health Probe initialisieren
health_probe = HolySheepHealthProbe(
api_key="YOUR_HOLYSHEEP_API_KEY",
check_interval=5
)
# Failover-Client erstellen
client = HolySheepFailoverClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
health_probe=health_probe,
config=FailoverConfig(
max_retries=3,
base_delay_ms=100,
exponential_base=2.0
)
)
# Failover-Callback
def on_failover(from_provider: str, to_provider: str):
print(f"🔄 FAILOVER: {from_provider} → {to_provider}")
# Chat-Completion mit Failover
try:
response = await client.chat_completion(
model="deepseek-v3.2", # Wird automatisch an besten Provider weitergeleitet
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre was Disaster Recovery für AI Agents bedeutet."}
],
temperature=0.7,
max_tokens=500,
on_failover=on_failover
)
print(f"\n✅ Antwort erhalten (Token: {response.get('usage', {}).get('total_tokens', 'N/A')})")
print(f"Modell: {response.get('model')}")
print(f"Antwort: {response['choices'][0]['message']['content'][:200]}...")
except RuntimeError as e:
print(f"❌ Alle Provider ausgefallen: {e}")
if __name__ == "__main__":
asyncio.run(main())Circuit Breaker Pattern
Für besonders kritische Systeme empfehle ich das Circuit Breaker Pattern, das verhindert, dass Sie bei einem komplett ausgefallenen Provider weiterhin Anfragen verschwenden.
import time
from dataclasses import dataclass, field
from typing import Dict
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal, Requests durchlassen
OPEN = "open" # Blockiert, keine Requests
HALF_OPEN = "half_open" # Test-Request, um Recovery zu prüfen
@dataclass
class CircuitBreaker:
"""
Circuit Breaker für HolySheep Provider
Verhindert Kaskadenausfälle durch automatische Isolation
"""
name: str
failure_threshold: int = 5 # Fehler bis OPEN
success_threshold: int = 3 # Erfolge in HALF_OPEN bis CLOSED
timeout_seconds: float = 30.0 # Zeit bis HALF_OPEN
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: float = 0.0
last_state_change: float = field(default_factory=time.time)
def record_success(self):
"""Erfolg recorded - Circuit Counter aktualisiert"""
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self._transition_to(CircuitState.CLOSED)
elif self.state == CircuitState.CLOSED:
self.failure_count = max(0, self.failure_count - 1)
def record_failure(self):
"""Fehler recorded"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self._transition_to(CircuitState.OPEN)
elif self.state == CircuitState.CLOSED:
if self.failure_count >= self.failure_threshold:
self._transition_to(CircuitState.OPEN)
def _transition_to(self, new_state: CircuitState):
"""Zustandsübergang mit Logging"""
old_state = self.state
self.state = new_state
self.last_state_change = time.time()
if new_state == CircuitState.CLOSED:
self.failure_count = 0
self.success_count = 0
elif new_state == CircuitState.OPEN:
self.success_count = 0
print(f"Circuit Breaker '{self.name}': {old_state.value} → {new_state.value}")
def can_execute(self) -> bool:
"""Prüft ob Request ausgeführt werden darf"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Timeout erreicht? → HALF_OPEN
if time.time() - self.last_failure_time >= self.timeout_seconds:
self._transition_to(CircuitState.HALF_OPEN)
return True
return False
# HALF_OPEN: Nur ein Test-Request erlaubt
return True
def get_status(self) -> dict:
"""Gibt aktuellen Status zurück"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"success_count": self.success_count,
"time_in_state": time.time() - self.last_state_change
}
class HolySheepCircuitBreakerManager:
"""
Verwaltet Circuit Breaker für alle Provider
"""
def __init__(self):
self.breakers: Dict[str, CircuitBreaker] = {}
def get_breaker(self, provider_name: str) -> CircuitBreaker:
if provider_name not in self.breakers:
self.breakers[provider_name] = CircuitBreaker(
name=provider_name,
failure_threshold=5,
timeout_seconds=30.0
)
return self.breakers[provider_name]
def get_available_providers(self) -> list:
"""Gibt nur Provider zurück, deren Circuit nicht OPEN ist"""
return [
name for name, breaker in self.breakers.items()
if breaker.can_execute()
]
def report_status(self):
"""Gibt Status aller Circuit Breaker aus"""
print("\n=== Circuit Breaker Status ===")
for name, breaker in self.breakers.items():
status = breaker.get_status()
emoji = "🟢" if status["state"] == "closed" else "🔴" if status["state"] == "open" else "🟡"
print(f"{emoji} {name}: {status['state']} (Failures: {status['failure_count']})")
Beispiel: Circuit Breaker testen
if __name__ == "__main__":
manager = HolySheepCircuitBreakerManager()
# Simuliere Fehler
breaker = manager.get_breaker("openai")
for i in range(7):
breaker.record_failure()
print(f"Nach Fehler {i+1}: {breaker.state.value}")
print(f"\nVerfügbar: {manager.get_available_providers()}")
# Warte auf Timeout-Simulation (in echtem Code würde hier time.sleep stehen)
# breaker.last_failure_time = time.time() - 31
# print(f"Nach Timeout: {breaker.can_execute()}")Häufige Fehler und Lösungen
1. Fehler: "Rate Limit Exceeded" bei allen Providern
# PROBLEM: Alle Provider geben 429 zurück
LÖSUNG: Request-Queue mit Priority und Backpressure
import asyncio
from typing import List
from dataclasses import dataclass
import time
@dataclass
class QueuedRequest:
priority: int # 1 = highest
created_at: float
future: asyncio.Future
model: str
messages: List[dict]
class RequestQueue:
"""
Priority Queue für AI-Requests bei Rate Limit
Verhindert Überlastung und priorisiert wichtige Requests
"""
def __init__(self, max_concurrent: int = 10):
self.max_concurrent = max_concurrent
self.queue: List[QueuedRequest] = []
self.active_requests = 0
self.lock = asyncio.Lock()
async def enqueue(
self,
model: str,
messages: List[dict],
priority: int = 5
) -> str:
"""Reiht Request in Queue ein und gibt Request-ID zurück"""
future = asyncio.Future()
request = QueuedRequest(
priority=priority,
created_at=time.time(),
future=future,
model=model,
messages=messages
)
async with self.lock:
self.queue.append(request)
# Sortiere nach Priority (niedrigste zuerst) und Alter
self.queue.sort(key=lambda x: (x.priority, x.created_at))
# Warte auf Ausführung
result = await future
return result
async def get_next(self) -> QueuedRequest:
"""Holt nächsten Request wenn Slot verfügbar"""
while True:
async with self.lock:
if self.active_requests < self.max_concurrent and self.queue:
self.active_requests += 1
return self.queue.pop(0)
await asyncio.sleep(0.1)
def complete(self):
"""Markiert Request als abgeschlossen"""
async with self.lock:
self.active_requests -= 12. Fehler: Provider antwortet, aber mit schlechter Qualität
# PROBLEM: DeepSeek V3.2 antwortet korrekt aber mit niedriger Qualität
LÖSUNG: Multi-Stage Validation und Auto-Upgrade
class QualityAwareFailover:
"""
Failover-Strategie die auch Antwortqualität berücksichtigt
"""
QUALITY_THRESHOLDS = {
"deepseek-v3.2": 0.6, # Niedrigere Schwelle (günstiger)
"gemini-2.5-flash": 0.7,
"claude-sonnet-4.5": 0.8, # Höhere Qualität erwartet
"gpt-4.1": 0.8
}
def validate_response(self, response: str, model: str) -> bool:
"""Prüft ob Antwortqualität ausreichend ist"""
# Minimale Prüfungen
if not response or len(response) < 20:
return False
# Hier könnten komplexere Quality Checks eingebaut werden:
# - Halluzinations-Detection
# - Syntax-Validierung
# - Sentiment-Analyse
threshold = self.QUALITY_THRESHOLDS.get(model, 0.7)
# Einfacher Qualitätsscore basierend auf Struktur
score = 0.5
if response.count(".") > 3:
score += 0.2
if response.count("\n") > 2:
score += 0.2
if len(response) > 100:
score += 0.1
return score >= threshold
async def smart_failover(
self,
client: HolySheepFailoverClient,
model: str,
messages: List[dict]
) -> dict:
"""
Führt Failover durch und prüft Qualität
Upgradet automatisch wenn nötig
"""
models_to_try = [model]
# Upgrade-Pfad basierend auf aktuellem Modell
upgrade_paths = {
"deepseek-v3.2": ["gemini-2.5-flash", "claude-sonnet-4.5", "gpt-4.1"],
"gemini-2.5-flash": ["claude-sonnet-4.5", "gpt-4.1"],
"claude-sonnet-4.5": ["gpt-4.1"]
}
if model in upgrade_paths:
models_to_try.extend(upgrade_paths[model])
for try_model in models_to_try:
response = await client.chat_completion(
model=try_model,
messages=messages
)
content = response["choices"][0]["message"]["content"]
if self.validate_response(content, try_model):
return response
# Letzter Versuch mit bester Qualität
return await client.chat_completion(
model="claude-sonnet-4.5",
messages=messages
)3. Fehler: Token-Limit erreicht bei langen Konversationen
# PROBLEM: Kontext-Fenster wird bei langen Chats überschritten
LÖSUNG: Smart Context Window Management
class ContextWindowManager:
"""
Verwaltet Kontext-Fenster und komprimiert bei Bedarf
"""
MAX_TOKENS = {
"deepseek-v3.2": 64000,
"gemini-2.5-flash": 32000,
"claude-sonnet-4.5": 200000,
"gpt-4.1": 128000
}
RESERVE_TOKENS = 2000 # Puffer für System-Prompt
def __init__(self, model: str):
self.model = model
self.max_context = self.MAX_TOKENS.get(model, 4000)
def estimate_tokens(self, messages: List[dict]) -> int:
"""Schätzt Token-Anzahl (vereinfacht: 1 Token ≈ 4 Zeichen)"""
total = 0
for msg in messages:
total += len(str(msg.get("content", ""))) // 4
total += 20 # Overhead pro Message
return total
def truncate_history(
self,
messages: List[dict],
max_tokens: Optional[int] = None
) -> List[dict]:
"""Kürzt Konversation wenn nötig, behält aber System-Prompt"""
if max_tokens is None:
max_tokens = self.max_context - self.RESERVE_TOKENS
current_tokens = self.estimate_tokens(messages)
if current_tokens <= max_tokens:
return messages
# System-Prompt behalten
system_prompt = None
if messages and messages[0].get("role") == "system":
system_prompt = messages[0]
messages = messages[1:]
# Nachrichten vom Ende entfernen (älteste zuerst)
truncated = messages
while self.estimate_tokens(truncated) > max_tokens and len(truncated) > 1:
truncated = truncated[1:]
# Zusammenfassung der entfernten Messages hinzufügen
removed_count = len(messages) - len(truncated)
if removed_count > 0 and truncated:
summary = f"[{removed_count} frühere Nachrichten wurden gekürzt]"
truncated.insert(