Das Szenario: Wenn Ihre Anwendung plötzlich den Dienst verweigert
Es ist 14:32 Uhr an einem Dienstagnachmittag. Ihre Produktions-API empfängt plötzlich Hunderte von Anfragen pro Sekunde — ein unerwarteter Traffic-Peak durch virale Social-Media-Posts. Innerhalb von Sekunden erscheint in Ihren Logs:
ConnectionError: timeout after 30s — API request failed
HTTP 429: Too Many Requests — Rate limit exceeded
Retry-After: 60 seconds
Was nun folgt, ist einpanicischer Sturm von Retry-Versuchen, der die Situation verschlimmert. Genau dieses Szenario erlebte unser Team vor drei Monaten bei einem Kundenprojekt. Die Lösung: ein robustes Rate-Limiting-System, das wir in diesem Tutorial vollständig implementieren werden.
Warum Rate Limiting existiert: Die HolySheep AI Perspektive
Als Entwickler der
HolySheep AI API verstehen wir die technischen Herausforderungen beider Seiten. Unsere Infrastruktur bietet eine durchschnittliche Latenz von unter 50ms — ein Wert, der nur durch strenge Rate-Limit-Policies maintainbar bleibt. Bei Wechselkursen von ¥1 pro Dollar (über 85% Ersparnis gegenüber westlichen Anbietern) ist unser GPT-4.1 Endpunkt zu $8/MTok zwar konkurrenzfähig, aber selbst diese Effizienz erfordert faire Nutzungslimits.
Grundlagen: Was bedeutet Rate Limiting?
Rate Limiting kontrolliert die Anzahl der API-Anfragen, die ein Client in einem bestimmten Zeitfenster senden darf. Die zwei dominierenden Algorithmen sind:
- Token Bucket: Erlaubt Bursts, aber begrenzt den langfristigen Durchsatz
- Sliding Window: Glattere Verteilung, präzisere Grenzen pro Zeitfenster
Token Bucket Algorithmus: Der Klassiker
Der Token Bucket funktioniert wie ein echter Eimer: Tokens (Anfragen) werden mit konstanter Rate nachgefüllt, bis der Eimer voll ist. Wenn der Eimer leer ist, werden Anfragen abgelehnt.
import time
import threading
from typing import Optional
from dataclasses import dataclass
@dataclass
class TokenBucketConfig:
"""Konfiguration für Token Bucket Rate Limiter"""
capacity: int # Maximale Anzahl Tokens im Eimer
refill_rate: float # Tokens pro Sekunde
tokens: float # Aktuelle Token-Anzahl
last_refill: float # Zeitstempel der letzten Auffüllung
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
class TokenBucketRateLimiter:
"""
Token Bucket Implementation für API Rate Limiting.
Erlaubt Bursts bis zur Kapazität, limitiert aber den Durchschnitt.
"""
def __init__(self, capacity: int = 100, refill_rate: float = 10.0):
"""
Args:
capacity: Maximale Anzahl Tokens (Burst-Größe)
refill_rate: Tokens die pro Sekunde hinzugefügt werden
"""
self._config = TokenBucketConfig(
capacity=capacity,
refill_rate=refill_rate,
tokens=float(capacity),
last_refill=time.time()
)
self._lock = threading.Lock()
def _refill(self) -> None:
"""Auffüllen basierend auf vergangener Zeit"""
now = time.time()
elapsed = now - self._config.last_refill
# Neue Tokens berechnen
new_tokens = elapsed * self._config.refill_rate
self._config.tokens = min(
self._config.capacity,
self._config.tokens + new_tokens
)
self._config.last_refill = now
def acquire(self, tokens: int = 1, blocking: bool = False, timeout: Optional[float] = None) -> bool:
"""
Versucht Tokens zu erwerben.
Args:
tokens: Anzahl benötigter Tokens
blocking: Ob auf verfügbare Tokens gewartet werden soll
timeout: Maximale Wartezeit in Sekunden
Returns:
True wenn Tokens erworben, False sonst
"""
deadline = time.time() + timeout if timeout else None
with self._lock:
while True:
self._refill()
if self._config.tokens >= tokens:
self._config.tokens -= tokens
return True
if not blocking:
return False
if deadline and time.time() >= deadline:
return False
# Wartezeit berechnen
wait_time = (tokens - self._config.tokens) / self._config.refill_rate
if deadline:
wait_time = min(wait_time, deadline - time.time())
# Lock kurz loslassen für andere Threads
self._lock.release()
try:
time.sleep(min(wait_time, 0.1))
finally:
self._lock.acquire()
def get_available_tokens(self) -> float:
"""Gibt die aktuelle Anzahl verfügbarer Tokens zurück"""
with self._lock:
self._refill()
return self._config.tokens
Beispiel: 100 Anfragen Burst, 10 Anfragen/Sekunde durchschnittlich
rate_limiter = TokenBucketRateLimiter(capacity=100, refill_rate=10)
Nicht-blockierender Erwerb
if rate_limiter.acquire(tokens=1):
print("Anfrage erlaubt - Token verbraucht")
else:
print("Rate Limit erreicht - bitte warten")
print(f"Verfügbare Tokens: {rate_limiter.get_available_tokens():.2f}")
Sliding Window Algorithmus: Präzise Kontrolle
Der Sliding Window Algorithmus teilt die Zeit in Segmente und berechnet die Anfragen im "gleitenden Fenster". Dies verhindert die Burst-Problematik am Fensterrand.
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Tuple
import threading
@dataclass
class SlidingWindowConfig:
"""Konfiguration für Sliding Window Rate Limiter"""
max_requests: int # Maximale Anfragen im Fenster
window_size: float # Fenstergröße in Sekunden
requests: Deque[float] = field(default_factory=deque)
def __post_init__(self):
self.requests = deque()
class SlidingWindowRateLimiter:
"""
Sliding Window Counter Implementation.
Zählt Anfragen im aktuellen Zeitfenster - keine Bursts möglich.
"""
def __init__(self, max_requests: int = 100, window_size: float = 60.0):
"""
Args:
max_requests: Maximale Anfragen im Zeitfenster
window_size: Fenstergröße in Sekunden
"""
self._config = SlidingWindowConfig(
max_requests=max_requests,
window_size=window_size
)
self._lock = threading.Lock()
self._last_cleanup = time.time()
def _cleanup_old_requests(self, current_time: float) -> None:
"""Entfernt Anfragen außerhalb des aktuellen Fensters"""
window_start = current_time - self._config.window_size
# Alle alten Timestamps entfernen
while self._config.requests and self._config.requests[0] < window_start:
self._config.requests.popleft()
self._last_cleanup = current_time
def _cleanup_if_needed(self) -> None:
"""Periodisches Cleanup optimieren"""
now = time.time()
# Nur alle 100ms aufräumen statt bei jedem Aufruf
if now - self._last_cleanup > 0.1:
self._cleanup_old_requests(now)
def acquire(self, requests: int = 1, blocking: bool = False,
timeout: Optional[float] = None) -> bool:
"""
Versucht Anfragen im aktuellen Fenster zu platzieren.
Args:
requests: Anzahl benötigter Anfragen-Slots
blocking: Ob auf freien Slot gewartet werden soll
timeout: Maximale Wartezeit
Returns:
True wenn Anfragen erlaubt, False sonst
"""
deadline = time.time() + timeout if timeout else None
while True:
with self._lock:
self._cleanup_if_needed()
current_time = time.time()
self._cleanup_old_requests(current_time)
# Prüfen ob genug Platz im Fenster
available = self._config.max_requests - len(self._config.requests)
if available >= requests:
# Anfragen registrieren
for _ in range(requests):
self._config.requests.append(current_time)
return True
if not blocking:
return False
if deadline and time.time() >= deadline:
return False
# Zeit bis ein Slot frei wird berechnen
oldest_request = self._config.requests[0]
wait_time = (oldest_request + self._config.window_size) - current_time
if deadline:
wait_time = min(wait_time, deadline - time.time())
# Außerhalb des Locks warten
if wait_time > 0:
time.sleep(min(wait_time, 0.1))
def get_current_usage(self) -> Tuple[int, float]:
"""Gibt aktuelle Nutzung und Sekunden bis Reset zurück"""
with self._lock:
self._cleanup_if_needed()
current_time = time.time()
self._cleanup_old_requests(current_time)
used = len(self._config.requests)
remaining = self._config.max_requests - used
# Zeit bis älteste Anfrage aus dem Fenster fällt
if self._config.requests:
oldest = self._config.requests[0]
reset_in = max(0, (oldest + self._config.window_size) - current_time)
else:
reset_in = 0.0
return remaining, reset_in
Beispiel: 100 Anfragen pro Minute, streng limitiert
sw_limiter = SlidingWindowRateLimiter(max_requests=100, window_size=60.0)
Blockierender Erwerb mit Timeout
if sw_limiter.acquire(requests=5, blocking=True, timeout=10.0):
print("5 Anfragen erfolgreich im Fenster platziert")
remaining, reset_in = sw_limiter.get_current_usage()
print(f"Verbleibend: {remaining}, Reset in: {reset_in:.1f}s")
else:
print("Timeout erreicht - Rate Limit streng eingehalten")
Hybrid-Implementierung: Token Bucket mit Sliding Window Log
Die mächtigste Lösung kombiniert beide Algorithmen für maximale Flexibilität:
import time
import threading
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, Optional, List
import hashlib
@dataclass
class RequestLog:
"""Log für Sliding Window Referenz"""
timestamps: Deque[float] = field(default_factory=deque)
class HybridRateLimiter:
"""
Kombiniert Token Bucket (für Bursts) mit Sliding Window Log (für Abrechnung).
Ideal für API Gateways mit unterschiedlichen Limit-Typen.
"""
def __init__(
self,
bucket_capacity: int = 50, # Burst-Kapazität
refill_rate: float = 5.0, # Tokens/Sekunde
window_max: int = 100, # Max im Sliding Window
window_size: float = 60.0 # Fenster in Sekunden
):
self._bucket_capacity = bucket_capacity
self._refill_rate = refill_rate
self._window_max = window_max
self._window_size = window_size
# Token Bucket State
self._tokens = float(bucket_capacity)
self._last_refill = time.time()
# Sliding Window Logs (pro Client)
self._request_logs: Dict[str, RequestLog] = {}
# Lock für Thread-Safety
self._lock = threading.Lock()
# Cleanup Intervalle
self._last_global_cleanup = time.time()
def _refill_bucket(self) -> None:
"""Auffüllen des Token Buckets"""
now = time.time()
elapsed = now - self._last_refill
self._tokens = min(
self._bucket_capacity,
self._tokens + elapsed * self._refill_rate
)
self._last_refill = now
def _cleanup_window(self, client_id: str, current_time: float) -> None:
"""Entfernt abgelaufene Timestamps aus dem Sliding Window"""
if client_id not in self._request_logs:
return
log = self._request_logs[client_id]
window_start = current_time - self._window_size
while log.timestamps and log.timestamps[0] < window_start:
log.timestamps.popleft()
# Leere Logs entfernen (Speicher-Optimierung)
if not log.timestamps:
del self._request_logs[client_id]
def _global_cleanup(self) -> None:
"""Periodisches Cleanup aller Client-Logs"""
now = time.time()
if now - self._last_global_cleanup < 10.0: # Alle 10s
return
current_window_start = now - self._window_size
expired_clients = []
for client_id, log in self._request_logs.items():
# Elemente vor dem Fenster entfernen
while log.timestamps and log.timestamps[0] < current_window_start:
log.timestamps.popleft()
if not log.timestamps:
expired_clients.append(client_id)
for client_id in expired_clients:
del self._request_logs[client_id]
self._last_global_cleanup = now
def check_rate_limit(
self,
client_id: str,
requests: int = 1,
blocking: bool = False,
timeout: Optional[float] = None
) -> Dict[str, any]:
"""
Prüft Rate Limit für einen Client.
Returns:
Dict mit 'allowed', 'reason', 'retry_after', 'limit_info'
"""
deadline = time.time() + timeout if timeout else None
while True:
with self._lock:
self._global_cleanup()
current_time = time.time()
# 1. Token Bucket prüfen
self._refill_bucket()
# 2. Sliding Window Log prüfen
self._cleanup_window(client_id, current_time)
if client_id not in self._request_logs:
self._request_logs[client_id] = RequestLog()
window_count = len(self._request_logs[client_id].timestamps)
# Beide Limits prüfen
bucket_allowed = self._tokens >= requests
window_allowed = (window_count + requests) <= self._window_max
if bucket_allowed and window_allowed:
# Anfrage erlauben
self._tokens -= requests
for _ in range(requests):
self._request_logs[client_id].timestamps.append(current_time)
return {
'allowed': True,
'reason': 'Both limits OK',
'retry_after': 0,
'limit_info': {
'tokens_remaining': self._tokens,
'window_remaining': self._window_max - window_count - requests,
'window_used': window_count + requests
}
}
# Limit erreicht - Gründe analysieren
if not window_allowed:
# Sliding Window ist das engere Limit
oldest = self._request_logs[client_id].timestamps[0]
retry_after = oldest + self._window_size - current_time
reason = f'Sliding window limit: {window_count}/{self._window_max}'
elif not bucket_allowed:
# Token Bucket ist das engere Limit
tokens_needed = requests - self._tokens
retry_after = tokens_needed / self._refill_rate
reason = f'Token bucket empty: {self._tokens:.1f}/{self._bucket_capacity}'
else:
retry_after = 1.0
reason = 'Unknown'
if not blocking or (deadline and time.time() >= deadline):
return {
'allowed': False,
'reason': reason,
'retry_after': max(0, retry_after),
'limit_info': {
'tokens_remaining': self._tokens,
'window_remaining': self._window_max - window_count
}
}
# Wartezeit optimieren
wait = min(retry_after, 0.1) if retry_after > 0 else 0.1
if deadline:
remaining = deadline - time.time()
if remaining <= 0:
return {
'allowed': False,
'reason': 'Timeout',
'retry_after': 0,
'limit_info': {}
}
wait = min(wait, remaining)
time.sleep(max(0, wait))
def get_client_status(self, client_id: str) -> Dict[str, any]:
"""Gibt den aktuellen Status eines Clients zurück"""
with self._lock:
self._refill_bucket()
current_time = time.time()
self._cleanup_window(client_id, current_time)
window_count = len(self._request_logs.get(client_id, RequestLog()).timestamps)
return {
'client_id': client_id,
'tokens_available': self._tokens,
'tokens_capacity': self._bucket_capacity,
'window_used': window_count,
'window_max': self._window_max,
'window_available': self._window_max - window_count
}
Praktisches Beispiel: Multi-Client Rate Limiter
class APIRateLimiter:
"""
Production-ready Rate Limiter für API Gateway Integration.
"""
def __init__(self):
# Verschiedene Limiter für verschiedene Tier
self._limiters: Dict[str, HybridRateLimiter] = {
'free': HybridRateLimiter(
bucket_capacity=10,
refill_rate=1.0,
window_max=50,
window_size=60.0
),
'pro': HybridRateLimiter(
bucket_capacity=100,
refill_rate=10.0,
window_max=1000,
window_size=60.0
),
'enterprise': HybridRateLimiter(
bucket_capacity=500,
refill_rate=50.0,
window_max=10000,
window_size=60.0
)
}
def check(self, client_id: str, tier: str = 'free', requests: int = 1) -> Dict:
"""Prüft Rate Limit für einen Client mit bestimmter Tier"""
limiter = self._limiters.get(tier, self._limiters['free'])
return limiter.check_rate_limit(client_id, requests, blocking=False)
def get_status(self, client_id: str, tier: str = 'free') -> Dict:
"""Gibt Status eines Clients zurück"""
limiter = self._limiters.get(tier, self._limiters['free'])
return limiter.get_client_status(client_id)
Verwendung mit HolySheep AI API
def call_holysheep_api(api_key: str, prompt: str, rate_limiter: APIRateLimiter):
"""Beispiel: Rate-Limited HolySheep AI API Aufruf"""
import json
import urllib.request
# Client ID aus API Key ableiten
client_id = hashlib.md5(api_key.encode()).hexdigest()[:16]
# Rate Limit prüfen
result = rate_limiter.check(client_id, tier='pro', requests=1)
if not result['allowed']:
raise Exception(f"Rate Limit: {result['reason']}. Retry after {result['retry_after']:.1f}s")
# API Request
payload = json.dumps({
'model': 'gpt-4.1',
'messages': [{'role': 'user', 'content': prompt}]
}).encode('utf-8')
req = urllib.request.Request(
'https://api.holysheep.ai/v1/chat/completions',
data=payload,
headers={
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
)
with urllib.request.urlopen(req, timeout=30) as response:
return json.loads(response.read().decode('utf-8'))
Initialisierung
rate_limiter = APIRateLimiter()
Test
try:
result = rate_limiter.check('test-client-123', tier='pro')
print(f"Rate Limit Check: {result}")
except Exception as e:
print(f"Error: {e}")
Integration in Production: Python Async Implementation
Für hochperformante Anwendungen empfehle ich die asyncio-basierte Variante:
import asyncio
import time
from typing import Dict, Optional
from dataclasses import dataclass
import hashlib
try:
import aiohttp
HAS_AIOHTTP = True
except ImportError:
HAS_AIOHTTP = False
@dataclass
class AsyncTokenBucket:
"""Async-fähiger Token Bucket mit Redis-kompatiblem Interface"""
capacity: int
refill_rate: float
tokens: float
last_update: float
@classmethod
async def create(cls, capacity: int, refill_rate: float):
now = time.time()
return cls(
capacity=capacity,
refill_rate=refill_rate,
tokens=float(capacity),
last_update=now
)
async def consume(self, tokens: int = 1) -> bool:
"""Consumiert Tokens wenn verfügbar"""
now = time.time()
# Auffüllen basierend auf vergangener Zeit
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
async def wait_for_token(self, tokens: int = 1, timeout: Optional[float] = None):
"""Wartet bis Tokens verfügbar sind"""
deadline = time.time() + timeout if timeout else None
while True:
if await self.consume(tokens):
return True
if deadline and time.time() >= deadline:
raise TimeoutError(f"Timeout waiting for {tokens} tokens")
# Wartezeit basierend auf benötigten Tokens
wait_time = (tokens - self.tokens) / self.refill_rate
await asyncio.sleep(min(wait_time, 0.1))
class AsyncAPIClient:
"""
Async API Client mit integriertem Rate Limiting.
Perfekt für Production-Workloads mit vielen parallelen Anfragen.
"""
def __init__(
self,
api_key: str,
base_url: str = 'https://api.holysheep.ai/v1',
requests_per_second: float = 10.0,
burst_capacity: int = 20
):
self.api_key = api_key
self.base_url = base_url
self._rate_limiter: Optional[AsyncTokenBucket] = None
self._rps = requests_per_second
self._burst = burst_capacity
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self._rate_limiter = await AsyncTokenBucket.create(
capacity=self._burst,
refill_rate=self._rps
)
if HAS_AIOHTTP:
self._session = aiohttp.ClientSession(
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_completions(
self,
model: str = 'gpt-4.1',
messages: list = None,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""
Sendet eine Chat-Completion Anfrage mit automatischem Rate Limiting.
"""
if messages is None:
messages = []
# Rate Limit abwarten
await self._rate_limiter.wait_for_token(1, timeout=30.0)
payload = {
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens
}
if HAS_AIOHTTP and self._session:
async with self._session.post(
f'{self.base_url}/chat/completions',
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 429:
raise Exception("Rate limit exceeded - please implement retry logic")
return await response.json()
else:
# Fallback zu sync mit httpx
import httpx
with httpx.Client(timeout=60) as client:
response = client.post(
f'{self.base_url}/chat/completions',
json=payload,
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
)
if response.status_code == 429:
raise Exception("Rate limit exceeded")
return response.json()
Production Beispiel: Batch-Verarbeitung mit Rate Limiting
async def process_batch(prompts: list, api_key: str):
"""Verarbeitet mehrere Prompts mit kontrolliertem Rate Limit"""
async with AsyncAPIClient(
api_key=api_key,
requests_per_second=5.0, # 5 Anfragen/Sekunde
burst_capacity=10 # Erlaubt Burst von 10
) as client:
tasks = []
for prompt in prompts:
task = client.chat_completions(
model='deepseek-v3.2', # $0.42/MTok - günstigste Option
messages=[{'role': 'user', 'content': prompt}]
)
tasks.append(task)
# Parallel mit Rate Limit Kontrolle
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
Benchmark: 100 Anfragen mit 10 RPS
async def benchmark_rate_limiting():
"""Testet Rate Limiter Performance"""
import time
limiter = await AsyncTokenBucket.create(capacity=100, refill_rate=10.0)
start = time.time()
success_count = 0
# Senden so schnell wie möglich - Limiter kontrolliert
tasks = []
for i in range(100):
task = limiter.consume(1)
tasks.append(task)
results = await asyncio.gather(*tasks)
success_count = sum(1 for r in results if r)
elapsed = time.time() - start
print(f"100 Anfragen in {elapsed:.2f}s")
print(f"Erfolgreich: {success_count}")
print(f"Tatsächliche Rate: {success_count/elapsed:.1f} req/s")
asyncio.run(benchmark_rate_limiting())
Häufige Fehler und Lösungen
Fehler 1: Race Condition bei concurrent Token-Zugriff
# FEHLERHAFT: Non-thread-safe Implementation
class UnsafeRateLimiter:
def __init__(self, max_per_second: int = 10):
self.max_per_second = max_per_second
self.current_count = 0
self.window_start = time.time()
def acquire(self):
now = time.time()
# PROBLEM: Kein Lock - Race Condition möglich
if now - self.window_start > 1.0:
self.current_count = 0
self.window_start = now
if self.current_count < self.max_per_second:
self.current_count += 1
return True
return False
LÖSUNG: Thread-safe mit Lock
import threading
class SafeRateLimiter:
def __init__(self, max_per_second: int = 10):
self.max_per_second = max_per_second
self.current_count = 0
self.window_start = time.time()
self._lock = threading.Lock() # Korrekt: Lock hinzufügen
def acquire(self):
with self._lock: # Korrekt: Lock verwenden
now = time.time()
if now - self.window_start > 1.0:
self.current_count = 0
self.window_start = now
if self.current_count < self.max_per_second:
self.current_count += 1
return True
return False
Alternativ: Atomic Operations mit threading.local
class AtomicRateLimiter:
def __init__(self, max_per_second: int = 10):
self.max_per_second = max_per_second
self._local = threading.local()
def _get_state(self):
if not hasattr(self._local, 'count'):
self._local.count = 0
self._local.start = time.time()
return self._local
def acquire(self):
state = self._get_state()
if time.time() - state.start > 1.0:
state.count = 0
state.start = time.time()
if state.count < self.max_per_second:
state.count += 1
return True
return False
Fehler 2: Unbehandeltes 429 Response ohne Exponential Backoff
# FEHLERHAFT: Kein Retry bei 429
def call_api_no_retry(endpoint: str, api_key: str):
req = urllib.request.Request(
endpoint,
headers={'Authorization': f'Bearer {api_key}'}
)
response = urllib.request.urlopen(req) # Crash bei 429!
return response.read()
LÖSUNG: Exponential Backoff mit Jitter
import random
def call_api_with_retry(
endpoint: str,
api_key: str,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0
) -> dict:
"""
Ruft API auf mit Exponential Backoff bei Rate Limits.
Args:
endpoint: API Endpoint URL
api_key: Ihr HolySheep AI API Key
max_retries: Maximale Wiederholungsversuche
base_delay: Basis-Wartezeit in Sekunden
max_delay: Maximale Wartezeit zwischen Versuchen
"""
import json
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
for attempt in range(max_retries + 1):
try:
req = urllib.request.Request(endpoint, headers=headers)
with urllib.request.urlopen(req, timeout=60) as response:
return json.loads(response.read().decode('utf-8'))
except urllib.error.HTTPError as e:
if e.code == 429:
# Rate Limit erreicht - Retry mit Backoff
retry_after = e.headers.get('Retry-After', base_delay)
try:
retry_after = float(retry_after)
except (ValueError, TypeError):
retry_after = base_delay
# Exponential Backoff: 1s, 2s, 4s, 8s, 16s...
delay = min(base_delay * (2 ** attempt), max_delay)
# Random Jitter hinzufügen (±25%)
jitter = delay * 0.25 * (random.random() * 2 - 1)
delay = delay + jitter
print(f"Rate limit hit. Attempt {attempt + 1}/{max_retries + 1}. "
f"Waiting {delay:.1f}s (retry-after: {retry_after:.1f}s)")
time.sleep(max(delay, retry_after))
elif e.code in (500, 502, 503, 504):
# Server Error - Retry erlaubt
delay = base_delay * (2 ** attempt)
print(f"Server error {e.code}. Retrying in {delay:.1f}s...")
time.sleep(delay)
elif e.code == 401:
raise Exception(f"Authentication failed. Check your API key: {api_key}")
else:
# Andere Fehler - nicht retry
raise Exception(f"API Error {e.code}: {e.read().decode()}")
except urllib.error.URLError as e:
if attempt < max_retries:
delay = base_delay * (2 ** attempt)
print(f"Connection error: {e.reason}. Retrying in {delay:.1f}s...")
time.sleep(delay)
else:
raise
raise Exception(f"Max retries ({max_retries}) exceeded")
Beispiel mit HolySheep AI
try:
result = call_api_with_retry(
'https://api.holysheep.ai/v1/chat/completions',
api_key='YOUR_HOLYSHEEP_API_KEY'
)
print(f"Success: {result}")
except Exception as e:
print(f"Final error: {e}")