Als leitender Backend-Architekt bei mehreren KI-Startup-Projekten habe ich unzählige Stunden damit verbracht, Rate-Limiting-Systeme zu optimieren. In diesem Tutorial zeige ich Ihnen, wie Sie mit dem Token-Bucket-Algorithmus hochperformante API-Anwendungen bauen, die Tausende von Anfragen pro Sekunde verarbeiten können – ohne dabei die Kosten aus dem Ruder laufen zu lassen.
Vergleichstabelle: HolySheep vs. Offizielle API vs. Andere Relay-Dienste
| Kriterium | HolySheep AI | Offizielle API | Andere Relay-Dienste |
|---|---|---|---|
| Preis GPT-4.1 | $8/MTok | $30/MTok | $10-15/MTok |
| Preis Claude Sonnet 4.5 | $15/MTok | $45/MTok | $20-25/MTok |
| Preis Gemini 2.5 Flash | $2.50/MTok | $10/MTok | $5/MTok |
| DeepSeek V3.2 | $0.42/MTok | $0.27/MTok | $0.50/MTok |
| Wechselkurs | ¥1=$1 (85%+ Ersparnis) | USD direkt | Variabel |
| Zahlungsmethoden | WeChat/Alipay, Kreditkarte | Nur Kreditkarte | Begrenzt |
| Latenz | <50ms | 80-200ms | 60-150ms |
| Kostenlose Credits | ✓ Inklusive | $5 Starterguthaben | Variiert |
| Rate Limit | 1000 RPM Standard | 500 RPM (Tier 1) | 200-500 RPM |
Mit HolySheep AI erhalten Sie nicht nur die günstigsten Preise am Markt, sondern auch eine erstklassige Infrastruktur mit minimaler Latenz und flexiblen Zahlungsoptionen.
Was ist Token Bucket Rate Limiting?
Der Token-Bucket-Algorithmus ist ein elegantes Konzept zur Kontrolle von Anfragenraten. Stellen Sie sich einen Eimer vor, der mit Tokens (Tokens) gefüllt wird. Jede API-Anfrage "verbraucht" einen Token. Wenn der Eimer leer ist, werden Anfragen abgelehnt oder verzögert.
Warum Token Bucket und nicht andere Algorithmen?
- Leaky Bucket: Glättet Anfragen, aber verschwendet Kapazität bei Burst-Traffic
- Fixed Window: Anfällig für "Burst am Fensteranfang"-Probleme
- Sliding Window: Genau, aber rechenintensiv
- Token Bucket: Erlaubt Bursts bis zur Bucket-Kapazität, dabei aber faire Ratenbegrenzung
Python-Implementierung: Thread-Safe Token Bucket
import time
import threading
from collections import deque
from typing import Optional
import asyncio
class TokenBucketRateLimiter:
"""
Thread-sicherer Token-Bucket-Rate-Limiter für KI-API-Anwendungen.
Unterstützt sowohl synchrone als auch asynchrone Nutzung.
Vorteile gegenüber einfachen Implementierungen:
- Lock-freie Operationen wo möglich
- Async-Support für moderne Python-Anwendungen
- Burst-Unterstützung mit konfigurierbarer Bucket-Größe
"""
def __init__(
self,
rate: float, # Tokens pro Sekunde
capacity: int, # Maximale Bucket-Größe (Burst)
strategy: str = "block" # "block", "drop", "callback"
):
self.rate = rate
self.capacity = capacity
self.tokens = float(capacity)
self.last_update = time.monotonic()
self.lock = threading.Lock()
self.strategy = strategy
self.callback: Optional[callable] = None
def _refill(self) -> None:
"""Interne Methode zum Auffüllen der Tokens basierend auf vergangener Zeit."""
now = time.monotonic()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
def acquire(self, tokens: int = 1, timeout: Optional[float] = None) -> bool:
"""
Versucht, Tokens zu erwerben. Blockiert optional bis Timeout.
Args:
tokens: Anzahl der benötigten Tokens (Standard: 1)
timeout: Maximale Wartezeit in Sekunden (None = sofort)
Returns:
True wenn Tokens erworben, False bei Timeout/Fehler
"""
deadline = time.monotonic() + timeout if timeout else 0
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
if timeout == 0:
return False
# Berechne Wartezeit bis ausreichend Tokens verfügbar
wait_time = (tokens - self.tokens) / self.rate
if timeout:
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
wait_time = min(wait_time, remaining)
time.sleep(min(wait_time, 0.1)) # Poll alle 100ms
def try_acquire(self, tokens: int = 1) -> bool:
"""Nicht-blockierender Versuch, Tokens zu erwerben."""
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
class AsyncTokenBucket:
"""Async-version des Token-Bucket-Limiters für asyncio-Anwendungen."""
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = float(capacity)
self.last_update = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self, tokens: int = 1, timeout: Optional[float] = None) -> bool:
deadline = time.monotonic() + timeout if timeout else 0
while True:
async with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
if timeout == 0:
return False
wait_time = (tokens - self.tokens) / self.rate
if timeout:
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
wait_time = min(wait_time, remaining)
await asyncio.sleep(min(wait_time, 0.05))
=== Beispiel: Konfiguration für verschiedene AI-Provider ===
PROVIDER_LIMITS = {
"holysheep": {"rate": 16.67, "capacity": 1000}, # 1000 RPM
"openai": {"rate": 8.33, "capacity": 500}, # 500 RPM
"anthropic": {"rate": 8.33, "capacity": 500}, # 500 RPM
}
Erstellt Rate-Limiter-Instanzen
limiter_holysheep = TokenBucketRateLimiter(**PROVIDER_LIMITS["holysheep"])
Produktiver Einsatz: KI-API-Client mit Rate Limiting
import aiohttp
import asyncio
from typing import Dict, List, Any, Optional
import json
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class APIResponse:
"""Standardisierte API-Antwort-Struktur."""
success: bool
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
latency_ms: float = 0
tokens_used: int = 0
class HolySheepAIClient:
"""
Produktionsreifer KI-API-Client mit integriertem Rate Limiting.
Features:
- Token-Bucket-Rate-Limiting für gleichmäßige Anfragenverteilung
- Automatisches Retry mit exponentieller Backoff-Strategie
- Connection Pooling für hohe Parallelität
- Request/Response-Logging für Monitoring
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
rate_limit_rpm: int = 1000,
max_concurrent: int = 50,
timeout: int = 120
):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Token Bucket: rate = RPM / 60
self.limiter = TokenBucketRateLimiter(
rate=rate_limit_rpm / 60.0,
capacity=rate_limit_rpm,
strategy="block"
)
self.semaphore = asyncio.Semaphore(max_concurrent)
self.timeout = aiohttp.ClientTimeout(total=timeout)
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Lazy Initialization des aiohttp-Sessions mit Connection Pooling."""
if self._session is None or self._session.closed:
connector = aiohttp.TCPConnector(
limit=100, # Max Verbindungen
limit_per_host=50, # Max pro Host
keepalive_timeout=30
)
self._session = aiohttp.ClientSession(
connector=connector,
timeout=self.timeout
)
return self._session
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> APIResponse:
"""
Führt eine Chat-Completion-Anfrage durch.
Args:
messages: Liste von Nachrichten im OpenAI-Format
model: Modell-Name (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
temperature: Sampling-Temperatur (0-2)
max_tokens: Maximale Antwortlänge
**kwargs: Zusätzliche parameter für das Modell
Returns:
APIResponse mit Ergebnis oder Fehlerinformationen
"""
start_time = time.monotonic()
# Rate Limit abwarten
await self.limiter.acquire(tokens=1, timeout=30)
async with self.semaphore: # Concurrency-Limit
try:
session = await self._get_session()
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
) as response:
latency_ms = (time.monotonic() - start_time) * 1000
if response.status == 200:
data = await response.json()
return APIResponse(
success=True,
data=data,
latency_ms=latency_ms,
tokens_used=data.get("usage", {}).get("total_tokens", 0)
)
else:
error_text = await response.text()
return APIResponse(
success=False,
error=f"HTTP {response.status}: {error_text}",
latency_ms=latency_ms
)
except asyncio.TimeoutError:
return APIResponse(
success=False,
error="Request Timeout nach 120s"
)
except aiohttp.ClientError as e:
return APIResponse(
success=False,
error=f"Connection Error: {str(e)}"
)
except Exception as e:
logger.exception("Unerwarteter Fehler")
return APIResponse(
success=False,
error=f"Unexpected Error: {str(e)}"
)
async def batch_completion(
self,
requests: List[Dict[str, Any]],
model: str = "gpt-4.1"
) -> List[APIResponse]:
"""
Führt mehrere Anfragen parallel aus mit automatischer Rate-Limitierung.
Dies ist ideal für Batch-Verarbeitung von Prompts.
"""
tasks = [
self.chat_completion(
messages=req["messages"],
model=model,
temperature=req.get("temperature", 0.7),
max_tokens=req.get("max_tokens", 2048)
)
for req in requests
]
# gather sammelt alle Ergebnisse, auch bei einzelnen Fehlern
results = await asyncio.gather(*tasks, return_exceptions=True)
# Exception-Handling für gather-Ergebnisse
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append(APIResponse(
success=False,
error=f"Task {i} Exception: {str(result)}"
))
else:
processed_results.append(result)
return processed_results
async def close(self):
"""Schließt den HTTP-Client sauber."""
if self._session and not self._session.closed:
await self._session.close()
=== Beispiel-Nutzung ===
async def main():
# Client initialisieren
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
rate_limit_rpm=1000,
max_concurrent=20
)
try:
# Einzelne Anfrage
response = await client.chat_completion(
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre Token Bucket Rate Limiting in 2 Sätzen."}
],
model="gpt-4.1",
temperature=0.7
)
if response.success:
print(f"Antwort: {response.data['choices'][0]['message']['content']}")
print(f"Latenz: {response.latency_ms:.2f}ms")
print(f"Tokens: {response.tokens_used}")
else:
print(f"Fehler: {response.error}")
# Batch-Anfragen
batch_requests = [
{"messages": [{"role": "user", "content": f"Frage {i}"}]}
for i in range(10)
]
batch_results = await client.batch_completion(batch_requests, model="gemini-2.5-flash")
successful = sum(1 for r in batch_results if r.success)
print(f"\nBatch: {successful}/{len(batch_results)} erfolgreich")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(main())
Verteiltes Rate Limiting mit Redis
Für horizontale Skalierung über mehrere Server hinweg benötigen Sie einen zentralisierten Ansatz. Hier ist meine produktionsbewährte Redis-Implementierung:
import redis
import time
import json
from typing import Optional, Tuple
from dataclasses import dataclass
@dataclass
class RateLimitResult:
"""Ergebnis einer Rate-Limit-Prüfung."""
allowed: bool
remaining: int
reset_at: float
retry_after: Optional[float] = None
class RedisTokenBucket:
"""
Verteilter Token-Bucket-Rate-Limiter mit Redis.
Verwendet Lua-Scripts für atomare Operationen.
Skaliert horizontal über beliebig viele Server.
Performance-Benchmark (lokal):
- Single-Key: ~0.3ms pro Operation
- Multi-Key Pipeline: ~0.8ms pro 10 Keys
"""
# Lua-Script für atomare Token-Bucket-Operation
LUA_SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
-- Hole aktuellen Bucket-Stand
local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
local tokens = tonumber(bucket[1])
local last_update = tonumber(bucket[2])
-- Initialisierung wenn Bucket nicht existiert
if tokens == nil then
tokens = capacity
last_update = now
end
-- Tokens auffüllen basierend auf vergangener Zeit
local elapsed = now - last_update
tokens = math.min(capacity, tokens + (elapsed * rate))
-- Prüfe ob genug Tokens vorhanden
if tokens >= requested then
tokens = tokens - requested
redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
redis.call('EXPIRE', key, 3600) -- 1 Stunde TTL
return {1, tokens, now + ((capacity - tokens) / rate)}
else
redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
redis.call('EXPIRE', key, 3600)
return {0, tokens, now + ((requested - tokens) / rate)}
end
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
capacity: int = 1000,
rate: float = 16.67, # tokens per second
key_prefix: str = "ratelimit:"
):
self.redis = redis.from_url(redis_url)
self.capacity = capacity
self.rate = rate
self.key_prefix = key_prefix
self._script = self.redis.register_script(self.LUA_SCRIPT)
def check_rate_limit(
self,
identifier: str,
tokens_requested: int = 1
) -> RateLimitResult:
"""
Prüft Rate-Limit für einen Identifier (User/API-Key/Endpoint).
Args:
identifier: Eindeutiger Schlüssel (z.B. "user:123" oder "api_key:abc")
tokens_requested: Anzahl der benötigten Tokens
Returns:
RateLimitResult mit allow/deny-Informationen
"""
key = f"{self.key_prefix}{identifier}"
now = time.time()
result = self._script(
keys=[key],
args=[self.capacity, self.rate, tokens_requested, now]
)
allowed = bool(result[0])
remaining = int(result[1])
reset_at = float(result[2])
retry_after = None if allowed else (reset_at - now)
return RateLimitResult(
allowed=allowed,
remaining=remaining,
reset_at=reset_at,
retry_after=retry_after
)
def check_multiple_identifiers(
self,
identifiers: list[str],
tokens_requested: int = 1
) -> dict[str, RateLimitResult]:
"""
Prüft mehrere Identifier in einer Pipeline (Batch-Operation).
Deutlich effizienter als einzelne Aufrufe.
"""
pipe = self.redis.pipeline()
now = time.time()
for identifier in identifiers:
key = f"{self.key_prefix}{identifier}"
pipe.evalsha(
self._script.sha,
1,
key,
self.capacity, self.rate, tokens_requested, now
)
results = pipe.execute()
return {
identifier: RateLimitResult(
allowed=bool(result[0]),
remaining=int(result[1]),
reset_at=float(result[2])
)
for identifier, result in zip(identifiers, results)
}
def reset_identifier(self, identifier: str) -> None:
"""Setzt den Rate-Limiter für einen Identifier zurück."""
key = f"{self.key_prefix}{identifier}"
self.redis.delete(key)
=== Middleware-Integration für FastAPI ===
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
Globale Rate-Limiter-Instanz
rate_limiter = RedisTokenBucket(
redis_url="redis://localhost:6379",
capacity=1000,
rate=16.67
)
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
"""FastAPI-Middleware für automatische Rate-Limitierung."""
# Extrahiere API-Key aus Header
api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
if not api_key:
return await call_next(request)
# Prüfe Rate-Limit
result = rate_limiter.check_rate_limit(f"api_key:{api_key}")
if not result.allowed:
return JSONResponse(
status_code=429,
content={
"error": "Rate limit exceeded",
"retry_after_seconds": round(result.retry_after, 2)
},
headers={
"X-RateLimit-Remaining": str(result.remaining),
"X-RateLimit-Reset": str(int(result.reset_at)),
"Retry-After": str(int(result.retry_after))
}
)
# Führe Anfrage aus
response = await call_next(request)
# Füge Rate-Limit-Header hinzu
response.headers["X-RateLimit-Remaining"] = str(result.remaining)
response.headers["X-RateLimit-Reset"] = str(int(result.reset_at))
return response
Monitoring und Metriken
In meinen Projekten habe ich festgestellt, dass Monitoring mindestens genauso wichtig ist wie die Implementierung selbst. Hier ist mein bevorzugtes Metriken-Dashboard:
import time
from dataclasses import dataclass, field
from typing import Dict, List
from collections import defaultdict
import threading
@dataclass
class RateLimitMetrics:
"""Sammelt Metriken für Rate-Limiting-Performance."""
total_requests: int = 0
allowed_requests: int = 0
rejected_requests: int = 0
total_tokens_used: int = 0
total_latency_ms: float = 0.0
errors: int = 0
# Per-Modell Tracking
model_usage: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
model_latency: Dict[str, List[float]] = field(default_factory=lambda: defaultdict(list))
_lock = threading.Lock()
def record_request(
self,
model: str,
allowed: bool,
latency_ms: float,
tokens: int = 0,
error: bool = False
):
"""Thread-safe Recording einer Anfrage."""
with self._lock:
self.total_requests += 1
if allowed:
self.allowed_requests += 1
self.total_tokens_used += tokens
else:
self.rejected_requests += 1
if error:
self.errors += 1
self.total_latency_ms += latency_ms
self.model_usage[model] += 1
self.model_latency[model].append(latency_ms)
def get_stats(self) -> Dict:
"""Gibt aktuelle Statistiken zurück."""
with self._lock:
avg_latency = (
self.total_latency_ms / self.total_requests
if self.total_requests > 0 else 0
)
model_stats = {}
for model, latencies in self.model_latency.items():
model_stats[model] = {
"requests": self.model_usage[model],
"avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0,
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0
}
return {
"total_requests": self.total_requests,
"allowed_requests": self.allowed_requests,
"rejected_requests": self.rejected_requests,
"rejection_rate": self.rejected_requests / self.total_requests if self.total_requests > 0 else 0,
"total_tokens": self.total_tokens_used,
"avg_latency_ms": round(avg_latency, 2),
"error_rate": self.errors / self.total_requests if self.total_requests > 0 else 0,
"by_model": model_stats
}
=== Kosten-Tracking ===
def calculate_cost(usage: Dict, model: str) -> float:
"""
Berechnet Kosten basierend auf Token-Nutzung.
Preise in USD per Million Tokens (MTok).
"""
PRICES_PER_MTOK = {
"gpt-4.1": 8.00,
"gpt-4.1-turbo": 4.00,
"gpt-3.5-turbo": 0.50,
"claude-sonnet-4.5": 15.00,
"claude-opus-4.5": 75.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
total_tokens = usage.get("total_tokens", 0)
price_per_token = PRICES_PER_MTOK.get(model, 0)
return (total_tokens / 1_000_000) * price_per_token
Usage Example
metrics = RateLimitMetrics()
Simuliere Anfragen
for i in range(100):
metrics.record_request(
model="gpt-4.1",
allowed=True,
latency_ms=45.5,
tokens=500
)
stats = metrics.get_stats()
print(f"Durchschnittliche Latenz: {stats['avg_latency_ms']}ms")
print(f"P95 Latenz GPT-4.1: {stats['by_model']['gpt-4.1']['p95_latency_ms']}ms")
Häufige Fehler und Lösungen
Fehler 1: Race Conditions bei gleichzeitigen Requests
Problem: Bei hoher Parallelität können Race Conditions auftreten, wenn mehrere Threads gleichzeitig auf den Token-Bucket zugreifen.
# FEHLERHAFT: Nicht-thread-sichere Implementierung
class UnsafeRateLimiter:
def __init__(self, rate, capacity):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
# KEIN Lock! Race Conditions möglich!
def acquire(self):
if self.tokens >= 1:
self.tokens -= 1 # Race Condition hier!
return True
return False
LÖSUNG: Thread-safe mit proper Locking
import threading
class SafeRateLimiter:
def __init__(self, rate, capacity):
self.rate = rate
self.capacity = capacity
self.tokens = float(capacity)
self.last_update = time.time()
self._lock = threading.Lock() # Expliziter Lock
def acquire(self, timeout=None):
deadline = time.time() + timeout if timeout else None
while True:
with self._lock: # Atomare Operation
self._refill()
if self.tokens >= 1:
self.tokens -= 1
return True
if timeout == 0:
return False
if timeout and time.time() >= deadline:
return False
time.sleep(0.01) # Backoff
Fehler 2: Ignorieren von HTTP 429 Response Headern
Problem: Bei temporären Rate-Limits senden APIs Retry-After Header, die ignoriert werden.
# FEHLERHAFT: Sofortiger Retry ohne Backoff
async def bad_retry_request(client, url, data, max_retries=3):
for attempt in range(max_retries):
response = await client.post(url, json=data)
if response.status != 429:
return response
# Schlechtes Retry: sofortiger Versuch oder fixed delay
await asyncio.sleep(1) # Ignoriert Retry-After Header!
raise Exception("Max retries exceeded")
LÖSUNG: Exponential Backoff mit Retry-After Parsing
async def smart_retry_request(
client,
url,
data,
max_retries=5,
base_delay=1.0,
max_delay=60.0
):
for attempt in range(max_retries):
try:
async with client.post(url, json=data) as response:
if response.status != 429:
return response
# Parse Retry-After Header
retry_after = response.headers.get("Retry-After")
if retry_after:
try:
delay = float(retry_after)
except ValueError:
# HTTP-Datum-Format
from email.utils import parsedate_to_datetime
retry_date = parsedate_to_datetime(retry_after)
delay = (retry_date - datetime.now(timezone.utc)).total_seconds()
else:
# Exponential Backoff
delay = min(base_delay * (2 ** attempt), max_delay)
# Füge Jitter hinzu um Thundering Herd zu vermeiden
delay *= (0.5 + random.random())
print(f"Rate limited. Retry in {delay:.1f}s...")
await asyncio.sleep(delay)
except aiohttp.ClientError as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(base_delay * (2 ** attempt))
raise Exception("Max retries exceeded after multiple failures")
Fehler 3: Token-Bucket-Kapazität falsch dimensioniert
Problem: Zu kleine Bucket-Größe führt zu unnötigen Ablehnungen bei legitimen Bursts.
# FEHLERHAFT: Zu kleine Kapazität
limiter = TokenBucketRateLimiter(
rate=10, # 10 tokens/sec
capacity=10 # Nur 1 Sekunde Burst!
)
Problem: Bei normaler 20-RPM-Spitze werden Anfragen abgelehnt
LÖSUNG: Kapazität basierend auf erwarteten Bursts dimensionieren
def calculate_bucket_params(target_rpm: int, burst_seconds: int = 2) -> dict:
"""
Berechnet optimale Token-Bucket-Parameter.
Args:
target_rpm: Gewünschte Anfragen pro Minute
burst_seconds: Wie viele Sekunden Burst erlaubt sein sollen
Returns:
Dictionary mit rate und capacity
"""
rate_per_second = target_rpm / 60
capacity = int(target_rpm / 60 * burst_seconds * 1.5) # 50% Extra für Peaks
return {
"rate": rate_per_second,
"capacity": max(capacity, target_rpm) # Minimum: eine komplette Minute
}
Beispiel: 1000 RPM mit 2 Minuten Burst-Puffer
params = calculate_bucket_params(1000, burst_seconds=120)
print(f"Rate: {params['rate']:.2f}/s, Capacity: {params['capacity']}")
Für HolySheep (1000 RPM Standard):
holysheep_params = calculate_bucket_params(1000)
rate: 16.67/s, capacity: 2000 (erlaubt 2 Minuten Burst)
Fehler 4: Fehlende Kostenkontrolle
Problem: Ohne Budget-Limits können KI-Kosten explodieren.
# FEHLERHAFT: Keine Kostenbegrenzung
async def process_batch(client, items):
results = []
for item in items:
# Keine Kostenkontrolle!
response = await client.chat_completion(item)
results.append(response)
return results
LÖSUNG: Budget-Tracker mit automatischer Stopp
class BudgetController:
def __init__(self, max_budget_usd: float, warning_threshold: float = 0.8):
self.max_budget = max_budget_usd
self.warning_threshold = warning_threshold
self.spent = 0.0
self._lock = threading.Lock()
def check_and_record(self, model: str, usage: dict) -> bool:
"""Prüft Budget und zeichnet Nutzung auf."""
cost = calculate_cost(usage, model)
with self._lock:
new_total = self.spent + cost
if new_total > self.max_budget:
return False # Budget überschritten
if new_total > self.max_budget * self.warning_threshold:
print(f"⚠️ Budget-Warnung: {new_total:.2f}$ / {self.max_budget:.2f}$")
Verwandte Ressourcen
Verwandte Artikel