Sechs LangGraph-Primitiven und der in jeder verborgene Ausfallmodus

Ein erster LangGraph-Agent entsteht in der Regel schnell: Er beantwortet eine Frage, präzisiert die Antwort und stoppt. Dann fügt jemand einen Versuchszweig hinzu, und plötzlich endet der Graph nicht mehr, wodurch API-Kredite verbraucht werden, bis der Prozess beendet wird. Die Lösung liegt oft in einem fehlenden Kante, doch das eigentliche Problem ist das Fehlen eines mentalen Modells dafür, warum sich der Graph so verhält.

Dieser Leitfaden konstruiert dieses Modell aus den sechs Primitiven, aus denen LangGraph besteht: Zustand, Knoten, direkte Kanten, bedingte Kanten, Checkpointing und menschliche Einmischung. Für jedes dieser Elemente erhalten Sie ein minimales Beispiel, den häufigsten Fehler, den Teams damit machen, sowie die Version, die veröffentlicht werden sollte. Wenn Sie einen umfassenderen Überblick über auf diesen Elementen basierende Agentenmuster wünschen, behandelt LangGraph in der Praxis: Zustand, Knoten, Kanten und fünf Agentenmuster dieses Thema; hier liegt der Fokus auf Ausfallmustern.

Warum ein Graph statt einer Kette

LangChains Piping-Syntax prompt | llm | parser ist für einen einzigen Durchlauf durch ein Modell praktisch. Sie funktioniert nur noch, solange der Agent eine Entscheidung treffen muss: suchen oder direkt antworten, erneut versuchen oder aufgeben, eine Person um Hilfe bitten oder fortfahren. Eine Kette hat kein Konzept von „es hängt davon ab“, weshalb Entwickler die Kettenaufrufe in if-Anweisungen einbetten – bald haben sie so eine selbst gebaute, undokumentierte und schwieriger zu debuggende Zustandsmaschine.

LangGraph macht diese Zustandsmaschine explizit. Es gibt Knoten, Kanten sowie ein gemeinsames Zustandsobjekt, das zu jedem Zeitpunkt überprüft werden kann. Daran ist nichts Magisches – und genau das ist der Vorteil: Jede Entscheidung des Agents entspricht etwas, was in der Graphendefinition zu finden ist.

1. Zustand: ein gemeinsames Objekt und relevante Reduzierfunktionen

Der State ist das einzige Objekt, aus dem jeder Node liest und in das er schreibt. Ohne ihn wird der Kontext oft als Funktionsargumente weitergegeben, wodurch es schwierig wird zu bestimmen, was ein bestimmter Schritt tatsächlich wusste. Die untenstehende Definition ist ein TypedDict mit einer Frage, einer Antwort und einer Liste von Nachrichten, deren Aktualisierungen durch den add_messages Reducer zusammengeführt werden.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add ist kein Nachrichten-Reducer

Viele Tutorials kennzeichnen das Feld für Nachrichten stattdessen mit operator.add. Das sieht richtig aus: add fügt dem Listeneintrag etwas hinzu, anstatt ihn zu überschreiben – was für ein wachsendes Gespräch notwendig ist. Das Problem ist jedoch, dass es blind zusammenfügt. Sobald man beispielsweise eine bestehende Nachricht aktualisieren oder entfernen muss, etwa um die Historie zu kürzen oder das Ergebnis eines Tool-Aufrufs zu ersetzen, fügt es stattdessen einen Duplikat ein, wodurch die Gesprächshistorie ohne jeglichen Fehler mit veralteten Einträgen gefüllt wird.

add_messages wurde speziell dafür entwickelt. Es vergleicht Nachrichten anhand der ID und ersetzt eine bereits vorhandene Nachricht, wenn diese ID vorliegt, fügt jedoch nur wirklich neue Nachrichten hinzu. Die Regel ist einfach: Verwenden Sie add_messages für Felder, die HumanMessage- und AIMessage-Objekte enthalten, und nutzen Sie weiterhin operator.add für einfache, ansammelnde Listen, wie beispielsweise eine fortlaufende Aufzeichnung der aufgerufenen Tools.

Halten Sie die Struktur des Zustands minimal

