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.
Persistencia y Puntos de Control
One of LangGraph's most powerful features is persistence — the ability to save and restore graph state at any point during execution. This enables long-running agents, human-in-the-loop workflows, and fault-tolerant systems.
Por qué Importa la Persistencia
Without persistence, every graph invocation starts from scratch. With persistence:
- State survives across invocations: Resume where you left off
- Human-in-the-loop: Pause execution, inspect state, provide input, resume
- Fault tolerance: Recover from crashes without losing progress
- Multi-turn conversations: Maintain context across multiple interactions
MemorySaver: Persistencia en Memoria
MemorySaver stores checkpoints in memory. It's simple and fast, but state is lost when the process exits.
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]
The checkpointer parameter is passed to .compile(). Once set, every node execution creates a checkpoint that can be retrieved later.
IDs de Hilo
Checkpoints are organized by thread ID. A thread represents a single conversation or workflow session:
# 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]
When using a checkpointer, the initial state you pass to invoke() is merged with the existing thread state. Keys that already exist in the thread keep their values; new keys are added.
Recuperando Puntos de Control
# 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
You can manually update state for a thread:
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() lets you intervene in an agent's state — useful for human feedback, error correction, or testing.
Reanudando desde un Punto de Control Específico
You can resume execution from a specific checkpoint:
# 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
For production, use database-backed checkpointers that persist across restarts:
SQLite (Local Development)
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 (Production)
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]
Always call checkpointer.setup() for SQLite/PostgreSQL checkpointers to ensure database tables exist. This is a one-time initialization.
Contenido del Punto de Control
Each checkpoint stores:
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_configRamificación desde Puntos de Control
Create alternative execution paths from any checkpoint:
# 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
By default, a checkpoint is saved after every node execution. You can control this:
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] For high-throughput graphs, checkpointing after every node can be expensive. Consider batching or reducing checkpoint frequency in production.
Ejemplo Completo: Chatbot Persistente
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 conversationPreguntas Prácticas
What parameter is passed to compile() to enable persistence?
What organizes checkpoints in LangGraph?
Which checkpointer is suitable for production deployment?
How often does LangGraph save checkpoints by default?
What does app.get_state(config) return?
How do you manually inject state into an existing thread?
When resuming a thread with invoke(), how is the new input combined with existing state?
What method lists all checkpoints for a thread?
What is the purpose of checkpointing in LangGraph?
How do you initialize database tables for SqliteSaver or PostgresSaver?
[!ÉXITO]
Conclusiones Clave
- MemorySaver provides in-memory persistence; SqliteSaver and PostgresSaver for production
- IDs de Hilo organize checkpoints into sessions
- Checkpoints are saved after every node execution by default
- get_state() retrieves current state; get_state_history() lists all checkpoints
- update_state() manually injects state into a thread
- Different threads have isolated state — they don't share context
- Database-backed checkpointers require checkpointer.setup() before first use
- Checkpointing enables multi-turn conversations, human-in-the-loop, and fault tolerance