Checkpointing con MemorySaver

Descripción de la cápsula

Tu Research Agent procesa 5 fuentes de investigación. Descompone la consulta, busca en la web, analiza papers, extrae datos de noticias, y sintetiza un reporte. El proceso toma 5 minutos y consume tokens de API en cada paso. Después de procesar 3 fuentes — crash. El proceso se cae. Sin checkpointing: repites las 5 fuentes desde cero. 5 minutos más. Doble costo de API. El usuario espera de nuevo.

Con checkpointing: el agente resume desde la fuente 4. 1 minuto. Sin repetir trabajo. Sin costo duplicado.

Checkpointing guarda el estado completo de tu grafo después de cada nodo. Es un snapshot automático: qué nodos se ejecutaron, qué estado tenían, qué mensajes se intercambiaron. Si el proceso se interrumpe, no empiezas de cero — retomas desde el último checkpoint.

En la cápsula anterior viste short-term memory: cómo el grafo mantiene el historial de mensajes durante una conversación. Eso resuelve la continuidad dentro de una sesión. Checkpointing resuelve algo diferente: la durabilidad de esa sesión. Si el proceso muere, la memoria muere con él — a menos que tengas checkpointing.

MemorySaver es el checkpointer más simple de LangGraph: guarda los checkpoints en memoria RAM. Es perfecto para desarrollo y testing. No necesitas base de datos, no necesitas configuración. Una línea de código y tu grafo ya tiene checkpointing. En la siguiente cápsula migrarás a PostgresSaver para producción — y vas a ver que el cambio es literalmente una línea.

El problema: estado que se pierde

Sin checkpointing, cada invocación de tu grafo es independiente. El grafo no sabe que ya corrió antes. No tiene contexto previo. Cada graph.invoke() empieza con un estado vacío.

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()

result1 = graph.invoke({"messages": [("user", "¿Qué es RAG?")]})
print(result1["messages"][-1].content)
# Output: "RAG (Retrieval-Augmented Generation) es una técnica que..."

result2 = graph.invoke({"messages": [("user", "Dame ejemplos")]})
print(result2["messages"][-1].content)
# Output: "¿Ejemplos de qué? No tengo contexto previo..."

El segundo invoke no sabe nada del primero. "Dame ejemplos" no tiene referente — el agente no sabe que hablaste de RAG hace 2 segundos. Cada invocación es una conversación nueva desde cero.

MemorySaver: checkpointing en una línea

MemorySaver guarda el estado de tu grafo en memoria después de cada nodo. Lo activas en dos pasos: crear el checkpointer y pasarlo al compilar.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user_001"}}

result1 = graph.invoke({"messages": [("user", "¿Qué es RAG?")]}, config)
print(result1["messages"][-1].content)
# Output: "RAG (Retrieval-Augmented Generation) es una técnica que..."

result2 = graph.invoke({"messages": [("user", "Dame ejemplos")]}, config)
print(result2["messages"][-1].content)
# Output: "Aquí tienes ejemplos de RAG: 1) Un chatbot que consulta
#          documentación interna... 2) Un asistente legal que busca
#          jurisprudencia relevante..."

Tres cambios respecto al código sin checkpointing:

1. checkpointer = MemorySaver() — creas el checkpointer
2. graph_builder.compile(checkpointer=checkpointer) — lo pasas al compilar
3. config = {"configurable": {"thread_id": "user_001"}} — identificas el thread

Ahora el segundo invoke sabe que "ejemplos" se refiere a RAG. El checkpointer guardó el estado completo después del primer turno (mensajes del usuario + respuesta del modelo), y el segundo turno lo recibió automáticamente.

Cómo funciona internamente

Cuando compilas con checkpointer, LangGraph intercepta la ejecución de cada nodo y guarda un snapshot del estado:

Ejecución con checkpointing:

[START] → Estado inicial guardado como Checkpoint 0
   ↓
[chatbot] ejecuta → Estado actualizado guardado como Checkpoint 1
   ↓