Der zweite häufige Fehler besteht darin, den Zustand wie ein Datenbankschema zu gestalten – mit einem Feld für jeden Bedarf, den jemand später haben könnte. Fügen Sie ein Feld nur dann hinzu, wenn ein Knoten es tatsächlich liest oder schreibt. Die Folgen des Ignorierens sind konkret: Betrachten Sie ein Graphen-System zur Verarbeitung von Dokumenten, das vollständige Rohantworten von Large Language Models samt Metadaten zur Tokenverwendung im Zustand speichert. Die Verarbeitung von 50 Dokumenten in einer Schleife führte dazu, dass jeder Checkpoint etwa 180 KB groß wurde, und die Schreibvorgänge in Postgres lagen bei über 400 ms – langsam genug, damit Nutzer, die auf eine Antwort warten, es bemerken. Die Lösung war unprätentiös: Reduzieren Sie den Zustand auf die drei Felder, die die nachgelagerten Knoten tatsächlich verwenden. Denken Sie daran, dass bei Verwendung eines Checkpointers alles im Zustand bei jedem Schritt serialisiert und gespeichert wird.

2. Knoten: Geben Sie nur das zurück, was sich geändert hat

Ein Knoten ist eine gewöhnliche Python-Funktion. Sie erhält den Zustand, führt ihre Arbeit aus und gibt ein Dictionary zurück, das nur die Felder enthält, die sie geändert hat. Das ist der gesamte Vertrag. Im ersten Beispiel wird ein OpenAI-Chat-Modell mit der Frage aufgerufen und die Antwort in answer geschrieben.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Während Sie die Graphenstruktur durchlaufen – was den größten Teil der anfänglichen Arbeit ausmacht – möchten Sie vermutlich nicht, dass jede Vorversion auf eine bezahlte API zugreift. Ein lokales Modell, das von Ollama bereitgestellt wird, implementiert dieselbe Schnittstelle, sodass der Knoteninhalt unverändert bleibt und das Debuggen keine Kosten verursacht:

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Diese Version erfordert, dass Ollama lokal mit dem heruntergeladenen Modell (ollama pull llama3.1) läuft und das Integrationspaket installiert ist (pip install langchain-ollama). Die Einstellung temperature=0 sorgt außerdem dafür, dass die Ausführungen reproduzierbarer sind, was beim Testen der Routing-Logik hilft.

Das Zurückgeben des gesamten Zustands überschreibt andere Aktualisierungen

Ein häufiger Fehler besteht darin, statt nur der geänderten Schlüssel das gesamte Zustandsdictionary aus einem Knoten zurückzugeben. In einem kleinen linearen Graphen scheint das zu funktionieren, weil sonst nichts diese Felder berührt. Sobald zwei Knoten sich überschneidende Felder aktualisieren, überwrite die vollständige Rückgabe eines Knotens die Änderungen des anderen mit alten Werten. Das Symptom sieht wie ein Routing-Problem aus, weshalb Entwickler dazu neigen, in der Kantenlogik zu suchen, obwohl die eigentliche Ursache darin liegt, dass ein Knoten zu viel zurückgibt. Das Zurückgeben einer minimalen Aktualisierung ermöglicht es außerdem den Reduzern, ihre Aufgabe zu erfüllen: Ein Feld ohne Reducer wird einfach durch das ersetzt, was der Knoten zurückgibt.

3. Direkte Kanten: Verbinden Sie immer den Ausgang

Die Kanten bestimmen, was als Nächstes ausgeführt wird. Eine direkte Kante ist bedingungslos: Wenn Knoten A abgeschlossen ist, wird Knoten B ausgeführt. Das untenstehende Diagramm zeigt zwei Knoten, verbindet answer mit refine, verbindet refine mit END, legt den Eingangspunkt fest und kompiliert das Ganze.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

Die END-Kante ist der Teil, den die Leute vergessen – und sie ist die klassische Ursache dafür, dass ein Graph scheinbar ewig läuft. Die zuverlässigste Vorgehensweise besteht darin, jeden Pfad durch den Graph explizit bei END enden zu lassen, damit man die Beendigung durch das Lesen der Definition nachvollziehen kann. Dies ist besonders wichtig, sobald Zyklen auftreten: Eine Wiederholungsschleife ohne Pfad zu END oder mit einer Bedingung, die niemals wahr wird, führt weiterhin zu Zyklen, bis LangGraphs Rekursionsschranken mit einem GraphRecursionError eingreifen. Diese Schranken dienen lediglich als Sicherheitsnetz, nicht als Designelement – jede dieser Iterationen kostet weiterhin Token. Wenn ein Graph scheinbar hängen bleibt, prüfen Sie zuerst die Definition des Graphs.

4. Bedingte Kanten: Wo der Agent tatsächlich entscheidet

Konditionale Kanten sorgen dafür, dass das Graph ein Agent statt eines festen Pipelines wird. Eine Routenfunktion prüft den Zustand und gibt eine Kennung zurück; eine Zuordnung wandelt jede Kennung in das nächste Knoten um. In diesem Beispiel wird eine kurze Antwort (unter 50 Zeichen) an refine gesendet, während alles andere an END gelangt.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Bemerken Sie, dass diese konditionale Kante die direkte answer-Kante zu refine aus dem vorherigen Beispiel ersetzt. Wenn Sie beide registrieren, werden beide Pfade genutzt, was selten das gewünschte Verhalten ist.

