Langlebige KI-Agenten: Zustand speichern mit LangGraph, DynamoDB und Cosmos DB

Ein Agent, der sich nicht merken kann, wo er aufgehört hat, ist nur eine Demo. Sobald die Idee des persistierten Zustands klar ist, beginnen die praktischen Fragen: Wo liegt dieser Zustand, wie findet ein Workflow ihn wieder, und was verdient es überhaupt, gespeichert zu werden? Dieser Beitrag beantwortet sie anhand von LangGraph und den beiden großen Clouds.

Der Plan: Erstens, wo ein Checkpoint in Produktion hingehört – an einen robusteren Ort als die lokale Festplatte, nämlich in AWS- oder Azure-Dienste. Zweitens, wie LangGraph die Fortsetzung über thread_id und austauschbare checkpointer-Backends regelt. Drittens, welche Teile des Zustands es wert sind, behalten zu werden, und welche sich billiger neu aufbauen lassen.

So wenden Sie es in einem Projekt an:

1. Einen Checkpointer passend zum Deployment wählen: lokale Entwicklung, eine einzelne Instanz oder mehrere Instanzen.
2. Ab dem ersten Tag eine thread_id an jeden graph.invoke()-Aufruf übergeben.
3. Jedes Feld des Zustandsmodells nach der Regel gegen Ende als „speichern“ oder „neu berechnen“ einordnen.

Ein WorkflowState-Modell nützt nur, wenn etwas es dort ablegt, wo ein Pod-Neustart es nicht löschen kann. Um dieses Etwas geht es als Nächstes.

Wo der Checkpoint liegt

Jedes Diagramm eines langlebigen Agenten hat einen Kasten mit der Aufschrift „Zustand speichern“. In Produktion ist dieser Kasten eine Architekturentscheidung, denn der Speicherort bestimmt, welche Ausfälle der Checkpoint übersteht.

Die lokale Festplatte übersteht einen abgestürzten Prozess. Sie übersteht keinen Container, der auf einen anderen Node verschoben wird, keinen gekillten und ersetzten Pod und keinen Autoscaler, der eine zweite Instanz startet, die von der ersten nichts weiß. Läuft Ihr Agent auch nur in der Nähe von Kubernetes – und dort landen die meisten irgendwann –, geben lokale Dateien eher ein falsches Sicherheitsgefühl als eine Strategie.

Also wandern Checkpoints in den Cloud-Speicher. Auf AWS ist die übliche Kombination ein schneller Key-Value-Store für den Zeiger und ein Objektspeicher für die Masse:

• DynamoDB hält Checkpoint-Metadaten und kleine Zustandsobjekte, mit task_id oder thread_id als Schlüssel und schnellen Lesezugriffen, wenn eine Aufgabe fortgesetzt wird.
• S3 speichert die großen Blobs, etwa lange Gesprächsverläufe oder angesammelte Tool-Ausgaben, die keinen Zugriff im Millisekundenbereich brauchen.

Azure bietet eine entsprechende Aufteilung:

• Cosmos DB oder Azure Table Storage für Metadaten und schnelle Abfragen.
• Azure Blob Storage für größere Nutzlasten.
• Entra ID, um zu steuern, welche Identitäten und Service Principals einen Checkpoint lesen oder schreiben dürfen – wichtig, sobald der Zustand etwas Sensibles enthält.

Die Abstraktion sieht auf jedem Backend gleich aus, nur der Client ändert sich:

from typing import Protocol

class Checkpointer(Protocol):
    def get(self, task_id: str) -> dict | None: ...
    def put(self, task_id: str, state: dict) -> None: ...class DynamoDBCheckpointer:
    def __init__(self, table):
        self.table = table    def get(self, task_id: str) -> dict | None:
        item = self.table.get_item(Key={"task_id": task_id}).get("Item")
        return item["state"] if item else None    def put(self, task_id: str, state: dict) -> None:
        self.table.put_item(Item={"task_id": task_id, "state": state})class AzureBlobCheckpointer:
    def __init__(self, container_client):
        self.container = container_client    def get(self, task_id: str) -> dict | None:
        blob = self.container.get_blob_client(f"{task_id}.json")
        if not blob.exists():
            return None
        return json.loads(blob.download_blob().readall())    def put(self, task_id: str, state: dict) -> None:
        self.container.upload_blob(
            f"{task_id}.json", json.dumps(state), overwrite=True
        )

Wer gegen ein Checkpointer-Protokoll programmiert und nicht gegen das SDK eines Anbieters, hält die Funktion load_or_initialize_state() unabhängig vom Speicher. Man tauscht die Implementierung aus und lässt die Workflow-Logik in Ruhe.

Zwei Kompromisse für die Produktion verschweigen Tutorials meist:

