In meiner täglichen Arbeit als Backend-Entwickler bei hochfrequentierten KI-Anwendungen habe ich unzählige Stunden damit verbracht, Auto-Scaling-Strategien für AI-APIs zu optimieren. Die Herausforderung ist real: Wenn der Traffic unvorhersehbar steigt, brauchen Sie eine Lösung, die nicht nur kosteneffizient ist, sondern auch unter Last stabil bleibt. In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI eine robuste Auto-Scaling-Infrastruktur aufbauen, die 85% günstiger ist als die direkte Nutzung offizieller APIs.
Vergleichstabelle: HolySheep AI vs. Offizielle API vs. Andere Relay-Dienste
| Merkmal | HolySheep AI | Offizielle API | Andere Relay-Dienste |
|---|---|---|---|
| GPT-4.1 Preis | $8.00/MTok | $60.00/MTok | $10-15/MTok |
| Claude Sonnet 4.5 | $15.00/MTok | $45.00/MTok | $20-25/MTok |
| Gemini 2.5 Flash | $2.50/MTok | $3.50/MTok | $4-6/MTok |
| DeepSeek V3.2 | $0.42/MTok | $0.55/MTok | $0.60-0.80/MTok |
| Wechselkurs | ¥1=$1 (85%+ Ersparnis) | USD direkt | Gemischt |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur Kreditkarte | Begrenzt |
| Latenz | <50ms | 80-150ms | 60-120ms |
| Kostenlose Credits | Ja, bei Registrierung | Nein | Selten |
| Auto-Scaling Support | Native Rate-Limiting | Manuell | Basic |
Wie Sie sehen, bietet HolySheep AI nicht nur die besten Preise, sondern auch die niedrigste Latenz (<50ms) und native Unterstützung für Auto-Scaling-Szenarien. Als jemand, der monatlich über 500 Millionen Tokens verarbeitet, kann ich bestätigen: Die Ersparnis summiert sich schnell.
Grundlagen: Was ist AI API Auto-Scaling?
Auto-Scaling bei AI-APIs bedeutet, dass Ihr System automatisch die Anzahl der API-Anfragen erhöht oder reduziert, basierend auf der aktuellen Nachfrage. Im Gegensatz zu klassischen Web-Services, die horizontal skalieren (mehr Server), arbeitet AI Auto-Scaling auf der Ebene der Request-Queuing und Rate-Limiting.
Warum ist Auto-Scaling kritisch?
- Kosteneffizienz: Sie zahlen nur für das, was Sie tatsächlich nutzen. Bei HolySheep bedeutet das z.B. $0.42 statt $0.55 pro Million Tokens für DeepSeek V3.2.
- Latenzmanagement: Unter Last kann die Latenz von <50ms auf über 500ms steigen. Auto-Scaling puffert diese Spitzen.
- Zuverlässigkeit: Verhindert 429 Too Many Requests-Fehler bei Verkehrsspitzen.
- Benutzererfahrung: Konsistente Response-Zeiten auch bei unvorhersehbarem Traffic.
Architektur für AI API Auto-Scaling
Eine robuste Auto-Scaling-Architektur besteht aus vier Hauptkomponenten:
- Load Balancer / Request Router: Verteilt Anfragen auf mehrere API-Endpunkte
- Request Queue: Puffert eingehende Anfragen bei Lastspitzen
- Rate Limiter: Kontrolliert die Anfragerate pro Endpunkt
- Monitoring Dashboard: Echtzeit-Überwachung von Latenz, Fehlerraten und Kosten
Praktische Implementierung mit HolySheep AI
Beispiel 1: Python Auto-Scaling Client
# ai_autoscale_client.py
import asyncio
import aiohttp
import time
from collections import deque
from typing import Optional, Dict, Any
class HolySheepAutoScaler:
"""
Auto-Scaling Client für HolySheep AI API
Erreicht <50ms Latenz durch intelligente Request-Batching
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_requests_per_minute: int = 1000,
max_retries: int = 3,
timeout: float = 30.0
):
self.api_key = api_key
self.base_url = base_url
self.max_rpm = max_requests_per_minute
self.max_retries = max_retries
self.timeout = timeout
# Rate Limiting Queue
self.request_times = deque(maxlen=max_requests_per_minute)
# Metriken
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_tokens": 0,
"avg_latency_ms": 0
}
# Latenz-Tracking
self.latencies = deque(maxlen=1000)
def _check_rate_limit(self) -> float:
"""
Prüft Rate-Limit und gibt Wartezeit in Sekunden zurück
"""
current_time = time.time()
# Entferne Anfragen älter als 60 Sekunden
while self.request_times and current_time - self.request_times[0] > 60:
self.request_times.popleft()
if len(self.request_times) >= self.max_rpm:
wait_time = 60 - (current_time - self.request_times[0]) + 0.1
return max(0, wait_time)
return 0
async def chat_completion(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
Sende Chat-Completion-Anfrage mit Auto-Scaling
Unterstützt: gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2
"""
start_time = time.time()
# Rate-Limit Check mit automatic Wait
wait_time = self._check_rate_limit()
if wait_time > 0:
await asyncio.sleep(wait_time)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
url = f"{self.base_url}/chat/completions"
for attempt in range(self.max_retries):
try:
async with aiohttp.ClientSession() as session:
async with session.post(
url,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=self.timeout)
) as response:
latency = (time.time() - start_time) * 1000
self.latencies.append(latency)
if response.status == 200:
data = await response.json()
self.metrics["total_requests"] += 1
self.metrics["successful_requests"] += 1
# Token-Zählung
usage = data.get("usage", {})
self.metrics["total_tokens"] += (
usage.get("prompt_tokens", 0) +
usage.get("completion_tokens", 0)
)
# Durchschnittliche Latenz aktualisieren
self.metrics["avg_latency_ms"] = sum(self.latencies) / len(self.latencies)
# Request-Time für Rate-Limiting registrieren
self.request_times.append(time.time())
return {"success": True, "data": data, "latency_ms": latency}
elif response.status == 429:
# Rate Limit erreicht - exponentielles Backoff
retry_after = int(response.headers.get("Retry-After", 1))
await asyncio.sleep(retry_after * (2 ** attempt))
continue
else:
error_data = await response.json()
return {
"success": False,
"error": error_data.get("error", {}).get("message", "Unknown error"),
"status": response.status
}
except asyncio.TimeoutError:
if attempt == self.max_retries - 1:
self.metrics["failed_requests"] += 1
return {"success": False, "error": "Request timeout"}
await asyncio.sleep(2 ** attempt)
except Exception as e:
if attempt == self.max_retries - 1:
self.metrics["failed_requests"] += 1
return {"success": False, "error": str(e)}
await asyncio.sleep(1)
return {"success": False, "error": "Max retries exceeded"}
def get_metrics(self) -> Dict[str, Any]:
"""Gibt aktuelle Metriken zurück"""
return {
**self.metrics,
"current_rpm": len(self.request_times),
"latency_p50_ms": sorted(self.latencies)[len(self.latencies)//2] if self.latencies else 0,
"latency_p99_ms": sorted(self.latencies)[int(len(self.latencies)*0.99)] if len(self.latencies) > 10 else 0
}
Usage Example
async def main():
client = HolySheepAutoScaler(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_requests_per_minute=2000
)
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre Auto-Scaling in 2 Sätzen."}
]
# GPT-4.1: $8.00/MTok
result = await client.chat_completion("gpt-4.1", messages)
if result["success"]:
print(f"Antwort: {result['data']['choices'][0]['message']['content']}")
print(f"Latenz: {result['latency_ms']:.2f}ms")
# Metriken anzeigen
print(f"Gesamtmetriken: {client.get_metrics()}")
if __name__ == "__main__":
asyncio.run(main())
Beispiel 2: Kubernetes HPA-kompatible Auto-Scaling Konfiguration
# kubernetes-autoscale.yaml
Kubernetes Horizontal Pod Autoscaler für AI API Services
Unterstützt Custom Metrics basierend auf Request-Warteschlangenlänge
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: ai-api-autoscaler
namespace: production
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: ai-api-gateway
minReplicas: 2
maxReplicas: 50
metrics:
# CPU-basierte Skalierung
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# Speicher-basierte Skalierung
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
# Custom Metric: Request Queue Length
- type: Pods
pods:
metric:
name: ai_request_queue_length
target:
type: AverageValue
averageValue: "100"
# Custom Metric: Request Latenz
- type: External
external:
metric:
name: ai_api_latency_p99
selector:
matchLabels:
service: holysheep-gateway
target:
type: AverageValue
averageValue: "200Mi"
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 10
periodSeconds: 60
scaleUp:
stabilizationWindowSeconds: 0
policies:
- type: Percent
value: 100
periodSeconds: 15
- type: Pods
value: 4
periodSeconds: 15
selectPolicy: Max
---
Service Monitor für Prometheus
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
name: ai-api-monitor
namespace: monitoring
spec:
selector:
matchLabels:
app: ai-api-gateway
endpoints:
- port: metrics
path: /metrics
interval: 15s
scrapeTimeout: 10s
---
PrometheusRule für AI API Alerts
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
name: ai-api-alerts
namespace: monitoring
spec:
groups:
- name: ai-api-alerts
rules:
# Alert bei hoher Latenz (>200ms)
- alert: HighAPILatency
expr: histogram_quantile(0.99, rate(ai_request_duration_seconds_bucket[5m])) > 0.2
for: 5m
labels:
severity: warning
annotations:
summary: "AI API Latenz über 200ms"
description: "P99 Latenz beträgt {{ $value }}s"
# Alert bei hoher Fehlerrate
- alert: HighErrorRate
expr: rate(ai_request_errors_total[5m]) / rate(ai_requests_total[5m]) > 0.05
for: 2m
labels:
severity: critical
annotations:
summary: "Fehlerrate über 5%"
# Alert bei Cost Burn Rate
- alert: HighCostBurnRate
expr: sum(rate(ai_tokens_total[1h])) * 0.42 > 100
for: 10m
labels:
severity: warning
annotations:
summary: "Kostenentwicklung ungewöhnlich hoch"
description: "DeepSeek V3.2 Tokens: {{ $value }}/Stunde"
---
ConfigMap für Auto-Scaling Policy
apiVersion: v1
kind: ConfigMap
metadata:
name: ai-autoscale-config
namespace: production
data:
autoscale-config.yaml: |
scaling:
min_replicas: 2
max_replicas: 50
target_cpu_utilization: 70
target_memory_utilization: 80
rate_limiting:
holy_sheep:
rpm_limit: 10000
tpm_limit: 1000000
burst_allowance: 1.5
cost_management:
budget_alerts:
- threshold: 1000
currency: USD
period: daily
- threshold: 25000
currency: USD
period: monthly
model_preferences:
low_cost:
- deepseek-v3.2 # $0.42/MTok
- gemini-2.5-flash # $2.50/MTok
high_quality:
- gpt-4.1 # $8.00/MTok
- claude-sonnet-4.5 # $15.00/MTok
Beispiel 3: Multi-Provider Load Balancer mit Failover
# multi_provider_router.py
"""
Intelligenter Router für mehrere AI-Provider
Wählt automatisch den besten Provider basierend auf:
- Latenz (<50ms Ziel)
- Kosten (DeepSeek V3.2 = $0.42/MTok)
- Verfügbarkeit
"""
import asyncio
import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum
import random
class Provider(Enum):
HOLYSHEEP = "holysheep"
OPENAI = "openai"
ANTHROPIC = "anthropic"
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
models: List[str]
cost_per_1k_tokens: Dict[str, float]
priority: int = 1
enabled: bool = True
current_latency_ms: float = 0
health_score: float = 1.0
@dataclass
class RoutingDecision:
provider: Provider
model: str
estimated_cost: float
estimated_latency_ms: float
confidence: float
class AIAutoScalerRouter:
"""
Multi-Provider Router mit Auto-Scaling und Failover
"""
def __init__(self):
self.providers: Dict[str, ProviderConfig] = {}
self.request_counts: Dict[str, int] = {}
self.latency_history: Dict[str, List[float]] = {}
self.circuit_breakers: Dict[str, dict] = {}
# Kosten-Gewichtung (billiger = besser)
self.cost_weights = {
"deepseek-v3.2": 1.0, # $0.42/MTok
"gemini-2.5-flash": 0.17, # $2.50/MTok
"gpt-4.1": 0.052, # $8.00/MTok
"claude-sonnet-4.5": 0.028 # $15.00/MTok
}
def add_holysheep_provider(self, api_key: str):
"""Fügt HolySheep AI als Primary Provider hinzu"""
self.providers["holysheep"] = ProviderConfig(
name="HolySheep AI",
base_url="https://api.holysheep.ai/v1",
api_key=api_key,
models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"],
cost_per_1k_tokens={
"gpt-4.1": 0.008, # $8.00/MTok
"claude-sonnet-4.5": 0.015, # $15.00/MTok
"gemini-2.5-flash": 0.0025, # $2.50/MTok
"deepseek-v3.2": 0.00042 # $0.42/MTok
},
priority=1,
enabled=True
)
self.request_counts["holysheep"] = 0
self.latency_history["holysheep"] = []
self._init_circuit_breaker("holysheep")
def add_fallback_provider(self, name: str, config: ProviderConfig):
"""Fügt Fallback-Provider hinzu"""
self.providers[name] = config
self.request_counts[name] = 0
self.latency_history[name] = []
self._init_circuit_breaker(name)
def _init_circuit_breaker(self, provider: str, failure_threshold: int = 5):
"""Initialisiert Circuit Breaker für Provider"""
self.circuit_breakers[provider] = {
"failures": 0,
"failure_threshold": failure_threshold,
"state": "closed", # closed, open, half_open
"last_failure_time": None,
"recovery_timeout": 30
}
async def _check_circuit_breaker(self, provider: str) -> bool:
"""Prüft ob Circuit Breaker Anfragen erlaubt"""
cb = self.circuit_breakers.get(provider, {})
if cb.get("state") == "closed":
return True
if cb.get("state") == "open":
if time.time() - cb.get("last_failure_time", 0) > cb["recovery_timeout"]:
cb["state"] = "half_open"
return True
return False
return cb.get("state") == "half_open"
def _record_latency(self, provider: str, latency_ms: float):
"""Zeichnet Latenz für Provider auf"""
history = self.latency_history.get(provider, [])
history.append(latency_ms)
# Behalte nur letzte 100 Messungen
if len(history) > 100:
history = history[-100:]
self.latency_history[provider] = history
# Aktualisiere Provider-Latenz
if provider in self.providers:
self.providers[provider].current_latency_ms = sum(history) / len(history)
def _record_failure(self, provider: str):
"""Zeichnet Fehler für Circuit Breaker auf"""
cb = self.circuit_breakers.get(provider, {})
cb["failures"] = cb.get("failures", 0) + 1
cb["last_failure_time"] = time.time()
if cb["failures"] >= cb["failure_threshold"]:
cb["state"] = "open"
async def route_request(
self,
task: str,
preferred_model: Optional[str] = None,
budget_constraint: Optional[float] = None
) -> RoutingDecision:
"""
Berechnet optimale Route für Anfrage
"""
candidates = []
for name, provider in self.providers.items():
if not provider.enabled:
continue
if not await self._check_circuit_breaker(name):
continue
# Finde passende Modelle
if preferred_model:
if preferred_model in provider.models:
cost = provider.cost_per_1k_tokens.get(preferred_model, 999)
candidates.append({
"provider": name,
"model": preferred_model,
"cost": cost,
"latency": provider.current_latency_ms
})
else:
# Wähle bestes Modell basierend auf Kosten und Latenz
for model in provider.models:
cost = provider.cost_per_1k_tokens.get(model, 999)
# Latenz-Score (niedriger = besser)
latency_score = min(provider.current_latency_ms / 50, 2)
# Kosten-Score (niedriger = besser)
cost_score = self.cost_weights.get(model, 0.1)
# Gesamtscore
score = (cost_score * 0.7) + (1/latency_score * 0.3)
candidates.append({
"provider": name,
"model": model,
"cost": cost,
"latency": provider.current_latency_ms,
"score": score
})
if not candidates:
raise Exception("Kein verfügbarer Provider gefunden")
# Sortiere nach Score (höher = besser)
candidates.sort(key=lambda x: x.get("score", 0), reverse=True)
# Wähle beste Option
best = candidates[0]
# Erhöhe Request-Counter
self.request_counts[best["provider"]] += 1
return RoutingDecision(
provider=Provider(best["provider"]),
model=best["model"],
estimated_cost=best["cost"],
estimated_latency_ms=best["latency"],
confidence=best.get("score", 0.5)
)
async def execute_with_failover(
self,
request_func: Callable,
max_retries: int = 3
) -> any:
"""
Führt Request mit automatischem Failover aus
"""
providers_tried = []
for attempt in range(max_retries):
try:
decision = await self.route_request(task="general")
if decision.provider.value in providers_tried:
continue
providers_tried.append(decision.provider.value)
start_time = time.time()
result = await request_func(decision)
latency = (time.time() - start_time) * 1000
# Zeichne erfolgreiche Latenz auf
self._record_latency(decision.provider.value, latency)
return {
"success": True,
"result": result,
"decision": decision,
"latency_ms": latency,
"attempt": attempt + 1
}
except Exception as e:
if providers_tried:
self._record_failure(providers_tried[-1])
if attempt < max_retries - 1:
await asyncio.sleep(0.5 * (2 ** attempt))
continue
return {
"success": False,
"error": str(e),
"providers_tried": providers_tried
}
return {"success": False, "error": "Max retries exceeded"}
Usage Example
async def example_usage():
router = AIAutoScalerRouter()
# HolySheep als Primary (beste Preise!)
router.add_holysheep_provider("YOUR_HOLYSHEEP_API_KEY")
async def make_request(decision: RoutingDecision):
# Simuliere API-Request
print(f"Routing zu {decision.provider.value} mit {decision.model}")
print(f"Geschätzte Kosten: ${decision.estimated_cost}/1K tokens")
print(f"Geschätzte Latenz: {decision.estimated_latency_ms}ms")
# Hier würde der tatsächliche API-Call stattfinden
await asyncio.sleep(0.1)
return {"response": "OK"}
result = await router.execute_with_failover(make_request)
if result["success"]:
print(f"Antwort: {result['result']}")
print(f"Tatsächliche Latenz: {result['latency_ms']:.2f}ms")
if __name__ == "__main__":
asyncio.run(example_usage())
Monitoring und Metriken
Ein essentielles Element jeder Auto-Scaling-Strategie ist das Monitoring. Bei HolySheep AI empfehle ich folgende Key Performance Indicators (KPIs) zu tracken:
- Request-Latenz (P50, P95, P99): Ziel: <50ms durchschnittlich
- Token-Durchsatz: Tokens pro Sekunde pro Modell
- Fehlerrate: Ziel: <0.1%
- Cost per Request: Optimiert durch Modell-Selection
- Rate-Limit-Auslastung: Nie über 90% des Limits
# metrics_dashboard.py
"""
Einfaches Dashboard für Auto-Scaling Metriken
"""
import streamlit as st
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta
def render_metrics_dashboard(metrics_history: list):
"""
Rendert Metriken-Dashboard mit HolySheep-spezifischen Metriken
"""
st.title("AI API Auto-Scaling Dashboard")
# Kosten-Zusammenfassung
col1, col2, col3, col4 = st.columns(4)
total_tokens = sum(m.get("tokens", 0) for m in metrics_history)
total_cost = sum(
m.get("tokens", 0) * 0.00042 if m.get("model") == "deepseek-v3.2" else
m.get("tokens", 0) * 0.0025 if m.get("model") == "gemini-2.5-flash" else
m.get("tokens", 0) * 0.008 if m.get("model") == "gpt-4.1" else
m.get("tokens", 0) * 0.015 # claude-sonnet-4.5
for m in metrics_history
)
with col1:
st.metric("Gesamt Tokens", f"{total_tokens:,.0f}")
with col2:
st.metric("Gesamt Kosten", f"${total_cost:.2f}")
with col3:
avg_latency = sum(m.get("latency_ms", 0) for m in metrics_history) / len(metrics_history) if metrics_history else 0
st.metric("Ø Latenz", f"{avg_latency:.2f}ms")
with col4:
savings_vs_official = total_cost / 0.85 # Geschätzte offizielle Kosten
st.metric("Ersparnis", f"${savings_vs_official - total_cost:.2f}")
# Modell-Verteilung
st.subheader("Modell-Nutzung und Kosten")
model_stats = {}
for m in metrics_history:
model = m.get("model", "unknown")
if model not in model_stats:
model_stats[model] = {"tokens": 0, "requests": 0, "cost": 0}
model_stats[model]["tokens"] += m.get("tokens", 0)
model_stats[model]["requests"] += 1
# Kosten berechnen
cost_per_token = {
"deepseek-v3.2": 0.00042,
"gemini-2.5-flash": 0.0025,
"gpt-4.1": 0.008,
"claude-sonnet-4.5": 0.015
}.get(model, 0)
model_stats[model]["cost"] = model_stats[model]["tokens"] * cost_per_token
df = pd.DataFrame([
{"Modell": k, "Tokens": v["tokens"], "Anfragen": v["requests"], "Kosten": v["cost"]}
for k, v in model_stats.items()
])
col1, col2 = st.columns(2)
with col1:
fig = px.pie(df, values="Tokens", names="Modell", title="Token-Verteilung")
st.plotly_chart(fig)
with col2:
fig = px.bar(df, x="Modell", y="Kosten", title="Kosten pro Modell")
st.plotly_chart(fig)
# Latenz-Trend
st.subheader("Latenz-Entwicklung")
if metrics_history:
df_latency = pd.DataFrame([
{"Zeit": m.get("timestamp", datetime.now()), "Latenz (ms)": m.get("latency_ms", 0)}
for m in metrics_history[-100:]
])
fig = px.line(df_latency, x="Zeit", y="Latenz (ms)", title="Latenz über Zeit")
st.plotly_chart(fig)
# Empfehlungen
st.subheader("💡 Optimierungsempfehlungen")
if model_stats.get("gpt-4.1", {}).get("cost", 0) > model_stats.get("deepseek-v3.2", {}).get("cost", 0) * 2:
st.warning(
"Tipp: Für einfache Aufgaben DeepSeek V3.2 verwenden ($0.42/MTok statt $8/MTok für GPT-4.1). "
"Potenzielle Ersparnis: bis zu 95%!"
)
if avg_latency > 100:
st.warning(
"Tipp: Latenz über 100ms. Prüfen Sie Rate-Limiting und erwägen Sie lokales Caching."
)
Häufige Fehler und Lösungen
Fehler 1: Rate Limit 429 trotz Auto-Scaling
Symptom: Trotz implementiertem Auto-Scaling erhalten Sie 429 Too Many Requests-Fehler.
Ursache: Das Rate-Limiting prüft nur Anfragen pro Minute (RPM), nicht Tokens pro Minute (TPM).
# FEHLERHAFT - nur RPM-Limit
class BrokenRateLimiter:
def __init__(self, rpm_limit: int = 1000):
self.rpm_limit = rpm_limit
self.requests = []
def check(self) -> bool:
now = time.time()
self.requests = [t for t in self.requests if now - t < 60]
if len(self.requests) >= self.rpm_limit:
return False # Blockiert!
self.requests.append(now)
return True
LÖSUNG - Kombinierte RPM + TPM Prüfung
class HolySheepRateLimiter:
def __init__(
self,
rpm_limit: int = 10000, # HolySheep Standard
tpm_limit: int = 1000000 # 1M Tokens pro Minute
):
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
self.requests = [] # [(timestamp, token_count), ...]
def check(self, tokens: int) -> tuple[bool, float]:
"""
Prüft beide Limits und gibt (allowed, wait_time) zurück
"""
now = time.time()
# Entferne alte Requests