17. State-Persistenz und Checkpointing

In Kapitel 15 haben wir die Agent-Schleife als StateGraph neu aufgebaut, und in Kapitel 16 haben wir vorgefertigte Komponenten und Multi-Branch-Routing verwendet, um etwas Ausgefeilteres zu bauen. Jeder Graph, den wir bisher geschrieben haben, teilt eine Einschränkung: Der Graph hält seinen State nicht fest.

Der Graph hält und verwaltet den State für die Dauer eines einzelnen invoke()-Aufrufs, und keinen Moment länger. Wenn Sie ihn aufrufen, erstellt LangGraph einen frischen State, führt die Nodes aus, führt jeden Rückgabewert gemäß den Reducer-Regeln in den State zusammen und gibt den finalen State an den Aufrufer zurück. Sobald es diesen State übergeben hat, erinnert sich der Graph nicht mehr daran. Der nächste invoke()-Aufruf beginnt mit einem brandneuen State, der keine Verbindung zum vorherigen Aufruf hat.

Daraus ergeben sich zwei Probleme. Erstens ist messages ebenfalls Teil des States, sodass sich der Agent an nichts erinnern kann, was Sie zuvor gesagt haben. Zweitens: Wenn ein Lauf mittendrin fehlschlägt, verschwindet alles, was bis zu diesem Punkt erreicht wurde. Angenommen, der dritte Node löst eine Ausnahme aus: Die von den ersten beiden Nodes erzeugten Ergebnisse gehen damit verloren, und Sie müssen von vorne beginnen. Bereits in 15.1 haben wir „das Wiederherstellen nach Unterbrechungen" als einen der Gründe aufgeführt, zu LangGraph zu greifen – dies ist das Problem, das wir dabei im Sinn hatten.

LangGraph behandelt dies auf Framework-Ebene. Hängen Sie einen Checkpointer an einen Graphen an, und LangGraph speichert bei jedem Ausführungsschritt automatisch einen Snapshot des States. Diese gespeicherten Snapshots überleben den invoke()-Aufruf, sodass der nächste Aufruf dort weitermachen kann, wo der letzte aufgehört hat. Diese Eigenschaft – dass der State über einen einzelnen Lauf hinaus fortbesteht – nennt man Persistenz.

Sie haben tatsächlich schon einmal einen Checkpointer verwendet. In Kapitel 11, als wir dem konversationsbasierten RAG-Agent ein Multi-Turn-Gedächtnis gegeben haben, übergaben wir create_agent(..., checkpointer=InMemorySaver()) und eine thread_id. Damals mussten Sie nur wissen, dass der Checkpointer den Gesprächsverlauf pro thread_id speichert; wir haben nie erklärt, wie. Und in Kapitel 16, als wir den checkpointer-Parameter eingeführt haben, sagten wir: „Wie das funktioniert, behandeln wir in Kapitel 17." Dies ist dieses Kapitel.

Es besteht aus drei Teilen. In 17.1 hängen wir einen Checkpointer an einen Graphen an und führen Multi-Turn-Gespräche mit thread_id. In 17.2 öffnen wir die gespeicherten Checkpoints, um zu sehen, was der Agent zu einem bestimmten Zeitpunkt wusste – Checkpoints sind Ihr wichtigstes Werkzeug, um aufzuspüren, warum sich ein Agent falsch verhalten hat. In 17.3 nehmen wir einen Graphen, der mitten im Lauf fehlgeschlagen ist, und setzen ihn von dort fort, wo er gestoppt hat, statt von vorne.

17.1) State über Aufrufe hinweg mitnehmen

17.1.1) Ein Graph, der vergisst

Wir haben dieses Kapitel damit begonnen zu sagen, dass ein Graph seinen State nicht festhält. Bestätigen wir das im Code.

Der folgende Graph hat dieselbe Form wie der say_hello-Graph aus 15.2. Der einzige Unterschied ist, dass der Node eine LLM-Antwort statt eines festen Strings zurückgibt.

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
graph = builder.compile()   # kein Checkpointer

Führen wir nun ein Gespräch über zwei Turns. Wir teilen dem Modell im ersten Aufruf unseren Namen mit und fragen es dann im zweiten Aufruf, wie unser Name ist.

# Erster Aufruf – wir nennen unseren Namen
graph.invoke({"messages": [{"role": "user", "content": "Hi, ich heiße Bob."}]})
 
# Zweiter Aufruf – wir fragen ihn wieder ab
result = graph.invoke({"messages": [{"role": "user", "content": "Wie heiße ich?"}]})
print(result["messages"][-1].content)

Ausgabe:

Es tut mir leid, aber ich kenne Ihren Namen nicht. Könnten Sie mir sagen, wie Sie heißen?

Die einzige Nachricht, die das Modell beim zweiten Aufruf erhalten hat, war "Wie heiße ich?". Der State aus dem ersten Aufruf ist nicht mehr im Graphen, sodass die früheren Nachrichten – diejenigen, die den Namen enthielten – es nie bis zum Modell geschafft haben.

Wir könnten das natürlich selbst beheben. Behalten Sie die vom ersten Aufruf zurückgegebenen messages und übergeben Sie sie zusammen mit dem zweiten. Genau so haben wir den Gesprächsverlauf in Kapitel 8 verwaltet. Aber dann würden wir unseren eigenen Code schreiben, um den Verlauf für jedes Gespräch und jeden Benutzer zu speichern und abzurufen. Das ist die Arbeit, die Ihnen ein Checkpointer abnimmt.

17.1.2) Checkpoints und Checkpointer

Ein Checkpointer ist ein Objekt, dessen Aufgabe es ist, State zu speichern. Sie erstellen eine Instanz – zum Beispiel InMemorySaver() – und übergeben sie an builder.compile(checkpointer=...), um sie an Ihren Graphen anzuhängen.

Sobald ein Checkpointer angehängt ist, kopiert der Graph den gesamten State und speichert ihn während der Ausführung. Jede dieser gespeicherten Kopien wird als Checkpoint bezeichnet. Stellen Sie es sich wie eine Fotografie vor: der gesamte State in diesem Augenblick, exakt so bewahrt, wie er war.

Ein Autosave in einem Videospiel ist das richtige Gedankenbild. Das Spiel zeichnet Ihren Fortschritt still auf, wann immer Sie einen bedeutsamen Punkt passieren, sodass Sie aufhören und später zurückkommen können oder sterben können, ohne von vorne beginnen zu müssen. Ein Checkpointer tut genau das für einen Graphen.

Wann ist also ein „bedeutsamer Punkt"? LangGraph teilt die Ausführung eines Graphen in Stufen auf, und jede Stufe wird als Super-Step bezeichnet. Ein Checkpoint wird jedes Mal gespeichert, wenn ein Super-Step abgeschlossen ist.

Der Grund, warum es ein Super-Step und nicht nur ein Step ist, liegt darin, dass eine einzelne Stufe mehrere Nodes gleichzeitig ausführen kann. In einem Graphen wie dem aus 17.1.1, in dem die Nodes eine gerade Linie bilden, ist das Ausführen eines Nodes ein Super-Step. Aber in einem Graphen, in dem mehrere Nodes parallel laufen, bilden all diese Nodes zusammen einen einzigen Super-Step.

Selbst ein einzelner Aufruf von invoke() hinterlässt also mehrere Checkpoints. Wir werden sie herausziehen und uns in 17.2 genau ansehen, was jeder einzelne enthält.

Der InMemorySaver, den wir als unser Beispiel verwendet haben, ist der einfachste Checkpointer, den es gibt. Wie der Name schon sagt, speichert er Checkpoints im Speicher des Prozesses (RAM). Es gibt nichts zu installieren und nichts zu konfigurieren, was ihn zu einer guten Wahl für das Lernen und die lokale Entwicklung macht. Der Nachteil ist, dass jeder gespeicherte Checkpoint verschwindet, wenn der Prozess neu startet. Die Produktionsalternativen betrachten wir in 17.1.5.

17.1.3) Einen Checkpointer hinzufügen

Das Anhängen eines Checkpointers erfordert nur zwei Dinge.

1. Erstellen Sie eine Checkpointer-Instanz und übergeben Sie sie an compile().
2. Übergeben Sie eine config, die eine thread_id enthält, wann immer Sie invoke() aufrufen.

Warum das zweite benötigt wird, klären wir gleich. Wissen Sie fürs Erste nur, dass es dem Checkpointer sagt, welches seiner gespeicherten Gespräche Sie fortsetzen möchten.

Wenden wir beides auf den Graphen aus 17.1.1 an.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
# 1. Einen Checkpointer erstellen und an compile() übergeben
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. Eine config mit einer thread_id an invoke() übergeben
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, ich heiße Bob."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Wie heiße ich?"}]},
    config,
)
print(result["messages"][-1].content)