[END] → Estado final guardado como Checkpoint 2

Cada checkpoint contiene:
- El estado completo del grafo en ese punto
- El nombre del nodo que produjo ese estado
- Un timestamp
- El ID del checkpoint padre (para navegar la historia)

Lo importante: el checkpoint guarda el estado completo, no un diff. Checkpoint 2 tiene todos los mensajes, no solo el último. Esto significa que puedes retomar desde cualquier checkpoint sin necesidad de los anteriores.

Flujo de dos turnos

Turno 1: "¿Qué es RAG?"
  1. Checkpointer busca checkpoint para thread "user_001" → no hay
  2. Nodo "chatbot" ejecuta → respuesta del modelo
  3. Checkpointer guarda: [user_msg, ai_msg]

Turno 2: "Dame ejemplos"
  1. Checkpointer busca checkpoint → ENCUENTRA [user_msg_1, ai_msg_1]
  2. MERGE: estado previo + nuevo input = [user_msg_1, ai_msg_1, user_msg_2]
  3. Nodo "chatbot" ejecuta con contexto completo → respuesta coherente
  4. Checkpointer guarda: [user_msg_1, ai_msg_1, user_msg_2, ai_msg_2]

El paso 1-2 del turno 2 es la magia: el checkpointer recupera el estado previo y lo combina con el nuevo input. El nodo chatbot recibe toda la conversación, no solo el último mensaje.

thread_id: aislamiento entre conversaciones

El thread_id es obligatorio cuando usas checkpointer. Es el identificador que separa las conversaciones. Sin él, LangGraph no sabe dónde guardar ni de dónde recuperar el checkpoint.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config_alice = {"configurable": {"thread_id": "alice_session_001"}}
config_bob = {"configurable": {"thread_id": "bob_session_001"}}

graph.invoke({"messages": [("user", "Soy Alice. Investigo sobre RAG.")]}, config_alice)
graph.invoke({"messages": [("user", "Soy Bob. Investigo sobre fine-tuning.")]}, config_bob)

result_alice = graph.invoke({"messages": [("user", "¿Qué estoy investigando?")]}, config_alice)
print(f"Alice: {result_alice['messages'][-1].content}")
# Output: "Estás investigando sobre RAG..."

result_bob = graph.invoke({"messages": [("user", "¿Qué estoy investigando?")]}, config_bob)
print(f"Bob: {result_bob['messages'][-1].content}")
# Output: "Estás investigando sobre fine-tuning..."

Cada thread_id tiene su propia línea de checkpoints independiente. Alice y Bob nunca se mezclan. Esto es la base del soporte multi-usuario: en producción, cada usuario (o cada sesión de un usuario) tiene su propio thread_id.

Patrones comunes para thread_id

Inspeccionar checkpoints: ver el estado del grafo

Los checkpoints no son solo para continuidad — son una herramienta de debugging. Puedes inspeccionar el estado exacto del grafo en cualquier punto de su ejecución.

get_state: ver el checkpoint actual

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "debug_session"}}
graph.invoke({"messages": [("user", "¿Qué es prompt engineering?")]}, config)
graph.invoke({"messages": [("user", "Dame 3 técnicas")]}, config)

state_snapshot = graph.get_state(config)

print(f"Número de mensajes: {len(state_snapshot.values['messages'])}")
print(f"Último nodo ejecutado: {state_snapshot.next}")
print(f"Config: {state_snapshot.config}")
print(f"Checkpoint ID: {state_snapshot.config['configurable']['checkpoint_id']}")

for msg in state_snapshot.values["messages"]:
    role = "User" if msg.type == "human" else "AI"
    preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
    print(f"  [{role}] {preview}")
# Output esperado:
# Número de mensajes: 4
# Último nodo ejecutado: ()
# Config: {'configurable': {'thread_id': 'debug_session', 'checkpoint_id': '...'}}
# Checkpoint ID: 1ef8a...
#   [User] ¿Qué es prompt engineering?
#   [AI] Prompt engineering es la disciplina de diseñar instrucciones efectivas...
#   [User] Dame 3 técnicas
#   [AI] 1) Zero-shot prompting: dar la instrucción sin ejemplos...

get_state() retorna un StateSnapshot con:

• values: el estado completo del grafo
• next: tupla de nodos pendientes (vacía si terminó)
• config: la configuración incluyendo el checkpoint_id
• metadata: información adicional del checkpoint
• parent_config: referencia al checkpoint anterior

get_state_history: navegar el historial completo

history = list(graph.get_state_history(config))

print(f"Total de checkpoints: {len(history)}")
for i, snapshot in enumerate(history):
    n_msgs = len(snapshot.values.get("messages", []))
    created = snapshot.metadata.get("created_by", "unknown")
    step = snapshot.metadata.get("step", "?")
    print(f"  Checkpoint {i}: step={step}, mensajes={n_msgs}, created_by={created}")
# Output esperado:
# Total de checkpoints: 5
#   Checkpoint 0: step=3, mensajes=4, created_by=loop
#   Checkpoint 1: step=2, mensajes=3, created_by=loop
#   Checkpoint 2: step=1, mensajes=2, created_by=loop
#   Checkpoint 3: step=0, mensajes=1, created_by=loop
#   Checkpoint 4: step=-1, mensajes=0, created_by=system

El historial está en orden cronológico inverso: el más reciente primero. Cada checkpoint corresponde a un paso en la ejecución del grafo. Checkpoint 4 (step=-1) es el estado inicial antes de que cualquier nodo ejecute.

Resumir desde un checkpoint específico (time-travel)

No solo puedes retomar desde el último checkpoint — puedes retomar desde cualquiera. Obtén el checkpoint deseado de get_state_history(), extrae su .config, y pasa ese config a graph.invoke():

history = list(graph.get_state_history(config))
old_checkpoint = history[-3]  # checkpoint después del turno 1
result = graph.invoke(
    {"messages": [("user", "Nueva pregunta")]},
    old_checkpoint.config  # retoma desde ese punto
)

Esto es time-travel debugging: retrocedes a un punto específico y creas una bifurcación. El historial original sigue intacto. El nuevo invoke desde el checkpoint antiguo crea una rama nueva. Los ejercicios 4 y 5 de esta cápsula lo implementan en detalle.

Checkpointing en grafos multi-nodo

El valor del checkpointing se amplifica en grafos con múltiples nodos. Cada nodo genera un checkpoint, así que si el proceso se interrumpe entre nodos, no pierdes el trabajo de los nodos anteriores.

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


class ResearchState(TypedDict):
    query: str
    sources_searched: list[str]
    findings: list[str]
    report: str
    status: str


def decompose_query(state: ResearchState) -> dict:
    print("[1/4] Descomponiendo query...")
    return {"status": "decomposed"}


def search_sources(state: ResearchState) -> dict:
    sources = ["web", "papers", "news"]
    findings = []
    for source in sources:
        print(f"[2/4] Buscando en {source}...")
        findings.append(f"Hallazgo de {source} sobre '{state['query']}'")
    return {
        "sources_searched": sources,
        "findings": findings,
        "status": "searched"
    }


def analyze_findings(state: ResearchState) -> dict:
    print(f"[3/4] Analizando {len(state['findings'])} hallazgos...")
    return {"status": "analyzed"}


def generate_report(state: ResearchState) -> dict:
    print("[4/4] Generando reporte...")
    report = f"Reporte: {len(state['findings'])} hallazgos de {len(state['sources_searched'])} fuentes"
    return {"report": report, "status": "completed"}


