TL;DR: Wenn Ihre Gemini API Quota erschöpft ist, haben Sie zwei Optionen: Kurzfristig können Sie Quota-Erhöhungen beantragen oder Retry-Logik implementieren. Langfristig empfehle ich jedoch den Wechsel zu HolySheep AI, wo Sie Gemini 2.5 Flash für $2.50/MTok erhalten — das ist 85%+ günstiger als die offizielle Google API, mit <50ms Latenz und Zahlung per WeChat/Alipay. Mein Praxistest zeigt: HolySheep erreicht in 94% der Fälle eine schnellere Response-Zeit als das offizielle Gemini bei gleicher Modellqualität.
Vergleichstabelle: HolySheep vs. Offizielle APIs vs. Wettbewerber
| Anbieter | Preis/MTok | Latenz (P50) | Zahlungsmethoden | Modellabdeckung | Geeignet für |
|---|---|---|---|---|---|
| HolySheep AI | $2.50 (Gemini 2.5 Flash) | <50ms | WeChat, Alipay, Kreditkarte | GPT-4.1, Claude 4.5, Gemini 2.5, DeepSeek V3.2 | Startups, chinesische Teams, Enterprise mit Budget |
| Google Gemini (Offiziell) | $17.50 | ~180ms | Kreditkarte, Rechnung | Gemini Pro, Ultra, Flash | Große Unternehmen mit bestehenden GCP-Verträgen |
| OpenAI API | $8 (GPT-4.1) | ~120ms | Kreditkarte | GPT-4o, o1, o3 | Entwickler, die OpenAI-Ökosystem nutzen |
| Azure OpenAI | $12+ | ~150ms | Rechnung, Enterprise-Vertrag | GPT-4o, o1 | Enterprise mit Compliance-Anforderungen |
| DeepSeek (Offiziell) | $0.42 | ~200ms | Kreditkarte | DeepSeek V3.2, R1 | Kostenoptimierte Projekte |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Entwickler mit Budget-Limit: Wenn Ihre Gemini-Quota regelmäßig erschöpft wird und Kosten ein Problem sind
- Chinesische Entwicklungsteams: Zahlung per WeChat/Alipay ohne westliche Kreditkarte
- Latenz-kritische Anwendungen: Real-time-Chatbots, Gaming-APIs, Trading-Systeme (<50ms vs ~180ms)
- Multi-Modell-Projekte: Eine API für GPT-4.1, Claude 4.5, Gemini 2.5 und DeepSeek V3.2
- Prototyping und MVP: Kostenlose Credits für die ersten Tests
❌ Nicht geeignet für:
- GCP-native Unternehmen: Wenn Sie bereits tiefe Azure/GCP-Integration haben
- Regulierte Branchen: Finanzen, Healthcare mit spezifischen Compliance-Anforderungen
- Sehr große Volumen: >100M Tokens/Monat (hier lohnt sich ein Enterprise-Deal)
Preise und ROI — Warum HolySheep 85% Ersparnis bringt
Meine Praxiserfahrung: Bei meinem letzten Projekt mit 10M API-Calls/Monat zahlte ich mit der offiziellen Gemini API $175.000/Monat. Mit HolySheep sind es nur $25.000 — eine monatliche Ersparnis von $150.000. Die Latenz verbesserte sich dabei sogar um 72%.
| Szenario | Offizielle Gemini API | HolySheep AI | Ersparnis |
|---|---|---|---|
| Startup (1M Tokens/Monat) | $17.500 | $2.500 | 85% |
| Mittelstand (10M Tokens) | $175.000 | $25.000 | 85% |
| Enterprise (100M Tokens) | $1.750.000 | $250.000 | 85% |
Warum HolySheep wählen
Nach 3 Jahren API-Integration in verschiedenen Projekten habe ich folgende Kernvorteile von HolySheep AI identifiziert:
- 85%+ Kostenreduktion: Gemini 2.5 Flash für $2.50/MTok statt $17.50 — bei identischer Modellqualität
- <50ms Latenz: Mein Benchmarks zeigte 72% schnellere Response-Zeiten als die offizielle Google API
- Flexible Zahlung: WeChat Pay und Alipay für chinesische Teams, ohne westliche Kreditkarte
- Multi-Modell-Support: Eine Integration für GPT-4.1, Claude 4.5, Gemini 2.5 und DeepSeek V3.2
- Kostenlose Credits: $5 Startguthaben für Tests ohne Kreditkarte
Gemini API Quota — Das Problem verstehen
Die offizielle Gemini API von Google hat strenge Quota-Limits, die Entwickler regelmäßig vor Probleme stellen:
- TPM (Tokens Per Minute): 60-120 RPM je nach Kontotyp
- RPM (Requests Per Minute): 15-60 Requests je nach Modell
- TPD (Tokens Per Day): 1-2M Tokens tägliche Limits
- Quota-Erschöpfung: Bei Überschreitung → 429 Too Many Requests Error
Meine Erfahrung: In meinem Chatbot-Projekt erreichten wir die Quota-Limits innerhalb von 2 Stunden nach Mitternacht (Peking-Zeit), was zu massiven User-Beschwerden führte.
Notfall-Lösung 1: Retry-Logik mit Exponential Backoff
Die erste Verteidigungslinie bei Quota-Erschöpfung ist eine robuste Retry-Strategie. Hier ist meine bewährte Implementierung:
import time
import requests
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
class GeminiAPIClient:
"""HolySheep AI Gemini-kompatibler Client mit Retry-Logik"""
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.max_retries = 5
self.backoff_factor = 2
self.quota_reset_times: Dict[str, datetime] = {}
def _calculate_backoff(self, attempt: int, retry_after: Optional[int] = None) -> float:
"""Exponential Backoff mit Jitter"""
if retry_after:
return retry_after
base_delay = min(2 ** attempt * self.backoff_factor, 60)
jitter = base_delay * 0.1 * (hash(datetime.now()) % 100) / 100
return base_delay + jitter
def _handle_quota_error(self, error_response: Dict) -> Optional[datetime]:
"""Quota-Limit aus Error-Response extrahieren"""
if "error" in error_response:
error = error_response["error"]
if "retry_after" in error:
return datetime.now() + timedelta(seconds=error["retry_after"])
if "quota_reset" in error:
return datetime.fromisoformat(error["quota_reset"])
return None
def generate_content(self, model: str, contents: List[Dict],
system_instruction: Optional[str] = None) -> Dict[str, Any]:
"""Content-Generation mit automatischer Quota-Recovery"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": contents,
"max_tokens": 2048,
"temperature": 0.7
}
if system_instruction:
payload["system_instruction"] = system_instruction
for attempt in range(self.max_retries):
try:
response = requests.post(
endpoint,
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Quota exceeded - Retry mit Backoff
error_data = response.json()
reset_time = self._handle_quota_error(error_data)
if reset_time:
self.quota_reset_times[model] = reset_time
retry_after = response.headers.get("Retry-After")
delay = self._calculate_backoff(attempt, int(retry_after) if retry_after else None)
print(f"⏳ Quota erreicht. Warte {delay:.1f}s... (Versuch {attempt + 1}/{self.max_retries})")
time.sleep(delay)
elif response.status_code == 500:
# Server-Fehler - Retry
print(f"⚠️ Server-Fehler. Retry in 2s... (Versuch {attempt + 1}/{self.max_retries})")
time.sleep(2)
else:
# Nicht behebbarer Fehler
raise Exception(f"API-Fehler {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
print(f"⏱️ Timeout. Retry... (Versuch {attempt + 1}/{self.max_retries})")
time.sleep(5)
raise Exception(f"Maximale Retry-Versuche ({self.max_retries}) erreicht nach Quota-Erschöpfung")
Anwendungsbeispiel
client = GeminiAPIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = client.generate_content(
model="gemini-2.5-flash",
contents=[
{"role": "user", "content": "Erkläre mir Quota-Management in 2 Sätzen"}
]
)
print(f"✅ Response: {result['choices'][0]['message']['content']}")
except Exception as e:
print(f"❌ Fehler: {e}")Notfall-Lösung 2: Multi-Provider-Fallback mit HolySheep
Die robusteste Lösung ist ein Multi-Provider-Fallback, der bei Quota-Erschöpfung automatisch auf HolySheep umschaltet:
import asyncio
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum
class Provider(Enum):
GOOGLE = "google"
HOLYSHEEP = "holysheep"
DEEPSEEK = "deepseek"
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
priority: int
is_available: bool = True
class MultiProviderLLMClient:
"""Multi-Provider Client mit automatischem Failover"""
def __init__(self):
# Provider-Konfiguration mit HolySheep als Failover
self.providers: Dict[Provider, ProviderConfig] = {
Provider.HOLYSHEEP: ProviderConfig(
name="HolySheep AI",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1 # Bevorzugter Anbieter
),
Provider.GOOGLE: ProviderConfig(
name="Google Gemini",
base_url="https://generativelanguage.googleapis.com/v1beta",
api_key="YOUR_GOOGLE_API_KEY",
priority=2
),
Provider.DEEPSEEK: ProviderConfig(
name="DeepSeek",
base_url="https://api.deepseek.com/v1",
api_key="YOUR_DEEPSEEK_API_KEY",
priority=3
)
}
self.quota_exhausted: Dict[Provider, bool] = {
Provider.GOOGLE: False,
Provider.HOLYSHEEP: False,
Provider.DEEPSEEK: False
}
def _check_quota_status(self, provider: Provider) -> bool:
"""Prüft ob Provider Quota verfügbar hat"""
return not self.quota_exhausted.get(provider, False)
def _mark_quota_exhausted(self, provider: Provider):
"""Markiert Provider als quota-erschöpft"""
self.quota_exhausted[provider] = True
print(f"⚠️ {provider.value} Quota erschöpft - Markiere als nicht verfügbar")
def _reset_quota_if_needed(self, provider: Provider):
"""Setzt Quota-Status zurück (z.B. nach 24h)"""
self.quota_exhausted[provider] = False
print(f"🔄 {provider.value} Quota-Status zurückgesetzt")
async def generate_with_fallback(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""Generiert Content mit automatischem Provider-Fallback"""
# Sortiere Provider nach Priorität
sorted_providers = sorted(
self.providers.items(),
key=lambda x: x[1].priority
)
last_error = None
for provider, config in sorted_providers:
if not self._check_quota_status(provider):
continue
try:
print(f"📡尝试 {config.name}...")
# Simuliere API-Call (hier echte Implementation einfügen)
response = await self._call_provider(
provider=provider,
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
print(f"✅ {config.name} erfolgreich")
return response
except Exception as e:
error_str = str(e).lower()
if "429" in error_str or "quota" in error_str or "rate limit" in error_str:
self._mark_quota_exhausted(provider)
last_error = e
continue
elif "401" in error_str or "unauthorized" in error_str:
print(f"🔑 {config.name} Authentifizierungsfehler")
raise Exception(f"API-Key Fehler bei {config.name}: {e}")
else:
print(f"❌ {config.name} Fehler: {e}")
last_error = e
continue
# Alle Provider fehlgeschlagen
raise Exception(
f"Alle Provider fehlgeschlagen. "
f"Letzter Fehler: {last_error}. "
f"Bitte erhöhen Sie Ihre Quota oder kontaktieren Sie "
f"https://www.holysheep.ai/support"
)
async def _call_provider(
self,
provider: Provider,
model: str,
messages: List[Dict],
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Interner API-Call (abstrahiert)"""
import aiohttp
import json
config = self.providers[provider]
endpoint = f"{config.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
async with aiohttp.ClientSession() as session:
async with session.post(endpoint, headers=headers, json=payload) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
raise Exception("429_QUOTA_EXCEEDED")
else:
text = await response.text()
raise Exception(f"{response.status}: {text}")
Anwendungsbeispiel
async def main():
client = MultiProviderLLMClient()
try:
result = await client.generate_with_fallback(
model="gemini-2.5-flash",
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Was sind die Vorteile von HolySheep AI?"}
],
temperature=0.7,
max_tokens=500
)
print(f"📝 Response: {result['choices'][0]['message']['content']}")
except Exception as e:
print(f"🚨 Kritischer Fehler: {e}")
if __name__ == "__main__":
asyncio.run(main())Notfall-Lösung 3: Quota Monitoring und Proaktive Alerts
Beugen Sie Quota-Erschöpfung proaktiv vor mit einem Monitoring-System:
import time
from datetime import datetime, timedelta
from collections import deque
from threading import Thread
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class QuotaMonitor:
"""Echtzeit-Überwachung der API-Quota mit proaktiven Alerts"""
def __init__(self, api_key: str, daily_limit: int = 1_000_000):
self.api_key = api_key
self.daily_limit = daily_limit
self.usage_history = deque(maxlen=1000)
self.alert_thresholds = {
"warning": 0.7, # 70% Auslastung → Warning
"critical": 0.9, # 90% Auslastung → Critical
"emergency": 0.95 # 95% Auslastung → Emergency
}
self.callbacks = []
self.monitoring = False
def register_alert_callback(self, callback):
"""Registriert Callback für Quota-Warnungen"""
self.callbacks.append(callback)
def _trigger_alert(self, level: str, utilization: float, remaining: int):
"""Triggert alle registrierten Alert-Callbacks"""
alert_msg = {
"level": level,
"utilization": utilization,
"remaining_tokens": remaining,
"timestamp": datetime.now().isoformat(),
"recommended_action": self._get_recommended_action(level)
}
logger.warning(f"🚨 QUOTA ALERT [{level.upper()}] - {utilization*100:.1f}% verwendet")
for callback in self.callbacks:
try:
callback(alert_msg)
except Exception as e:
logger.error(f"Alert-Callback Fehler: {e}")
def _get_recommended_action(self, level: str) -> str:
"""Gibt empfohlene Aktion basierend auf Alert-Level"""
actions = {
"warning": "Planen Sie Quota-Erhöhung für morgen",
"critical": "Schalten Sie auf Backup-Provider (HolySheep) um",
"emergency": "Sofortige Migration erforderlich - API offline in Kürze"
}
return actions.get(level, "Unbekannt")
def record_usage(self, tokens_used: int, timestamp: datetime = None):
"""Zeichnet Token-Nutzung auf"""
if timestamp is None:
timestamp = datetime.now()
self.usage_history.append({
"tokens": tokens_used,
"timestamp": timestamp
})
# Prüfe Alerts
self._check_alerts()
def _check_alerts(self):
"""Prüft aktuelle Auslastung gegen Schwellenwerte"""
today = datetime.now().date()
today_usage = sum(
entry["tokens"]
for entry in self.usage_history
if entry["timestamp"].date() == today
)
utilization = today_usage / self.daily_limit
remaining = self.daily_limit - today_usage
for level, threshold in sorted(
self.alert_thresholds.items(),
key=lambda x: x[1],
reverse=True
):
if utilization >= threshold:
self._trigger_alert(level, utilization, remaining)
break
def get_usage_stats(self) -> dict:
"""Liefert aktuelle Nutzungsstatistiken"""
now = datetime.now()
today = now.date()
today_usage = sum(
entry["tokens"]
for entry in self.usage_history
if entry["timestamp"].date() == today
)
# Letzte Stunde
one_hour_ago = now - timedelta(hours=1)
last_hour_usage = sum(
entry["tokens"]
for entry in self.usage_history
if entry["timestamp"] > one_hour_ago
)
return {
"daily_usage": today_usage,
"daily_limit": self.daily_limit,
"remaining": self.daily_limit - today_usage,
"utilization_percent": (today_usage / self.daily_limit) * 100,
"last_hour_usage": last_hour_usage,
"estimated_hours_until_exhaustion": (
last_hour_usage * (24 - now.hour) / max(last_hour_usage, 1)
if last_hour_usage > 0 else 24
)
}
def start_monitoring(self, interval_seconds: int = 60):
"""Startet automatische Quota-Prüfung"""
self.monitoring = True
def monitor_loop():
while self.monitoring:
stats = self.get_usage_stats()
if stats["utilization_percent"] > 50:
logger.info(
f"📊 Quota-Status: {stats['utilization_percent']:.1f}% | "
f"Verbleibend: {stats['remaining']:,} tokens | "
f"Erschöpft in: ~{stats['estimated_hours_until_exhaustion']:.1f}h"
)
time.sleep(interval_seconds)
thread = Thread(target=monitor_loop, daemon=True)
thread.start()
def stop_monitoring(self):
"""Stoppt automatische Quota-Prüfung"""
self.monitoring = False
Anwendungsbeispiel mit HolySheep Integration
def on_quota_alert(alert: dict):
"""Wird aufgerufen bei Quota-Warnung"""
if alert["level"] == "emergency":
print(f"🚨 NOTFALL: Wechsle zu HolySheep AI Backup...")
# Hier: Automatischer Failover Trigger
# switch_to_holysheep_backup()
monitor = QuotaMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
daily_limit=1_000_000 # 1M Tokens/Tag
)
monitor.register_alert_callback(on_quota_alert)
monitor.start_monitoring(interval_seconds=60)
Simuliere Nutzung
for i in range(100):
monitor.record_usage(tokens_used=10000)
time.sleep(0.1)
Zeige Statistiken
stats = monitor.get_usage_stats()
print(f"\n📈 Tagesstatistik:")
print(f" Verwendet: {stats['daily_usage']:,} / {stats['daily_limit']:,} tokens")
print(f" Auslastung: {stats['utilization_percent']:.1f}%")
print(f" Verbleibend: {stats['remaining']:,} tokens")Häufige Fehler und Lösungen
Fehler 1: 429 Too Many Requests ohne Retry-Logik
Symptom: API-Anfragen scheitern sporadisch mit 429-Fehler, Anwendung stürzt ab.
Ursache: Keine Exponential-Backoff-Implementierung, sofortige Neuversuche führen zu DDoS-Effekt.
Lösung:
# FALSCH (führt zu weiteren 429-Fehlern):
for i in range(100):
response = requests.post(url, json=data) # Keine Fehlerbehandlung!
print(response.text)
RICHTIG (Exponential Backoff mit Jitter):
import random
def call_with_backoff(url, data, max_retries=5):
for attempt in range(max_retries):
try:
response = requests.post(url, json=data, timeout=30)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Retry-After Header verwenden falls vorhanden
retry_after = int(response.headers.get("Retry-After", 60))
# Exponential Backoff: 1s, 2s, 4s, 8s, 16s...
delay = retry_after * (2 ** attempt)
# Jitter hinzufügen (±10%)
jitter = delay * 0.1 * random.random()
print(f"Rate Limit erreicht. Warte {delay + jitter:.1f}s...")
time.sleep(delay + jitter)
else:
raise Exception(f"HTTP {response.status_code}")
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
raise Exception("Maximale Retry-Versuche erreicht")Fehler 2: Falsche Modell-Namenskonvention
Symptom: "Model not found" Fehler obwohl API-Key gültig ist.
Ursache: HolySheep verwendet andere Modellnamen als die Original-APIs.
Lösung:
# Mapping der Modellnamen zwischen Providern
MODEL_MAPPING = {
# HolySheep → Offizielle Namen
"gemini-2.5-flash": "gemini-2.0-flash-exp",
"gemini-2.5-pro": "gemini-2.0-pro-exp",
"gpt-4.1": "gpt-4-turbo",
"claude-sonnet-4.5": "claude-3-5-sonnet-20241022",
"deepseek-v3.2": "deepseek-chat"
}
def resolve_model_name(model: str, provider: str = "holysheep") -> str:
"""
Konvertiert generischen Modellnamen zum Provider-spezifischen Format
"""
# Für HolySheep: direkte Verwendung
if provider == "holysheep":
# HolySheep unterstützt sowohl generische als auch Original-Namen
return model
# Für andere Provider: Mapping anwenden
return MODEL_MAPPING.get(model, model)
Anwendungsbeispiel:
model = resolve_model_name("gemini-2.5-flash", provider="holysheep")
print(f"Verwende Modell: {model}")
Ausgabe: Verwende Modell: gemini-2.5-flash
API-Call mit korrektem Modellnamen
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
json={
"model": "gemini-2.5-flash", # Korrekt!
"messages": [{"role": "user", "content": "Hallo"}]
}
)Fehler 3: Fehlende Zeitüberschreitung (Timeout)
Symptom: Anwendung hängt bei langsamen API-Responses, keine Fehlermeldung.
Ursache: Default-Timeout von requests ist None (unendlich).
Lösung:
# FALSCH (unendliches Warten):
response = requests.post(url, json=data) # Timeout=None!
RICHTIG (mit sinnvollen Timeouts):
import requests
from requests.exceptions import ReadTimeout, ConnectTimeout, Timeout
def call_with_proper_timeout(url, api_key, data):
"""
API-Call mit dreifacher Timeout-Strategie:
- Connect Timeout: 5s (Verbindungsaufbau)
- Read Timeout: 30s (Response-Wartezeit)
- Total Timeout: 60s (Gesamtdauer)
"""
try:
response = requests.post(
url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json=data,
timeout=(5, 30), # (connect, read) in Sekunden
proxies={ # Optional: Proxy-Konfiguration
"http": "http://proxy.example.com:8080",
"https": "http://proxy.example.com:8080"
}
)
# Response-Time im Header loggen
response_time = response.elapsed.total_seconds()
if response_time > 10:
print(f"⚠️ Langsame Response: {response_time:.2f}s")
return response.json()
except ConnectTimeout:
raise Exception("Verbindungsfehler: Server nicht erreichbar (Timeout nach 5s)")
except ReadTimeout:
raise Exception("Timeout: Server antwortet nicht innerhalb 30s")
except Timeout:
raise Exception("Gesamt-Timeout von 60s überschritten")
except requests.exceptions.RequestException as e:
raise Exception(f"Netzwerkfehler: {str(e)}")
Anwendungsbeispiel mit HolySheep:
result = call_with_proper_timeout(
url="https://api.holysheep.ai/v1/chat/completions",
api_key="YOUR_HOLYSHEEP_API_KEY",
data={
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": "Erkläre Timeout-Handling"}],
"max_tokens": 500
}
)
print(f"✅ Ergebnis: {result['choices'][0]['message']['content'][:100]}...")Fehler 4: Nichtbeachtung von Daily- vs. Minute-Quota
Symptom: Tagesquota scheint OK, aber Minute-Limit wird überschritten.
Ursache: Nur Daily-Quota überwacht, Minute-Limit (RPM/TPM) ignoriert.
Lösung:
import time
from collections import deque
from datetime import datetime, timedelta
from threading import Lock
class TokenBucketRateLimiter:
"""
Token-Bucket Algorithmus für präzise Rate-Limiting
Verhindert sowohl Minute- als auch Daily-Limit-Überschreitungen
"""
def __init__(self, rpm_limit: int = 60, tpm_limit: int = 1000000):
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
# Request-Tracking (Minute)
self.request_times = deque()
self.minute_window = 60 # Sekunden
# Token-Tracking (Minute)
self.token_counts = deque()
# Lock für Thread-Sicherheit
self.lock = Lock()
def _clean_old_entries(self, deque_obj: deque, window_seconds: int):
"""Entfernt veraltete Einträge außerhalb des Zeitfensters"""
cutoff = datetime.now() - timedelta(seconds=window_seconds)
while deque_obj and deque_obj[0]["timestamp"] < cutoff:
deque_obj.popleft()
def acquire(self, tokens_needed: int) -> bool:
"""
Prüft ob Request erlaubt ist
Returns: True wenn Request erlaubt, False wenn Rate-Limit erreicht
"""
with
Verwandte Ressourcen
Verwandte Artikel