Ausgabe:

Sie heißen Bob.

Dieselben zwei Aufrufe wie in 17.1.1, und ein anderes Ergebnis. Diesmal bleibt der Name erhalten.

Hier ist der Grund. Ein Graph mit angehängtem Checkpointer lädt den gespeicherten State, bevor er den llm_call-Node ausführt. Dieser State enthält bereits den ersten Austausch. Die neue Nachricht, die wir übergeben haben, wird dann darin zusammengeführt. Wie wir in 15.2.2 gesehen haben, trägt das Feld messages den add_messages-Reducer, sodass die neue Nachricht an die bestehende Liste angehängt wird. Das Modell erhält am Ende drei Nachrichten: die Begrüßung, seine eigene erste Antwort und die neue Frage.

Wir senden nur die neue Nachricht, und LangGraph lädt das frühere Gespräch aus dem letzten Checkpoint. Der Gesprächsverlauf, den wir in Kapitel 8 von Hand verwaltet haben, wird nun von LangGraph verwaltet.

17.1.4) thread_id: Der Identifikator, der Gespräche trennt

Eine thread_id ist ein Identifikator, der ein Gespräch von einem anderen unterscheidet. Sie wählen den Wert. Wir haben in 17.1.3 "1" verwendet, aber jeder beliebige String funktioniert. Rufen Sie den Graphen mit derselben thread_id auf, und Sie setzen dieses Gespräch fort; rufen Sie ihn mit einer anderen auf, und Sie beginnen ein separates Gespräch.

Bestätigen wir das. Wir lassen Alice und Bob unterschiedliche Gespräche über denselben Graphen führen.

def send(thread_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    return result["messages"][-1].content
 
# Alices Gespräch
send("alice", "Meine Lieblingsfarbe ist Blaugrün.")
 
# Bobs Gespräch – eine andere thread_id
send("bob", "Meine Lieblingsfarbe ist Orange.")
 
# Jeden von ihnen erneut fragen
print("Alice:", send("alice", "Was ist meine Lieblingsfarbe?"))
print("Bob:  ", send("bob", "Was ist meine Lieblingsfarbe?"))

Ausgabe:

Alice: Ihre Lieblingsfarbe ist Blaugrün.
Bob:   Ihre Lieblingsfarbe ist Orange.

Beide Gespräche liefen über dasselbe graph-Objekt und denselben Checkpointer, und dennoch vermischten sie sich nie. Die thread_id ist der Primärschlüssel, den der Checkpointer verwendet, um State zu speichern und nachzuschlagen. Unterschiedliche Schlüssel, vollständig getrennter Speicher.

Was passiert also, wenn Sie den Wert weglassen?

graph.invoke({"messages": [{"role": "user", "content": "Hallo"}]})

Ausgabe:

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

Der Graph läuft überhaupt nicht. Sobald ein Checkpointer angehängt ist, ist thread_id nicht optional – es ist erforderlich.

Angesichts dessen, was wir gerade gesehen haben, ergibt das Sinn. Bevor ein Node ausgeführt wird, muss der Checkpointer den gespeicherten State laden – und die thread_id ist das, was ihm sagt, welchen Gesprächs-State er laden soll.

Dies ist die Grundform eines Chatbot-Dienstes: ein Graph, ein Checkpointer und eine thread_id pro Benutzer oder pro Chatraum.

17.1.5) Die Grenzen von InMemorySaver und Produktionsalternativen

Wir haben zuvor gesagt, dass InMemorySaver Checkpoints im Speicher hält. Zwei Einschränkungen kommen mit dieser Wahl.

Starten Sie den Prozess neu, und alles ist weg. Deployen Sie den Dienst neu oder fahren Sie den Server wieder hoch, und jedes bisher angesammelte Gespräch verschwindet.

Separate Prozesse können ihn nicht gemeinsam nutzen. Ein echter Dienst verteilt eingehende Anfragen auf mehrere Prozesse. Jeder Prozess hat seinen eigenen Speicher, sodass ein von Prozess A gespeichertes Gespräch für Prozess B unsichtbar ist. Ein Benutzer kann jedes Mal dieselbe thread_id senden und trotzdem zusehen, wie das Gespräch auseinanderfällt, je nachdem, welcher Prozess die Anfrage zufällig aufnimmt.

