Persistencia, Checkpointing y Threads
Explora MemorySaver, checkpointing por thread, reproducción desde checkpoints, bifurcación de estados pasados y PostgresSaver para producción.
Persistencia, Checkpointing y Threads
Una de las características más poderosas de LangGraph es la persistencia. Cada paso de la ejecución de un grafo puede guardarse como un checkpoint, permitiendo reproducción, rollback y bifurcación desde cualquier estado anterior.
Mermaid: Ciclo de Vida del Checkpoint
Cada llamada invoke() con un thread_id dispara un guardado de checkpoint después de cada nodo. Los checkpoints se almacenan secuencialmente dentro de un thread, formando una línea de tiempo append-only.
MemorySaver
MemorySaver es el backend de checkpointing más simple. Almacena checkpoints en memoria y es ideal para prototipado.
from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph, START, END
# Crea un guardador de checkpoint
memory = MemorySaver()
# Pásalo al compilar
app = builder.compile(checkpointer=memory)
# Cada invocación necesita una configuración de thread
config = {"configurable": {"thread_id": "sesion-1"}}
result = app.invoke({"messages": ["Hola"]}, config)MemorySaver es efímero — todos los checkpoints se pierden cuando el proceso Python termina. Usa PostgresSaver o un guardador personalizado para cargas de trabajo de producción.
Comparación: Backends de Checkpoint
| Característica | MemorySaver | PostgresSaver | Guardador Personalizado |
|---|---|---|---|
| Persistencia | En memoria | PostgreSQL | Definido por el usuario |
| Listo para producción | No | Sí | Depende de la impl. |
| Aislamiento de thread | Sí | Sí | Sí |
| Soporte de replay | Sí | Sí | Debe implementar |
| Soporte de bifurcación | Sí | Sí | Debe implementar |
| Complejidad de setup | Ninguna | Requiere esquema DB | Alta |
| Supervivencia entre procesos | No | Sí | Depende de la impl. |
| Escalabilidad | Proceso único | Multi-proceso | Definido por el usuario |
| Costo | Gratuito | Costos de almacenamiento | Variable |
Mermaid: Diagrama de Estado del Thread
Cada thread transiciona entre estados de ejecución activa y puntos de pausa con checkpoint. Las ramas crean linajes de thread completamente nuevos desde un checkpoint padre.
Checkpointing de Estados por Thread
Un thread es una conversación o sesión de ejecución identificada por un thread_id. LangGraph almacena un nuevo checkpoint después de cada ejecución de nodo dentro de un thread.
# Mismo grafo, mismo thread — el estado se acumula
app.invoke({"messages": ["Turno 1"]}, {"configurable": {"thread_id": "t1"}})
app.invoke({"messages": ["Turno 2"]}, {"configurable": {"thread_id": "t1"}})
# Thread diferente — estado aislado
app.invoke({"messages": ["Inicio thread 2"]}, {"configurable": {"thread_id": "t2"}})El estado tiene namespace por thread. Los checkpoints forman una línea de tiempo dentro de cada thread.
El aislamiento de thread es crítico en producción. Cada sesión de usuario recibe su propio thread_id — nunca compartas un thread_id entre usuarios. Usa un identificador único como user_id:conversation_id como ID del thread para garantizar aislamiento.
Reproduciendo desde Checkpoints
Puedes reproducir la ejecución desde un checkpoint específico proporcionando un checkpoint_id.
# Obtén el ID del checkpoint padre de la última ejecución
parent_id = result["__run"]["checkpoint_id"]
# Reproduce desde ese checkpoint
replayed = app.invoke(
{"messages": ["Nuevo mensaje"]},
{"configurable": {"thread_id": "t1", "checkpoint_id": parent_id}}
)La reproducción no re-ejecuta nodos anteriores al checkpoint — reanuda desde ese estado exacto.
El replay es invaluable para pruebas y depuración. Puedes reproducir un checkpoint específico con entrada modificada para ver cómo se comportaría el grafo con datos diferentes en ese estado exacto.
Reproducción de Checkpoint con thread_id
import uuid
def replay_thread(app, thread_id: str, checkpoint_id: str, new_input: dict):
"""Función utilitaria para reproducir un checkpoint."""
config = {
"configurable": {
"thread_id": thread_id,
"checkpoint_id": checkpoint_id,
}
}
return app.invoke(new_input, config)
# Uso
result = replay_thread(
app,
thread_id="user-123",
checkpoint_id="1ef345ab...",
new_input={"messages": ["Consulta corregida"]}
)Bifurcando desde Estados Pasados
Puedes bifurcar un thread en cualquier checkpoint, creando una bifurcación que diverge de la línea de tiempo original.
# Bifurca desde un checkpoint anterior
fork_config = {
"configurable": {
"thread_id": "t1-rama-1",
"checkpoint_id": parent_id
}
}
fork_result = app.invoke({"messages": ["Mensaje de la rama"]}, fork_config)La rama comienza con el estado del checkpoint padre y prosigue independientemente. Esto es útil para análisis de "qué pasaría si" o correcciones humanas.
Ejemplo de Creación de Bifurcación
def create_branch(app, original_thread: str, checkpoint_id: str, branch_suffix: str, new_input: dict):
"""Crea una bifurcación desde un checkpoint y la ejecuta."""
branch_thread = f"{original_thread}-{branch_suffix}"
config = {
"configurable": {
"thread_id": branch_thread,
"checkpoint_id": checkpoint_id,
}
}
return app.invoke(new_input, config)
# Compara dos ramas del mismo checkpoint
branch_a = create_branch(app, "sesion-1", ckpt_id, "rollback", {"messages": ["Probar A"]})
branch_b = create_branch(app, "sesion-1", ckpt_id, "experimento", {"messages": ["Probar B"]})
# Analiza qué rama produjo mejores resultadosLa bifurcación permite pruebas A/B de decisiones de agente. Ejecuta múltiples ramas desde el mismo checkpoint con diferentes prompts, parámetros o rutas, luego compara resultados para optimizar tu agente.
Costos de Almacenamiento de Checkpoint
Cada ejecución de nodo crea un checkpoint de estado completo. Si tu estado es grande (vectores de embedding, historiales completos de conversación), el almacenamiento de checkpoint puede crecer rápidamente. Considera:
- Usar PostgresSaver con particionamiento de tabla por thread_id
- Implementar política de retención que limpie checkpoints antiguos
- Mantener esquemas de estado ligeros — almacena solo lo que los nodos downstream necesitan
- Usar guardadores personalizados con políticas de ciclo de vida S3/GCS
PostgresSaver para Producción
PostgresSaver persiste checkpoints en una base de datos PostgreSQL, sobreviviendo a reinicios.
from langgraph.checkpoint import PostgresSaver
import asyncpg
# Conecta a PostgreSQL
conn = await asyncpg.connect("postgresql://user:pass@localhost/langgraph")
saver = PostgresSaver(conn)
# Compila con el guardador de producción
app = builder.compile(checkpointer=saver)
# El estado sobrevive a reinicios del proceso
result = await app.ainvoke({"messages": ["Hola"]}, {"configurable": {"thread_id": "prod-1"}})# Configuración del esquema (ejecutar una vez)
pip install langgraph-checkpoint-postgres
python -c "from langgraph.checkpoint import PostgresSaver; PostgresSaver.create_tables('postgresql://user:pass@localhost/langgraph')"Configuración de Producción con PostgresSaver
import asyncpg
from langgraph.checkpoint import PostgresSaver
from contextlib import asynccontextmanager
@asynccontextmanager
async def get_graph_app():
"""Grafo listo para producción con persistencia Postgres."""
conn = await asyncpg.connect(
user="app_user",
password="app_password",
host="postgres.example.com",
port=5432,
database="langgraph_prod",
# Pool de conexiones es recomendado para producción
min_size=5,
max_size=20,
)
try:
saver = PostgresSaver(conn)
app = builder.compile(checkpointer=saver)
yield app
finally:
await conn.close()
# Uso
async with get_graph_app() as app:
result = await app.ainvoke(
{"messages": ["Procesar pedido 12345"]},
{"configurable": {"thread_id": "pedido:12345:usuario:678"}}
)Mermaid: Línea de Tiempo de Checkpoints
Cada checkpoint es una instantánea del estado completo. Las ramas bifurcan desde un checkpoint padre y crean su propia línea de tiempo.
Mermaid: Visualización de Aislamiento de Thread
El aislamiento de thread garantiza que la conversación del Usuario 1 nunca se filtre al estado del Usuario 2. Cada thread tiene su propia cadena de checkpoint independiente.
¿Qué guardador es apropiado para uso en producción?
¿Qué identifica una sesión de ejecución única en LangGraph?
¿Qué sucede cuando reproduces desde un checkpoint?
¿Qué es una bifurcación en el checkpointing de LangGraph?
¿Cuál NO es una característica de MemorySaver?
Escenario: Necesitas depurar por qué un agente dio una respuesta incorrecta hace dos turnos. Tienes los IDs de checkpoints. ¿Qué hacer?
Conclusiones Clave
- MemorySaver almacena checkpoints en memoria para prototipado.
- Cada thread (
thread_id) tiene estado independiente y línea de tiempo de checkpoints. - Reproducir desde un checkpoint reanuda la ejecución sin re-ejecutar pasos anteriores.
- La bifurcación divide una línea de tiempo desde cualquier checkpoint pasado para experimentación.
- PostgresSaver proporciona persistencia de nivel de producción con replay y bifurcación completos.
- Los checkpoints se almacenan después de cada ejecución de nodo por defecto.
- Los guardadores personalizados pueden implementar cualquier backend (Redis, S3, etc.).
- El aislamiento de thread es esencial para sistemas de producción multi-inquilino.
- Usa IDs de thread únicos por sesión de usuario para prevenir fugas de estado.
- Monitorea costos de almacenamiento de checkpoint con esquemas de estado grandes.