Die Integration von Large Language Models (LLMs) in Produktionsumgebungen bringt erhebliche Herausforderungen mit sich: Unvorhersehbare Rate-Limits, gelegentliche Serverausfälle und die Notwendigkeit, auf API-Fehler reagieren zu müssen, ohne den gesamten Workflow zu unterbrechen. In diesem umfassenden Tutorial konfigurieren wir ein vollständiges Monitoring- und Alerting-System für die HolySheep AI API, das automatische Wiederholungsmechanismen bei HTTP 429-, 502- und 503-Fehlern implementiert und Slack-Benachrichtigungen bei kritischen Ausfällen versendet.
Vergleichstabelle: HolySheep vs. Offizielle API vs. Andere Relay-Dienste
| Merkmal | HolySheep AI | Offizielle API (OpenAI/Anthropic) | Andere Relay-Dienste |
|---|---|---|---|
| Preis GPT-4.1 | $8/MTok | $15-30/MTok | $10-20/MTok |
| Preis Claude Sonnet 4.5 | $15/MTok | $18/MTok | $16-22/MTok |
| DeepSeek V3.2 | $0.42/MTok | Nicht verfügbar | $0.50-1.00/MTok |
| Latenz (P99) | <50ms | 100-300ms | 80-200ms |
| Rate-Limit-Handling | Exponential Backoff integriert | Manuelle Konfiguration | Basic Retry |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur Kreditkarte | Oft nur Kreditkarte |
| Kostenlose Credits | ✓ Ja | ✗ Nein | Selten |
| Wechselkurs | ¥1≈$1 (85%+ Ersparnis) | USD direkt | USD oder EUR |
| Monitoring-Dashboard | ✓ Inklusive | Basic | Extra kostenpflichtig |
| Slack-Integration | ✓ Nativ unterstützt | ✗ Nicht integriert | Webhook-only |
Warum Monitoring und Alerting essentiell sind
In meiner dreijährigen Erfahrung mit API-Integrationen habe ich unzählige Produktionsausfälle erlebt, die durch mangelndes Monitoring verursacht wurden. Ein typisches Szenario: Ein LLM-Client sendet 10.000 Anfragen pro Stunde, aber bei einem temporären 502-Fehler gehen alle fehlgeschlagenen Requests verloren, weil keine automatische Wiederholung implementiert wurde. Dies führt nicht nur zu Datenverlusten, sondern auch zu negativen Kundenerfahrungen und potenziellen Umsatzeinbußen.
Die HolySheep AI Plattform bietet hier entscheidende Vorteile: Dank der <50ms Latenz und der integrierten Rate-Limit-Handhabung reduziert sich das Risiko von 429-Fehlern erheblich. Dennoch ist ein robustes Monitoring-System unerlässlich – selbst bei höchster Zuverlässigkeit können externe Faktoren wie Netzwerkprobleme oder temporäre Überlastungen auftreten.
Architektur des Monitoring-Systems
Unser System besteht aus vier Hauptkomponenten:
- API-Client mit Retry-Logik: Implementiert exponentielles Backoff bei 429, 502, 503
- Rate-Limiter: Verhindert das Überschreiten von API-Limits
- Alert-Manager: Sendet Slack-Benachrichtigungen bei kritischen Fehlern
- Metrics-Collector: Erfasst Latenz, Fehlerraten und Token-Verbrauch
Python-Client-Implementierung mit Retry-Mechanismus
Der folgende Code implementiert einen produktionsreifen API-Client mit automatischer Wiederholung bei temporären Fehlern:
#!/usr/bin/env python3
"""
HolySheep AI API Client mit automatischer Wiederholung und Slack-Alerting
Version: 2.0 | Stand: 2026-05-12
"""
import time
import json
import logging
import httpx
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import asyncio
import aiohttp
Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("HolySheepMonitor")
class ErrorType(Enum):
RATE_LIMIT = 429
BAD_GATEWAY = 502
SERVICE_UNAVAILABLE = 503
TIMEOUT = 408
SERVER_ERROR = 500
CLIENT_ERROR = 400
SUCCESS = 200
@dataclass
class APIResponse:
"""Strukturierte API-Antwort"""
status_code: int
content: Optional[Dict[str, Any]] = None
error_type: ErrorType = ErrorType.SUCCESS
retry_count: int = 0
latency_ms: float = 0.0
timestamp: datetime = field(default_factory=datetime.now)
@property
def is_success(self) -> bool:
return 200 <= self.status_code < 300
@property
def is_retryable(self) -> bool:
return self.status_code in [429, 502, 503, 408, 500]
class HolySheepAPIClient:
"""
Produktionsreifer API-Client für HolySheep AI mit:
- Automatischem Retry mit Exponential Backoff
- Rate-Limiting
- Slack-Benachrichtigungen
- Metriken-Sammlung
"""
BASE_URL = "https://api.holysheep.ai/v1"
MAX_RETRIES = 5
BASE_DELAY = 1.0 # Sekunden
MAX_DELAY = 60.0 # Sekunden
def __init__(
self,
api_key: str,
slack_webhook_url: Optional[str] = None,
rate_limit_per_minute: int = 60
):
self.api_key = api_key
self.slack_webhook_url = slack_webhook_url
self.rate_limit_per_minute = rate_limit_per_minute
# Rate-Limiter State
self.request_times: List[float] = []
# Metriken
self.metrics = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'retries': 0,
'rate_limits': 0,
'latencies': [],
'errors_by_type': defaultdict(int)
}
# Slack Alert Thresholds
self.alert_thresholds = {
'error_rate_percent': 5.0, # Alert wenn >5% Fehlerrate
'latency_p95_ms': 2000, # Alert wenn P95 >2s
'consecutive_failures': 3 # Alert nach 3 konsekutiven Fehlern
}
self._consecutive_failures = 0
logger.info(f"HolySheep Client initialisiert mit Rate-Limit: {rate_limit_per_minute}/min")
def _check_rate_limit(self) -> bool:
"""Prüft ob Rate-Limit erreicht wurde"""
now = time.time()
# Entferne Anfragen, die älter als 1 Minute sind
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.rate_limit_per_minute:
sleep_time = 60 - (now - self.request_times[0])
if sleep_time > 0:
logger.warning(f"Rate-Limit erreicht. Warte {sleep_time:.1f}s")
time.sleep(sleep_time)
self.request_times = [t for t in self.request_times if time.time() - t < 60]
self.request_times.append(time.time())
return True
def _calculate_backoff(self, retry_count: int, status_code: int) -> float:
"""Berechnet Wartezeit für Exponential Backoff"""
# Erhöhte Wartezeit bei Rate-Limits
if status_code == 429:
base_delay = self.BASE_DELAY * 2
else:
base_delay = self.BASE_DELAY
# Exponentielles Backoff mit Jitter
delay = min(base_delay * (2 ** retry_count), self.MAX_DELAY)
jitter = delay * 0.1 * (hash(str(time.time())) % 10) / 10
return delay + jitter
def _send_slack_alert(
self,
title: str,
message: str,
severity: str = "warning",
error_details: Optional[Dict] = None
):
"""Sendet Alert an Slack"""
if not self.slack_webhook_url:
logger.debug("Slack-Webhook nicht konfiguriert, kein Alert")
return
emoji = {
"critical": "🔴",
"warning": "🟡",
"info": "ℹ️"
}.get(severity, "⚪")
payload = {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{emoji} {title}",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Zeit:* {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n*Nachricht:* {message}"
}
}
]
}
if error_details:
payload["blocks"].append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"``json\n{json.dumps(error_details, indent=2)}\n``"
}
})
try:
response = httpx.post(
self.slack_webhook_url,
json=payload,
timeout=10.0
)
response.raise_for_status()
logger.info(f"Slack-Alert gesendet: {title}")
except Exception as e:
logger.error(f"Fehler beim Senden des Slack-Alerts: {e}")
def _record_error(self, error_type: ErrorType):
"""Zeichnet Fehler für Metriken auf"""
self._consecutive_failures += 1
self.metrics['failed_requests'] += 1
self.metrics['errors_by_type'][error_type.name] += 1
# Check für Alert
if self._consecutive_failures >= self.alert_thresholds['consecutive_failures']:
self._send_slack_alert(
title="Konsekutive API-Fehler",
message=f"{self._consecutive_failures} aufeinanderfolgende Fehler erkannt",
severity="critical",
error_details={
"last_error_type": error_type.name,
"consecutive_failures": self._consecutive_failures
}
)
def _record_success(self):
"""Zeichnet erfolgreiche Anfrage auf"""
self._consecutive_failures = 0
self.metrics['successful_requests'] += 1
async def chat_completion_async(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000
) -> APIResponse:
"""Asynchroner API-Aufruf mit Retry-Logik"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
for attempt in range(self.MAX_RETRIES):
self._check_rate_limit()
start_time = time.time()
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
latency = (time.time() - start_time) * 1000
self.metrics['total_requests'] += 1
self.metrics['latencies'].append(latency)
response_data = await response.json()
if response.status == 200:
self._record_success()
return APIResponse(
status_code=200,
content=response_data,
latency_ms=latency
)
error_type = ErrorType(response.status)
if not error_type == ErrorType.RATE_LIMIT:
self._record_error(error_type)
# Retry-Logik
if attempt < self.MAX_RETRIES - 1 and response.status in [429, 502, 503, 500]:
delay = self._calculate_backoff(attempt, response.status)
self.metrics['retries'] += 1
logger.warning(
f"Retry {attempt + 1}/{self.MAX_RETRIES} nach "
f"{response.status} in {delay:.1f}s"
)
await asyncio.sleep(delay)
continue
# Finaler Fehler
return APIResponse(
status_code=response.status,
error_type=error_type,
latency_ms=latency
)
except aiohttp.ClientError as e:
logger.error(f"Netzwerkfehler: {e}")
if attempt < self.MAX_RETRIES - 1:
delay = self._calculate_backoff(attempt, 0)
await asyncio.sleep(delay)
continue
return APIResponse(
status_code=0,
error_type=ErrorType.SERVER_ERROR
)
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000
) -> APIResponse:
"""Synchroner Wrapper für API-Aufruf"""
return asyncio.run(self.chat_completion_async(
model, messages, temperature, max_tokens
))
def get_metrics_report(self) -> Dict[str, Any]:
"""Generiert Metriken-Bericht"""
latencies = self.metrics['latencies']
if latencies:
latencies_sorted = sorted(latencies)
p50 = latencies_sorted[len(latencies_sorted) // 2]
p95 = latencies_sorted[int(len(latencies_sorted) * 0.95)]
p99 = latencies_sorted[int(len(latencies_sorted) * 0.99)]
else:
p50 = p95 = p99 = 0
total = self.metrics['total_requests']
error_rate = (self.metrics['failed_requests'] / total * 100) if total > 0 else 0
# Alert bei hoher Fehlerrate
if error_rate > self.alert_thresholds['error_rate_percent']:
self._send_slack_alert(
title="Hohe API-Fehlerrate",
message=f"Fehlerrate von {error_rate:.1f}% erkannt (Schwellwert: {self.alert_thresholds['error_rate_percent']}%)",
severity="warning"
)
return {
"timestamp": datetime.now().isoformat(),
"total_requests": total,
"successful": self.metrics['successful_requests'],
"failed": self.metrics['failed_requests'],
"retries": self.metrics['retries'],
"error_rate_percent": round(error_rate, 2),
"latency_p50_ms": round(p50, 2),
"latency_p95_ms": round(p95, 2),
"latency_p99_ms": round(p99, 2),
"errors_by_type": dict(self.metrics['errors_by_type'])
}
==================== BEISPIEL-NUTZUNG ====================
if __name__ == "__main__":
# API-Key aus Umgebung oder direkt
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
# Client initialisieren
client = HolySheepAPIClient(
api_key=API_KEY,
slack_webhook_url=SLACK_WEBHOOK,
rate_limit_per_minute=100
)
# Beispiel: Chat-Completion aufrufen
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre mir die Vorteile von Exponential Backoff."}
]
response = client.chat_completion(
model="gpt-4.1",
messages=messages,
temperature=0.7,
max_tokens=500
)
if response.is_success:
print(f"✓ Erfolgreich in {response.latency_ms:.0f}ms")
print(f"Antwort: {response.content['choices'][0]['message']['content']}")
else:
print(f"✗ Fehlgeschlagen: {response.status_code} - {response.error_type.name}")
# Metriken ausgeben
print("\n📊 Metriken-Bericht:")
print(json.dumps(client.get_metrics_report(), indent=2))Rate-Limiter und Request-Queue mit Priority
Für komplexere Anwendungen mit unterschiedlichen Prioritäten implementieren wir einen erweiterten Rate-Limiter mit Request-Queue und Priority-System:
#!/usr/bin/env python3
"""
Erweiterter Rate-Limiter mit Priority-Queue für HolySheep API
Unterstützt burst-Handling und automatische Lastverteilung
"""
import time
import threading
import queue
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Callable, Any, List
from enum import IntEnum
import logging
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("RateLimiter")
class Priority(IntEnum):
"""Prioritätsstufen für API-Anfragen"""
CRITICAL = 0 # Sofortige Verarbeitung
HIGH = 1 # Hohe Priorität
NORMAL = 2 # Standard-Priorität
LOW = 3 # Niedrige Priorität (Batch-Verarbeitung)
@dataclass
class QueuedRequest:
"""Struktur für gereihte Anfragen"""
id: str
priority: Priority
created_at: datetime = field(default_factory=datetime.now)
callback: Optional[Callable] = None
args: tuple = ()
kwargs: dict = field(default_factory=dict)
retry_count: int = 0
def __lt__(self, other):
# Prioritäts-Warteschlange sortiert nach Priorität, dann nach Zeit
if self.priority != other.priority:
return self.priority < other.priority
return self.created_at < other.created_at
class HolySheepRateLimiter:
"""
Token-Bucket Rate Limiter mit Priority-Queue
Features:
- Token-Bucket Algorithmus für smooth rate limiting
- Priority-Queue für wichtige Anfragen
- Burst-Handling für plötzliche Lastspitzen
- Thread-safe für Multi-Threading-Anwendungen
"""
def __init__(
self,
requests_per_minute: int = 60,
requests_per_second: int = 10,
burst_size: int = 20
):
self.rpm_limit = requests_per_minute
self.rps_limit = requests_per_second
self.burst_size = burst_size
# Token-Buckets
self.tokens_per_minute = requests_per_minute
self.tokens_per_second = requests_per_second
self.bucket_minute = float(requests_per_minute)
self.bucket_second = float(requests_per_second)
# Timing
self.last_refill_minute = time.time()
self.last_refill_second = time.time()
# Priority Queue
self.request_queue = queue.PriorityQueue()
self.processing_lock = threading.Lock()
# Metriken
self.stats = {
'total_requests': 0,
'processed': 0,
'rejected': 0,
'queued': 0,
'avg_wait_time': 0.0,
'queue_depth': []
}
# Worker-Thread
self._running = False
self._worker_thread: Optional[threading.Thread] = None
logger.info(
f"RateLimiter initialisiert: {requests_per_minute}/min, "
f"{requests_per_second}/s, Burst: {burst_size}"
)
def _refill_buckets(self):
"""Füllt die Token-Buckets basierend auf vergangener Zeit auf"""
now = time.time()
# Minute-Bucket auffüllen
elapsed_minute = now - self.last_refill_minute
if elapsed_minute >= 1.0:
refill_amount = elapsed_minute * self.tokens_per_minute / 60.0
self.bucket_minute = min(
self.tokens_per_minute,
self.bucket_minute + refill_amount
)
self.last_refill_minute = now
# Second-Bucket auffüllen
elapsed_second = now - self.last_refill_second
if elapsed_second >= 0.1: # Alle 100ms auffüllen
refill_amount = elapsed_second * self.tokens_per_second
self.bucket_second = min(
self.burst_size,
self.bucket_second + refill_amount
)
self.last_refill_second = now
def _can_proceed(self) -> bool:
"""Prüft ob Anfrage durchgeführt werden kann"""
self._refill_buckets()
return self.bucket_minute >= 1.0 and self.bucket_second >= 1.0
def _consume_tokens(self):
"""Verbraucht ein Token aus beiden Buckets"""
self.bucket_minute -= 1.0
self.bucket_second -= 1.0
def _wait_for_token(self, timeout: Optional[float] = None) -> bool:
"""Blockiert bis Token verfügbar oder Timeout erreicht"""
start_time = time.time()
while True:
if self._can_proceed():
self._consume_tokens()
return True
if timeout and (time.time() - start_time) > timeout:
return False
# Wartezeit basierend auf Bucket-Füllstand
wait_time = min(
(1.0 - self.bucket_second) / self.tokens_per_second + 0.01,
(1.0 - self.bucket_minute) / (self.tokens_per_minute / 60.0) + 0.01
)
time.sleep(max(0.01, wait_time))
def enqueue(
self,
request_id: str,
priority: Priority = Priority.NORMAL,
callback: Optional[Callable] = None,
timeout: float = 60.0,
*args,
**kwargs
) -> QueuedRequest:
"""Reiht Anfrage in Priority-Queue ein"""
request = QueuedRequest(
id=request_id,
priority=priority,
callback=callback,
args=args,
kwargs=kwargs
)
self.request_queue.put(request)
self.stats['queued'] += 1
logger.debug(f"Anfrage {request_id} eingereiht (Priorität: {priority.name})")
return request
def execute_blocking(
self,
request_id: str,
api_call_func: Callable,
priority: Priority = Priority.NORMAL,
*args,
**kwargs
) -> Any:
"""
Führt API-Call mit Rate-Limiting aus (blockierend)
Args:
request_id: Eindeutige ID für die Anfrage
api_call_func: Funktion die den API-Call ausführt
priority: Prioritätsstufe
Returns:
Ergebnis des API-Calls
"""
self.stats['total_requests'] += 1
# Priority-Anpassung für kritische Anfragen
effective_priority = priority
if priority == Priority.CRITICAL:
# Kritische Anfragen überspringen das Rate-Limit nicht komplett,
# erhalten aber bevorzugten Zugang
effective_priority = Priority.HIGH
# Warten auf Rate-Limit-Freigabe
if not self._wait_for_token(timeout=kwargs.pop('timeout', 60.0)):
logger.warning(f"Anfrage {request_id} wegen Timeout abgelehnt")
self.stats['rejected'] += 1
raise TimeoutError(f"Rate-Limit Timeout für Anfrage {request_id}")
# API-Call ausführen
try:
result = api_call_func(*args, **kwargs)
self.stats['processed'] += 1
return result
except Exception as e:
logger.error(f"API-Call fehlgeschlagen: {e}")
raise
def start_background_worker(self, api_call_func: Callable):
"""Startet Hintergrund-Worker für Queue-Verarbeitung"""
self._running = True
def worker():
while self._running:
try:
# Timeout damit wir regelmäßig prüfen können ob wir stoppen sollen
request = self.request_queue.get(timeout=1.0)
start_time = time.time()
# Warten auf Rate-Limit
if self._wait_for_token(timeout=request.kwargs.get('timeout', 60.0)):
try:
result = api_call_func(*request.args, **request.kwargs)
if request.callback:
request.callback(result=result, error=None)
self.stats['processed'] += 1
except Exception as e:
if request.callback:
request.callback(result=None, error=e)
logger.error(f"Queue-Anfrage fehlgeschlagen: {e}")
else:
if request.callback:
request.callback(
result=None,
error=TimeoutError("Rate-Limit Timeout")
)
self.stats['rejected'] += 1
# Queue-Depth tracken
self.stats['queue_depth'].append(self.request_queue.qsize())
# Durchschnittliche Wartezeit berechnen
wait_time = time.time() - start_time
total_wait = self.stats['avg_wait_time'] * (self.stats['processed'] - 1)
self.stats['avg_wait_time'] = (total_wait + wait_time) / self.stats['processed']
except queue.Empty:
continue
except Exception as e:
logger.error(f"Worker-Fehler: {e}")
self._worker_thread = threading.Thread(target=worker, daemon=True)
self._worker_thread.start()
logger.info("Hintergrund-Worker gestartet")
def stop(self):
"""Stoppt den Rate-Limiter und Worker"""
self._running = False
if self._worker_thread:
self._worker_thread.join(timeout=5.0)
logger.info("Rate-Limiter gestoppt")
def get_stats(self) -> dict:
"""Gibt aktuelle Statistiken zurück"""
return {
'total_requests': self.stats['total_requests'],
'processed': self.stats['processed'],
'rejected': self.stats['rejected'],
'queued': self.request_queue.qsize(),
'avg_wait_time_ms': self.stats['avg_wait_time'] * 1000,
'current_bucket_minute': round(self.bucket_minute, 2),
'current_bucket_second': round(self.bucket_second, 2),
'queue_depth_avg': sum(self.stats['queue_depth']) / len(self.stats['queue_depth']) if self.stats['queue_depth'] else 0
}
==================== MONITORING-DASHBOARD ====================
class MonitoringDashboard:
"""Einfaches Terminal-basiertes Monitoring Dashboard"""
def __init__(self, rate_limiter: HolySheepRateLimiter, api_client):
self.rate_limiter = rate_limiter
self.api_client = api_client
self._running = False
def start(self, interval_seconds: int = 10):
"""Startet das Monitoring Dashboard"""
self._running = True
print("\n" + "="*60)
print(" HolySheep API Monitoring Dashboard")
print("="*60)
while self._running:
stats = self.rate_limiter.get_stats()
metrics = self.api_client.get_metrics_report()
print(f"\n📈 Zeit: {datetime.now().strftime('%H:%M:%S')}")
print("-" * 40)
print(f"Requests: {stats['total_requests']} | "
f"Erfolgreich: {stats['processed']} | "
f"Abgelehnt: {stats['rejected']}")
print(f"Queue-Tiefe: {stats['queued']} | "
f"Ø Wartezeit: {stats['avg_wait_time_ms']:.0f}ms")
print("-" * 40)
print(f"Latenz P50: {metrics['latency_p50_ms']:.0f}ms | "
f"P95: {metrics['latency_p95_ms']:.0f}ms | "
f"P99: {metrics['latency_p99_ms']:.0f}ms")
print(f"Fehlerrate: {metrics['error_rate_percent']:.2f}%")
print("-" * 40)
if metrics['errors_by_type']:
print("Fehler nach Typ:")
for error_type, count in metrics['errors_by_type'].items():
print(f" {error_type}: {count}")
print("\nBucket-Status:")
print(f" Minute: {stats['current_bucket_minute']:.0f}/{self.rate_limiter.tokens_per_minute}")
print(f" Second: {stats['current_bucket_second']:.0f}/{self.rate_limiter.burst_size}")
time.sleep(interval_seconds)
def stop(self):
"""Stoppt das Dashboard"""
self._running = False
==================== INTEGRATIONS-BEISPIEL ====================
if __name__ == "__main__":
import httpx
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# Rate-Limiter initialisieren
rate_limiter = HolySheepRateLimiter(
requests_per_minute=120, # 120 Anfragen pro Minute
requests_per_second=5, # Max 5 pro Sekunde
burst_size=10 # Burst von bis zu 10
)
# API-Call Funktion
def make_api_call(model: str, messages: list):
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages
}
response = httpx.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=30.0
)
response.raise_for_status()
return response.json()
# Beispiel: Verschiedene Anfragen mit Prioritäten
messages = [
{"role": "user", "content": "Hallo, wie geht es dir?"}
]
# Kritische Anfrage (z.B. User-Login)
try:
result = rate_limiter.execute_blocking(
Verwandte Ressourcen
Verwandte Artikel