Deshalb verwendet die Produktion Checkpointer, die Checkpoints in einer Datenbank speichern.

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) – speichert alles in einer einzigen Datei. Eine gute Wahl für einen kleinen Dienst, der auf einem Server läuft, oder für einen lokalen Prototyp.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) – speichert Checkpoints in einem Datenbankserver. Fügen Sie mehr Server hinzu, und jeder Prozess sieht weiterhin dieselben Checkpoints. Dies ist der Checkpointer, den die LangGraph-Dokumentation für die Produktion empfiehlt.

Alle von ihnen implementieren dieselbe Schnittstelle wie InMemorySaver. Ihr Graph-Code, Ihre Nodes und die Art, wie Sie thread_id verwenden, bleiben genau so, wie sie sind. Das Einzige, was sich ändert, ist die Art, wie Sie den Checkpointer erstellen.

create_agent, den Sie in Kapitel 16 kennengelernt haben, verwendet Checkpointer auf die gleiche Weise. Übergeben Sie einen Checkpointer an dessen checkpointer-Parameter und übergeben Sie eine config mit einer thread_id an invoke(). Genau das hat den konversationsbasierten RAG-Agent in Kapitel 11 dazu gebracht, sich an frühere Turns zu erinnern.

Wir verwenden für den Rest dieses Kapitels weiterhin InMemorySaver. Wo auch immer die Checkpoints zufällig liegen, die Art, wie Sie mit ihnen arbeiten, ist dieselbe.

17.2) State inspizieren und Debugging

In 17.1.2 haben wir gesagt, dass der Checkpointer State bei jedem Super-Step speichert. Ziehen wir diese Checkpoints heraus und sehen sie uns an.

Das Öffnen gespeicherter Checkpoints ist keine bloße Neugier. Agents verhalten sich falsch. Sie rufen Tools in einer Schleife auf, die nie endet, sie verlieren den Überblick über frühere Turns, sie nehmen einen Branch, den Sie nie erwartet haben. Um herauszufinden, warum, müssen Sie wissen, wie der State in diesem Moment aussah – und die Checkpoints haben die Antwort.

LangGraph gibt Ihnen zwei Methoden.

• graph.get_state(config) – gibt den neuesten Checkpoint für dieses Gespräch zurück.
• graph.get_state_history(config) – gibt jeden Checkpoint für dieses Gespräch zurück, den neuesten zuerst.

Beide benötigen eine thread_id in der config, da sie wissen müssen, wessen Checkpoints Sie anfordern.

Die Checkpoints, die diese beiden Methoden zurückgeben, werden als StateSnapshot-Objekte dargestellt.

17.2.1) StateSnapshot: Was in einem Checkpoint steckt

Sehen wir uns an, was ein Checkpoint tatsächlich enthält. Wir werden für dieses Beispiel kein LLM verwenden. Ein LLM gibt jedes Mal etwas anderes zurück, was nicht hilfreich ist, wenn wir jedes Feld einzeln durchgehen möchten. Stattdessen bauen wir einen kleinen Graphen, der feste Werte zurückgibt.

Der State hat zwei Arten von Feldern. foo hat keinen Reducer, sodass es überschrieben wird; bar hat einen, sodass es akkumuliert. Es ist dieselbe Anordnung wie beim add_messages-Reducer, den wir in 15.2.2 an messages angehängt haben – hier verwenden wir Pythons operator.add als Reducer, um Listen zu verketten.

from operator import add
from typing_extensions import TypedDict, Annotated
 
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    foo: str                        # kein Reducer → überschreiben
    bar: Annotated[list[str], add]  # add-Reducer → akkumulieren
 
def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}
 
def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}
 
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
 
snapshot = graph.get_state(config)
print(snapshot)

Ausgabe:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                             'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
    metadata={'source': 'loop', 'step': 2, 'parents': {}},
    created_at='2026-07-12T03:17:33.637368+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                                    'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
    tasks=(),
    interrupts=()
)

Acht Felder. Nehmen wir sie eins nach dem anderen.

