Persistance de LangGraph, partie 2 : PostgresSavers, HITL et backends en production

Des points de contrôle en RAM aux systèmes de stockage durables

La première partie expliquait pourquoi la persistance dans LangGraph est nécessaire : l’état, les points de contrôle, les outils de sauvegarde, les threads et InMemorySaver. Cette partie rend ces mêmes concepts prêts à être utilisés en production. Les outils de sauvegarde en mémoire disparaissent à la redémarrage du processus ; les agents réels ont besoin d’un stockage qui survive au processus Python.

PostgresSaver pour une durabilité en production

Une base de données se trouve en dehors du processus d’application. Les pannes, les déploiements répétés et les changements de machine permettent au historique des points de contrôle de rester intact. PostgresSaver enregistre ces points de contrôle dans PostgreSQL plutôt que dans la RAM.

PostgreSQL est un choix courant car il est fiable, largement proposé sous forme de services cloud gérés, adapté aux serveurs d’application concurrents, et capable de s’adapter à l’augmentation du volume des points de contrôle avec la hausse du trafic.

Installez le paquet dédié :

pip install langgraph-checkpoint-postgres

Connexions de base :

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 ouvre une connexion gérée par le contexte. setup() crée une seule fois des tables telles que checkpoints et checkpoint_blobs — considérez-le comme une migration et non comme une procédure à effectuer à chaque requête. compile(checkpointer=...) utilise la même API que le mode en mémoire ; changer de backend ne réécrit pas le graphe.

Pools de connexions

Ouvrir une nouvelle connexion à chaque requête est coûteux en temps. Préférez un pool de connexions prêtes à l’emploi :

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)

Variante asynchrone

Les applications de type FastAPI devraient utiliser AsyncPostgresSaver :

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)

Si vous créez manuellement des connexions psycopg, définissez autocommit=True et row_factory=dict_row ; sinon, la fonction setup() pourrait ne pas conserver les tables. Gardez thread_id sous 255 caractères ; les UUID constituent un choix par défaut sûr.

Autres backends

SQLite

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

Un seul fichier local survit aux redémarrages et convient aux prototypes ou aux scripts à un processus. Le verrouillage des fichiers rend inadapté son utilisation dans les déploiements web multi-travailleurs.

Redis

Rapide, basé sur la mémoire avec une persistance disque optionnelle — utile pour des produits de chat à fort débit utilisant de nombreux threads simultanés.

Postgres cloud géré

RDS, Cloud SQL, Supabase, Neon et d’autres solutions utilisent le PostgreSQL standard, ce qui permet à PostgresSaver de fonctionner tandis que le fournisseur s’occupe des sauvegardes et du temps de disponibilité.

Points de contrôle personnalisés

Déclinez la sous-classe BaseCheckpointSaver et implémentez les méthodes put, get_tuple, list (ainsi que leurs équivalents asynchrones) lorsque le stockage interne est obligatoire.

Sélection rapide

Script local durable → SQLite. Environnement de production avec plusieurs travailleurs → Postgres/AsyncPostgres. Forte concurrence dans les opérations de pointage d’état → Redis. Postgres géré existant → Postgres hébergé dans le cloud. Stockage interne forcé → sauveur personnalisé.

Cinq exemples progressifs

Exemple 1 — pas de persistance (amnésie)

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}

La deuxième invocation ne peut pas voir la première : pas de pointeur de vérification, pas d’ID de thread.

Exemple 2 — continuité avec 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}

Les mêmes entrées reprennent à partir du dernier pointage d’état plutôt que de se réinitialiser.

Exemple 3 — reprise après une panne simulée

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

Le partage d’un dictionnaire en cours d’exécution simule un redémarrage ; en production, il suffit de pointer un nouveau processus vers la même base de données Postgres pour reprendre l’exécution via thread_id.

