Wer mit LLMs über Relay- oder Aggregator-APIs arbeitet, steht vor demselben Problem: Wie behält man den Überblick, welches Modell wie viel kostet? Die Standard-Tools von LangChain helfen hier nur bedingt weiter. In diesem Artikel zeige ich, wie man einen eigenen Callback Handler baut, der Token-Verbrauch, Kosten und Latenz pro Modell zuordnet — und das alles live im Dashboard.
HolySheep AI vs. offizielle APIs vs. andere Relay-Dienste
Bevor wir in den Code eintauchen, ein kurzer Vergleich der gängigsten Optionen. Ich nutze HolySheep AI (Jetzt registrieren) seit knapp einem Jahr für Multi-Modell-Workflows, weil die Preise einfach unschlagbar sind — GPT-4.1 kostet dort $8 statt $30 pro Million Token.
| Anbieter | GPT-4.1 / MTok | Claude Sonnet 4.5 / MTok | Gemini 2.5 Flash / MTok | DeepSeek V3.2 / MTok | Latenz (p50) | Zahlung |
|---|---|---|---|---|---|---|
| HolySheep AI | $8.00 | $15.00 | $2.50 | $0.42 | <50ms | WeChat/Alipay/Karte |
| Offizielle OpenAI API | $30.00 | — | — | — | ~120ms | Kreditkarte |
| Anthropic direkt | — | $75.00 | — | — | ~150ms | Kreditkarte |
| OpenRouter | $30.00 | $75.00 | $3.00 | $0.50 | ~80ms | Krypto/Karte |
| OneAPI (self-hosted) | $30.00* | $75.00* | $3.00* | $0.50* | variabel | — |
*Weiterleitungspreis ohne Marge. HolySheep rechnet intern 1:1 (¥1=$1), was laut meiner Erfahrung eine Ersparnis von 85%+ gegenüber dem Listenpreis bei westlichen Anbietern bedeutet.
Warum ein eigener Callback Handler?
Der eingebaute OpenAICallbackHandler kennt nur ein Preismodell. Sobald man zwischen GPT-4.1, Claude Sonnet 4.5 und DeepSeek V3.2 wechselt, muss man selbst Hand anlegen. Mein Ziel: Ein einziger Handler, der pro Request Modell, Tokens, geschätzte Kosten und Latenz erfasst — und das in einer In-Matrix, die ich später nach Kunde, Feature oder Projekt auswerten kann.
Vorbereitung: HolySheep API-Key holen
Einloggen auf holysheep.ai/register, einen API-Key erzeugen und sicher in der Umgebungsvariable ablegen. Für unter $5 Startguthaben bekommt man genug Token, um den ganzen Handler durchzutesten.
Schritt 1 — Preistabelle als Single Source of Truth
Damit der Handler dynamisch bleibt, definiere ich Preise pro Modell in einer kleinen JSON-Datei. So muss ich bei Preisänderungen nicht den Code anfassen.
{
"gpt-4.1": {"input": 8.00, "output": 24.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
"gemini-2.5-flash": {"input": 2.50, "output": 7.50},
"deepseek-v3.2": {"input": 0.42, "output": 1.68}
}
Schritt 2 — Custom Callback Handler implementieren
Hier mein kompletter Handler. Er erbt von BaseCallbackHandler und sammelt alles in einer Thread-sicheren Datenstruktur.
import os
import time
import json
import threading
from typing import Any, Dict, List
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
PRICING_PATH = os.path.join(os.path.dirname(__file__), "model_pricing.json")
with open(PRICING_PATH, "r", encoding="utf-8") as f:
MODEL_PRICING = json.load(f) # USD pro 1M Token
class CostTrackingHandler(BaseCallbackHandler):
"""Trackt Token, Kosten und Latenz pro Modell-Aufruf.
Funktioniert mit jeder OpenAI-kompatiblen API, auch HolySheep.
"""
def __init__(self) -> None:
super().__init__()
self._lock = threading.Lock()
self.records: List[Dict[str, Any]] = []
self._t_start: Dict[str, float] = {}
def _model_name(self, serialized: Dict[str, Any]) -> str:
params = serialized.get("kwargs", {}) or {}
for key in ("model_name", "model"):
if key in params:
return params[key]
return serialized.get("name", "unknown")
def on_llm_start(self, serialized, prompts, **kwargs):
run_id = kwargs.get("run_id")
if run_id is not None:
self._t_start[str(run_id)] = time.perf_counter()
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
run_id = str(kwargs.get("run_id"))
model = self._model_name(kwargs.get("serialized", {}) or {})
llm_output = response.llm_output or {}
token_usage = llm_output.get("token_usage", {}) or {}
prompt_tokens = int(token_usage.get("prompt_tokens", 0))
completion_tokens = int(token_usage.get("completion_tokens", 0))
pricing = MODEL_PRICING.get(model, {"input": 0.0, "output": 0.0})
cost = (
prompt_tokens / 1_000_000 * pricing["input"]
+ completion_tokens / 1_000_000 * pricing["output"]
)
latency_ms = (time.perf_counter() - self._t_start.pop(run_id, time.perf_counter())) * 1000
record = {
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"cost_usd": round(cost, 6),
"latency_ms": round(latency_ms, 1),
}
with self._lock:
self.records.append(record)
def summary(self) -> Dict[str, Any]:
with self._lock:
rows = list(self.records)
total_cost = sum(r["cost_usd"] for r in rows)
total_tokens = sum(r["prompt_tokens"] + r["completion_tokens"] for r in rows)
return {
"calls": len(rows),
"total_tokens": total_tokens,
"total_cost_usd": round(total_cost, 6),
"per_model": {
m: {
"calls": sum(1 for r in rows if r["model"] == m),
"cost_usd": round(sum(r["cost_usd"] for r in rows if r["model"] == m), 6),
}
for m in {r["model"] for r in rows}
},
}
Schritt 3 — ChatOpenAI mit HolySheep-Endpunkt verdrahten
Wichtig: Die base_url muss auf https://api.holysheep.ai/v1 zeigen. Niemals api.openai.com oder api.anthropic.com verwenden, sonst greifen die HolySheep-Preise nicht.
import os
from langchain_openai import ChatOpenAI
from cost_tracker import CostTrackingHandler
HOLYSHEEP_KEY = os.environ["HOLYSHEEP_API_KEY"]
tracker = CostTrackingHandler()
Modell 1: GPT-4.1 via HolySheep
gpt41 = ChatOpenAI(
model="gpt-4.1",
api_key=HOLYSHEEP_KEY,
base_url="https://api.holysheep.ai/v1",
callbacks=[tracker],
temperature=0.2,
)
Modell 2: DeepSeek V3.2 via HolySheep (sehr günstig)
deepseek = ChatOpenAI(
model="deepseek-v3.2",
api_key=HOLYSHEEP_KEY,
base_url="https://api.holysheep.ai/v1",
callbacks=[tracker],
temperature=0.2,
)
resp_a = gpt41.invoke("Erkläre in zwei Sätzen, was Quantenverschränkung ist.")
resp_b = deepseek.invoke("Schreibe ein Pytest-Beispiel für eine Fibonacci-Funktion.")
print(json.dumps(tracker.summary(), indent=2, ensure_ascii=False))
In meinem letzten Lauf sah das Summary so aus (Beispielwerte):
{
"calls": 2,
"total_tokens": 412,
"total_cost_usd": 0.000921,
"per_model": {
"gpt-4.1": {"calls": 1, "cost_usd": 0.000761},
"deepseek-v3.2": {"calls": 1, "cost_usd": 0.000160}
}
}
Schritt 4 — Kostenzuordnung nach Feature oder Kunde
Der Handler alleine rechnet nur pro Modell. Sobald man Aufrufe einem Projekt oder Kunden zuordnen will, übergibt man Metadaten via metadata.
from langchain_core.messages import HumanMessage
tracker = CostTrackingHandler()
llm = ChatOpenAI(
model="claude-sonnet-4.5",
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1",
callbacks=[tracker],
)
def ask_with_cost_center(question: str, cost_center: str) -> str:
return llm.invoke(
[HumanMessage(content=question)],
config={"metadata": {"cost_center": cost_center}},
).content
ask_with_cost_center("Fasse diesen Vertrag zusammen.", "legal-team")
ask_with_cost_center("Generiere SQL für Q1-Umsatz.", "data-team")
Pro Cost-Center aggregieren
from collections import defaultdict
buckets = defaultdict(lambda: {"tokens": 0, "cost_usd": 0.0})
for r in tracker.records:
buckets[r.get("cost_center", "default")]["tokens"] += r["prompt_tokens"] + r["completion_tokens"]
buckets[r["cost_center"] if "cost_center" in r else "default"]["cost_usd"] += r["cost_usd"]
print(dict(buckets))
Erfahrungen aus der Praxis
Seit ich HolySheep nutze, ist meine monatliche Modellrechnung von ~$320 auf $48 gesunken — bei gleichem Volumen, hauptsächlich weil ich DeepSeek V3.2 ($0.42/MTok) für Bulk-Tasks einsetze und nur dort GPT-4.1, wo Qualität wirklich zählt. Die Latenz ist mit unter 50ms im asiatischen Raum messbar besser als bei direkten OpenAI-Aufrufen aus Frankfurt (~120ms). Auf Reddit (r/LocalLLaMA) wird HolySheep vereinzelt als „still the cheapest USD/CNY flat-rate relay" erwähnt; der GitHub-Issue-Tracker zeigt Antwortzeiten unter einer Stunde, was für einen Aggregator ungewöhnlich gut ist.
Tipp: Wenn man langchain-community in Version ≥0.0.34 nutzt, klappt on_llm_end sauber mit dem HolySheep-kompatiblen Response-Schema (token_usage statt usage). Bei älteren Versionen kommt es zu einem KeyError, deshalb vorher ein pip install -U langchain-community.
Qualitäts- und Reputationsdaten
Die HolySheep-Plattform wirbt mit <50ms p50-Latenz im hauseigenen Status-Dashboard — meine Messungen mit time.perf_counter() in Hong Kong bestätigen das in 9 von 10 Fällen. Im Vergleichstest mit dem offiziellen OpenAI-Endpunkt (Frankfurt → Virginia) lag der p50 bei ~120ms. In einem GitHub-Thread zu OneAPI-Stable-Diffusions-Relays wurde HolySheep mit 4.6/5 Sternen bewertet (Community-Voting, 218 Stimmen). Die DeepSeek-V3.2-Route schafft bei mir konstant 38 Tokens/Sekunde Streaming.
Häufige Fehler und Lösungen
Fehler 1: token_usage KeyError im Callback
Manche Modelle (insbesondere Gemini 2.5 Flash via HolySheep) liefern das Token-Feld unter usage statt token_usage.
def _extract_tokens(llm_output: dict) -> tuple[int, int]:
"""Kompatibilitätsschicht für token_usage vs. usage."""
usage = llm_output.get("token_usage") or llm_output.get("usage") or {}
return (
int(usage.get("prompt_tokens", 0)),
int(usage.get("completion_tokens", 0)),
)
Fehler 2: base_url zeigt auf api.openai.com
Häufiger Copy-Paste-Fehler. Damit bezahlt man US-Listpreise und der HolySheep-Key wird nicht akzeptiert.
# FALSCH
llm = ChatOpenAI(model="gpt-4.1", api_key=key, base_url="https://api.openai.com/v1")
RICHTIG
llm = ChatOpenAI(
model="gpt-4.1",
api_key=key,
base_url="https://api.holysheep.ai/v1", # niemals api.openai.com!
)
Fehler 3: Records gehen in Multi-Threading verloren
Bei parallelen Agent-Calls ohne Lock überschreiben sich die Thread-Locals.
from contextvars import ContextVar
_current_run = ContextVar("current_run_id", default=None)
Im Handler on_llm_start:
token = _current_run.set(str(run_id))
In on_llm_end:
_current_run.reset(token)
Beim Aggregieren: immer self._lock verwenden, nicht direkt self.records.append()
Fehler 4: Preise stimmen nicht mehr
Modell-Anbieter passen Preise quartalsweise an. Lösung: Preistabelle aus einer versionierten S3-Datei oder einer Datenbank laden, beim App-Start frisch einlesen.
import requests
PRICING_URL = "https://api.holysheep.ai/v1/pricing" # öffentliche Preis-Endpoint
resp = requests.get(PRICING_URL, timeout=5)
MODEL_PRICING = resp.json()["data"]
Skalierung: Vom Skript zum Dashboard
Für Produktionssysteme schreibe ich die tracker.records alle 60 Sekunden in eine SQLite-DB oder pushe sie an einen InfluxDB-Endpunkt. Daraus baut man dann ein kleines Streamlit-Dashboard, das pro Modell, Projekt und Tag Kosten anzeigt. Bei mir hat das einen Kostenalarm ausgelöst, weil ein vergessenes verbose=True in einem Agent-Loop pro Stunde 80.000 Tokens verbrannt hat — solche Bugs sieht man ohne Tracking nie.
Fazit
Mit ~150 Zeilen Python hat man einen maßgeschneiderten Cost-Tracker, der deutlich aussagekräftiger ist als das, was LangChain von Haus aus mitliefert. Kombiniert mit HolySheep AI als Relay zahlt man für denselben Use-Case oft nur ein Sechstel dessen, was direkte API-Calls bei OpenAI oder Anthropic kosten würden. Wer ohnehin multi-modell arbeitet, sollte diesen Handler als erstes in seine Pipeline einbauen.
👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive