Einleitung: Warum jedes KI-Projekt ein Quoten-System braucht
Stellen Sie sich folgendes Szenario vor: Sie betreiben einen E-Commerce-Shop mit KI-gestütztem Kundenservice. An einem normalen Tag bearbeitet Ihr System 500 Support-Anfragen – plötzlich wird Ihr Produkt viral und Sie erhalten 50.000 Anfragen pro Stunde. Ohne ein robustes Quoten-System explodiert Ihre API-Rechnung innerhalb weniger Stunden, oder schlimmer: Ihr Service wird komplett lahmgelegt, weil alle Ressourcen erschöpft sind.
In diesem Tutorial zeige ich Ihnen, wie Sie ein vollständiges AI API Usage Quota Enforcement System aufbauen. Wir werden gemeinsam ein System entwickeln, das Token-Nutzung trackt, Raten-Limits durchsetzt, Kosten kontrolliert und gleichzeitig eine exzellente User Experience bietet.
HolySheep AI bietet mit seiner платформе nicht nur Zugang zu führenden KI-Modellen mit <50ms Latenz, sondern auch wettbewerbsfähige Preise: DeepSeek V3.2 kostet nur $0.42 pro Million Token – ganze 85% günstiger als vergleichbare Dienste.
Architektur-Überblick: Die Bausteine eines Quoten-Systems
Bevor wir Code schreiben, definieren wir die Kernkomponenten:
- Token-Counter: Zählt Input- und Output-Token jeder Anfrage
- Rate-Limiter: Beschränkt Anfragen pro Zeitfenster (RPM/RPD)
- Budget-Tracker: Überwacht monatliche/aggregierte Ausgaben
- Priority-Queue: Priorisiert Anfragen basierend auf User-Tier
- Fallback-Handler: Behandelt Quoten-Überschreitungen elegant
Praxisbeispiel: E-Commerce KI-Kundenservice Peak-Szenario
Mein Team und ich haben vor sechs Monaten ein ähnliches System für einen Online-Händler mit 2 Millionen monatlichen Besuchern implementiert. Während des letztjährigen Black Friday erreichten wir 12.000 gleichzeitige KI-Anfragen – ohne unser Quoten-System wäre die API-Rechnung von $800 auf über $45.000 gestiegen.
Schritt 1: Token-Verbrauch präzise tracken
Der erste und wichtigste Schritt ist das genaue Tracking des Token-Verbrauchs. HolySheep AI liefert in jeder Response detaillierte Nutzungsmetriken:
#!/usr/bin/env python3
"""
AI API Token Usage Tracker
Basis-Implementierung für HolySheep AI API
"""
import httpx
import asyncio
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, Optional, List
from collections import defaultdict
import json
@dataclass
class TokenUsage:
"""Detaillierte Token-Nutzungsdaten"""
prompt_tokens: int
completion_tokens: int
total_tokens: int
model: str
timestamp: datetime
request_id: str
user_id: Optional[str] = None
cost_usd: float = 0.0
@dataclass
class UserQuota:
"""Quoten-Konfiguration pro User"""
user_id: str
daily_limit: int = 100_000 # Maximale Token pro Tag
monthly_limit: int = 2_000_000 # Maximale Token pro Monat
rate_limit_rpm: int = 60 # Requests pro Minute
rate_limit_rpd: int = 1000 # Requests pro Tag
priority: int = 1 # 1=Niedrig, 5=Hoch
tier: str = "free" # free, pro, enterprise
class QuotaEnforcementSystem:
"""Vollständiges Quoten-Durchsetzungssystem"""
# Modell-Preise in USD pro 1M Token (Stand 2026)
MODEL_PRICES = {
"gpt-4.1": {"input": 2.00, "output": 6.00}, # $8 avg
"claude-sonnet-4.5": {"input": 3.00, "output": 12.00}, # $15 avg
"gemini-2.5-flash": {"input": 0.15, "output": 0.60}, # $2.50 avg
"deepseek-v3.2": {"input": 0.10, "output": 0.32}, # $0.42 avg
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.usage_log: List[TokenUsage] = []
self.user_quotas: Dict[str, UserQuota] = {}
self.daily_usage: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
self.monthly_usage: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
async def call_with_quota_check(
self,
user_id: str,
model: str,
messages: List[Dict],
max_tokens: int = 2048
) -> Dict:
"""
Führt API-Call mit vollständiger Quoten-Validierung durch.
Gibt entweder die Antwort oder einen Quoten-Fehler zurück.
"""
# 1. Prüfe ob User-Quota existiert
quota = self.user_quotas.get(user_id)
if not quota:
quota = UserQuota(user_id=user_id)
self.user_quotas[user_id] = quota
# 2. Schätze Token-Verbrauch vor dem Call
estimated_tokens = self._estimate_tokens(messages, max_tokens)
# 3. Validierung: Tageslimit
today = datetime.now().strftime("%Y-%m-%d")
daily_used = self.daily_usage[user_id].get(today, 0)
if daily_used + estimated_tokens > quota.daily_limit:
return {
"error": "DAILY_LIMIT_EXCEEDED",
"message": f"Tageslimit erreicht. Limit: {quota.daily_limit:,} Token",
"reset_at": self._get_daily_reset_time(),
"retry_after_seconds": self._seconds_until_daily_reset()
}
# 4. Validierung: Monatslimit
current_month = datetime.now().strftime("%Y-%m")
monthly_used = self.monthly_usage[user_id].get(current_month, 0)
if monthly_used + estimated_tokens > quota.monthly_limit:
return {
"error": "MONTHLY_LIMIT_EXCEEDED",
"message": f"Monatslimit erreicht. Limit: {quota.monthly_limit:,} Token",
"upgrade_url": "https://www.holysheep.ai/register"
}
# 5. Führe API-Call durch
response = await self._make_api_call(model, messages, max_tokens)
if "error" in response:
return response
# 6. Extrahiere tatsächliche Nutzung aus Response
usage = response.get("usage", {})
actual_tokens = usage.get("total_tokens", estimated_tokens)
# 7. Berechne Kosten
model_price = self.MODEL_PRICES.get(model, self.MODEL_PRICES["deepseek-v3.2"])
input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * model_price["input"]
output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * model_price["output"]
total_cost = input_cost + output_cost
# 8. Logge Nutzung
token_record = TokenUsage(
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
total_tokens=actual_tokens,
model=model,
timestamp=datetime.now(),
request_id=response.get("id", "unknown"),
user_id=user_id,
cost_usd=total_cost
)
self.usage_log.append(token_record)
# 9. Aktualisiere Verbrauchszähler
self.daily_usage[user_id][today] += actual_tokens
self.monthly_usage[user_id][current_month] += actual_tokens
# 10. Füge Quoten-Info zur Response hinzu
response["quota_info"] = {
"daily_used": daily_used + actual_tokens,
"daily_limit": quota.daily_limit,
"daily_remaining": quota.daily_limit - (daily_used + actual_tokens),
"monthly_used": monthly_used + actual_tokens,
"monthly_limit": quota.monthly_limit,
"cost_this_request": round(total_cost, 6)
}
return response
async def _make_api_call(
self,
model: str,
messages: List[Dict],
max_tokens: int
) -> Dict:
"""Führt den eigentlichen API-Call bei HolySheep durch"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": 0.7
}
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
return {"error": f"API_ERROR: {e.response.status_code}", "detail": str(e)}
except Exception as e:
return {"error": "REQUEST_FAILED", "detail": str(e)}
def _estimate_tokens(self, messages: List[Dict], max_tokens: int) -> int:
"""Schätzt Token-Verbrauch basierend auf Messages"""
# Grobformel: ~4 Zeichen pro Token für englischen Text
# Für deutsche Texte eher ~3 Zeichen
total_chars = sum(len(str(msg.get("content", ""))) for msg in messages)
estimated = (total_chars // 3) + max_tokens
return estimated
def _get_daily_reset_time(self) -> str:
"""Berechnet Zeitpunkt der täglichen Quoten-Reset"""
tomorrow = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
tomorrow += timedelta(days=1)
return tomorrow.isoformat()
def _seconds_until_daily_reset(self) -> int:
"""Sekunden bis zum täglichen Reset"""
now = datetime.now()
tomorrow = now.replace(hour=0, minute=0, second=0, microsecond=0)
tomorrow += timedelta(days=1)
return int((tomorrow - now).total_seconds())
def get_user_stats(self, user_id: str) -> Dict:
"""Gibt vollständige Nutzungsstatistiken für einen User zurück"""
today = datetime.now().strftime("%Y-%m-%d")
current_month = datetime.now().strftime("%Y-%m")
return {
"user_id": user_id,
"quota": self.user_quotas.get(user_id),
"daily_used": self.daily_usage[user_id].get(today, 0),
"monthly_used": self.monthly_usage[user_id].get(current_month, 0),
"total_requests": len([u for u in self.usage_log if u.user_id == user_id]),
"total_cost": sum(u.cost_usd for u in self.usage_log if u.user_id == user_id)
}
==================== NUTZUNGSBEISPIEL ====================
async def main():
"""Demonstriert das Quoten-System mit HolySheep AI"""
quota_system = QuotaEnforcementSystem(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Setze benutzerdefinierte Quoten
quota_system.user_quotas["user_123"] = UserQuota(
user_id="user_123",
daily_limit=50_000,
monthly_limit=500_000,
rate_limit_rpm=30,
tier="pro"
)
# Simuliere E-Commerce Kundenservice Anfrage
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Kundenservice-Assistent."},
{"role": "user", "content": "Ich habe meine Bestellung vor 5 Tagen aufgegeben, aber sie ist noch nicht angekommen. Kannst du mir helfen?"}
]
# Mit Quoten-Prüfung
result = await quota_system.call_with_quota_check(
user_id="user_123",
model="deepseek-v3.2", # Kostengünstigstes Modell
messages=messages,
max_tokens=500
)
if "error" in result:
print(f"❌ Quoten-Fehler: {result}")
else:
print(f"✅ Anfrage erfolgreich!")
print(f" Quoten-Info: {result.get('quota_info')}")
print(f" Response: {result.get('choices', [{}])[0].get('message', {}).get('content', '')[:100]}...")
if __name__ == "__main__":
asyncio.run(main())
Schritt 2: Rate Limiting mit Sliding Window
Token-Limits sind nur eine Dimension. Sie müssen auch die Anfragenrate kontrollieren, um Missbrauch zu verhindern und eine faire Ressourcenverteilung zu gewährleisten. Das Sliding-Window-Algorithmus bietet hier die beste Balance zwischen Genauigkeit und Performance:
#!/usr/bin/env python3
"""
Advanced Rate Limiter mit Sliding Window Algorithmus
Für HolySheep AI API Integration
"""
import time
import asyncio
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, Optional, Callable
from datetime import datetime, timedelta
import hashlib
@dataclass
class RateLimitConfig:
"""Rate-Limit Konfiguration"""
requests_per_minute: int = 60
requests_per_hour: int = 1000
requests_per_day: int = 10000
burst_allowance: int = 10 # Erlaubt kurze Spitzen
@dataclass
class SlidingWindowRateLimiter:
"""
Effizienter Rate-Limiter mit Sliding Window Counter.
Bietet granulare Kontrolle über Anfragenraten.
"""
config: RateLimitConfig = field(default_factory=RateLimitConfig)
_minute_windows: Dict[str, deque] = field(default_factory=dict)
_hour_windows: Dict[str, deque] = field(default_factory=dict)
_day_windows: Dict[str, deque] = field(default_factory=dict)
_locks: Dict[str, asyncio.Lock] = field(default_factory=dict)
def _get_key(self, identifier: str, endpoint: str = "") -> str:
"""Generiert eindeutigen Key für Rate-Limiting"""
return hashlib.sha256(f"{identifier}:{endpoint}".encode()).hexdigest()[:16]
def _get_lock(self, key: str) -> asyncio.Lock:
"""Thread-sicherer Lock für jeden Key"""
if key not in self._locks:
self._locks[key] = asyncio.Lock()
return self._locks[key]
async def check_and_record(
self,
identifier: str,
endpoint: str = "chat/completions"
) -> Dict:
"""
Prüft Rate-Limit und zeichnet Anfrage auf.
Gibt Status und Metriken zurück.
"""
key = self._get_key(identifier, endpoint)
lock = self._get_lock(key)
now = time.time()
async with lock:
# Initialisiere Windows falls nötig
if key not in self._minute_windows:
self._minute_windows[key] = deque()
if key not in self._hour_windows:
self._hour_windows[key] = deque()
if key not in self._day_windows:
self._day_windows[key] = deque()
# Bereinige alte Einträge (älter als 1 Minute)
self._clean_old_entries(self._minute_windows[key], now, 60)
self._clean_old_entries(self._hour_windows[key], now, 3600)
self._clean_old_entries(self._day_windows[key], now, 86400)
# Prüfe Minute-Limit
minute_count = len(self._minute_windows[key])
if minute_count >= self.config.requests_per_minute:
oldest = self._minute_windows[key][0]
retry_after = int(60 - (now - oldest)) + 1
return {
"allowed": False,
"reason": "MINUTE_RATE_LIMIT",
"retry_after_seconds": max(1, retry_after),
"limit": self.config.requests_per_minute,
"current": minute_count
}
# Prüfe Hour-Limit
hour_count = len(self._hour_windows[key])
if hour_count >= self.config.requests_per_hour:
oldest = self._hour_windows[key][0]
retry_after = int(3600 - (now - oldest)) + 1
return {
"allowed": False,
"reason": "HOUR_RATE_LIMIT",
"retry_after_seconds": max(1, retry_after),
"limit": self.config.requests_per_hour,
"current": hour_count
}
# Prüfe Day-Limit
day_count = len(self._day_windows[key])
if day_count >= self.config.requests_per_day:
oldest = self._day_windows[key][0]
retry_after = int(86400 - (now - oldest)) + 1
return {
"allowed": False,
"reason": "DAY_RATE_LIMIT",
"retry_after_seconds": max(1, retry_after),
"limit": self.config.requests_per_day,
"current": day_count
}
# Alle Checks bestanden - zeichne Anfrage auf
self._minute_windows[key].append(now)
self._hour_windows[key].append(now)
self._day_windows[key].append(now)
return {
"allowed": True,
"minute_remaining": self.config.requests_per_minute - minute_count - 1,
"hour_remaining": self.config.requests_per_hour - hour_count - 1,
"day_remaining": self.config.requests_per_day - day_count - 1
}
def _clean_old_entries(self, window: deque, now: float, window_seconds: int):
"""Entfernt veraltete Einträge aus dem Window"""
cutoff = now - window_seconds
while window and window[0] < cutoff:
window.popleft()
def get_current_status(self, identifier: str) -> Dict:
"""Gibt aktuellen Rate-Limit Status zurück (ohne zu recorden)"""
key = self._get_key(identifier)
now = time.time()
return {
"requests_last_minute": len(self._minute_windows.get(key, [])),
"requests_last_hour": len(self._hour_windows.get(key, [])),
"requests_last_day": len(self._day_windows.get(key, [])),
"minute_limit": self.config.requests_per_minute,
"hour_limit": self.config.requests_per_hour,
"day_limit": self.config.requests_per_day
}
class IntelligentAPIGateway:
"""
Intelligentes API-Gateway das Rate-Limiting und Quoten-Enforcement
für HolySheep AI kombiniert.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.rate_limiter = SlidingWindowRateLimiter(
config=RateLimitConfig(
requests_per_minute=60,
requests_per_hour=1000,
requests_per_day=10000
)
)
# User-spezifische Limits
self.user_limits: Dict[str, RateLimitConfig] = {}
async def make_request(
self,
user_id: str,
model: str,
messages: list,
max_tokens: int = 2048,
callback: Optional[Callable] = None
) -> Dict:
"""
Kombiniert Rate-Limiting mit API-Request.
Behandelt automatisch Limits und Backoff.
"""
# Hole user-spezifisches Limit oder Default
limit_config = self.user_limits.get(user_id, self.rate_limiter.config)
# Erstelle temporären Rate-Limiter mit User-spezifischen Limits
user_rate_limiter = SlidingWindowRateLimiter(config=limit_config)
# Prüfe Rate-Limit
rate_status = await user_rate_limiter.check_and_record(user_id)
if not rate_status["allowed"]:
return {
"success": False,
"error": "RATE_LIMIT_EXCEEDED",
"rate_limit_info": rate_status,
"retry_after": rate_status["retry_after_seconds"]
}
# Führe Request durch
try:
import httpx
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens
}
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
# Callback für Tracking
if callback:
await callback(user_id, model, result)
return {
"success": True,
"data": result,
"rate_limit_info": rate_status
}
except Exception as e:
return {
"success": False,
"error": str(e),
"rate_limit_info": rate_status
}
==================== BEISPIEL-NUTZUNG ====================
async def example_gateway_usage():
"""Demonstriert das Gateway mit mehreren Anfragen"""
gateway = IntelligentAPIGateway(api_key="YOUR_HOLYSHEEP_API_KEY")
# Premium User mit höheren Limits
gateway.user_limits["premium_user_1"] = RateLimitConfig(
requests_per_minute=120,
requests_per_hour=5000,
requests_per_day=50000
)
messages = [
{"role": "user", "content": "Erkläre mir die Vorteile von HolySheep AI in 3 Sätzen."}
]
# Simuliere 5 Anfragen
for i in range(5):
result = await gateway.make_request(
user_id="premium_user_1",
model="deepseek-v3.2",
messages=messages
)
if result["success"]:
print(f"✅ Anfrage {i+1}: Erfolgreich")
print(f" Rate-Status: {result['rate_limit_info']}")
else:
print(f"❌ Anfrage {i+1}: {result['error']}")
if "retry_after" in result:
print(f" Retry nach: {result['retry_after']}s")
await asyncio.sleep(0.5) # Kurze Pause zwischen Anfragen
if __name__ == "__main__":
asyncio.run(example_gateway_usage())
Schritt 3: Budget-Tracking und Kostenkontrolle
In meiner Praxis hat sich gezeigt, dass Token-Limits allein nicht ausreichen. Sie brauchen auch ein präzises Budget-Tracking, um unerwartete Kosten zu vermeiden. HolySheep AI's Preisstruktur mit DeepSeek V3.2 zu $0.42/MTok macht kosteneffizientes Skalieren möglich:
#!/usr/bin/env python3
"""
Budget Tracker und Cost Alert System
Überwacht AI API Ausgaben in Echtzeit
"""
import asyncio
import httpx
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Callable
from enum import Enum
import json
class AlertLevel(Enum):
"""Warnstufen für Budget-Überschreitungen"""
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
EMERGENCY = "emergency"
@dataclass
class BudgetConfig:
"""Budget-Konfiguration"""
daily_budget_usd: float = 10.0
weekly_budget_usd: float = 50.0
monthly_budget_usd: float = 200.0
per_request_max_usd: float = 0.50
# Alert-Schwellenwerte (Prozent des Budgets)
alert_threshold_warning: float = 0.70 # 70%
alert_threshold_critical: float = 0.90 # 90%
alert_threshold_emergency: float = 0.95 # 95%
@dataclass
class CostRecord:
"""Einzelne Kostenaufzeichnung"""
timestamp: datetime
user_id: str
model: str
input_tokens: int
output_tokens: int
cost_usd: float
request_id: str
class BudgetTracker:
"""
Echtzeit-Budget-Tracker mit Alert-System.
Verhindert Kosten-Überraschungen.
"""
# HolySheep AI Preise (USD per Million Token) - Stand 2026
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 2.00, "output": 6.00}, # $8 avg
"claude-sonnet-4.5": {"input": 3.00, "output": 12.00}, # $15 avg
"gemini-2.5-flash": {"input": 0.15, "output": 0.60}, # $2.50 avg
"deepseek-v3.2": {"input": 0.10, "output": 0.32}, # $0.42 avg
}
def __init__(self, config: Optional[BudgetConfig] = None):
self.config = config or BudgetConfig()
self.cost_records: List[CostRecord] = []
self.alerts: List[Dict] = []
self.alert_callbacks: List[Callable] = []
# Aggregierte Kosten
self._daily_costs: Dict[str, float] = {}
self._weekly_costs: Dict[str, float] = {}
self._monthly_costs: Dict[str, float] = {}
def calculate_request_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Berechnet Kosten für einen Request in USD"""
pricing = self.HOLYSHEEP_PRICING.get(model, self.HOLYSHEEP_PRICING["deepseek-v3.2"])
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6) # 6 Dezimalstellen für Präzision
async def record_request(
self,
user_id: str,
model: str,
input_tokens: int,
output_tokens: int,
request_id: str
) -> Dict:
"""
Zeichnet Request auf und prüft Budget-Limits.
Gibt Tuple zurück: (allowed, alert_info)
"""
cost = self.calculate_request_cost(model, input_tokens, output_tokens)
now = datetime.now()
# Prüfe maximale Request-Kosten
if cost > self.config.per_request_max_usd:
alert = self._create_alert(
level=AlertLevel.WARNING,
message=f"Request-Kosten ({cost:.4f}$) überschreiten Limit ({self.config.per_request_max_usd}$)",
user_id=user_id
)
self.alerts.append(alert)
return False, alert
# Aktualisiere aggregierte Kosten
self._update_costs(user_id, cost)
# Prüfe alle Budget-Limits
today_key = now.strftime("%Y-%m-%d")
week_key = now.strftime("%Y-W%W")
month_key = now.strftime("%Y-%m")
# Tägliches Budget
daily_total = self._daily_costs.get(today_key, 0)
if daily_total > self.config.daily_budget_usd:
alert = self._create_alert(
level=AlertLevel.EMERGENCY,
message=f"Tagesbudget überschritten: {daily_total:.2f}$ / {self.config.daily_budget_usd}$",
user_id=user_id,
current_spend=daily_total,
budget_limit=self.config.daily_budget_usd
)
self.alerts.append(alert)
return False, alert
# Wöchentliches Budget
weekly_total = self._weekly_costs.get(week_key, 0)
if weekly_total > self.config.weekly_budget_usd:
alert = self._create_alert(
level=AlertLevel.CRITICAL,
message=f"Weekly-Budget erreicht: {weekly_total:.2f}$ / {self.config.weekly_budget_usd}$",
user_id=user_id
)
self.alerts.append(alert)
# Monatliches Budget
monthly_total = self._monthly_costs.get(month_key, 0)
if monthly_total > self.config.monthly_budget_usd:
alert = self._create_alert(
level=AlertLevel.EMERGENCY,
message=f"Monatsbudget überschritten: {monthly_total:.2f}$ / {self.config.monthly_budget_usd}$",
user_id=user_id
)
self.alerts.append(alert)
return False, alert
# Record the request
record = CostRecord(
timestamp=now,
user_id=user_id,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost,
request_id=request_id
)
self.cost_records.append(record)
# Check alert thresholds
await self._check_alert_thresholds(monthly_total, user_id)
return True, {"cost": cost, "daily_remaining": self.config.daily_budget_usd - daily_total}
def _update_costs(self, user_id: str, cost: float):
"""Aktualisiert aggregierte Kosten"""
now = datetime.now()
today_key = now.strftime("%Y-%m-%d")
week_key = now.strftime("%Y-W%W")
month_key = now.strftime("%Y-%m")
self._daily_costs[today_key] = self._daily_costs.get(today_key, 0) + cost
self._weekly_costs[week_key] = self._weekly_costs.get(week_key, 0) + cost
self._monthly_costs[month_key] = self._monthly_costs.get(month_key, 0) + cost
def _create_alert(
self,
level: AlertLevel,
message: str,
user_id: str,
**kwargs
) -> Dict:
"""Erstellt Alert-Record"""
return {
"level": level.value,
"message": message,
"user_id": user_id,
"timestamp": datetime.now().isoformat(),
**kwargs
}
async def _check_alert_thresholds(self, current_spend: float, user_id: str):
"""Prüft ob Alert-Schwellen erreicht wurden"""
monthly_budget = self.config.monthly_budget_usd
percentage = current_spend / monthly_budget
if percentage >= self.config.alert_threshold_emergency:
alert = self._create_alert(
level=AlertLevel.EMERGENCY,
message=f"Notfall: 95%+ des Monatsbudgets verbraucht",
user_id=user_id,
percentage=round(percentage * 100, 1)
)
await self._trigger_callbacks(alert)
elif percentage >= self.config.alert_threshold_critical:
alert = self._create_alert(
level=AlertLevel.CRITICAL,
message=f"Kritisch: 90%+ des Monatsbudgets verbraucht",
user_id=user_id,
percentage=round(percentage * 100, 1)
)
await self._trigger_callbacks(alert)
elif percentage >= self.config.alert_threshold_warning:
alert = self._create_alert(
level=AlertLevel.WARNING,
message=f"Warnung: 70%+ des Monatsbudgets verbraucht",
user_id=user_id,
percentage=round(percentage * 100, 1)
)
await self._trigger_callbacks(alert)
def register_alert_callback(self, callback: Callable):
"""Registriert Callback für Alerts"""
self.alert_callbacks.append(callback)
async def _trigger_callbacks(self, alert: Dict):
"""Triggert alle registrierten Alert-Callbacks"""
for callback in self.alert_callbacks:
try:
if asyncio.iscoroutinefunction(callback):
await callback(alert)
else:
callback(alert)
except Exception as e:
print(f"Alert callback error: {e}")
def get_budget_summary(self, user_id: Optional[str] = None) -> Dict:
"""Gibt vollständige