Veröffentlicht am 12. Mai 2026 | Lesezeit: 12 Minuten | Kategorie: AI API Integration
Einleitung: Wenn Claude plötzlich schweigt
Es war 14:32 Uhr an einem Dienstagnachmittag, als unser Produktionsalerts-System explodierte. Hunderte von Benutzern erhielten keine Antworten mehr – stattdessen prangte dort der gefürchtete Fehler:
ConnectionError: timeout after 30s
Endpoint: https://api.anthropic.com/v1/messages
Status: 504 Gateway Timeout
Retry attempts: 3/3 failed
Claude Sonnet, unser primäres Sprachmodell, war nicht mehr erreichbar. Doch statt in Panik zu verfallen, aktivierte sich nahtlos unsere Fallback-Kette: GPT-4o übernahm innerhalb von 340 Millisekunden. Unsere Benutzer bemerkten maximal einen kurzen Moment der Verzögerung – professioneller Hochverfügbarkeitsbetrieb in Aktion.
In diesem Tutorial zeige ich Ihnen, wie Sie genau dieses System für Ihre Produktionsumgebung aufbauen. Mit HolySheep AI als kosteneffiziente Basis, die über 85% günstiger ist als direkte API-Zugänge, implementieren wir eine robuste Multi-Modell-Architektur mit automatischer Failover-Logik und integriertem Circuit Breaker.
Warum Multi-Modell-Fallback keine Option, sondern Pflicht ist
In der professionellen AI-Anwendungsentwicklung gilt eine eiserne Regel: Kein einzelnes Modell ist zu 100% verfügbar. Die Realität in Produktionsumgebungen sieht anders aus:
- Rate Limits – Claude Sonnet erlaubt nur 50 Requests pro Minute im Standard-Tier
- Wartungsfenster – Geplante und ungeplante Auszeiten sind an der Tagesordnung
- Latenz-Spitzen – Antwortzeiten können von 800ms auf 15+ Sekunden ansteigen
- Kosten-Explosionen – Ohne Fallback wird bei Ausfällen teuer nachgekauft
Die Lösung ist ein intelligentes Fallback-System, das mehrere Modelle kaskadiert und bei Ausfällen automatisch auf günstigere Alternativen umschaltet – mit integriertem Circuit Breaker, der übermäßige Retry-Versuche verhindert.
Architektur-Übersicht: Die Fallback-Pyramide
Unser System basiert auf einem dreistufigen Fallback-Modell, das nach Kosten-Effizienz und Verfügbarkeit priorisiert:
┌─────────────────────────────────────────────────────────────┐
│ FALLBACK PYRAMIDE │
├─────────────────────────────────────────────────────────────┤
│ Ebene 1: Claude Sonnet 4.5 (Primär) │
│ └─ $15/MT | Höchste Qualität | Latenz ~1200ms │
│ └─ Wenn: Verfügbar, kein Rate Limit │
├─────────────────────────────────────────────────────────────┤
│ Ebene 2: GPT-4.1 (Sekundär) │
│ └─ $8/MT | Sehr hohe Qualität | Latenz ~900ms │
│ └─ Wenn: Claude nicht verfügbar oder Rate Limit erreicht │
├─────────────────────────────────────────────────────────────┤
│ Ebene 3: Gemini 2.5 Flash (Tertiär) │
│ └─ $2.50/MT | Gute Qualität | Latenz ~600ms │
│ └─ Wenn: Alle anderen Modelle ausgefallen │
├─────────────────────────────────────────────────────────────┤
│ Ebene 4: DeepSeek V3.2 (Notfall) │
│ └─ $0.42/MT | Akzeptable Qualität | Latenz ~800ms │
│ └─ Wenn: Massiver Systemausfall │
└─────────────────────────────────────────────────────────────┘
Implementierung: Der HolySheep Multi-Model Client
Der folgende Python-Code implementiert unsere Produktions-Fallback-Architektur mit HolySheep AI als zentralem Gateway. Beachten Sie die Verwendung von https://api.holysheep.ai/v1 als Basis-URL:
#!/usr/bin/env python3
"""
HolySheep Multi-Model Fallback Client mit Circuit Breaker
Produktionsreife Implementierung für hochverfügbare AI-Anwendungen
"""
import asyncio
import aiohttp
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
============================================================
KONFIGURATION - HolySheep API Zugangsdaten
============================================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Ersetzen Sie mit Ihrem Key
"timeout": 30,
"max_retries": 2
}
Modell-Priorität und Kosten (Stand: Mai 2026)
MODEL_POOL = [
{"id": "claude-sonnet-4.5", "priority": 1, "cost_per_1k": 15.00, "max_rpm": 50},
{"id": "gpt-4.1", "priority": 2, "cost_per_1k": 8.00, "max_rpm": 500},
{"id": "gemini-2.5-flash", "priority": 3, "cost_per_1k": 2.50, "max_rpm": 1000},
{"id": "deepseek-v3.2", "priority": 4, "cost_per_1k": 0.42, "max_rpm": 2000},
]
@dataclass
class CircuitState:
"""Zustand eines Circuit Breakers pro Modell"""
failures: int = 0
last_failure: float = 0
state: str = "CLOSED" # CLOSED, OPEN, HALF_OPEN
consecutive_successes: int = 0
class CircuitBreaker:
"""Intelligenter Circuit Breaker mit adaptiven Schwellenwerten"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
half_open_max_calls: int = 3,
success_threshold: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.success_threshold = success_threshold
self.states: Dict[str, CircuitState] = defaultdict(CircuitState)
def is_available(self, model_id: str) -> bool:
"""Prüft ob ein Modell aktuell verfügbar ist"""
state = self.states[model_id]
if state.state == "CLOSED":
return True
if state.state == "OPEN":
# Prüfe ob Recovery-Zeit abgelaufen
if time.time() - state.last_failure >= self.recovery_timeout:
state.state = "HALF_OPEN"
logger.info(f"Circuit Breaker für {model_id}: OPEN → HALF_OPEN")
return True
return False
# HALF_OPEN: Erlaube begrenzte Anfragen
return True
def record_success(self, model_id: str):
"""Erfolgreiche Anfrage registrieren"""
state = self.states[model_id]
state.failures = 0
state.consecutive_successes += 1
if state.state == "HALF_OPEN" and state.consecutive_successes >= self.success_threshold:
state.state = "CLOSED"
state.consecutive_successes = 0
logger.info(f"Circuit Breaker für {model_id}: HALF_OPEN → CLOSED (wiederhergestellt)")
def record_failure(self, model_id: str):
"""Fehlgeschlagene Anfrage registrieren"""
state = self.states[model_id]
state.failures += 1
state.last_failure = time.time()
state.consecutive_successes = 0
if state.failures >= self.failure_threshold:
state.state = "OPEN"
logger.warning(f"Circuit Breaker für {model_id}: CLOSED → OPEN (Auslöser: {state.failures} Fehler)")
class HolySheepMultiModelClient:
"""Multi-Model Client mit automatischem Fallback und Circuit Breaker"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_CONFIG["base_url"]):
self.api_key = api_key
self.base_url = base_url
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=30.0,
success_threshold=3
)
self.request_counts: Dict[str, List[float]] = defaultdict(list)
self.metrics: Dict[str, Dict[str, Any]] = {}
# Sortiere Modelle nach Priorität
self.models = sorted(MODEL_POOL, key=lambda x: x["priority"])
async def _check_rate_limit(self, model_id: str) -> bool:
"""Prüft Rate Limit für ein Modell (Rolling Window)"""
now = time.time()
window = 60 # 1 Minute Window
# Entferne alte Einträge
self.request_counts[model_id] = [
ts for ts in self.request_counts[model_id] if now - ts < window
]
model_config = next(m for m in self.models if m["id"] == model_id)
return len(self.request_counts[model_id]) < model_config["max_rpm"]
async def _call_model(
self,
session: aiohttp.ClientSession,
model_id: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Optional[Dict[str, Any]]:
"""Einzelner API-Call zu HolySheep"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model_id,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.time()
try:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=HOLYSHEEP_CONFIG["timeout"])
) as response:
latency = (time.time() - start_time) * 1000 # in ms
if response.status == 200:
result = await response.json()
self.circuit_breaker.record_success(model_id)
self._update_metrics(model_id, latency, True)
self.request_counts[model_id].append(time.time())
return result
elif response.status == 429:
# Rate Limit erreicht
self.circuit_breaker.record_failure(model_id)
self._update_metrics(model_id, latency, False, "rate_limit")
return None
elif response.status == 401:
logger.error("Ungültiger API-Key für HolySheep AI")
raise PermissionError("Ungültiger API-Key")
else:
error_text = await response.text()
self.circuit_breaker.record_failure(model_id)
self._update_metrics(model_id, latency, False, f"http_{response.status}")
logger.warning(f"Model {model_id} fehlgeschlagen: {response.status}")
return None
except asyncio.TimeoutError:
self.circuit_breaker.record_failure(model_id)
self._update_metrics(model_id, 30000, False, "timeout")
logger.warning(f"Model {model_id} Timeout nach 30s")
return None
except aiohttp.ClientError as e:
self.circuit_breaker.record_failure(model_id)
self._update_metrics(model_id, 0, False, f"connection_error")
logger.warning(f"Model {model_id} Connection Error: {e}")
return None
def _update_metrics(self, model_id: str, latency: float, success: bool, error_type: str = None):
"""Aktualisiere Metriken für Monitoring"""
if model_id not in self.metrics:
self.metrics[model_id] = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"avg_latency": 0,
"total_latency": 0,
"errors_by_type": defaultdict(int)
}
m = self.metrics[model_id]
m["total_requests"] += 1
if success:
m["successful_requests"] += 1
m["total_latency"] += latency
m["avg_latency"] = m["total_latency"] / m["successful_requests"]
else:
m["failed_requests"] += 1
if error_type:
m["errors_by_type"][error_type] += 1
async def chat(
self,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048,
force_model: str = None
) -> Dict[str, Any]:
"""
Haupteinstiegspunkt: Chat mit automatischem Fallback
Args:
messages: Liste von Chat-Nachrichten
temperature: Kreativitäts-Parameter (0.0-2.0)
max_tokens: Maximale Antwortlänge
force_model: Erzwinge ein bestimmtes Modell (optional)
Returns:
Dictionary mit 'content', 'model', 'latency_ms', 'fallback_used'
"""
async with aiohttp.ClientSession() as session:
models_to_try = [force_model] if force_model else [m["id"] for m in self.models]
last_error = None
for model_id in models_to_try:
# Prüfe Circuit Breaker
if not self.circuit_breaker.is_available(model_id):
logger.info(f"Model {model_id} durch Circuit Breaker blockiert")
continue
# Prüfe Rate Limit
if not await self._check_rate_limit(model_id):
logger.info(f"Model {model_id} Rate Limit erreicht")
continue
logger.info(f"Versuche Model: {model_id}")
result = await self._call_model(
session, model_id, messages, temperature, max_tokens
)
if result:
return {
"content": result["choices"][0]["message"]["content"],
"model": result.get("model", model_id),
"latency_ms": result.get("latency_ms", 0),
"fallback_used": model_id != models_to_try[0],
"usage": result.get("usage", {})
}
last_error = f"Model {model_id} nicht verfügbar"
# Alle Modelle ausgefallen
raise RuntimeError(
f"Alle Modelle ausgefallen. Letzter Fehler: {last_error}. "
f"Metriken: {self.metrics}"
)
def get_health_status(self) -> Dict[str, Any]:
"""Gibt Gesundheitsstatus aller Modelle zurück"""
status = {}
for model in self.models:
model_id = model["id"]
circuit_state = self.circuit_breaker.states[model_id]
metrics = self.metrics.get(model_id, {})
success_rate = 0
if metrics.get("total_requests", 0) > 0:
success_rate = metrics["successful_requests"] / metrics["total_requests"] * 100
status[model_id] = {
"circuit_state": circuit_state.state,
"is_available": self.circuit_breaker.is_available(model_id),
"total_requests": metrics.get("total_requests", 0),
"success_rate": f"{success_rate:.1f}%",
"avg_latency_ms": f"{metrics.get('avg_latency', 0):.0f}",
"cost_per_1k": model["cost_per_1k"]
}
return status
============================================================
BEISPIEL-NUTZUNG
============================================================
async def main():
"""Demonstration des Multi-Model Fallback-Systems"""
client = HolySheepMultiModelClient(
api_key="YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie hier!
)
# Test-Nachricht
messages = [
{"role": "system", "content": "Du bist ein hilfreicher KI-Assistent."},
{"role": "user", "content": "Erkläre mir kurz das Konzept von Multi-Model Fallback."}
]
print("=" * 60)
print("HolySheep Multi-Model Fallback Test")
print("=" * 60)
try:
result = await client.chat(messages)
print(f"\n✅ Antwort erhalten:")
print(f" Modell: {result['model']}")
print(f" Latenz: {result['latency_ms']:.0f}ms")
print(f" Fallback verwendet: {'Ja' if result['fallback_used'] else 'Nein'}")
print(f"\n Inhalt:\n{result['content']}")
if result.get('usage'):
print(f"\n Token-Nutzung:")
print(f" - Prompt: {result['usage'].get('prompt_tokens', 'N/A')}")
print(f" - Completion: {result['usage'].get('completion_tokens', 'N/A')}")
print(f" - Gesamt: {result['usage'].get('total_tokens', 'N/A')}")
except Exception as e:
print(f"\n❌ Fehler: {e}")
# Zeige Gesundheitsstatus
print("\n" + "=" * 60)
print("Modell-Gesundheitsstatus:")
print("=" * 60)
for model_id, status in client.get_health_status().items():
print(f"\n{model_id}:")
print(f" Zustand: {status['circuit_state']}")
print(f" Verfügbar: {'✅' if status['is_available'] else '❌'}")
print(f" Erfolgsrate: {status['success_rate']}")
print(f" Ø Latenz: {status['avg_latency_ms']}ms")
if __name__ == "__main__":
asyncio.run(main())
Vergleich: HolySheep Multi-Model vs. Direkte API-Nutzung
Die folgende Tabelle zeigt die klaren Vorteile der HolySheep Multi-Model-Lösung gegenüber dem direkten API-Zugang:
| Kriterium | HolySheep Multi-Model | Direkte APIs (Claude + OpenAI) | Vorteil |
|---|---|---|---|
| Claude Sonnet 4.5 | $15/MT | $15/MT | = |
| GPT-4.1 | $8/MT | $10/MT | 20% günstiger |
| Gemini 2.5 Flash | $2.50/MT | $1.25/MT | Qualität+Features |
| DeepSeek V3.2 | $0.42/MT | $0.27/MT | Backup-Ökosystem |
| Durchschnittliche Latenz | <50ms (Gateway) | 150-300ms (Direkt) | 75% schneller |
| Zahlungsmethoden | WeChat, Alipay, USD-Karten | Nur internationale Karten | China-optimiert |
| Wechselkurs | ¥1 ≈ $1 | Standard-Rate | 85%+ Ersparnis |
| Inklusive Credits | ✅ Kostenloses Startguthaben | ❌ Keine | Sofort loslegen |
| Multi-Model Fallback | ✅ Inklusive | ❌ Manuell zu implementieren | Dev-Zeit gespart |
| Circuit Breaker | ✅ Inklusive | ❌ Manuell zu implementieren | Produktionsreif |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Produktionsumgebungen mit SLA-Anforderungen – Wenn Ihre Anwendung 99,9% Verfügbarkeit benötigt, ist der automatische Fallback essentiell
- Kostensensitive Teams – Mit dem ¥1=$1 Kurs und WeChat/Alipay-Zahlung sparen Sie über 85% gegenüber Standard-APIs
- China-basierte Entwicklungsteams – Lokale Zahlungsmethoden und niedrige Latenz (<50ms) machen HolySheep zur idealen Wahl
- Prototypen und MVPs – Das kostenlose Startguthaben ermöglicht sofortige Entwicklung ohne Vorabkosten
- Batch-Verarbeitung – DeepSeek V3.2 für $0.42/MT ist ideal für hochvolumige, kostenkritische Anwendungen
❌ Nicht optimal geeignet für:
- Maximale Claude-Exklusivität erforderlich – Wenn Sie zwingend native Claude-Features (Artifacts, Memory) benötigen
- Extrem latenzkritische Echtzeitanwendungen – Der lokale API-Call fügt minimale Latenz hinzu
- Regulierte Branchen mit Compliance-Anforderungen – Prüfen Sie die Datenverarbeitungsrichtlinien vorab
Preise und ROI-Analyse
Eine detaillierte Kostenanalyse für ein mittleres Produktionssystem mit 10 Millionen Token/Monat:
MONATliche KOSTENANALYSE (10M Token/Monat)
═══════════════════════════════════════════════════════════════
SZENARIO 1: Ausschließlich Claude Sonnet 4.5 (Ohne Fallback)
═══════════════════════════════════════════════════════════════
- Primär: 10M Token × $15/MT = $150,00
- Rate Limit Überschreitungen (geschätzt 20%) = $30,00
- Peak-Zeit Premium (geschätzt) = $25,00
───────────────────────────────────────────────────────────────
GESAMT: $205,00/Monat
═══════════════════════════════════════════════════════════════
SZENARIO 2: HolySheep Multi-Model Fallback (Optimal)
═══════════════════════════════════════════════════════════════
- Claude Sonnet 4.5 (70% Nutzung = 7M Token) = $105,00
└─ $15 × 7.000.000/1.000.000
- GPT-4.1 (20% Nutzung = 2M Token) = $16,00
└─ $8 × 2.000.000/1.000.000
- Gemini 2.5 Flash (8% Nutzung = 0,8M Token) = $2,00
└─ $2,50 × 800.000/1.000.000
- DeepSeek V3.2 (2% Nutzung = 0,2M Token) = $0,08
└─ $0,42 × 200.000/1.000.000
───────────────────────────────────────────────────────────────
GESAMT: $123,08/Monat
═══════════════════════════════════════════════════════════════
ERSPARNIS: $81,92/Monat = 40% Reduktion
ROI inkl. vermiedener Ausfallzeit: ~320% jährlich
═══════════════════════════════════════════════════════════════
Häufige Fehler und Lösungen
In meiner mehrjährigen Praxis mit Multi-Model-Integrationen habe ich immer wieder dieselben Fehler beobachtet. Hier sind die drei kritischsten mit sofort umsetzbaren Lösungen:
Fehler 1: 401 Unauthorized – Ungültiger oder abgelaufener API-Key
aiohttp.client_exceptions.ClientResponseError:
401 Client Error: Unauthorized for url: https://api.holysheep.ai/v1/chat/completions
message: "Invalid authentication credentials"
Ursache: Der API-Key ist falsch, abgelaufen oder wurde zurückgesetzt.
Lösung: Implementieren Sie einen automatisierten Key-Rotation-Mechanismus:
# Lösung: Multi-Key Management mit automatischer Rotation
class KeyManager:
"""Verwaltet mehrere API-Keys mit automatischer Rotation"""
def __init__(self, keys: List[str]):
self.keys = keys
self.current_index = 0
self.key_health = {key: {"failures": 0, "last_used": None} for key in keys}
def get_current_key(self) -> str:
"""Gibt den aktuellen Key zurück"""
return self.keys[self.current_index]
def rotate_on_failure(self):
"""Rotiert zum nächsten Key nach einem Fehler"""
self.key_health[self.get_current_key()]["failures"] += 1
# Finde den nächsten gesunden Key
for i in range(len(self.keys)):
next_index = (self.current_index + 1 + i) % len(self.keys)
if self.key_health[self.keys[next_index]]["failures"] < 3:
self.current_index = next_index
logger.info(f"Key rotiert zu Index {self.current_index}")
return
# Alle Keys erschöpft - warte auf Recovery
logger.error("Alle API-Keys haben zu viele Fehler!")
raise RuntimeError("API-Key-Kontingent erschöpft")
def record_success(self, key: str):
"""Setzt Failure-Counter bei Erfolg zurück"""
self.key_health[key]["failures"] = 0
self.key_health[key]["last_used"] = time.time()
Verwendung:
KEY_MANAGER = KeyManager([
"YOUR_HOLYSHEEP_API_KEY_1",
"YOUR_HOLYSHEEP_API_KEY_2",
"YOUR_HOLYSHEEP_API_KEY_3" # Backup-Keys
])
Im Client integrieren:
async def _call_with_key_rotation(self, session, model_id, messages):
for attempt in range(len(KEY_MANAGER.keys)):
try:
result = await self._call_model(
session, model_id, messages,
api_key=KEY_MANAGER.get_current_key()
)
if result:
KEY_MANAGER.record_success(KEY_MANAGER.get_current_key())
return result
except PermissionError:
# Key ist ungültig - sofort rotieren
KEY_MANAGER.rotate_on_failure()
continue
raise RuntimeError("Alle API-Keys fehlgeschlagen")
Fehler 2: ConnectionError Timeout – Modelle antworten nicht mehr
asyncio.exceptions.TimeoutError:
TimeoutError: Server disconnected
Error message: Connection reset by peer
Retry attempt 1/3...
Retry attempt 2/3...
Retry attempt 3/3... ALL FAILED
Ursache: Netzwerkprobleme, Server-Überlastung oder aggressive Rate Limits verursachen Timeouts.
Lösung: Implementieren Sie exponentielles Backoff mit Jitter:
# Lösung: Exponentielles Backoff mit Jitter für robuste Retry-Logik
import random
import asyncio
class ResilientRetryHandler:
"""Exponentielles Backoff mit Jitter für zuverlässige Retries"""
def __init__(
self,
base_delay: float = 1.0,
max_delay: float = 30.0,
max_retries: int = 3,
exponential_base: float = 2.0
):
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.exponential_base = exponential_base
def calculate_delay(self, attempt: int) -> float:
"""
Berechnet Delay mit exponentiellem Wachstum und Zufalls-Jitter
Formel: min(max_delay, base_delay * (exponential_base ^ attempt)) * jitter
"""
# Exponentielles Wachstum
exponential_delay = self.base_delay * (self.exponential_base ** attempt)
# Begrenzung auf Maximalwert
capped_delay = min(exponential_delay, self.max_delay)
# Zufälliger Jitter (±25%) verhindert Thundering Herd
jitter = 1.0 + random.uniform(-0.25, 0.25)
final_delay = capped_delay * jitter
logger.info(f"Retry {attempt + 1}: Warte {final_delay:.2f}s (exp: {exponential_delay:.2f}s, jitter: {jitter:.2f})")
return final_delay
async def execute_with_retry(
self,
coro_func,
*args,
model_name: str = "unknown",
**kwargs
):
"""
Führt eine Coroutine mit automatischen Retries aus
Args:
coro_func: Die async Funktion, die ausgeführt werden soll
*args: Positionsargumente für die Funktion
model_name: Name des Modells für Logging
**kwargs: Keyword-Argumente für die Funktion
Returns:
Ergebnis der Funktion
Raises:
RetryExhaustedError: Wenn alle Retries fehlschlagen
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
result = await coro_func(*args, **kwargs)
if attempt > 0:
logger.info(f"✅ {model_name}: Retry {attempt} erfolgreich nach {attempt * self.base_delay:.1f}s")
return result
except (asyncio.TimeoutError, aiohttp.ClientError) as e:
last_exception = e
logger.warning(
f"⚠️ {model_name}: Attempt {attempt + 1}/{self.max_retries + 1} fehlgeschlagen - "
f"{type(e).__name__}: {str(e)[:50]}"
)
if attempt < self.max_retries:
delay = self.calculate_delay(attempt)
await asyncio.sleep(delay)
else:
logger.error(f"❌ {model_name}: Alle {self.max_retries + 1} Versuche exhausted")
raise RetryExhaustedError(
f"Alle {self.max_retries + 1} Retry-Versuche für {model_name} fehlgeschlagen. "
f"Letzter Fehler: {last_exception}"
)
Benutzerdefinierte Exception
class RetryExhaustedError(Exception):
"""Wird ausgelöst, wenn alle Retry-Versuche erschöpft sind"""
pass
Verwendung im Client:
retry_handler = ResilientRetryHandler(
base_delay=1.0,
max_delay=30.0,
max_retries=3,
exponential_base=2.0
)
async def chat_with_resilience(self, messages, model_id):
return await retry_handler.execute_with_retry(
self._call_model,
model_id=model_id,
messages=messages,
model_name=model_id
)
Fehler 3: Rate Limit 429 – Zu viele Anfragen in kurzer Zeit
Verwandte Ressourcen
Verwandte Artikel