TL;DR: Dieser Praxisartikel zeigt, wie Sie mit LangChain Callbacks einen robusten Monitoring-Stack für Ihre KI-Anwendungen aufbauen. Wir messen Latenz, Erfolgsquoten und Kosten in Echtzeit – mit konkretem Benchmark-Code und meinen persönlichen Erfahrungen aus 18 Monaten Produktionseinsatz.
Getestete Konfiguration: HolySheep AI (Jetzt registrieren) mit Base-URL https://api.holysheep.ai/v1, Latenz unter 50ms, 85% Kostenersparnis gegenüber Alternativen.
Warum Callbacks in LangChain unverzichtbar sind
Bei meinen ersten LangChain-Projekten habe ich Callbacks ignoriert. Das rächte sich: Unvorhersehbare Latenz-Spitzen, fehlende Token-Statistiken und keine Ahnung, warum certain Prompts fehlschlugen. Callback-Handler transformierten mein Debugging von Raterei zu präziser Observability.
Architektur: So funktioniert das LangChain Callback-System
LangChain verwendet ein ereignisbasiertes Callback-System mit zwei Kernkomponenten:
- CallbackHandler-Interface: Abstrakte Klasse mit Methoden für jede Chain-Phase
- CallbackManager: Koordiniert mehrere Handler und verteilt Events
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
from typing import Any, Dict, List, Optional
from datetime import datetime
import time
class ProductionMonitoringHandler(BaseCallbackHandler):
"""Produktionsreifer Monitoring-Handler mit Metriken-Sammlung."""
def __init__(self):
super().__init__()
self.metrics = {
"llm_calls": 0,
"total_tokens": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"total_latency_ms": 0,
"errors": 0,
"chain_starts": 0,
"chain_ends": 0,
"retries": 0
}
self.latencies = []
self.start_time = None
def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
**kwargs: Any
) -> None:
"""Wird aufgerufen, bevor LLM-Anfrage startet."""
self.metrics["llm_calls"] += 1
self.start_time = time.perf_counter()
print(f"[LLM START] Call #{self.metrics['llm_calls']} - Prompts: {len(prompts)}")
def on_llm_end(
self,
response: LLMResult,
**kwargs: Any
) -> None:
"""Wird aufgerufen, nachdem LLM-Antwort empfangen wurde."""
if self.start_time:
latency_ms = (time.perf_counter() - self.start_time) * 1000
self.latencies.append(latency_ms)
self.metrics["total_latency_ms"] += latency_ms
# Token-Zählung aus Response extrahieren
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
self.metrics["prompt_tokens"] += usage.get("prompt_tokens", 0)
self.metrics["completion_tokens"] += usage.get("completion_tokens", 0)
self.metrics["total_tokens"] += usage.get("total_tokens", 0)
print(f"[LLM END] Latenz: {latency_ms:.2f}ms - Tokens: {self.metrics['total_tokens']}")
def on_chain_start(
self,
serialized: Dict[str, Any],
inputs: Dict[str, Any],
**kwargs: Any
) -> None:
"""Wird aufgerufen, wenn Chain-Ausführung beginnt."""
self.metrics["chain_starts"] += 1
chain_name = serialized.get("name", "Unknown")
print(f"[CHAIN START] {chain_name}")
def on_chain_end(
self,
outputs: Dict[str, Any],
**kwargs: Any
) -> None:
"""Wird aufgerufen, wenn Chain-Ausführung endet."""
self.metrics["chain_ends"] += 1
print(f"[CHAIN END] Outputs: {list(outputs.keys())}")
def on_tool_start(
self,
serialized: Dict[str, Any],
input_str: str,
**kwargs: Any
) -> None:
"""Wird aufgerufen, bevor Tool ausgeführt wird."""
tool_name = serialized.get("name", "Unknown")
print(f"[TOOL START] {tool_name}: {input_str[:100]}...")
def on_tool_end(
self,
output: str,
**kwargs: Any
) -> None:
"""Wird aufgerufen, nachdem Tool ausgeführt wurde."""
print(f"[TOOL END] Output: {output[:100]}...")
def get_summary(self) -> Dict[str, Any]:
"""Gibt zusammengefasste Metriken zurück."""
avg_latency = (
self.metrics["total_latency_ms"] / self.metrics["llm_calls"]
if self.metrics["llm_calls"] > 0 else 0
)
p95_latency = sorted(self.latencies)[int(len(self.latencies) * 0.95)] if self.latencies else 0
return {
**self.metrics,
"avg_latency_ms": round(avg_latency, 2),
"p95_latency_ms": round(p95_latency, 2),
"success_rate": round(
(self.metrics["llm_calls"] - self.metrics["errors"]) /
self.metrics["llm_calls"] * 100, 2
) if self.metrics["llm_calls"] > 0 else 0
}
Integration mit HolySheep AI: Vollständiges Benchmark-Beispiel
Das folgende Beispiel zeigt die vollständige Integration mit HolySheep AI. Die API bietet unter 50ms Latenz und akzeptiert WeChat/Alipay für chinesische Entwickler:
import os
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.callbacks.base import BaseCallbackHandler
import time
HolySheep AI Konfiguration - 85%+ Ersparnis gegenüber OpenAI
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
WICHTIG: Niemals api.openai.com verwenden!
llm = ChatOpenAI(
model_name="gpt-4o", # $8/MTok bei HolySheep vs. $15 bei OpenAI
openai_api_base="https://api.holysheep.ai/v1", # Direkt hier
temperature=0.7,
max_tokens=500
)
class LatencyBenchmark