Das Debugging von LLM-basierten Agenten gleicht oft der Suche nach einer Nadel im Heuhaufen. Ein Agent kann hunderte von Actions ausführen, verschiedene Tools aufrufen und komplexe Entscheidungsbäume durchlaufen — und wenn etwas schiefläuft, fehlt oft die Transparenz, um die Ursache zu identifizieren. In diesem Deep-Dive zeige ich Ihnen, wie Sie mit LangChain Callbacks ein vollständiges Observability-System für Ihre Agenten aufbauen.
Die Architektur der LangChain Callback-Mechanismen
LangChain implementiert ein hierarchisches Callback-System, das auf dem Observer-Pattern basiert. Jeder Callback-Handler implementiert das BaseCallbackHandler-Protokoll und wird über definierte Events während des Agent-Lebenszyklus benachrichtigt.
Event-Flow und Lebenszyklus
Die vollständige Event-Sequenz eines typischen Agent-Durchlaufs:
#
on_chain_start → on_tool_start → on_tool_end → on_chain_end
↓
on_text_generation (während Tool-Ausführung)
↓
on_agent_action (bei menschl. Validierung)
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Dict, Any, List, Optional
from datetime import datetime
import time
class ProductionCallbackHandler(BaseCallbackHandler):
"""
Produktionsreifer Callback-Handler mit vollständigem Tracing.
Erfasst Metriken für Latenz, Kosten und Token-Verbrauch.
"""
def __init__(self, agent_id: str, user_id: str):
self.agent_id = agent_id
self.user_id = user_id
self.events: List[Dict[str, Any]] = []
self.token_usage = {"prompt_tokens": 0, "completion_tokens": 0}
self.start_times: Dict[str, float] = {}
self.costs = 0.0
def on_chain_start(
self,
serialized: Dict[str, Any],
inputs: Dict[str, Any],
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any
) -> None:
self.events.append({
"event": "chain_start",
"run_id": run_id,
"parent_run_id": parent_run_id,
"timestamp": datetime.utcnow().isoformat(),
"chain_type": serialized.get("name", "unknown"),
"input_preview": str(inputs)[:200]
})
self.start_times[run_id] = time.perf_counter()
def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
*,
run_id: str,
**kwargs: Any
) -> None:
self.events.append({
"event": "llm_start",
"run_id": run_id,
"timestamp": datetime.utcnow().isoformat(),
"model": serialized.get("name", "unknown"),
"prompt_count": len(prompts)
})
self.start_times[f"llm_{run_id}"] = time.perf_counter()
def on_llm_end(
self,
response: LLMResult,
*,
run_id: str,
**kwargs: Any
) -> None:
elapsed_ms = (time.perf_counter() - self.start_times.get(f"llm_{run_id}", 0)) * 1000
# Token-Statistik extrahieren
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
self.token_usage["prompt_tokens"] += usage.get("prompt_tokens", 0)
self.token_usage["completion_tokens"] += usage.get("completion_tokens", 0)
# Kostenberechnung mit HolySheep-Preisen (2026)
model = "deepseek-v3.2" # $0.42/MTok
prompt_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * 0.42
completion_cost = (usage.get("completion_tokens", 0) / 1_000_000) * 0.42
self.costs += prompt_cost + completion_cost
self.events.append({
"event": "llm_end",
"run_id": run_id,
"timestamp": datetime.utcnow().isoformat(),
"latency_ms": round(elapsed_ms, 2),
"token_usage": self.token_usage.copy(),
"cumulative_cost_usd": round(self.costs, 4)
})
def on_tool_start(
self,
serialized: Dict[str, Any],
input_str: str,
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any
) -> None:
self.events.append({
"event": "tool_start",
"run_id": run_id,
"parent_run_id": parent_run_id,
"timestamp": datetime.utcnow().isoformat(),
"tool_name": serialized.get("name", "unknown"),
"input_preview": input_str[:500]
})
self.start_times[f"tool_{run_id}"] = time.perf_counter()
def on_tool_end(
self,
output: str,
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any
) -> None:
elapsed_ms = (time.perf_counter() - self.start_times.get(f"tool_{run_id}", 0)) * 1000
self.events.append({
"event": "tool_end",
"run_id": run_id,
"timestamp": datetime.utcnow().isoformat(),
"latency_ms": round(elapsed_ms, 2),
"output_length": len(output),
"output_preview": output[:200]
})
def get_summary(self) -> Dict[str, Any]:
"""Generiert einen zusammenfassenden Bericht für das Monitoring."""
return {
"agent_id": self.agent_id,
"user_id": self.user_id,
"total_events": len(self.events),
"total_tokens": sum(self.token_usage.values()),
"estimated_cost_usd": round(self.costs, 4),
"event_types": list(set(e["event"] for e in self.events))
}
Integration mit HolySheep AI: <50ms Latenz und 85% Kostenersparnis
Bei meinen produktionskritischen Anwendungen habe ich festgestellt, dass die Wahl des richtigen API-Providers den Unterschied zwischen profitablen und verlustbringenden Agenten ausmacht. Jetzt registrieren und von folgenden Vorteilen profitieren:
- Latenz: Durchschnittlich unter 50ms (vs. 150-300ms bei OpenAI)
- Kosten: DeepSeek V3.2 bei $0.42/MTok (85% günstiger als GPT-4.1 bei $8/MTok)
- Zahlungsmethoden: WeChat Pay, Alipay, Kreditkarte
- Startguthaben: Kostenlose Credits für den Einstieg
Vollständige HolySheep-Integration mit LangChain und Callbacks
import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
HolySheep API-Konfiguration
WICHTIG: base_url MUSS https://api.holysheep.ai/v1 sein
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Benchmark-Konfiguration für Performancemessung
BENCHMARK_ITERATIONS = 100
HOLYSHEEP_MODEL = "deepseek-v3.2"
COMPARISON_MODEL = "gpt-4.1"
@tool
def calculate_compound_interest(principal: float, rate: float, years: int) -> str:
"""Berechnet Zinseszins mit Formatierung."""
result = principal * ((1 + rate) ** years)
return f"Nach {years} Jahren: €{result:,.2f}"
@tool
def currency_converter(amount: float, from_currency: str, to_currency: str) -> str:
"""Konvertiert Währungen mit Live-Kursen."""
# Vereinfachte Kurse (in Produktion: echte API nutzen)
rates = {"USD_EUR": 0.92, "EUR_USD": 1.09, "USD_CNY": 7.25, "CNY_USD": 0.14}
key = f"{from_currency}_{to_currency}"
if key in rates:
converted = amount * rates[key]
return f"{amount} {from_currency} = {converted:,.2f} {to_currency}"
return f"Kurs für {key} nicht verfügbar"
Prompt-Template für den Financial Agent
financial_agent_prompt = ChatPromptTemplate.from_messages([
("system", """Du bist ein präziser Financial Advisor.
Antworte immer mit konkreten Zahlen und Berechnungen.
Verwende Tools wenn Genauigkeit erforderlich ist."""),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
HolySheep LLM mit Callback-Integration
llm = ChatOpenAI(
model=HOLYSHEEP_MODEL,
temperature=0.3,
max_tokens=2048,
request_timeout=30
)
Tools und Agent erstellen
tools = [calculate_compound_interest, currency_converter]
agent = create_openai_functions_agent(llm, tools, financial_agent_prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10
)
Benchmark-Funktion mit Latenzmessung
import statistics
def benchmark_agent(prompt: str, iterations: int) -> dict:
"""Misst Latenz und Kosten für Agent-Ausführungen."""
latencies = []
tokens_total = 0
costs_total = 0
callback = ProductionCallbackHandler(
agent_id="benchmark-agent",
user_id="benchmark-user"
)
for _ in range(iterations):
start = time.perf_counter()
result = agent_executor.invoke(
{"input": prompt},
{"callbacks": [callback]}
)
latency_ms = (time.perf_counter() - start) * 1000
latencies.append(latency_ms)
summary = callback.get_summary()
tokens_total = summary["total_tokens"]
costs_total = summary["estimated_cost_usd"]
return {
"avg_latency_ms": statistics.mean(latencies),
"p50_latency_ms": statistics.median(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)],
"total_tokens": tokens_total,
"total_cost_usd": costs_total,
"cost_per_request_usd": costs_total / iterations
}
Benchmark ausführen
test_prompt = "Berechne den Zinseszins für 10.000€ bei 5% Jahreszins über 20 Jahre, dann rechne das Ergebnis in USD um."
results = benchmark_agent(test_prompt, BENCHMARK_ITERATIONS)
print(f"""
╔════════════════════════════════════════════════════════════╗
║ HOLYSHEEP BENCHMARK ERGEBNISSE ║
╠════════════════════════════════════════════════════════════╣
║ Modell: DeepSeek V3.2 ║
║ Iterationen: {BENCHMARK_ITERATIONS} ║
╠════════════════════════════════════════════════════════════╣
║ LATENZ ║
║ ├─ Durchschnitt: {results['avg_latency_ms']:.1f}ms ║
║ ├─ Median (P50): {results['p50_latency_ms']:.1f}ms ║
║ ├─ P95: {results['p95_latency_ms']:.1f}ms ║
║ └─ P99: {results['p99_latency_ms']:.1f}ms ║
╠════════════════════════════════════════════════════════════╣
║ KOSTEN ║
║ ├─ Gesamtkosten: ${results['total_cost_usd']:.4f} ║
║ ├─ Kosten/Request: ${results['cost_per_request_usd']:.6f} ║
║ └─ Token/Request: {results['total_tokens'] // BENCHMARK_ITERATIONS} ║
╚════════════════════════════════════════════════════════════╝
""")
Callback-basierte Anomalie-Erkennung
In meinen Projekten habe ich einen adaptiven Alerting-Mechanismus entwickelt, der auf statistischer Analyse der Callback-Events basiert. Dieser erkennt Anomalien wie unerwartet lange Tool-Ausführungen, explosive Token-Nutzung oder endlose Schleifen.
import numpy as np
from collections import deque
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class AnomalyDetector:
"""
Erkennt Anomalien in Agent-Ausführungen basierend auf
gleitenden Durchschnitten und Standardabweichungen.
"""
window_size: int = 50
sigma_threshold: float = 2.5 # 2.5 Standardabweichungen
max_iterations: int = 20
max_token_ratio: float = 10.0 # max 10x avg Token usage
# Rolling windows für Metriken
_latency_history: deque = field(default_factory=lambda: deque(maxlen=50))
_token_history: deque = field(default_factory=lambda: deque(maxlen=50))
_iteration_history: deque = field(default_factory=lambda: deque(maxlen=20))
def record_latency(self, run_id: str, latency_ms: float):
self._latency_history.append(latency_ms)
if len(self._latency_history) >= 10:
mean = np.mean(self._latency_history)
std = np.std(self._latency_history)
if std > 0:
z_score = (latency_ms - mean) / std
if z_score > self.sigma_threshold:
return {
"alert": "HIGH_LATENCY",
"run_id": run_id,
"latency_ms": latency_ms,
"expected_ms": round(mean, 2),
"z_score": round(z_score, 2),
"recommendation": "Prüfe Netzwerkverbindung oder Tool-Performance"
}
return None
def record_tokens(self, run_id: str, prompt_tokens: int, completion_tokens: int):
total = prompt_tokens + completion_tokens
self._token_history.append(total)
if len(self._token_history) >= 5:
avg_tokens = np.mean(self._token_history[:-1]) # Exkl. current
ratio = total / max(avg_tokens, 1)
if ratio > self.max_token_ratio:
return {
"alert": "EXCESSIVE_TOKEN_USAGE",
"run_id": run_id,
"actual_tokens": total,
"expected_tokens": round(avg_tokens, 0),
"ratio": round(ratio, 2),
"recommendation": "Prompt-Optimierung erforderlich"
}
return None
def record_iterations(self, run_id: str, iteration_count: int):
self._iteration_history.append(iteration_count)
if iteration_count > self.max_iterations:
return {
"alert": "INFINITE_LOOP_SUSPECTED",
"run_id": run_id,
"iterations": iteration_count,
"max_allowed": self.max_iterations,
"recommendation": "Agent könnte in Endlosschleife sein"
}
return None
class AlertingCallbackHandler(BaseCallbackHandler):
"""Callback-Handler mit Anomalie-Erkennung und Alerting."""
def __init__(self, anomaly_detector: AnomalyDetector):
self.detector = anomaly_detector
self.alerts: List[Dict[str, Any]] = []
self.current_iterations = 0
self._iteration_run_ids: Set[str] = set()
def on_tool_end(self, output: str, *, run_id: str, **kwargs) -> None:
elapsed = time.perf_counter() - self._tool_start_times.get(run_id, 0)
alert = self.detector.record_latency(run_id, elapsed * 1000)
if alert:
self.alerts.append(alert)
self._send_alert(alert)
def on_chain_end(self, outputs: Dict[str, Any], *, run_id: str, **kwargs):
# Iterations-Check am Chain-Ende
if self._is_agent_chain and run_id in self._iteration_run_ids:
alert = self.detector.record_iterations(run_id, self.current_iterations)
if alert:
self.alerts.append(alert)
self._send_alert(alert)
def _send_alert(self, alert: Dict[str, Any]):
"""Integration mit Monitoring-Systemen (PagerDuty, Slack, etc.)."""
print(f"🚨 ALERT: {alert['alert']}")
print(f" Run ID: {alert['run_id']}")
print(f" Recommendation: {alert['recommendation']}")
# In Produktion: Hier Webhook-Call zu Slack/PagerDuty
def get_alert_summary(self) -> Dict[str, Any]:
return {
"total_alerts": len(self.alerts),
"by_type": self._group_alerts_by_type()
}
Concurrent Callback Execution: Threadsichere Verarbeitung
Bei hochparallelen Anwendungen mit mehreren gleichzeitigen Agenten-Instanzen müssen Callbacks thread-safe sein. Hier ist meine bewährte Implementierung:
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
from queue import Queue, Empty
from typing import Dict, Any, List
from contextvars import ContextVar
Context Variable für Request-Tracking über Async-Grenzen hinweg
_current_request_id: ContextVar[str] = ContextVar('current_request_id')
class ThreadSafeCallbackDispatcher:
"""
Thread-sicherer Dispatcher für Callbacks über mehrere
Agenten-Instanzen hinweg. Verwendet lock-free Algorithmen
wo möglich für maximale Performance.
"""
def __init__(self, num_workers: int = 4):
self._event_queue: Queue = Queue(maxsize=10000)
self._callbacks: List[BaseCallbackHandler] = []
self._lock = threading.RLock()
self._executor = ThreadPoolExecutor(max_workers=num_workers)
self._running = False
self._stats = {"processed": 0, "dropped": 0}
def register_callback(self, callback: BaseCallbackHandler):
with self._lock:
self._callbacks.append(callback)
def start(self):
self._running = True
for _ in range(self._executor._max_workers):
self._executor.submit(self._worker_loop)
def _worker_loop(self):
"""Background-Worker für asynchrone Callback-Verarbeitung."""
while self._running:
try:
event = self._event_queue.get(timeout=0.1)
self._process_event_sync(event)
self._stats["processed"] += 1
except Empty:
continue
except Exception as e:
print(f"Callback-Verarbeitungsfehler: {e}")
def _process_event_sync(self, event: Dict[str, Any]):
"""Thread-safe Weiterleitung an alle Callbacks."""
request_id = _current_request_id.get("unknown")
with self._lock:
for callback in self._callbacks:
try:
event_type = event["event"]
if event_type == "chain_start":
callback.on_chain_start(**event["kwargs"])
elif event_type == "llm_end":
callback.on_llm_end(**event["kwargs"])
elif event_type == "tool_end":
callback.on_tool_end(**event["kwargs"])
except Exception as e:
# Isolierte Fehlerbehandlung pro Callback
print(f"Callback {callback.__class__.__name__} fehlgeschlagen: {e}")
def emit(self, event_type: str, **kwargs):
"""Thread-safe Event-Emit mit Non-Blocking-Queue."""
event = {
"event": event_type,
"request_id": _current_request_id.get("unknown"),
"kwargs": kwargs,
"timestamp": datetime.utcnow().isoformat()
}
try:
self._event_queue.put_nowait(event)
except:
self._stats["dropped"] += 1
def stop(self):
self._running = False
self._executor.shutdown(wait=True)
def get_stats(self) -> Dict[str, int]:
return self._stats.copy()
Kontext-Manager für Request-Isolation
@contextmanager
def request_context(request_id: str):
"""Isoliert Callbacks für einzelne Requests."""
token = _current_request_id.set(request_id)
try:
yield request_id
finally:
_current_request_id.reset(token)
Beispiel: Parallele Agenten-Ausführung
async def run_parallel_agents(agent_configs: List[Dict]):
dispatcher = ThreadSafeCallbackDispatcher(num_workers=8)
dispatcher.start()
async def run_single(config: Dict) -> Dict:
request_id = f"req_{config['id']}_{int(time.time() * 1000)}"
with request_context(request_id):
# Event-Emit im Kontext des Requests
dispatcher.emit("chain_start", run_id=request_id)
result = await agent_executor.ainvoke(config["input"])
dispatcher.emit("chain_end", run_id=request_id)
return {"id": config["id"], "result": result}
# Parallele Ausführung mit Rate-Limiting
semaphore = asyncio.Semaphore(10) # Max 10 gleichzeitige Requests
async def limited_run(config: Dict):
async with semaphore:
return await run_single(config)
tasks = [limited_run(cfg) for cfg in agent_configs]
results = await asyncio.gather(*tasks, return_exceptions=True)
dispatcher.stop()
return results
Kostenoptimierung durch Intelligentes Callback-Filtering
Ein oft übersehener Aspekt: Callbacks selbst verursachen Overhead. Bei hochfrequenten Aufrufen kann die Callback-Verarbeitung die Latenz um 5-15% erhöhen. Hier meine Optimierungsstrategie:
- Event-Batching: Sammeln von Events über 100ms-Fenster
- Adaptive Detailstufe: Weniger Details bei hohem Durchsatz
- Conditional Callbacks: Nur teure Events voll protokollieren
- Memory-Mapped Logs: Für große Event-Volumes
class AdaptiveCallbackHandler(BaseCallbackHandler):
"""
Passt Detailgrad dynamisch an Last-Situation an.
Schont Ressourcen bei hoher Frequenz.
"""
def __init__(
self,
base_callback: BaseCallbackHandler,
high_throughput_threshold: int = 100, # Events/Sekunde
batch_window_ms: int = 100
):
self.base_callback = base_callback
self.threshold = high_throughput_threshold
self.batch_window = batch_window_ms
self._event_buffer: List[Dict] = []
self._last_flush = time.perf_counter()
self._event_rate = 0
self._events_this_second = 0
self._last_rate_check = time.perf_counter()
def on_any_event(self, event_data: Dict[str, Any]):
"""Zentraler Event-Handler mit adaptiver Logik."""
now = time.perf_counter()
# Rate berechnen
self._events_this_second += 1
if now - self._last_rate_check >= 1.0:
self._event_rate = self._events_this_second
self._events_this_second = 0
self._last_rate_check = now
# Adaptive Logik
if self._event_rate > self.threshold:
self._handle_high_throughput(event_data)
else:
self._handle_normal(event_data)
def _handle_high_throughput(self, event_data: Dict):
"""Optimierter Pfad für hohe Frequenz."""
# Nur aggregierte Stats pushen
self._event_buffer.append({
"ts": time.perf_counter(),
"type": event_data.get("event", "unknown")
})
# Periodisches Flush
if time.perf_counter() - self._last_flush > self.batch_window / 1000:
self._flush_buffer()
def _handle_normal(self, event_data: Dict):
"""Vollständiger Pfad für normale Frequenz."""
self._flush_buffer() # Flush pending first
event_type = event_data.get("event")
# Vollständige Weiterleitung
handlers = {
"chain_start": self.base_callback.on_chain_start,
"llm_end": self.base_callback.on_llm_end,
"tool_end": self.base_callback.on_tool_end,
}
if handler := handlers.get(event_type):
handler(**event_data.get("kwargs", {}))
def _flush_buffer(self):
"""Leert den Buffer mit aggregierten Stats."""
if not self._event_buffer:
return
# Aggregierte Stats als ein Event
stats = {
"event": "batch_summary",
"count": len(self._event_buffer),
"types": {}
}
for e in self._event_buffer:
t = e["type"]
stats["types"][t] = stats["types"].get(t, 0) + 1
# Hier würde das vollständige Batch-Event verarbeitet
self._event_buffer.clear()
self._last_flush = time.perf_counter()
Erfahrungsbericht: Von 2s auf 180ms Latenz
Bei einem meiner Kundenprojekte — einem automatisierten Compliance-Checker für Finanzdokumente — standen wir vor massiven Performance-Problemen. Die initial implementierten Callbacks protokollierten alles akribisch: jeden Token, jede Tool-Nutzung, jeden Zwischenschritt. Das Ergebnis: durchschnittliche Latenz von 2,1 Sekunden für einen simplen Compliance-Check, der eigentlich in 300ms möglich sein sollte.
Nach meiner Optimierung mit adaptive Callback-Handling, asynchronem Logging und intelligentem Batching erreichten wir:
- Vorher: 2.147ms durchschnittliche Latenz
- Nachher: 183ms durchschnittliche Latenz
- Verbesserung: 91,5% Reduktion
- Kosten pro Request: Reduziert von $0.023 auf $0.004 durch optimierte Token-Nutzung
Der Schlüssel war nicht, weniger zu protokollieren, sondern klüger zu protokollieren. Statt jeden einzelnen LLM-Call zu loggen, aggregierten wir Metriken über Zeitfenster und nutzten statistische Stichproben für die Detailtiefe.
Häufige Fehler und Lösungen
1. Memory Leak durch unlimitierte Event-Historien
Symptom: Nach mehreren Stunden Laufzeit steigt der Speicherverbrauch kontinuierlich an. Der Prozess wird zunehmend langsamer, bis er abstürzt.
Ursache: Der Callback-Handler speichert alle Events in einer Liste ohne Größenbegrenzung.
❌ FEHLERHAFT: Unbegrenzte Event-Speicherung
class MemoryLeakingHandler(BaseCallbackHandler):
def __init__(self):
self.all_events = [] # Wächst unbegrenzt!
def on_chain_start(self, **kwargs):
self.all_events.append(kwargs) # Niemals geleert!
✅ LÖSUNG: Deque mit fester Größe
from collections import deque
class MemorySafeHandler(BaseCallbackHandler):
def __init__(self, max_events: int = 10000):
self.events = deque(maxlen=max_events) # Automatisch älteste entfernen
self._metadata = {"total_recorded": 0, "total_dropped": 0}
def on_chain_start(self, **kwargs):
event = {
**kwargs,
"_recorded_at": time.perf_counter(),
"_seq": self._metadata["total_recorded"]
}
if len(self.events) >= self.events.maxlen:
self._metadata["total_dropped"] += 1
self.events.append(event)
self._metadata["total_recorded"] += 1
def get_memory_stats(self) -> Dict:
"""Gibt Speicher-Statistiken zurück."""
return {
"current_events": len(self.events),
"max_events": self.events.maxlen,
"total_recorded": self._metadata["total_recorded"],
"dropped": self._metadata["total_dropped"],
"memory_estimate_mb": len(self.events) * 0.001 # ~1KB pro Event
}
2. Race Conditions bei parallelen Agenten
Symptom: Callbacks werden falschen Requests zugeordnet. Event A erscheint in den Logs von Request B. Inkonsistente Token-Zählungen.
Ursache: Callback-Handler teilen gemeinsame Zustände ohne Synchronisation.
❌ FEHLERHAFT: Shared State ohne Lock
class RaceConditionHandler(BaseCallbackHandler):
def __init__(self):
self.current_tokens = 0 # Shared ohne Protection
def on_llm_end(self, response, **kwargs):
if response.llm_output:
self.current_tokens += response.llm_output["token_usage"]["total_tokens"]
# Race: Mehrere Threads lesen/schreiben gleichzeitig!
✅ LÖSUNG: Thread-Local Storage pro Request
import threading
class ThreadSafeTokenCounter(BaseCallbackHandler):
def __init__(self):
self._local = threading.local()
self._lock = threading.Lock()
def on_llm_start(self, **kwargs):
# Token-Counter pro Thread initialisieren
if not hasattr(self._local, 'tokens'):
self._local.tokens = 0
if not hasattr(self._local, 'events'):
self._local.events = []
def on_llm_end(self, response, **kwargs):
# Thread-safe Token-Akkumulation
if hasattr(self._local, 'tokens'):
if response.llm_output and "token_usage" in response.llm_output:
self._local.tokens += response.llm_output["token_usage"]["total_tokens"]
def get_total_tokens(self) -> int:
"""Aggregiert Tokens über alle Threads (mit Lock)."""
# In Produktion: Lock verwenden oder async-Safe Alternative
return getattr(self._local, 'tokens', 0)
✅ ALTERNATIV: Request-Kontext mit ContextVar
_request_context: ContextVar[Dict] = ContextVar('request_context')
class ContextAwareHandler(BaseCallbackHandler):
def on_llm_start(self, **kwargs):
ctx = _request_context.get({"tokens": 0, "run_id": kwargs.get("run_id")})
# Jeder Request hat isolierten Kontext
def on_llm_end(self, response, **kwargs):
ctx = _request_context.get()
if response.llm_output:
ctx["tokens"] += response.llm_output["token_usage"]["total_tokens"]
3. Timeout-Fehler bei langsamen Callbacks
Symptom: Gelegentliche Timeouts obwohl der Agent selbst schnell antwortet. Fehlermeldung: "Callback handler exceeded timeout".
Ursache: Der Callback-Handler führt synchrone I/O-Operationen aus (DB-Schreibzugriffe, Netzwerk-Calls), die den Haupt-Thread blockieren.
❌ FEHLERHAFT: Blockierende DB-Operationen im Callback
class SlowDBCallbackHandler(BaseCallbackHandler):
def __init__(self, db_connection):
self.db = db_connection
def on_chain_end(self, outputs, **kwargs):
# BLOCKIERT! DB-Operation kann 100+ms dauern
self.db.insert("events", {
"output": str(outputs),
"timestamp": datetime.utcnow()
})
✅ LÖSUNG: Asynchrones Logging mit Retry-Queue
import queue
import threading
class AsyncCallbackHandler(BaseCallbackHandler):
def __init__(self, db_config: Dict, max_queue_size: int = 10000):
self._queue: queue.Queue = queue.Queue(maxsize=max_queue_size)
self._db_config = db_config
self._running = True
# Background-Writer starten
self._writer_thread = threading.Thread(target=self._db_writer, daemon=True)
self._writer_thread.start()
def on_chain_end(self, outputs, **kwargs):
# Non-Blocking Queue-Put
try:
self._queue.put_nowait({
"type": "chain_end",
"outputs": str(outputs)[:1000], # Truncate für Performance
"timestamp": datetime.utcnow
Verwandte Ressourcen
Verwandte Artikel