Verfolgen eines LangGraph-Blog-Pipelines mit Langfuse: Spans, Scores, Warnungen
Einrichten eines mehr-Agenten-LangGraph-Pipelines mit Langfuse: ein Trace pro Artikel, verschachtelte Spans, Warnungen, Scores sowie eine Debugging-Routine für schwache Entwürfe.
Quellen: Building Multi-Agent System From Scratch — Part 6: Observability and Debugging
Schutzmechanismen (Quellenprüfungen, Zitierungsüberprüfungen, begrenzte Wiederholversuche, Eskalation an Menschen) helfen nur dann, wenn ein fehlgeschlagener Ablauf später rekonstruiert werden kann. Ein mehragentenbasiertes Blog-Pipeline-System wirft in diesem Zusammenhang zahlreiche Fragen auf: Waren die Forschungsquellen unzuverlässig? Hat der Autor Warnungen bezüglich fehlender Zitate ignoriert? War die Ablehnung durch den Reviewer berechtigt? Ist die Suchanfrage abgelaufen? Wie viel kostete der Ablauf in Bezug auf Modellaufrufe und Tokens?
Ohne Überwachungsfähigkeit ist das System eine Blackbox, die weitere Blackboxes umschließt. Das Nachverfolgen mehrerer Agenten ist ein besonders schwieriger Aspekt dieser Aufgabe. Diese Konfiguration fügt Langfuse zu einer LangGraph-Blog-Pipeline hinzu und erzeugt für jeden durchgeführten Artikel einen Trace, der Übergaben, Modell- und Toolaufrufe, Zeiten, Kosten, Fehler sowie Bewertungsergebnisse enthält.
Bestimmung dessen, was ein Trace ist
Vor dem Bearbeiten des Codes wählen Sie die Einheit aus, die überprüft werden soll. In diesem Workflow entspricht eine Spur einem Versuch, einen Artikel zu erstellen. Sie beginnt, wenn ein Thema eingereicht wird, und endet, wenn der Workflow abgeschlossen ist, eine Pause einlegt, weil ein Mensch eingreifen muss, oder fehlschlägt. Darin sind Beobachtungen zu jeder durchgeführten Aufgabe enthalten:
blog-pipeline
├── research-agent
│ ├── search-web
│ └── research-model
├── writer-agent
│ └── writer-model
├── citation-check
│ └── citation-review-model
└── reviewer-agent
└── reviewer-model
Langfuse arbeitet mit drei Substantiven. Eine Trace ist eine vollständige Ausführung von Anfang bis Ende. Eine Observation ist ein einzelner Schritt darin, der entweder eine Zeitspanne (Anwendungslogik), eine Generierung (einen Aufruf eines LLMs) oder ein Ereignis (einen kleinen Marker) sein kann. Eine Session verbindet mehrere Traces miteinander, wie beispielsweise alle Versuche zu demselben Blog-Thema.
Die Hierarchie spiegelt den Ablauf des Debuggens wider: Man beginnt mit der vollständigen Ausführung und geht anschließend auf den spezifischen Agenten, das Tool oder den Modellaufruf ein, der Probleme verursacht hat.
Einrichten von Langfuse
Erstellen Sie ein Langfuse-Projekt und geben Sie dessen Anmeldeinformationen über Umgebungsvariablen an. Der Geheimcode sollte niemals in Quelldateien geschrieben werden.
pip install -U langfuse
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"
Das langfuse-Paket ist das Python SDK. Die beiden Schlüssel authentifizieren die Anwendung gegenüber dem Projekt, und LANGFUSE_BASE_URL wählt die Bereitstellungsinstanz aus. Das Beispiel richtet sich auf die EU-Cloud-Region – ersetzen Sie diese bei Bedarf durch eine andere Region oder eine selbst gehostete Instanz.
Eine Umgebungsbezeichnung sorgt dafür, dass Entwicklungsdaten von Produktionsdaten getrennt bleiben. Kurze, stabile Namen wie development, staging und production eignen sich hier gut. Ohne sie landen Experimente und tatsächliche Artikelverarbeitungen in derselben Ansicht, was das Debuggen erschwert.
Hinzufügen des Callback-Handlers
Langfuse liefert einen LangChain-Callback-Handler mit. LangGraph baut auf den von LangChain bereitgestellten Mechanismen für Ausführung und Callbacks auf, sodass derselbe Handler Ereignisse aus Graphen, Ketten, Tools und Modellen erfassen kann:
from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler
langfuse = get_client()
langfuse_handler = CallbackHandler()
get_client() erstellt einen Client aus den Umgebungsvariablen. Der Handler sollte beim Start einmal erstellt und bei jedem Ausführungsvorgang wiederverwendet werden. Er lauscht auf LangChain-Ereignisse und leitet sie als Beobachtungen weiter, wodurch eine manuelle Nachverfolgung jeder Modellaufruf nicht mehr notwendig ist.
Dann kommt propagate_attributes ins Spiel. Es setzt dieselben Labels auf jede Beobachtung innerhalb eines Ausführungsvorgangs, wodurch eine Suchanfrage, eine Erstellungsphase durch einen Schreiber sowie eine Überprüfungsphase durch einen Reviewer als Bestandteile eines einzigen Artikelverarbeitungspipelines gefiltert werden können.
Die Aufrufung des Graphen wird in einen Root-Span eingebettet, wobei der Handler über die LangGraph-Konfiguration weitergegeben wird:
def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
initial_state = {
"topic": topic,
"audience": "developers new to agent systems",
"research_brief": "",
"sources": [],
"open_questions": [],
"article_draft": "",
"review_feedback": "",
"citation_issues": [],
"approved": False,
"revision_count": 0,
"status": "researching",
}
with langfuse.start_as_current_observation(
as_type="span",
name="blog-pipeline",
input={"topic": topic, "audience": initial_state["audience"]},
) as pipeline_span:
with propagate_attributes(
trace_name="blog-pipeline",
session_id=blog_id,
user_id=user_id,
tags=["blog-pipeline", "langgraph"],
version="1.0.0",
metadata={"workflow": "research-write-review"},
):
trace_id = langfuse.get_current_trace_id()
result = graph.invoke(
initial_state,
config={"callbacks": [langfuse_handler]},
)
pipeline_span.update(
output={
"status": result["status"],
"approved": result["approved"],
"revision_count": result["revision_count"],
}
)
return result, trace_id
Mehrere Details dieses Wrappers sind wichtig. Der äußere Aufruf von start_as_current_observation() erstellt die Wurzelbeobachtung, wobei die Eingabedaten absichtlich kompakt gehalten werden – lediglich Thema und Zielgruppe. Das ist auf einen Blick in der Spurliste sichtbar, weshalb eine Ausgabe des internen Zustands dort nicht erforderlich ist.
propagate_attributes() verteilt alle verschachtelten Beobachtungen. blog_id dient als Sitzungs-ID und gruppiert Wiederholungsversuche bezüglich eines Artikels. Ein version-Wert identifiziert den Workflow sowie die Veröffentlichung der Anfrage hinter einem Ergebnis, sodass bei Verhaltensänderungen die Spuren nach Version gefiltert werden können – anstatt durch Vermutungen.
Der Handler durchläuft LangGraphs reguläre config-Einstellungen und sammelt dort automatisch detaillierte Informationen auf Framework-Ebene: jeden Knotenaufruf, die Generierung durch den LLM, Tool-Aufrufe, Eingaben, Ausgaben, Latenzzeiten, Tokenzahlen sowie Kosten.
Zum Schluss wird der Wurzelelementspan mit drei Ergebnissen auf Pipeline-Ebene aktualisiert: ob die Ausführung abgeschlossen wurde, ob der Entwurf genehmigt wurde und wie viele Überarbeitungen durchgeführt wurden. Der zurückgegebene trace_id sollte in den Anwendungsprotokollen oder der Benutzeroberfläche gespeichert werden, um einen direkten Zugriff vom Artikelausführungsprozess auf dessen Langfuse-Trace zu ermöglichen.
Business-Ebene-Spans innerhalb von Agenten
Automatisches Tracing belegt lediglich, dass ein Modell aufgerufen wurde. Es zeigt nicht an, was jeder Spezialist an den nächsten weitergegeben hat. Dafür sollte ein leichter Span innerhalb eines Agenten-Nodes hinzugefügt werden, wobei das config-Argument in verschachtelte LangChain- oder LangGraph-Aufrufe weitergeleitet wird, damit der Handler denselben Trace-Kontext beibehält:
from langchain_core.runnables import RunnableConfig
def research_node(state: BlogState, config: RunnableConfig) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="research-agent",
input={"topic": state["topic"], "audience": state["audience"]},
) as span:
result = research_agent.invoke(
{"topic": state["topic"], "audience": state["audience"]},
config=config,
)
span.update(
output={
"source_count": len(result["sources"]),
"open_question_count": len(result["open_questions"]),
"brief": result["research_brief"],
}
)
return {
"research_brief": result["research_brief"],
"sources": result["sources"],
"open_questions": result["open_questions"],
"status": "writing",
}
Der Span ergänzt den Callback-Handler und ersetzt ihn nicht. Er fügt die Bedeutung hinzu, die das Framework nicht erraten kann – genauer gesagt, was „Forschung erfolgreich“ in diesem Workflow bedeutet. Hier werden die Anzahl der Quellen, die Zahl der ungelösten Fragen sowie die Anweisungen an den Autor erfasst.
Durch das Übergeben von config: RunnableConfig an research_agent.invoke() werden Callbacks sowie der Trace-Context mit übertragen. Wenn man das vergisst, entsteht ein klassischer Fehler im Bereich der Überwachbarkeit: Der übergeordnete Graph wird verfolgt, doch der verschachtelte Aufruf erscheint als unzusammenhängender Trace oder verschwindet ganz.
Halten Sie die Namen der Überwachungsobjekte konstant: research-agent, writer-agent, citation-check. Ein Thema, ein Modellname oder eine Benutzer-ID sollten nicht in einen Namen eingebaut werden, da diese Daten zu den Eingaben, Modellattributen oder Metadaten gehören. Konstante Namen sorgen dafür, dass Dashboards und Filter auch bei Implementierungsänderungen weiterhin funktionieren.
Warnungen und Fehler
Nicht jedes unerwartete Ergebnis bedeutet einen Absturz. Eine zweifelhafte Quelle, ein aufgebrauchtes Überarbeitungsbudget oder eine nicht unterstützte Zitierung sind ebenfalls von Bedeutung und sollten als Warnung angezeigt werden.
def record_source_assessment(assessment: SourceAssessment) -> None:
if assessment.suspicious_content:
langfuse.update_current_span(
level="WARNING",
status_message="Untrusted source contained agent-directed instructions.",
)
def record_pipeline_failure(error: Exception) -> None:
langfuse.update_current_span(
level="ERROR",
status_message=f"Pipeline failed: {type(error).__name__}",
)
Langfuse unterstützt Log-Ebenen für Beobachtungen. WARNING wird für einen abgeschlossenen Lauf verwendet, der jedoch überprüft werden muss, während ERROR für echte Fehler gilt. Statusnachrichten sollten das betriebliche Problem beschreiben, ohne sensible Quelltexte oder Geheimnisse in die Telemetriedaten aufzunehmen.
Der Quellbewertungs-Recorder gehört zum Quellprüfungsweg, und der Fehler-Recorder in einen Fehlerhandler, der tatsächlich wiederherstellen oder melden kann. Jede Ausnahme nur zum Zweck des Eintragens in die Logs zu erfassen, ist eine schlechte Idee – unerwartete Fehler sollten dennoch zur Fehlerrückverfolgung gelangen und deutlich gemeldet werden.
Bewertung des Ergebnisses
Ein Trace beschreibt, was passiert ist, und eine Bewertung gibt an, ob das Ergebnis gut war. Sobald ein Lauf abgeschlossen ist, wird eine erste, deterministische Bewertung auf der Grundlage des Pipeline-Zustands hinzugefügt:
result, trace_id = run_blog_pipeline(
topic="How AI agents use tools",
blog_id="blog-ai-tools-001",
user_id="philip",
)
if trace_id:
langfuse.create_score(
trace_id=trace_id,
name="review_approved",
value=1 if result["approved"] else 0,
data_type="BOOLEAN",
comment=result["status"],
)
langfuse.create_score(
trace_id=trace_id,
name="revision_count",
value=float(result["revision_count"]),
data_type="NUMERIC",
)
Die erste Bewertung gibt das strukturierte Ja-oder-Nein-Urteil des Reviewers wieder, während die zweite angibt, wie viel Überarbeitung ein Entwurf erfordert hat. Im Laufe der Zeit liefern sie mehr als nur Anekdoten. Welches Modell, welche Prompt-Version, welche Themenkategorie oder welcher Informationsquellenansatz führt zu weniger Überarbeitungen und mehr Zustimmungen?
Auch diese sind keine absoluten Wahrheiten, da ein Reviewer sich irren kann. Eine von einem Menschen nach dem Lesen des Endentwurfs abgegebene Bewertung kann später erfolgen. Entscheidend ist vielmehr die Gewohnheit, Qualität messbar zu machen und verschiedene Fälle miteinander zu vergleichen, anstatt sich auf ein paar in Erinnerung gebliebene Beispiele zu verlassen.
Eine Debugging-Routine
Falls ein Artikel schwach ausfällt, öffnen Sie seine Dokumentation und stellen Sie nacheinander folgende Fragen:
- Hat der Prozess den erwarteten Weg eingeschlagen? Achten Sie auf Schritte wie Forschung, Verfassen, Zitatenprüfung und Reviewing sowie auf mögliche Überarbeitungsschleifen.
Deshalb ist die Struktur der Nachverfolgung wichtig. Ein gut benannter Baumwandlungsprozess verwandelt „der Blog lag falsch“ in eine präzise Feststellung wie „der Autor fügte nach der zweiten Überarbeitung eine Behauptung hinzu, die im Entwurf fehlte.“
Sensible Daten ausschließen
Tracing kann Aufforderungen, Ausgaben, Tool-Ergebnisse sowie Metadaten aufzeichnen, was auch bedeutet, dass es Informationen erfassen kann, die nicht gespeichert werden sollten. API-Schlüssel, Authentifizierungs-Header sowie vollständige Geheimkonfigurationen dürfen niemals an Langfuse übermittelt werden. Falls Benutzerdaten später in den Pipeline gelangen, prüfen Sie, was aufgezeichnet wird, und richten Sie vor dem Export der Spuren eine Maskierung ein. Langfuse bietet Maskierungsfunktionen, die sensible Werte vor ihrem Verlassen der Anwendung entfernen.
Die Faustregel lautet: Aufzeichnen Sie so viel, wie nötig ist, um eine Entscheidung zu erklären, aber nicht mehr, als sicher aufbewahrt werden kann.
Vorteile des Pipelines
Jeder Ausführungsvorgang hinterlässt nun eine Betriebsaufzeichnung:
- Jeder Artikelversuch ist eine einzelne Spur.
- Agenten, Modelle und Tools erscheinen als verschachtelte Beobachtungen.
- Konstante Namen zusammen mit Metadaten, Version, Sitzung und Umgebung erleichtern das Filtern.
Das Debuggen ist dann nicht mehr nur das Durchscrollen von Terminalprotokollen und Raten, sondern beinhaltet die genaue Untersuchung des Punkts, an dem etwas schiefgelaufen ist.