Nicht übereinstimmende Routenkennungen führen zu deutlichen, aber unklaren Fehlern

Der häufige Fehler hier ist eine Routing-Funktion, die einen in dem Zuordnungs-Wörterbuch nicht vorhandenen String zurückgibt. Der resultierende Fehler ist ein ziemlich allgemeiner Schlüsselfehler, der mehrere Ebenen tief im Stack-Trace versteckt ist, und schon etwas Kleines wie ein nachgestellter Leerzeichen kann erhebliche Zeitkosten verursachen. Eine zuverlässige Gewohnheit: Schreiben Sie zunächst die Zuordnungen und erstellen Sie anschließend den Router, indem Sie die genauen Schlüssel daraus kopieren. Noch besser ist es, die Labels einmal als Konstanten zu definieren oder den Rückgabetyp des Routers mit Literal["refine", "done"] zu annotieren, damit Typprüfer und Leser die zulässigen Werte sofort erkennen.

5. Checkpointing: Speicher, der zwischen Aufrufen erhalten bleibt

Ein Checkpointer verwandelt einen zustandslosen Funktionsaufruf in ein Gespräch mit dem Speicher. Ohne einen beginnt jeder app.invoke()-Aufruf von vorne. Mit einem wird der Zustand pro Thread gespeichert, und jeder Aufruf, der im Konfigurationsdaten denselben thread_id übermittelt, setzt dort fort, wo der vorherige aufgehört hat. Im Beispiel erinnert sich der zweite Aufruf im Thread user-session-42 an die erste Frage.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver eignet sich nur für die lokale Entwicklung und nichts anderes. Er befindet sich im Prozessspeicher, wodurch ein Neustart des Servers jedes „Gespräch“ löscht. Alles, worauf echte Benutzer angewiesen sind, benötigt einen persistenten Backend: SQLite für einen einzelnen Server oder Postgres, wenn mehrere Instanzen den Zustand teilen müssen.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Jeder Backend wird als eigenes Paket bereitgestellt, wie die Installationsanleitungen zeigen. In den aktuellen Versionen werden diese Speichererzeuger in der Regel aus einer Verbindungszeichenkette erstellt (zum Beispiel über from_conn_string), und Postgres benötigt einen einmaligen Aufruf von setup(), um seine Tabellen anzulegen. Prüfen Sie daher die Dokumentation zu Checkpointer auf die genaue Initialisierung in Ihrer Version.

Der Fehlerfall tritt auf, wenn im Produktivbetrieb der Speichererzeuger im Arbeitsspeicher verwendet wird und dies erst festgestellt wird, wenn ein Neustart einer Staging-Umgebung eine Live-Demo löscht. Die gute Nachricht ist, dass der Umstieg kostengünstig ist, sofern das Diagramm ansonsten gut strukturiert ist: Checkpointer ist ein Kompilierungsparameter, kein Neuentwurf, und der Wechsel zu SqliteSaver kann in unter einer Stunde abgeschlossen werden.

6. Mensch im Prozess: statische Breakpoints gegenüber dynamischen Unterbrechungen

Das Muster, das die meisten Tutorials zeigen, ist interrupt_before – eine Liste von Knotennamen, an denen das kompilierte Graphen vor der Ausführung pausiert:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Es funktioniert und ist leicht zu erklären, doch es ist statisch. Der Pausepunkt wird durch den Knotennamen festgelegt; man kann ihn nicht bedingungsabhängig gestalten und auch kein Payload hinzufügen, der beschreibt, was der Prüfer betrachten soll. Reale Anforderungen übersteigen es schnell, denn „vor diesem Knoten pausieren“ und „nur dann pausieren, wenn die Rückerstattung mehr als 500 Dollar beträgt“ sind unterschiedliche Regeln, wobei nur die erste auf diese Weise ausgedrückt werden kann.

Pause innerhalb des Knotens mit interrupt()

Das flexiblere Muster besteht darin, interrupt() direkt aus dem Knoten heraus aufzurufen. Der untenstehende Knoten überprüft den Rückerstattungsbetrag; bei einem Betrag über 500 Dollar pausiert er und zeigt den Entwurf sowie den Betrag einem Menschen an. Der erste invoke-Aufruf läuft bis zu dieser Pause weiter. Der zweite Aufruf übermittelt Command(resume="approve") im selben Thread, wobei der dem resume-Parameter übergebene Wert zum Rückgabewert von interrupt() wird. Dadurch geht der Knoten entweder zur Weiterleitung über oder gibt einen abgebrochenen Status zurück. Ein Checkpointer ist erforderlich, da der pausierte Zustand während des Wartens irgendwo gespeichert werden muss.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