• values – der State, wie er bei diesem Checkpoint stand. bar ist ['a', 'b'], weil der add-Reducer akkumuliert hat, was beide Nodes zurückgegeben haben; foo ist 'b', weil es keinen Reducer hat, sodass der letzte Schreibvorgang gewonnen hat. Dies ist das Feld, das die Frage beantwortet: „Wie sah der State in diesem Moment aus?"
• next – ein Tupel von Node-Namen, die nach diesem Checkpoint ausgeführt werden sollen. Ein leeres () bedeutet, dass nichts mehr auszuführen ist, d. h. der Graph ist fertig. ('node_b',) bedeutet, dass node_b noch bevorsteht.
• config – die Adresse dieses Checkpoints. thread_id identifiziert das Gespräch, und checkpoint_id identifiziert, welcher Moment innerhalb davon. LangGraph weist die checkpoint_id jedes Mal automatisch zu, wenn es einen Checkpoint speichert.
• metadata – Buchführung über den Lauf. source sagt Ihnen, woher der Checkpoint stammt: "input" bedeutet, dass er aus der Eingabe erstellt wurde, die Sie an invoke() übergeben haben, und "loop" bedeutet, dass er erzeugt wurde, während der Graph lief. step ist die Super-Step-Nummer.
• created_at – wann der Checkpoint gespeichert wurde. Praktisch, wenn Sie die Dinge mit Ihren Logs abgleichen.
• parent_config – die config des Checkpoints unmittelbar vor diesem. Folgen Sie ihr, und Sie können rückwärts durch den Lauf gehen. Sie ist None für den allerersten Checkpoint.
• tasks – die Ausführungsaufzeichnung für die in next aufgeführten Nodes. In dem Moment, in dem ein Checkpoint gespeichert wird, sind diese Nodes noch nicht ausgeführt worden; sobald sie es tun, wird ihr Ergebnis an diesen Checkpoint angehängt. Ein Node, der erfolgreich war, hinterlässt seinen Rückgabewert in result, und ein Node, der fehlgeschlagen ist, hinterlässt seine Ausnahme in error.
• interrupts – wo der Graph pausiert hat, um die Kontrolle an eine Person zurückzugeben. LangGraph kann mitten im Lauf stoppen und warten, dass jemand einen Schritt genehmigt oder einen Wert liefert, und dieses Feld zeichnet diese Pausen auf.

Sie lesen die Felder als Attribute aus. metadata ist ein Dictionary, sodass Sie Werte mit einem Schlüssel daraus abrufen.

snapshot = graph.get_state(config)
 
print(snapshot.values)            # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next)              # ()
print(snapshot.metadata["step"])  # 2

Von diesen ist das, wonach Sie beim Debugging am häufigsten greifen werden, next. Wenn next nicht leer ist, hat der Graph es nicht bis zum Ende geschafft – er hat irgendwo in der Mitte gestoppt. Und wenn wir in 17.3 einen fehlgeschlagenen Graphen fortsetzen, ist dieses Feld der Punkt, an dem wir beginnen.

17.2.2) Die Checkpoint-Historie durchlaufen

get_state() zeigt Ihnen nur den neuesten Checkpoint. Aber Debugging bedeutet oft, zu fragen: „Wie sind wir hier gelandet?" – und dafür brauchen Sie die gesamte Verlaufskurve des Laufs. get_state_history() liefert sie Ihnen.

for snap in graph.get_state_history(config):
    print(f"step={snap.metadata['step']:>2}  "
          f"next={str(snap.next):<16}  values={snap.values}")

Ausgabe:

step= 2  next=()                values={'foo': 'b', 'bar': ['a', 'b']}
step= 1  next=('node_b',)       values={'foo': 'a', 'bar': ['a']}
step= 0  next=('node_a',)       values={'foo': '', 'bar': []}
step=-1  next=('__start__',)    values={'bar': []}

Der neueste Checkpoint kommt zuerst, arbeiten Sie also von unten nach oben, um dem Lauf in der Reihenfolge zu folgen.

• step -1 – direkt nachdem invoke() die Eingabe erhalten hat. Beachten Sie, dass das {"foo": "", "bar": []}, das wir übergeben haben, nicht in values erscheint. Das Einbringen der Eingabe in den State ist selbst eine Stufe, und diese Stufe ist noch nicht gelaufen. Das __start__ in next ist der interne Node, der das erledigt.

  bar erscheint als [], aber das ist nicht der Wert, den wir übergeben haben. Ein Feld mit einem Reducer beginnt mit einem leeren Wert, in den sich Schreibvorgänge akkumulieren können. foo hat keinen Reducer, sodass es überhaupt keinen Startwert hat – weshalb es hier nicht erscheint.

• step 0 – __start__ ist gelaufen und die Eingabe ist nun im State. foo='' und bar=[] sind die Werte, die wir übergeben haben. node_a steht als Nächstes an.
• step 1 – das Ergebnis der Ausführung von node_a. foo ist nun 'a' und bar ist ['a'], mit node_b als Nächstem.
• step 2 – das Ergebnis der Ausführung von node_b. next ist leer, sodass der Graph fertig ist.

