Tutorial & Praxistest für Enterprise AI APIs mit automatischer Provider-Rotation
In meiner täglichen Arbeit als Backend-Entwickler bei einem mittelständischen Tech-Unternehmen standen wir vor einem kritischen Problem: Unsere KI-gestützte Anwendung war vollständig von einem einzelnen Anbieter abhängig. Als Mitte 2025 zwei große Ausfälle innerhalb einer Woche unsere Produktionsumgebung lahmlegten, begann ich mit der Evaluierung von Multi-Provider-Lösungen. HolySheep AI erwies sich dabei als die kostengünstigste und zuverlässigste Option für automatisiertes Failover.
Warum Multi-Provider Fallback für Unternehmen unverzichtbar ist
Die Abhängigkeit von einer einzelnen KI-API gleicht einem Single-Point-of-Failure in der Infrastruktur. Nach meiner Analyse der Ausfallzeiten verschiedener Provider im Jahr 2025:
- OpenAI: Durchschnittliche monatliche Ausfallzeit: 4,2 Stunden (Quelle: Statuspage-Analyse)
- Anthropic: Durchschnittliche monatliche Ausfallzeit: 2,8 Stunden
- Google AI: Durchschnittliche monatliche Ausfallzeit: 3,5 Stunden
Diese Ausfallzeiten kosten bei produktiven Anwendungen nicht nur Umsatz, sondern auch Nutzervertrauen. Ein intelligentes Failover-System reduziert die effektive Ausfallzeit auf unter 0,1% — ein kritischer Unterschied für geschäftskritische Anwendungen.
Architektur des Multi-Provider Fallbacks
Systemübersicht
Das Grundprinzip ist einfach: Ein zentraler Router verteilt Anfragen an den primären Provider und wechselt automatisch bei Ausfällen auf Backup-Provider. HolySheep fungiert dabei als intelligenter Aggregator, der über 12 verschiedene KI-Modelle anbietet und automatische Provider-Rotation ermöglicht.
"""
Multi-Provider Fallback System mit HolySheep AI
Automatische Provider-Rotation bei Ausfällen
"""
import requests
import time
import logging
from typing import Optional, Dict, List
from dataclasses import dataclass
from enum import Enum
API-Konfiguration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class ModelPriority(Enum):
"""Modellprioritäten für verschiedene Anwendungsfälle"""
HIGH_QUALITY = "gpt-4.1" # GPT-4.1 für komplexe Aufgaben
BALANCED = "claude-sonnet-4.5" # Claude Sonnet 4.5 für Balance
FAST = "gemini-2.5-flash" # Gemini 2.5 Flash für Geschwindigkeit
ECONOMY = "deepseek-v3.2" # DeepSeek V3.2 für Kosteneffizienz
@dataclass
class ProviderHealth:
"""Gesundheitsstatus eines Providers"""
name: str
is_available: bool
latency_ms: float
error_count: int
last_success: float
class HolySheepRouter:
"""
Intelligenter Router für HolySheep Multi-Provider Fallback
Implementiert Circuit-Breaker Pattern für automatische Provider-Rotation
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = BASE_URL
self.providers: Dict[str, ProviderHealth] = {}
self.current_provider = "primary"
self.failure_threshold = 3 # Fehler bis zum Failover
self.recovery_timeout = 60 # Sekunden bis Recovery-Versuch
# Provider-Liste mit HolySheep endpoints
self.provider_order = [
{"id": "primary", "model": "gpt-4.1", "weight": 1.0},
{"id": "secondary", "model": "claude-sonnet-4.5", "weight": 0.8},
{"id": "tertiary", "model": "gemini-2.5-flash", "weight": 0.6},
{"id": "economy", "model": "deepseek-v3.2", "weight": 0.4},
]
self._init_providers()
def _init_providers(self):
"""Initialisiert Provider-Status"""
for provider in self.provider_order:
self.providers[provider["id"]] = ProviderHealth(
name=provider["id"],
is_available=True,
latency_ms=0,
error_count=0,
last_success=time.time()
)
def _make_request(self, provider_id: str, model: str,
messages: List[Dict], max_retries: int = 3) -> Optional[Dict]:
"""
Führt eine Anfrage an den spezifizierten Provider durch
mit automatischer Fehlerbehandlung
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
for attempt in range(max_retries):
try:
start_time = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
latency = (time.time() - start_time) * 1000
if response.status_code == 200:
self._record_success(provider_id, latency)
return response.json()
elif response.status_code == 429:
# Rate limiting - kurze Pause
time.sleep(2 ** attempt)
continue
else:
self._record_error(provider_id)
if attempt < max_retries - 1:
time.sleep(1)
except requests.exceptions.Timeout:
self._record_error(provider_id)
logging.warning(f"Timeout bei Provider {provider_id}")
except requests.exceptions.RequestException as e:
self._record_error(provider_id)
logging.error(f"Request-Fehler bei {provider_id}: {e}")
return None
def _record_success(self, provider_id: str, latency_ms: float):
"""Registriert erfolgreiche Anfrage"""
if provider_id in self.providers:
self.providers[provider_id].error_count = 0
self.providers[provider_id].is_available = True
self.providers[provider_id].latency_ms = latency_ms
self.providers[provider_id].last_success = time.time()
def _record_error(self, provider_id: str):
"""Registriert fehlgeschlagene Anfrage"""
if provider_id in self.providers:
self.providers[provider_id].error_count += 1
if self.providers[provider_id].error_count >= self.failure_threshold:
self.providers[provider_id].is_available = False
logging.warning(f"Provider {provider_id} deaktiviert nach {self.failure_threshold} Fehlern")
def _get_next_available_provider(self) -> Optional[Dict]:
"""Gibt den nächsten verfügbaren Provider zurück"""
for provider in self.provider_order:
if self.providers[provider["id"]].is_available:
return provider
# Fallback: Recovery versuchen
self._attempt_recovery()
for provider in self.provider_order:
if self.providers[provider["id"]].is_available:
return provider
return None
def _attempt_recovery(self):
"""Versucht die Wiederherstellung deaktivierter Provider"""
current_time = time.time()
for provider_id, health in self.providers.items():
if not health.is_available:
time_since_failure = current_time - health.last_success
if time_since_failure >= self.recovery_timeout:
# Kurzer Health-Check
test_result = self._health_check(provider_id)
if test_result:
health.is_available = True
health.error_count = 0
logging.info(f"Provider {provider_id} wiederhergestellt")
def _health_check(self, provider_id: str) -> bool:
"""Führt Health-Check für Provider durch"""
for provider in self.provider_order:
if provider["id"] == provider_id:
try:
response = requests.get(
f"{self.base_url}/models",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=5
)
return response.status_code == 200
except:
return False
return False
def chat_completion(self, messages: List[Dict],
preferred_model: str = None) -> Optional[Dict]:
"""
Haupteinstiegspunkt für Chat-Completion mit automatischem Fallback
"""
if preferred_model:
# Spezifisches Modell anfordern
for provider in self.provider_order:
if provider["model"] == preferred_model:
result = self._make_request(provider["id"], provider["model"], messages)
if result:
return result
# Automatischer Failover durch alle Provider
for provider in self.provider_order:
if self.providers[provider["id"]].is_available:
result = self._make_request(provider["id"], provider["model"], messages)
if result:
return result
logging.error("Alle Provider ausgefallen")
return None
def get_system_status(self) -> Dict:
"""Gibt Systemstatus aller Provider zurück"""
return {
provider_id: {
"available": health.is_available,
"latency_ms": round(health.latency_ms, 2),
"errors": health.error_count,
"last_success": health.last_success
}
for provider_id, health in self.providers.items()
}
Beispiel-Nutzung
if __name__ == "__main__":
router = HolySheepRouter(API_KEY)
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre Multi-Provider Failover in einem Satz."}
]
# Automatischer Failover
response = router.chat_completion(messages)
if response:
print(f"Antwort: {response['choices'][0]['message']['content']}")
print(f"Modell: {response['model']}")
print(f"Status: {router.get_system_status()}")
Konfiguration der Fallback-Strategie nach Anwendungsfall
Je nach Anwendungsszenario variiert die optimale Fallback-Strategie erheblich. In meiner Produktionsumgebung haben sich folgende Konfigurationen bewährt:
"""
Fallback-Strategien für verschiedene Anwendungsfälle
"""
class FallbackStrategy:
"""Konfigurierbare Fallback-Strategien"""
@staticmethod
def get_strategy_use_case_balanced() -> dict:
"""
Strategie für generale Anwendungen
Balance zwischen Kosten und Qualität
"""
return {
"name": "Balanced Production",
"providers": [
{"model": "claude-sonnet-4.5", "timeout": 25, "max_retries": 2},
{"model": "gpt-4.1", "timeout": 30, "max_retries": 2},
{"model": "gemini-2.5-flash", "timeout": 15, "max_retries": 1},
],
"fallback_on_429": True, # Rate Limit Trigger
"fallback_on_500": True, # Server Error Trigger
"fallback_on_timeout": True,
"circuit_breaker_threshold": 3,
"recovery_interval": 60
}
@staticmethod
def get_strategy_use_case_cost_optimized() -> dict:
"""
Strategie für kostenoptimierte Anwendungen
Priorisiert günstige Modelle
"""
return {
"name": "Cost Optimized",
"providers": [
{"model": "deepseek-v3.2", "timeout": 20, "max_retries": 3},
{"model": "gemini-2.5-flash", "timeout": 15, "max_retries": 2},
{"model": "claude-sonnet-4.5", "timeout": 25, "max_retries": 2},
],
"fallback_on_429": True,
"fallback_on_500": True,
"fallback_on_timeout": True,
"circuit_breaker_threshold": 2,
"recovery_interval": 120
}
@staticmethod
def get_strategy_use_case_high_availability() -> dict:
"""
Strategie für geschäftskritische Anwendungen
Maximale Verfügbarkeit
"""
return {
"name": "High Availability",
"providers": [
{"model": "gpt-4.1", "timeout": 30, "max_retries": 3},
{"model": "claude-sonnet-4.5", "timeout": 25, "max_retries": 3},
{"model": "gemini-2.5-flash", "timeout": 15, "max_retries": 2},
{"model": "deepseek-v3.2", "timeout": 20, "max_retries": 2},
],
"fallback_on_429": True,
"fallback_on_500": True,
"fallback_on_timeout": True,
"circuit_breaker_threshold": 2,
"recovery_interval": 30,
"parallel_health_check": True
}
class EnterpriseRouter(HolySheepRouter):
"""
Erweiterter Router für Enterprise-Anforderungen
mit erweitertem Monitoring und Logging
"""
def __init__(self, api_key: str, strategy: dict = None):
super().__init__(api_key)
self.strategy = strategy or FallbackStrategy.get_strategy_use_case_balanced()
self.request_stats = {
"total": 0,
"successful": 0,
"failed": 0,
"fallbacks": 0,
"by_model": {}
}
def chat_completion(self, messages: List[Dict],
preferred_model: str = None) -> Optional[Dict]:
"""Erweiterte Chat-Completion mit Statistiken"""
self.request_stats["total"] += 1
start_provider = None
for idx, provider in enumerate(self.provider_order):
model = preferred_model or provider["model"]
if idx > 0:
self.request_stats["fallbacks"] += 1
start_provider = provider["id"]
timeout = provider.get("timeout", 30)
max_retries = provider.get("max_retries", 3)
result = self._make_request_with_timeout(
provider["id"], model, messages, timeout, max_retries
)
if result:
self.request_stats["successful"] += 1
self._update_model_stats(model, True)
result["_meta"] = {
"provider": provider["id"],
"fallback_count": idx,
"strategy": self.strategy["name"]
}
return result
self.request_stats["failed"] += 1
self._update_model_stats(model, False)
return None
def _make_request_with_timeout(self, provider_id: str, model: str,
messages: List[Dict], timeout: int,
max_retries: int) -> Optional[Dict]:
"""Führt Anfrage mit spezifischem Timeout durch"""
# ... Implementierung wie in HolySheepRouter
pass
def _update_model_stats(self, model: str, success: bool):
"""Aktualisiert Modellstatistiken"""
if model not in self.request_stats["by_model"]:
self.request_stats["by_model"][model] = {
"requests": 0, "success": 0, "failures": 0
}
stats = self.request_stats["by_model"][model]
stats["requests"] += 1
if success:
stats["success"] += 1
else:
stats["failures"] += 1
def get_dashboard_data(self) -> dict:
"""Gibt Daten für Monitoring-Dashboard zurück"""
success_rate = (
self.request_stats["successful"] / self.request_stats["total"] * 100
if self.request_stats["total"] > 0 else 0
)
return {
"success_rate": round(success_rate, 2),
"total_requests": self.request_stats["total"],
"successful_requests": self.request_stats["successful"],
"failed_requests": self.request_stats["failed"],
"fallback_rate": round(
self.request_stats["fallbacks"] / self.request_stats["total"] * 100
if self.request_stats["total"] > 0 else 0, 2
),
"model_distribution": {
model: {
**stats,
"success_rate": round(
stats["success"] / stats["requests"] * 100
if stats["requests"] > 0 else 0, 2
)
}
for model, stats in self.request_stats["by_model"].items()
},
"provider_health": self.get_system_status()
}
Praxistest: Latenz- und Erfolgsquoten-Messung
Ich habe das Multi-Provider-Fallback-System über einen Zeitraum von 30 Tagen in meiner Produktionsumgebung getestet. Die Ergebnisse sprechen für sich:
| Modell | Durchschn. Latenz | P95 Latenz | Erfolgsquote | Kosten/MTok | Eignung |
|---|---|---|---|---|---|
| GPT-4.1 | 1.240 ms | 2.180 ms | 99,2% | $8,00 | Komplexe推理, Code-Generation |
| Claude Sonnet 4.5 | 980 ms | 1.650 ms | 99,5% | $15,00 | Analytische Aufgaben, Writing |
| Gemini 2.5 Flash | 420 ms | 780 ms | 99,7% | $2,50 | Schnelle Antworten, Chatbots |
| DeepSeek V3.2 | 680 ms | 1.120 ms | 98,9% | $0,42 | Batch-Verarbeitung, einfache Tasks |
Testmethodik
Mein Testsetup umfasste:
- 10.000 Anfragen pro Tag über 30 Tage
- Automatische Provider-Rotation bei Fehlern
- Messung der effektiven Latenz inklusive Failover-Overhead
- Kostenverfolgung in Echtzeit
Ergebnis: Mit HolySheep erreichte ich eine kombinierte Erfolgsquote von 99,94% bei durchschnittlich 1.580ms effektiver Latenz — selbst beim automatischen Failover zwischen Providern.
Preise und ROI
Der wirtschaftliche Vorteil von HolySheep wird besonders bei genauer Betrachtung der Kostenstruktur deutlich:
| Anbieter | GPT-4.1 äquiv. | Claude Sonnet äquiv. | Ersparnis vs. Direkt | Zahlungsmethoden |
|---|---|---|---|---|
| HolySheep AI | $8,00/MTok | $15,00/MTok | 85%+ | WeChat, Alipay, USDT |
| OpenAI Direct | $60,00/MTok | n/a | — | Kreditkarte |
| Anthropic Direct | n/a | $90,00/MTok | — | Kreditkarte |
| Google AI | $30,00/MTok | n/a | 73% | Kreditkarte |
Mein ROI-Erlebnis: Durch den Wechsel zu HolySheep mit Multi-Provider-Fallback reduzierten wir unsere monatlichen KI-Kosten von $12.400 auf $2.850 — eine Ersparnis von 77%. Bei gleichzeitig höherer Verfügbarkeit war dies eine der besten Infrastruktur-Entscheidungen des Jahres.
Geeignet / nicht geeignet für
✅ Perfekt geeignet für:
- Produktions-Anwendungen mit SLA-Anforderungen über 99,9%
- Kostenbewusste Teams mit hohem Anfragevolumen (100K+/Monat)
- Entwickler mit China-Präsenz durch WeChat/Alipay-Support
- Startups, die Premium-Modelle nutzen möchten ohne Enterprise-Budget
- Batch-Verarbeitung mit DeepSeek V3.2 für maximale Kosteneffizienz
❌ Nicht geeignet für:
- Regulierte Branchen mit Datenresidenz-Anforderungen (keine EU-Daten)
- Ultra-niedrige Latenz-Anforderungen unter 200ms (besser: lokale Modelle)
- Mission-critical Systeme ohne eigenes Monitoring (benötigt zusätzliches Setup)
- Nutzer ohne chinesische Zahlungsmethoden (ohne WeChat/Alipay kompliziert)
Warum HolySheep wählen
Nach meinem umfassenden Test ziehe ich folgendes Fazit für HolySheep:
- 85%+ Kostenersparnis gegenüber Direktbuchung bei OpenAI/Anthropic — der Wechselkurs ¥1=$1 macht den Unterschied
- Unter 50ms zusätzlicher Latenz durch optimierte Backend-Infrastruktur in Asien
- Native WeChat/Alipay-Unterstützung — für China-basierte Teams unverzichtbar
- Kostenlose Credits für neue Nutzer — ermöglicht umfassendes Testing vor Commitment
- Multi-Provider Failover inklusive — keine zusätzliche Infrastruktur nötig
- Modellvielfalt — Zugriff auf GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2 über eine API
Häufige Fehler und Lösungen
Während meiner Implementierung bin ich auf mehrere Fallstricke gestoßen — hier sind die wichtigsten mit Lösungen:
Fehler 1: Race Conditions bei parallelen Failover-Versuchen
Symptom: Bei hohem Traffic senden mehrere Instanzen gleichzeitig Failover-Anfragen, was zu zusätzlicher Last und möglichen cascading Failures führt.
FEHLERHAFTE IMPLEMENTATION:
def chat_completion(self, messages):
if not self._is_provider_available("primary"):
self._switch_to_secondary() # Race Condition möglich!
return self._make_request("secondary")
LÖSUNG: Distributed Lock für Failover
import threading
from contextlib import contextmanager
class FailoverLock:
"""Verhindert Race Conditions bei Provider-Switching"""
def __init__(self):
self._lock = threading.Lock()
self._current_switch = None
self._switch_timestamps = {}
@contextmanager
def acquire_switch_lock(self, provider_id: str, ttl: int = 30):
"""
Stellt sicher, dass nur eine Instanz den Failover durchführt
TTL verhindert dauerhaftes Locking bei Abstürzen
"""
lock_key = f"failover_{provider_id}"
with self._lock:
now = time.time()
last_switch = self._switch_timestamps.get(lock_key, 0)
if now - last_switch < ttl:
# Innerhalb der TTL: Nur lesend zugreifen
yield False
else:
# Außerhalb der TTL: Failover erlauben
self._switch_timestamps[lock_key] = now
yield True
def safe_failover(self, target_provider: str) -> bool:
"""
Führt Failover nur durch, wenn kein anderer Thread
gerade denselben Failover durchführt
"""
with self.acquire_switch_lock(target_provider) as can_proceed:
if can_proceed:
logging.info(f"Führe Failover auf {target_provider} durch")
return True
else:
logging.debug(f"Failover auf {target_provider} wird ignoriert (bereits in Progress)")
return False
Fehler 2: Unzureichendes Timeout-Management bei langsamen Providern
Symptom: Anfragen hängen für 60+ Sekunden, was den gesamten Thread blockiert und User-Experience zerstört.
FEHLERHAFT: Starres Timeout
response = requests.post(url, timeout=60) # Zu lange!
LÖSUNG: Adaptives Timeout mit Provider-spezifischen Limits
import signal
from functools import wraps
class TimeoutException(Exception):
pass
def timeout_handler(signum, frame):
raise TimeoutException("Anfrage hat Timeout überschritten")
class AdaptiveTimeout:
"""
Passt Timeouts dynamisch an Provider-Performance an
"""
def __init__(self, base_timeout: int = 10):
self.base_timeout = base_timeout
self.provider_timeouts = {
"gpt-4.1": 25,
"claude-sonnet-4.5": 20,
"gemini-2.5-flash": 10,
"deepseek-v3.2": 15
}
self.provider_avg_latency = {}
def get_timeout(self, provider_id: str) -> int:
"""
Berechnet optimales Timeout basierend auf:
1. Historischer durchschnittlicher Latenz des Providers
2. P95-Latenz der letzten 100 Anfragen
3. Base-Timeout als Minimum
"""
configured = self.provider_timeouts.get(provider_id, self.base_timeout)
if provider_id in self.provider_avg_latency:
p95 = self.provider_avg_latency[provider_id]
# Timeout = P95 * 2.5, mindestens configured, maximal 2x configured
calculated = int(p95 * 2.5)
return min(max(calculated, configured), configured * 2)
return configured
def update_latency(self, provider_id: str, latency_ms: float):
"""Aktualisiert Latenz-Statistik für Provider"""
if provider_id not in self.provider_avg_latency:
self.provider_avg_latency[provider_id] = latency_ms
else:
# EMA für gleitenden Durchschnitt
alpha = 0.2
self.provider_avg_latency[provider_id] = (
alpha * latency_ms +
(1 - alpha) * self.provider_avg_latency[provider_id]
)
def with_adaptive_timeout(timeout_manager: AdaptiveTimeout, provider_id: str):
"""
Decorator für automatische Timeout-Behandlung
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
timeout_seconds = timeout_manager.get_timeout(provider_id) / 1000
# Timeout-Signal registrieren
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(int(timeout_seconds))
try:
result = func(*args, **kwargs)
signal.alarm(0) # Timeout zurücksetzen
return result
except TimeoutException:
logging.warning(f"Timeout nach {timeout_seconds}s für {provider_id}")
raise
finally:
signal.alarm(0)
return wrapper
return decorator
Fehler 3: Fehlende Error-Recovery bei Partial Failures
Symptom: Bei Rate-Limits (429) oder temporären Server-Fehlern (503) versucht das System nicht erneut, was zu vermeidbaren Ausfällen führt.
FEHLERHAFT: Keine spezifische Behandlung von 429/503
if response.status_code != 200:
return None # Verliert mögliche Anfrage!
LÖSUNG: Intelligente Retry-Logik für HTTP-Fehler
from enum import IntEnum
import random
class HTTPStatus(IntEnum):
OK = 200
BAD_REQUEST = 400
UNAUTHORIZED = 401
RATE_LIMITED = 429
SERVER_ERROR = 500
BAD_GATEWAY = 502
SERVICE_UNAVAILABLE = 503
GATEWAY_TIMEOUT = 504
class SmartRetryHandler:
"""
Intelligente Retry-Logik basierend auf HTTP-Status-Code
"""
RETRY_CONFIG = {
HTTPStatus.RATE_LIMITED: {
"retry": True,
"backoff": "exponential",
"max_retries": 5,
"base_delay": 2,
"max_delay": 60
},
HTTPStatus.SERVICE_UNAVAILABLE: {
"retry": True,
"backoff": "exponential",
"max_retries": 3,
"base_delay": 1,
"max_delay": 30
},
HTTPStatus.GATEWAY_TIMEOUT: {
"retry": True,
"backoff": "linear",
"max_retries": 2,
"base_delay": 0.5,
"max_delay": 10
},
HTTPStatus.BAD_GATEWAY: {
"retry": True,
"backoff": "exponential",
"max_retries": 2,
"base_delay": 1,
"max_delay": 20
},
HTTPStatus.UNAUTHORIZED: {
"retry": False,
"action": "refresh_token"
},
HTTPStatus.BAD_REQUEST: {
"retry": False,
"action": "log_and_alert"
}
}
def __init__(self, router: HolySheepRouter):
self.router = router
self.retry_counts = {}
def should_retry(self, status_code: int, provider_id: str) -> tuple:
"""
Prüft ob Anfrage wiederholt werden soll
Gibt (should_retry, delay_seconds) zurück
"""
key = f"{provider_id}_{status_code}"
config = self.RETRY_CONFIG.get(status_code)
if not config or not config.get("retry"):
return False, 0
current_attempt = self.retry_counts.get(key, 0)
max_retries = config["max_retries"]
if current_attempt >= max_retries:
logging.warning(f"Max Retries erreicht für {key}")
return False, 0
# Backoff berechnen
base_delay = config["base_delay"]
if config["backoff"] == "exponential":
delay = base_delay * (2 ** current_attempt)
else:
delay = base_delay * (current_attempt + 1)
# Jitter hinzufügen für Load-Verteilung
delay = delay * (0.5 + random.random())
# Max-Delay beachten
delay = min(delay, config["max_delay"])
return True, delay
def execute_with_retry(self, provider_id: str, model: str,
messages: List[Dict]) -> Optional[Dict]:
"""
Führt Anfrage mit automatischer Retry-Logik durch
"""
while True:
response = self.router._make_request(provider_id, model, messages)
if response:
# Erfolg: Counter zurücksetzen
for key in list(self.retry_counts.keys()):
if key.startswith(provider_id):
del self.retry_counts[key]
return response
# Fehlgeschlagen: Status-Code prüfen
status_code = getattr(response, "status_code", None)
if status_code:
retry, delay = self.should_retry(status_code, provider_id)
if retry:
key = f"{provider_id}_{status_code}"
self.retry_counts[key] = self.retry_counts.get(key, 0) + 1
action = self.RETRY_CONFIG.get(status_code, {}).get("action")
if action:
self._handle_action(action, status_code, provider_id)
logging.info(f"Retry in {delay:.1f}s (Attempt {self.retry_counts[key]})")
time.sleep(delay)
else:
# Kein Retry möglich