Exemple 4 — intervention humaine avec 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(...) pause l’exécution et crée des points de contrôle. L’appelant voit un chargement sous __interrupt__. Plus tard, Command(resume=...) reprend l’exécution. Lors du redémarrage, LangGraph exécute à nouveau tout le nœud depuis le début — les effets secondaires doivent être conservés après l’interruption, et non avant.

Exemple 5 — chatbot basé sur Postgres

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']}

Arrêtez le script, relancez-le le lendemain avec le même identifiant de thread, et la conversation se poursuit car la base de données contient les points de contrôle.

Pourquoi HITL dépend de la persistance

Trois éléments : interrupt() pour faire une pause avec un chargement, une attente indéfinie sans consommation de CPU tant que l’état est stocké, et Command(resume=...) pour reprendre. Sans point de contrôle, la méthode pause-attente ne dispose d’aucun endroit pour stocker le graphe figé.

Les vrais scénarios d’approbation — décisions de prêt dépassant un seuil, suggestions cliniques avant l’enregistrement des données, révision de clauses juridiques, dépassements de budget — nécessitent souvent des attentes allant de quelques minutes à plusieurs semaines. La persistance rend ces attentes peu coûteuses et sûres.

Persistance entre les capacités des agents

Les conversations longues, les plans en plusieurs étapes qui survivent aux pannes, les boucles d’outils qui ne doivent pas oublier les résultats précédents, ainsi que les workflows avec branches et rejointures dépendent tous de points de contrôle fiables. Les identifiants de thread sont les références que les opérateurs et les interfaces utilisateur utilisent pour reprendre l’exécution appropriée.

Erreurs qui persistent

• Appel de setup() à chaque demande.

Utilisation de SQLite avec plusieurs travailleurs Gunicorn/Uvicorn.

Valeurs trop grandes pour thread_id.

Introduction d’effets secondaires irréversibles avant l’appel à interrupt().

Oubli de transmettre le même thread_id lors de la reprise.

Présomption que les démos en mémoire ont la même durabilité que celles en production.

Pratiques recommandées

Exécutez la configuration du schéma sous forme de migration. Préférez les pools. Utilisez des sauvegardes asynchrones dans les applications asynchrones. Gardez les identifiants de thread courts et stables. Documentez quels environnements utilisent quel backend. Ajoutez des tests de fumée permettant de planter puis de reprendre le processus. Surveillez la croissance et la conservation de la table des points de contrôle.

Résumé

La persistance de LangGraph repose principalement sur le principe « même API graphique, stockage durable différent ». Postgres est le choix par défaut pour les entreprises ; SQLite et Redis occupent des niches spécifiques ; des mécanismes de sauvegarde personnalisés couvrent les infrastructures imposées. L’intervention humaine et la récupération en cas de panne ne sont pas des fonctionnalités optionnelles — c’est pour cela qu’il existe des outils de vérification. Connectez le backend dès le début afin que l’interruption et la reprise soient réelles avant même que l’agent ne rencontre les utilisateurs.

Manuel opérationnel pour les backends de vérification

Traitez le stockage des points de vérification comme les données principales de l’application. Les plannings de sauvegarde, les exercices de restauration et les politiques de conservation doivent faire partie des mêmes discussions que celles concernant les données clients. Un produit de chat qui ne peut pas restaurer les conversations après un basculement régional n’a en réalité pas livré de mémoire — il a seulement livré de l’espoir.

La planification de la capacité doit suivre les points de contrôle par thread actif par jour, la taille moyenne des blobs dans les états à forte charge de messages, ainsi que l’évolution sous charge maximale. Les gros ensembles de données insérés dans l’état domineront la taille de Postgres bien avant que le nombre de lignes ne devienne alarmant. Préférez stocker les fichiers volumineux dans un système de stockage d’objets et conserver des pointeurs dans l’état.

Le contrôle d’accès est crucial : l’utilisateur de la base de données chargé des points de contrôle ne doit pas être le même que celui qui migre des schémas non liés en utilisant des fonctions DDL puissantes. Rotiez les identifiants, privilégiez l’authentification IAM lorsque cela est possible chez les fournisseurs de cloud, et auditez ceux qui peuvent SELECT l’état des conversations. Les identifiants des threads correspondent souvent à des utilisateurs ; traitez les fichiers exportés comme des données personnelles sensibles.