Ein nicht leeres next bei step 0 oder step 1 bedeutet nicht, dass der Graph dort gestoppt hat. Ein Checkpoint, der aufgenommen wird, während der Graph noch läuft, hat naturgemäß einen Node als Nächstes eingereiht. Als 17.2.1 sagte „ein nicht leeres next bedeutet, dass der Graph gestoppt hat", ging es um den Checkpoint an dem Punkt, an dem etwas schiefgelaufen ist. In der Mitte der Historie zeigt next Ihnen einfach nur, welchen Pfad der Graph genommen hat.

17.3) Ab dem Fehlerpunkt fortsetzen

Da der State am Ende jedes Super-Steps gespeichert wird, nimmt ein Fehler mitten im Lauf die abgeschlossene Arbeit nicht mit – sie ist immer noch in den Checkpoints vorhanden. Es gibt keinen Grund, von vorne zu beginnen. Sie machen dort weiter, wo es gestoppt hat.

17.3.1) Fortsetzen mit invoke(None, config)

Das Fortsetzen ist einfach: Übergeben Sie None dort, wo die Eingabe hingehört.

graph.invoke(None, config)

Es bedeutet „es gibt keine neue Eingabe; setze vom gespeicherten State fort." Die config benötigt natürlich immer noch eine thread_id, da LangGraph wissen muss, welches Gespräch fortgesetzt werden soll.

Inszenieren wir einen Fehler und setzen von ihm fort. Wir bauen einen Zwei-Node-Graphen, dessen zweiter Node nur bei seinem ersten Lauf fehlschlägt. Er muss beim erneuten Versuch erfolgreich sein, sonst würden wir das Fortsetzen nie tatsächlich funktionieren sehen.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    step_1_done: bool
    step_2_done: bool
 
first_try = True   # Flag, um den ersten Lauf fehlschlagen zu lassen
 
def step_1(state: State):
    print("step_1 läuft (aufwändige Arbeit)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 fehlgeschlagen (API-Timeout)")
        raise RuntimeError("External API timed out")
    print("step_2 läuft")
    return {"step_2_done": True}
 
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
 
try:
    graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
    print("Fehlgeschlagen:", e)

Ausgabe:

step_1 läuft (aufwändige Arbeit)
step_2 fehlgeschlagen (API-Timeout)
Fehlgeschlagen: External API timed out

step_1 war erfolgreich und step_2 hat eine Ausnahme ausgelöst. Finden wir heraus, wo der Graph gestoppt hat, mit dem get_state(), das wir in 17.2 gelernt haben.

snapshot = graph.get_state(config)
print("next   =", snapshot.next)
print("values =", snapshot.values)

Ausgabe:

next   = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}

next ist ('step_2',), was uns sagt, dass der Graph mitten in der Ausführung von step_2 gestoppt hat. Und in values ist step_1_done True – das Ergebnis von step_1 ist immer noch im Checkpoint vorhanden.

Nun setzen wir mit None fort.

result = graph.invoke(None, config)
print("Final =", result)

Ausgabe:

step_2 läuft
Final = {'step_1_done': True, 'step_2_done': True}

step_1 läuft (aufwändige Arbeit) wurde nie ausgegeben. step_1 lief kein zweites Mal. LangGraph hat den gespeicherten State geladen und bei step_2 weitergemacht. Wir haben nicht zweimal für diesen aufwändigen ersten Schritt gezahlt.

17.3.2) Worauf beim Fortsetzen zu achten ist

Wenn Sie fortsetzen, läuft der Node, der fehlgeschlagen ist, erneut. Wenn dieser Node ein LLM aufruft oder eine externe API anspricht, geschehen diese Aufrufe ebenfalls erneut – und sie könnten mit etwas anderem zurückkommen.

Da liegt die Falle. Wenn step_2 eine E-Mail gesendet hat und dann fehlgeschlagen ist, sendet das Fortsetzen eine zweite E-Mail. LangGraph garantiert nur, dass es die Nodes, die erfolgreich waren, nicht erneut ausführt.

Daher muss jeder Node, der möglicherweise erneut läuft, idempotent sein: dasselbe zweimal zu tun, sollte Sie an denselben Ort führen. Prüfen Sie, ob die E-Mail bereits versandt wurde, bevor Sie sie senden; setzen Sie einen eindeutigen Schlüssel auf die Datenbanktabelle, damit ein doppeltes Insert nicht passieren kann.