LangGraph Persistenz Teil 2: PostgresSavers, HITL und Produktions-Backends
Verlegen Sie LangGraph von InMemorySaver zu PostgresSaver mit PostgreSQL-Pools und asynchronen Saverns, vergleichen Sie SQLite und Redis sowie führen Sie Unterbrechen und Wiederaufnahme im Produktivumfeld durch.
Von RAM-Checkpointen zu langlebigen Backend-Lösungen
Teil eins erklärte, warum die Persistenz in LangGraph notwendig ist – Zustand, Checkpointen, Checkpointer, Threads und InMemorySaver. In diesem Teil werden dieselben Konzepte für die Produktion bereitgestellt. In-Memory-Saver verschwinden beim Neustart des Prozesses; echte Agenten benötigen eine Speicherlösung, die länger besteht als der Python-Prozess.
PostgresSaver für Produktionsbeständigkeit
Eine Datenbank befindet sich außerhalb des Anwendungsprozesses. Abstürze, Neuverteilungen und Maschinenwechsel lassen die Checkpoint-Geschichte unberührt. PostgresSaver schreibt diese Checkpointen in PostgreSQL statt in den RAM.
PostgreSQL ist eine gängige Wahl, weil es langlebig ist, weit verbreitet als verwaltete Cloud-Dienste angeboten wird, sicher für konkurrierende Anwendungsserver ist und mit steigendem Traffic sowie der Menge an Checkpointen wachsen kann.
Installieren Sie das spezielle Paket:
pip install langgraph-checkpoint-postgres
Einfache Einrichtung:
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# This creates the required tables the first time you use it.
# Only needs to run once per database.
checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "session-99"}}
result = graph.invoke(
{"messages": [{"role": "user", "content": "Hello"}]},
config=config,
)
from_conn_string eröffnet eine vom Kontext verwaltete Verbindung. setup() erstellt einmalig Tabellen wie checkpoints und checkpoint_blobs – betrachten Sie dies als Migration und nicht als Vorgang pro Anfrage. compile(checkpointer=...) verwendet dieselbe API wie der In-Memory-Modus; ein Wechsel der Backend-Plattform führt nicht zur Neuverwaltung des Graphen.
Verbindungs-Pools
Die Eröffnung einer neuen Verbindung pro Anfrage ist zeitaufwändig. Verwenden Sie lieber einen Pool mit bereits verfügbaren Verbindungen:
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:password@localhost:5432/mydb"with ConnectionPool(DB_URI) as pool:
checkpointer = PostgresSaver(pool)
checkpointer.setup() # one-time setup
graph = builder.compile(checkpointer=checkpointer)
Asynchrone Variante
Apps im FastAPI-Stil sollten AsyncPostgresSaver verwenden:
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
await checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)
Falls Sie psycopg-Verbindungen manuell erstellen, setzen Sie autocommit=True und row_factory=dict_row; andernfalls könnte setup() Tabellen nicht persistent speichern. Halten Sie thread_id unter 255 Zeichen – UUIDs sind ein sicheres Standardwert.
Weitere Backend-Systeme
SQLite
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
Eine einzige lokale Datei überdauert Neustarts und eignet sich für Prototypen oder einprozessige Skripte. Durch Dateilocking eignet sie sich jedoch nicht für Web-Anwendungen mit mehreren Arbeitsprozessen.
Redis
Schnell, memory-basiert mit optionaler Festplattenspeicherung – nützlich für hochleistungsstarke Chat-Produkte mit vielen gleichzeitigen Threads.
Gemanagter Cloud-Postgres
RDS, Cloud SQL, Supabase, Neon und ähnliche Dienste verwenden standardmäßig PostgreSQL; daher funktioniert PostgresSaver, während der Anbieter die Backups und den Betriebsstatus übernimmt.
Kundenspezifische Checkpointer
Erstelle eine Unterklasse von BaseCheckpointSaver und implementiere put, get_tuple, list (sowie die asynchronen Entsprechungen), wenn eine interne Speicherung erforderlich ist.
Schnelle Auswahl
Lokaler, dauerhafter Script-Code → SQLite. Produktion mit mehreren Arbeitern → Postgres/AsyncPostgres. Extrem hohes Konkurrenzniveau bei Checkpoint-Transaktionen → Redis. Bestehender, verwalteter Postgres → cloud-basierter Postgres. Erzwungene interne Speicherung → benutzerdefinierter Speicherer.
Fünf schrittweise Beispiele
Beispiel 1 – keine Persistenz (Amnesie)
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
count: intdef increment(state: State) -> dict:
return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)graph = builder.compile() # No checkpointer passed in!print(graph.invoke({"count": 0})) # {'count': 1}
print(graph.invoke({"count": 0})) # {'count': 1} again — no memory of the last call!
{'count': 1}
{'count': 1}
Der zweite Aufruf kann den ersten nicht erkennen: Es gibt weder einen Checkpointer noch eine Thread-ID.
Beispiel 2 – Kontinuität mit InMemorySaver
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class State(TypedDict):
count: intdef increment(state: State) -> dict:
return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "counter-1"}}print(graph.invoke({"count": 0}, config)) # {'count': 1}
print(graph.invoke({"count": 0}, config)) # {'count': 2}
{'count': 1}
{'count': 2}
Die gleichen Eingaben setzen nun an dem vorherigen Checkpoint fort, anstatt neu zu starten.
Beispiel 3 – Wiederaufnahme nach einem simulierten Absturz
from langgraph.checkpoint.memory import InMemorySaver
# Shared storage dictionary simulates a database that survives across "restarts"
shared_storage = {}checkpointer_1 = InMemorySaver()
checkpointer_1.storage = shared_storage # pretend this is durable storagegraph = builder.compile(checkpointer=checkpointer_1)
config = {"configurable": {"thread_id": "resume-demo"}}graph.invoke({"count": 0}, config) # count becomes 1
print("Process 'crashes' here...")# Simulate a restart: a brand new checkpointer object, same underlying storage
checkpointer_2 = InMemorySaver()
checkpointer_2.storage = shared_storage
graph_after_restart = builder.compile(checkpointer=checkpointer_2)# Resume using the SAME thread_id
result = graph_after_restart.invoke({"count": 0}, config)
print(result) # {'count': 2} — continued, did NOT reset to 0
Das Teilen eines im Prozess vorhandenen Wörterbuchs simuliert einen Neustart; in der Produktion weist man einfach einen neuen Prozess auf dieselbe Postgres-Datenbank hin und setzt den Ablauf mit thread_id fort.
Beispiel 4 – Mensch im Prozess mit interrupt()
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class State(TypedDict):
task: str
approved: booldef request_approval(state: State) -> dict:
decision = interrupt(f"Approve this task? '{state['task']}' (yes/no)")
return {"approved": decision == "yes"}builder = StateGraph(State)
builder.add_node("request_approval", request_approval)
builder.add_edge(START, "request_approval")
builder.add_edge("request_approval", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "approval-1"}}# Step 1: Start the graph — it will pause at interrupt()
result = graph.invoke({"task": "Send $500 refund", "approved": False}, config)
print(result) # Contains an '__interrupt__' key with our question# ... hours later, a human reviews and approves ...# Step 2: Resume with the human's answer
final_result = graph.invoke(Command(resume="yes"), config)
print(final_result) # {'task': 'Send $500 refund', 'approved': True}
interrupt(...) pausiert den Ablauf und erstellt Checkpoints. Der Aufrufer sieht die übermittelten Daten unter __interrupt__. Später setzt Command(resume=...) den Ablauf fort. Beim Wiederaufnehmen läuft LangGraph den gesamten Knoten erneut von oben durch – Nebeneffekte bleiben nach der Unterbrechung bestehen, nicht davor.
Beispiel 5 – Chatbot mit Postgres-Unterstützung
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
class State(TypedDict):
messages: listdef chatbot_node(state: State) -> dict:
last_message = state["messages"][-1]
reply = f"You said: {last_message}"
return {"messages": state["messages"] + [reply]}builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # run once
graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-42"}}
result = graph.invoke({"messages": ["Hi there"]}, config)
print(result)
{'messages': ['Hi there', 'You said: Hi there']}
Stoppen Sie das Skript, starten Sie es morgen mit derselben Thread-ID – die Konversation setzt sich fort, da die Datenbank die Checkpoints speichert.
Warum HITL von Persistenz abhängt
Drei Komponenten: interrupt() zur Pause mit einem Payload, unbegrenztes Warten ohne CPU-Auslastung, während der Zustand im Speicher bleibt, sowie Command(resume=...) zur Fortsetzung. Ohne Checkpoint gibt es keinen Ort, an dem der eingefrorene Zustand gespeichert werden kann.
Echte Genehmigungsprozesse – Kreditentscheidungen über einem Schwellenwert, klinische Empfehlungen vor der Dokumentation, Überprüfung rechtlicher Klauseln, Ausgabenüberschreitungen – dauern oft Minuten bis Wochen. Durch Persistenz wird dieses Warten kostengünstig und sicher.
Persistenz über Agentfähigkeiten hinweg
Lange Gespräche, mehrstufige Pläne, die auch nach Abstürzen erhalten bleiben, Tool-Loops, die vorherige Ergebnisse nicht vergessen dürfen, sowie Workflows mit Branching und Wiederverbindung setzen alle auf zuverlässige Checkpoints. Thread-IDs sind die Identifikatoren, die von Operatoren und UIs verwendet werden, um den richtigen Ablauf fortzusetzen.
Fehler, die immer wieder auftauchen
- Aufruf von
setup()bei jeder Anfrage.
thread_id.interrupt().thread_id zu übergeben.Praktiken, die gut funktionieren
Führen Sie die Schema-Einrichtung als Migration durch. Verwenden Sie Pools. Nutzen Sie asynchrone Speicherfunktionen in asynchronen Anwendungen. Halten Sie Thread-IDs kurz und konstant. Dokumentieren Sie, welche Umgebungen welchen Backend verwenden. Fügen Sie Smoke-Tests hinzu, die Abstürze simulieren und wieder aufnehmen. Überwachen Sie das Wachstum sowie die Speicherung der Checkpoint-Tabelle.
Zusammenfassung
Die Persistenz von LangGraph in der Produktion besteht im Wesentlichen aus „derselben Graph-API, unterschiedlichem dauerhaftem Speicher“. Postgres ist die Standardwahl für Unternehmen; SQLite und Redis besetzen spezielle Nischen; eigene Speichermethoden decken vorgeschriebene Infrastrukturen ab. Menschliche Einmischung und Wiederherstellung nach Ausfällen sind keine optionalen Funktionen – sie sind der Grund für die Existenz von Checkpoint-Systemen. Verbinden Sie den Backend frühzeitig, damit Unterbrechungen und Wiederaufnahmen tatsächlich möglich sind, bevor der Agent mit den Nutzern in Kontakt tritt.
Betriebshandbuch für Checkpoint-Backends
Behandeln Sie den Speicher für Checkpoints wie die primären Daten der Anwendung. Backup-Pläne, Wiederherstellungsübungen und Aufbewahrungsrichtlinien gehören zu denselben Gesprächen wie Kundendaten. Ein Chat-Produkt, das nach einem regionalen Failover keine Gesprächsverläufe wiederherstellen kann, hat eigentlich kein echtes Gedächtnis geliefert – es hat nur Hoffnung geliefert.
Die Kapazitätsplanung sollte Kontrollpunkte pro aktiverem Thread pro Tag, die durchschnittliche Größe der Blöcke in Zuständen mit vielen Nachrichten sowie das Wachstum bei hoher gleichzeitiger Nutzung erfassen. Große Tool-Pakete, die im Zustandsspeicher gespeichert werden, dominieren die Größe von Postgres bereits lange bevor die Anzahl der Zeilen beunruhigend wird. Es ist besser, umfangreiche Daten in Objektspeichern zu lagern und nur Pointer im Zustandsspeicher zu behalten.
Der Zugriffskontroll ist wichtig: Der Datenbankbenutzer für die Kontrollpunkte sollte nicht dieselbe Rolle sein wie die, die mit umfangreichen DDL-Befehlen unverwandte Schemata migriert. Wechseln Sie regelmäßig die Zugangsdaten, bevorzugen Sie IAM-Authentifizierung, sofern dies von Cloud-Anbietern ermöglicht wird, und überprüfen Sie, wer den SELECT-Befehl auf den Konversationszustand ausführen kann. Thread-IDs beziehen sich oft auf Benutzer; betrachten Sie Ausdrucke solcher Daten als personenbezogene Informationen.
Beim Umstieg von InMemorySaver auf Postgres in einem Live-System ist eine doppelte Schreibvorgangssteuerung nur selten notwendig, sofern man akzeptiert, dass die RAM-Threads während des Vorgangs einmal zurückgesetzt werden. Dokumentieren Sie ein Wartungsfenster, implementieren Sie den neuen Checkpointer und überwachen Sie die Fehlerraten nach Wiederaufnahme. Bei Blue/Green-Umstellungen stellen Sie sicher, dass beide Umgebungen auf dieselbe Datenbank verweisen, andernfalls entstehen Inkonsistenzen in den Thread-Historien.
Tipps für Tests: Führen Sie Postgres in CI (oder in einem einmaligen Container) aus, rufen Sie setup() einmal auf, führen Sie die fünf Beispielstrukturen als Integrationstests aus und überprüfen Sie, ob ein zweiter Prozess einen von dem ersten erstellten Thread fortsetzen kann. Dieser einzige Test zwischen Prozessen erfasst mehr Fehler in der Produktion als jedes In-Memory-Unit-Test.
Zum Schluss sollten die On-Call-Mitarbeiter den Unterschied zwischen „Graph failed“ und „Checkpointer unreachable“ verstanden werden. Die Symptome ähneln sich – es tritt ein Fehler auf – doch die Lösungswege unterscheiden sich. Dashboards, die Anwendungsfehler von Datenbank-Timeouts trennen, sparen während von Vorfällen Stunden ein.
Thread-Identität und Produkt-UX
Thread-IDs sind nicht nur Datenbank-Schlüssel; sie sind auch Produktobjekte. Verknüpfen Sie sie bewusst mit Benutzer-Sessions, Support-Tickets oder Dokument-IDs. Stellen Sie über die Wiederverwendung der ID eine Funktion „Konversation fortsetzen“ in der Benutzeroberfläche bereit. Erstellen Sie eine neue ID, um „von vorne anfangen“ zu ermöglichen. Die Kombination dieser Funktionen in derselben ID führt zu problematischen Chats, in denen alte Tool-Ergebnisse erneut auftauchen.
Für Nutzer mit mehreren Geräten erfordert die Kontinuität der Thread-IDs über verschiedene Clients authentifizierungsbedingte Abfragen. Akzeptieren Sie niemals eine vom Client bereitgestellte Thread-ID, ohne deren Eigentumsverhältnis zu überprüfen. Checkpointers speichern alles, was man ihnen gibt; sie stellen keine Eigentumsregeln durch.
Schema-Entwicklung für geparkte Zustände
State TypedDicts entwickeln sich weiter. Das Hinzufügen optioneller Felder ist einfach; das Umbenennen oder Entfernen von Feldern zerstört alte Checkpoints. Bevorzugen Sie additive Änderungen, Leser, die fehlende Schlüssel standardmäßig ersetzen, sowie explizite Migrationsstrategien bei zerstörerischen Änderungen. Speichern Sie eine schema_version im Zustand und nutzen Sie diese bei Bedarf für die Logik der Branchen-Node.
Große Nachrichtenlisten erfordern Strategien zum Kürzen: Fassen Sie ältere Nachrichten in ein Feld zusammen, behalten Sie aktuelle Rohnachrichten bei und testen Sie das Fortsetzen nach dem Kürzen. Eine unbegrenzte Historie ist der häufigste Grund dafür, dass die Checkpoint-Tabellen in Postgres überladen werden.
UX-Muster bei Unterbrechungen
Die Freigabenummernschilder sollten die Unterbrechungsinhalte, den relevanten Zustandsausschnitt sowie Fristen anzeigen. Speichern Sie die Identität des Freigabeberechtigten im Zustand beim Wiederaufnehmen. Falls Freigaben ablaufen, führen Sie einen Überwachungsdienst aus, der entweder annulliert oder die Angelegenheit weiterleitet, anstatt die Threads endlos pausieren zu lassen. Die Persistenz ermöglicht lange Wartezeiten; die Produktrichtlinien müssen diese begrenzen.
Katastrophenwiederherstellungsübungen
Quartalsweise soll eine Sicherungskopie in das Staging-Umfeld wiederhergestellt und eine bekannte Thread-ID wiederaufgenommen werden. Überprüfen Sie, ob Nachrichten, Status der Unterbrechung sowie Tool-Überreste übereinstimmen. Dokumentieren Sie das RPO/RTO für die Checkpoint-Datenbank zusammen mit der primären Anwendungsdatenbank. Agenten, die sich an Benutzer erinnern, haben dieselben Erwartungen an Kontinuität wie E-Mail-Systeme.
Leistungshinweise
Verbindungspools sollten nach der Anzahl der gleichzeitigen Graph-Worker, nicht nur nach den Web-Handlern, dimensioniert werden. Achten Sie auf N+1-Muster, bei denen jeder Knoten unnötigerweise einen Checkpoint schreibt – passen Sie die Verwendung der Checkpointer an die Standardeinstellungen der LangGraph-Version an. Komprimieren oder externisieren Sie umfangreiche Ausgaben binärer Tools.
Sicherheitshinweise
Chiffrieren Sie Festplatten, verwenden Sie TLS für Postgres, beschränken Sie die Netzwerkpfade von den Workern zum Speicher und entfernen Sie sensible Daten aus dem Zustand vor dem Checkpoint, falls Tools jemals Rohschlüssel sehen. Überlegen Sie sich Middleware zur Löschung von Daten auf Feldebene. Bei Compliance-Prüfungen werden Fragen gestellt – bereiten Sie Antworten mit Diagrammen des Datenflusses von Benutzer → Graph → Checkpointer → Festplatte vor.
Diese Betriebslagen verwandeln die Codebeispiele in ein System, auf das man sich verlassen kann. Die APIs sind einfach; die dahinterliegende Disziplin ist das eigentliche Produktivitätsmerkmal.
Koordination von Checkpointern mit horizontaler Automaskierung
Wenn Kubernetes die Graph-Worker von zwei Pods auf zwanzig erweitert, muss jeder Pod mit demselben dauerhaften Checkpointer kommunizieren. Sticky Sessions sind unnötig, wenn die Thread-IDs mit jeder Anfrage übertragen werden. Warm Pools sollten nicht von lokalen SQLite-Dateien ausgehen. Health Checks sollten die Datenbankverbindung überprüfen, nicht nur den Prozesszustand. Während von Deployments ausgeführt wird, sollten die abgeschalteten Worker entweder fertiggestellt oder während des Laufs gecheckpt werden, damit Unterbrechungen wieder aufgenommen werden können. Führen Sie Chaos-Tests durch, indem Sie Pods während der interrupt()-Wartezeiten beenden und die Wiederaufnahme auf einem anderen Pod überprüfen. Dieses einzige Experiment validiert die Architektur besser als ein Dutzend Unit-Tests.
Dokumenten-Backpressure: Wenn der Checkpoint-Speicher langsamer wird, sollten die Worker Last abgeben oder in einer Warteschlange warten? Es ist besser, einen expliziten Fehler zu melden, anstatt stille In-Memory-Workarounds zu verwenden, die zwischen den Pods nicht konsistent sind. Operator dürfen in der Produktion niemals „vorübergehend“ auf InMemorySaver wechseln, um die Latenz zu beheben – dieser Wechsel zerstört das Produktversprechen.
Checkliste, bevor man die Persistenz als abgeschlossen betrachtet
Bestätigen Sie, dass setup() als Migration ausgeführt wird; Pools sind konfiguriert; Asynchronität und Synchronisation werden dem Web-Framework angepasst; die Eigentumsverhältnisse der Thread-IDs werden durchgesetzt; Unterbrechungen und Wiederaufnahmen wurden über Prozesse hinweg getestet; Backups konnten einmal erfolgreich wiederhergestellt werden; die Retentionsaufgaben sind dokumentiert; die Handhabung personenbezogener Daten wurde überprüft; Dashboards unterscheiden zwischen Datenbankfehlern und Graphenfehlern; Runbooks verknüpfen Thread-IDs mit den Tools für Benutzerunterstützung. Wenn jede Kategorie abgehakt ist, ist PostgresSaver (oder Ihr gewählter Backend) nicht mehr nur ein Codeausschnitt – er ist Teil der Speichergarantie des Produkts. Bis dahin sollten Demos weiterhin als Demos betrachtet werden und Feature-Flags für langlaufende HITL-Flüsse deaktiviert bleiben.
Eine kurze Geschichte über ein gescheitertes Wiederaufnehmen
Ein On-Call-Ingenieur sah einmal, wie Genehmigungen stundenlang blockiert blieben. Der Graph funktionierte einwandfrei; das Passwort des Checkpointers wurde regelmäßig geändert, doch die Mitarbeiter speicherten weiterhin alte Zugangsdaten. Neue Unterbrechungen scheiterten, während alte Threads weiterhin Daten lasen. Die Lösung bestand aus einem Aktualisieren der Zugangsdaten sowie einer Anweisung im Handbuch: Die Authentifizierung des Checkpointers vor dem Neustart der Graph-Pods überprüfen. Bewahren Sie diese Lektion zusammen mit Ihren PostgresSaver-Beispielen auf, damit bei der nächsten Änderung kein Ausfall mehr auftritt. Zu dauerhafter Erinnerung gehört auch das operative Wissen – die menschlichen Erfahrungen, die in Handbüchern festgehalten werden.
Dieser Vorfall machte außerdem die Überwachung klarer: Es sollten Alerts bei Fehlern bei der Authentifizierung des Checkpointers als eigene Seite ausgegeben werden, nicht in einen allgemeinen 500-Fehler-Kanal. Wenn das Gedächtnis das Produkt ist, dann ist der Datenbankpfad eine Abhängigkeit ersten Grades. Mit dieser Disziplin wird das Ersetzen von InMemorySaver durch PostgresSaver nicht mehr zu einer beängstigenden Migration, sondern zu einem routinemäßigen Schritt auf dem Weg von Notizbüchern zu Diensten.