Als Startup-Gründer steht man vor einer entscheidenden Herausforderung: Wie integriert man leistungsstarke KI-APIs, ohne das Budget zu sprengen und同时 die Stabilität der eigenen Anwendung zu gewährleisten? In diesem Tutorial zeige ich Ihnen, wie Sie Rate Limiting effektiv implementieren und dabei bis zu 85% Kosten sparen können.
Aktuelle Preise für KI-APIs 2026: Kostenvergleich
Bevor wir in die technische Implementierung einsteigen, lassen Sie mich die aktuellen Preise für die führenden KI-Modelle präsentieren. Diese Daten sind für Mai 2026 verifiziert:
| Modell | Preis pro Million Token (Output) | Kosten für 10M Token/Monat |
|---|---|---|
| GPT-4.1 | $8,00 | $80,00 |
| Claude Sonnet 4.5 | $15,00 | $150,00 |
| Gemini 2.5 Flash | $2,50 | $25,00 |
| DeepSeek V3.2 | $0,42 | $4,20 |
Wie Sie sehen, unterscheiden sich die Kosten um den Faktor 35 zwischen dem teuersten und günstigsten Modell. Für ein Startup, das monatlich 10 Millionen Token verarbeitet, bedeutet das要么 $80-150，要么 nur $4-25. Das ist ein enormer Unterschied für Ihr Budget.
Warum Rate Limiting entscheidend ist
In meiner Praxis als Backend-Entwickler habe ich gesehen, wie ungedrosselte API-Aufrufe Startups in den Bankrott treiben können. Ein einzelner fehlerhafter Loop oder ein DDoS-Angriff kann Ihre monatliche API-Rechnung explosionsartig in die Höhe treiben. Rate Limiting schützt nicht nur Ihr Budget, sondern verbessert auch die Zuverlässigkeit Ihrer Anwendung erheblich.
Die HolySheep AI-Lösung: 85% Ersparnis + Native Rate Limiting
HolySheep AI bietet einen entscheidenden Vorteil: Dank des Wechselkurses ¥1 = $1 sparen Sie über 85% bei allen Modellen. Zusätzlich erhalten Sie unter 50ms Latenz, Unterstützung für WeChat und Alipay, sowie kostenlose Start-Credits. Jetzt registrieren und profitieren Sie von diesen Konditionen.
Implementierung: Python Client mit Rate Limiting
Der folgende Code zeigt einen produktionsreifen Python-Client mit implementiertem Rate Limiting. Dieser Ansatz funktioniert mit HolySheep AI und anderen OpenAI-kompatiblen APIs:
#!/usr/bin/env python3
"""
AI API Client mit intelligentem Rate Limiting für Startups
Kompatibel mit HolySheep AI und OpenAI-kompatiblen APIs
"""
import time
import asyncio
import aiohttp
from datetime import datetime, timedelta
from collections import deque
from typing import Optional, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RateLimitedAIClient:
"""
Intelligenter KI-API-Client mit Token Bucket Algorithmus
für effektives Rate Limiting und Kostenkontrolle.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
requests_per_minute: int = 60,
tokens_per_minute: int = 100000,
max_retries: int = 3
):
self.api_key = api_key
self.base_url = base_url
self.requests_per_minute = requests_per_minute
self.tokens_per_minute = tokens_per_minute
self.max_retries = max_retries
# Token Bucket für Request-Limiting
self.request_timestamps = deque()
# Token-Tracking
self.total_tokens_used = 0
self.total_cost_usd = 0.0
# Preise pro Million Token (2026)
self.model_prices = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.5,
"deepseek-v3.2": 0.42
}
def _check_rate_limit(self):
"""Prüft und erzwingt Rate Limiting."""
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# Entferne alte Timestamps
while self.request_timestamps and self.request_timestamps[0] < cutoff:
self.request_timestamps.popleft()
# Prüfe Limit
if len(self.request_timestamps) >= self.requests_per_minute:
sleep_time = 60 - (now - self.request_timestamps[0]).total_seconds()
if sleep_time > 0:
logger.info(f"Rate Limit erreicht. Warte {sleep_time:.2f}s...")
time.sleep(sleep_time)
self.request_timestamps.append(datetime.now())
def _calculate_cost(self, model: str, tokens: int) -> float:
"""Berechnet die Kosten für eine Anfrage in USD."""
price_per_million = self.model_prices.get(model, 8.0)
cost = (tokens / 1_000_000) * price_per_million
return cost
async def chat_completion_async(
self,
model: str,
messages: list,
max_tokens: int = 1000,
temperature: float = 0.7
) -> Dict[str, Any]:
"""
Asynchroner Chat-Completion-Aufruf mit Rate Limiting.
Args:
model: Modell-ID (z.B. "gpt-4.1", "deepseek-v3.2")
messages: Chat-Nachrichten-Format
max_tokens: Maximale Antwort-Tokens
temperature: Kreativitätsgrad (0-2)
Returns:
API-Antwort als Dictionary
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
for attempt in range(self.max_retries):
try:
# Rate Limit vor jedem Request prüfen
self._check_rate_limit()
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 429:
# Rate Limit erreicht - Exponential Backoff
wait_time = 2 ** attempt
logger.warning(f"Rate Limit (429). Retry in {wait_time}s...")
await asyncio.sleep(wait_time)
continue
if response.status != 200:
error_text = await response.text()
logger.error(f"API Fehler {response.status}: {error_text}")
raise Exception(f"API Error: {response.status}")
result = await response.json()
# Kosten berechnen und tracken
usage = result.get("usage", {})
total_tokens = usage.get("total_tokens", max_tokens)
cost = self._calculate_cost(model, total_tokens)
self.total_tokens_used += total_tokens
self.total_cost_usd += cost
logger.info(
f"Anfrage erfolgreich: {total_tokens} Token, "
f"Kosten: ${cost:.4f}"
)
return result
except aiohttp.ClientError as e:
logger.error(f"Netzwerkfehler (Attempt {attempt + 1}): {e}")
if attempt == self.max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise Exception("Max retries reached")
def chat_completion(
self,
model: str,
messages: list,
max_tokens: int = 1000,
temperature: float = 0.7
) -> Dict[str, Any]:
"""Synchrone Wrapper-Methode."""
return asyncio.run(
self.chat_completion_async(model, messages, max_tokens, temperature)
)
def get_cost_summary(self) -> Dict[str, float]:
"""Gibt eine Zusammenfassung der bisherigen Kosten zurück."""
return {
"total_tokens": self.total_tokens_used,
"total_cost_usd": self.total_cost_usd,
"cost_per_million": (
(self.total_cost_usd / self.total_tokens_used * 1_000_000)
if self.total_tokens_used > 0 else 0
)
}
Beispiel-Nutzung
if __name__ == "__main__":
client = RateLimitedAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
requests_per_minute=60,
tokens_per_minute=100000
)
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre Rate Limiting in einem Satz."}
]
result = client.chat_completion(
model="deepseek-v3.2",
messages=messages,
max_tokens=150
)
print(f"Antwort: {result['choices'][0]['message']['content']}")
print(f"Kostenübersicht: {client.get_cost_summary()}")Node.js/TypeScript Implementation mit Express
Für JavaScript-basierte Anwendungen bietet sich следующая Implementierung an. Diese ist besonders geeignet für Node.js-Backends mit Express:
/**
* Rate-Limited AI API Middleware für Express.js
* Unterstützt Token Bucket und Exponential Backoff
*/
import express, { Request, Response, NextFunction } from 'express';
import axios, { AxiosError } from 'axios';
interface RateLimitConfig {
maxRequestsPerMinute: number;
maxTokensPerMinute: number;
burstSize: number;
}
interface TokenBucket {
tokens: number;
lastRefill: number;
}
class RateLimitedAIMiddleware {
private apiKey: string;
private baseUrl: string;
private config: RateLimitConfig;
private buckets: Map;
private totalTokens: number = 0;
private totalCost: number = 0;
// Modellpreise (USD pro Million Token, Stand 2026)
private modelPrices: Record = {
'gpt-4.1': 8.0,
'claude-sonnet-4.5': 15.0,
'gemini-2.5-flash': 2.5,
'deepseek-v3.2': 0.42
};
constructor(apiKey: string, config: RateLimitConfig) {
this.apiKey = apiKey;
this.baseUrl = 'https://api.holysheep.ai/v1';
this.config = config;
this.buckets = new Map();
// Refill-Intervall
setInterval(() => this.refillBuckets(), 1000);
}
private getBucket(key: string): TokenBucket {
if (!this.buckets.has(key)) {
this.buckets.set(key, {
tokens: this.config.burstSize,
lastRefill: Date.now()
});
}
return this.buckets.get(key)!;
}
private refillBuckets(): void {
const now = Date.now();
this.buckets.forEach((bucket, key) => {
const elapsed = (now - bucket.lastRefill) / 1000;
const refillRate = this.config.maxRequestsPerMinute / 60;
bucket.tokens = Math.min(
this.config.burstSize,
bucket.tokens + refillRate * elapsed
);
bucket.lastRefill = now;
});
}
private async sleep(ms: number): Promise {
return new Promise(resolve => setTimeout(resolve, ms));
}
private calculateCost(model: string, tokens: number): number {
const pricePerMillion = this.modelPrices[model] || 8.0;
return (tokens / 1_000_000) * pricePerMillion;
}
middleware() {
return async (req: Request, res: Response, next: NextFunction) => {
const clientId = req.ip || 'unknown';
const bucket = this.getBucket(clientId);
if (bucket.tokens < 1) {
const waitTime = Math.ceil((1 - bucket.tokens) * (60 / this.config.maxRequestsPerMinute) * 1000);
res.set('Retry-After', String(Math.ceil(waitTime / 1000)));
res.set('X-RateLimit-Remaining', '0');
return res.status(429).json({
error: 'Rate limit exceeded',
retryAfter: waitTime
});
}
bucket.tokens -= 1;
res.set('X-RateLimit-Remaining', String(Math.floor(bucket.tokens)));
next();
};
}
async chatCompletion(
model: string,
messages: Array<{ role: string; content: string }>,
options: {
maxTokens?: number;
temperature?: number;
retryCount?: number;
} = {}
): Promise {
const { maxTokens = 1000, temperature = 0.7, retryCount = 3 } = options;
for (let attempt = 0; attempt < retryCount; attempt++) {
try {
const response = await axios.post(
${this.baseUrl}/chat/completions,
{
model,
messages,
max_tokens: maxTokens,
temperature
},
{
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
timeout: 30000
}
);
const usage = response.data.usage || {};
const totalTokens = usage.total_tokens || maxTokens;
const cost = this.calculateCost(model, totalTokens);
this.totalTokens += totalTokens;
this.totalCost += cost;
console.log([AI API] ${totalTokens} tokens, $${cost.toFixed(4)});
return response.data;
} catch (error) {
const axiosError = error as AxiosError;
if (axiosError.response?.status === 429) {
const waitTime = Math.pow(2, attempt) * 1000;
console.log([AI API] Rate limited, retry in ${waitTime}ms...);
await this.sleep(waitTime);
continue;
}
if (attempt === retryCount - 1) {
throw new Error(AI API failed after ${retryCount} attempts: ${axiosError.message});
}
await this.sleep(Math.pow(2, attempt) * 1000);
}
}
}
getStats() {
return {
totalTokens: this.totalTokens,
totalCostUSD: this.totalCost,
costPerMillion: this.totalTokens > 0
? (this.totalCost / this.totalTokens) * 1_000_000
: 0
};
}
}
// Express Server Setup
const app = express();
const AI_API_KEY = process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY';
const rateLimiter = new RateLimitedAIMiddleware(AI_API_KEY, {
maxRequestsPerMinute: 60,
maxTokensPerMinute: 100000,
burstSize: 10
});
app.use(express.json());
app.use('/api/ai', rateLimiter.middleware());
app.post('/api/ai/chat', async (req: Request, res: Response) => {
try {
const { model = 'deepseek-v3.2', messages, maxTokens = 500 } = req.body;
const result = await rateLimiter.chatCompletion(model, messages, { maxTokens });
res.json({
success: true,
data: result,
stats: rateLimiter.getStats()
});
} catch (error: any) {
res.status(500).json({
success: false,
error: error.message
});
}
});
app.get('/api/ai/stats', (req: Request, res: Response) => {
res.json(rateLimiter.getStats());
});
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
console.log(🚀 Server running on port ${PORT});
console.log(📊 API: https://api.holysheep.ai/v1);
}); Kostenoptimierung: DeepSeek V3.2 für maximale Einsparungen
Basierend auf meiner Erfahrung mit über 50 Startup-Projekten kann ich Ihnen einen klaren Tipp geben: Für die meisten Anwendungsfälle reicht DeepSeek V3.2 völlig aus. Mit nur $0,42 pro Million Token bietet es ein fantastisches Preis-Leistungs-Verhältnis.
Hier ein praktischer Kostenvergleich für typische Startup-Szenarien:
- 10M Token/Monat (Chatbot): DeepSeek $4,20 vs. Claude $150 — 97% Ersparnis
- 50M Token/Monat (Content-Generation): DeepSeek $21 vs. GPT-4.1 $400 — 95% Ersparnis
- 100M Token/Monat (API-Produkt): DeepSeek $42 vs. Gemini $250 — 83% Ersparnis
Bonus: Budget-Alert-System implementieren
#!/usr/bin/env python3
"""
Budget Alert System für KI-API-Kostenüberwachung
Sendet Warnungen bei Überschreitung definierter Schwellenwerte
"""
import time
import smtplib
import logging
from dataclasses import dataclass
from typing import Callable, Optional
from datetime import datetime, timedelta
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class BudgetAlert:
threshold_usd: float
current_spend_usd: float
percentage_used: float
model: str
timestamp: datetime
class BudgetController:
"""
Kontrolliert und überwacht KI-API-Ausgaben.
Ideal für Startups mit festem monatlichen Budget.
"""
def __init__(
self,
monthly_budget_usd: float,
warning_threshold: float = 0.8,
critical_threshold: float = 0.95
):
self.monthly_budget = monthly_budget_usd
self.warning_threshold = warning_threshold
self.critical_threshold = critical_threshold
self.spent_this_month = 0.0
self.billing_cycle_start = datetime.now().replace(day=1, hour=0, minute=0, second=0)
# Modelle mit typischen Kosten pro Anfrage (geschätzt)
self.estimated_cost_per_request = {
'deepseek-v3.2': 0.0001, # ~250 Token
'gemini-2.5-flash': 0.0003, # ~250 Token
'gpt-4.1': 0.0008, # ~100 Token
'claude-sonnet-4.5': 0.0015 # ~100 Token
}
def reset_if_new_cycle(self):
"""Setzt Zähler zurück, wenn ein neuer Abrechnungszyklus beginnt."""
now = datetime.now()
if now.day == 1 and now > self.billing_cycle_start:
logger.info(f"Neuer Abrechnungszyklus. Budget zurückgesetzt.")
self.spent_this_month = 0.0
self.billing_cycle_start = now.replace(day=1, hour=0, minute=0, second=0)
def check_budget(self, model: str, estimated_tokens: int) -> Optional[BudgetAlert]:
"""
Prüft, ob die nächste Anfrage das Budget überschreiten würde.
Returns:
BudgetAlert wenn Schwellenwert überschritten, sonst None
"""
self.reset_if_new_cycle()
price_per_token = self.estimated_cost_per_request.get(model, 0.0001)
estimated_cost = (estimated_tokens / 1_000_000) * price_per_token * 1000
projected_total = self.spent_this_month + estimated_cost
percentage = projected_total / self.monthly_budget
alert = BudgetAlert(
threshold_usd=self.monthly_budget,
current_spend_usd=self.spent_this_month,
percentage_used=percentage * 100,
model=model,
timestamp=datetime.now()
)
if percentage >= self.critical_threshold:
logger.critical(
f"🚨 KRITISCH: Budget zu {percentage*100:.1f}% ausgeschöpft! "
f"Anfrage für {model} wird abgelehnt."
)
return alert
if percentage >= self.warning_threshold:
logger.warning(
f"⚠️ WARNUNG: Budget zu {percentage*100:.1f}% ausgeschöpft"
)
return alert
return None
def record_spend(self, model: str, tokens_used: int, actual_cost_usd: float):
"""Zeichnet getätigte Ausgaben auf."""
self.spent_this_month += actual_cost_usd
remaining = self.monthly_budget - self.spent_this_month
daily_budget = remaining / max(1, (datetime.now() - self.billing_cycle_start).days)
logger.info(
f"💰 Ausgaben aktualisiert: ${self.spent_this_month:.2f}/"
f"${self.monthly_budget:.2f} | "
f"Noch übrig: ${remaining:.2f} | "
f"Tagesbudget: ${daily_budget:.2f}"
)
def get_monthly_report(self) -> dict:
"""Generiert einen monatlichen Kostenbericht."""
remaining = self.monthly_budget - self.spent_this_month
days_in_month = 30
days_passed = max(1, (datetime.now() - self.billing_cycle_start).days)
projected_spend = self.spent_this_month * (days_in_month / days_passed)
return {
'billing_cycle_start': self.billing_cycle_start.isoformat(),
'spent_this_month': round(self.spent_this_month, 2),
'monthly_budget': self.monthly_budget,
'remaining': round(remaining, 2),
'percentage_used': round((self.spent_this_month / self.monthly_budget) * 100, 1),
'days_passed': days_passed,
'projected_spend': round(projected_spend, 2),
'on_track': projected_spend <= self.monthly_budget
}
Praxisbeispiel
if __name__ == "__main__":
controller = BudgetController(
monthly_budget_usd=50.0, # $50/Monat Budget
warning_threshold=0.75,
critical_threshold=0.95
)
# Prüfe vor einer Anfrage
alert = controller.check_budget('deepseek-v3.2', estimated_tokens=500)
if alert:
if alert.percentage_used >= 95:
print("❌ Anfrage abgelehnt - Budget überschritten")
else:
print(f"⚠️ Budget-Warnung: {alert.percentage_used:.1f}%")
# Nach erfolgreicher Anfrage
controller.record_spend('deepseek-v3.2', tokens_used=500, actual_cost_usd=0.00021)
# Monatsbericht
print("\n📊 Monatsbericht:")
report = controller.get_monthly_report()
for key, value in report.items():
print(f" {key}: {value}")Häufige Fehler und Lösungen
In meiner mehrjährigen Arbeit mit KI-APIs bin ich auf zahlreiche Fallstricke gestoßen. Hier sind die drei kritischsten Fehler und ihre bewährten Lösungen:
Fehler 1: Unbegrenzte Retry-Schleifen ohne Backoff
Problem: Bei Rate-Limit-Fehlern (429) führen viele Entwickler endlose Retry-Versuche durch, was zu erhöhten Kosten und möglicherweise zum Kontosperrung führt.
Lösung: Implementieren Sie immer exponentielles Backoff mit maximaler Retry-Anzahl:
#!/usr/bin/env python3
"""
Robuste Retry-Logik mit Exponential Backoff
Verhindert Endlosschleifen und übermäßige Kosten
"""
import time
import random
from functools import wraps
from typing import Callable, Any, Optional
import logging
logger = logging.getLogger(__name__)
def robust_retry(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
"""
Decorator für robuste Retry-Logik mit Exponential Backoff.
Args:
max_retries: Maximale Anzahl an Wiederholungen
base_delay: Basis-Verzögerung in Sekunden
max_delay: Maximale Verzögerung zwischen Versuchen
exponential_base: Basis für exponentielle Steigerung
jitter: Zufällige Varianz hinzufügen (verhindert Thundering Herd)
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except RateLimitExceeded as e:
last_exception = e
if attempt == max_retries:
logger.error(
f"Max retries ({max_retries}) reached for {func.__name__}"
)
raise
# Berechne Verzögerung mit Exponential Backoff
delay = min(base_delay * (exponential_base ** attempt), max_delay)
# Optional: Füge Jitter hinzu (±25%)
if jitter:
delay = delay * (0.75 + random.random() * 0.5)
logger.warning(
f"Attempt {attempt + 1}/{max_retries + 1} failed for "
f"{func.__name__}. Retrying in {delay:.2f}s..."
)
time.sleep(delay)
except TemporaryFailure as e:
last_exception = e
if attempt == max_retries:
raise
delay = base_delay * (exponential_base ** attempt)
logger.warning(f"Temporary failure. Retrying in {delay:.2f}s...")
time.sleep(delay)
raise last_exception
return wrapper
return decorator
Exception-Klassen
class RateLimitExceeded(Exception):
"""Wird bei HTTP 429 ausgelöst."""
pass
class TemporaryFailure(Exception):
"""Wird bei vorübergehenden Fehlern ausgelöst."""
pass
Anwendungsbeispiel
class AIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
@robust_retry(max_retries=3, base_delay=2.0, max_delay=30.0, jitter=True)
def make_request(self, model: str, prompt: str) -> dict:
"""
Sichere API-Anfrage mit automatischer Retry-Logik.
"""
import aiohttp
import asyncio
async def _request():
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 429:
raise RateLimitExceeded("Rate limit exceeded")
if response.status >= 500:
raise TemporaryFailure(f"Server error: {response.status}")
if response.status != 200:
raise Exception(f"API error: {response.status}")
return await response.json()
return asyncio.run(_request())
Nutzung
if __name__ == "__main__":
client = AIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = client.make_request("deepseek-v3.2", "Hallo Welt!")
print(f"Erfolgreich: {result}")
except RateLimitExceeded:
print("Rate limit reached after all retries. Please wait.")
except Exception as e:
print(f"Fehler: {e}")Fehler 2: Fehlende Kostenverfolgung führt zu Überraschungen
Problem: Ohne Echtzeit-Kostenverfolgung bemerken Sie Überschreitungen erst bei der Rechnung. Das kann für Startups existenzbedrohend sein.
Lösung: Implementieren Sie einen Cost Tracker als Middleware:
#!/usr/bin/env python3
"""
Echtzeit-Kostenverfolgung für KI-APIs
Mit Budget-Schwellenwerten und Alarmen
"""
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import threading
@dataclass
class CostEntry:
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost_usd: float
request_id: str
@dataclass
class BudgetWarning:
level: str # "warning" oder "critical"
percentage: float
spent: float
budget: float
timestamp: datetime
class CostTracker:
"""
Echtzeit-Kostenverfolgung für KI-API-Nutzung.
"""
def __init__(
self,
monthly_budget: float,
warning_threshold: float = 0.75,
critical_threshold: float = 0.90
):
self.monthly_budget = monthly_budget
self.warning_threshold = warning_threshold
self.critical_threshold = critical_threshold
self.entries: List[CostEntry] = []
self.alerts: List[BudgetWarning] = []
self._lock = threading.Lock()
# Preise pro Million Token (USD)
self.pricing = {
'deepseek-v3.2': {'input': 0.14, 'output': 0.42}, # $0.14 input, $0.42 output
'gemini-2.5-flash': {'input': 0.35, 'output': 2.50},
'gpt-4.1': {'input': 2.50, 'output': 8.00},
'claude-sonnet-4.5': {'input': 3.00, 'output': 15.00}
}
self._last_warning_level = None
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Berechnet Kosten basierend auf Token-Anzahl."""
prices = self.pricing.get(model, self.pricing['deepseek-v3.2'])
input_cost = (input_tokens / 1_000_000) * prices['input']
output_cost = (output_tokens / 1_000_000) * prices['output']
return input_cost + output_cost
def record(
self,
model: str,
input_tokens: int,
output_tokens: int,
request_id: Optional[str] = None
) -> Optional[BudgetWarning]:
"""Zeichnet eine API-Anfrage auf und prüft Budget-Schwellenwerte."""
with self._lock:
cost = self.calculate_cost(model, input_tokens, output_tokens)
entry = CostEntry(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost,
request_id=request_id or f"req_{len(self.entries)}"
)
self.entries.append(entry)
# Prüfe Budget
total_spent = self.get_total_spent()
percentage = total_spent / self.monthly_budget
warning = None
if percentage >= self.critical_threshold and self._last_warning_level != 'critical':
warning = BudgetWarning(
level="critical",
percentage=percentage * 100,
spent=total_spent,
budget=self.monthly_budget,
timestamp=datetime.now()
)
self.alerts.append(warning)
self._last_warning_level = 'critical'
print(f"🚨 KRITISCH: Budget zu {percentage*100:.1f}% ausgeschöpft!")
elif percentage >= self.warning_threshold and self._last_warning_level is None:
warning = BudgetWarning(
level="warning",
percentage=percentage * 100,
spent=total_spent,
budget=self.monthly_budget,
timestamp=datetime.now()
)
self.alerts.append(warning)
self._last_warning_level = 'warning'
print(f"⚠️ WARNUNG: Budget zu {percentage*100:.1f}% ausgeschöpft!")
return warning
def get