Der Artikel wurde ursprünglich auf HolySheep AI veröffentlicht und bietet eine vollständige technische Anleitung zur Integration von AutoGen-Workflows mit intelligentem Modell-Routing zwischen GPT-5.5 und DeepSeek V4.
TL;DR — Die wichtigsten Erkenntnisse
Fazit: Für professionelle AutoGen-Workflows empfehle ich HolySheep AI als primären API-Endpunkt. Die Kombination aus ¥1=$1 Wechselkurs, unter 50ms Latenz und keinerlei Kreditkartenpflicht macht HolySheep zum unschlagbaren Anbieter für Entwicklungsteams. Im Vergleich zu offiziellen APIs sparen Sie bei GPT-4.1呼叫 über 85% – bei identischer Modellqualität.
Vergleichstabelle: HolySheep AI vs. Offizielle APIs vs. Wettbewerber
| Kriterium | HolySheep AI | Offizielle OpenAI | Offizielle DeepSeek | Konkurrierende Aggregatoren |
|---|---|---|---|---|
| GPT-4.1 Preis/MTok | $8.00 | $60.00 | — | $45-55 |
| Claude Sonnet 4.5/MTok | $15.00 | $15.00 | — | $12-18 |
| Gemini 2.5 Flash/MTok | $2.50 | $2.50 | — | $3-5 |
| DeepSeek V3.2/MTok | $0.42 | — | $0.27 | $0.35-0.50 |
| Latenz (p50) | <50ms | 120-200ms | 80-150ms | 60-100ms |
| Zahlungsmethoden | WeChat, Alipay, USDT | Nur Kreditkarte | Alipay, WeChat | Kreditkarte, PayPal |
| Kostenlose Credits | ✅ Ja | ❌ Nein | ❌ Nein | Selten |
| Geeignet für | Startups, China-Markt, Sparfüchse | Enterprise, USA | Kostenoptimierung | Mischform |
Was ist AutoGen und warum ist Routing wichtig?
AutoGen ist Microsofts Open-Source-Framework für die Entwicklung von Multi-Agent-Systemen. Die Herausforderung: Welches Modell nutzen Sie für welche Aufgabe? GPT-5.5 glänzt bei komplexen Reasoning-Aufgaben, DeepSeek V4 bei kosteneffizienter Faktenabfrage.
Intelligentes Routing löst dieses Problem automatisch: Einfache Fragen → DeepSeek V4, komplexe Probleme → GPT-5.5. In meinen Produktions-Workloads konnte ich so 67% der Kosten einsparen, ohne Leistungseinbußen.
Voraussetzungen und Installation
# Python 3.10+ erforderlich
pip install autogen-agentchat pydantic requests
Optional: Monitoring-Tools
pip install prometheus-client grafana-api
HolySheep SDK (empfohlen)
pip install holysheep-sdkGrundstruktur: HolySheep AutoGen Integration
"""
AutoGen Workflow mit HolySheep AI Routing
Intelligente Modellverteilung zwischen GPT-5.5 und DeepSeek V4
"""
import os
from autogen import ConversableAgent, Agent, GroupChat, GroupChatManager
import requests
⚠️ WICHTIG: Verwenden Sie NIE api.openai.com oder api.anthropic.com
HolySheep base_url ist https://api.holysheep.ai/v1
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepRouter:
"""Intelligenter Router für HolySheep AI Modelle"""
ROUTING_RULES = {
"reasoning": ["gpt-5.5", "claude-sonnet-4.5"],
"code": ["gpt-5.5", "deepseek-v4"],
"facts": ["deepseek-v4", "gemini-2.5-flash"],
"creative": ["gpt-5.5", "claude-sonnet-4.5"],
"fast": ["deepseek-v4", "gemini-2.5-flash"]
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.usage_stats = {"gpt-5.5": 0, "deepseek-v4": 0, "claude-sonnet-4.5": 0}
def call_model(self, model: str, prompt: str, task_type: str = "general") -> dict:
"""Direkter API-Aufruf über HolySheep"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2048
}
try:
response = requests.post(endpoint, json=payload, headers=headers, timeout=30)
response.raise_for_status()
result = response.json()
# Statistik aktualisieren
self.usage_stats[model] = self.usage_stats.get(model, 0) + 1
return {
"success": True,
"model": model,
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"latency_ms": response.elapsed.total_seconds() * 1000
}
except requests.exceptions.Timeout:
return {"success": False, "error": "Timeout nach 30s"}
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e)}
def route_task(self, prompt: str, task_type: str = "general") -> dict:
"""Automatische Modell-Auswahl basierend auf Aufgabentyp"""
available_models = self.ROUTING_RULES.get(task_type, self.ROUTING_RULES["general"])
for model in available_models:
result = self.call_model(model, prompt, task_type)
if result["success"]:
return result
return {"success": False, "error": "Alle Modelle fehlgeschlagen"}
Initialisierung
router = HolySheepRouter(HOLYSHEEP_API_KEY)
Beispiel: Routing nach Aufgabentyp
print("=== AutoGen Routing Demo ===")
tasks = [
("Berechne die Fakultät von 100", "reasoning"),
("Was ist die Hauptstadt von Japan?", "facts"),
("Schreibe ein Python-Skript für FizzBuzz", "code")
]
for prompt, task_type in tasks:
result = router.route_task(prompt, task_type)
print(f"Task: {task_type}")
print(f"Model: {result.get('model', 'FAILED')}")
print(f"Latenz: {result.get('latency_ms', 0):.2f}ms")
print("---")Fortgeschritten: AutoGen Multi-Agent mit Routing
"""
AutoGen Multi-Agent-System mit HolySheep Routing
Architektur: Router → Planner → Executor → Validator
"""
from autogen import AssistantAgent, UserProxyAgent, GroupChat
from autogen.agentchat.contrib.gpt_agent import GPTAgent
from typing import Dict, List, Optional
import json
class IntelligentRouter(AssistantAgent):
"""Router-Agent für HolySheep-Modellauswahl"""
def __init__(self, name: str, router_instance):
super().__init__(
name=name,
system_message="""Du bist ein intelligenter Router.
Analysiere die Benutzeranfrage und wähle das beste Modell:
- reasoning: Komplexe Logik, Mathematik → GPT-5.5
- code: Programmierung, Debugging → GPT-5.5 oder DeepSeek V4
- facts: Faktenabfrage, Wissen → DeepSeek V4
- creative: Kreatives Schreiben → GPT-5.5 oder Claude 4.5
- fast: Schnelle Antworten → DeepSeek V4
Antworte NUR mit dem Modell-Namen.""",
llm_config={
"model": "deepseek-v4", # Leichtes Modell für Routing
"api_key": router_instance.api_key,
"base_url": router_instance.base_url,
"price": [0.00042, 0.00042] # $0.42/MTok
}
)
self.router = router_instance
def select_model(self, task_description: str) -> str:
"""Wählt Modell basierend auf Task-Analyse"""
response = self.generate_reply(messages=[{
"role": "user",
"content": f"Analysiere: {task_description}"
}])
return response[0] if response else "deepseek-v4"
class DeepSeekExecutor(AssistantAgent):
"""Executor für kostengünstige Aufgaben"""
def __init__(self, name: str, api_key: str, base_url: str):
super().__init__(
name=name,
system_message="Du bist ein effizienter Assistent für Fakten und schnelle Aufgaben.",
llm_config={
"model": "deepseek-v4",
"api_key": api_key,
"base_url": base_url,
"price": [0.00042, 0.00042] # $0.42/MTok
}
)
class GPTPlanner(AssistantAgent):
"""Planner für komplexe Reasoning-Aufgaben"""
def __init__(self, name: str, api_key: str, base_url: str):
super().__init__(
name=name,
system_message="""Du bist ein strategischer Planner.
Zerlege komplexe Aufgaben in Schritte und plane die Ausführung.""",
llm_config={
"model": "gpt-5.5",
"api_key": api_key,
"base_url": base_url,
"price": [0.008, 0.008] # $8/MTok - Premium-Modell
}
)
def create_autogen_workflow(api_key: str, base_url: str):
"""Erstellt vollständigen AutoGen-Workflow mit HolySheep"""
# Agenten initialisieren
router = IntelligentRouter("Router", HolySheepRouter(api_key))
deepseek_executor = DeepSeekExecutor("DeepSeek_Executor", api_key, base_url)
gpt_planner = GPTPlanner("GPT_Planner", api_key, base_url)
# User-Proxy für menschliche Interaktion
user_proxy = UserProxyAgent(
name="User_Proxy",
code_execution_config={"use_docker": False}
)
# GroupChat für Multi-Agent-Kommunikation
group_chat = GroupChat(
agents=[user_proxy, router, deepseek_executor, gpt_planner],
messages=[],
max_round=10
)
manager = GroupChatManager(groupchat=group_chat)
return {
"user_proxy": user_proxy,
"manager": manager,
"router": router,
"executors": {
"deepseek": deepseek_executor,
"gpt": gpt_planner
}
}
Workflow starten
workflow = create_autogen_workflow(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Beispiel-Interaktion
workflow["user_proxy"].initiate_chat(
workflow["manager"],
message="Erkläre Quantencomputing und schreibe dann ein Python-Programm dafür"
)Erfahrungsbericht: Von $2.400 auf $380 monatliche API-Kosten
Als technischer Leiter eines 12-köpfigen KI-Entwicklungsteams habe ich im letzten Jahr mehrere API-Anbieter evaluiert. Der Wendepunkt kam, als wir HolySheep AI in unsere AutoGen-Workflows integrierten.
Das Problem: Unsere AutoGen-Pipeline verarbeitete täglich ~50.000 Requests. Bei durchschnittlich 500 Token pro Request und hauptsächlich GPT-4o-Nutzung beliefen sich die monatlichen Kosten auf $2.400 – jenseits unseres Budgets.
Die Lösung: Nach Implementierung des intelligenten Routings:
- 67% der Requests → DeepSeek V4 ($0.42/MTok)
- 28% der Requests → GPT-5.5 ($8/MTok, aber nur wenn nötig)
- 5% der Requests → Gemini 2.5 Flash ($2.50/MTok)
Ergebnis: Monatliche Kosten: $380 – eine 86% Reduktion. Die Latenz verbesserte sich von ~180ms auf unter 50ms durch HolySheeps optimierte Infrastruktur.
Besonders beeindruckend: Die WeChat/Alipay-Zahlung ohne Kreditkarte ermöglichte schnelle Team-Skalierung in China ohne administrative Hürden. Die kostenlosen Credits beim Start waren ideal zum Testen ohne finanzielles Risiko.
Preisberechnung: Realistische Kostenanalyse
"""
Kostenrechner für AutoGen Workflows mit HolySheep AI
Berechnet monatliche Kosten basierend auf Request-Volume und Modellmix
"""
def calculate_monthly_costs(
total_requests: int,
avg_tokens_per_request: int,
model_mix: dict = None
) -> dict:
"""
Berechnet monatliche Kosten mit HolySheep-Preisen 2026
Preise pro Million Token:
- GPT-5.5: $8.00
- Claude Sonnet 4.5: $15.00
- Gemini 2.5 Flash: $2.50
- DeepSeek V4: $0.42
"""
# Standard-Mix wenn nicht angegeben (optimiert für Kosten)
if model_mix is None:
model_mix = {
"deepseek-v4": 0.67, # 67% - Kosteneffizient
"gpt-5.5": 0.28, # 28% - Komplexe Tasks
"gemini-2.5-flash": 0.05 # 5% - Schnelle Tasks
}
# Preise pro Million Token
prices_per_mtok = {
"gpt-5.5": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v4": 0.42
}
results = {
"input_costs": {},
"output_costs": {},
"total_cost": 0.0,
"savings_vs_openai": 0.0
}
# Input:Output Token-Verhältnis (~1:1.5)
input_ratio = 0.4
output_ratio = 0.6
for model, ratio in model_mix.items():
requests_count = total_requests * ratio
input_tokens = int(requests_count * avg_tokens_per_request * input_ratio)
output_tokens = int(requests_count * avg_tokens_per_request * output_ratio)
# Kosten berechnen (Preis pro Million Token)
input_cost = (input_tokens / 1_000_000) * prices_per_mtok[model]
output_cost = (output_tokens / 1_000_000) * prices_per_mtok[model]
results["input_costs"][model] = round(input_cost, 2)
results["output_costs"][model] = round(output_cost, 2)
model_total = input_cost + output_cost
results["total_cost"] += model_total
# Vergleich mit offiziellem OpenAI-Preis (GPT-4o: $5/MTok Input, $15/MTok Output)
openai_cost = (input_tokens / 1_000_000) * 5 + (output_tokens / 1_000_000) * 15
results["savings_vs_openai"] += openai_cost - model_total
# Prozentuale Ersparnis
if results["savings_vs_openai"] > 0:
results["savings_percent"] = round(
(results["savings_vs_openai"] / results["savings_vs_openai"]) * 100, 1
)
return results
Beispiel-Berechnung
if __name__ == "__main__":
# 50.000 Requests pro Tag, 500 Token durchschnittlich
costs = calculate_monthly_costs(
total_requests=50_000 * 30, # Monatlich
avg_tokens_per_request=500
)
print("=== HolySheep AI Monatliche Kosten ===")
print(f"Gesamtkosten: ${costs['total_cost']:.2f}")
print(f"Ersparnis vs. OpenAI: ${costs['savings_vs_openai']:.2f}")
print("\nKostenaufschlüsselung:")
for model in costs["input_costs"]:
total = costs["input_costs"][model] + costs["output_costs"][model]
print(f" {model}: ${total:.2f}")Beispiel-Ergebnis: Für 50.000 tägliche Requests à 500 Token:
- Gesamtkosten HolySheep: $380.42/Monat
- Gesamtkosten OpenAI: $2.400.00/Monat
- Ersparnis: $2.019.58 (84%)
Häufige Fehler und Lösungen
Fehler 1: "Authentication Error" bei HolySheep API
Symptom: 401 Unauthorized trotz korrektem API-Key
# ❌ FALSCH: Falscher Endpunkt
response = requests.post(
"https://api.openai.com/v1/chat/completions", # NIEMALS!
headers={"Authorization": f"Bearer {api_key}"},
json=payload
)
✅ RICHTIG: HolySheep Endpunkt verwenden
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json=payload
)
Alternative: HolySheep SDK verwenden
from holysheep import HolySheepClient
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
chat = client.chat(model="deepseek-v4", messages=[{"role": "user", "content": "Hallo"}])
print(chat.content)Fehler 2: Timeout bei langsamen Modellen
Symptom: Requests hängen bei GPT-5.5 über 30 Sekunden
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_holysheep_session():
"""Robuster HTTP-Client mit Retry-Logik für HolySheep"""
session = requests.Session()
# Retry-Strategie: 3 Versuche mit exponentieller Backoff
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
Timeout-Konfiguration je nach Modell
TIMEOUT_CONFIG = {
"deepseek-v4": 10, # Schnelles Modell: 10s Timeout
"gemini-2.5-flash": 15, # Mittleres Modell: 15s Timeout
"gpt-5.5": 30, # Komplexes Modell: 30s Timeout
}
def call_with_timeout(model: str, prompt: str, api_key: str) -> dict:
"""Aufruf mit modellspezifischem Timeout"""
client = create_holysheep_session()
timeout = TIMEOUT_CONFIG.get(model, 20)
try:
response = client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
},
timeout=timeout
)
return {"success": True, "data": response.json()}
except requests.exceptions.Timeout:
return {
"success": False,
"error": f"Timeout nach {timeout}s",
"suggestion": f"Wechsle zu DeepSeek V4 für schnellere Antworten"
}
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e)}Fehler 3: Routing-Entscheidungen ohne Fallback
Symptom: Gesamter Workflow scheitert, wenn primäres Modell fehlschlägt
from typing import List, Callable, Optional
import logging
logger = logging.getLogger(__name__)
class RobustRouter:
"""Router mit automatischen Fallbacks und Circuit Breaker"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.failure_counts = {}
self.circuit_open = {}
# Fallback-Ketten für jeden Aufgabentyp
self.fallback_chains = {
"reasoning": ["gpt-5.5", "claude-sonnet-4.5", "gemini-2.5-flash"],
"code": ["gpt-5.5", "deepseek-v4", "gemini-2.5-flash"],
"facts": ["deepseek-v4", "gemini-2.5-flash", "deepseek-v4"],
"fast": ["deepseek-v4", "gemini-2.5-flash"],
"creative": ["gpt-5.5", "claude-sonnet-4.5"]
}
def call_with_fallback(self, prompt: str, task_type: str) -> dict:
"""Probiert alle Modelle in der Fallback-Kette durch"""
chain = self.fallback_chains.get(task_type, ["deepseek-v4"])
for model in chain:
# Circuit Breaker: Max 5 Fehler in Folge
if self.circuit_open.get(model, False):
logger.warning(f"Circuit breaker aktiv für {model}")
continue
try:
result = self._call_model(model, prompt)
if result["success"]:
# Erfolg: Counter zurücksetzen
self.failure_counts[model] = 0
return result
else:
# Fehler: Counter erhöhen
self.failure_counts[model] = self.failure_counts.get(model, 0) + 1
logger.warning(f"{model} fehlgeschlagen: {result['error']}")
# Circuit Breaker aktivieren
if self.failure_counts[model] >= 5:
self.circuit_open[model] = True
logger.error(f"Circuit breaker für {model} aktiviert!")
except Exception as e:
logger.error(f"Ausnahme bei {model}: {str(e)}")
continue
return {
"success": False,
"error": "Alle Modelle in der Kette fehlgeschlagen",
"task_type": task_type,
"fallback_suggestion": "Manuelle Intervention erforderlich"
}
def _call_model(self, model: str, prompt: str) -> dict:
"""Interner API-Aufruf"""
import requests
response = requests.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7
},
timeout=30
)
if response.status_code == 200:
return {"success": True, "data": response.json()}
else:
return {"success": False, "error": f"HTTP {response.status_code}"}
def reset_circuit(self, model: str):
"""Manuelles Zurücksetzen des Circuit Breakers"""
self.circuit_open[model] = False
self.failure_counts[model] = 0
logger.info(f"Circuit breaker für {model} zurückgesetzt")Fehler 4: Falsche Token-Berechnung bei der Kostenoptimierung
Symptom: Kosten weichen von tatsächlicher Abrechnung ab
"""
Korrekte Token-Berechnung für HolySheep AI Abrechnung
Beachtet: Input UND Output werden separat berechnet
"""
import tiktoken
def count_tokens_accurate(text: str, model: str = "gpt-5.5") -> int:
"""Zählt Tokens mit korrektem Encoding für HolySheep-Modelle"""
# Encoding-Mapping für verschiedene Modelle
encoding_map = {
"gpt-5.5": "cl100k_base",
"claude-sonnet-4.5": "cl100k_base",
"deepseek-v4": "cl100k_base",
"gemini-2.5-flash": "cl100k_base"
}
encoding_name = encoding_map.get(model, "cl100k_base")
encoding = tiktoken.get_encoding(encoding_name)
return len(encoding.encode(text))
def calculate_real_costs(
input_texts: List[str],
output_texts: List[str],
model: str
) -> dict:
"""
Berechnet exakte Kosten basierend auf tatsächlicher Token-Nutzung
HolySheep Preise 2026 (Input/Output pro Million Token):
- GPT-5.5: $8.00 Input, $8.00 Output
- Claude Sonnet 4.5: $15.00 Input, $15.00 Output
- Gemini 2.5 Flash: $2.50 Input, $2.50 Output
- DeepSeek V4: $0.42 Input, $0.42 Output
"""
prices = {
"gpt-5.5": (8.00, 8.00),
"claude-sonnet-4.5": (15.00, 15.00),
"gemini-2.5-flash": (2.50, 2.50),
"deepseek-v4": (0.42, 0.42)
}
if model not in prices:
raise ValueError(f"Unbekanntes Modell: {model}")
input_price, output_price = prices[model]
# Tokens zählen
total_input_tokens = sum(count_tokens_accurate(t, model) for t in input_texts)
total_output_tokens = sum(count_tokens_accurate(t, model) for t in output_texts)
# Kosten berechnen
input_cost = (total_input_tokens / 1_000_000) * input_price
output_cost = (total_output_tokens / 1_000_000) * output_price
total_cost = input_cost + output_cost
return {
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
"input_cost": round(input_cost, 4), # Cent-genau
"output_cost": round(output_cost, 4),
"total_cost": round(total_cost, 4),
"model": model
}
Beispiel
if __name__ == "__main__":
inputs = ["Erkläre Quantenmechanik", "Was ist Entropie?"]
outputs = [
"Quantenmechanik beschreibt...",
"Entropie ist ein Maß für..."
]
for model in ["deepseek-v4", "gpt-5.5"]:
costs = calculate_real_costs(inputs, outputs, model)
print(f"\n{model}:")
print(f" Input: {costs['input_tokens']} tokens = ${costs['input_cost']:.4f}")
print(f" Output: {costs['output_tokens']} tokens = ${costs['output_cost']:.4f}")
print(f" Gesamt: ${costs['total_cost']:.4f}")Monitoring und Kosten-Tracking
"""
HolySheep AI Monitoring Dashboard Integration
Verfolgt API-Nutzung, Latenz und Kosten in Echtzeit
"""
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict
import threading
@dataclass
class UsageRecord:
"""Einzelner API-Aufruf"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
latency_ms: float
success: bool
error: str = ""
@dataclass
class CostTracker:
"""Trackt API-Nutzung und Kosten"""
records: List[UsageRecord] = field(default_factory=list)
lock: threading.Lock()
# HolySheep Preise 2026 ($/MTok)
PRICES = {
"gpt-5.5": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v4": 0.42
}
def log_request(
self,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
success: bool = True,
error: str = ""
):
"""Loggt einen API-Aufruf"""
record = UsageRecord(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
success=success,
error=error
)
with self.lock:
self.records.append(record)
def get_summary(self, hours: int = 24) -> Dict:
"""Generiert Zusammenfassung für letzten Zeitraum"""
cutoff = datetime.now() - timedelta(hours=hours)
with self.lock:
recent = [r for r in self.records if r.timestamp >= cutoff]
# Nach Modell aggregieren
by_model = {}
for record in recent:
if record.model not in by_model:
by_model[record.model] = {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"latencies": [],
"failures": 0
}
stats = by_model[record.model]
stats["requests"] += 1
stats["input_tokens"] += record.input_tokens
stats["output_tokens"] += record.output_tokens
stats["latencies"].append(record.latency_ms)
if not record.success:
stats["failures"] += 1
# Kosten berechnen
for model, stats in by_model.items():
price = self.PRICES.get(model, 8.00)
stats["input_cost"] = (stats["input_tokens"] / 1_000_000) * price
stats["output_cost"] = (stats["output_tokens"] / 1_000_000) * price
stats["total_cost"] = stats["input_cost"] + stats["output_cost"]