builder = StateGraph(ResearchState)
builder.add_node("decompose", decompose_query)
builder.add_node("search", search_sources)
builder.add_node("analyze", analyze_findings)
builder.add_node("report", generate_report)
builder.add_edge(START, "decompose")
builder.add_edge("decompose", "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "research_001"}}
result = graph.invoke(
    {"query": "Estado del arte en RAG", "sources_searched": [], "findings": [], "report": "", "status": ""},
    config
)
# Output:
# [1/4] Descomponiendo query...
# [2/4] Buscando en web...
# [2/4] Buscando en papers...
# [2/4] Buscando en news...
# [3/4] Analizando 3 hallazgos...
# [4/4] Generando reporte...

history = list(graph.get_state_history(config))
print(f"Checkpoints creados: {len(history)}")
# Output esperado: Checkpoints creados: 6
# (1 por nodo + 1 input + 1 sistema)

6 checkpoints para 4 nodos. Si el proceso se hubiera caído entre search y analyze, podrías retomar desde el checkpoint con status=searched — sin repetir la descomposición ni las búsquedas.

Qué contiene un checkpoint

Cada checkpoint almacena información completa para reconstruir el estado exacto del grafo:

La limitación crítica: MemorySaver es volátil

MemorySaver almacena todo en la RAM del proceso. Esto significa:

Proceso Python activo:
  ✅ Checkpoints disponibles
  ✅ Puedes resumir conversaciones
  ✅ Historial completo accesible

Proceso Python se reinicia (deploy, crash, Ctrl+C):
  ❌ TODOS los checkpoints se pierden
  ❌ Todas las conversaciones empiezan de cero
  ❌ El historial completo desaparece

Esto NO es un defecto — es una decisión de diseño. MemorySaver está diseñado para:

Para producción necesitas un checkpointer durable. En la siguiente cápsula implementarás PostgresSaver — y vas a ver que el cambio es una sola línea de código. Todo lo que aprendiste aquí sobre thread_id, get_state(), get_state_history(), y el flujo de checkpointing funciona exactamente igual. Solo cambia dónde se guardan los datos.

Troubleshooting

Problema 1: "RunnableConfigurableFieldsSpec" o error al invocar sin config

Síntoma: Error críptico al hacer graph.invoke({"messages": [...]}) sin pasar config. Causa: Cuando compilas con checkpointer, el thread_id es obligatorio. Solución: Siempre pasa config = {"configurable": {"thread_id": "algún_id"}} como segundo argumento a invoke().

Problema 2: El agente no recuerda turnos anteriores

Síntoma: Cada invoke parece una conversación nueva. Causa 1: Estás usando un thread_id diferente en cada invoke. Causa 2: El checkpointer no se pasó al compilar. Solución: Verifica que (1) el mismo thread_id se use en ambos invokes y (2) graph_builder.compile(checkpointer=checkpointer) incluya el checkpointer.

Problema 3: Checkpoints desaparecen después de reiniciar el script

Síntoma: Ejecutas el script, funciona. Lo ejecutas de nuevo, el agente no recuerda nada. Causa: MemorySaver es in-memory. Cuando el proceso de Python termina, los checkpoints se pierden. Solución: Esto es comportamiento esperado de MemorySaver. Para persistencia entre reinicios, usa PostgresSaver (cápsula 04).

Problema 4: Mezcla de conversaciones entre usuarios

Síntoma: Un usuario recibe contexto de otro usuario. Causa: Estás reutilizando el mismo thread_id para diferentes usuarios. Solución: Usa un thread_id único por usuario o por sesión. Un patrón seguro: f"user_{user_id}_session_{session_id}".

Problema 5: El historial crece sin límite y el modelo falla

Síntoma: Después de muchos turnos, el modelo lanza error de context window. Causa: El checkpointer guarda todos los mensajes. Después de 50+ turnos, la lista de mensajes excede el límite del modelo. Solución: Implementa message trimming antes de llamar al modelo. La cápsula 02 de este módulo cubre trim_messages() para esto.

Ejercicios

Ejercicio 1: Chatbot con memoria básica (Fácil)

Crea un grafo con un solo nodo chatbot que use MessagesState y MemorySaver. Haz 3 invocaciones con el mismo thread_id: (1) "Me llamo Carlos", (2) "¿Cuál es mi nombre?", (3) "¿Cuántos mensajes llevamos?". Verifica que el modelo recuerda el nombre y puede contar los turnos.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "carlos_001"}}