Lors du passage de InMemorySaver à Postgres dans un système en production, l’écriture double est rarement nécessaire si l’on accepte que les threads en mémoire vive se réinitialisent une seule fois. Documentez une fenêtre de maintenance, déployez le nouveau checkpointer et surveillez les taux d’erreur lors de la reprise. Pour les basculements blue/green, assurez-vous que les deux environnements font référence à la même base de données, sinon vous risquerez des incohérences dans les historiques des threads.

Conseils de test : lancez Postgres dans un CI (ou dans un conteneur jetable), exécutez setup() une fois, lancez les cinq exemples en tant que tests d’intégration, et vérifiez qu’un second processus peut reprendre un thread créé par le premier. Ce seul test inter-processus permet de détecter plus d’erreurs en production que n’importe quel test unitaire en mémoire vive.

Finalement, enseignez aux équipes de surveillance la différence entre « graph failed » et « checkpointer unreachable ». Les symptômes sont similaires — ils déclenchent une erreur — mais les solutions diffèrent. Les tableaux de bord qui séparent les exceptions des applications des temps d’attente de la base de données permettent d’économiser des heures en cas d’incidents.

Identité de la thread et expérience utilisateur du produit

Les identifiants de thread ne sont pas seulement des clés de base de données ; ce sont aussi des objets du produit. Associez-les délibérément à des sessions d’utilisateurs, à des tickets de support ou à des identifiants de documents. Affichez l’option « continuer la conversation » dans l’interface en réutilisant cet identifiant. Affichez l’option « recommencer » en en créant un nouveau. Mélanger ces fonctionnalités dans un seul identifiant crée des conversations perturbées où les résultats d’outils anciens réapparaissent.

Pour les utilisateurs multi-appareils, la persistance des identifiants de thread entre différents clients exige des recherches tenant compte de l’authentification. Ne jamais accepter un identifiant de thread fourni par le client sans vérifier son appartenance. Les checkpointers stockent tout ce que vous leur donnez ; ils ne garantissent pas la propriété exclusive.

Évolution du schéma pour l’état en points de contrôle

Les TypedDicts représentant l’état évoluent. Ajouter des champs optionnels est simple ; renommer ou supprimer des champs endommage les anciens points de contrôle. Préférez des modifications additives, des lecteurs qui utilisent des valeurs par défaut pour les clés manquantes, ainsi que des migrations explicites pour les changements destructeurs. Stockez une schema_version dans l’état et basez la logique des nœuds de branche sur cette version lorsque c’est nécessaire.

Les listes de messages volumineuses nécessitent des politiques de suppression : résumez les messages anciens dans un champ, conservez les messages récents bruts, et testez la reprise après suppression. Un historique illimité est la cause la plus fréquente de l’explosion des tables de points de contrôle de Postgres.

Modèles d’expérience utilisateur en cas d’interruption

Les interfaces utilisateur de validation doivent afficher la charge utile de l’interruption, le fragment d’état pertinent ainsi que les délais. Stocker l’identité de l’approbateur dans l’état lors de la reprise. Si les validations expirent, mettre en œuvre un mécanisme de surveillance qui annule ou escalade la situation au lieu de laisser les threads en pause indéfiniment. La persistance permet des attentes prolongées ; la politique du produit doit en limiter la durée.

Exercices de reprise après sinistre

Triennalement, restaurer une sauvegarde dans l’environnement de préparation et reprendre un identifiant de thread connu. Vérifier que les messages, l’état des interruptions et les résidus des outils correspondent. Documenter le RPO/RTO pour la base de données de points de contrôle ainsi que pour la base de données principale de l’application. Les agents qui se souviennent des utilisateurs doivent respecter les mêmes attentes de continuité que les systèmes de messagerie.