• Schreibhäufigkeit gegen Latenz. Nach jedem Node zu speichern ist am sichersten und am langsamsten, weil jeder Schreibvorgang ein Netzwerk-Roundtrip ist. Echte Systeme speichern meist nach bedeutsamen Übergängen, etwa einem abgeschlossenen Tool-Aufruf oder einer menschlichen Freigabe, und überspringen die internen Schritte dazwischen.
• Kosten pro Checkpoint. Eine Pipeline mit fünf Nodes, die an jedem Node speichert und täglich über Tausende Aufgaben läuft, erzeugt Tausende Schreibvorgänge, bevor überhaupt ein Ergebnis vorliegt. Planen Sie das ein wie LLM-Tokens, denn im großen Maßstab ist es ein vergleichbarer Posten auf der Rechnung.

Umsetzung in LangGraph

Bisher ging es um Architektur. In LangGraph wird daraus Code mit erstaunlich wenigen Zeilen.

Thread-IDs dienen als Schlüssel für die Fortsetzung. LangGraph erkennt einen bestimmten Lauf – ein Gespräch, einen Bericht, eine Aufgabe – an seiner thread_id und nicht an der Definition des Graphen. Stellen Sie sich den Graphen als Bauplan vor und den Thread als eine konkrete Instanz davon. Jeder Aufruf erhält die ID über config:

config = {"configurable": {"thread_id": "task_123"}}

# First call - no prior state for this thread_id, so LangGraph cold-starts it
result = graph.invoke({"task_id": "task_123"}, config)# Later call, same thread_id - LangGraph loads the last checkpoint
# and resumes instead of starting over
result = graph.invoke(None, config)

Der zweite Aufruf mit None als Eingabe erledigt, was load_or_initialize_state() von Hand gemacht hat: Nachschlagen und Verzweigen übernimmt der Checkpointer. Verwenden Sie dieselbe thread_id erneut, geht die Ausführung mitten im Graphen weiter, eine neue ID löst einen Kaltstart aus.

Checkpointer sind die austauschbare Speicherschicht. Sie hängen direkt mit den Cloud-Optionen oben zusammen. LangGraph definiert eine checkpointer-Schnittstelle, und das übergebene Objekt bestimmt, wo der Zustand eines Threads liegt:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# Local dev / tests - gone when the process exits
checkpointer = MemorySaver()# Single-instance production - survives process restarts
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")# Multi-instance production - survives pod restarts, works across replicas
checkpointer = PostgresSaver.from_conn_string(POSTGRES_URL)graph = builder.compile(checkpointer=checkpointer)

MemorySaver reproduziert das Problem der lokalen Festplatte exakt. Für Demos taugt er, mit dem Ende des Prozesses ist alles weg. PostgresSaver oder eine eigene Klasse auf Basis von DynamoDB oder Cosmos DB, aufgebaut wie das Checkpointer-Protokoll oben, lässt einen Thread einen Pod überleben, der auf einen anderen Node verschoben wird.

Stellen Sie sich das ganze System als einen Fluss vor: eine Verzweigung zwischen Kaltstart und Fortsetzung, eine Pipeline mit fünf Nodes, der Checkpointer als gemeinsame Lese- und Schreibschicht und drei Speicher-Backends, auf die er zeigen kann.

Nicht alles speichern

Noch ein Kompromiss, den Tutorials selten behandeln. Ein Checkpoint sollte enthalten, was teuer oder unmöglich neu zu erzeugen ist – nicht alles, was zu diesem Zeitpunkt im Graphen zufällig im Speicher lag.

Speichern Sie:

• Ergebnisse von Tool-Aufrufen, die Geld kosten oder ein Rate Limit verbrauchen
• Freigabeentscheidungen von Menschen
• den aktuellen Schritt bzw. die Node-Position
• alles, von dem ein Mensch erwartet, dass das System es sich merkt

Berechnen Sie neu:

• Prompt-Strings, die aus Templates und Zustand gebaut werden
• abgeleitete Werte wie eine Wortzahl, die sich aus anderem Zustand ergeben
• alles, was ein frischer Funktionsaufruf in Millisekunden fast kostenlos neu aufbaut
• abgerufene Dokumente, solange Sie ihre IDs speichern, denn die IDs brauchen Sie, um sie erneut zu holen

Die Frage, nach der ein Feld eingeordnet wird, ist einfach. Würde das Fortsetzen ohne dieses Feld im Checkpoint echtes Geld, echte Verzögerung oder einen wiederholten Seiteneffekt kosten? Oder wäre die Neuerzeugung ein einziger billiger Funktionsaufruf? Ersteres speichern, Letzteres neu aufbauen. Ein mit neu berechenbaren Daten vollgestopfter Checkpoint liest sich langsamer, schreibt sich langsamer und kostet mehr Speicher – und bringt im Gegenzug keine zusätzliche Ausfallsicherheit.