r1 = graph.invoke({"messages": [("user", "Me llamo Carlos")]}, config)
print(f"Turno 1: {r1['messages'][-1].content}")

r2 = graph.invoke({"messages": [("user", "¿Cuál es mi nombre?")]}, config)
print(f"Turno 2: {r2['messages'][-1].content}")

r3 = graph.invoke({"messages": [("user", "¿Cuántos mensajes llevamos en esta conversación?")]}, config)
print(f"Turno 3: {r3['messages'][-1].content}")

state = graph.get_state(config)
print(f"\nMensajes totales en el checkpoint: {len(state.values['messages'])}")
assert len(state.values["messages"]) == 6, "Debe haber 6 mensajes (3 user + 3 AI)"
print("✅ Checkpoint contiene toda la conversación")
# Output esperado:
# Turno 1: ¡Hola Carlos! Encantado de conocerte...
# Turno 2: Tu nombre es Carlos...
# Turno 3: Llevamos 3 intercambios (6 mensajes en total)...
# Mensajes totales en el checkpoint: 6
# ✅ Checkpoint contiene toda la conversación

Ejercicio 2: Aislamiento multi-usuario (Fácil)

Crea un chatbot con MemorySaver. Usa dos thread_id diferentes ("user_A" y "user_B"). En thread A, dile al modelo "Mi lenguaje favorito es Python". En thread B, dile "Mi lenguaje favorito es Rust". Luego pregunta en cada thread "¿Cuál es mi lenguaje favorito?" y verifica que las respuestas son correctas y no se mezclan.

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END


def chatbot(state: MessagesState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config_a = {"configurable": {"thread_id": "user_A"}}
config_b = {"configurable": {"thread_id": "user_B"}}

graph.invoke({"messages": [("user", "Mi lenguaje favorito es Python")]}, config_a)
graph.invoke({"messages": [("user", "Mi lenguaje favorito es Rust")]}, config_b)

result_a = graph.invoke({"messages": [("user", "¿Cuál es mi lenguaje favorito?")]}, config_a)
result_b = graph.invoke({"messages": [("user", "¿Cuál es mi lenguaje favorito?")]}, config_b)

print(f"User A: {result_a['messages'][-1].content}")
print(f"User B: {result_b['messages'][-1].content}")

assert "python" in result_a["messages"][-1].content.lower()
assert "rust" in result_b["messages"][-1].content.lower()
print("\n✅ Las conversaciones están correctamente aisladas")
# Output esperado:
# User A: Tu lenguaje favorito es Python...
# User B: Tu lenguaje favorito es Rust...
# ✅ Las conversaciones están correctamente aisladas

Ejercicio 3: Inspeccionar el historial de checkpoints (Medio)

Construye un grafo con 3 nodos secuenciales (analyze → enrich → summarize), cada uno agrega un campo al estado. Usa get_state_history() para verificar que cada checkpoint tiene progresivamente más datos.

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


class PipelineState(TypedDict):
    input_text: str
    analysis: str
    enrichment: str
    summary: str


def analyze(state: PipelineState) -> dict:
    return {"analysis": f"Análisis de: {state['input_text'][:30]}"}

def enrich(state: PipelineState) -> dict:
    return {"enrichment": f"Datos para: {state['analysis'][:30]}"}

def summarize(state: PipelineState) -> dict:
    return {"summary": f"Resumen: {state['analysis'][:20]} + {state['enrichment'][:20]}"}


builder = StateGraph(PipelineState)
builder.add_node("analyze", analyze)
builder.add_node("enrich", enrich)
builder.add_node("summarize", summarize)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "enrich")
builder.add_edge("enrich", "summarize")
builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "pipeline_001"}}
graph.invoke({"input_text": "LangGraph para agentes", "analysis": "", "enrichment": "", "summary": ""}, config)

history = list(graph.get_state_history(config))
print(f"Total checkpoints: {len(history)}\n")
for i, snap in enumerate(history):
    filled = [k for k, v in snap.values.items() if v]
    print(f"  Checkpoint {i}: campos={filled}")
