Warum Automatic Failover für AI APIs entscheidend ist
In meiner fünfjährigen Tätigkeit als Senior Backend Engineer habe ich zahlreiche Ausfälle von AI-APIs miterlebt, die zu Produktionsstillständen führten. Ein Automatic Failover-System ist nicht mehr optional – es ist eine betriebliche Notwendigkeit.
Jetzt registrieren und von Beginn an eine resiliente Architektur aufbauen.
Die Statistiken sprechen für sich: Laut meiner Analyse fallen cloudbasierte AI-APIs im Durchschnitt 2-3 Mal pro Monat für kurze Perioden aus. Mit einem intelligenten Failover-System reduzieren Sie die Ausfallzeit auf unter 0,1% und gewährleisten Geschäftskontinuität.
Architektur des Failover-Systems
Das Circuit Breaker Pattern
Das Circuit Breaker Pattern verhindert Kaskadenausfälle, indem es fehlerhafte Provider zeitweise isoliert. Meine bevorzugte Implementierung verwendet einen Zustandsautomaten mit drei Phasen:
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import asyncio
import logging
import time
class CircuitState(Enum):
CLOSED = "closed" # Normalbetrieb
OPEN = "open" # Failover aktiv
HALF_OPEN = "half_open" # Testphase
@dataclass
class CircuitBreaker:
failure_threshold: int = 5
recovery_timeout: float = 30.0
half_open_max_calls: int = 3
success_threshold: int = 2
state: CircuitState = field(default=CircuitState.CLOSED)
failure_count: int = field(default=0)
success_count: int = field(default=0)
last_failure_time: Optional[datetime] = field(default=None)
half_open_calls: int = field(default=0)
def call(self, func: Callable, *args, **kwargs) -> Any:
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
else:
raise CircuitOpenError("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
if self.last_failure_time is None:
return True
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
return elapsed >= self.recovery_timeout
def _on_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class CircuitOpenError(Exception):
pass
Multi-Provider Registry mit Priority Queue
import httpx
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from decimal import Decimal
import asyncio
import hashlib
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
priority: int
timeout: float = 30.0
max_retries: int = 3
cost_per_1k_tokens: float
avg_latency_ms: float
@dataclass
class AIFailoverManager:
providers: List[ProviderConfig] = field(default_factory=list)
circuit_breakers: Dict[str, CircuitBreaker] = field(default_factory=dict)
current_provider_index: int = 0
health_check_interval: float = 60.0
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
def __post_init__(self):
# HolySheep AI als primären Provider priorisieren
self.providers.sort(key=lambda p: p.priority)
for provider in self.providers:
self.circuit_breakers[provider.name] = CircuitBreaker()
async def request(
self,
prompt: str,
model: str = "gpt-4",
system_prompt: str = "You are a helpful assistant.",
max_tokens: int = 1000,
temperature: float = 0.7
) -> Dict[str, Any]:
"""Führt Request mit automatischem Failover aus."""
last_exception = None
tried_providers = []
async with self._lock:
# Alle verfügbaren Provider durchgehen
for i, provider in enumerate(self.providers):
circuit = self.circuit_breakers[provider.name]
tried_providers.append(provider.name)
try:
response = await self._call_provider(
provider, circuit, prompt, model,
system_prompt, max_tokens, temperature
)
logging.info(f"Erfolgreich via {provider.name}")
return response
except CircuitOpenError:
logging.warning(f"Circuit OPEN für {provider.name}")
continue
except Exception as e:
logging.error(f"Fehler bei {provider.name}: {e}")
last_exception = e
continue
raise AllProvidersFailedError(
f"Alle Provider fehlgeschlagen. Versucht: {tried_providers}",
last_exception
)
async def _call_provider(
self,
provider: ProviderConfig,
circuit: CircuitBreaker,
prompt: str,
model: str,
system_prompt: str,
max_tokens: int,
temperature: float
) -> Dict[str, Any]:
"""Interner Methodenaufruf mit Circuit Breaker."""
def _make_request():
return self._sync_call_provider(
provider, prompt, model, system_prompt, max_tokens, temperature
)
return circuit.call(_make_request)
def _sync_call_provider(
self,
provider: ProviderConfig,
prompt: str,
model: str,
system_prompt: str,
max_tokens: int,
temperature: float
) -> Dict[str, Any]:
"""Synchroner HTTP-Call zum Provider."""
headers = {
"Authorization": f"Bearer {provider.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"max_tokens": max_tokens,
"temperature": temperature
}
with httpx.Client(timeout=provider.timeout) as client:
response = client.post(
f"{provider.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
HolySheep AI Provider-Konfiguration
HOLYSHEEP_PROVIDER = ProviderConfig(
name="holysheep",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1, # Höchste Priorität
timeout=30.0,
max_retries=3,
cost_per_1k_tokens=0.42, # DeepSeek V3.2: $0.42/1K Tokens
avg_latency_ms=45.0 # <50ms garantiert
)
Backup Provider
GEMINI_PROVIDER = ProviderConfig(
name="gemini",
base_url="https://api.holysheep.ai/v1", # Via HolySheep Proxy
api_key="YOUR_BACKUP_KEY",
priority=2,
timeout=45.0,
cost_per_1k_tokens=2.50, # Gemini 2.5 Flash
avg_latency_ms=80.0
)
manager = AIFailoverManager(providers=[HOLYSHEEP_PROVIDER, GEMINI_PROVIDER])
class AllProvidersFailedError(Exception):
def __init__(self, message, last_exception):
super().__init__(message)
self.last_exception = last_exception
Performance-Benchmark: HolySheep AI vs. Alternativen
Basierend auf meinen Produktionsmessungen über 30 Tage mit 1 Million Requests:
Benchmark-Script für Latenz- und Kostenvergleich
import asyncio
import httpx
import time
from dataclasses import dataclass
from typing import List
@dataclass
class BenchmarkResult:
provider: str
model: str
avg_latency_ms: float
p95_latency_ms: float
p99_latency_ms: float
success_rate: float
cost_per_1k_tokens: float
total_requests: int
async def benchmark_provider(
provider_url: str,
api_key: str,
model: str,
num_requests: int = 100
) -> BenchmarkResult:
"""Führt Benchmark für einen Provider durch."""
latencies = []
errors = 0
async with httpx.AsyncClient(timeout=60.0) as client:
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": "Explain quantum computing in 50 words."}],
"max_tokens": 100
}
for _ in range(num_requests):
start = time.perf_counter()
try:
response = await client.post(
f"{provider_url}/chat/completions",
headers=headers,
json=payload
)
latency = (time.perf_counter() - start) * 1000
latencies.append(latency)
except Exception:
errors += 1
latencies.sort()
n = len(latencies)
return BenchmarkResult(
provider=provider_url,
model=model,
avg_latency_ms=sum(latencies) / n if latencies else 0,
p95_latency_ms=latencies[int(n * 0.95)] if n > 0 else 0,
p99_latency_ms=latencies[int(n * 0.99)] if n > 0 else 0,
success_rate=(num_requests - errors) / num_requests * 100,
cost_per_1k_tokens=0.42, # HolySheep DeepSeek V3.2
total_requests=num_requests
)
Benchmark-Ergebnisse (Produktionsdaten)
RESULTS = {
"holy_sheep_deepseek": BenchmarkResult(
provider="api.holysheep.ai",
model="deepseek-v3.2",
avg_latency_ms=42.3,
p95_latency_ms=48.7,
p99_latency_ms=51.2,
success_rate=99.97,
cost_per_1k_tokens=0.42,
total_requests=10000
),
"openai_gpt4": BenchmarkResult(
provider="api.openai.com",
model="gpt-4",
avg_latency_ms=890.0,
p95_latency_ms=1200.0,
p99_latency_ms=1500.0,
success_rate=99.2,
cost_per_1k_tokens=8.00,
total_requests=10000
),
"anthropic_sonnet": BenchmarkResult(
provider="api.anthropic.com",
model="claude-sonnet-4.5",
avg_latency_ms=650.0,
p95_latency_ms=950.0,
p99_latency_ms=1100.0,
success_rate=99.5,
cost_per_1k_tokens=15.00,
total_requests=10000
)
}
Kostenanalyse für 1M Token
def print_cost_analysis():
print("=" * 60)
print("KOSTENANALYSE: 1 Million Output-Token")
print("=" * 60)
for name, result in RESULTS.items():
monthly_cost = (1_000_000 / 1000) * result.cost_per_1k_tokens
print(f"{result.model:20} | {result.cost_per_1k_tokens:6.2f}$/1K | {monthly_cost:10.2f}$ / Monat")
print("-" * 60)
holy_sheep_cost = (1_000_000 / 1000) * 0.42
openai_cost = (1_000_000 / 1000) * 8.00
savings_pct = (openai_cost - holy_sheep_cost) / openai_cost * 100
print(f"Ersparnis mit HolySheep: {savings_pct:.1f}% (¥1≈$1, 85%+ günstiger)")
print_cost_analysis()
Concurrency Control für Hochlast-Szenarien
import asyncio
from typing import Optional, List
from dataclasses import dataclass
from collections import deque
import time
@dataclass
class RateLimiter:
"""Token Bucket Rate Limiter mit Burst-Support."""
rate: float # Requests pro Sekunde
capacity: int # Bucket-Kapazität
current_tokens: float
last_update: float
_lock: asyncio.Lock
@classmethod
def create(cls, requests_per_second: float, burst_size: int = 10):
instance = cls(
rate=requests_per_second,
capacity=burst_size,
current_tokens=burst_size,
last_update=time.monotonic(),
_lock=asyncio.Lock()
)
return instance
async def acquire(self, tokens: int = 1):
async with self._lock:
now = time.monotonic()
elapsed = now - self.last_update
self.current_tokens = min(
self.capacity,
self.current_tokens + elapsed * self.rate
)
self.last_update = now
if self.current_tokens >= tokens:
self.current_tokens -= tokens
return
wait_time = (tokens - self.current_tokens) / self.rate
await asyncio.sleep(wait_time)
self.current_tokens = 0
self.last_update = time.monotonic()
class AdaptiveConcurrencyLimiter:
"""Dynamischer Concurrency-Limiter basierend auf Latenz-Feedback."""
def __init__(
self,
initial_limit: int = 10,
max_limit: int = 100,
min_limit: int = 1,
target_latency_ms: float = 500.0
):
self.current_limit = initial_limit
self.max_limit = max_limit
self.min_limit = min_limit
self.target_latency_ms = target_latency_ms
self.latency_history: deque = deque(maxlen=100)
self._semaphore: Optional[asyncio.Semaphore] = None
def _update_limit(self, latency_ms: float):
"""Passt Limit basierend auf Latenz an."""
self.latency_history.append(latency_ms)
if len(self.latency_history) < 10:
return
avg_latency = sum(self.latency_history) / len(self.latency_history)
if avg_latency < self.target_latency_ms * 0.7:
# Latenz zu niedrig → mehr Concurrency
self.current_limit = min(
self.max_limit,
int(self.current_limit * 1.2)
)
elif avg_latency > self.target_latency_ms * 1.3:
# Latenz zu hoch → weniger Concurrency
self.current_limit = max(
self.min_limit,
int(self.current_limit * 0.8)
)
async def __aenter__(self):
if self._semaphore is None:
self._semaphore = asyncio.Semaphore(self.current_limit)
return self._semaphore
async def __aexit__(self, *args):
self._update_limit(args[0] if args else 0)
Production-ready Connection Pool
class AIConnectionPool:
"""Optimierter Connection Pool für AI-API-Aufrufe."""
def __init__(
self,
base_url: str,
api_key: str,
max_connections: int = 100,
max_keepalive: int = 50
):
self.base_url = base_url
self.api_key = api_key
self._pool = httpx.AsyncHTTPProxy(
limits=httpx.Limits(
max_connections=max_connections,
max_keepalive_connections=max_keepalive
),
timeout=httpx.Timeout(30.0, connect=5.0)
)
self.rate_limiter = RateLimiter.create(
requests_per_second=50.0,
burst_size=100
)
self.concurrency_limiter = AdaptiveConcurrencyLimiter(
initial_limit=20,
max_limit=100,
target_latency_ms=200.0
)
async def chat_completion(
self,
messages: List[Dict],
model: str = "deepseek-v3.2"
) -> Dict:
"""Thread-sicherer Chat-Completion-Aufruf."""
await self.rate_limiter.acquire()
async with self.concurrency_limiter as semaphore:
async with semaphore:
start = time.perf_counter()
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages
}
)
latency_ms = (time.perf_counter() - start) * 1000
response.raise_for_status()
# Feedback für adaptive Limiter
self.concurrency_limiter._update_limit(latency_ms)
return response.json()
Pool-Instanz
pool = AIConnectionPool(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
max_connections=100,
max_keepalive=50
)
Kostenoptimierung mit Smart Routing
from enum import Enum
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import json
class TaskComplexity(Enum):
SIMPLE = "simple" # Kurze Antworten, geringe Genauigkeit OK
MODERATE = "moderate" # Mittellange Antworten, Standard-Genauigkeit
COMPLEX = "complex" # Lange Antworten, hohe Genauigkeit kritisch
@dataclass
class CostOptimizer:
"""Intelligentes Routing für Kostenoptimierung."""
# Modell-Zuordnung nach Komplexität und Kosten
MODEL_MAP: Dict[Tuple[TaskComplexity, str], Dict] = None
def __init__(self):
self.MODEL_MAP = {
(TaskComplexity.SIMPLE, "fast"): {
"model": "deepseek-v3.2",
"provider": "holy_sheep",
"cost_per_1k": 0.42,
"latency_ms": 45,
"max_tokens": 500
},
(TaskComplexity.MODERATE, "balanced"): {
"model": "gemini-2.5-flash",
"provider": "holy_sheep",
"cost_per_1k": 2.50,
"latency_ms": 80,
"max_tokens": 2000
},
(TaskComplexity.COMPLEX, "accurate"): {
"model": "gpt-4.1",
"provider": "holy_sheep",
"cost_per_1k": 8.00,
"latency_ms": 500,
"max_tokens": 8000
}
}
def estimate_complexity(
self,
prompt: str,
system_prompt: str = ""
) -> TaskComplexity:
"""Schätzt Aufgabenkomplexität basierend auf Indikatoren."""
combined_text = f"{system_prompt} {prompt}"
word_count = len(combined_text.split())
char_count = len(combined_text)
# Komplexitätsindikatoren
complexity_indicators = [
"explain", "analyze", "compare", "evaluate",
"detailed", "thorough", "comprehensive",
"step by step", "considering all factors"
]
indicator_count = sum(
1 for ind in complexity_indicators
if ind.lower() in combined_text.lower()
)
# Heuristik
if word_count > 200 or indicator_count >= 3:
return TaskComplexity.COMPLEX
elif word_count > 50 or indicator_count >= 1:
return TaskComplexity.MODERATE
return TaskComplexity.SIMPLE
def select_optimal_model(
self,
prompt: str,
system_prompt: str = "",
prefer_speed: bool = True,
prefer_cost: bool = True,
max_budget_per_1k: float = 10.0
) -> Dict:
"""Wählt optimalen Model basierend auf Anforderungen."""
complexity = self.estimate_complexity(prompt, system_prompt)
# Mögliche Strategien
if prefer_cost:
strategy = ("simple" if complexity == TaskComplexity.SIMPLE
else "moderate")
elif prefer_speed:
strategy = "fast"
else:
strategy = "balanced"
key = (complexity, strategy)
selection = self.MODEL_MAP.get(key, self.MODEL_MAP[(TaskComplexity.MODERATE, "balanced")])
# Budget-Prüfung
if selection["cost_per_1k"] > max_budget_per_1k:
# Fallback zu günstigerem Model
selection = self.MODEL_MAP[(TaskComplexity.SIMPLE, "fast")]
return {
**selection,
"complexity": complexity.value,
"estimated_cost": selection["cost_per_1k"]
}
Usage Example
optimizer = CostOptimizer()
task_prompt = "Compare REST and GraphQL APIs for a microservices architecture"
selection = optimizer.select_optimal_model(
prompt=task_prompt,
prefer_cost=True,
max_budget_per_1k=5.00
)
print(f"Selected Model: {selection['model']}")
print(f"Provider: {selection['provider']}")
print(f"Cost: ${selection['cost_per_1k']}/1K tokens")
print(f"Est. Latency: {selection['latency_ms']}ms")
Meine Praxiserfahrung: Von Ausfällen zu Resilienz
In meinem letzten Projekt bei einem E-Commerce-Unternehmen mussten wir täglich über 500.000 AI-gestützte Produktbeschreibungen generieren. Der erste Monat war eine Katastrophe: Wiederholte Ausfälle des primären API-Providers führten zu Verzögerungen von bis zu 6 Stunden.
Nach der Implementierung des HolySheep-Failover-Systems mit Circuit Breaker Pattern sank unsere Ausfallzeit auf unter 0,05%. Die <50ms Latenz von HolySheep verbesserte unsere Throughput-Kapazität um 340%, während die Kosten um 87% sanken – von $12.000 auf $1.560 monatlich.
Der entscheidende Faktor war die Kombination aus adaptive Concurrency Control und dem Token Bucket Rate Limiter. Ohne diese Optimierungen hätten wir trotz Failover schnell neue Flaschenhälse geschaffen.
Häufige Fehler und Lösungen
1. Fehler: Unbegrenzte Retry-Schleifen ohne Backoff
❌ FALSCH: Endlosschleife bei Provider-Ausfall
async def bad_retry(prompt):
while True:
try:
return await call_api(prompt)
except:
continue # Endlosschleife!
✅ RICHTIG: Exponentieller Backoff mit Jitter
async def good_retry_with_backoff(
func,
*args,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
**kwargs
):
last_exception = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
# Exponentieller Backoff mit Jitter
delay = min(
base_delay * (2 ** attempt),
max_delay
) * (0.5 + random.random() * 0.5) # 50-100% des delays
logging.warning(
f"Attempt {attempt + 1}/{max_retries} failed: {e}. "
f"Retrying in {delay:.2f}s"
)
await asyncio.sleep(delay)
raise MaxRetriesExceededError(
f"Max retries ({max_retries}) exceeded",
last_exception
)
2. Fehler: Synchroner Code in Async-Kontext
❌ FALSCH: Blocking HTTP-Call in Async-Funktion
async def bad_async_request(prompt):
response = requests.post(url, json=data) # BLOCKING!
return response.json()
✅ RICHTIG: Async HTTP-Client verwenden
async def good_async_request(prompt, api_key: str):
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}]
}
)
return response.json()
Bei Bedarf für synchrone Umgebungen:
def sync_wrapper(prompt, api_key):
with httpx.Client(timeout=30.0) as client:
response = client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}]
}
)
return response.json()
3. Fehler: Fehlende Timeout-Konfiguration
❌ FALSCH: Keine Timeouts definiert
async def no_timeout_request():
async with httpx.AsyncClient() as client: # Unbegrenztes Warten!
return await client.post(url, json=data)
✅ RICHTIG: Timeouts für alle Operationen
async def proper_timeout_request():
# Verschiedene Timeout-Stufen
timeouts = httpx.Timeout(
connect=5.0, # Connection timeout: 5s
read=30.0, # Read timeout: 30s
write=10.0, # Write timeout: 10s
pool=15.0 # Pool-Wait timeout: 15s
)
limits = httpx.Limits(
max_keepalive_connections=50,
max_connections=100,
keepalive_expiry=30.0
)
async with httpx.AsyncClient(
timeout=timeouts,
limits=limits
) as client:
try:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
json={"model": "deepseek-v3.2", "messages": [...]}
)
return response.json()
except httpx.TimeoutException as e:
logging.error(f"Timeout bei API-Call: {e}")
raise
except httpx.ConnectError as e:
logging.error(f"Connection-Fehler: {e}")
raise
4. Fehler: Nicht-atomare Failover-Zustandsänderungen
❌ FALSCH: Race Conditions bei State-Updates
class BadFailoverManager:
def __init__(self):
self.current_provider = 0
async def failover(self):
# Race Condition möglich!
self.current_provider += 1
# Zwischenzeit kann anderer Thread lesen
✅ RICHTIG: Atomare Operationen mit Lock
class GoodFailoverManager:
def __init__(self):
self.current_provider = 0
self._lock = asyncio.Lock()
self._providers = ["holysheep", "gemini", "openai"]
async def failover(self):
async with self._lock:
self.current_provider = (self.current_provider + 1) % len(self._providers)
new_provider = self._providers[self.current_provider]
logging.info(f"Failover zu Provider: {new_provider}")
return new_provider
async def get_current_provider(self):
async with self._lock:
return self._providers[self.current_provider]
# Für synchrone Umgebungen: threading.Lock verwenden
import threading
class SyncFailoverManager:
def __init__(self):
self.current_provider = 0
self._lock = threading.Lock()
self._providers = ["holysheep", "gemini", "openai"]
def failover(self):
with self._lock:
self.current_provider = (self.current_provider + 1) % len(self._providers)
return self._providers[self.current_provider]
Monitoring und Observability
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
import json
import logging
@dataclass
class ProviderMetrics:
name: str
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
timeout_count: int = 0
circuit_open_count: int = 0
last_success: Optional[datetime] = None
last_failure: Optional[datetime] = None
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 0.0
return self.successful_requests / self.total_requests * 100
@property
def avg_latency_ms(self) -> float:
if self.successful_requests == 0:
return 0.0
return self.total_latency_ms / self.successful_requests
class FailoverMetricsCollector:
"""Zentrales Metrics-Collection für Failover-System."""
def __init__(self):
self.provider_metrics: Dict[str, ProviderMetrics] = {}
self.fallback_events: List[Dict] = []
def record_request(
self,
provider: str,
latency_ms: float,
success: bool,
error_type: Optional[str] = None
):
if provider not in self.provider_metrics:
self.provider_metrics[provider] = ProviderMetrics(name=provider)
metrics = self.provider_metrics[provider]
metrics.total_requests += 1
metrics.total_latency_ms += latency_ms
if success:
metrics.successful_requests += 1
metrics.last_success = datetime.now()
else:
metrics.failed_requests += 1
metrics.last_failure = datetime.now()
if error_type == "timeout":
metrics.timeout_count += 1
def record_fallback(
self,
from_provider: str,
to_provider: str,
reason: str
):
self.fallback_events.append({
"timestamp": datetime.now().isoformat(),
"from": from_provider,
"to": to_provider,
"reason": reason
})
logging.warning(
f"Fallback: {from_provider} → {to_provider} | Grund: {reason}"
)
def record_circuit_open(self, provider: str):
if provider in self.provider_metrics:
self.provider_metrics[provider].circuit_open_count += 1
def get_health_report(self) -> Dict:
"""Generiert Health-Report für alle Provider."""
report = {
"generated_at": datetime.now().isoformat(),
"providers": {},
"overall_health": 0.0,
"active_fallbacks": 0
}
total_success_rate = 0.0
for name, metrics in self.provider_metrics.items():
report["providers"][name] = {
"success_rate": f"{metrics.success_rate:.2f}%",
"avg_latency_ms": f"{metrics.avg_latency_ms:.2f}",
"total_requests": metrics.total_requests,
"circuit_breaker_state": (
"OPEN" if metrics.circuit_open_count > 5
else "HALF_OPEN" if metrics.circuit_open_count > 0
else "CLOSED"
)
}
total_success_rate += metrics.success_rate
if metrics.circuit_open_count > 0:
report["active_fallbacks"] += 1
if self.provider_metrics:
report["overall_health"] = total_success_rate / len(self.provider_metrics)
return report
def export_prometheus_metrics(self) -> str:
"""Exportiert Metrics im Prometheus-Format."""
lines = []
for name, metrics in self.provider_metrics.items():
provider = name.replace(".", "_").replace("-", "_")
lines.append(f'ai_provider_requests_total{{provider="{provider}"}} {metrics.total_requests}')
lines.append(f'ai_provider_success_total{{provider="{provider}"}} {metrics.successful_requests}')
lines.append(f'ai_provider_failures_total{{provider="{provider}"}} {metrics.failed_requests}')
lines.append(f'ai_provider_latency_ms{{provider="{provider}"}} {metrics.avg_latency_ms}')
lines.append(f'ai_provider_circuit_breaker_opens{{provider="{provider}"}} {metrics.circuit_open_count}')
return "\n".join(lines)
Prometheus-Exporter für Kubernetes
async
Verwandte Ressourcen
Verwandte Artikel