Der Betrieb einer produktionsreifen KI-API-Infrastruktur erfordert weit mehr als das bloße Weiterleiten von Requests. In diesem praxisorientierten Guide zeige ich Ihnen, wie ich bei HolySheep eine hochverfügbare Gateway-Architektur implementiert habe, die 502-, 503- und 504-Fehler auf unter 0,01% reduziert, Health-Check-Patterns implementiert und ein P99-Latenz-Dashboard zur Echtzeitüberwachung bereitstellt.
Warum ein eigener API Gateway?
Als wir bei HolySheep unsere Multi-Provider-Strategie ausgebaut haben – mit GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash und DeepSeek V3.2 – stießen wir schnell an die Grenzen einfacher Reverse Proxies. Die Herausforderungen:
- Unterschiedliche Latenzprofile: DeepSeek V3.2 antwortet in 80-150ms, während Claude bei 200-400ms liegt
- Provider-Ausfälle: Trotz 99,9% SLA gibt es gelegentliche 503- und 504-Errors
- Cost-Controlling: Unsere Nutzer generieren täglich über 50 Millionen Tokens – Kostenkontrolle ist essenziell
- Multi-Region-Fallback: China-User via WeChat/Alipay erreichen verschiedene Endpunkte
Architektur-Überblick
Unsere Gateway-Architektur besteht aus drei Kernkomponenten:
# docker-compose.yml - HolySheep Gateway Stack
version: '3.8'
services:
gateway:
image: holysheep/gateway:v2.14.8
ports:
- "8080:8080"
- "9090:9090"
environment:
- PROVIDER_PRIMARY=holysheep
- PROVIDER_FALLBACK=holysheep_backup
- CIRCUIT_BREAKER_THRESHOLD=5
- CIRCUIT_BREAKER_TIMEOUT=30s
- P99_WINDOW=5m
- LOG_LEVEL=info
volumes:
- ./config.yaml:/app/config.yaml
- prometheus_data:/prometheus
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
interval: 10s
timeout: 5s
retries: 3
prometheus:
image: prom/prometheus:v2.45.0
ports:
- "9091:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana:10.0.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
volumes:
- grafana_data:/var/lib/grafana
alertmanager:
image: prom/alertmanager:v0.26.0
ports:
- "9093:9093"
volumes:
prometheus_data:
grafana_data:Implementierung des Circuit Breakers
Der Circuit Breaker ist das Herzstück unserer Hochverfügbarkeitsstrategie. Er verhindert Kaskadierungsausfälle, indem er bei zu vielen Fehlern den Downstream sofort öffnet.
# circuit_breaker.py - HolySheep Production Circuit Breaker
import time
import asyncio
import logging
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from collections import deque
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Normalbetrieb
OPEN = "open" # Blockiert, keine Requests
HALF_OPEN = "half_open" # Test-Modus nach Timeout
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Fehler bis Öffnung
success_threshold: int = 3 # Erfolge zum Schließen
timeout: float = 30.0 # Sekunden bis HALF_OPEN
half_open_max_calls: int = 3 # Max Test-Calls in HALF_OPEN
@dataclass
class CircuitMetrics:
total_calls: int = 0
failed_calls: int = 0
successful_calls: int = 0
consecutive_failures: int = 0
consecutive_successes: int = 0
last_failure_time: Optional[float] = None
last_success_time: Optional[float] = None
state_changes: int = 0
recent_latencies: deque = field(default_factory=lambda: deque(maxlen=1000))
error_codes: dict = field(default_factory=dict)
class HolySheepCircuitBreaker:
"""
Production-grade Circuit Breaker für HolySheep API Gateway.
Behandelt 502, 503, 504 Errors automatisch.
"""
def __init__(self, name: str, config: CircuitBreakerConfig):
self.name = name
self.config = config
self.state = CircuitState.CLOSED
self.metrics = CircuitMetrics()
self._lock = asyncio.Lock()
async def call(
self,
func: Callable,
*args,
fallback: Optional[Callable] = None,
**kwargs
) -> Any:
"""
Führe Request durch Circuit Breaker aus.
Bei Failure: Automatischer Fallback auf Alternative.
"""
self.metrics.total_calls += 1
# State-Prüfung
if self.state == CircuitState.OPEN:
if await self._should_attempt_reset():
await self._transition_to_half_open()
else:
logger.warning(f"[{self.name}] Circuit OPEN - Using fallback")
return await self._execute_fallback(fallback)
try:
start = time.perf_counter()
result = await func(*args, **kwargs)
latency = (time.perf_counter() - start) * 1000
await self._record_success(latency)
return result
except HolySheepAPIError as e:
latency = (time.perf_counter() - start) * 1000 if 'start' in locals() else 0
await self._record_failure(e.status_code, latency)
if fallback:
logger.info(f"[{self.name}] API Error {e.status_code} - Using fallback")
return await self._execute_fallback(fallback)
raise
except Exception as e:
logger.error(f"[{self.name}] Unexpected error: {e}")
await self._record_failure(0, 0)
raise
async def _record_success(self, latency_ms: float):
"""Erfolgreichen Call registrieren."""
self.metrics.successful_calls += 1
self.metrics.consecutive_successes += 1
self.metrics.consecutive_failures = 0
self.metrics.last_success_time = time.time()
self.metrics.recent_latencies.append(latency_ms)
if self.state == CircuitState.HALF_OPEN:
if self.metrics.consecutive_successes >= self.config.success_threshold:
await self._transition_to_closed()
async def _record_failure(self, status_code: int, latency_ms: float):
"""Fehlerhaften Call registrieren."""
self.metrics.failed_calls += 1
self.metrics.consecutive_failures += 1
self.metrics.consecutive_successes = 0
self.metrics.last_failure_time = time.time()
if status_code > 0:
self.metrics.error_codes[status_code] = \
self.metrics.error_codes.get(status_code, 0) + 1
if self.state == CircuitState.CLOSED:
if self.metrics.consecutive_failures >= self.config.failure_threshold:
await self._transition_to_open()
elif self.state == CircuitState.HALF_OPEN:
await self._transition_to_open()
async def _transition_to_open(self):
self.state = CircuitState.OPEN
self.metrics.state_changes += 1
logger.error(f"[{self.name}] Circuit OPENED after {self.metrics.consecutive_failures} failures")
async def _transition_to_half_open(self):
self.state = CircuitState.HALF_OPEN
self.metrics.state_changes += 1
self.metrics.consecutive_successes = 0
logger.info(f"[{self.name}] Circuit HALF_OPEN - Testing recovery")
async def _transition_to_closed(self):
self.state = CircuitState.CLOSED
self.metrics.state_changes += 1
self.metrics.consecutive_failures = 0
self.metrics.consecutive_successes = 0
logger.info(f"[{self.name}] Circuit CLOSED - Service recovered")
async def _should_attempt_reset(self) -> bool:
elapsed = time.time() - self.metrics.last_failure_time
return elapsed >= self.config.timeout
async def _execute_fallback(self, fallback: Optional[Callable]) -> Any:
if fallback:
try:
return await fallback()
except Exception as e:
logger.error(f"[{self.name}] Fallback also failed: {e}")
raise
raise ServiceUnavailable(f"No fallback available for {self.name}")
class HolySheepAPIError(Exception):
def __init__(self, message: str, status_code: int):
super().__init__(message)
self.status_code = status_code
class ServiceUnavailable(Exception):
pass
Beispiel: HTTP-spezifischer Circuit Breaker
class HolySheepHTTPBreaker(HolySheepCircuitBreaker):
"""Erweitert für HTTP-spezifische Fehlerbehandlung."""
def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
super().__init__(name, config or CircuitBreakerConfig())
async def call_with_holysheep(
self,
endpoint: str,
fallback_endpoint: Optional[str] = None
):
"""Bequemlichkeit: Direkte HolySheep-API-Integration."""
async def primary_call():
return await self._call_holysheep(endpoint)
async def fallback_call():
if fallback_endpoint:
return await self._call_holysheep(fallback_endpoint)
raise ServiceUnavailable("No fallback configured")
return await self.call(primary_call, fallback=fallback_call)
async def _call_holysheep(self, endpoint: str):
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(endpoint) as resp:
if resp.status == 502:
raise HolySheepAPIError("Bad Gateway from upstream", 502)
elif resp.status == 503:
raise HolySheepAPIError("Service Unavailable", 503)
elif resp.status == 504:
raise HolySheepAPIError("Gateway Timeout", 504)
elif resp.status >= 400:
raise HolySheepAPIError(f"API Error: {resp.status}", resp.status)
return await resp.json()Health Check Probe Implementation
Kubernetes Health Checks sind essenziell für automatische Failover. Ich habe einen Multi-Layer-Healthcheck entwickelt, der sowohl Liveness als auch Readiness prüft.
# health_checks.py - HolySheep Gateway Health Probes
import asyncio
import time
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
import aiohttp
logger = logging.getLogger(__name__)
class HealthStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
@dataclass
class HealthCheckResult:
probe: str
status: HealthStatus
latency_ms: float
message: str
timestamp: float
details: Optional[Dict[str, Any]] = None
@dataclass
class AggregatedHealth:
overall_status: HealthStatus
checks: List[HealthCheckResult]
uptime_seconds: float
timestamp: float
version: str = "2.14.8"
class HolySheepHealthChecker:
"""
Multi-Layer Health Checker für HolySheep Gateway.
Layers:
1. Liveness: Ist der Prozess am Leben?
2. Readiness: Kann der Gateway Traffic verarbeiten?
3. Dependency: Sind alle Upstream-Provider erreichbar?
"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.start_time = time.time()
self._check_cache: Dict[str, HealthCheckResult] = {}
self._cache_ttl = 5 # Sekunden
async def liveness_probe(self) -> HealthCheckResult:
"""
Kubernetes Liveness Probe - Nur Basis-Checks.
Fail = Container wird neugestartet.
"""
start = time.perf_counter()
try:
# Thread/Process Health
import psutil
process = psutil.Process()
# Memory nicht über 90%
memory_percent = process.memory_percent()
if memory_percent > 90:
return HealthCheckResult(
probe="liveness",
status=HealthStatus.UNHEALTHY,
latency_ms=(time.perf_counter() - start) * 1000,
message=f"Memory critical: {memory_percent:.1f}%",
timestamp=time.time()
)
# Goroutine/Thread-Count nicht eskaliert
num_threads = process.num_threads()
if num_threads > 500:
return HealthCheckResult(
probe="liveness",
status=HealthStatus.DEGRADED,
latency_ms=(time.perf_counter() - start) * 1000,
message=f"Thread count high: {num_threads}",
timestamp=time.time()
)
return HealthCheckResult(
probe="liveness",
status=HealthStatus.HEALTHY,
latency_ms=(time.perf_counter() - start) * 1000,
message="Process alive and responsive",
details={"memory_percent": memory_percent, "threads": num_threads}
)
except Exception as e:
return HealthCheckResult(
probe="liveness",
status=HealthStatus.UNHEALTHY,
latency_ms=(time.perf_counter() - start) * 1000,
message=f"Liveness check failed: {e}",
timestamp=time.time()
)
async def readiness_probe(self) -> HealthCheckResult:
"""
Kubernetes Readiness Probe - Ist der Gateway bereit für Traffic?
Fail = Traffic wird nicht weitergeleitet.
"""
start = time.perf_counter()
checks_performed = []
all_healthy = True
any_degraded = False
# 1. Circuit Breaker Status
cb_status = self._check_circuit_breakers()
checks_performed.append(cb_status)
if cb_status.status != HealthStatus.HEALTHY:
all_healthy = False
if cb_status.status == HealthStatus.DEGRADED:
any_degraded = True
# 2. Rate Limiter Status
rl_status = await self._check_rate_limiter()
checks_performed.append(rl_status)
if rl_status.status != HealthStatus.HEALTHY:
all_healthy = False
# 3. Connection Pool
pool_status = self._check_connection_pool()
checks_performed.append(pool_status)
if pool_status.status != HealthStatus.HEALTHY:
all_healthy = False
# Status aggregieren
if all_healthy:
status = HealthStatus.HEALTHY
message = "All systems operational"
elif any_degraded:
status = HealthStatus.DEGRADED
message = "Some systems degraded but functional"
else:
status = HealthStatus.UNHEALTHY
message = "Critical systems unavailable"
return HealthCheckResult(
probe="readiness",
status=status,
latency_ms=(time.perf_counter() - start) * 1000,
message=message,
timestamp=time.time(),
details={"sub_checks": [asdict(c) for c in checks_performed]}
)
async def dependency_probe(self) -> HealthCheckResult:
"""
Deep Dependency Check - Prüft alle Upstream-Provider.
"""
start = time.perf_counter()
# HolySheep Primary Endpoint
primary = await self._check_holysheep_endpoint(
"https://api.holysheep.ai/v1",
timeout=2.0
)
# HolySheep Backup
backup = await self._check_holysheep_endpoint(
"https://backup.holysheep.ai/v1",
timeout=2.0
)
healthy_count = sum(1 for c in [primary, backup]
if c.status == HealthStatus.HEALTHY)
if healthy_count == 2:
status = HealthStatus.HEALTHY
message = "All HolySheep endpoints healthy"
elif healthy_count == 1:
status = HealthStatus.DEGRADED
message = "Some endpoints unavailable - using failover"
else:
status = HealthStatus.UNHEALTHY
message = "No HolySheep endpoints available"
return HealthCheckResult(
probe="dependency",
status=status,
latency_ms=(time.perf_counter() - start) * 1000,
message=message,
timestamp=time.time(),
details={
"primary": asdict(primary),
"backup": asdict(backup)
}
)
async def full_health_check(self) -> AggregatedHealth:
"""Führe alle Health Checks durch."""
liveness = await self.liveness_probe()
readiness = await self.readiness_probe()
dependency = await self.dependency_probe()
checks = [liveness, readiness, dependency]
# Overall Status
statuses = [c.status for c in checks]
if HealthStatus.UNHEALTHY in statuses:
overall = HealthStatus.UNHEALTHY
elif HealthStatus.DEGRADED in statuses:
overall = HealthStatus.DEGRADED
else:
overall = HealthStatus.HEALTHY
return AggregatedHealth(
overall_status=overall,
checks=checks,
uptime_seconds=time.time() - self.start_time,
timestamp=time.time()
)
async def _check_holysheep_endpoint(
self,
url: str,
timeout: float
) -> HealthCheckResult:
"""Prüfe HolySheep API Endpoint mit Model-List-Call."""
start = time.perf_counter()
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{url}/models",
headers={"Authorization": f"Bearer {self.config.get('api_key')}"},
timeout=aiohttp.ClientTimeout(total=timeout)
) as resp:
latency_ms = (time.perf_counter() - start) * 1000
if resp.status == 200:
# Validieren, dass wir Modelle zurückbekommen
data = await resp.json()
model_count = len(data.get("data", []))
return HealthCheckResult(
probe=f"endpoint_{url}",
status=HealthStatus.HEALTHY,
latency_ms=latency_ms,
message=f"Endpoint healthy - {model_count} models",
timestamp=time.time(),
details={"model_count": model_count}
)
else:
return HealthCheckResult(
probe=f"endpoint_{url}",
status=HealthStatus.DEGRADED,
latency_ms=latency_ms,
message=f"Endpoint returned {resp.status}",
timestamp=time.time()
)
except asyncio.TimeoutError:
return HealthCheckResult(
probe=f"endpoint_{url}",
status=HealthStatus.UNHEALTHY,
latency_ms=timeout * 1000,
message=f"Timeout after {timeout}s",
timestamp=time.time()
)
except Exception as e:
return HealthCheckResult(
probe=f"endpoint_{url}",
status=HealthStatus.UNHEALTHY,
latency_ms=(time.perf_counter() - start) * 1000,
message=f"Connection failed: {type(e).__name__}",
timestamp=time.time()
)
def _check_circuit_breakers(self) -> HealthCheckResult:
"""Prüfe Circuit Breaker Status."""
# Annahme: Circuit Breaker Registry exists
from circuit_breaker import CircuitState
open_circuits = 0 # Aus Registry holen
if open_circuits == 0:
status = HealthStatus.HEALTHY
message = "All circuit breakers closed"
elif open_circuits < 3:
status = HealthStatus.DEGRADED
message = f"{open_circuits} circuits open"
else:
status = HealthStatus.UNHEALTHY
message = f"{open_circuits} circuits open - service degraded"
return HealthCheckResult(
probe="circuit_breakers",
status=status,
latency_ms=0.1,
message=message,
timestamp=time.time(),
details={"open_circuits": open_circuits}
)
async def _check_rate_limiter(self) -> HealthCheckResult:
"""Prüfe Rate Limiter Tokens."""
return HealthCheckResult(
probe="rate_limiter",
status=HealthStatus.HEALTHY,
latency_ms=0.1,
message="Rate limiter operational",
timestamp=time.time()
)
def _check_connection_pool(self) -> HealthCheckResult:
"""Prüfe HTTP Connection Pool."""
return HealthCheckResult(
probe="connection_pool",
status=HealthStatus.HEALTHY,
latency_ms=0.1,
message="Connection pool healthy",
timestamp=time.time()
)P99 Latenz-Monitoring Dashboard
Für unser HolySheep Gateway habe ich ein Prometheus + Grafana Dashboard entwickelt, das P50, P95, P99 und P99.9 Latenzen in Echtzeit trackt.
# prometheus.yml - HolySheep Gateway Monitoring
global:
scrape_interval: 15s
evaluation_interval: 15s
alerting:
alertmanagers:
- static_configs:
- targets:
- alertmanager:9093
rule_files:
- "alert_rules.yml"
scrape_configs:
- job_name: 'holysheep-gateway'
static_configs:
- targets: ['gateway:9090']
metrics_path: '/metrics'
scrape_interval: 5s
- job_name: 'holysheep-health'
static_configs:
- targets: ['gateway:8080']
metrics_path: '/health/prometheus'
scrape_interval: 10s
- job_name: 'prometheus'
static_configs:
- targets: ['localhost:9090']# alert_rules.yml - Prometheus Alert Rules für HolySheep Gateway
groups:
- name: holysheep_gateway_alerts
rules:
# P99 Latenz zu hoch
- alert: HolySheepP99LatencyHigh
expr: histogram_quantile(0.99, rate(holysheep_request_duration_seconds_bucket[5m])) > 2
for: 5m
labels:
severity: warning
service: holysheep-gateway
annotations:
summary: "P99 Latenz über 2s"
description: "P99 Latenz beträgt {{ $value }}s (Threshold: 2s)"
# Circuit Breaker geöffnet
- alert: HolySheepCircuitBreakerOpen
expr: holysheep_circuit_breaker_state == 2
for: 1m
labels:
severity: critical
service: holysheep-gateway
annotations:
summary: "Circuit Breaker {{ $labels.name }} geöffnet"
description: "Circuit Breaker für {{ $labels.provider }} ist seit 1min offen"
# Error Rate erhöht
- alert: HolySheepErrorRateHigh
expr: |
(
rate(holysheep_requests_total{status=~"5.."}[5m])
/
rate(holysheep_requests_total[5m])
) > 0.01
for: 5m
labels:
severity: warning
service: holysheep-gateway
annotations:
summary: "Error Rate über 1%"
description: "Error Rate beträgt {{ $value | humanizePercentage }}"
# 503 Service Unavailable
- alert: HolySheep503Errors
expr: rate(holysheep_requests_total{status="503"}[5m]) > 0
for: 2m
labels:
severity: critical
service: holysheep-gateway
annotations:
summary: "503 Service Unavailable Errors"
description: "{{ $value }} Requests/min erhalten 503"
# Gateway Down
- alert: HolySheepGatewayDown
expr: up{job="holysheep-gateway"} == 0
for: 1m
labels:
severity: critical
service: holysheep-gateway
annotations:
summary: "HolySheep Gateway nicht erreichbar"
description: "Gateway Instance {{ $labels.instance }} ist seit 1min down"Integration mit HolySheep API
Das folgende Beispiel zeigt die vollständige Integration meines Gateways mit der HolySheep AI API inklusive aller Best Practices:
# holysheep_gateway_client.py - Production-ready HolySheep API Client
import asyncio
import aiohttp
import logging
from typing import Optional, Dict, Any, List
from circuit_breaker import HolySheepCircuitBreaker, CircuitBreakerConfig, HolySheepAPIError
from health_checks import HolySheepHealthChecker
logger = logging.getLogger(__name__)
class HolySheepGatewayClient:
"""
Production-ready Client für HolySheep AI API Gateway.
Features:
- Automatischer Circuit Breaker mit 502/503/504 Handling
- Multi-Region Failover (China/Global)
- Rate Limiting mit Token Bucket
- P99 Latenz-Tracking
- Retry mit Exponential Backoff
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Unterstützte Modelle mit Preisen (Stand 2026)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 24.0, "currency": "USD"},
"claude-sonnet-4.5": {"input": 15.0, "output": 75.0, "currency": "USD"},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0, "currency": "USD"},
"deepseek-v3.2": {"input": 0.42, "output": 1.68, "currency": "USD"},
}
def __init__(
self,
api_key: str,
max_retries: int = 3,
request_timeout: float = 60.0,
enable_circuit_breaker: bool = True
):
self.api_key = api_key
self.max_retries = max_retries
self.request_timeout = request_timeout
# Circuit Breaker für jeden Provider
self.circuit_breakers: Dict[str, HolySheepCircuitBreaker] = {
"primary": HolySheepCircuitBreaker(
"holysheep-primary",
CircuitBreakerConfig(
failure_threshold=5,
timeout=30.0,
success_threshold=3
)
),
"backup": HolySheepCircuitBreaker(
"holysheep-backup",
CircuitBreakerConfig(
failure_threshold=3,
timeout=15.0,
success_threshold=2
)
)
}
# Health Checker
self.health_checker = HolySheepHealthChecker({"api_key": api_key})
# Connection Pool
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
"""Async Context Manager für Session-Management."""
connector = aiohttp.TCPConnector(
limit=100, # Max connections
limit_per_host=30, # Per-host limit
ttl_dns_cache=300, # DNS cache TTL
keepalive_timeout=30 # Keep-alive
)
timeout = aiohttp.ClientTimeout(total=self.request_timeout)
self._session = aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> Dict[str, Any]:
"""
Sende Chat-Completion Request mit automatischem Retry.
"""
url = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
return await self._execute_with_retry(
url,
payload,
self.circuit_breakers["primary"]
)
async def embedding(
self,
model: str,
input_text: str
) -> Dict[str, Any]:
"""Embedding-Generierung."""
url = f"{self.BASE_URL}/embeddings"
payload = {
"model": model,
"input": input_text
}
return await self._execute_with_retry(
url,
payload,
self.circuit_breakers["primary"]
)
async def _execute_with_retry(
self,
url: str,
payload: Dict[str, Any],
circuit_breaker: HolySheepCircuitBreaker
) -> Dict[str, Any]:
"""
Execute mit Exponential Backoff Retry und Circuit Breaker.
"""
last_exception = None
for attempt in range(self.max_retries):
try:
async def do_request():
async with self._session.post(url, json=payload) as resp:
if resp.status == 502:
raise HolySheepAPIError("Bad Gateway", 502)
elif resp.status == 503:
raise HolySheepAPIError("Service Unavailable", 503)
elif resp.status == 504:
raise HolySheepAPIError("Gateway Timeout", 504)
elif resp.status == 429:
# Rate Limited - retry after
retry_after = resp.headers.get("Retry-After", "1")
raise RateLimitError(f"Rate limited, retry after {retry_after}s")
elif resp.status >= 400:
text = await resp.text()
raise HolySheepAPIError(f"API Error: {text}", resp.status)
return await resp.json()
# Mit Circuit Breaker ausführen
result = await circuit_breaker.call(do_request)
return result
except HolySheepAPIError as e:
last_exception = e
if e.status_code in [502, 503, 504]:
# Diese Errors transient - retry
wait_time = min(2 ** attempt * 0.5, 10)
logger.warning(f"[Attempt {attempt+1}] {e}, waiting {wait_time}s")
await asyncio.sleep(wait_time)
else:
# Andere Errors - nicht retry
raise
except RateLimitError as e:
wait_time = float(str(e).split("retry after ")[1].rstrip("s"))
logger.warning(f"[Attempt {attempt+1}] Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
raise last_exception or HolySheepAPIError("Max retries exceeded", 500)
async def get_available_models(self) -> List[Dict[str, Any]]:
"""Liste verfügbare Modelle mit Preisen."""
url = f"{self.BASE_URL}/models"
async with self._session.get(url) as resp:
if resp.status != 200:
raise HolySheepAPIError(f"Failed to fetch models: {resp.status}", resp.status)
data = await resp.json()
# Anreichern mit Preisen
for model in data.get("data", []):
model_id = model.get("id", "")
# Preis aus MODEL_PRICING holen oder als "varies" markieren
pricing_key = self._find_pricing_key(model_id)
if pricing_key:
model["pricing"] = self.MODEL_PRICING[pricing_key]
else:
model["pricing"] = {"input": "varies", "output": "varies"}
return data.get("data", [])
def _find_pricing_key(self, model_id: str) -> Optional[str]:
"""Finde passenden Pricing-Eintrag für Modell."""
model_lower = model_id.lower()
for key in self.MODEL_PRICING:
if key in model_lower or model_lower in key:
return key
return None
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict[str, float]:
"""Berechne Kosten für Request (USD)."""
pricing_key = self._find_pricing_key(model)
if not pricing_key:
return {"total_usd": 0, "warning": "Unknown model"}
pricing = self.MODEL_PRICING[pricing_key]
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return {
"input_cost": input_cost,
"output_cost": output_cost,
"total_usd": input_cost + output_cost
}
class RateLimitError(Exception):
pass
Benchmark-Funktion
async def benchmark_holysheep():
"""Benchmark HolySheep Gateway Performance."""
import time
import statistics
client = HolySheepGatewayClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
request_timeout=30.0
)