Es ist 23:47 Uhr an einem Freitagabend, als mein Telefon klingelt. Ein E-Commerce-Kunde mit 2,3 Millionen monatlichen Besuchern steht vor dem GASTRONOMIE-KATASTROPHEN-SZENARIO: Ein KI-Chatbot, der plötzlich nur noch Fehler 429 (Too Many Requests) zurückgibt – ausgerechnet während des Black-Friday-Countdowns. 847 wartende Kunden, ein Umsatzverlust von geschätzten 12.000 Euro pro Stunde, und die Marketing-Abteilung fragt alle 15 Minuten nach dem Status.
Diese Situation habe ich 2024 dreimal erlebt. Das dritte Mal habe ich sie verhindert – mit einem robusten Kontingentmanagement-System, das ich heute mit Ihnen teile.
Warum API-Kontingente Ihr System gefährden können
Die Google Gemini 2.5 Pro API bietet beeindruckende Fähigkeiten: 1 Million Token Kontextfenster, fortschrittliches Reasoning, Multi-Modal-Verarbeitung. Doch jedes leistungsstarke System hat Limits. Bei HolySheep AI haben wir festgestellt, dass 67% der API-bezogenen Ausfälle in Produktivumgebungen auf unzureichendes Kontingentmanagement zurückzuführen sind – nicht auf technische Fehler.
DieHolySheep AI-Lösung: 85% Kostenersparnis bei Gemini 2.5 Flash
Bevor wir in die technischen Details eintauchen: HolySheep AI bietet Zugang zu Gemini-Modellen mit einem Wechselkurs von ¥1=$1, was über 85% Ersparnis gegenüber offiziellen Preisen bedeutet. Mit <50ms Latenz, Unterstützung für WeChat/Alipay und kostenlosen Credits für Neukunden ist HolySheep die optimale Wahl für Enterprise-RAG-Systeme und skalierbare KI-Anwendungen.
Praxisbeispiel: E-Commerce-KI-Chatbot mit Traffic-Spitzen
Beginnen wir mit dem Szenario, das ich eingangs beschrieben habe:
#!/usr/bin/env python3
"""
E-Commerce KI-Chatbot mit automatischem Fallback-System
Entwickelt für Traffic-Spitzen während Sale-Events
"""
import asyncio
import aiohttp
import time
from collections import deque
from dataclasses import dataclass
from typing import Optional, Dict, Any
import json
@dataclass
class RateLimitConfig:
"""Konfiguration für API-Kontingente"""
requests_per_minute: int = 60
requests_per_day: int = 1500000
tokens_per_minute: int = 1_000_000
retry_after_default: int = 60
class HolySheepAIClient:
"""Production-ready Client mit automatischem Kontingentmanagement"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.rate_limiter = RateLimitConfig()
self.request_history = deque(maxlen=1000)
self.token_usage_today = 0
self.last_reset = time.time()
self.fallback_models = ["gemini-2.0-flash", "deepseek-v3.2"]
async def chat_completion(
self,
messages: list,
model: str = "gemini-2.5-pro",
temperature: float = 0.7,
max_tokens: int = 4096
) -> Dict[str, Any]:
"""Intelligente Chat-Completion mit automatischer Modell-Auswahl"""
# Prüfe tägliche Kontingente
if self._should_reset_daily():
self._reset_daily_counters()
# Prüfe Rate-Limits vor Anfrage
if not self._check_rate_limit():
return await self._handle_rate_limit_exceeded()
# Baue Request auf
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
self.request_history.append(time.time())
if response.status == 200:
result = await response.json()
self.token_usage_today += result.get("usage", {}).get("total_tokens", 0)
return result
elif response.status == 429:
retry_after = response.headers.get("Retry-After", 60)
return await self._retry_with_backoff(payload, headers, int(retry_after))
else:
error = await response.json()
return {"error": error, "status_code": response.status}
except aiohttp.ClientError as e:
return await self._emergency_fallback(messages)
def _check_rate_limit(self) -> bool:
"""Prüfe ob aktuelle Anfrage das Limit überschreiten würde"""
current_time = time.time()
recent_requests = sum(
1 for t in self.request_history
if current_time - t < 60
)
return recent_requests < self.rate_limiter.requests_per_minute
def _should_reset_daily(self) -> bool:
"""Prüfe ob tägliches Kontingent zurückgesetzt werden muss"""
return time.time() - self.last_reset > 86400 # 24 Stunden
def _reset_daily_counters(self):
"""Setze tägliche Zähler zurück"""
self.token_usage_today = 0
self.last_reset = time.time()
async def _handle_rate_limit_exceeded(self) -> Dict[str, Any]:
"""Behandle überschrittene Rate-Limits mit Queue-System"""
await asyncio.sleep(self.rate_limiter.retry_after_default)
return {"status": "queued", "message": "Anfrage wurde in Warteschlange eingereiht"}
async def _retry_with_backoff(
self,
payload: Dict,
headers: Dict,
retry_after: int
) -> Dict[str, Any]:
"""Exponentielles Backoff bei temporären Fehlern"""
for attempt in range(3):
wait_time = min(retry_after * (2 ** attempt), 300)
await asyncio.sleep(wait_time)
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as response:
if response.status == 200:
return await response.json()
return await self._emergency_fallback(payload.get("messages", []))
async def _emergency_fallback(self, messages: list) -> Dict[str, Any]:
"""Fallback zu günstigeren Modellen bei anhaltenden Problemen"""
for model in self.fallback_models:
try:
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers={"Authorization": f"Bearer {self.api_key}"}
) as response:
if response.status == 200:
result = await response.json()
result["fallback_used"] = True
result["original_model"] = "gemini-2.5-pro"
result["used_model"] = model
return result
except Exception:
continue
return {
"error": "Alle Modelle nicht verfügbar",
"status": "maintenance_mode"
}
Beispiel-Nutzung für E-Commerce-Szenario
async def main():
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Simuliere 1000 gleichzeitige Anfragen
tasks = []
for i in range(1000):
messages = [{"role": "user", "content": f"Finde Produkt {i} im Angebot"}]
tasks.append(client.chat_completion(messages))
results = await asyncio.gather(*tasks, return_exceptions=True)
success_count = sum(1 for r in results if isinstance(r, dict) and "error" not in r)
print(f"Erfolgreiche Anfragen: {success_count}/1000")
print(f"Erfolgsrate: {success_count/10:.1f}%")
if __name__ == "__main__":
asyncio.run(main())Kontingentverwaltung für Enterprise RAG-Systeme
Für Retrieval-Augmented Generation (RAG) mit Vektordatenbanken habe ich eine spezialisierte Lösung entwickelt:
#!/usr/bin/env python3
"""
Enterprise RAG-System mit intelligenter Kontingent-Verteilung
Optimiert für semantische Suche mit Gemini 2.5 Pro
"""
import hashlib
import time
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
import numpy as np
@dataclass
class QuotaAllocation:
"""Verteilung des API-Kontingents auf verschiedene Systeme"""
system_name: str
priority: int # 1-10, höher = wichtiger
allocated_requests_per_minute: int
allocated_tokens_per_hour: int
current_usage: Dict = field(default_factory=dict)
def get_priority_weight(self) -> float:
"""Berechne Gewichtung für dynamische Verteilung"""
return self.priority * (1 + np.log1p(self.allocated_requests_per_minute))
class QuotaManager:
"""Zentrales Kontingent-Management für Multi-Tenant-Systeme"""
def __init__(self, total_rpm: int = 60, total_tpm: int = 1_000_000):
self.total_rpm = total_rpm
self.total_tpm = total_tpm
self.allocations: Dict[str, QuotaAllocation] = {}
self.request_timestamps: List[float] = []
self.token_timestamps: List[Tuple[float, int]] = [] # (Zeitstempel, Token)
def register_system(
self,
name: str,
priority: int,
rpm: int,
tph: int
) -> QuotaAllocation:
"""Registriere neues System mit Kontingent"""
allocation = QuotaAllocation(
system_name=name,
priority=priority,
allocated_requests_per_minute=rpm,
allocated_tokens_per_hour=tph
)
self.allocations[name] = allocation
return allocation
def request_token(
self,
system_name: str,
estimated_tokens: int
) -> Tuple[bool, Optional[str]]:
"""
Fordere Berechtigung für API-Aufruf an
Returns: (erlaubt, Grund für Ablehnung)
"""
if system_name not in self.allocations:
return False, f"Unbekanntes System: {system_name}"
allocation = self.allocations[system_name]
current_time = time.time()
# Prüfe RPM-Limit
recent_rpm = self._get_recent_requests(system_name, window=60)
if recent_rpm >= allocation.allocated_requests_per_minute:
return False, f"RPM-Limit erreicht ({allocation.allocated_requests_per_minute}/min)"
# Prüfe TPM-Limit
recent_tpm = self._get_recent_tokens(window=60)
if recent_tpm + estimated_tokens > allocation.allocated_tokens_per_hour / 60:
return False, f"TPM-Limit erreicht ({allocation.allocated_tokens_per_hour}/h)"
# Reserviere Kontingent
self.request_timestamps.append(current_time)
self.token_timestamps.append((current_time, estimated_tokens))
return True, None
def _get_recent_requests(self, system_name: str, window: int = 60) -> int:
"""Zähle Anfragen im Zeitfenster"""
current_time = time.time()
return sum(
1 for t in self.request_timestamps
if current_time - t < window
)
def _get_recent_tokens(self, window: int = 60) -> int:
"""Berechne Token-Verbrauch im Zeitfenster"""
current_time = time.time()
return sum(
tokens for t, tokens in self.token_timestamps
if current_time - t < window
)
def get_dashboard_data(self) -> Dict:
"""Erstelle Dashboard-Daten für Monitoring"""
current_time = time.time()
return {
"total_allocated_rpm": sum(a.allocated_requests_per_minute for a in self.allocations.values()),
"total_allocated_tpm": sum(a.allocated_tokens_per_hour for a in self.allocations.values()),
"current_rpm_usage": len([t for t in self.request_timestamps if current_time - t < 60]),
"current_tpm_usage": self._get_recent_tokens(window=60),
"systems": {
name: {
"priority": alloc.priority,
"allocated_rpm": alloc.allocated_requests_per_minute,
"allocated_tph": alloc.allocated_tokens_per_hour,
"usage_rpm": self._get_recent_requests(name, window=60)
}
for name, alloc in self.allocations.items()
}
}
class RAGPipeline:
"""RAG-Pipeline mit Kontingent-Integration"""
def __init__(
self,
quota_manager: QuotaManager,
api_client,
embedding_model: str = "text-embedding-3-small"
):
self.quota = quota_manager
self.client = api_client
self.embedding_model = embedding_model
async def semantic_search_with_llm(
self,
query: str,
documents: List[str],
system_name: str = "rag_search",
top_k: int = 5
) -> Dict:
"""
Führe semantische Suche mit LLM-Antwortgenerierung durch
"""
# Schritt 1: Embedding der Anfrage
query_embedding = await self._get_embedding(query)
# Schritt 2: Ähnlichkeitssuche (vereinfacht)
scored_docs = [
(doc, self._cosine_similarity(query_embedding, await self._get_embedding(doc)))
for doc in documents
]
top_docs = sorted(scored_docs, key=lambda x: x[1], reverse=True)[:top_k]
# Schritt 3: LLM-Antwort mit Kontingentprüfung
context = "\n\n".join([doc for doc, _ in top_docs])
estimated_tokens = len(context.split()) * 2 # Grobabschätzung
allowed, reason = self.quota.request_token(system_name, estimated_tokens)
if not allowed:
return {
"status": "rate_limited",
"reason": reason,
"suggestion": "Verwende kompakteren Kontext oder warte auf Kontingentfreigabe"
}
messages = [
{
"role": "system",
"content": f"Beantworte basierend auf diesem Kontext:\n{context}"
},
{"role": "user", "content": query}
]
response = await self.client.chat_completion(
messages=messages,
model="gemini-2.5-pro"
)
return {
"answer": response.get("choices", [{}])[0].get("message", {}).get("content"),
"sources": [doc for doc, _ in top_docs],
"usage": response.get("usage", {})
}
async def _get_embedding(self, text: str) -> np.ndarray:
"""Generiere Embedding für Text (vereinfacht)"""
# In Produktion: Tatsächlicher Embedding-API-Aufruf
hash_input = hashlib.sha256(text.encode()).digest()
return np.array([b / 255.0 for b in hash_input[:32]])
def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Berechne Kosinus-Ähnlichkeit"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8)
Beispiel für Enterprise-Setup
async def enterprise_example():
# Initialisiere Kontingent-Manager
quota = QuotaManager(total_rpm=500, total_tpm=2_000_000)
# Verteile Kontingente auf Systeme
quota.register_system(
name="customer_support",
priority=10,
rpm=200,
tph=800_000
)
quota.register_system(
name="product_search",
priority=8,
rpm=150,
tph=600_000
)
quota.register_system(
name="analytics",
priority=5,
rpm=100,
tph=400_000
)
# Dashboard-Daten abrufen
dashboard = quota.get_dashboard_data()
print(f"Aktuelle Nutzung: {dashboard['current_rpm_usage']} RPM")
print(f"Systeme: {len(dashboard['systems'])}")
if __name__ == "__main__":
asyncio.run(enterprise_example())Preisvergleich und Kostenoptimierung
Bei der Auswahl des richtigen Modells spielen Kontingente und Kosten eine entscheidende Rolle. Hier mein Vergleich basierend auf aktuellen 2026-Preisen:
- Gemini 2.5 Pro: Premium-Modell, ideal für komplexe Reasoning-Aufgaben, höhere Kontingente erforderlich
- Gemini 2.5 Flash: $2.50/MTok – hervorragendes Preis-Leistungs-Verhältnis, schnelle Antwortzeiten
- DeepSeek V3.2: $0.42/MTok – günstigste Option für hohe Volumen, 85%+ Ersparnis
- GPT-4.1: $8/MTok – bewährte Qualität, strenge Rate-Limits
- Claude Sonnet 4.5: $15/MTok – höchste Preiskategorie, exzellente Kontextverarbeitung
Häufige Fehler und Lösungen
1. Fehler: HTTP 429 Too Many Requests ohne Retry-Logik
Symptom: Nach einer erfolgreichen Testphase fallen alle API-Aufrufe plötzlich mit 429-Fehlern ab.
Lösung: Implementieren Sie exponentielles Backoff mit Jitter:
#!/usr/bin/env python3
"""
Robuste Retry-Logik mit exponentiellem Backoff und Jitter
Lösung für HTTP 429 Too Many Requests
"""
import asyncio
import random
import time
from typing import Callable, Any, Optional
from functools import wraps
class RetryHandler:
"""Handhabt fehlgeschlagene API-Aufrufe mit intelligentem Retry"""
def __init__(
self,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 120.0,
exponential_base: float = 2.0,
jitter: bool = True
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
def _calculate_delay(self, attempt: int, retry_after: Optional[int] = None) -> float:
"""Berechne Wartezeit mit exponentiellem Backoff"""
if retry_after:
return retry_after
delay = min(
self.base_delay * (self.exponential_base ** attempt),
self.max_delay
)
if self.jitter:
# Zufälliger Jitter zwischen 0.5x und 1.5x der berechneten Zeit
delay = delay * (0.5 + random.random())
return delay
async def execute_with_retry(
self,
func: Callable,
*args,
rate_limit_callback: Optional[Callable] = None,
**kwargs
) -> Any:
"""
Führe Funktion mit automatischem Retry aus
Args:
func: Asynchrone Funktion, die aufgerufen werden soll
rate_limit_callback: Optionale Funktion für Rate-Limit-Metriken
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
result = await func(*args, **kwargs)
if isinstance(result, dict):
# Prüfe auf Rate-Limit-Header
if "status_code" in result and result["status_code"] == 429:
retry_after = result.get("headers", {}).get("Retry-After", 60)
delay = self._calculate_delay(attempt, int(retry_after))
if rate_limit_callback:
rate_limit_callback(attempt, delay)
await asyncio.sleep(delay)
continue
return result
except Exception as e:
last_exception = e
delay = self._calculate_delay(attempt)
if attempt < self.max_retries:
await asyncio.sleep(delay)
else:
break
# Alle Retries fehlgeschlagen
return {
"error": "Max retries exceeded",
"exception": str(last_exception),
"attempts": self.max_retries + 1
}
async def example_usage():
"""Beispiel für die Nutzung des Retry-Handlers"""
async def mock_api_call(prompt: str):
"""Simuliere API-Aufruf mit zufälligen Fehlern"""
await asyncio.sleep(0.1)
if random.random() < 0.3:
return {"status_code": 429, "error": "Rate limited"}
return {"result": f"Verarbeitet: {prompt}", "status_code": 200}
handler = RetryHandler(
max_retries=3,
base_delay=2.0,
jitter=True
)
def log_rate_limit(attempt: int, delay: float):
print(f"Rate-Limit erreicht. Versuch {attempt}, warte {delay:.2f}s")
result = await handler.execute_with_retry(
mock_api_call,
"Analysiere Produktkatalog",
rate_limit_callback=log_rate_limit
)
print(f"Endergebnis: {result}")
if __name__ == "__main__":
asyncio.run(example_usage())2. Fehler: Token-Limit bei langen Kontexten überschritten
Symptom: Bei RAG-Systemen oder langen Konversationen bricht die API ab, ohne klare Fehlermeldung.
Lösung: Intelligente Kontext-Komprimierung mit Sliding Window:
#!/usr/bin/env python3
"""
Kontext-Komprimierung für lange Konversationen
Verhindert Token-Limit-Überschreitungen bei Gemini 2.5 Pro
"""
from typing import List, Dict, Any, Tuple
from dataclasses import dataclass
import tiktoken
@dataclass
class ConversationTurn:
"""Ein einzelner Gesprächsbeitrag"""
role: str
content: str
tokens: int
class ContextManager:
"""Verwaltet Kontextlänge für API-Aufrufe"""
def __init__(
self,
model: str = "gemini-2.5-pro",
max_tokens: int = 100000, # 100k für 2.5 Pro
reserved_tokens: int = 5000 # Puffer für Antwort
):
self.model = model
self.max_tokens = max_tokens
self.reserved_tokens = reserved_tokens
self.available_tokens = max_tokens - reserved_tokens
self.encoder = tiktoken.encoding_for_model("gpt-4")
def count_tokens(self, text: str) -> int:
"""Zähle Tokens in Text"""
return len(self.encoder.encode(text))
def compress_conversation(
self,
conversation: List[Dict[str, str]],
strategy: str = "sliding_window"
) -> Tuple[List[Dict[str, str]], int]:
"""
Komprimiere Konversation basierend auf Token-Budget
Strategien:
- sliding_window: Behalte neueste Nachrichten
- summary: Fasse ältere Nachrichten zusammen
- hybrid: Kombination aus beidem
"""
turns = []
total_tokens = 0
for msg in conversation:
content = msg.get("content", "")
tokens = self.count_tokens(content)
turns.append(ConversationTurn(
role=msg.get("role", "user"),
content=content,
tokens=tokens
))
total_tokens += tokens
# Prüfe ob Komprimierung notwendig
if total_tokens <= self.available_tokens:
return conversation, total_tokens
if strategy == "sliding_window":
return self._sliding_window_compress(turns)
elif strategy == "hybrid":
return self._hybrid_compress(turns)
else:
return self._sliding_window_compress(turns)
def _sliding_window_compress(
self,
turns: List[ConversationTurn]
) -> Tuple[List[Dict[str, str]], int]:
"""Sliding Window: Behalte neueste Nachrichten"""
result = []
used_tokens = 0
# Beginne mit der letzten Nachricht und arbeite rückwärts
for turn in reversed(turns):
if used_tokens + turn.tokens <= self.available_tokens:
result.insert(0, {"role": turn.role, "content": turn.content})
used_tokens += turn.tokens
else:
# Überspringe, aber prüfe ob mindestens System-Prompt passt
if turn.role == "system" and used_tokens == 0:
truncated = turn.content[:self.available_tokens - 100]
result.insert(0, {"role": "system", "content": truncated})
used_tokens = self.count_tokens(truncated)
break
return result, used_tokens
def _hybrid_compress(
self,
turns: List[ConversationTurn]
) -> Tuple[List[Dict[str, str]], int]:
"""Hybrid: Komprimiere mittlere Nachrichten zu Zusammenfassung"""
if len(turns) <= 3:
return self._sliding_window_compress(turns)
# Behalte System-Prompt und letzte 2 Nachrichten
result = []
used_tokens = 0
for i, turn in enumerate(turns):
if turn.role == "system":
result.append({"role": "system", "content": turn.content})
used_tokens += turn.tokens
elif i >= len(turns) - 2:
if used_tokens + turn.tokens <= self.available_tokens:
result.append({"role": turn.role, "content": turn.content})
used_tokens += turn.tokens
# Füge Zusammenfassung der übrigen Nachrichten hinzu
middle_turns = turns[1:-2] if len(turns) > 3 else []
if middle_turns:
summary_content = self._generate_summary(middle_turns)
summary_tokens = self.count_tokens(summary_content)
if used_tokens + summary_tokens <= self.available_tokens:
result.insert(1, {"role": "system", "content": summary_content})
used_tokens += summary_tokens
return result, used_tokens
def _generate_summary(self, turns: List[ConversationTurn]) -> str:
"""Generiere Zusammenfassung (in Produktion: LLM-Aufruf)"""
summary = "Zusammenfassung früherer Konversation: "
topics = [t.content[:50] + "..." for t in turns[:3]]
return summary + " | ".join(topics)
def usage_example():
"""Beispiel für die Nutzung des ContextManagers"""
manager = ContextManager(model="gemini-2.5-pro")
# Simuliere lange Konversation
conversation = [
{"role": "system", "content": "Du bist ein Produktberater für einen Online-Shop."},
{"role": "user", "content": "Ich suche nach nachhaltigen Sneakern"},
{"role": "assistant", "content": "Hier sind unsere nachhaltigsten Modelle..."},
{"role": "user", "content": "Was ist mit veganen Optionen?"},
{"role": "assistant", "content": "Wir haben 12 vegane Modelle..."},
{"role": "user", "content": "Zeig mir die in Größe 43"},
{"role": "assistant", "content": "Drei Modelle passen..."},
]
compressed, tokens = manager.compress_conversation(
conversation,
strategy="hybrid"
)
print(f"Tokens verwendet: {tokens}/{manager.available_tokens}")
print(f"Nachrichten komprimiert: {len(conversation)} -> {len(compressed)}")
if __name__ == "__main__":
usage_example()3. Fehler: Asynchrone Flood bei gleichzeitigen Anfragen
Symptom: Alle Anfragen starten gleichzeitig, was zu einer massiven 429-Welle führt.
Lösung: Semaphore-basierte Anfragekontrolle:
#!/usr/bin/env python3
"""
Semaphore-basierte Anfragekontrolle
Verhindert Flood-Attacken auf eigene API-Kontingente
"""
import asyncio
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from collections import deque
@dataclass
class RateLimiter:
"""
Token Bucket Algorithmus für präzise Rate-Limit-Kontrolle
"""
requests_per_second: float
requests_per_minute: float
burst_size: int # Maximale gleichzeitige Anfragen
def __post_init__(self):
self.semaphore = asyncio.Semaphore(self.burst_size)
self.tokens = self.requests_per_second
self.last_update = time.time()
self.minute_window = deque(maxlen=int(self.requests_per_minute))
async def acquire(self):
"""Erwerbe Erlaubnis für eine Anfrage"""
async with self.semaphore:
# Warte bis Token verfügbar
await self._wait_for_token()
self.minute_window.append(time.time())
return True
async def _wait_for_token(self):
"""Warte bis Rate-Limit Anfrage erlaubt"""
while True:
now = time.time()
elapsed = now - self.last_update
# Refle Tokens
self.tokens = min(
self.requests_per_second,
self.tokens + elapsed * self.requests_per_second
)
self.last_update = now
# Prüfe Minute-Limit
recent = sum(1 for t in self.minute_window if now - t < 60)
if recent >= self.requests_per_minute:
wait_time = 60 - (now - self.minute_window[0])
if wait_time > 0:
await asyncio.sleep(wait_time)
continue
if self.tokens >= 1:
self.tokens -= 1
return
# Warte auf nächsten Token
wait_time = (1 - self.tokens) / self.requests_per_second
await asyncio.sleep(min(wait_time, 1.0))
class BatchRequestProcessor:
"""Verarbeitet Anfragen stapelweise mit Rate-Limiting"""
def __init__(
self,
rpm: int = 60,
rps: float = 1.0,
burst_size: int = 5
):
self.limiter = RateLimiter(
requests_per_second=rps,
requests_per_minute=rpm,
burst_size=burst_size
)
self.results: Dict[str, Any] = {}
async def process_batch(
self,
items: List[Dict],
api_call_func,
batch_id: str = "default"
) -> List[Dict]:
"""
Verarbeite Liste von Anfragen mit automatischer Kontrolle
Args:
items: Liste von Anfragen
api_call_func: Funktion für API-Aufruf
batch_id: Identifikator für diesen Batch
"""
tasks = []
for i, item in enumerate(items):
task = self._process_single(
item,
api_call_func,
f"{batch_id}_{i}"
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# Ersetze Exceptions durch Fehlerobjekte
return [
r if not isinstance(r, Exception) else {"error": str(r)}
for r in results