langchain-ai/langgraph-persistence

¿Qué hace langgraph-persistence?

INVOCA ESTA HABILIDAD cuando tu LangGraph necesite persistir estado, recordar conversaciones, viajar a través del historial o configurar el alcance del checkpointer en subgrafos. Cubre checkpointers, thread_id, viaje en el tiempo, Store y modos de persistencia en subgrafos.

Descripción general

Esta habilidad proporciona las herramientas y configuraciones necesarias para que los grafos de LangGraph mantengan estado entre ejecuciones. Permite almacenar y recuperar el estado de la conversación, navegar por versiones anteriores (time travel) y gestionar la persistencia a nivel de subgrafo.

Plataformas compatibles

• Claude Code
• Cursor
• Codex
• Cualquier entorno que ejecute LangGraph

Funcionalidades principales

Checkpointers

Los checkpointers son el mecanismo central para persistir el estado en LangGraph. La habilidad cubre:

• MemorySaver: checkpointers en memoria (útil para pruebas)
• SqliteSaver: persistencia en archivos SQLite
• PostgresSaver: persistencia en bases de datos PostgreSQL
• AsyncSqliteSaver: versión asíncrona de SQLite

Puedes configurar un checkpointer al crear el grafo:

from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph

checkpointer = MemorySaver()
graph = StateGraph(MyState)
graph.set_checkpointer(checkpointer)

thread_id

Cada conversación o sesión se identifica con un thread_id. Esto permite que diferentes hilos de conversación tengan estados separados:

config = {"configurable": {"thread_id": "conversacion-1"}}
result = graph.invoke(input_data, config)

Viaje en el tiempo (y bifurcación)

Con el checkpointer habilitado, puedes recuperar estados anteriores y bifurcar la ejecución desde cualquier punto:

# Obtener historial de estados
states = graph.get_state_history(config)

# Recuperar un estado específico
target_config = next(s for s in states if s.id == "id-deseado")
graph.invoke(input_data, target_config)

Store

El Store permite datos que persisten entre diferentes hilos y sesiones, a diferencia del checkpointer que almacena el estado del grafo por hilo:

from langgraph.store import InMemoryStore

store = InMemoryStore()
graph = StateGraph(MyState)
graph.set_store(store)

Persistencia en subgrafos

Cuando usas subgrafos, puedes controlar el alcance del checkpointer:

• Modo por hilo (por defecto): cada subgrafo comparte el mismo thread_id que el grafo padre.
• Modo global (usando checkpointer padre): el subgrafo utiliza el checkpointer del padre pero con su propio namespace.
• Modo independiente: el subgrafo tiene su propio checkpointer separado.

Configuración de ejemplo:

from langgraph.graph import StateGraph

def subgraph_builder():
    builder = StateGraph(SubState)
    # ... agregar nodos
    return builder.compile(checkpointer=MemorySaver(), checkpointer_mode="parent")

Ejemplos de uso

Persistencia de conversación

from langgraph.checkpoint import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    graph = StateGraph(MyState).compile(checkpointer=saver)
    # Ejecutar con thread_id para persistir
    graph.invoke({"messages": [("user", "Hola")]}, {"configurable": {"thread_id": "1"}})

Viaje en el tiempo para depuración

graph = StateGraph(MyState).compile(checkpointer=MemorySaver())
graph.invoke({...}, {"configurable": {"thread_id": "debug-1"}})
# ... más invocaciones ...
for state in graph.get_state_history({"configurable": {"thread_id": "debug-1"}}):
    print(state.values)

Enlaces útiles

• Documentación oficial de LangGraph Persistence
• Checkpointers en LangGraph
• Referencia de la API