Praktische Hinweise: Erstellung eines Multi-Agenten-Systems von Grund auf – Teil 6

Die folgenden Notizen skizzieren einen praktischen Weg durch den Abschnitt „Building Multi-Agent System From Scratch — Part 6: Observability and Debugging“. Der Schwerpunkt liegt auf Verträgen, Überprüfungen sowie Code-Platzhaltern, anstatt auf motivierenden Erläuterungen. Während der Übersichtsphase sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Man sollte kleine, testbare Einheiten vorzugsweise verwenden statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema.

Was sollte eine Trace darstellen?

Die Phase „Was sollte man nachverfolgen?“ funktioniert am besten, wenn sie als messbare Oberfläche betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Strukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

blog-pipeline
├── research-agent
│   ├── search-web
│   └── research-model
├── writer-agent
│   └── writer-model
├── citation-check
│   └── citation-review-model
└── reviewer-agent
    └── reviewer-model

Langfuse mit LangGraph verbinden

Die Phase des Anschlusses von Langfuse an LangGraph funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein erfolgreiches Beispiel, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Protokollieren Sie die Laufzeiten sowie die Kosten pro Token oder Abfrage zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Halten Sie den Zustand des Graphen strukturiert und typisiert – verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

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"

from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler


langfuse = get_client()
langfuse_handler = CallbackHandler()

Verfolgen Sie einen vollständigen Artikelverarbeitungsvorgang

Die Trace one complete article-Ebene funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Rollback-Anmerkung, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne Durchsicht des gesamten Graphen prüfen können. Halten Sie den Zustand des Graphen strukturiert und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen bei der Fortsetzung der Verarbeitung. Die Trace one complete article-Ebene funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Rollback-Anmerkung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

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

Fügen Sie Beobachtungen hinzu, die den Übergang erklären.

Für den Abschnitt „Hinzufügen von Beobachtungen zur Erklärung der Phase“ sollten die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingabedaten und den validierten Ausgabedaten. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Kompilierzeit-Verkabelung entspricht nicht der Geschäftsabschlussfähigkeit.

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",
    }

Markieren Sie die Signale, die Aufmerksamkeit erfordern

Für den Mark-Prozess sollten die Signale der jeweiligen Phase, die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erhalten Sie neben den funktionalen Ergebnissen auch Aufzeichnungen der Laufzeiten sowie der Kosten für Tokens oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozesspfad von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Setzen Sie menschliche Freigabe für Schritte voraus, bei denen Geld ausgegeben wird oder Produktionsdaten geändert werden. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

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__}",
    )

Entscheidungen der Prüfer in Bewertungswerte umwandeln

Um die Entscheidungen des Turn Reviewers in eine bestimmte Phase umzusetzen, sollten vor der Codeänderung die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode liegen. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Codeverlauf durchlesen zu müssen. Menschliche Freigabe sollte für Schritte erforderlich sein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung. Um die Entscheidungen des Turn Reviewers in eine bestimmte Phase umzusetzen, sollten vor der Codeänderung die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bevorzugen Sie kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf mehrere.

gewinkelte Pipeline.

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",
    )

Fehlerhafte Ausführung in fünf Fragen debuggen

Beim Bearbeiten der Phase „Fehlerhafte Ausführung debuggen“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie was bei einem teilweisen Versagen geschieht. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Zwischenpunkte nach aufwändigen Schritten. Die Wiederaufnahme sollte keine doppelten LLM-Aufrufe veranlassen, wenn ein Operator einen späteren Knoten erneut ausführt.

Auch die Beobachtbarkeit hat Grenzen hinsichtlich der Privatsphäre

Beim Arbeiten an der Überwachbarkeit gibt es ebenfalls eine Phase, in der man zunächst den Vertrag aufschreiben sollte: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Checkpoint an. Das Wiederaufnehmen des Prozesses sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

Was wir gebaut haben

Beim Arbeiten in der Phase „Was wir gebaut haben“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Operator:innen überprüfen können, ohne den gesamten Ablauf durchlesen zu müssen. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Arbeiten in der Phase „Was wir gebaut haben“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf einen verworrenen Ablauf.

Operative Checkliste

Zur Phase der Betriebskontrollliste sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen.

Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallplan. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen.

Fügen Sie menschliche Freigabe bei Schritten ein, die Geld ausgeben oder Produktionsdaten ändern. Kompilierzeitbezogene Verbindungen bedeuten noch keine vollständige Geschäftsabwicklung.

Schreiben Sie ein kurzes Handbuch: Wie werden Schlüssel ausgetauscht, wie wird die Warteschlange geleert und wie wird der letzte Eingang rückgängig gemacht?

Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema.

Fügen Sie menschliche Freigabe bei Schritten ein, die Geld ausgeben oder Produktionsdaten ändern. Compile-time Wiring bedeutet nicht automatisch vollständige Geschäftsabdeckung.

Vor der Einführung des Stack sollten Versionen eingefroren werden, ein „goldener Transkript“ für den kritischen Pfad erstellt und Rollback-Schritte bestätigt werden. Gemeinsame Umgebungen benötigen Rate Limits, Überprüfungen der Nutzerrechte sowie einen klaren Verantwortlichen für die Rotation von Geheimnissen. Wählen Sie langweilige Zuverlässigkeit statt cleverer, einmaliger Demonstrationen.

Batch-Hinweis für cf19385cb4a9: Halten Sie Provider-Schlüssel außerhalb des Repositories, legen Sie eine Obergrenze für Session-Tokens fest und speichern Sie Transkripte neben den Eval-Fixtures, damit spätere Modellwechsel vergleichbar bleiben.