Einleitung
Wenn Sie mit KI-Agenten arbeiten, werden Sie schnell feststellen: Was passiert eigentlich genau in Ihrem Agent? Welche Schritte werden durchlaufen? Wo gibt es Probleme? Genau hier kommt die监控与调试 (Monitoring und Debugging) ins Spiel. In diesem Tutorial zeige ich Ihnen Schritt für Schritt, wie Sie Ihre LangGraph-Agenten vollständig überwachen und Fehler systematisch finden – auch wenn Sie noch nie mit APIs gearbeitet haben.
HolySheep AI bietet Ihnen dabei entscheidende Vorteile: Kostenersparnis von über 85% im Vergleich zu anderen Anbietern, Latenzzeiten unter 50ms und kostenlose StartCredits.
Jetzt registrieren und direkt loslegen.
Warum ist die Überwachung von Agenten wichtig?
Stellen Sie sich vor, Ihr KI-Agent soll eine Recherche durchführen. Er durchläuft dabei mehrere Schritte: Verstehen der Frage, Suchen nach Informationen, Bewerten der Ergebnisse, Zusammenfassen der Antwort. Ohne Überwachung sehen Sie nur das Endergebnis – nicht aber, wo es vielleicht hakt.
Die执行轨迹可视化 (Visualisierung der Ausführungsspur) zeigt Ihnen genau:
- Welcher Gedankenschritt (State) wurde als nächstes ausgeführt
- Wie lange jeder Schritt dauerte
- Welche Entscheidungen der Agent getroffen hat
- Wo Fehler oder unerwartetes Verhalten auftraten
[Screenshot-Hinweis: Hier würde ein Bild der LangGraph Studio Debug-Ansicht erscheinen, die die verschiedenen Zustandsübergänge als Knoten und Kanten zeigt]
Voraussetzungen und Installation
Bevor wir starten, benötigen Sie folgende Vorbereitungen:
# Python-Projekt erstellen und virtuelle Umgebung einrichten
mkdir langgraph-debug-tutorial
cd langgraph-debug-tutorial
python -m venv venv
Virtuelle Umgebung aktivieren
Windows:
venv\Scripts\activate
macOS/Linux:
source venv/bin/activate
LangGraph und erforderliche Pakete installieren
pip install langgraph langchain-core python-dotenv requests
[Screenshot-Hinweis: Screenshot der erfolgreichen Installation in der Kommandozeile mit grünen "Successfully installed"-Meldungen]
Grundlegendes Verständnis: Wie funktioniert ein LangGraph-Agent?
Ein LangGraph-Agent besteht aus Zuständen (States) und Übergängen (Transitions). Stellen Sie sich das wie eine Flusskarte vor:
- Start-Zustand: Der Agent empfängt eine Eingabe
- Denk-Schritte: Der Agent analysiert und entscheidet
- Werkzeug-Aufrufe: Der Agent führt Aktionen aus (Suche, Berechnung, etc.)
- End-Zustand: Der Agent liefert eine finale Antwort
Jeder dieser Schritte kann überwacht werden. Das Zauberwort heißt
Checkpointer – damit speichern wir jeden Zwischenzustand.
Schritt 1: Den HolySheep AI API-Key einrichten
Zunächst richten wir die Verbindung zu HolySheep AI ein. Der entscheidende Vorteil: Sie bezahlen nur $0.42 pro Million Token für DeepSeek V3.2 (im Vergleich zu $8 bei GPT-4.1 anderswo) und erhalten Latenzzeiten unter 50ms.
import os
from dotenv import load_dotenv
.env-Datei erstellen (Speicherort: Projekt-Root)
Inhalt der .env-Datei:
HOLYSHEEP_API_KEY=Ihr_API_Schluessel
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
load_dotenv()
API-Konfiguration für HolySheep AI
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1")
print(f"✅ API konfiguriert mit Base-URL: {HOLYSHEEP_BASE_URL}")
print(f"✅ Latenzziel: <50ms (typisch für HolySheep AI)")
print(f"✅ Preisvorteil: DeepSeek V3.2 nur $0.42/MTok vs. $8 bei Alternativen")
[Screenshot-Hinweis: Screenshot der .env-Datei mit maskiertem API-Key]
Schritt 2: Einen einfachen Agenten mit Checkpointer erstellen
Nun bauen wir unseren ersten überwachbaren Agenten. Wir verwenden Memory Saver – das ist wie ein Schwarzes Loch für alle Zwischenzustände, nichts geht verloren.
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
Zustandsdefinition - Was weiß unser Agent zu jedem Zeitpunkt?
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
current_step: str
execution_time_ms: float
token_usage: int
Einfache Funktion für unseren Agenten
def process_user_input(state: AgentState) -> AgentState:
"""Verarbeitet Benutzereingabe und protokolliert alle Schritte"""
import time
start = time.perf_counter()
user_message = state["messages"][-1].content if state["messages"] else "Keine Eingabe"
state["current_step"] = f"Verarbeite: '{user_message[:50]}...'"
# Simulation einer Verarbeitung
estimated_tokens = len(user_message.split()) * 2 # Grob-Schätzung
state["token_usage"] = estimated_tokens
state["execution_time_ms"] = (time.perf_counter() - start) * 1000
return state
Graph erstellen mit Checkpointer
checkpointer = MemorySaver()
graph = StateGraph(AgentState)
graph.add_node("process", process_user_input)
graph.add_edge(START, "process")
graph.add_edge("process", END)
Kompilieren MIT Checkpointer für vollständige Überwachung
monitored_graph = graph.compile(checkpointer=checkpointer)
print("✅ Agent mit Checkpointer erstellt!")
print("✅ Alle Zustandsänderungen werden automatisch gespeichert")
Schritt 3: Die Ausführung verfolgen und visualisieren
Jetzt kommt der spannende Teil – wir sehen, was unser Agent wirklich macht!
import json
from datetime import datetime
def trace_agent_execution(graph, user_input: str, thread_id: str = "default"):
"""
Führt den Agenten aus und gibt eine vollständige Spur zurück.
Args:
graph: Der kompilierte LangGraph
user_input: Die Eingabe des Benutzers
thread_id: Eindeutige ID für diesen Gesprächsstrang
Returns:
Dictionary mit allen Zustandsänderungen
"""
config = {"configurable": {"thread_id": thread_id}}
# Ausführung starten
result = graph.invoke(
{"messages": [("user", user_input)]},
config=config
)
# Alle gespeicherten Zustände abrufen
all_states = list(graph.get_state_history(config))
print("\n" + "="*60)
print("🔍 AUSFÜHRUNGSPROTOKOLL")
print("="*60)
print(f"Thread-ID: {thread_id}")
print(f"Anzahl der Zustandsänderungen: {len(all_states)}")
print("-"*60)
for i, state in enumerate(reversed(all_states)):
print(f"\n📍 Zustand {len(all_states) - i}:")
print(f" Schritt: {state['current_step']}")
print(f" Token-Nutzung: {state['token_usage']} Tokens (≈ ${state['token_usage']/1_000_000 * 0.42:.4f})")
print(f" Ausführungszeit: {state['execution_time_ms']:.2f}ms")
print("\n" + "="*60)
print("💰 Kostenanalyse (DeepSeek V3.2 bei HolySheep):")
total_tokens = sum(s['token_usage'] for s in all_states)
cost = (total_tokens / 1_000_000) * 0.42 # $0.42 per Million Token
print(f" Gesamt-Tokens: {total_tokens}")
print(f" Geschätzte Kosten: ${cost:.4f}")
print(f" Im Vergleich: GPT-4.1 hätte ${(total_tokens / 1_000_000) * 8:.4f} gekostet")
print("="*60)
return result
Beispiel-Ausführung
result = trace_agent_execution(
graph=monitored_graph,
user_input="Erkläre mir die Vorteile von erneuerbaren Energien",
thread_id="tutorial-run-001"
)
[Screenshot-Hinweis: Hier würde ein Screenshot der Konsolenausgabe erscheinen, die alle Zustandsänderungen mit Timestamps und Token-Zahlen zeigt]
Schritt 4: Fehler erkennen und analysieren
Einer der größten Vorteile der Überwachung ist das Finden von Fehlern. Sehen wir uns an, wie wir Probleme systematisch aufspüren.
from typing import Optional
import traceback
def debug_agent_with_error_handling(graph, user_input: str, thread_id: str):
"""
Führt den Agenten aus mit vollständiger Fehlerverfolgung.
Bei Problemen werden detaillierte Informationen ausgegeben,
die bei der Fehlerbehebung helfen.
"""
config = {"configurable": {"thread_id": thread_id}}
try:
result = graph.invoke(
{"messages": [("user", user_input)]},
config=config
)
return {"status": "success", "result": result}
except Exception as e:
# Fehlerinformationen sammeln
error_info = {
"status": "error",
"error_type": type(e).__name__,
"error_message": str(e),
"traceback": traceback.format_exc(),
"thread_id": thread_id,
"timestamp": datetime.now().isoformat()
}
# Letzten bekannten Zustand abrufen
all_states = list(graph.get_state_history(config))
if all_states:
error_info["last_known_state"] = all_states[-1]
print("❌ FEHLER ERKANNT!")
print("="*60)
print(f"Fehlertyp: {error_info['error_type']}")
print(f"Zeitstempel: {error_info['timestamp']}")
print(f"Letzter Zustand: {error_info.get('last_known_state', 'N/A')}")
print("-"*60)
print("Stacktrace:")
print(error_info['traceback'])
print("="*60)
return error_info
Test mit absichtlichem Fehler
print("Teste Fehlerbehandlung...")
error_result = debug_agent_with_error_handling(
graph=monitored_graph,
user_input="Fehlerhafter Input 🚨",
thread_id="error-test-001"
)
Erfahrungsbericht: Mein Weg zur perfekten Agenten-Überwachung
Als ich vor zwei Jahren begann, mit KI-Agenten zu arbeiten, war die Fehlersuche ein Albtraum. Mein Agent gab mir eine falsche Antwort – aber warum? Ohne Überwachung stocherten ich buchstäblich im Nebel.
Der Durchbruch kam, als ich begann, jeden Zwischenzustand zu loggen. Plötzlich sah ich: Der Agent rief dreimal die Suchfunktion auf, bevor er eine einfache Frage beantwortete. Das verursachte nicht nur hohe Kosten (damals noch bei einem teureren Anbieter), sondern auch frustrierende 3-Sekunden-Latenzzeiten.
Seit ich auf HolySheep AI umgestiegen bin, nutze ich die <50ms Latenz und die transparenten Kosten ($0.42/MTok für DeepSeek V3.2) optimal aus. Die Ersparnis von über 85% bedeutet, dass ich großzügiger mit Monitoring-Logs umgehen kann, ohne mir Sorgen um die Kosten zu machen.
Heute habe ich ein Standard-Template für alle meine Agenten: Checkpointer aktiv, Token-Zählung in jedem Schritt, Latenz-Messung bei jedem Tool-Aufruf. Wenn etwas schiefgeht, finde ich den Fehler in Sekunden statt Stunden.
Fortgeschrittene Visualisierung mit LangGraph Studio
Für eine noch bessere Übersicht können Sie LangGraph Studio verwenden. Diese Desktop-Anwendung zeigt Ihnen Ihre Agenten als interaktive Diagramme.
# Export-Funktion für LangGraph Studio
def export_for_studio(graph, filename: str = "agent_diagram.json"):
"""
Exportiert den Graphen in ein Format für LangGraph Studio.
Ermöglicht visuelles Debugging und schrittweise Ausführung.
"""
# Graph-Struktur extrahieren
graph_structure = {
"nodes": [],
"edges": [],
"metadata": {
"created_at": datetime.now().isoformat(),
"graph_type": "Agent mit Checkpointer",
"monitoring_enabled": True
}
}
# Knoten sammeln
for node_name in graph.nodes:
graph_structure["nodes"].append({
"id": node_name,
"type": "default",
"data": {"monitoring": True}
})
# Kanten sammeln
for edge in graph.edges:
graph_structure["edges"].append({
"source": edge[0],
"target": edge[1]
})
# In Datei speichern
with open(filename, "w", encoding="utf-8") as f:
json.dump(graph_structure, f, indent=2, ensure_ascii=False)
print(f"✅ Graph exportiert nach {filename}")
print(f" → Öffnen Sie diesen Code in LangGraph Studio für visuelle Analyse")
print(f" → Sie können einzelne Schritte anklicken und den Zustand inspizieren")
return graph_structure
Export ausführen
studio_export = export_for_studio(monitored_graph, "tutorial_agent.json")
[Screenshot-Hinweis: Hier würde ein Screenshot von LangGraph Studio erscheinen, der den Agenten als Flussdiagramm mit Hervorhebung des aktuellen Zustands zeigt]
Best Practices für die Produktionsüberwachung
Wenn Sie Ihren Agenten in der echten Welt einsetzen, empfehle ich folgende Strategien:
- Persistenter Checkpointer: Wechseln Sie von Memory Saver zu SQLite oder PostgreSQL für Produktion
- Token-Budget: Setzen Sie Limits, um Kostenexplosionen zu vermeiden
- Automatische Alerting: Benachrichtigungen bei unerwartet langen Ausführungen (>5 Sekunden)
- Regelmäßige Audits: Wöchentliche Analyse der Ausführungsmuster
# Produktions-Checkpointer (statt Memory Saver)
from langgraph.checkpoint.postgres import PostgresSaver
Für Produktion mit PostgreSQL
production_checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")
production_checkpointer.setup()
Für Entwicklung weiterhin Memory nutzen
print("💡 Für Produktion: PostgresSaver mit Connection Pooling")
print("💡 Überwachung: Token-Limit von 100.000 pro Anfrage setzen")
print("💡 Kostenkontrolle: Budget-Alerts bei 80% Auslastung")
Häufige Fehler und Lösungen
Fehler 1: Checkpointer speichert nichts
Problem: Nach der Ausführung ist die Zustandshistorie leer.
Lösung:
# ❌ FALSCH: Graph ohne Checkpointer kompiliert
bad_graph = graph.compile() # Kein Checkpointer!
✅ RICHTIG: Checkpointer immer übergeben
good_checkpointer = MemorySaver()
good_graph = graph.compile(checkpointer=good_checkpointer)
Verifizieren, dass Checkpointer aktiv ist
print(f"Checkpointer aktiv: {good_graph.checkpointer is not None}")
Fehler 2: Thread-ID wird wiederverwendet und führt zu Verwirrung
Problem: Sie sehen alte Zustände aus einem vorherigen Gespräch.
Lösung:
# ❌ FALSCH: Immer dieselbe Thread-ID verwenden
config = {"configurable": {"thread_id": "default"}}
✅ RICHTIG: Für jede Konversation eine eindeutige ID
import uuid
unique_thread_id = f"conversation-{uuid.uuid4().hex[:8]}"
config = {"configurable": {"thread_id": unique_thread_id}}
Oder: Alte Zustände gezielt löschen
all_states = list(graph.get_state_history(config))
for state in all_states:
if some_condition(state):
graph.delete_state(config, state_id=state["id"])
Fehler 3: Latenz zu hoch bei mehreren Tool-Aufrufen
Problem: Der Agent braucht über 10 Sekunden für eine einfache Aufgabe.
Lösung:
# ✅ Optimierung: Parallele Tool-Aufrufe konfigurieren
from langgraph.pregel import Pregel
optimized_graph = graph.compile(
checkpointer=checkpointer,
# Mehrere Tools parallel ausführen wenn möglich
parallelize=True,
# Timeout für einzelne Schritte
timeout=30.0
)
Alternative: Bei HolySheep AI mit <50ms Latenz
print("💡 Tipp: HolySheep AI bietet <50ms Latenz")
print(" → Wechseln Sie für schnellere Tool-Aufrufe")
print(" → DeepSeek V3.2: $0.42/MTok mit optimaler Geschwindigkeit")
Fehler 4: Token-Kosten nicht nachvollziehbar
Problem: Die API-Kosten sind höher als erwartet, aber Sie wissen nicht warum.
Lösung:
# ✅ Vollständige Token-Verfolgung implementieren
def create_cost_tracking_agent():
"""Agent mit vollständiger Kostenüberwachung"""
class TrackedState(TypedDict):
messages: Annotated[list, add_messages]
total_tokens: int
total_cost_usd: float
step_count: int
# Token-Preise für verschiedene Modelle
MODEL_PRICES = {
"gpt-4.1": 8.00, # $8.00/MTok
"claude-sonnet-4.5": 15.00, # $15.00/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok (HolySheep)
"gemini-2.5-flash": 2.50 # $2.50/MTok
}
def track_tokens(state: TrackedState, model: str = "deepseek-v3.2") -> TrackedState:
# Tokens schätzen
last_message = state["messages"][-1].content if state["messages"] else ""
tokens = len(last_message.split()) * 2 # Grob-Schätzung
state["step_count"] = state.get("step_count", 0) + 1
state["total_tokens"] = state.get("total_tokens", 0) + tokens
state["total_cost_usd"] = (state["total_tokens"] / 1_000_000) * MODEL_PRICES[model]
return state
# ... Rest des Graphen ...
return tracked_graph
tracked_agent = create_cost_tracking_agent()
print("✅ Token-Tracking aktiv
Verwandte Ressourcen
Verwandte Artikel