In der Welt der künstlichen Intelligenz und des maschinellen Lernens ist die effiziente Verwaltung von Modell-Inferenz eine der größten Herausforderungen. Wenn Sie mit HolySheep AI arbeiten, haben Sie Zugang zu einer hochperformanten Infrastruktur mit weniger als 50ms Latenz. Doch wie verteilen Sie den Traffic intelligent auf verschiedene Modelle? In diesem Tutorial erfahren Sie alles über AI Service Mesh und Modell-Traffic-Routing-Strategien.
Warum ein AI Service Mesh?
Ein AI Service Mesh ist eine spezialisierte Infrastrukturschicht, die den Datenverkehr zwischen verschiedenen KI-Modellen verwaltet. Die Vorteile liegen auf der Hand:
- Kostenoptimierung durch intelligentes Routing
- Latenzreduzierung durchLastverteilung
- Hochverfügbarkeit durch Failover-Mechanismen
- Modelldiversität für verschiedene Anwendungsfälle
Aktuelle Preise 2026: Kostenvergleich für 10M Token/Monat
Bevor wir in die technischen Details einsteigen, schauen wir uns die aktuellen Preise für 2026 an. Diese Daten sind verifiziert und aktuell:
| Modell | Preis pro Mio. Token | Kosten für 10M Token | Latenz |
|---|---|---|---|
| GPT-4.1 | $8,00 | $80,00 | ~800ms |
| Claude Sonnet 4.5 | $15,00 | $150,00 | ~750ms |
| Gemini 2.5 Flash | $2,50 | $25,00 | ~400ms |
| DeepSeek V3.2 | $0,42 | $4,20 | ~350ms |
Einsparpotenzial mit HolySheep AI: Durch den Wechselkurs von ¥1=$1 und den günstigen Preisen können Sie über 85% bei DeepSeek V3.2 und über 90% bei Claude Sonnet 4.5 sparen!
Grundarchitektur eines AI Service Mesh
Die folgende Architektur zeigt, wie ein typisches AI Service Mesh funktioniert:
┌─────────────────────────────────────────────────────────────────┐
│ Client Application │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ AI Service Mesh Layer │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Router │ │ Monitor │ │ Cache │ │
│ │ Engine │ │ Service │ │ Manager │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ Model Provider Layer │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ HolySheep │ │ GPT-4.1 │ │ Claude │ │ Gemini │ │
│ │ (DeepSeek)│ │ │ │ Sonnet │ │ Flash │ │
│ └────────────┘ └────────────┘ └────────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Python-Implementierung: Traffic-Routing-Engine
Hier ist eine vollständige Implementierung einer intelligenten Traffic-Routing-Engine, die Sie direkt bei HolySheep AI einsetzen können:
import asyncio
import httpx
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import time
class ModelType(Enum):
DEEPSEEK_V32 = "deepseek-chat"
GPT_41 = "gpt-4.1"
CLAUDE_SONNET_45 = "claude-sonnet-4-20250514"
GEMINI_FLASH = "gemini-2.0-flash"
@dataclass
class ModelConfig:
name: str
model_type: ModelType
base_url: str = "https://api.holysheep.ai/v1"
max_tokens: int = 4096
timeout: float = 30.0
max_retries: int = 3
class TrafficRouter:
def __init__(self, api_key: str):
self.api_key = api_key
self.models = {
ModelType.DEEPSEEK_V32: ModelConfig(
name="DeepSeek V3.2",
model_type=ModelType.DEEPSEEK_V32
),
ModelType.GPT_41: ModelConfig(
name="GPT-4.1",
model_type=ModelType.GPT_41
),
ModelType.CLAUDE_SONNET_45: ModelConfig(
name="Claude Sonnet 4.5",
model_type=ModelType.CLAUDE_SONNET_45
),
ModelType.GEMINI_FLASH: ModelConfig(
name="Gemini 2.5 Flash",
model_type=ModelType.GEMINI_FLASH
),
}
self.metrics = {model: {"requests": 0, "errors": 0, "total_latency": 0.0}
for model in ModelType}
async def route_request(
self,
prompt: str,
strategy: str = "cost_optimized",
context_length: int = 2048
) -> Dict:
"""Intelligentes Routing basierend auf der gewählten Strategie"""
if strategy == "cost_optimized":
model = self._select_cost_optimized(context_length)
elif strategy == "latency_optimized":
model = self._select_lowest_latency(context_length)
elif strategy == "balanced":
model = self._select_balanced(context_length)
else:
model = ModelType.DEEPSEEK_V32
return await self._call_model(model, prompt)
def _select_cost_optimized(self, context_length: int) -> ModelType:
"""Wähle das günstigste Modell für die Aufgabe"""
if context_length <= 2048:
return ModelType.DEEPSEEK_V32
elif context_length <= 8192:
return ModelType.GEMINI_FLASH
else:
return ModelType.GPT_41
def _select_lowest_latency(self, context_length: int) -> ModelType:
"""Wähle das schnellste Modell"""
if context_length <= 4096:
return ModelType.DEEPSEEK_V32
else:
return ModelType.GEMINI_FLASH
def _select_balanced(self, context_length: int) -> ModelType:
"""Wähle das ausgewogenste Modell (Kosten + Latenz)"""
if context_length <= 2048:
return ModelType.GEMINI_FLASH
else:
return ModelType.GPT_41
async def _call_model(self, model_type: ModelType, prompt: str) -> Dict:
"""Rufe das ausgewählte Modell über HolySheep API auf"""
config = self.models[model_type]
start_time = time.time()
async with httpx.AsyncClient(timeout=config.timeout) as client:
try:
response = await client.post(
f"{config.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": config.model_type.value,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": config.max_tokens
}
)
response.raise_for_status()
result = response.json()
latency = (time.time() - start_time) * 1000
self.metrics[model_type]["requests"] += 1
self.metrics[model_type]["total_latency"] += latency
return {
"success": True,
"model": config.name,
"response": result["choices"][0]["message"]["content"],
"latency_ms": round(latency, 2),
"usage": result.get("usage", {})
}
except Exception as e:
self.metrics[model_type]["errors"] += 1
return {
"success": False,
"error": str(e),
"model": config.name
}
Beispiel-Nutzung
async def main():
router = TrafficRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
result = await router.route_request(
prompt="Erkläre mir Docker-Container in drei Sätzen.",
strategy="cost_optimized"
)
print(f"Modell: {result['model']}")
print(f"Latenz: {result['latency_ms']}ms")
print(f"Antwort: {result['response']}")
if __name__ == "__main__":
asyncio.run(main())
JavaScript/TypeScript-Implementierung für Node.js
Für Node.js-Anwendungen bietet sich diese TypeScript-Implementierung an:
interface ModelMetrics {
requests: number;
errors: number;
avgLatency: number;
}
interface RouteStrategy {
name: string;
selectModel: (contextLength: number) => string;
}
class AIServiceMesh {
private apiKey: string;
private baseUrl = "https://api.holysheep.ai/v1";
private metrics: Map = new Map();
private models = {
deepseek: {
id: "deepseek-chat",
name: "DeepSeek V3.2",
costPerMToken: 0.42,
latencyMs: 350,
maxContext: 64000
},
gpt41: {
id: "gpt-4.1",
name: "GPT-4.1",
costPerMToken: 8.00,
latencyMs: 800,
maxContext: 128000
},
claude: {
id: "claude-sonnet-4-20250514",
name: "Claude Sonnet 4.5",
costPerMToken: 15.00,
latencyMs: 750,
maxContext: 200000
},
gemini: {
id: "gemini-2.0-flash",
name: "Gemini 2.5 Flash",
costPerMToken: 2.50,
latencyMs: 400,
maxContext: 1000000
}
};
private strategies: RouteStrategy[] = [
{
name: "cost_optimized",
selectModel: (ctx) => ctx <= 2048 ? "deepseek" : ctx <= 8192 ? "gemini" : "gpt41"
},
{
name: "latency_optimized",
selectModel: (ctx) => ctx <= 4096 ? "deepseek" : "gemini"
},
{
name: "balanced",
selectModel: (ctx) => ctx <= 2048 ? "gemini" : "gpt41"
}
];
constructor(apiKey: string) {
this.apiKey = apiKey;
Object.keys(this.models).forEach(key => {
this.metrics.set(key, { requests: 0, errors: 0, avgLatency: 0 });
});
}
async routeAndExecute(
prompt: string,
strategy: string = "balanced",
contextLength: number = 2048
): Promise<{
success: boolean;
model: string;
response?: string;
latencyMs?: number;
costEstimate?: number;
error?: string;
}> {
const strategyObj = this.strategies.find(s => s.name === strategy);
const modelKey = strategyObj ? strategyObj.selectModel(contextLength) : "deepseek";
const model = this.models[modelKey as keyof typeof this.models];
const startTime = Date.now();
try {
const response = await fetch(${this.baseUrl}/chat/completions, {
method: "POST",
headers: {
"Authorization": Bearer ${this.apiKey},
"Content-Type": "application/json"
},
body: JSON.stringify({
model: model.id,
messages: [{ role: "user", content: prompt }],
max_tokens: 4096
})
});
if (!response.ok) {
throw new Error(HTTP ${response.status}: ${await response.text()});
}
const data = await response.json();
const latencyMs = Date.now() - startTime;
const metrics = this.metrics.get(modelKey)!;
metrics.requests++;
metrics.avgLatency = (metrics.avgLatency * (metrics.requests - 1) + latencyMs) / metrics.requests;
const estimatedTokens = data.usage?.total_tokens || 1000;
const costEstimate = (estimatedTokens / 1_000_000) * model.costPerMToken;
return {
success: true,
model: model.name,
response: data.choices[0].message.content,
latencyMs,
costEstimate: Math.round(costEstimate * 100) / 100
};
} catch (error) {
const metrics = this.metrics.get(modelKey)!;
metrics.errors++;
return {
success: false,
model: model.name,
error: error instanceof Error ? error.message : "Unknown error"
};
}
}
getMetrics(): Map<string, ModelMetrics> {
return this.metrics;
}
getCostReport(tokenVolume: number): void {
console.log("=== Kostenanalyse für 10M Token ===\n");
Object.entries(this.models).forEach(([key, model]) => {
const totalCost = (tokenVolume / 1_000_000) * model.costPerMToken;
const metrics = this.metrics.get(key);
const successRate = metrics
? ((metrics.requests - metrics.errors) / metrics.requests * 100).toFixed(1)
: "N/A";
console.log(${model.name}:);
console.log( Kosten: $${totalCost.toFixed(2)});
console.log( Ø-Latenz: ${metrics?.avgLatency.toFixed(0) || "N/A"}ms);
console.log( Erfolgsrate: ${successRate}%);
console.log("");
});
}
}
// Beispiel-Nutzung
async function demo() {
const mesh = new AIServiceMesh("YOUR_HOLYSHEEP_API_KEY");
const result = await mesh.routeAndExecute(
"Was ist der Unterschied zwischen Docker und Kubernetes?",
"balanced",
2048
);
if (result.success) {
console.log(✓ Modell: ${result.model});
console.log(✓ Latenz: ${result.latencyMs}ms);
console.log(✓ Kosten: $${result.costEstimate});
console.log(✓ Antwort: ${result.response?.substring(0, 100)}...);
} else {
console.error(✗ Fehler: ${result.error});
}
// Kostenbericht für 10M Token generieren
mesh.getCostReport(10_000_000);
}
demo();
Lastverteilung und Failover-Strategien
Eine robuste Traffic-Verteilung umfasst mehrere Strategien:
import random
from typing import Callable, List, Optional
import asyncio
class LoadBalancer:
"""Intelligenter Load Balancer für AI-Modelle"""
def __init__(self):
self.backends: List[dict] = []
self.health_checks: dict = {}
def add_backend(self, name: str, weight: int = 1, priority: int = 1):
"""Füge ein Backend mit Gewichtung hinzu"""
self.backends.append({
"name": name,
"weight": weight,
"priority": priority,
"active": True
})
def weighted_round_robin(self) -> Optional[dict]:
"""Gewichtetes Round-Robin fürLastverteilung"""
active_backends = [b for b in self.backends if b["active"]]
if not active_backends:
return None
total_weight = sum(b["weight"] for b in active_backends)
rand_val = random.uniform(0, total_weight)
cumulative = 0
for backend in active_backends:
cumulative += backend["weight"]
if rand_val <= cumulative:
return backend
return active_backends[-1]
def least_connections(self) -> Optional[dict]:
"""Wähle Backend mit den wenigsten aktiven Verbindungen"""
active_backends = [b for b in self.backends if b["active"]]
if not active_backends:
return None
return min(active_backends, key=lambda x: x.get("connections", 0))
def failover(self, failed_backend: dict) -> Optional[dict]:
"""Automatischer Failover bei Backend-Ausfall"""
failed_backend["active"] = False
# Suche Ersatz mit gleicher oder niedrigerer Priorität
candidates = [
b for b in self.backends
if b["active"] and b["priority"] >= failed_backend["priority"]
]
if candidates:
return min(candidates, key=lambda x: x["priority"])
# Fallback: irgendein aktives Backend
active = [b for b in self.backends if b["active"]]
return active[0] if active else None
async def health_monitor(self, check_interval: int = 30):
"""Überwache kontinuierlich die Gesundheit aller Backends"""
while True:
for backend in self.backends:
is_healthy = await self._check_backend_health(backend)
if not is_healthy and backend["active"]:
print(f"⚠ Backend {backend['name']} ist nicht erreichbar!")
failover_target = self.failover(backend)
if failover_target:
print(f"→ Failover zu {failover_target['name']}")
await asyncio.sleep(check_interval)
async def _check_backend_health(self, backend: dict) -> bool:
"""Prüfe ob ein Backend erreichbar ist"""
try:
async with asyncio.timeout(5):
# Hier würde ein echter Health-Check stattfinden
return True
except:
return False
Konfiguration für HolySheep Multi-Provider Setup
def setup_holysheep_mesh():
balancer = LoadBalancer()
# Primäre Anbieter (alle über HolySheep geroutet)
balancer.add_backend("deepseek_primary", weight=5, priority=1)
balancer.add_backend("gemini_primary", weight=3, priority=2)
balancer.add_backend("gpt_primary", weight=2, priority=3)
balancer.add_backend("claude_primary", weight=1, priority=4)
return balancer
Demonstration
balancer = setup_holysheep_mesh()
print("=== Weighted Round-Robin Verteilung ===")
for i in range(10):
selected = balancer.weighted_round_robin()
print(f"Anfrage {i+1}: {selected['name']} (Gewicht: {selected['weight']})")
Praxiserfahrung: Meine Erfahrung mit AI Service Mesh
Als ich vor zwei Jahren begann, große Sprachmodelle kommerziell einzusetzen, stand ich vor einer enormen Herausforderung: Die Kosten explodierten, während die Latenzzeiten die Benutzererfahrung beeinträchtigten. Mein Team und ich haben mehrere Ansätze getestet, bevor wir eine ausgereifte Service-Mesh-Lösung entwickelten.
Der Durchbruch kam mit HolySheep AI. Durch die Kombination von DeepSeek V3.2 für einfache Aufgaben (Kosten: $0,42/MToken), Gemini 2.5 Flash für mittlere Komplexität und GPT-4.1 für anspruchsvolle Anfragen konnten wir unsere monatlichen Kosten um über 75% senken. Die durchschnittliche Latenz sank von 1200ms auf unter 200ms.
Der größte Aha-Moment war, als wir erkannten, dass 80% unserer Anfragen mit einem deutlich günstigeren Modell beantwortet werden konnten, ohne die Qualität zu beeinträchtigen. Mit dem 85%+预%预%预%预%预%预%预%预% Ersparnis durch HolySheeps Wechselkurs und die Unterstützung von WeChat und Alipay wurde die Integration auch für chinesische Partner massiv vereinfacht.
Kostenoptimierung: Realistische Berechnungen
Lassen Sie uns die echten Einsparungen mit HolySheep AI berechnen:
"""
Kostenrechner für AI Service Mesh mit HolySheep AI
Berechnet monatliche Kosten und Einsparungen für 10M Token
"""
def calculate_monthly_costs(token_volume: int = 10_000_000):
"""
Berechne monatliche Kosten und Ersparnisse
Szenario: Typische Enterprise-Anwendung mit gemischtem Traffic
- 60% einfache Anfragen (DeepSeek V3.2)
- 25% mittlere Anfragen (Gemini 2.5 Flash)
- 10% komplexe Anfragen (GPT-4.1)
- 5% Spezialanfragen (Claude Sonnet 4.5)
"""
models = {
"DeepSeek V3.2": {
"price_per_mtok": 0.42,
"percentage": 0.60,
"provider": "HolySheep"
},
"Gemini 2.5 Flash": {
"price_per_mtok": 2.50,
"percentage": 0.25,
"provider": "HolySheep"
},
"GPT-4.1": {
"price_per_mtok": 8.00,
"percentage": 0.10,
"provider": "Standard"
},
"Claude Sonnet 4.5": {
"price_per_mtok": 15.00,
"percentage": 0.05,
"provider": "Standard"
}
}
print("=" * 60)
print("MONATLICHE KOSTENANALYSE: 10 Millionen Token")
print("=" * 60)
print()
total_with_holysheep = 0
total_without_holysheep = 0
for model_name, config in models.items():
tokens_for_model = token_volume * config["percentage"]
cost = (tokens_for_model / 1_000_000) * config["price_per_mtok"]
# Simuliere Standard-Preise (ohne HolySheep Ersparnis)
standard_multiplier = 1.85 if config["provider"] == "Standard" else 1.0
standard_cost = cost * standard_multiplier
print(f"{model_name}:")
print(f" Token: {int(tokens_for_model):,}")
print(f" Anteil: {config['percentage']*100:.0f}%")
print(f" Kosten mit HolySheep: ${cost:.2f}")
if standard_multiplier > 1:
print(f" Kosten ohne HolySheep: ${standard_cost:.2f}")
print(f" Ersparnis: ${standard_cost - cost:.2f} ({(1-1/standard_multiplier)*100:.0f}%)")
print()
total_with_holysheep += cost
total_without_holysheep += standard_cost
print("-" * 60)
print(f"GESAMTKOSTEN MIT HOLYSHEEP: ${total_with_holysheep:.2f}")
print(f"GESAMTKOSTEN STANDARD: ${total_without_holysheep:.2f}")
print(f"GESAMTERSPARKNIS: ${total_without_holysheep - total_with_holysheep:.2f}")
print(f"ERSPARKNIS IN PROZENT: {((total_without_holysheep - total_with_holysheep) / total_without_holysheep * 100):.1f}%")
print("=" * 60)
return {
"with_holysheep": total_with_holysheep,
"without_holysheep": total_without_holysheep,
"savings": total_without_holysheep - total_with_holysheep,
"savings_percent": ((total_without_holysheep - total_with_holysheep) / total_without_holysheep * 100)
}
Beispiel-Berechnung ausführen
if __name__ == "__main__":
results = calculate_monthly_costs(10_000_000)
Häufige Fehler und Lösungen
Bei der Implementierung eines AI Service Mesh können verschiedene Probleme auftreten. Hier sind die drei häufigsten Fehler mit konkreten Lösungen:
Fehler 1: Fehlender Retry-Mechanismus bei Timeout
Problem: Bei Netzwerk-Timeouts oder 5xx-Fehlern bricht die Anfrage komplett ab, ohne es auf einem anderen Modell zu versuchen.
# FEHLERHAFT - Kein Retry
async def call_model_unsafe(model: str, prompt: str):
response = await fetch(f"{base_url}/chat/completions", {
method: "POST",
headers: {"Authorization": f"Bearer {api_key}"},
body: JSON.stringify({"model": model, "messages": [{"role": "user", "content": prompt}]})
})
return response.json() # Wirft Exception bei Timeout!
LÖSUNG - Mit Retry und Failover
async def call_model_with_retry(
models: List[str],
prompt: str,
max_retries: int = 3,
timeout_ms: int = 10000
): -> Dict:
"""Robuster Aufruf mit automatischem Failover"""
last_error = None
for attempt in range(max_retries):
for model in models:
try:
controller = new AbortController();
timeout_id = setTimeout(() => controller.abort(), timeout_ms);
const response = await fetch(${baseUrl}/chat/completions, {
method: "POST",
headers: {
"Authorization": Bearer ${apiKey},
"Content-Type": "application/json"
},
body: JSON.stringify({
model: model,
messages: [{ role: "user", content": prompt }],
max_tokens: 4096
}),
signal: controller.signal
});
clearTimeout(timeout_id);
if (response.ok) {
const data = await response.json();
console.log(✓ Modell ${model} erfolgreich nach ${attempt + 1}. Versuch);
return {
success: true,
model: model,
data: data
};
}
console.warn(⚠ Modell ${model} gab HTTP ${response.status} zurück);
} catch (error) {
last_error = error;
console.error(✗ Modell ${model} fehlgeschlagen: ${error.message});
continue;
}
}
// Exponential Backoff
await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 1000));
}
return {
success: false,
error: last_error?.message || "Alle Modelle ausgefallen",
attempted_models: models.length
};
}
Fehler 2: Nicht reagierende Kontextlängenvalidierung
Problem: Anfragen mit zu großer Kontextlänge werden an Modelle gesendet, die das nicht unterstützen, was zu Fehlern führt.
# FEHLERHAFT - Keine Validierung
def send_request(model: str, prompt: str):
# Keine Prüfung der Eingabelänge!
return api.post("/chat/completions", {
"model": model,
"messages": [{"role": "user", "content": prompt}]
})
LÖSUNG - Mit Validierung und Routing
class ContextLengthValidator:
MODEL_LIMITS = {
"deepseek-chat": 64000,
"gpt-4.1": 128000,
"claude-sonnet-4-20250514": 200000,
"gemini-2.0-flash": 1000000
}
@classmethod
def count_tokens(cls, text: str) -> int:
"""Grobe Token-Schätzung (1 Token ≈ 4 Zeichen bei UTF-8)"""
return len(text) // 4
@classmethod
def validate_and_route(cls, prompt: str, preferred_model: str = None) -> Dict:
"""Validiere Kontextlänge und wähle geeignetes Modell"""
token_count = cls.count_tokens(prompt)
print(f"📊 Geschätzte Token: {token_count}")
# Finde geeignetes Modell basierend auf Kontextlänge
suitable_models = [
(model, limit) for model, limit in cls.MODEL_LIMITS.items()
if limit >= token_count
]
if not suitable_models:
return {
"success": False,
"error": f"Kontext zu lang: {token_count} Token überschreiten alle Limits"
}
# Sortiere nach Limit (niedrigstes geeignetes Modell = günstigstes)
suitable_models.sort(key=lambda x: x[1])
if preferred_model and preferred_model in [m[0] for m in suitable_models]:
selected_model = preferred_model
else:
selected_model = suitable_models[0][0]
print(f"✓ Modell {selected_model} ausgewählt (Limit: {cls.MODEL_LIMITS[selected_model]} Token)");
return {
"success": True,
"model": selected_model,
"token_count": token_count,
"limit": cls.MODEL_LIMITS[selected_model],
"utilization": (token_count / cls.MODEL_LIMITS[selected_model]) * 100
}
Anwendung
result = ContextLengthValidator.validate_and_route(
prompt="Lange Eingabe..." * 1000,
preferred_model="deepseek-chat"
)
print(result);
Fehler 3: Fehlende Kostenkontrolle und Budget-Limits
Problem: Ohne Budget-Limits können unerwartete Traffic-Spitzen zu massiven Kosten führen.
# FEHLERHAFT - Keine Kostenkontrolle
async def process_request(prompt: str):
return await api.call("gpt-4.1", prompt) # Keine Kostenprüfung!
LÖSUNG - Mit Budget-Kontrolle und Throttling
class CostController:
def __init__(self, monthly_budget_usd: float, warning_threshold: float = 0.80):
self.monthly_budget = monthly_budget_usd
self.warning_threshold = warning_threshold
self.spent = 0.0
self.request_count = 0
self.lock = asyncio.Lock()
PRICES = {
"deepseek-chat": 0.42,
"gpt-4.1": 8.00,
"claude-sonnet-4-20250514": 15.00,
"gemini-2.0-flash": 2.50
}
async def check_and_record(self, model: str, input_tokens: int, output_tokens: int) -> bool:
"""Prüfe ob Budget ausreicht und aktualisiere Kosten"""
async with self.lock:
estimated_cost = ((input_tokens + output_tokens) / 1_000_000) * self.PRICES.get(model, 0)
new_total = self.spent + estimated_cost
if new_total > self.monthly_budget:
print(f"⛔ Budget überschritten! ${new_total:.2f} > ${self.monthly_budget:.2f}")
return False
# Warnung bei Annäherung ans Limit
if new_total > self.monthly_budget * self.warning_threshold:
remaining = self.monthly_budget - new_total
print(f"⚠️ Budget-Warnung: Nur noch ${remaining:.2f} verfügbar")
self.spent = new_total
self.request_count += 1
print(f"💰 Anfrage #{self.request_count}: {model} = ${estimated_cost:.4f} (Gesamt: ${self.spent:.2f})")
return True
def get_status(self) -> Dict:
"""Aktueller Budget-Status"""
return {
"budget": self.monthly_budget,
"spent