# Output esperado:
# Total checkpoints: 5
#   Checkpoint 0: campos=['input_text', 'analysis', 'enrichment', 'summary']
#   Checkpoint 1: campos=['input_text', 'analysis', 'enrichment']
#   ...

assert len(history) == 5
print("✅ Progresión de checkpoints verificada")

Ejercicio 4: Time-travel — bifurcar desde un checkpoint anterior (Medio)

Usa el grafo del Ejercicio 3. Obtén el checkpoint después de analyze (antes de enrich). Invoca desde ese checkpoint con graph.invoke(None, checkpoint.config). Verifica que enrich y summarize se re-ejecutan pero analysis se preserva.

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


class PipelineState(TypedDict):
    input_text: str
    analysis: str
    enrichment: str
    summary: str

def analyze(state: PipelineState) -> dict:
    return {"analysis": f"Análisis de: {state['input_text'][:30]}"}

def enrich(state: PipelineState) -> dict:
    return {"enrichment": f"Enriquecido: {state['analysis'][:30]}"}

def summarize(state: PipelineState) -> dict:
    return {"summary": f"Resumen: {state['analysis'][:20]} + {state['enrichment'][:20]}"}


builder = StateGraph(PipelineState)
for name, fn in [("analyze", analyze), ("enrich", enrich), ("summarize", summarize)]:
    builder.add_node(name, fn)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "enrich")
builder.add_edge("enrich", "summarize")
builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "time_travel_ex"}}
original = graph.invoke(
    {"input_text": "LangGraph para agentes", "analysis": "", "enrichment": "", "summary": ""}, config
)

history = list(graph.get_state_history(config))
after_analyze = next(s for s in history if s.values.get("analysis") and not s.values.get("enrichment"))

branched = graph.invoke(None, after_analyze.config)

assert branched["analysis"] == original["analysis"], "Analysis debe preservarse"
assert branched["summary"] != "", "Summary debe regenerarse"
print(f"Analysis preservado: {branched['analysis']}")
print(f"Enrichment regenerado: {branched['enrichment']}")
print("✅ Time-travel bifurcation exitosa")
# Output esperado:
# Analysis preservado: Análisis de: LangGraph para agentes
# Enrichment regenerado: Enriquecido: Análisis de: LangGraph para ag
# ✅ Time-travel bifurcation exitosa

Ejercicio 5: Simular crash y recovery (Avanzado)

Crea un grafo con 4 nodos secuenciales (step_1 → step_2 → step_3 → step_4) donde cada nodo agrega su nombre a una lista steps_completed en el estado. Ejecuta el grafo normalmente y guarda el resultado. Luego simula un "crash" obteniendo el checkpoint después de step_2 y reanudando desde ahí. Verifica que la ejecución reanudada completa step_3 y step_4 sin repetir step_1 y step_2.

import operator
from typing import TypedDict, Annotated
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class CrashState(TypedDict):
    task: str
    steps_completed: Annotated[list[str], operator.add]
    output: str


def step_1(state: CrashState) -> dict:
    print("[step_1] Ejecutando...")
    return {"steps_completed": ["step_1"]}


def step_2(state: CrashState) -> dict:
    print("[step_2] Ejecutando...")
    return {"steps_completed": ["step_2"]}


def step_3(state: CrashState) -> dict:
    print("[step_3] Ejecutando...")
    return {"steps_completed": ["step_3"]}


def step_4(state: CrashState) -> dict:
    print("[step_4] Ejecutando...")
    all_steps = state["steps_completed"] + ["step_4"]
    return {"steps_completed": ["step_4"], "output": f"Completado: {', '.join(all_steps)}"}


builder = StateGraph(CrashState)
for name, fn in [("step_1", step_1), ("step_2", step_2), ("step_3", step_3), ("step_4", step_4)]:
    builder.add_node(name, fn)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", "step_4")