Der Beispielzustand verwendet Felder wie refund_amount, draft und task, die im früheren AgentState nicht vorhanden sind; in einem echten Graphen würden Sie diese dort deklarieren.

Die Wiederaufnahme führt zum Neustart des gesamten Knotens

Das Verhalten, das Menschen überrascht: In LangGraph wird nach einem interrupt()-Aufruf nicht dort weitergemacht, wo man aufgehört hat. Stattdessen wird der gesamte Knoten von vorne ausgeführt, und diesmal gibt interrupt() den Fortsetzungswert zurück anstatt eine Pause einzulegen. Jeder Code, der vor dem Aufruf stand, wird erneut ausgeführt. Ein Knoten, der vor dem Unterbrechen einen Zähler erhöht, wird diesen für jede Genehmigung zweimal erhöhen. Stellen Sie sicher, dass alles vor einem interrupt()-Aufruf idempotent ist, oder verlagern Sie Nebeneffekte auf einen früheren Knoten. Dasselbe Prinzip gilt für API-Aufrufe oder Datenbank-Einschreibungen, die vor der Pause platziert werden.

Ein Modell, das sich selbst genehmigt, ist kein Human-in-the-Loop-Verfahren

Egal, welchen Mechanismus Sie wählen – das Modell mit der Frage „Soll ich fortfahren?“ zu befragen und der Antwort zu vertrauen, stellt keine menschliche Überwachung dar, unabhängig davon, wie es bezeichnet wird. Es handelt sich vielmehr um das Agent, das seine eigene Entscheidung bestätigt. Ein echter Genehmigungsprozess übergibt die Kontrolle an eine Person außerhalb des Graphen und wartet auf deren Antwort.

Die sechs Primitiven im Überblick

Die folgende Zusammenfassung verbindet jedes Konzept mit seiner Funktion sowie dem typischen Fehler, der damit verbunden ist.

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Eine sinnvolle Build-Reihenfolge

Für das erste echte Graph-Modell sollten Sie zunächst den vollständigen Schleifenprozess mit InMemorySaver und ohne Unterbrechungen zum Laufen bringen. Halten Sie den Zustand klein und beschränken Sie ihn auf das, was die Knoten benötigen, und stellen Sie sicher, dass jede bedingte Kante genau die Labels zurückgibt, die ihre Zuordnung erwartet. Erst wenn das reibungslos funktioniert, sollten Sie einen persistenten Checkpointer einsetzen und an dem Schritt eine Unterbrechung hinzufügen, der tatsächlich menschliche Intervention erfordert – typischerweise alles, was Geld bewegt, eine externe E-Mail sendet oder Daten löscht.

Mehr fortgeschrittene Funktionen, darunter benutzerdefinierte Reducer über add_messages hinaus, Subgraphen, die ein großes Graph in testbare Teile aufteilen, sowie Streamen auf Token-Ebene, basieren alle auf derselben Grundstruktur. Sie sind viel leichter zu übernehmen, nachdem Sie bereits ein Graph mit nur diesen sechs Konzepten erstellt, kaputtgemacht und repariert haben.

Kernpunkte

• Verwenden Sie add_messages für den Chatverlauf und operator.add nur für einfache Listen, und halten Sie den Zustand schlank, da er bei jedem Schritt gespeichert wird.
• Geben Sie nur die geänderten Felder der Knoten zurück; ein Rückgabewert mit vollständigem Zustand überschreibt stumm parallele oder frühere Aktualisierungen.
• Weisen Sie jedem Pfad einen expliziten Weg zum END zu und nutzen Sie begrenzte Wiederholungsschleifen, anstatt sich auf die Rekursionsgrenze zu verlassen.
• Erstellen Sie Routing-Labels aus der Zuordnung, damit sie nicht auseinanderdriften.
• Betrachten Sie InMemorySaver ausschließlich als Entwicklungstool; der Checkpointer-Austausch ist kostengünstig, daher führen Sie ihn durch, bevor Benutzer vom Graphen abhängig werden.
• Verwenden Sie lieber eine dynamische interrupt()-Funktion für bedingte Freigaben, und halten Sie den Code idempotent, da ein Neustart den Knoten erneut ausführt.

Referenzdokumentation: die Graph-API-Dokumentation für LangGraph, die Anleitung zu Interrupts sowie die interrupt()-API-Referenz.