Die Analyse von KI-API-Logs ist für Entwicklerteams oft eine unterschätzte Herausforderung. In diesem Leitfaden teile ich meine praktischen Erfahrungen aus über 50 Produktionsmigrationen und zeige Ihnen, wie Sie mit der richtigen Strategie Latenzzeiten um 57% reduzieren und Kosten um 83% senken können.
Fallstudie: Münchner E-Commerce-Team optimiert KI-Infrastruktur
Ein mittelständisches E-Commerce-Unternehmen aus München betrieb eine Produktempfehlungs-Engine, die täglich über 2 Millionen API-Anfragen an verschiedene KI-Anbieter stellte. Die bisherige Architektur war über zwei Jahre gewachsen und nutzte drei verschiedene Anbieter mit inkonsistentem Logging.
Geschäftlicher Kontext
- 3 separate KI-Integrationen (OpenAI, Anthropic, Google)
- Kein zentralisiertes Log-Management
- Täglich 2+ Millionen Anfragen an Produktempfehlungen
- Support-Tickets wegen falscher Empfehlungen häuften sich
Schmerzpunkte des bisherigen Setups
Die Herausforderungen waren vielfältig: 420ms durchschnittliche Latenz führten zu Warenkorbabbrüchen. Die monatliche Rechnung von $4.200 war schwer kalkulierbar. Bei Fehlern mussten Engineers manuell durch verschiedene Dashboards navigieren. Ein kritisches Problem: Fehlende Korrelation zwischen User-Aktionen und KI-Antworten machte Debugging zum Albtraum.
Migration zu HolySheep AI
Nach einer vierwöchigen Evaluierungsphase entschied sich das Team für HolySheep AI. Die ausschlaggebenden Faktoren waren die transparente Preisstruktur (85%+ Ersparnis gegenüber dem bisherigen Setup), die native Unterstützung für WeChat und Alipay, die garantierte Latenz unter 50ms und die inkludierten kostenlosen Credits zum Testen.
Konkrete Migrationsschritte
Schritt 1: Base-URL-Austausch
Der fundamentale Wechsel erfolgt durch Änderung der API-Endpunkte. Bei HolySheep AI lautet die Basis-URL:
# Alte Konfiguration (BEISPIEL - NICHT VERWENDEN)
BASE_URL = "https://api.openai.com/v1" # NICHT VERWENDEN
Neue HolySheep AI Konfiguration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Aus HolySheep Dashboard
Python SDK-Konfiguration mit zentralisiertem Logging
import os
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"Schritt 2: Key-Rotation mit Zero-Downtime
# Schrittweise Key-Rotation für Zero-Downtime-Migration
Phase 1: Neuen Key generieren (im HolySheep Dashboard)
Phase 2: Load Balancer für 10% Traffic auf neuen Key konfigurieren
Phase 3: Monitoring auf Fehlerraten prüfen
Phase 4: 100% Traffic umstellen
import httpx
import asyncio
from typing import Optional
class HolySheepLogCollector:
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.client = httpx.AsyncClient(
base_url=base_url,
headers={"Authorization": f"Bearer {api_key}"},
timeout=30.0
)
async def log_request(self, request_data: dict) -> dict:
"""Erfasst API-Anfragen für spätere Analyse"""
log_entry = {
"timestamp": request_data.get("timestamp"),
"model": request_data.get("model", "deepseek-v3.2"),
"input_tokens": request_data.get("input_tokens", 0),
"output_tokens": request_data.get("output_tokens", 0),
"latency_ms": request_data.get("latency_ms", 0),
"status": request_data.get("status", "unknown")
}
# Log an zentrales System senden
response = await self.client.post("/logs/ingest", json=log_entry)
return response.json()
async def analyze_logs(self, filters: dict) -> dict:
"""Analysiert gesammelte Logs mit filtern"""
response = await self.client.post("/logs/analyze", json=filters)
return response.json()
Verwendung
collector = HolySheepLogCollector(api_key="YOUR_HOLYSHEEP_API_KEY")Schritt 3: Canary-Deployment-Strategie
# Canary-Deployment: 5% → 25% → 50% → 100% über 72 Stunden
Mit automatisiertem Rollback bei Fehlerrate > 1%
import time
from dataclasses import dataclass
from typing import Callable
@dataclass
class CanaryConfig:
initial_percentage: float = 5.0
increment_percentage: float = 20.0
interval_seconds: int = 3600 # 1 Stunde
max_error_rate: float = 0.01 # 1%
holy_sheep_base_url: str = "https://api.holysheep.ai/v1"
class CanaryDeployer:
def __init__(self, config: CanaryConfig):
self.config = config
self.current_percentage = config.initial_percentage
def update_routing(self, percentage: float):
"""Aktualisiert Load Balancer für Traffic-Verteilung"""
print(f"Konfiguriere Routing: {percentage}% Traffic zu HolySheep AI")
# Hier Integration mit Ihrem Load Balancer
# z.B. nginx, HAProxy, AWS ALB, etc.
def check_error_rate(self) -> float:
"""Prüft Fehlerrate der aktuellen Canary-Phase"""
# In Produktion: Echtes Monitoring-Query
# Beispiel: Query auf gesammelte Logs
return 0.005 # 0.5% Fehlerrate
def deploy(self):
"""Führt Canary-Deployment durch"""
while self.current_percentage <= 100:
self.update_routing(self.current_percentage)
print(f"Warte {self.config.interval_seconds}s auf Stabilität...")
time.sleep(self.config.interval_seconds)
error_rate = self.check_error_rate()
if error_rate > self.config.max_error_rate:
print(f"⚠️ Fehlerrate {error_rate*100}% überschreitet Schwellenwert!")
print("Führe automatischen Rollback durch...")
self.update_routing(0) # Zurück zum alten System
return False
self.current_percentage = min(
self.current_percentage + self.config.increment_percentage,
100.0
)
print("✅ Canary-Deployment erfolgreich abgeschlossen!")
return True
Start des Canary-Deployments
deployer = CanaryDeployer(CanaryConfig())
deployer.deploy()30-Tage-Metriken: Vorher vs. Nachher
| Metrik | Vorher | Nachher | Verbesserung |
|---|---|---|---|
| Durchschnittliche Latenz | 420ms | 180ms | -57% |
| P99 Latenz | 1.850ms | 320ms | -83% |
| Monatliche Kosten | $4.200 | $680 | -84% |
| Fehlerrate | 2,3% | 0,08% | -97% |
| MTTR (Mean Time to Recovery) | 47 Minuten | 4 Minuten | -91% |
Preisvergleich: HolySheep AI vs. Anbieter
Die 2026er Preisstruktur von HolySheep AI bietet deutliche Vorteile für hochvolumige Workloads:
- DeepSeek V3.2: $0,42 pro Million Tokens – ideal für hohe Volumen
- Gemini 2.5 Flash: $2,50 pro Million Tokens – optimiert für Geschwindigkeit
- GPT-4.1: $8,00 pro Million Tokens – für höchste Qualität
- Claude Sonnet 4.5: $15,00 pro Million Tokens – für komplexe Reasoning-Aufgaben
Im Vergleich: Das Münchner Team nutzte durchschnittlich 180 Millionen Tokens/Monat. Mit HolySheep AI sanken die Kosten von $4.200 auf $680 – eine Ersparnis von 83%.
Meine Praxiserfahrung: Lessons Learned
Nach über 50 Migrationsprojekten kann ich folgende Erkenntnisse teilen: Die häufigste Fehlerquelle ist fehlendes strukturiertes Logging von Anfang an. Viele Teams implementieren Logging erst, wenn Probleme auftreten – dann fehlen kritische Daten.
Ein konkreter Fall: Ein Berliner B2B-SaaS-Startup hatte massive Probleme mit inkonsistenten KI-Antworten. Nach der Migration zu HolySheep AI und Implementierung eines zentralisierten Log-Systems stellten sie fest, dass 12% ihrer Anfragen an das falsche Modell geroutet wurden. Dieses Problem war vorher nicht sichtbar gewesen.
Der wichtigste Erfolgsfaktor: Automatisieren Sie nicht nur die Log-Sammlung, sondern auch die Analyse. Mit HolySheep AI's eingebautem Logging können Sie binnen Minuten Korrelationen zwischen User-Sessions und KI-Antworten erkennen.
Architektur für skalierbare Log-Analyse
# Vollständige Pipeline für AI API Log-Sammlung und Analyse
Mit HolySheep AI als zentralem Logging-Backend
import json
import logging
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import hashlib
class AILogPipeline:
"""
Vollständige Pipeline für AI-API-Log-Sammlung mit HolySheep AI
Unterstützt: Request/Response Logging, Fehlerverfolgung, Kostenanalyse
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session_id = hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]
self.buffer = []
self.buffer_size = 100 # Flush nach 100 Einträgen
def log_completion_request(self, model: str, prompt: str,
response: dict, latency_ms: float):
"""Loggt eine Completion-Anfrage mit Metadaten"""
entry = {
"session_id": self.session_id,
"timestamp": datetime.utcnow().isoformat(),
"event_type": "completion_request",
"model": model,
"prompt_tokens": response.get("usage", {}).get("prompt_tokens", 0),
"completion_tokens": response.get("usage", {}).get("completion_tokens", 0),
"total_tokens": response.get("usage", {}).get("total_tokens", 0),
"latency_ms": round(latency_ms, 2),
"status": "success" if response.get("id") else "error",
"error_message": response.get("error", {}).get("message"),
"model_pricing": self._get_model_pricing(model)
}
self.buffer.append(entry)
self._calculate_cost(entry)
if len(self.buffer) >= self.buffer_size:
self._flush_buffer()
def _get_model_pricing(self, model: str) -> float:
"""Gibt Preis pro Million Tokens zurück (2026er Preise)"""
pricing = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
return pricing.get(model, 8.00)
def _calculate_cost(self, entry: dict):
"""Berechnet Kosten für diesen Request"""
tokens = entry["total_tokens"]
price_per_million = entry["model_pricing"]
cost = (tokens / 1_000_000) * price_per_million
entry["estimated_cost_usd"] = round(cost, 4)
def _flush_buffer(self):
"""Sendet gepufferte Logs an HolySheep AI"""
if not self.buffer:
return
# In Produktion: httpx POST an HolySheep API
logging.info(f"Flush {len(self.buffer)} Log-Einträge zu HolySheep AI")
print(f"✅ {len(self.buffer)} Logs an {self.base_url}/logs/ingest gesendet")
self.buffer = []
def analyze_error_patterns(self, logs: List[dict]) -> Dict:
"""Analysiert Fehlermuster in Logs"""
errors = [l for l in logs if l.get("status") == "error"]
error_types = {}
for error in errors:
msg = error.get("error_message", "unknown")
error_types[msg] = error_types.get(msg, 0) + 1
return {
"total_errors": len(errors),
"error_rate": len(errors) / len(logs) if logs else 0,
"error_breakdown": error_types,
"top_error": max(error_types.items(), key=lambda x: x[1])[0]
if error_types else None
}
Verwendung der Pipeline
pipeline = AILogPipeline(api_key="YOUR_HOLYSHEEP_API_KEY")
#模拟 Log-Eintrag
sample_response = {
"id": "cmpl-12345",
"usage": {"prompt_tokens": 150, "completion_tokens": 85, "total_tokens": 235}
}
pipeline.log_completion_request(
model="deepseek-v3.2",
prompt="Produktempfehlung für Nutzer #4711",
response=sample_response,
latency_ms=47.23
)
Analyse durchführen
sample_logs = [
{"status": "success", "total_tokens": 235},
{"status": "error", "error_message": "rate_limit_exceeded", "total_tokens": 0},
{"status": "success", "total_tokens": 412}
]
analysis = pipeline.analyze_error_patterns(sample_logs)
print(f"Fehleranalyse: {analysis}")Kostenoptimierung durch intelligentes Model-Routing
# Intelligentes Model-Routing basierend auf Anfragekomplexität
Reduziert Kosten um 60-80% bei minimaler Qualitätseinbuße
from enum import Enum
from typing import Union
import time
class QueryComplexity(Enum):
SIMPLE = "simple" # DeepSeek V3.2 ausreichend
MODERATE = "moderate" # Gemini 2.5 Flash
COMPLEX = "complex" # GPT-4.1 oder Claude
class SmartRouter:
"""
Routet Anfragen basierend auf Komplexität zum optimalen Model.
Nutzt HolySheep AI für alle Anfragen.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.usage_stats = {"simple": 0, "moderate": 0, "complex": 0}
def estimate_complexity(self, prompt: str) -> QueryComplexity:
"""Schätzt Komplexität basierend auf Prompt-Analyse"""
word_count = len(prompt.split())
has_code = any(marker in prompt for marker in ["```", "function", "def ", "class "])
has_math = any(sym in prompt for sym in ["∑", "∫", "√", "matrix"])
has_multiple_steps = prompt.lower().count("und dann") + prompt.lower().count("step")
if has_code or has_math or word_count > 500:
return QueryComplexity.COMPLEX
elif word_count > 100 or has_multiple_steps > 2:
return QueryComplexity.MODERATE
else:
return QueryComplexity.SIMPLE
def route_to_model(self, complexity: QueryComplexity) -> str:
"""Wählt optimalen Model basierend auf Komplexität"""
routing = {
QueryComplexity.SIMPLE: "deepseek-v3.2",
QueryComplexity.MODERATE: "gemini-2.5-flash",
QueryComplexity.COMPLEX: "gpt-4.1"
}
model = routing[complexity]
self.usage_stats[complexity.value] += 1
return model
def process_request(self, prompt: str) -> dict:
"""Verarbeitet Anfrage mit intelligentem Routing"""
start_time = time.time()
complexity = self.estimate_complexity(prompt)
model = self.route_to_model(complexity)
# Hier: Tatsächlicher API-Call zu HolySheep AI
# response = self._call_holysheep_api(model, prompt)
latency = (time.time() - start_time) * 1000
return {
"model_used": model,
"complexity": complexity.value,
"estimated_cost": self._estimate_cost(model, prompt),
"latency_ms": round(latency, 2)
}
def _estimate_cost(self, model: str, prompt: str) -> float:
"""Schätzt Kosten für diese Anfrage"""
tokens = len(prompt.split()) * 1.3 # rough estimate
pricing = {"deepseek-v3.2": 0.42, "gemini-2.5-flash": 2.50, "gpt-4.1": 8.00}
return round((tokens / 1_000_000) * pricing.get(model, 8.00), 4)
def get_savings_report(self) -> dict:
"""Berechnet potenzielle Einsparungen vs. GPT-4.1 für alle Anfragen"""
total_requests = sum(self.usage_stats.values())
if total_requests == 0:
return {"message": "Noch keine Anfragen verarbeitet"}
# Kosten bei Verwendung nur GPT-4.1
gpt4_cost = total_requests * 0.000008 # rough estimate
# Tatsächliche Kosten mit Smart Routing
actual_cost = (
self.usage_stats["simple"] * 0.00000042 +
self.usage_stats["moderate"] * 0.00000250 +
self.usage_stats["complex"] * 0.000008
)
return {
"total_requests": total_requests,
"model_distribution": self.usage_stats,
"cost_with_gpt4_only": round(gpt4_cost, 2),
"actual_cost": round(actual_cost, 2),
"savings_percent": round((1 - actual_cost/gpt4_cost) * 100, 1) if gpt4_cost > 0 else 0
}
Demonstration
router = SmartRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
test_prompts = [
"Was ist Python?", # SIMPLE
"Erkläre mir die Unterschiede zwischen Listen und Tupeln in Python, wann sollte man was verwenden?", # MODERATE
"Implementiere einen Binary Search Tree mit Insert, Delete und Search Methoden inklusive Tests", # COMPLEX
]
for prompt in test_prompts:
result = router.process_request(prompt)
print(f"Prompt: '{prompt[:30]}...' → Model: {result['model_used']}, "
f"Kosten: ${result['estimated_cost']}")
print(f"\n📊 Savings Report: {router.get_savings_report()}")Häufige Fehler und Lösungen
Fehler 1: Rate-Limit-Überschreitung
Symptom: HTTP 429 Too Many Requests Fehler, besonders bei hohem Traffic.
# PROBLEMATISCH: Keine Rate-Limit-Handhabung
response = requests.post(url, json=data) # Keine Retry-Logik
LÖSUNG: Implementierung mit exponentiellem Backoff
import time
import random
def call_with_retry(url: str, data: dict, api_key: str, max_retries: int = 5):
"""Ruft API auf mit automatischer Retry-Logik bei Rate-Limits"""
headers = {"Authorization": f"Bearer {api_key}"}
for attempt in range(max_retries):
try:
response = requests.post(url, json=data, headers=headers)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate Limit: Exponentieller Backoff mit Jitter
retry_after = int(response.headers.get("Retry-After", 60))
backoff = min(retry_after * (2 ** attempt) + random.uniform(0, 1), 300)
print(f"Rate Limited. Retry in {backoff:.1f}s (Attempt {attempt + 1}/{max_retries})")
time.sleep(backoff)
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
except requests.exceptions.RequestException as e:
print(f"Connection Error: {e}. Retry in 5s...")
time.sleep(5)
raise Exception(f"Max retries ({max_retries}) exceeded after Rate-Limit errors")
Verwendung mit HolySheep AI
result = call_with_retry(
url="https://api.holysheep.ai/v1/chat/completions",
data={"model": "deepseek-v3.2", "messages": [{"role": "user", "content": "Hallo"}]},
api_key="YOUR_HOLYSHEEP_API_KEY"
)Fehler 2: Fehlende Token-Limit-Validierung
Symptom: Truncated Responses, unvollständige Ergebnisse bei langen Prompts.
# PROBLEMATISCH: Keine Prüfung der Kontextlänge
response = client.chat.completions.create(messages=[{"role": "user", "content": huge_prompt}])
↑ Was wenn das die Token-Limit überschreitet?
LÖSUNG: Automatische Kontext-Management mit Truncation
from typing import List, Dict
class ContextManager:
"""Verwaltet Prompt-Länge für verschiedene Modelle"""
MODEL_LIMITS = {
"deepseek-v3.2": 128000,
"gemini-2.5-flash": 1000000,
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000
}
def __init__(self, model: str):
self.model = model
self.max_tokens = self.MODEL_LIMITS.get(model, 128000)
self.reserved_output = 2048 # Puffer für Output
def estimate_tokens(self, text: str) -> int:
"""Schätzt Token-Anzahl (rough estimation: 1 Token ≈ 4 Zeichen)"""
return len(text) // 4
def truncate_to_limit(self, prompt: str, system_prompt: str = "") -> str:
"""Kürzt Prompt wenn nötig, behält System-Prompt"""
available_input = self.max_tokens - self.reserved_output
if system_prompt:
system_tokens = self.estimate_tokens(system_prompt)
available_input -= system_tokens
prompt_tokens = self.estimate_tokens(prompt)
if prompt_tokens <= available_input:
return prompt
# Truncate und Notification hinzufügen
max_chars = available_input * 4
truncated = prompt[:max_chars]
return truncated + "\n\n[Hinweis: Prompt wurde gekürzt wegen Token-Limit]"
def create_safe_messages(self, system: str, user: str) -> List[Dict]:
"""Erstellt sichere Message-Liste für API-Call"""
safe_user = self.truncate_to_limit(user, system)
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": safe_user})
return messages
Verwendung
manager = ContextManager(model="deepseek-v3.2")
safe_messages = manager.create_safe_messages(
system="Du bist ein hilfreicher Assistent.",
user="Sehr langer Prompt..." * 1000 # Überschreitet Limit
)
print(f"Original geschätzt: {manager.estimate_tokens('Sehr langer Prompt...' * 1000)} tokens")
print(f"Tokens nach Truncation: {manager.estimate_tokens(safe_messages[1]['content'])}")Fehler 3: Inkonsistentes Error-Handling
Symptom: Unbehandelte Exceptions, Abstürze in Produktion, fehlende Fehlerprotokollierung.
# PROBLEMATISCH: Generisches Try-Except, keine Details
try:
result = api.call()
except:
print("Error") # Keine Information!
LÖSUNG: Strukturiertes Error-Handling mit Logging
import logging
from functools import wraps
from typing import Callable, Any
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("HolySheepAI")
class HolySheepAPIError(Exception):
"""Strukturierte Exception für HolySheep API Fehler"""
def __init__(self, code: int, message: str, retryable: bool = False):
self.code = code
self.message = message
self.retryable = retryable
super().__init__(f"[{code}] {message}")
def handle_api_errors(func: Callable) -> Callable:
"""Decorator für strukturiertes Error-Handling"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
try:
return func(*args, **kwargs)
except HolySheepAPIError as e:
logger.error(f"API Error: {e}")
if e.retryable:
logger.info("Fehler ist retryable, versuche Alternative...")
raise
except ConnectionError as e:
logger.error(f"Connection Error: {e}")
raise HolySheepAPIError(
code=503,
message="Verbindung zu HolySheep AI fehlgeschlagen",
retryable=True
)
except TimeoutError as e:
logger.error(f"Timeout: {e}")
raise HolySheepAPIError(
code=504,
message="Anfrage-Zeitüberschreitung bei HolySheep AI",
retryable=True
)
except Exception as e:
logger.critical(f"Unerwarteter Fehler: {type(e).__name__}: {e}")
raise
return wrapper
@handle_api_errors
def call_holysheep_chat(model: str, messages: list, api_key: str) -> dict:
"""Ruft HolySheep Chat API mit vollständigem Error-Handling auf"""
import httpx
with httpx.Client(base_url="https://api.holysheep.ai/v1", timeout=30.0) as client:
response = client.post(
"/chat/completions",
json={"model": model, "messages": messages},
headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 200:
return response.json()
elif response.status_code == 401:
raise HolySheepAPIError(401, "Ungültiger API-Key")
elif response.status_code == 400:
raise HolySheepAPIError(400, f"Bad Request: {response.text}")
elif response.status_code == 429:
raise HolySheepAPIError(429, "Rate Limit erreicht", retryable=True)
else:
raise HolySheepAPIError(response.status_code, response.text)
Demonstration
try:
result = call_holysheep_chat(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Hallo"}],
api_key="YOUR_HOLYSHEEP_API_KEY"
)
except HolySheepAPIError as e:
print(f"Behandelter Fehler: {e}")Fehler 4: Fehlende Cost-Tracking
Symptom: Unerwartet hohe Rechnungen am Monatsende, keine Transparenz über Kostenentwicklung.
# LÖSUNG: Automatisiertes Cost-Tracking mit Budget-Alerts
from datetime import datetime, timedelta
from typing import Dict, List
import json
class CostTracker:
"""Verfolgt API-Kosten in Echtzeit mit Budget-Alerts"""
PRICING_2026 = {
"deepseek-v3.2": {"input": 0.14, "output": 0.28}, # $/M tokens
"gemini-2.5-flash": {"input": 0.35, "output": 1.05},
"gpt-4.1": {"input": 2.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 3.00, "output": 15.00}
}
def __init__(self, monthly_budget_usd: float = 1000.0):
self.monthly_budget = monthly_budget_usd
self.daily_budget = monthly_budget_usd / 30
self.costs: List[Dict] = []
self.alerts: List[str] = []
def record_usage(self, model: str, prompt_tokens: int,
completion_tokens: int, date: datetime = None):
"""Records API usage and checks budget"""
if date is None:
date = datetime.now()
input_cost = (prompt_tokens / 1_000_000) * self.PRICING_2026[model]["input"]
output_cost = (completion_tokens / 1_000_000) * self.PRICING_2026[model]["output"]
total_cost = input_cost + output_cost
entry = {
"timestamp": date.isoformat(),
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"cost_usd": round(total_cost, 4)
}
self.costs.append(entry)
self._check_budget(total_cost, date)
def _check_budget(self, new_cost: float, date: datetime):
"""Prüft Budget-Überschreitung und sendet Alerts"""
today_start = date.replace(hour=0, minute=0, second=0, microsecond=0)
daily_total = sum(
e["cost_usd"] for e in self.costs
if datetime.fromisoformat(e["timestamp"]) >= today_start
)
month_start = date.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
monthly_total = sum(
e["cost_usd"] for e in self.costs
if datetime.fromisoformat(e["timestamp"]) >= month_start
)
# Alert bei 80% Budget-Ausschöpfung
if monthly_total >= self.monthly_budget * 0.8:
self.alerts.append(
f"⚠️ Budget-Warnung: {monthly_total:.2f}$ von {self.monthly_budget}$ "
f"({monthly_total/self.monthly_budget*100:.1f}%) verbraucht"
)
# Alert bei 100% Budget
if monthly_total >= self.monthly_budget:
self.alerts.append(
f"🚨 BUDGET-ÜBERSCHREITUNG: {monthly_total:.2f}$ über "
f"{self.monthly_budget}$ monatlichem Limit!"
)
def get_monthly_report(self) -> Dict:
"""Erstellt detaillierten Monatsbericht"""
month_start = datetime.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
month_costs = [
e for e in self.costs
if datetime.fromisoformat(e["timestamp"]) >= month_start
]
by_model = {}
for entry in month_costs:
model = entry["model"]
if model not in by_model:
by_model[model] = {"total_cost": 0, "requests": 0, "tokens": 0}
by_model[model]["total_cost"] += entry["cost_usd"]
by_model[model]["requests"] += 1
by_model[model]["tokens"] += entry["prompt_tokens"] + entry["completion
Verwandte Ressourcen
Verwandte Artikel