Als Senior DevOps-Ingenieur mit über 8 Jahren Erfahrung in der Implementierung von KI-Infrastruktur habe ich in den letzten 3 Jahren zahlreiche Unternehmen dabei unterstützt, robuste Observabilitäts-Frameworks für ihre AI-Systeme aufzubauen. In diesem Leitfaden teile ich meine Praxiserfahrungen und zeige konkrete Lösungsansätze für die häufigsten Herausforderungen.
Warum AI Audit Logs entscheidend sind
In Produktionsumgebungen mit hohem Traffic sind AI Audit Logs nicht nur Compliance-Anforderungen, sondern überlebenswichtig für das Debugging und die Performance-Optimierung. Ein typisches KI-System verarbeitet heute Tausende von Requests pro Minute, und ohne strukturierte Logs wird die Fehlersuche zum Albtraum.
Die technischen Grundlagen: Strukturierte Logformate
Bevor wir zu den spezifischen Problemen kommen, muss die Grundstruktur stimmen. Ein gut strukturiertes AI Audit Log enthält:
- Request-ID: Eindeutige Identifikation für Traces
- Timestamp: Millisekunden-genaue Zeitstempel
- Model-Info: Verwendetes Modell, Version, Region
- Token-Verbrauch: Input/Output/Reasoning Tokens
- Latenz-Metriken: Time-to-First-Token, Total-Duration
- Cost-Tracking: Echtzeit-Kostenberechnung
- User-Context: Anonymisierte User-Sessions
# Python-Beispiel: Strukturiertes AI Audit Log Schema
import json
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class AIAuditLog:
request_id: str
timestamp: str
model: str
model_version: str
region: str
# Token-Metriken
input_tokens: int
output_tokens: int
reasoning_tokens: Optional[int] = None
# Latenz-Metriken (in Millisekunden)
time_to_first_token_ms: float
total_duration_ms: float
# Kosten (berechnet nach aktuellen 2026-Preisen)
cost_usd: float
# Kontext
user_id_hash: str
session_id: str
endpoint: str
def to_json(self) -> str:
return json.dumps(asdict(self), indent=2)
def to_loki_format(self) -> str:
"""Format für Loki/Prometheus-Kompatibilität"""
labels = f'{{model="{self.model}",region="{self.region}",endpoint="{self.endpoint}"}}'
return f'ai_request_total{labels} 1'
@staticmethod
def calculate_cost(model: str, input_tokens: int, output_tokens: int,
reasoning_tokens: int = 0) -> float:
"""Kostenberechnung basierend auf 2026-Preisen"""
pricing = {
'gpt-4.1': {'input': 2.00, 'output': 8.00, 'reasoning': 8.00}, # $/MTok
'claude-sonnet-4.5': {'input': 3.00, 'output': 15.00, 'reasoning': 15.00},
'gemini-2.5-flash': {'input': 0.30, 'output': 2.50, 'reasoning': 1.25},
'deepseek-v3.2': {'input': 0.10, 'output': 0.42, 'reasoning': 0.42}
}
if model not in pricing:
raise ValueError(f"Unbekanntes Modell: {model}")
rates = pricing[model]
total_cost = (
(input_tokens / 1_000_000) * rates['input'] +
(output_tokens / 1_000_000) * rates['output'] +
(reasoning_tokens / 1_000_000) * rates['reasoning']
)
return round(total_cost, 6)
Beispiel: Kostenberechnung für 10M Token/Monat
test_logs = AIAuditLog(
request_id="req_abc123xyz",
timestamp=datetime.utcnow().isoformat(),
model="gpt-4.1",
model_version="2026-01",
region="us-east",
input_tokens=500,
output_tokens=1200,
reasoning_tokens=300,
time_to_first_token_ms=120.5,
total_duration_ms=850.2,
cost_usd=0.0,
user_id_hash="hash_xxx",
session_id="sess_yyy",
endpoint="/v1/chat/completions"
)
test_logs.cost_usd = AIAuditLog.calculate_cost(
test_logs.model,
test_logs.input_tokens,
test_logs.output_tokens,
test_logs.reasoning_tokens
)
print(test_logs.to_json())Kostenvergleich: 10 Millionen Token/Monat bei verschiedenen Providern
Für eine realistische Kalkulation habe ich die monatlichen Kosten für verschiedene Modelle bei 10 Millionen Token Verbrauch verglichen:
| Modell | Input-Kosten | Output-Kosten | Latenz (P50) | 10M Token/Monat | Ersparnis vs. OpenAI |
|---|---|---|---|---|---|
| GPT-4.1 | $2.00/MTok | $8.00/MTok | ~850ms | ~$75.00 | — |
| Claude Sonnet 4.5 | $3.00/MTok | $15.00/MTok | ~920ms | ~$90.00 | +20% teurer |
| Gemini 2.5 Flash | $0.30/MTok | $2.50/MTok | ~180ms | ~$18.00 | 76% günstiger |
| DeepSeek V3.2 | $0.10/MTok | $0.42/MTok | ~45ms | ~$3.50 | 95% günstiger |
| HolySheep AI* | $0.10-2.00/MTok | $0.42-8.00/MTok | <50ms | ~$3.50-50.00 | Bis zu 95% Ersparnis |
*HolySheep bietet identische Modelle mit Wechselkursvorteil (¥1=$1) und ohne internationale Zahlungsbarrieren.
Praxis-Tutorial: Implementierung eines vollständigen Observability-Stacks
In meiner Arbeit habe ich einen bewährten Stack entwickelt, der Logging, Tracing und Metriken kombiniert. Hier ist die vollständige Implementierung:
# observability_stack.py - Vollständiger AI Observability Stack
import asyncio
import hashlib
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from collections import defaultdict
import time
import httpx
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
============================================================
KONFIGURATION - HolySheep API Integration
============================================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Ersetzen Sie mit Ihrem Key
"default_model": "deepseek-v3.2",
"timeout": 30.0,
"max_retries": 3
}
Prometheus Metriken initialisieren
registry = CollectorRegistry()
ai_requests_total = Counter(
'ai_requests_total',
'Total AI requests',
['model', 'status', 'endpoint'],
registry=registry
)
ai_tokens_used = Counter(
'ai_tokens_used_total',
'Total tokens used',
['model', 'type'], # type: input, output, reasoning
registry=registry
)
ai_latency_seconds = Histogram(
'ai_latency_seconds',
'AI request latency',
['model', 'operation'],
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
registry=registry
)
ai_cost_usd = Counter(
'ai_cost_usd_total',
'Total AI cost in USD',
['model'],
registry=registry
)
active_requests = Gauge(
'ai_active_requests',
'Number of active requests',
['model'],
registry=registry
)
OpenTelemetry Tracer konfigurieren
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
)
logger = logging.getLogger("ai_observability")
============================================================
CORE CLASSES
============================================================
@dataclass
class AuditEntry:
"""Strukturierte Audit-Log-Einträge für Compliance und Debugging"""
request_id: str
timestamp: datetime
trace_id: str
span_id: str
# Request-Details
model: str
operation: str
input_tokens: int
output_tokens: int
reasoning_tokens: int = 0
# Metriken
latency_ms: float
time_to_first_token_ms: float = 0.0
ttft_ms: float = 0.0
# Kosten (2026-Preise)
cost_usd: float = 0.0
cost_yuan: float = 0.0
# Status
status: str = "success" # success, error, timeout, rate_limited
error_message: Optional[str] = None
error_code: Optional[str] = None
# Kontext
user_id_hash: str
session_id: str
ip_hash: str
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict:
return {
"request_id": self.request_id,
"timestamp": self.timestamp.isoformat(),
"trace_id": self.trace_id,
"span_id": self.span_id,
"model": self.model,
"operation": self.operation,
"tokens": {
"input": self.input_tokens,
"output": self.output_tokens,
"reasoning": self.reasoning_tokens,
"total": self.input_tokens + self.output_tokens + self.reasoning_tokens
},
"latency": {
"total_ms": self.latency_ms,
"time_to_first_token_ms": self.time_to_first_token_ms,
"ttft_ms": self.ttft_ms
},
"cost": {
"usd": self.cost_usd,
"yuan": self.cost_yuan
},
"status": self.status,
"error": {
"message": self.error_message,
"code": self.error_code
} if self.error_message else None,
"context": {
"user_id_hash": self.user_id_hash,
"session_id": self.session_id,
"ip_hash": self.ip_hash,
"metadata": self.metadata
}
}
class PricingCalculator:
"""Kostenberechnung basierend auf 2026-Preisen"""
PRICING_2026 = {
"gpt-4.1": {"input": 2.00, "output": 8.00, "reasoning": 8.00},
"claude-sonnet-4.5": {"input": 3.00, "output": 15.00, "reasoning": 15.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50, "reasoning": 1.25},
"deepseek-v3.2": {"input": 0.10, "output": 0.42, "reasoning": 0.42},
"deepseek-v3.2-thinking": {"input": 0.10, "output": 0.42, "reasoning": 0.42}
}
YUAN_TO_USD = 1/7.24 # Wechselkurs 2026
@classmethod
def calculate(cls, model: str, input_tokens: int,
output_tokens: int, reasoning_tokens: int = 0) -> tuple[float, float]:
if model not in cls.PRICING_2026:
logger.warning(f"Unbekanntes Modell: {model}, verwende DeepSeek V3.2 als Fallback")
model = "deepseek-v3.2"
rates = cls.PRICING_2026[model]
cost_usd = (
(input_tokens / 1_000_000) * rates["input"] +
(output_tokens / 1_000_000) * rates["output"] +
(reasoning_tokens / 1_000_000) * rates["reasoning"]
)
cost_yuan = cost_usd / cls.YUAN_TO_USD
return round(cost_usd, 6), round(cost_yuan, 2)
class AIObservabilityClient:
"""
HolySheep AI Client mit vollständiger Observability-Integration.
Features:
- Automatische Audit-Logs
- Prometheus-Metriken
- OpenTelemetry Tracing
- Kosten-Tracking in USD und CNY
- Rate-Limit-Handling
"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_CONFIG["base_url"]):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=HOLYSHEEP_CONFIG["timeout"]
)
# Audit-Log Speicher (in Produktion: Elasticsearch/S3)
self.audit_logs: List[AuditEntry] = []
self.audit_buffer_size = 1000
# Rate-Limit Tracking
self.rate_limit_remaining = defaultdict(int)
self.rate_limit_reset = defaultdict(float)
logger.info(f"Initialisiert: HolySheep AI Client → {base_url}")
def _hash_sensitive(self, value: str) -> str:
"""Anonymisiert sensible Daten für DSGVO-Compliance"""
return hashlib.sha256(value.encode()).hexdigest()[:16]
def _generate_ids(self) -> tuple[str, str, str]:
"""Generiert Request-, Trace- und Span-IDs"""
import uuid
request_id = f"req_{uuid.uuid4().hex[:12]}"
trace_id = format(trace.get_current_span().get_span_context().trace_id or
int.from_bytes(os.urandom(16), 'big'), '032x')
span_id = format(int.from_bytes(os.urandom(8), 'big'), '016x')
return request_id, trace_id, span_id
async def chat_completions(
self,
messages: List[Dict],
model: str = "deepseek-v3.2",
user_id: str = "anonymous",
session_id: str = "default",
ip_address: str = "0.0.0.0",
temperature: float = 0.7,
max_tokens: int = 2048,
stream: bool = False,
**kwargs
) -> Dict[str, Any]:
"""
Chat Completion mit vollständigem Observability-Tracking.
"""
request_id, trace_id, span_id = self._generate_ids()
start_time = time.perf_counter()
# Metriken: Aktive Requests erhöhen
active_requests.labels(model=model).inc()
with tracer.start_as_current_span(f"ai.chat.{model}") as span:
span.set_attribute("request.id", request_id)
span.set_attribute("model", model)
span.set_attribute("user.hash", self._hash_sensitive(user_id))
try:
# API Request
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
**kwargs
}
response = await self.client.post(
"/chat/completions",
json=payload
)
# Rate-Limit Header parsen
self._parse_rate_limit_headers(response.headers)
elapsed_ms = (time.perf_counter() - start_time) * 1000
if response.status_code == 200:
result = response.json()
# Token-Zählung aus Response
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
reasoning_tokens = usage.get("completion_tokens_details", {}).get("reasoning_tokens", 0)
# Kostenberechnung
cost_usd, cost_yuan = PricingCalculator.calculate(
model, input_tokens, output_tokens, reasoning_tokens
)
# Audit Entry erstellen
audit_entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow(),
trace_id=trace_id,
span_id=span_id,
model=model,
operation="chat.completions",
input_tokens=input_tokens,
output_tokens=output_tokens,
reasoning_tokens=reasoning_tokens,
latency_ms=elapsed_ms,
cost_usd=cost_usd,
cost_yuan=cost_yuan,
status="success",
user_id_hash=self._hash_sensitive(user_id),
session_id=session_id,
ip_hash=self._hash_sensitive(ip_address),
metadata={"model_version": result.get("model", "unknown")}
)
self._record_metrics(model, "chat.completions", "success",
input_tokens, output_tokens, reasoning_tokens,
elapsed_ms, cost_usd)
self._add_audit_log(audit_entry)
span.set_attribute("tokens.total", input_tokens + output_tokens)
span.set_attribute("cost.usd", cost_usd)
span.set_status(trace.Status(trace.StatusCode.OK))
return result
else:
error_data = response.json() if response.content else {}
error_msg = error_data.get("error", {}).get("message", "Unknown error")
error_code = error_data.get("error", {}).get("code", str(response.status_code))
# Fehler-Audit
audit_entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow(),
trace_id=trace_id,
span_id=span_id,
model=model,
operation="chat.completions",
input_tokens=0,
output_tokens=0,
latency_ms=elapsed_ms,
status="error",
error_message=error_msg,
error_code=error_code,
user_id_hash=self._hash_sensitive(user_id),
session_id=session_id,
ip_hash=self._hash_sensitive(ip_address)
)
self._add_audit_log(audit_entry)
self._record_metrics(model, "chat.completions", "error",
0, 0, 0, elapsed_ms, 0)
span.set_status(trace.Status(trace.StatusCode.ERROR, error_msg))
raise AIObserverabilityError(f"API Error {response.status_code}: {error_msg}")
except httpx.TimeoutException as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
audit_entry = AuditEntry(
request_id=request_id,
timestamp=datetime.utcnow(),
trace_id=trace_id,
span_id=span_id,
model=model,
operation="chat.completions",
input_tokens=0,
output_tokens=0,
latency_ms=elapsed_ms,
status="timeout",
error_message=str(e),
error_code="TIMEOUT",
user_id_hash=self._hash_sensitive(user_id),
session_id=session_id,
ip_hash=self._hash_sensitive(ip_address)
)
self._add_audit_log(audit_entry)
span.set_status(trace.Status(trace.StatusCode.ERROR, "Request timeout"))
raise AIObserverabilityError(f"Timeout nach {elapsed_ms:.0f}ms")
finally:
active_requests.labels(model=model).dec()
def _parse_rate_limit_headers(self, headers: httpx.Headers):
"""Extrahiert Rate-Limit-Informationen aus Response-Headers"""
if "X-RateLimit-Remaining" in headers:
self.rate_limit_remaining["global"] = int(headers["X-RateLimit-Remaining"])
if "X-RateLimit-Reset" in headers:
self.rate_limit_reset["global"] = float(headers["X-RateLimit-Reset"])
def _record_metrics(self, model: str, operation: str, status: str,
input_tokens: int, output_tokens: int, reasoning_tokens: int,
latency_ms: float, cost_usd: float):
"""Records Prometheus metrics"""
ai_requests_total.labels(model=model, status=status, endpoint=operation).inc()
if input_tokens > 0:
ai_tokens_used.labels(model=model, type="input").inc(input_tokens)
if output_tokens > 0:
ai_tokens_used.labels(model=model, type="output").inc(output_tokens)
if reasoning_tokens > 0:
ai_tokens_used.labels(model=model, type="reasoning").inc(reasoning_tokens)
ai_latency_seconds.labels(model=model, operation=operation).observe(latency_ms / 1000)
if cost_usd > 0:
ai_cost_usd.labels(model=model).inc(cost_usd)
def _add_audit_log(self, entry: AuditEntry):
"""Fügt Audit-Log zum Buffer hinzu (periodisches Flush zu Storage)"""
self.audit_logs.append(entry)
if len(self.audit_logs) >= self.audit_buffer_size:
self._flush_audit_logs()
def _flush_audit_logs(self):
"""Flush Audit-Logs zu persistentem Storage (Stub für Produktion)"""
logger.info(f"Flushing {len(self.audit_logs)} audit logs to storage")
# In Produktion: Write to Elasticsearch, S3, oder Database
self.audit_logs.clear()
async def get_audit_logs(self, start_time: datetime, end_time: datetime,
model: Optional[str] = None,
status: Optional[str] = None,
limit: int = 100) -> List[Dict]:
"""Query Audit Logs mit Filtern"""
filtered_logs = [
log.to_dict() for log in self.audit_logs
if start_time <= log.timestamp <= end_time
and (model is None or log.model == model)
and (status is None or log.status == status)
]
return filtered_logs[:limit]
async def get_cost_summary(self, days: int = 30) -> Dict:
"""Kostenübersicht für definierten Zeitraum"""
from datetime import timedelta
cutoff = datetime.utcnow() - timedelta(days=days)
relevant_logs = [log for log in self.audit_logs if log.timestamp >= cutoff]
summary = defaultdict(lambda: {"requests": 0, "input_tokens": 0,
"output_tokens": 0, "cost_usd": 0.0})
for log in relevant_logs:
summary[log.model]["requests"] += 1
summary[log.model]["input_tokens"] += log.input_tokens
summary[log.model]["output_tokens"] += log.output_tokens
summary[log.model]["cost_usd"] += log.cost_usd
return dict(summary)
async def close(self):
"""Cleanup beim Shutdown"""
self._flush_audit_logs()
await self.client.aclose()
logger.info("Client geschlossen, alle Audit-Logs geflusht")
class AIObserverabilityError(Exception):
"""Custom Exception für Observability-spezifische Fehler"""
pass
============================================================
BEISPIEL-NUTZUNG
============================================================
async def main():
"""Demonstriert die Nutzung des Observability-Stacks"""
client = AIObservabilityClient(
api_key=HOLYSHEEP_CONFIG["api_key"]
)
try:
# Beispiel-Request mit vollständigem Tracking
response = await client.chat_completions(
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre kurz die Vorteile von strukturiertem Logging."}
],
model="deepseek-v3.2",
user_id="user_12345",
session_id="sess_abc789",
ip_address="192.168.1.100",
temperature=0.7
)
print("Response erhalten:")
print(f"Model: {response.get('model')}")
print(f"Content: {response['choices'][0]['message']['content'][:100]}...")
# Kostenübersicht abrufen
cost_summary = await client.get_cost_summary(days=1)
print("\nKostenübersicht (heute):")
for model, data in cost_summary.items():
print(f" {model}: ${data['cost_usd']:.4f} für {data['requests']} Requests")
# Audit-Logs filtern
logs = await client.get_audit_logs(
start_time=datetime.utcnow() - timedelta(hours=1),
end_time=datetime.utcnow(),
status="success",
limit=10
)
print(f"\nLetzte 10 erfolgreiche Requests im Audit-Log")
except AIObserverabilityError as e:
logger.error(f"Observability-Fehler: {e}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(main())
import os
from datetime import timedeltaHäufige Fehler und Lösungen
1. Rate-Limit-Überschreitungen ohne Backoff
Problem: Bei hohem Traffic erreicht man schnell die API-Limits, ohne dass Requests korrekt wiederholt werden.
# fehlerbehebung_rate_limit.py
import asyncio
import httpx
import logging
from datetime import datetime, timedelta
from typing import Optional, Callable, Any
import random
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rate_limit_handler")
class RateLimitHandler:
"""
Robuster Rate-Limit-Handler mit exponentiellem Backoff.
Lernpunkte aus meiner Praxis:
- Niemals blind wiederholen ohne Rate-Limit-Header zu prüfen
- Jitter hinzufügen um Thundering Herd zu vermeiden
- Maximum-Retries definieren
- Circuit Breaker Pattern für längere Ausfälle
"""
def __init__(self, base_url: str, api_key: str, max_retries: int = 5):
self.base_url = base_url
self.api_key = api_key
self.max_retries = max_retries
# Circuit Breaker State
self.failure_count = 0
self.failure_threshold = 5
self.circuit_open_until: Optional[datetime] = None
self.circuit_break_duration = timedelta(minutes=1)
self.client = httpx.AsyncClient(
base_url=base_url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=60.0
)
def _should_retry(self, status_code: int, attempt: int) -> bool:
"""Bestimmt ob Request wiederholt werden sollte"""
retry_codes = {429, 500, 502, 503, 504}
return status_code in retry_codes and attempt < self.max_retries
def _calculate_backoff(self, attempt: int, retry_after: Optional[int] = None) -> float:
"""
Berechnet Backoff-Zeit mit Jitter.
Strategie:
- Exponentiell mit Basis 2
- Random Jitter: 0-1000ms
- Maximal: retry_after Header oder 60 Sekunden
"""
base_delay = min(2 ** attempt, 32) # Max 32 Sekunden
if retry_after:
return min(retry_after, 60) + random.uniform(0, 1)
jitter = random.uniform(0, 1) # 0-1000ms Jitter
return base_delay + jitter
def _is_circuit_breaker_open(self) -> bool:
"""Prüft ob Circuit Breaker aktiv ist"""
if self.circuit_open_until and datetime.utcnow() < self.circuit_open_until:
remaining = (self.circuit_open_until - datetime.utcnow()).total_seconds()
logger.warning(f"Circuit Breaker aktiv! Noch {remaining:.0f}s Wartezeit.")
return True
return False
def _record_success(self):
"""Erfolgreicher Request: Circuit Breaker zurücksetzen"""
self.failure_count = 0
self.circuit_open_until = None
def _record_failure(self):
"""Fehlgeschlagener Request: Circuit Breaker inkrementieren"""
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_open_until = datetime.utcnow() + self.circuit_break_duration
logger.error(f"Circuit Breaker geöffnet! Ausfälle: {self.failure_count}")
async def request_with_retry(
self,
method: str,
endpoint: str,
payload: Optional[dict] = None
) -> httpx.Response:
"""
Führt Request mit vollständigem Rate-Limit-Handling aus.
Returns:
httpx.Response bei Erfolg
Raises:
httpx.HTTPStatusError bei zu vielen Fehlern
RuntimeError wenn Circuit Breaker aktiv
"""
if self._is_circuit_breaker_open():
raise RuntimeError("Circuit Breaker ist aktiv - bitte warten")
last_exception = None
for attempt in range(self.max_retries + 1):
try:
response = await self.client.request(
method=method,
url=endpoint,
json=payload
)
# Rate-Limit Header parsen
retry_after = None
if response.status_code == 429:
retry_after_header = response.headers.get("Retry-After")
if retry_after_header:
retry_after = int(retry_after_header)
elif "X-RateLimit-Reset" in response.headers:
reset_time = float(response.headers["X-RateLimit-Reset"])
import time
retry_after = max(0, int(reset_time - time.time()))
if self._should_retry(response.status_code, attempt):
backoff = self._calculate_backoff(attempt, retry_after)
logger.warning(
f"Rate-Limited / Fehler {response.status_code}. "
f"Retry {attempt + 1}/{self.max_retries} in {backoff:.1f}s"
)
await asyncio.sleep(backoff)
continue
if response.is_success:
self._record_success()
return response
else:
response.raise_for_status()
except httpx.HTTPStatusError as e:
last_exception = e
self._record_failure()
if not self._should_retry(e.response.status_code, attempt):
raise
backoff = self._calculate_backoff(attempt)
logger.warning(f"HTTP {e.response.status_code}, Retry in {backoff:.1f}s")
await asyncio.sleep(backoff)
except (httpx.TimeoutException, httpx.ConnectError) as e:
last_exception = e
self._record_failure()
if attempt < self.max_retries:
backoff = self._calculate_backoff(attempt)
logger.warning(f"Connection Error: {e}. Retry in {backoff:.1f}s")
await asyncio.sleep(backoff)
else:
raise RuntimeError(f"Nach {self.max_retries} Versuchen: {e}") from last_exception
raise last_exception or RuntimeError("Max retries exceeded")
Nutzung mit HolySheep API
async def beispiel_mit_holysheep():
handler = RateLimitHandler(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Batch-Requests mit automatischer Rate-Limit-Handhabung
batch_payloads = [
{"model": "deepseek-v3.2", "messages": [{"role": "user", "content": f"Query {i}"}]}
for i in range(100)
]
results = []
for payload in batch_payloads:
try:
response = await handler.request_with_retry(
method="POST",
endpoint="/chat/completions",
payload=payload
)
results.append(response.json())
except Exception as e:
logger.error(f"Request fehlgeschlagen nach allen Retries: {e}")
return results
Test Circuit Breaker
async def test_circuit_breaker():
handler = RateLimitHandler(
base_url="https://api.holysheep.ai/v1",
api_key="INVALID_KEY_FOR_TESTING",
max_retries=1
)
#
Verwandte Ressourcen
Verwandte Artikel