Persistencia y Puntos de Control

Intermedio⏱30 min📖3 min de lecturaLección 2 de 10

Persistencia y Puntos de Control

Aprende a persistir el estado del grafo con MemorySaver, guardar y cargar puntos de control, y usar gestión de estado basada en hilos.

Una de las funciones más potentes de LangGraph es la persistencia: la capacidad de guardar y restaurar el estado del grafo en cualquier momento de la ejecución. Esto permite agentes de larga duración, flujos de trabajo con humano en el bucle y sistemas tolerantes a fallos.

Por qué Importa la Persistencia

Sin persistencia, cada invocación de grafo empieza desde cero. Con persistencia:

El estado sobrevive a las invocaciones: Reanuda donde lo dejaste
Humano en el bucle: Pausa la ejecución, inspecciona el estado, aporta entrada y reanuda
Tolerancia a fallos: Recupérate de los fallos sin perder el progreso
Conversaciones con varios turnos: Mantén el contexto entre múltiples interacciones

MemorySaver: Persistencia en Memoria

MemorySaver guarda checkpoints en memoria. Es simple y rápido, pero el estado se pierde cuando el proceso termina.

python

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class State(TypedDict):
    messages: list
    turn_count: int

def node_a(state: State) -> dict:
    return {
        "messages": state["messages"] + ["Processed by A"],
        "turn_count": state["turn_count"] + 1
    }

# Add persistence
checkpointer = MemorySaver()

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
builder.add_edge("a", END)

# Compile with checkpointer
app = builder.compile(checkpointer=checkpointer)

ℹ️Nota

El parámetro checkpointer se pasa a .compile(). Una vez definido, cada ejecución de nodo crea un checkpoint que puede recuperarse después.

IDs de Hilo

Los checkpoints se organizan por thread ID. Un hilo representa una única conversa o sesión de flujo de trabajo:

python

# First invocation — creates a new thread
config1 = {"configurable": {"thread_id": "thread-1"}}
result1 = app.invoke(
    {"messages": [], "turn_count": 0},
    config1
)

# Second invocation on the same thread — continues from where we left off
config2 = {"configurable": {"thread_id": "thread-1"}}
result2 = app.invoke(
    {"messages": ["New message"], "turn_count": 0},  # Initial state is merged
    config2
)

# Different thread — starts fresh
config3 = {"configurable": {"thread_id": "thread-2"}}
result3 = app.invoke(
    {"messages": [], "turn_count": 0},
    config3
)

📌Importante

Al usar un checkpointer, el estado inicial que pasas a invoke() se fusiona con el estado existente del hilo. Las claves que ya existen en el hilo conservan sus valores; las claves nuevas se añaden.

Recuperando Puntos de Control

python

# Get the latest state for a thread
latest = app.get_state(config1)
print(latest.values)      # Current state dict
print(latest.next)        # Nodes to execute next (empty if finished)

# List all checkpoints for a thread
for checkpoint in app.get_state_history(config1):
    print(f"Checkpoint: {checkpoint.config}")
    print(f"  State: {checkpoint.values}")
    print(f"  Next: {checkpoint.next}")
    print(f"  Time: {checkpoint.created_at}")

Actualizando Estado Durante la Ejecución

Puedes actualizar manualmente el estado de un hilo:

python

config = {"configurable": {"thread_id": "thread-1"}}

# Inject new state values
app.update_state(
    config,
    {"messages": ["Injected message"], "turn_count": 99}
)

# Now when you invoke, it continues from the updated state
result = app.invoke({"messages": ["New message"]}, config)

✅Éxito

update_state() te permite intervenir en el estado de un agente; es útil para comentarios humanos, corrección de errores o pruebas.

Reanudando desde un Punto de Control Específico

Puedes reanudar la ejecución desde un checkpoint específico:

python

# Get a specific checkpoint from history
history = list(app.get_state_history(config1))
if len(history) > 1:
    previous_checkpoint = history[1]  # Second-to-last checkpoint

    # Resume from that checkpoint
    result = app.invoke(
        None,  # No new input — use the checkpoint's state
        previous_checkpoint.config
    )

Checkpointers SQLite y PostgreSQL

Para producción, usa checkpointers basados en base de datos que persistan entre reinicios:

SQLite (desarrollo local)

python

from langgraph.checkpoint.sqlite import SqliteSaver

# File-based SQLite
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# Or in-memory SQLite
checkpointer = SqliteSaver.from_conn_string(":memory:")

PostgreSQL (producción)

python

from langgraph.checkpoint.postgres import PostgresSaver

# Requires a PostgreSQL connection string
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)

# Initialize tables (run once)
checkpointer.setup()

⚠️Advertencia

Llama siempre a checkpointer.setup() en los checkpointers SQLite/PostgreSQL para asegurar que las tablas de la base de datos existan. Es una inicialización única.

Contenido del Punto de Control

Cada checkpoint guarda:

python

checkpoint = app.get_state(config)

# The full state dict
state_values = checkpoint.values

# Which nodes are pending execution
next_nodes = checkpoint.next

# Checkpoint metadata
metadata = checkpoint.config
# Includes: thread_id, checkpoint_id, checkpoint_ns

# Parent checkpoint (for branching)
parent = checkpoint.parent_config

Ramificación desde Puntos de Control

Crea rutas de ejecución alternativas desde cualquier checkpoint:

python

# Get the state at a specific checkpoint
checkpoint = list(app.get_state_history(config))[2]

# Create a new thread that continues from this checkpoint
branch_config = {"configurable": {"thread_id": "branch-thread"}}

# Copy the checkpoint's state to the new thread
app.update_state(branch_config, checkpoint.values)

# Now branch-thread continues from where the checkpoint was made
result = app.invoke({"messages": ["Alternative path"]}, branch_config)

Configurando Frecuencia de Puntos de Control

De forma predeterminada, se guarda un checkpoint después de cada ejecución de nodo. Puedes controlar esto:

python

from langgraph.checkpoint.memory import MemorySaver

# Configure checkpoint interval
checkpointer = MemorySaver()

# Option 1: Always checkpoint (default)
app = builder.compile(checkpointer=checkpointer)

# Option 2: Pass checkpoint config at invocation
config = {
    "configurable": {
        "thread_id": "my-thread",
        "checkpoint_ns": "custom"
    }
}

💡Consejo

Para grafos de alto rendimiento, hacer checkpoint después de cada nodo puede ser costoso. Considera agrupar o reducir la frecuencia de checkpointing en producción.

Ejemplo Completo: Chatbot Persistente

python

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from typing import Annotated, List, Any
from operator import add

llm = ChatOpenAI(model="gpt-4o-mini")

class ChatState(TypedDict):
    messages: Annotated[List[Any], add]

def chat_node(state: ChatState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

checkpointer = MemorySaver()

builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

app = builder.compile(checkpointer=checkpointer)

# Multi-turn conversation
session = {"configurable": {"thread_id": "alice-session"}}

app.invoke({"messages": [("human", "Hi!")]}, session)
app.invoke({"messages": [("human", "What's my name?")]}, session)
# The bot remembers the context because state persists

# Different session — starts fresh
session2 = {"configurable": {"thread_id": "bob-session"}}
app.invoke({"messages": [("human", "What's my name?")]}, session2)
# Bob's session doesn't know about Alice's conversation