builder.add_edge("step_4", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "crash_sim"}}
full_result = graph.invoke({"task": "test", "steps_completed": [], "output": ""}, config)
print(f"\nFull run: {full_result['steps_completed']}")

history = list(graph.get_state_history(config))
after_step_2 = None
for snap in history:
    completed = snap.values.get("steps_completed", [])
    if "step_2" in completed and "step_3" not in completed:
        after_step_2 = snap
        break

assert after_step_2 is not None, "No se encontró checkpoint post-step_2"
print(f"\n--- Simulando crash y recovery desde step_2 ---")
print(f"Estado en checkpoint: steps_completed={after_step_2.values['steps_completed']}")

recovered = graph.invoke(None, after_step_2.config)
print(f"\nRecovered steps: {recovered['steps_completed']}")
print(f"Output: {recovered['output']}")

assert "step_1" in recovered["steps_completed"]
assert "step_2" in recovered["steps_completed"]
assert "step_3" in recovered["steps_completed"]
assert "step_4" in recovered["steps_completed"]
print("\n✅ Recovery exitoso: step_3 y step_4 ejecutados sin repetir step_1 y step_2")
# Output esperado:
# [step_1] Ejecutando...
# [step_2] Ejecutando...
# [step_3] Ejecutando...
# [step_4] Ejecutando...
#
# Full run: ['step_1', 'step_2', 'step_3', 'step_4']
#
# --- Simulando crash y recovery desde step_2 ---
# Estado en checkpoint: steps_completed=['step_1', 'step_2']
# [step_3] Ejecutando...
# [step_4] Ejecutando...
#
# Recovered steps: ['step_1', 'step_2', 'step_3', 'step_4']
# Output: Completado: step_1, step_2, step_3, step_4
#
# ✅ Recovery exitoso: step_3 y step_4 ejecutados sin repetir step_1 y step_2

Resumen

En esta cápsula aprendiste:

• Checkpointing guarda el estado completo del grafo después de cada nodo. No es un diff, es un snapshot. Si el proceso se interrumpe, puedes retomar desde el último checkpoint sin perder trabajo previo
• MemorySaver es el checkpointer para desarrollo. Una línea (checkpointer = MemorySaver()), sin dependencias externas, sin configuración. Perfecto para prototipar y testear
• thread_id es obligatorio y es el mecanismo de aislamiento. Cada thread tiene su propia cadena de checkpoints. Múltiples usuarios = múltiples thread_ids = historiales completamente separados
• get_state() y get_state_history() son herramientas de debugging. Puedes ver el estado exacto del grafo en cualquier punto de su ejecución — qué mensajes había, qué nodos se ejecutaron, qué datos tenía
• Time-travel: puedes retomar desde cualquier checkpoint, no solo el último. Esto crea bifurcaciones en la historia — útil para debugging y para explorar caminos alternativos
• MemorySaver es volátil: si el proceso se reinicia, todos los checkpoints se pierden. Esto es esperado y aceptable para desarrollo. Para producción, necesitas persistencia durable

Próxima cápsula: Persistencia con PostgresSaver y Redis — migrar de MemorySaver a PostgresSaver es cambiar una línea de código. Todo lo demás (thread_id, get_state, get_state_history) funciona exactamente igual. El valor: tus checkpoints sobreviven a crashes, restarts, y deploys.

Recursos adicionales

1. LangGraph — Persistence — Documentación oficial de checkpointing en LangGraph. Cubre MemorySaver, PostgresSaver, y la arquitectura general del sistema de persistencia
2. LangGraph — How to add thread-level persistence — Guía paso a paso para implementar checkpointing con thread_id
3. LangGraph — How to manage conversation history — Estrategias para mantener el historial de mensajes bajo control cuando usas checkpointing
4. LangGraph — Time travel — Guía oficial de time-travel debugging: navegar checkpoints, hacer replay, y bifurcar ejecuciones
5. LangGraph — MemorySaver Reference — API reference del MemorySaver con todos los métodos disponibles

Módulo 8 — LangChain & LangGraph: From Chains to Agents