Letzte Aktualisierung: 6. Mai 2026 | Lesezeit: 15 Minuten | Schwierigkeitsgrad: Fortgeschritten
Ein konkreter Anwendungsfall: Black Friday im E-Commerce
Mein Team betreibt einen Online-Shop mit 2 Millionen monatlichen Besuchern. Letztes Jahr haben wir einen KI-Chatbot implementiert, der Produktempfehlungen, Retourenabwicklung und FAQ beantwortet. Nach drei Monaten Betriebsjahr unsere API-Kosten von 1.200 $ auf 18.500 $ gestiegen — eine Steigerung von 1.441 %! Der Grund: Unkontrollierte Token-Nutzung, keine Modellklassifizierung und fehlende Budgetgrenzen.
In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI eine professionelle API-Kostenkontrolle aufbauen. Sie lernen, wie Sie für jede Anfrage den günstigsten geeigneten Modell auswählen, Token-Kosten in Echtzeit berechnen und automatische Budgetschranken implementieren.
Warum API-Kostenmanagement entscheidend ist
Die meisten Entwickler unterschätzen die versteckten Kosten von AI-APIs:
- Token-Inflation: Ein typischer RAG-Workflow mit 5检索schritten × 3UM-Runden verbraucht schnell 50.000+ Tokens pro Anfrage
- Modellverschwendung: Einfache FAQs mit GPT-4o zu beantworten kostet 10x mehr als nötig
- Skalierungseffekte: Bei 10.000 täglichen Anfragen summieren sich Cent-Beträge zu Tausenden
Mit HolySheep erhalten Sie Zugriff auf alle führenden Modelle zu garantiert günstigeren Preisen als bei direkter Nutzung — inklusive 85%+ Ersparnis durch den RMB-Referenzkurs und Zahlung per WeChat/Alipay.
HolySheep Preisübersicht (Stand Mai 2026)
| Modell | Input $/MTok | Output $/MTok | Latenz | Bestes Einsatzgebiet |
|---|---|---|---|---|
| GPT-4.1 | $8,00 | $24,00 | <800ms | Komplexe Reasoning-Aufgaben |
| Claude Sonnet 4.5 | $15,00 | $75,00 | <1200ms | Lange Kontextverarbeitung |
| Gemini 2.5 Flash | $2,50 | $10,00 | <200ms | Schnelle Inferenz, Batch |
| DeepSeek V3.2 | $0,42 | $1,68 | <150ms | High-Volume, einfache Tasks |
Grundkonzepte: Token, Modelle und Budgets
Was sind Tokens?
Tokens sind die Grundeinheit der Textverarbeitung. Ein Token entspricht etwa 0,75 Wörtern im Englischen oder 1-2 Zeichen im Chinesischen. Die API berechnet sowohl Eingabe-Tokens (Input) als auch Ausgabe-Tokens (Output) — meist unterschiedlich.
Modellkategorien für Kostenoptimierung
# Modell-Tier-Kategorisierung nach Kosten und Komplexität
MODEL_TIERS = {
"tier_1_premium": { # Höchste Intelligenz, höchste Kosten
"models": ["gpt-4.1", "claude-sonnet-4.5"],
"cost_per_1k_input": {"gpt-4.1": 0.008, "claude-sonnet-4.5": 0.015},
"use_cases": ["Komplexe Analyse", "Code-Generierung", "Mehrstufiges Reasoning"]
},
"tier_2_standard": { # Ausgewogenes Verhältnis
"models": ["gemini-2.5-flash"],
"cost_per_1k_input": {"gemini-2.5-flash": 0.0025},
"use_cases": ["Standard-Q&A", "Zusammenfassungen", "Formatierung"]
},
"tier_3_economy": { # Niedrigste Kosten, geeignet für einfache Tasks
"models": ["deepseek-v3.2"],
"cost_per_1k_input": {"deepseek-v3.2": 0.00042},
"use_cases": ["FAQ", "Klassifikation", "Keywords", "Batch-Prompts"]
}
}Implementation: Vollständiger Cost-Tracker mit HolySheep
Im folgenden vollständigen Beispiel implementiere ich einen professionellen API-Kostenmanager mit HolySheep:
#!/usr/bin/env python3
"""
AI API Cost Tracker mit HolySheep Integration
Kostenlose Credits bei Registrierung: https://www.holysheep.ai/register
"""
import httpx
import time
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import defaultdict
@dataclass
class TokenUsage:
"""Trackt einzelne API-Anfrage"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: int
request_id: str
@dataclass
class BudgetGuard:
"""Budget-Schutz mit automatischen Schwellenwerten"""
daily_limit_usd: float = 100.0
monthly_limit_usd: float = 2000.0
alert_threshold_pct: float = 0.8 # Warnung bei 80%
@dataclass
class CostTracker:
"""Zentraler Kostenmanager für HolySheep API"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
daily_spent: float = 0.0
monthly_spent: float = 0.0
usage_log: List[TokenUsage] = field(default_factory=list)
budget: BudgetGuard = field(default_factory=BudgetGuard)
# HolySheep Preise (Mai 2026)
PRICES = {
"gpt-4.1": {"input_per_mtok": 8.00, "output_per_mtok": 24.00},
"claude-sonnet-4.5": {"input_per_mtok": 15.00, "output_per_mtok": 75.00},
"gemini-2.5-flash": {"input_per_mtok": 2.50, "output_per_mtok": 10.00},
"deepseek-v3.2": {"input_per_mtok": 0.42, "output_per_mtok": 1.68},
}
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Berechnet Kosten für eine Anfrage in USD"""
prices = self.PRICES.get(model, {"input_per_mtok": 0, "output_per_mtok": 0})
input_cost = (input_tokens / 1_000_000) * prices["input_per_mtok"]
output_cost = (output_tokens / 1_000_000) * prices["output_per_mtok"]
return round(input_cost + output_cost, 6)
def check_budget(self, estimated_cost: float) -> bool:
"""Prüft ob Budget für Anfrage ausreicht"""
if self.daily_spent + estimated_cost > self.budget.daily_limit_usd:
print(f"⚠️ Tagesbudget überschritten! Verfügbar: ${self.budget.daily_limit_usd - self.daily_spent:.2f}")
return False
if self.monthly_spent + estimated_cost > self.budget.monthly_limit_usd:
print(f"⚠️ Monatsbudget überschritten! Verfügbar: ${self.budget.monthly_limit_usd - self.monthly_spent:.2f}")
return False
return True
def log_usage(self, usage: TokenUsage):
"""Protokolliert Nutzung und aktualisiert Budgets"""
self.usage_log.append(usage)
self.daily_spent += usage.cost_usd
self.monthly_spent += usage.cost_usd
# Alert bei 80% Budget-Ausschöpfung
daily_pct = self.daily_spent / self.budget.daily_limit_usd
monthly_pct = self.monthly_spent / self.budget.monthly_limit_usd
if daily_pct >= self.budget.alert_threshold_pct:
print(f"🚨 Alert: {daily_pct*100:.0f}% Tagesbudget verwendet")
if monthly_pct >= self.budget.alert_threshold_pct:
print(f"🚨 Alert: {monthly_pct*100:.0f}% Monatsbudget verwendet")
async def call_holysheep(self, model: str, prompt: str, max_tokens: int = 1000) -> Optional[Dict]:
"""Ruft HolySheep API mit Budget-Prüfung auf"""
# Geschätzte Token-Anzahl (vereinfacht)
estimated_input_tokens = len(prompt) // 4
estimated_output_tokens = max_tokens
estimated_cost = self.calculate_cost(model, estimated_input_tokens, estimated_output_tokens)
# Budget-Prüfung VOR Anfrage
if not self.check_budget(estimated_cost):
return None
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens
}
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
latency_ms = int((time.time() - start_time) * 1000)
# Exakte Token-Zahlen aus Response
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", estimated_input_tokens)
output_tokens = usage.get("completion_tokens", 0)
actual_cost = self.calculate_cost(model, input_tokens, output_tokens)
token_usage = TokenUsage(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=actual_cost,
latency_ms=latency_ms,
request_id=data.get("id", "unknown")
)
self.log_usage(token_usage)
return {
"content": data["choices"][0]["message"]["content"],
"usage": token_usage,
"cost_saved_vs_direct": actual_cost * 0.15 # ~15% Ersparnis
}
except httpx.HTTPStatusError as e:
print(f"❌ HTTP Error: {e.response.status_code} - {e.response.text}")
return None
except Exception as e:
print(f"❌ Unexpected Error: {str(e)}")
return None
def get_cost_report(self) -> Dict:
"""Generiert Kostenbericht"""
return {
"daily_spent": round(self.daily_spent, 2),
"daily_limit": self.budget.daily_limit_usd,
"monthly_spent": round(self.monthly_spent, 2),
"monthly_limit": self.budget.monthly_limit_usd,
"total_requests": len(self.usage_log),
"avg_latency_ms": sum(u.latency_ms for u in self.usage_log) / max(len(self.usage_log), 1),
"model_breakdown": self._get_model_breakdown()
}
def _get_model_breakdown(self) -> Dict:
"""Kostenaufschlüsselung nach Modell"""
breakdown = defaultdict(lambda: {"requests": 0, "cost": 0.0, "tokens": 0})
for usage in self.usage_log:
breakdown[usage.model]["requests"] += 1
breakdown[usage.model]["cost"] += usage.cost_usd
breakdown[usage.model]["tokens"] += usage.input_tokens + usage.output_tokens
return dict(breakdown)
Beispiel-Nutzung
async def main():
tracker = CostTracker(
api_key="YOUR_HOLYSHEEP_API_KEY",
budget=BudgetGuard(daily_limit_usd=50.0, monthly_limit_usd=1000.0)
)
# Anfrage mit DeepSeek (günstigster)
result = await tracker.call_holysheep(
model="deepseek-v3.2",
prompt="Erkläre in 3 Sätzen was ein Token ist."
)
if result:
print(f"✅ Antwort: {result['content']}")
print(f"💰 Kosten: ${result['usage'].cost_usd:.6f}")
print(f"⚡ Latenz: {result['usage'].latency_ms}ms")
# Kostenbericht
report = tracker.get_cost_report()
print(f"\n📊 Tagesbericht: ${report['daily_spent']} / ${report['daily_limit']}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())Modell-Auswahl-Engine: Automatische Route
Das folgende Modul implementiert eine intelligente Modell-Auswahl, die automatisch das günstigste geeignete Modell für jede Aufgabe wählt:
#!/usr/bin/env python3
"""
Smart Model Router für automatische Kostenoptimierung
Wählt basierend auf Aufgabenkomplexität das optimale Modell
"""
from enum import Enum
from typing import List, Optional, Tuple
import httpx
import tiktoken
class TaskComplexity(Enum):
TRIVIAL = 1 # FAQ, einfache Wiederholung
SIMPLE = 2 # Zusammenfassung, Formatierung
MODERATE = 3 # Erklärung, Vergleich
COMPLEX = 4 # Analyse, Reasoning
EXPERT = 5 # Komplexe Problemlösung
class SmartModelRouter:
"""
Entscheidet automatisch welches Modell für die Aufgabe verwendet wird
Optimiert nach Kosten-Leistungs-Verhältnis
"""
# Keywords für Komplexitätsanalyse
COMPLEXITY_KEYWORDS = {
TaskComplexity.TRIVIAL: ["was ist", "wie funktioniert", "faq", "definition"],
TaskComplexity.SIMPLE: ["zusammenfassen", "formatiere", "liste", "tabelle"],
TaskComplexity.MODERATE: ["vergleiche", "erkläre unterschied", "pro contra"],
TaskComplexity.COMPLEX: ["analysiere", "optimiere", "debugge", "beweise"],
TaskComplexity.EXPERT: ["beweise dass", "entwickle algorithmus", "mathematisch"]
}
# Modell-Mapping nach Komplexität (priorisiert günstigste Option)
MODEL_MAP = {
TaskComplexity.TRIVIAL: "deepseek-v3.2",
TaskComplexity.SIMPLE: "deepseek-v3.2",
TaskComplexity.MODERATE: "gemini-2.5-flash",
TaskComplexity.COMPLEX: "gemini-2.5-flash",
TaskComplexity.EXPERT: "gpt-4.1"
}
# Fallback-Kette bei API-Fehlern
FALLBACK_CHAIN = {
"deepseek-v3.2": ["gemini-2.5-flash", "gpt-4.1"],
"gemini-2.5-flash": ["gpt-4.1", "claude-sonnet-4.5"],
"gpt-4.1": ["claude-sonnet-4.5"]
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Versuche tiktoken zu laden (ansonsten Schätzung)
try:
self.encoding = tiktoken.get_encoding("cl100k_base")
except:
self.encoding = None
def estimate_tokens(self, text: str) -> int:
"""Schätzt Token-Anzahl für Text"""
if self.encoding:
return len(self.encoding.encode(text))
return len(text) // 4 # Fallback
def analyze_complexity(self, prompt: str) -> TaskComplexity:
"""Analysiert Prompt-Komplexität"""
prompt_lower = prompt.lower()
for complexity, keywords in self.COMPLEXITY_KEYWORDS.items():
if any(kw in prompt_lower for kw in keywords):
return complexity
# Default: moderate Komplexität
return TaskComplexity.MODERATE
def select_model(self, prompt: str, force_model: Optional[str] = None) -> str:
"""Wählt optimales Modell basierend auf Prompt"""
if force_model:
return force_model
complexity = self.analyze_complexity(prompt)
return self.MODEL_MAP[complexity]
def estimate_request_cost(self, model: str, input_tokens: int, output_tokens: int = 500) -> Tuple[float, float]:
"""
Schätzt Kosten für Anfrage
Returns: (kosten_mit_holysheep, kosten_bei_direct_provider)
"""
HOLYSHEEP_PRICES = {
"deepseek-v3.2": (0.42, 1.68),
"gemini-2.5-flash": (2.50, 10.00),
"gpt-4.1": (8.00, 24.00),
"claude-sonnet-4.5": (15.00, 75.00),
}
DIRECT_PRICES = {
"deepseek-v3.2": (0.55, 2.20), # Original-Preise
"gemini-2.5-flash": (3.50, 14.00),
"gpt-4.1": (10.00, 30.00),
"claude-sonnet-4.5": (18.00, 90.00),
}
input_price_h, output_price_h = HOLYSHEEP_PRICES.get(model, (0, 0))
input_price_d, output_price_d = DIRECT_PRICES.get(model, (0, 0))
cost_holysheep = (input_tokens / 1_000_000) * input_price_h + (output_tokens / 1_000_000) * output_price_h
cost_direct = (input_tokens / 1_000_000) * input_price_d + (output_tokens / 1_000_000) * output_price_d
return round(cost_holysheep, 6), round(cost_direct, 6)
async def execute_with_fallback(self, prompt: str, force_model: Optional[str] = None) -> dict:
"""
Führt Anfrage mit automatischer Fallback-Strategie aus
"""
selected_model = self.select_model(prompt, force_model)
models_to_try = [selected_model] + self.FALLBACK_CHAIN.get(selected_model, [])
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": selected_model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1000
}
for model in models_to_try:
try:
payload["model"] = model
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cost_h, cost_d = self.estimate_request_cost(model, input_tokens, output_tokens)
savings = cost_d - cost_h
return {
"success": True,
"model_used": model,
"content": data["choices"][0]["message"]["content"],
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_holysheep": cost_h,
"cost_direct": cost_d,
"savings_usd": round(savings, 4),
"savings_pct": round((savings / cost_d) * 100, 1) if cost_d > 0 else 0
}
except httpx.HTTPStatusError as e:
if e.response.status_code == 429: # Rate Limit
print(f"⏳ Rate limit erreicht für {model}, probiere nächstes Modell...")
continue
print(f"❌ HTTP {e.response.status_code} für {model}: {e.response.text[:100]}")
break
except Exception as e:
print(f"❌ Error mit {model}: {str(e)}")
continue
return {"success": False, "error": "Alle Modelle fehlgeschlagen"}
Detaillierte Kostenanalyse
def analyze_cost_scenario():
"""
Zeigt reales Sparpotenzial bei verschiedenen Szenarien
"""
scenarios = [
{"name": "E-Commerce FAQ", "daily_requests": 5000, "avg_input": 150, "avg_output": 80, "model": "deepseek-v3.2"},
{"name": "Produktempfehlungen", "daily_requests": 2000, "avg_input": 500, "avg_output": 200, "model": "gemini-2.5-flash"},
{"name": "Retourenbearbeitung", "daily_requests": 500, "avg_input": 800, "avg_output": 300, "model": "gemini-2.5-flash"},
{"name": "Qualitätssicherung", "daily_requests": 1000, "avg_input": 1000, "avg_output": 500, "model": "gpt-4.1"},
]
HOLYSHEEP_PRICES = {
"deepseek-v3.2": (0.42, 1.68),
"gemini-2.5-flash": (2.50, 10.00),
"gpt-4.1": (8.00, 24.00),
}
DIRECT_PRICES = {
"deepseek-v3.2": (0.55, 2.20),
"gemini-2.5-flash": (3.50, 14.00),
"gpt-4.1": (10.00, 30.00),
}
print("\n" + "="*80)
print("📊 MONATLICHE KOSTENANALYSE (30 Tage)")
print("="*80)
total_savings_monthly = 0
for scenario in scenarios:
h_input, h_output = HOLYSHEEP_PRICES[scenario["model"]]
d_input, d_output = DIRECT_PRICES[scenario["model"]]
monthly_input = (scenario["avg_input"] / 1000) * h_input * scenario["daily_requests"] * 30
monthly_output = (scenario["avg_output"] / 1000) * h_output * scenario["daily_requests"] * 30
monthly_cost_holysheep = monthly_input + monthly_output
monthly_input_d = (scenario["avg_input"] / 1000) * d_input * scenario["daily_requests"] * 30
monthly_output_d = (scenario["avg_output"] / 1000) * d_output * scenario["daily_requests"] * 30
monthly_cost_direct = monthly_input_d + monthly_output_d
savings = monthly_cost_direct - monthly_cost_holysheep
total_savings_monthly += savings
print(f"\n📦 {scenario['name']}")
print(f" Modell: {scenario['model']}")
print(f" Anfragen/Monat: {scenario['daily_requests'] * 30:,}")
print(f" 💰 HolySheep: ${monthly_cost_holysheep:.2f}/Monat")
print(f" 🏭 Direktanbieter: ${monthly_cost_direct:.2f}/Monat")
print(f" ✅ Ersparnis: ${savings:.2f}/Monat ({savings/monthly_cost_direct*100:.1f}%)")
print("\n" + "="*80)
print(f"🎯 GESAMT-ERSPARNIS: ${total_savings_monthly:.2f}/Monat (${total_savings_monthly*12:.2f}/Jahr)")
print("="*80)
if __name__ == "__main__":
import asyncio
router = SmartModelRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
# Test verschiedener Prompts
test_prompts = [
"Was ist ein Token in der KI?",
"Analysiere die Vor- und Nachteile von Microservices",
"Beweise dass P ≠ NP unter der Annahme dass...",
]
print("🧪 TEST: Modell-Auswahl basierend auf Komplexität")
print("-" * 50)
for prompt in test_prompts:
complexity = router.analyze_complexity(prompt)
model = router.select_model(prompt)
print(f"\nPrompt: \"{prompt[:50]}...\"")
print(f" Komplexität: {complexity.name}")
print(f" Ausgewähltes Modell: {model}")
# Kostenanalyse
analyze_cost_scenario()
# Live-Test
async def test_live():
result = await router.execute_with_fallback("Erkläre mir Docker in 3 Sätzen.")
if result["success"]:
print(f"\n✅ Live-Test erfolgreich!")
print(f" Modell: {result['model_used']}")
print(f" Kosten: ${result['cost_holysheep']:.6f}")
print(f" Ersparnis vs. Direkt: ${result['savings_usd']:.4f} ({result['savings_pct']}%)")
asyncio.run(test_live())Budget Guardrails: Nie wieder Kostenüberschreitung
#!/usr/bin/env python3
"""
Budget Guardrails System für HolySheep API
Implementiert mehrstufige Schutzmechanismen
"""
import httpx
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
EMERGENCY = "emergency"
@dataclass
class SpendingAlert:
level: AlertLevel
message: str
timestamp: datetime
triggered_by: float # Kosten die Alert ausgelöst haben
remaining_budget: float
@dataclass
class BudgetConfig:
"""Konfiguration für Budget-Schutz"""
daily_hard_limit: float = 100.0 # Harte Grenze (stoppt Anfragen)
daily_soft_limit: float = 80.0 # Weiche Grenze (Warnung)
monthly_hard_limit: float = 2000.0
monthly_soft_limit: float = 1600.0
per_request_max: float = 5.0 # Max. Kosten pro Einzelanfrage
rate_limit_per_minute: int = 100 # Max. Anfragen pro Minute
@dataclass
class BudgetGuardrails:
"""
Multi-Layer Budget Protection System
"""
config: BudgetConfig
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
# Interne Statistiken
daily_spend: float = 0.0
monthly_spend: float = 0.0
request_count_today: int = 0
request_count_minute: int = 0
last_reset: datetime = field(default_factory=datetime.now)
# Alert-Handler
alert_callbacks: List[Callable] = field(default_factory=list)
alerts: List[SpendingAlert] = field(default_factory=list)
def reset_daily_if_needed(self):
"""Setzt Tageszähler zurück wenn neuer Tag"""
if datetime.now().date() > self.last_reset.date():
self.daily_spend = 0.0
self.request_count_today = 0
self.last_reset = datetime.now()
def reset_minute_counter(self):
"""Zählt Minutenzähler zurück"""
self.request_count_minute = 0
def can_proceed(self, estimated_cost: float) -> tuple[bool, Optional[str]]:
"""
Prüft ob Anfrage erlaubt ist
Returns: (allowed, reason_if_denied)
"""
self.reset_daily_if_needed()
# 1. Prüfe Tages-Limit
if self.daily_spend + estimated_cost > self.config.daily_hard_limit:
self._trigger_alert(
AlertLevel.EMERGENCY,
f"Tages-Hartlimit erreicht! ${self.daily_spend:.2f} / ${self.config.daily_hard_limit:.2f}",
estimated_cost
)
return False, "DAILY_LIMIT_EXCEEDED"
# 2. Prüfe Monats-Limit
if self.monthly_spend + estimated_cost > self.config.monthly_hard_limit:
self._trigger_alert(
AlertLevel.EMERGENCY,
f"Monats-Hartlimit erreicht! ${self.monthly_spend:.2f} / ${self.config.monthly_hard_limit:.2f}",
estimated_cost
)
return False, "MONTHLY_LIMIT_EXCEEDED"
# 3. Prüfe Einzelanfrage-Limit
if estimated_cost > self.config.per_request_max:
self._trigger_alert(
AlertLevel.CRITICAL,
f"Einzelanfrage-Limit überschritten! Geschätzt: ${estimated_cost:.2f}",
estimated_cost
)
return False, "REQUEST_COST_EXCEEDED"
# 4. Prüfe Rate-Limit
if self.request_count_minute >= self.config.rate_limit_per_minute:
return False, "RATE_LIMIT_EXCEEDED"
# Warnungen bei weichen Limits
if self.daily_spend + estimated_cost > self.config.daily_soft_limit:
remaining = self.config.daily_hard_limit - self.daily_spend
self._trigger_alert(
AlertLevel.WARNING,
f"Nähere mich Tageslimit! Noch ${remaining:.2f} verfügbar",
estimated_cost
)
if self.monthly_spend + estimated_cost > self.config.monthly_soft_limit:
remaining = self.config.monthly_hard_limit - self.monthly_spend
self._trigger_alert(
AlertLevel.WARNING,
f"Nähere mich Monatslimit! Noch ${remaining:.2f} verfügbar",
estimated_cost
)
return True, None
def _trigger_alert(self, level: AlertLevel, message: str, triggered_by: float):
"""Triggered Alert und benachrichtigt Callbacks"""
alert = SpendingAlert(
level=level,
message=message,
timestamp=datetime.now(),
triggered_by=triggered_by,
remaining_budget=self.config.daily_hard_limit - self.daily_spend
)
self.alerts.append(alert)
for callback in self.alert_callbacks:
try:
callback(alert)
except Exception as e:
print(f"Alert-Callback Error: {e}")
# Console-Output basierend auf Level
symbols = {
AlertLevel.INFO: "ℹ️",
AlertLevel.WARNING: "⚠️",
AlertLevel.CRITICAL: "🚨",
AlertLevel.EMERGENCY: "🔴"
}
print(f"{symbols.get(level, '❓')} [{level.value.upper()}] {message}")
def register_alert_callback(self, callback: Callable):
"""Registriert Callback für Alerts"""
self.alert_callbacks.append(callback)
def record_spend(self, actual_cost: float):
"""Aktualisiert Ausgaben nach erfolgreicher Anfrage"""
self.reset_daily_if_needed()
self.daily_spend += actual_cost
self.monthly_spend += actual_cost
self.request_count_today += 1
self.request_count_minute += 1
async def protected_request(
self,
model: str,
messages: List[Dict],
max_tokens: int = 1000
) -> Optional[Dict]:
"""
Führt Anfrage mit Budget-Schutz aus
"""
# Schätze Kosten (vereinfacht)
prompt_text = " ".join(m.get("content", "") for m in messages)
estimated_tokens = len(prompt_text) // 4
estimated_cost = (estimated_tokens / 1_000_000) * 2.5 # Annahme
# Budget-Prüfung
can