Persistencia de LangGraph Parte 2: PostgresSavers, HITL y servidores backend en producción
Mueva LangGraph de InMemorySaver a PostgresSaver con grupos de PostgreSQL y salvadores asíncronos, compare SQLite y Redis, y ejecute funciones de interrupción y reanudación en entorno de producción.
De los puntos de control en RAM a backends duraderos
La primera parte explicó por qué existe la persistencia en LangGraph: estado, puntos de control, herramientas para generar puntos de control, hilos y InMemorySaver. Esta parte adapta esas mismas ideas para su uso en entornos de producción. Los salvadores en memoria desaparecen al reiniciar el proceso; los agentes reales necesitan un almacenamiento que sobreviva más allá del proceso de Python.
PostgresSaver para durabilidad en producción
Una base de datos se encuentra fuera del proceso de la aplicación. Los fallos, las redespliegues y los cambios de máquina mantienen intacta la historia de los puntos de control. PostgresSaver escribe esos puntos de control en PostgreSQL en lugar de en RAM.
PostgreSQL es una opción común porque es duradero, está ampliamente disponible como servicio en la nube gestionado, seguro para servidores de aplicaciones concurrentes y capaz de escalar con el volumen de puntos de control a medida que aumenta el tráfico.
Instale el paquete dedicado:
pip install langgraph-checkpoint-postgres
Conexión básica:
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 abre una conexión gestionada por el contexto. setup() crea una sola vez tablas como checkpoints y checkpoint_blobs; hay que tratarlo como una migración, no como un procedimiento por solicitud. compile(checkpointer=...) utiliza la misma API que el método en memoria; cambiar de backend no modifica el grafo.
Piscinas de conexiones
Abrir una conexión nueva por solicitud desperdicia tiempo. Es mejor utilizar una piscina de conexiones listas:
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 asíncrona
Las aplicaciones al estilo FastAPI deben usar 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 crea las conexiones con psycopg manualmente, establezca autocommit=True y row_factory=dict_row; de lo contrario, setup() podría no persistir las tablas. Mantenga thread_id por debajo de 255 caracteres; los UUID son un valor predeterminado seguro.
Otros backends
SQLite
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
Un único archivo local sobrevive a los reinicios y es adecuado para prototipos o scripts de un solo proceso. El bloqueo de archivos hace que su uso en despliegues web con varios trabajadores no sea adecuado.
Redis
Rápido, basado en la memoria con persistencia opcional en disco; útil para productos de chat de alto rendimiento con muchos hilos simultáneos.
Postgres en la nube gestionado
RDS, Cloud SQL, Supabase, Neon y otros servicios utilizan PostgreSQL estándar, por lo que PostgresSaver funciona mientras el proveedor se encarga de las copias de seguridad y del tiempo de actividad.
Comprobadores personalizados
Derive la subclase BaseCheckpointSaver e implemente put, get_tuple, list (además de sus equivalentes asíncronos) cuando sea obligatorio utilizar almacenamiento interno.
Elegir rápidamente
Script local duradero → SQLite. Entorno de producción con múltiples trabajadores → Postgres/AsyncPostgres. Tráfico extremadamente concurrente de puntos de control → Redis. Postgres gestionado existente → Postgres alojado en la nube. Almacenamiento interno forzoso → salvador personalizado.
Cinco ejemplos progresivos
Ejemplo 1: sin persistencia (amnesia)
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 segunda llamada no puede ver la primera: no hay punto de control ni identificador de hilo.
Ejemplo 2: continuidad con 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}
Los mismos datos de entrada ahora continúan desde el punto de control anterior en lugar de reiniciarse.
Ejemplo 3: reanudación tras una caída simulada
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
Compartir un diccionario dentro del mismo proceso simula un reinicio; en producción simplemente se dirige a un nuevo proceso hacia la misma base de datos Postgres y se reanuda mediante thread_id.
Ejemplo 4: intervención humana con 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(...) pausa y crea puntos de control. Quien llama ve un payload bajo __interrupt__. Más tarde, Command(resume=...) continúa. Al reanudarse, LangGraph vuelve a ejecutar todo el nodo desde el principio; los efectos secundarios se mantienen después de la interrupción, no antes.
Ejemplo 5: chatbot respaldado por 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']}
Detenga el script, arránquelo al día siguiente con el mismo ID de hilo, y la conversación continuará porque la base de datos almacena los puntos de control.
Por qué HITL depende de la persistencia
Tres componentes: interrupt() para pausar con un payload, espera indefinida sin consumo de CPU mientras el estado permanece almacenado, y Command(resume=...) para continuar. Sin un punto de control, la técnica de pausa y espera no tiene dónde almacenar el grafo congelado.
Los patrones reales de aprobación —decisiones de préstamo por encima de un umbral, sugerencias clínicas antes de registrar datos, revisión de cláusulas legales, excesos en gastos— suelen requerir esperas de minutos a semanas. La persistencia hace que esa espera sea económica y segura.
Persistencia entre capacidades de los agentes
Las conversaciones largas, los planes de múltiples pasos que sobreviven a caídas del sistema, los bucles de herramientas que no deben olvidar resultados anteriores y los flujos de trabajo con ramificaciones dependen todos de puntos de control fiables. Los identificadores de hilos son los mecanismos que utilizan los operadores y las interfaces de usuario para reanudar la ejecución correcta.
Errores que siguen apareciendo
- Llamar a
setup()en cada solicitud.
thread_id.interrupt().thread_id al reanudar la ejecución.Prácticas recomendadas
Ejecutar la configuración del esquema como una migración. Preferir grupos de conexiones. Utilizar salvadores asíncronos en aplicaciones asíncronas. Mantener los identificadores de hilos cortos y estables. Documentar qué entornos utilizan qué backend. Agregar pruebas de smoke que permitan reiniciar la aplicación tras un fallo. Monitorear el crecimiento y almacenamiento de la tabla de puntos de control.
Resumen
La persistencia de LangGraph en producción consiste principalmente en “la misma API de gráficos, pero un almacenamiento duradero diferente”. Postgres es la opción empresarial por defecto; SQLite y Redis cubren nichos específicos; los mecanismos de guardado personalizados se utilizan para la infraestructura requerida por normativas. La intervención humana y la recuperación tras fallos no son características opcionales: son la razón de existir los sistemas de puntos de control. Conecte el backend lo antes posible para que la interrupción y la reanudación sean posibles antes de que el agente interactúe con los usuarios.
Manual operativo para backends de puntos de control
Trate el almacenamiento de los puntos de control como si fuera la información principal de la aplicación. Los horarios de copia de seguridad, las pruebas de restauración y las políticas de retención deben abordarse en las mismas conversaciones que los registros de clientes. Un producto de chat que no puede restaurar las conversaciones tras un fallo regional en realidad no ha entregado funcionalidades reales, sino solo expectativas.
La planificación de capacidad debe hacer un seguimiento de los puntos de control por hilo activo al día, del tamaño promedio de los bloques en estados con muchas mensajes y del crecimiento bajo alta concurrencia. Las grandes cargas de herramientas almacenadas en el estado dominarán el tamaño de Postgres mucho antes de que la cantidad de filas parezca preocupante. Es preferible almacenar los archivos voluminosos en almacenamiento de objetos y mantener los punteros en el estado.
El control de acceso es importante: el usuario de la base de datos encargado de los puntos de control no debe ser el mismo que tiene las capacidades avanzadas de DDL para migrar esquemas no relacionados. Rotar las credenciales, preferir la autenticación IAM cuando lo permitan los proveedores de nube y auditar quién puede SELECT el estado de las conversaciones. Los IDs de los hilos a menudo se corresponden con usuarios; trate los datos exportados como PII.
Al migrar de InMemorySaver a Postgres en un sistema en funcionamiento, rara vez es necesario utilizar escritura dual si se acepta que los hilos en RAM se reinicien una sola vez. Documente un período de mantenimiento, implemente el nuevo checkpointer y supervise las tasas de error al reanudar. En los cambios tipo blue/green, asegúrese de que ambos entornos apunten a la misma base de datos; de lo contrario, habrá incoherencias en los historiales de hilos.
Sugerencias para pruebas: ejecute Postgres en CI (o en un contenedor desechable), llame a setup() una vez, ejecute las cinco formas de ejemplo como pruebas de integración y verifique que un segundo proceso pueda reanudar un hilo creado por el primero. Esa única prueba entre procesos detecta más errores en producción que cualquier prueba unitaria en memoria.
Finalmente, enseñe a los técnicos de soporte en turno la diferencia entre “graph failed” y “checkpointer unreachable”. Los síntomas son similares: se genera un error, pero las soluciones difieren. Las pantallas de control que separan las excepciones de la aplicación de los tiempos de espera de la base de datos ahorran horas durante los incidentes.
Identidad del hilo y experiencia de usuario del producto
Los identificadores de hilos no son solo claves de la base de datos; también representan objetos del producto. Asócielos intencionadamente con sesiones de usuario, tickets de soporte o identificadores de documentos. Exponga la opción “continuar conversación” en la interfaz reutilizando dicho identificador, y exponga “empezar de nuevo” al crear uno nuevo. Mezclar estos comportamientos en un mismo identificador genera conversaciones problemáticas donde reaparecen resultados de herramientas anteriores.
Para usuarios con múltiples dispositivos, la persistencia de los identificadores de hilos entre clientes requiere búsquedas que tengan en cuenta la autenticación. Nunca acepte un identificador de hilo proporcionado por el cliente sin verificar su pertenencia. Los puntos de control almacenan todo lo que se les proporciona; no aplican restricciones relacionadas con el propietario del dispositivo.
Evolución del esquema para el estado con puntos de control
Los TypedDicts del estado evolucionan. Añadir campos opcionales es sencillo; renombrar o eliminar campos daña los puntos de control anteriores. Es preferible realizar cambios adicionales, utilizar lectores que asignen valores por defecto a las claves faltantes y aplicar migraciones explícitas para cambios destructivos. Almacene una schema_version en el estado y basee la lógica de los nodos de ramificación en ella cuando sea necesario.
Las listas grandes de mensajes requieren políticas de reducción: resuma los mensajes antiguos en un único campo, conserve los mensajes recientes sin procesar y pruebe la reanudación después de realizar dicha reducción. Un historial ilimitado es la causa más común por la cual las tablas de puntos de control de Postgres se saturan.
Patrones de interrupción en la experiencia de usuario
Las interfaces de aprobación deben mostrar la carga útil de la interrupción, el fragmento de estado relevante y las fechas límite. Al reanudar, guarde la identidad del aprobador en el estado. Si las aprobaciones vencen, ejecute un mecanismo de vigilancia que cancele o escalone la situación en lugar de dejar las hilos en pausa indefinidamente. La persistencia permite esperas prolongadas; la política del producto debe establecer límites para ellas.
Ejercicios de recuperación ante desastres
Cada trimestre, restaure una copia de seguridad en el entorno de pruebas y reanude un identificador de hilo conocido. Confirme que los mensajes, el estado de la interrupción y los residuos de las herramientas coincidan. Documente los valores RPO/RTO para la base de datos de puntos de control junto con la base de datos principal de la aplicación. Los agentes que recuerdan a los usuarios heredan las mismas expectativas de continuidad que los sistemas de correo electrónico.
Notas sobre el rendimiento
Los grupos de conexiones deben dimensionarse en función del número de trabajadores del grafo que operan simultáneamente, y no solo de los manejadores web. Preste atención a los patrones N+1 en los que cada nodo escribe un punto de control innecesariamente; ajuste el uso de los puntos de control según las configuraciones predeterminadas de la versión de LangGraph. Comprima o externalice los resultados binarios voluminosos generados por las herramientas.
Notas de seguridad
Cifre los discos, utilice TLS con Postgres, restrinja las rutas de red desde los trabajadores hasta el almacenamiento, y elimine cualquier información confidencial del estado antes de crear un punto de control si las herramientas llegan a ver claves en bruto. Considere utilizar middleware de censura a nivel de campo. Las revisiones de cumplimiento harán preguntas; prepare respuestas con diagramas que muestren el flujo de datos desde el usuario → grafo → punto de control → disco.
Estas capas operativas convierten los ejemplos de código en un sistema en el que se puede confiar plenamente. Las APIs son simples; la disciplina de implementación es la verdadera ventaja en entornos de producción.
Coordinación de los puntos de control con el escalado automático horizontal
Cuando Kubernetes escala los trabajadores del grafo de dos pods a veinte, cada pod debe comunicarse con el mismo punto de verificación duradero. Las sesiones pegajosas son innecesarias si los identificadores de hilos se transmiten con cada solicitud. Los grupos de trabajadores en estado caliente no deben asumir la existencia de archivos SQLite locales. Las verificaciones de salud deben confirmar la conectividad a la base de datos, y no solo el estado de actividad del proceso. Durante los despliegues, el vaciado de trabajadores debe completarse o hacerse un punto de verificación durante su ejecución para que las interrupciones puedan reanudarse. Realice pruebas de caos eliminando pods mientras se espera la función interrupt() y verificando que la operación pueda reanudarse en otro pod. Ese único experimento valida la arquitectura mejor que una docena de pruebas unitarias.
Presión de retroalimentación del documento: si el almacén de puntos de control se ralentiza, ¿deben los trabajadores reducir la carga o ponerla en cola? Es preferible un fallo explícito a soluciones silenciosas basadas en memoria que generen discrepancias entre los pods. Los operadores nunca deben cambiar “temporalmente” a InMemorySaver en entornos de producción para “corregir la latencia”; ese cambio anula la promesa del producto.
Lista de verificación antes de considerar que la persistencia está completada
Se confirma que setup() se ejecuta como parte de la migración; los grupos están configurados; el modo asíncrono frente al síncrono se ajusta al framework web; se aplica el control de propiedad de los identificadores de hilos; se ha probado la reanudación de interrupciones entre procesos; las copias de seguridad se restauraron con éxito una vez; está documentado el trabajo de retención de datos; se revisó el manejo de información personal identificable; los paneles de control distinguen entre errores de la base de datos y errores gráficos; los manuales de operación vinculan los identificadores de hilos con las herramientas de soporte al usuario. Cuando se marque cada casilla, PostgresSaver (o el backend que haya elegido) deja de ser un fragmento de código y pasa a formar parte de la garantía de memoria del producto. Hasta entonces, trate las demostraciones como tales y mantenga desactivadas las banderas de funcionalidad para flujos HITL de larga duración.
Una breve historia sobre una reanudación que falló
Un ingeniero de soporte técnico vio en una ocasión que las aprobaciones se quedaban atascadas durante horas. El gráfico funcionaba bien; la contraseña del checkpointer se rotaba y los trabajadores guardaban en caché credenciales antiguas. Las nuevas interrupciones fallaban mientras los hilos antiguos seguían leyendo datos. La solución fue actualizar las credenciales además de incluir una instrucción en el manual: verificar la autenticación del checkpointer antes de reiniciar los pods del gráfico. Guarde esa lección junto a sus fragmentos de PostgresSaver para que la próxima rotación no repita el corte de servicio. La memoria duradera incluye la memoria operativa: esa sabiduría humana plasmada en los manuales de procedimientos.
Ese incidente también aclaró la importancia del monitoreo: generar alertas por fallas en la autenticación del checkpointer en una página separada, y no dentro de la categoría genérica 500. Cuando la memoria es el producto final, la ruta de la base de datos constituye una dependencia de nivel cero. Con esa disciplina, reemplazar InMemorySaver por PostgresSaver deja de ser una migración intimidante y se convierte en un paso rutinario en el proceso de transformar un cuaderno de notas en un servicio.
Lecturas relacionadas
- El registro de chat, los hechos, el estado del flujo de trabajo y los puntos de control son cuatro almacenes distintos — Deje de llamar memoria a todo. Separe las transcripciones de sesiones, los hechos duraderos, el estado del flujo de trabajo de los tickets y los puntos de control de LangGraph, aplicando reglas de retención y autenticación para cada uno.
- Refuerzo de los Grafos Agentes: Contratos de Estado, Aislamiento y Puertas HITL — Fiabilidad en producción para LangGraph: estado basado en TypedDict plano, nodos de uso exclusivo, aristas para intentos y errores, revisión humana y capacidades de observabilidad.