Remarques sur les performances

Les pools de connexions doivent être dimensionnés en fonction du nombre d’ouvriers graphiques simultanés, et non seulement des gestionnaires web. Faites attention aux schémas N+1 où chaque nœud écrit inutilement un point de contrôle — ajustez l’utilisation des outils de pointage en fonction des paramètres par défaut de la version LangGraph. Comprimez ou externalisez les sorties binaires volumineuses générées par les outils.

Notes de sécurité

Chiffrez les disques, utilisez TLS avec Postgres, restreignez les chemins réseau entre les ouvriers et le stockage, et supprimez tous les secrets du système avant de créer un point de contrôle si des outils peuvent accéder aux clés brutes. Envisagez l’utilisation de middleware de censure au niveau des champs. Les audits de conformité poseront des questions ; préparez des réponses accompagnées de diagrammes montrant le flux de données du utilisateur → le graphe → l’outil de pointage → le disque.

Ces couches opérationnelles transforment les exemples de code en un système fiable sur lequel on peut compter. Les API sont simples ; c’est la discipline qui entoure leur utilisation qui constitue véritablement l’avantage en environnement de production.

Coordonner les outils de pointage avec l’auto-échelle horizontale

Lorsque Kubernetes échelle les travailleurs graphiques de deux pods à vingt, chaque pod doit communiquer avec le même point de contrôle persistant. Les sessions collantes ne sont pas nécessaires si les identifiants des threads sont transmis avec chaque requête. Les pools préchauffés ne doivent pas faire appel à des fichiers SQLite locaux. Les vérifications de santé doivent s’assurer de la connectivité à la base de données, et non seulement du bon fonctionnement des processus. Lors des déploiements, l’évacuation des travailleurs doit s’achever ou créer un point de contrôle en cours d’exécution afin que les interruptions puissent être reprises. Testez la résilience en supprimant des pods pendant que le système attend une interrupt() et en vérifiant que la reprise est possible sur un autre pod. Cet unique essai valide l’architecture mieux que une douzaine de tests unitaires.

Résistance du système en cas de surcharge : si le stockage des points de contrôle ralentit, les travailleurs doivent-ils réduire la charge ou l’insérer dans une file d’attente ? Il vaut mieux une défaillance explicite que des solutions de secours en mémoire silencieuses qui entraînent des incohérences entre les pods. Les opérateurs ne doivent jamais passer « temporairement » à InMemorySaver en production pour « résoudre le problème de latence » — ce changement annule la promesse de qualité du produit.

Liste de contrôle avant de déclarer la persistance terminée

Confirmer que setup() s’exécute en tant que migration ; les pools sont configurés ; le mode asynchrone versus synchrone est adapté au framework web ; la propriété des identifiants de thread est respectée ; la reprise après interruption a été testée entre différents processus ; les sauvegardes ont été restaurées avec succès une fois ; la tâche de conservation des données est documentée ; le traitement des PII a été examiné ; les tableaux de bord distinguent les erreurs de base de données des erreurs graphiques ; les guides d’utilisation associent les identifiants de thread aux outils d’assistance au utilisateur. Lorsque chaque case est cochée, PostgresSaver (ou le backend choisi) n’est plus un simple extrait de code — il fait partie de la garantie de mémoire du produit. Jusqu’à ce moment-là, considérez les démos comme telles et maintenez les indicateurs de fonctionnalité désactivés pour les flux HITL à long terme.

Une courte histoire d’une reprise qui a échoué

Cet incident a également clarifié l’importance du suivi : déclencher une alerte pour les échecs d’authentification du checkpointer sur une page distincte, et non dans la catégorie générale des erreurs 500. Lorsque la mémoire est le produit principal, l’emplacement de la base de données constitue une dépendance de niveau zéro. Avec cette discipline, remplacer InMemorySaver par PostgresSaver cesse d’être une migration effrayante pour devenir une étape routine dans le passage du notebook au service.