Módulo 8: Memoria y Persistencia

Time-Travel Debugging

Siguiente →

Descripción de la cápsula

Time-travel debugging te permite navegar el historial completo de ejecución de tu agente — cada estado en cada paso. Puedes retroceder a cualquier punto y entender exactamente qué sabía el agente y por qué tomó una decisión específica. Puedes hacer replay desde cualquier paso, bifurcar la ejecución para explorar caminos alternativos, y diagnosticar problemas sin re-ejecutar nada.

Esto no es una curiosidad ni un feature de demo. Es la herramienta de debugging más poderosa que existe para agentes. En programación tradicional, cuando algo falla, agregas print statements, re-ejecutas, y tratas de adivinar qué pasó. Con agentes, esa estrategia no funciona — una ejecución puede tomar minutos, costar dinero en API calls, y depender de respuestas no deterministas de un LLM. Re-ejecutar no garantiza que veas el mismo bug.

Con time-travel debugging: navegas directamente al paso donde algo salió mal, inspeccionas el estado completo, entiendes la causa raíz, corriges, y haces replay para verificar. Sin gastar tiempo ni dinero extra.

El escenario que lo hace imprescindible

Tu Research Assistant generó un reporte sobre "AI Safety" y la sección de resumen dice algo raro — mezcla datos de dos temas diferentes. ¿Qué haces?

Sin time-travel debugging:

1. Miras el output final → "el resumen está mal"
2. ¿Dónde falló? No sabes → agregas print statements
3. Re-ejecutas → $0.50 en API calls, 3 minutos de espera
4. El LLM responde diferente esta vez → no puedes reproducir el bug
5. Repites 3-4 veces → $2.00 y 12 minutos perdidos
6. Sigues sin entender qué pasó

Con time-travel debugging:

1. Miras el output final → "el resumen está mal"
2. get_state_history() → ves los 8 pasos de la ejecución
3. Navegas al paso 6 (síntesis) → inspeccionas qué fuentes tenía
4. Descubres: el nodo de búsqueda retornó datos del tema equivocado en paso 3
5. Navegas al paso 3 → ves exactamente el query que se envió y la respuesta
6. Causa raíz: el query se construyó mal. Fix en 2 minutos, $0.00 extra

La diferencia no es solo eficiencia — es que puedes diagnosticar bugs que son imposibles de reproducir porque dependen del estado específico que tenía el agente en ese momento.

Checkpoint history: la línea temporal de tu agente

Cada vez que un nodo completa, LangGraph guarda un checkpoint. El historial de checkpoints es una lista ordenada de todos los estados por los que pasó el agente:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sources: Annotated[list[str], operator.add]
    step_count: int

def search(state: State) -> dict:
    return {
        "sources": ["Wikipedia: AI safety es un campo de investigación..."],
        "step_count": state.get("step_count", 0) + 1,
    }

def analyze(state: State) -> dict:
    return {
        "sources": ["Análisis: 3 tendencias principales identificadas..."],
        "step_count": state.get("step_count", 0) + 1,
    }

def synthesize(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(state["sources"])
    response = model.invoke(
        f"Sintetiza brevemente (2 oraciones):\n\n{context}"
    )
    return {
        "messages": [response],
        "step_count": state.get("step_count", 0) + 1,
    }

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("synthesize", synthesize)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "synthesize")
graph_builder.add_edge("synthesize", END)

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

config = {"configurable": {"thread_id": "debug-001"}}
result = graph.invoke(
    {"messages": [HumanMessage(content="Investiga AI safety")], "sources": [], "step_count": 0},
    config,
)

print("=== Historial de checkpoints ===\n")
for i, state in enumerate(graph.get_state_history(config)):
    node = state.metadata.get("source", "unknown")
    step = state.metadata.get("step", -1)
    checkpoint_id = state.config["configurable"]["checkpoint_id"]
    print(f"Checkpoint {i}:")
    print(f"  ID: {checkpoint_id[:16]}...")
    print(f"  Nodo: {node}")
    print(f"  Step: {step}")
    print(f"  Sources: {len(state.values.get('sources', []))}")
    print(f"  Messages: {len(state.values.get('messages', []))}")
    print(f"  Siguiente: {state.next}")
    print()
# Output esperado:
# === Historial de checkpoints ===
#
# Checkpoint 0:
#   ID: 1ef8a1b2c3d4e5f6...
#   Nodo: synthesize
#   Step: 3
#   Sources: 2
#   Messages: 2
#   Siguiente: ()
#
# Checkpoint 1:
#   ID: 1ef8a1b2c3d4e5f5...
#   Nodo: analyze
#   Step: 2
#   Sources: 2
#   Messages: 1
#   Siguiente: ('synthesize',)
#
# Checkpoint 2:
#   ID: 1ef8a1b2c3d4e5f4...
#   Nodo: search
#   Step: 1
#   Sources: 1
#   Messages: 1
#   Siguiente: ('analyze',)
#
# Checkpoint 3:
#   ID: 1ef8a1b2c3d4e5f3...
#   Nodo: __start__
#   Step: 0
#   Sources: 0
#   Messages: 1
#   Siguiente: ('search',)

get_state_history retorna los checkpoints en orden inverso (el más reciente primero). Cada checkpoint contiene:

  • ✅ values: el estado completo del grafo en ese punto
  • ✅ metadata: qué nodo generó este checkpoint, el step number, timestamp
  • ✅ config: incluye el checkpoint_id único para poder navegar a este punto
  • ✅ next: qué nodo iba a ejecutarse después (tupla vacía si la ejecución terminó)

Inspeccionar un checkpoint específico

Cuando encuentras un checkpoint sospechoso en el historial, puedes inspeccionar su estado completo:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    query: str
    search_results: list[str]
    analysis: str
    final_report: str

def search(state: State) -> dict:
    results = [
        f"Fuente 1: Datos sobre '{state['query']}' de Wikipedia",
        f"Fuente 2: Paper sobre '{state['query']}' de arXiv",
        f"Fuente 3: Artículo sobre '{state['query']}' de TechCrunch",
    ]
    return {"search_results": results}

def analyze(state: State) -> dict:
    analysis = f"Análisis de {len(state['search_results'])} fuentes: "
    analysis += "convergencia en 3 tendencias principales."
    return {"analysis": analysis}

def report(state: State) -> dict:
    report_text = f"Reporte: {state['analysis']} "
    report_text += f"Basado en {len(state['search_results'])} fuentes."
    return {"final_report": report_text}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("report", report)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "inspect-demo"}}
result = graph.invoke(
    {
        "messages": [HumanMessage(content="Investiga quantum computing")],
        "query": "quantum computing",
        "search_results": [],
        "analysis": "",
        "final_report": "",
    },
    config,
)

history = list(graph.get_state_history(config))

print("=== Inspección del checkpoint después de 'search' ===\n")
for checkpoint in history:
    if checkpoint.metadata.get("source") == "search":
        print(f"Checkpoint ID: {checkpoint.config['configurable']['checkpoint_id'][:20]}...")
        print(f"Nodo que lo generó: {checkpoint.metadata.get('source')}")
        print(f"Siguiente nodo: {checkpoint.next}")
        print(f"\nEstado completo en este punto:")
        print(f"  query: {checkpoint.values['query']}")
        print(f"  search_results ({len(checkpoint.values['search_results'])}):")
        for r in checkpoint.values["search_results"]:
            print(f"    - {r}")
        print(f"  analysis: '{checkpoint.values['analysis']}'")
        print(f"  final_report: '{checkpoint.values['final_report']}'")
        break
# Output esperado:
# === Inspección del checkpoint después de 'search' ===
#
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...
# Nodo que lo generó: search
# Siguiente nodo: ('analyze',)
#
# Estado completo en este punto:
#   query: quantum computing
#   search_results (3):
#     - Fuente 1: Datos sobre 'quantum computing' de Wikipedia
#     - Fuente 2: Paper sobre 'quantum computing' de arXiv
#     - Fuente 3: Artículo sobre 'quantum computing' de TechCrunch
#   analysis: ''
#   final_report: ''

Observa que después de search:

  • search_results tiene 3 fuentes (el nodo search las produjo)
  • analysis está vacío (el nodo analyze aún no se ejecutó)
  • final_report está vacío (el nodo report aún no se ejecutó)

Puedes ver exactamente qué sabía el agente en cada punto de la ejecución. Si el reporte final tiene un error, navegas hacia atrás checkpoint por checkpoint hasta encontrar dónde apareció el dato incorrecto.

Replay desde un checkpoint: "¿qué hubiera pasado si...?"

Replay es la capacidad de re-ejecutar el grafo desde un checkpoint histórico. Esto tiene dos usos principales:

  1. Verificar un fix: cambiaste la lógica de un nodo y quieres ver si el mismo input produce un resultado correcto
  2. Explorar alternativas: "¿qué hubiera pasado si el agente tenía estos datos en vez de estos?"
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    input: str
    step_a_result: str
    step_b_result: str
    final: str
    path_taken: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
    return {
        "step_a_result": f"A procesó: '{state['input']}'",
        "path_taken": ["step_a"],
    }

def step_b(state: State) -> dict:
    return {
        "step_b_result": f"B analizó: '{state['step_a_result']}'",
        "path_taken": ["step_b"],
    }

def step_final(state: State) -> dict:
    return {
        "final": f"Resultado: {state['step_b_result']}",
        "path_taken": ["final"],
    }

graph_builder = StateGraph(State)
graph_builder.add_node("a", step_a)
graph_builder.add_node("b", step_b)
graph_builder.add_node("final", step_final)

graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", "final")
graph_builder.add_edge("final", END)

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

config = {"configurable": {"thread_id": "replay-demo"}}
result = graph.invoke(
    {"input": "datos originales", "step_a_result": "", "step_b_result": "", "final": "", "path_taken": []},
    config,
)
print(f"Ejecución original: {result['final']}")
print(f"Path: {result['path_taken']}")

history = list(graph.get_state_history(config))
target_checkpoint = None
for cp in history:
    if cp.metadata.get("source") == "a":
        target_checkpoint = cp
        break

if target_checkpoint:
    checkpoint_id = target_checkpoint.config["configurable"]["checkpoint_id"]
    print(f"\n--- Replay desde checkpoint después de step_a ---")
    print(f"Checkpoint ID: {checkpoint_id[:20]}...")
    print(f"Estado en este punto: step_a_result = '{target_checkpoint.values['step_a_result']}'")

    replay_config = {
        "configurable": {
            "thread_id": "replay-demo",
            "checkpoint_id": checkpoint_id,
        }
    }

    replay_result = graph.invoke(None, replay_config)
    print(f"\nReplay resultado: {replay_result['final']}")
    print(f"Replay path: {replay_result['path_taken']}")
# Output esperado:
# Ejecución original: Resultado: B analizó: 'A procesó: 'datos originales''
# Path: ['step_a', 'step_b', 'final']
#
# --- Replay desde checkpoint después de step_a ---
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...
# Estado en este punto: step_a_result = 'A procesó: 'datos originales''
#
# Replay resultado: Resultado: B analizó: 'A procesó: 'datos originales''
# Replay path: ['step_a', 'step_b', 'final']

La mecánica del replay:

  1. Obtienes el checkpoint_id del punto al que quieres retroceder
  2. Invocas el grafo con None como input (el estado viene del checkpoint) y el checkpoint_id en la config
  3. El grafo ejecuta desde ese punto hacia adelante

Pasas None porque no estás proporcionando input nuevo — el estado completo ya existe en el checkpoint.

Forking: bifurcar desde un punto histórico

Forking va un paso más allá del replay. En vez de re-ejecutar con el mismo estado, modificas el estado en un punto histórico y ejecutas desde ahí. Esto crea una rama alternativa de la ejecución:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    search_data: str
    analysis: str
    path: Annotated[list[str], operator.add]

def search(state: State) -> dict:
    return {
        "search_data": f"Resultados básicos sobre '{state['query']}'",
        "path": ["search"],
    }

def analyze(state: State) -> dict:
    return {
        "analysis": f"Análisis de: {state['search_data']}",
        "path": ["analyze"],
    }

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", END)

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

config = {"configurable": {"thread_id": "fork-demo"}}
original = graph.invoke(
    {"query": "LangGraph", "search_data": "", "analysis": "", "path": []},
    config,
)
print(f"Original: {original['analysis']}")

history = list(graph.get_state_history(config))
search_checkpoint = None
for cp in history:
    if cp.metadata.get("source") == "search":
        search_checkpoint = cp
        break

if search_checkpoint:
    checkpoint_id = search_checkpoint.config["configurable"]["checkpoint_id"]

    fork_config = {
        "configurable": {
            "thread_id": "fork-demo-branch",
            "checkpoint_id": checkpoint_id,
        }
    }

    graph.update_state(
        fork_config,
        {"search_data": "Resultados PREMIUM con 50 papers y datos exclusivos sobre 'LangGraph'"},
    )

    fork_result = graph.invoke(None, fork_config)
    print(f"Fork:     {fork_result['analysis']}")

    original_state = graph.get_state(config)
    print(f"\nOriginal no cambió: {original_state.values['analysis']}")
# Output esperado:
# Original: Análisis de: Resultados básicos sobre 'LangGraph'
# Fork:     Análisis de: Resultados PREMIUM con 50 papers y datos exclusivos sobre 'LangGraph'
#
# Original no cambió: Análisis de: Resultados básicos sobre 'LangGraph'

El forking permite responder preguntas como:

  • "¿Qué hubiera pasado si la búsqueda retornaba datos diferentes?"
  • "¿Qué pasa si cambio el prompt del nodo de análisis?"
  • "¿Qué resultado produce si le doy más contexto al agente en el paso 3?"

Y lo más importante: la ejecución original no se modifica. El fork crea una rama independiente con un thread_id diferente.

Workflow práctico de debugging

Aquí está el workflow completo que usarás cada vez que tu agente produzca un resultado inesperado:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    query: str
    raw_data: str
    processed_data: str
    summary: str

def fetch_data(state: State) -> dict:
    return {"raw_data": f"[DATOS] Información sobre '{state['query']}': AI agents usan LLMs para razonar."}

def process_data(state: State) -> dict:
    processed = state["raw_data"].replace("[DATOS]", "[PROCESADO]")
    processed += " NOTA: datos verificados."
    return {"processed_data": processed}

def summarize(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Resume en una oración: {state['processed_data']}"
    )
    return {"summary": response.content, "messages": [response]}

graph_builder = StateGraph(State)
graph_builder.add_node("fetch", fetch_data)
graph_builder.add_node("process", process_data)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "fetch")
graph_builder.add_edge("fetch", "process")
graph_builder.add_edge("process", "summarize")
graph_builder.add_edge("summarize", END)

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

config = {"configurable": {"thread_id": "debug-workflow"}}
result = graph.invoke(
    {
        "messages": [HumanMessage(content="Investiga AI agents")],
        "query": "AI agents",
        "raw_data": "",
        "processed_data": "",
        "summary": "",
    },
    config,
)

print("PASO 1: El output no es lo que esperábamos")
print(f"  Summary: {result['summary']}\n")

print("PASO 2: Obtener historial completo")
history = list(graph.get_state_history(config))
print(f"  Total checkpoints: {len(history)}\n")

print("PASO 3: Inspeccionar cada paso")
for cp in reversed(history):
    node = cp.metadata.get("source", "?")
    print(f"  [{node}]")
    if cp.values.get("raw_data"):
        print(f"    raw_data: {cp.values['raw_data'][:60]}...")
    if cp.values.get("processed_data"):
        print(f"    processed_data: {cp.values['processed_data'][:60]}...")
    if cp.values.get("summary"):
        print(f"    summary: {cp.values['summary'][:60]}...")

print("\nPASO 4: Encontrar el checkpoint sospechoso")
for cp in history:
    if cp.metadata.get("source") == "process":
        print(f"  Después de 'process':")
        print(f"    processed_data = '{cp.values['processed_data']}'")
        print(f"    ¿Los datos se procesaron correctamente? → Inspeccionar lógica del nodo")
        break

print("\nPASO 5: Verificar fix con replay (después de corregir el nodo)")
print("  Usarías: graph.invoke(None, config_con_checkpoint_id)")
# Output esperado:
# PASO 1: El output no es lo que esperábamos
#   Summary: AI agents use LLMs to reason and make decisions...
#
# PASO 2: Obtener historial completo
#   Total checkpoints: 4
#
# PASO 3: Inspeccionar cada paso
#   [__start__]
#   [fetch]
#     raw_data: [DATOS] Información sobre 'AI agents': AI agents usan L...
#   [process]
#     raw_data: [DATOS] Información sobre 'AI agents': AI agents usan L...
#     processed_data: [PROCESADO] Información sobre 'AI agents': AI agents u...
#   [summarize]
#     raw_data: [DATOS] Información sobre 'AI agents': AI agents usan L...
#     processed_data: [PROCESADO] Información sobre 'AI agents': AI agents u...
#     summary: AI agents use large language models (LLMs) for reasonin...
#
# PASO 4: Encontrar el checkpoint sospechoso
#   Después de 'process':
#     processed_data = '[PROCESADO] Información sobre 'AI agents'...'
#     ¿Los datos se procesaron correctamente? → Inspeccionar lógica del nodo
#
# PASO 5: Verificar fix con replay (después de corregir el nodo)
#   Usarías: graph.invoke(None, config_con_checkpoint_id)

El workflow en resumen

1. Agente produce output inesperado
     ↓
2. get_state_history(config) → lista de todos los checkpoints
     ↓
3. Iterar checkpoints: inspeccionar estado en cada paso
     ↓
4. Encontrar el paso donde apareció el dato incorrecto
     ↓
5. Comparar estado ANTES y DESPUÉS del nodo sospechoso
     ↓
6. Causa raíz identificada → corregir lógica del nodo
     ↓
7. Replay desde el checkpoint anterior al error → verificar fix

Comparación: debugging tradicional vs time-travel

Para que quede claro por qué esto es transformador:

AspectoDebugging tradicionalTime-travel debugging
Ver estado intermedioprint() + re-ejecutarNavegar al checkpoint
Costo de debuggingRe-ejecutar = API calls + tiempo$0.00 (datos ya guardados)
ReproducibilidadNo garantizada (LLM no determinista)Estado exacto preservado
"¿Qué pasó en el paso 5?"Agregar logging, re-ejecutar, esperarget_state_history() → inspeccionar
"¿Qué pasa si cambio X?"Modificar código, re-ejecutar todoFork + update_state + replay
Debugging en producciónLogs, métricas, adivinarNavegar el historial exacto
Tiempo para diagnósticoMinutos a horasSegundos a minutos

Metadata de checkpoints: contexto adicional

Cada checkpoint incluye metadata que ayuda a entender el contexto de la ejecución:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    log: Annotated[list[str], operator.add]

def node_a(state: State) -> dict:
    return {"data": "resultado de A", "log": ["A ejecutado"]}

def node_b(state: State) -> dict:
    return {"data": "resultado de B", "log": ["B ejecutado"]}

graph_builder = StateGraph(State)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", END)

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

config = {"configurable": {"thread_id": "metadata-demo"}}
graph.invoke({"data": "", "log": []}, config)

print("=== Metadata de cada checkpoint ===\n")
for cp in graph.get_state_history(config):
    print(f"Nodo: {cp.metadata.get('source', '?')}")
    print(f"  step: {cp.metadata.get('step')}")
    print(f"  writes: {cp.metadata.get('writes')}")
    print(f"  checkpoint_id: {cp.config['configurable']['checkpoint_id'][:20]}...")
    parent = cp.config['configurable'].get('checkpoint_ns', '')
    print(f"  parent_checkpoint_id: {cp.parent_config}")
    print()
# Output esperado:
# === Metadata de cada checkpoint ===
#
# Nodo: b
#   step: 2
#   writes: {'b': {'data': 'resultado de B', 'log': ['B ejecutado']}}
#   checkpoint_id: 1ef8a1b2c3d4e5f6ab...
#   parent_checkpoint_id: {'configurable': {'thread_id': 'metadata-demo', 'checkpoint_id': '...'}}
#
# Nodo: a
#   step: 1
#   writes: {'a': {'data': 'resultado de A', 'log': ['A ejecutado']}}
#   checkpoint_id: 1ef8a1b2c3d4e5f5ab...
#   parent_checkpoint_id: {'configurable': {'thread_id': 'metadata-demo', 'checkpoint_id': '...'}}
#
# Nodo: __start__
#   step: 0
#   writes: {'__start__': {'data': '', 'log': []}}
#   checkpoint_id: 1ef8a1b2c3d4e5f4ab...
#   parent_checkpoint_id: None

Los campos más útiles para debugging:

  • ✅ source: qué nodo generó este checkpoint
  • ✅ step: número secuencial del paso
  • ✅ writes: exactamente qué escribió el nodo en el estado (el "diff")
  • ✅ parent_config: el checkpoint del paso anterior (para navegar hacia atrás)

El campo writes es especialmente poderoso — te muestra no solo el estado completo, sino qué cambió en este paso específico.

Debugging con agentes reales: un ejemplo completo

Veamos un caso más realista donde el time-travel debugging resuelve un problema concreto:

from dotenv import load_dotenv
load_dotenv()

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

class ResearchState(TypedDict):
    topic: str
    sources: Annotated[list[dict], operator.add]
    filtered_sources: list[dict]
    analysis: str
    report: str

def gather_sources(state: ResearchState) -> dict:
    return {"sources": [
        {"name": "Wikipedia", "data": f"Info enciclopédica sobre {state['topic']}", "relevance": 0.9},
        {"name": "arXiv", "data": f"Papers sobre {state['topic']}", "relevance": 0.8},
        {"name": "Reddit", "data": "Memes sobre gatos", "relevance": 0.1},
    ]}

def filter_sources(state: ResearchState) -> dict:
    threshold = 0.5
    filtered = [s for s in state["sources"] if s["relevance"] >= threshold]
    return {"filtered_sources": filtered}

def analyze_sources(state: ResearchState) -> dict:
    source_names = [s["name"] for s in state["filtered_sources"]]
    source_data = " | ".join(s["data"] for s in state["filtered_sources"])
    return {"analysis": f"Análisis basado en {source_names}: {source_data}"}

def generate_report(state: ResearchState) -> dict:
    return {"report": f"REPORTE: {state['analysis']}. Total fuentes: {len(state['filtered_sources'])}."}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("gather", gather_sources)
graph_builder.add_node("filter", filter_sources)
graph_builder.add_node("analyze", analyze_sources)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "gather")
graph_builder.add_edge("gather", "filter")
graph_builder.add_edge("filter", "analyze")
graph_builder.add_edge("analyze", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "real-debug"}}
result = graph.invoke(
    {"topic": "transformer architectures", "sources": [], "filtered_sources": [], "analysis": "", "report": ""},
    config,
)

print("=== El reporte final parece bien, pero ¿está correcta la filtración? ===")
print(f"Reporte: {result['report']}\n")

print("=== Debugging: verificar qué pasó en cada paso ===\n")

history = list(graph.get_state_history(config))

for cp in history:
    node = cp.metadata.get("source", "?")

    if node == "gather":
        print(f"[gather] Fuentes recolectadas: {len(cp.values['sources'])}")
        for s in cp.values["sources"]:
            print(f"  - {s['name']} (relevance: {s['relevance']}): {s['data'][:40]}...")

    elif node == "filter":
        print(f"\n[filter] Fuentes después de filtrar: {len(cp.values['filtered_sources'])}")
        removed = [s for s in cp.values["sources"] if s not in cp.values["filtered_sources"]]
        for s in cp.values["filtered_sources"]:
            print(f"  ✅ {s['name']} (relevance: {s['relevance']})")
        for s in removed:
            print(f"  ❌ {s['name']} (relevance: {s['relevance']}) — FILTRADA")

    elif node == "analyze":
        print(f"\n[analyze] Análisis generado:")
        print(f"  {cp.values['analysis'][:80]}...")

    elif node == "report":
        print(f"\n[report] Reporte final:")
        print(f"  {cp.values['report'][:80]}...")
# Output esperado:
# === El reporte final parece bien, pero ¿está correcta la filtración? ===
# Reporte: REPORTE: Análisis basado en ['Wikipedia', 'arXiv']: Info enciclopédica sobre transformer architectures | Papers sobre transformer architectures. Total fuentes: 2.
#
# === Debugging: verificar qué pasó en cada paso ===
#
# [gather] Fuentes recolectadas: 3
#   - Wikipedia (relevance: 0.9): Info enciclopédica sobre transformer arc...
#   - arXiv (relevance: 0.8): Papers sobre transformer architectures...
#   - Reddit (relevance: 0.1): Memes sobre gatos...
#
# [filter] Fuentes después de filtrar: 2
#   ✅ Wikipedia (relevance: 0.9)
#   ✅ arXiv (relevance: 0.8)
#   ❌ Reddit (relevance: 0.1) — FILTRADA
#
# [analyze] Análisis generado:
#   Análisis basado en ['Wikipedia', 'arXiv']: Info enciclopédica sobre transfo...
#
# [report] Reporte final:
#   REPORTE: Análisis basado en ['Wikipedia', 'arXiv']: Info enciclopédica sobr...

Con este debugging puedes verificar:

  • ✅ ¿Se recolectaron las fuentes correctas? → Sí, 3 fuentes
  • ✅ ¿La filtración fue correcta? → Sí, Reddit (relevance 0.1) fue filtrada
  • ✅ ¿El análisis usó las fuentes correctas? → Sí, Wikipedia y arXiv
  • ✅ ¿El reporte refleja el análisis? → Sí

Si algo estuviera mal (por ejemplo, Reddit pasó el filtro), sabrías exactamente en qué paso ocurrió el error.

Forking avanzado: comparar decisiones alternativas

Un caso de uso poderoso: tu agente filtró fuentes con un threshold de 0.5. ¿Qué pasa si cambias el threshold a 0.3? En vez de re-ejecutar todo, haces fork desde antes del filtro:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    sources: Annotated[list[dict], operator.add]
    filtered: list[dict]
    threshold: float
    summary: str

def gather(state: State) -> dict:
    return {"sources": [
        {"name": "Source A", "score": 0.9},
        {"name": "Source B", "score": 0.4},
        {"name": "Source C", "score": 0.7},
        {"name": "Source D", "score": 0.2},
    ]}

def filter_sources(state: State) -> dict:
    threshold = state.get("threshold", 0.5)
    filtered = [s for s in state["sources"] if s["score"] >= threshold]
    return {"filtered": filtered}

def summarize(state: State) -> dict:
    names = [s["name"] for s in state["filtered"]]
    return {"summary": f"Resumen basado en {len(names)} fuentes: {', '.join(names)}"}

graph_builder = StateGraph(State)
graph_builder.add_node("gather", gather)
graph_builder.add_node("filter", filter_sources)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "gather")
graph_builder.add_edge("gather", "filter")
graph_builder.add_edge("filter", "summarize")
graph_builder.add_edge("summarize", END)

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

config_original = {"configurable": {"thread_id": "compare-original"}}
original = graph.invoke(
    {"topic": "AI", "sources": [], "filtered": [], "threshold": 0.5, "summary": ""},
    config_original,
)
print(f"Original (threshold=0.5): {original['summary']}")

history = list(graph.get_state_history(config_original))
gather_checkpoint = None
for cp in history:
    if cp.metadata.get("source") == "gather":
        gather_checkpoint = cp
        break

if gather_checkpoint:
    checkpoint_id = gather_checkpoint.config["configurable"]["checkpoint_id"]

    config_fork = {
        "configurable": {
            "thread_id": "compare-fork",
            "checkpoint_id": checkpoint_id,
        }
    }

    graph.update_state(config_fork, {"threshold": 0.3})

    fork_result = graph.invoke(None, config_fork)
    print(f"Fork (threshold=0.3):    {fork_result['summary']}")

    original_check = graph.get_state(config_original)
    print(f"\nOriginal sin cambios:    {original_check.values['summary']}")
# Output esperado:
# Original (threshold=0.5): Resumen basado en 2 fuentes: Source A, Source C
# Fork (threshold=0.3):    Resumen basado en 3 fuentes: Source A, Source B, Source C
#
# Original sin cambios:    Resumen basado en 2 fuentes: Source A, Source C

Con threshold 0.5 se incluyeron 2 fuentes. Con threshold 0.3 se incluyeron 3 (Source B ahora pasa). Puedes comparar ambos resultados sin haber re-ejecutado el paso de gather (que podría haber costado API calls).

Construir una función de debugging reutilizable

Para que el debugging sea práctico en tu día a día, encapsula la lógica en una función:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    input: str
    step_a: str
    step_b: str
    output: str
    log: Annotated[list[str], operator.add]

def debug_execution(graph, config, fields_to_show=None):
    """Muestra el historial completo de una ejecución para debugging."""
    history = list(graph.get_state_history(config))

    print(f"{'='*60}")
    print(f"Thread: {config['configurable']['thread_id']}")
    print(f"Total checkpoints: {len(history)}")
    print(f"{'='*60}\n")

    for i, cp in enumerate(reversed(history)):
        node = cp.metadata.get("source", "?")
        step = cp.metadata.get("step", "?")
        checkpoint_id = cp.config["configurable"]["checkpoint_id"][:12]

        print(f"Step {step} | Nodo: {node} | ID: {checkpoint_id}...")

        writes = cp.metadata.get("writes", {})
        if writes:
            for node_name, node_writes in writes.items():
                if isinstance(node_writes, dict):
                    for key, value in node_writes.items():
                        if fields_to_show is None or key in fields_to_show:
                            val_str = str(value)
                            if len(val_str) > 80:
                                val_str = val_str[:80] + "..."
                            print(f"  → {key} = {val_str}")

        if cp.next:
            print(f"  siguiente: {cp.next}")
        else:
            print(f"  [EJECUCIÓN COMPLETA]")
        print()

def node_a(state: State) -> dict:
    return {"step_a": f"A procesó: {state['input']}", "log": ["a_done"]}

def node_b(state: State) -> dict:
    return {"step_b": f"B analizó: {state['step_a']}", "log": ["b_done"]}

def node_output(state: State) -> dict:
    return {"output": f"Final: {state['step_b']}", "log": ["output_done"]}

graph_builder = StateGraph(State)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_node("output", node_output)

graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", "output")
graph_builder.add_edge("output", END)

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

config = {"configurable": {"thread_id": "debug-util"}}
graph.invoke({"input": "datos de prueba", "step_a": "", "step_b": "", "output": "", "log": []}, config)

debug_execution(graph, config, fields_to_show=["step_a", "step_b", "output"])
# Output esperado:
# ============================================================
# Thread: debug-util
# Total checkpoints: 4
# ============================================================
#
# Step 0 | Nodo: __start__ | ID: 1ef8a1b2c3d4...
#   siguiente: ('a',)
#
# Step 1 | Nodo: a | ID: 1ef8a1b2c3d5...
#   → step_a = A procesó: datos de prueba
#   siguiente: ('b',)
#
# Step 2 | Nodo: b | ID: 1ef8a1b2c3d6...
#   → step_b = B analizó: A procesó: datos de prueba
#   siguiente: ('output',)
#
# Step 3 | Nodo: output | ID: 1ef8a1b2c3d7...
#   → output = Final: B analizó: A procesó: datos de prueba
#   [EJECUCIÓN COMPLETA]

Puedes usar debug_execution() en cualquier grafo. Solo necesitas el grafo compilado y el config. El parámetro fields_to_show te permite filtrar qué campos del estado quieres ver (útil cuando el estado tiene muchos campos).

Troubleshooting

Problema 1: "get_state_history retorna una lista vacía"

Síntoma: Llamas a graph.get_state_history(config) y no obtienes checkpoints.

Causa: El grafo no tiene checkpointer, o el thread_id no coincide.

Solución: Verifica que compilaste con checkpointer y que el thread_id es correcto:

# ❌ Sin checkpointer — no hay historial
graph = graph_builder.compile()

# ✅ Con checkpointer
graph = graph_builder.compile(checkpointer=MemorySaver())

# Verificar thread_id
config = {"configurable": {"thread_id": "mi-thread"}}  # mismo ID que la ejecución

Problema 2: "Replay produce un resultado diferente al original"

Síntoma: Haces replay desde un checkpoint y el resultado es distinto.

Causa: Los nodos contienen lógica no determinista (llamadas a LLM, timestamps, random) que produce resultados diferentes cada vez.

Solución: Esto es esperado con LLMs. El replay re-ejecuta los nodos desde el checkpoint, y un LLM puede responder diferente. Para debugging, lo importante es que el estado de entrada sea el mismo — puedes comparar el input y entender por qué el output cambió.

Problema 3: "No puedo encontrar el checkpoint donde ocurrió el error"

Síntoma: Tienes muchos checkpoints y no sabes cuál inspeccionar.

Causa: No estás usando la metadata para filtrar.

Solución: Usa metadata['source'] para filtrar por nodo:

for cp in graph.get_state_history(config):
    if cp.metadata.get("source") == "nodo_sospechoso":
        print(f"Estado: {cp.values}")
        break

Problema 4: "update_state no tiene efecto en el fork"

Síntoma: Haces graph.update_state() pero el fork ejecuta con el estado original.

Causa: El checkpoint_id o thread_id no coinciden, o la actualización se hizo en la config incorrecta.

Solución: Verifica que la config del fork tenga el checkpoint_id correcto:

fork_config = {
    "configurable": {
        "thread_id": "nuevo-thread-para-fork",
        "checkpoint_id": checkpoint_id_del_historial,
    }
}
graph.update_state(fork_config, {"campo": "nuevo_valor"})
result = graph.invoke(None, fork_config)

Ejercicios

Ejercicio 1: Explorar el historial de checkpoints (Fácil)

Crea un grafo de 3 nodos secuenciales. Ejecuta una invocación y luego usa get_state_history() para imprimir: el nombre del nodo, el step number, y el next de cada checkpoint. Verifica que el historial tiene 4 checkpoints (start + 3 nodos).

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    log: Annotated[list[str], operator.add]

def node_1(state: State) -> dict:
    return {"data": "resultado_1", "log": ["n1"]}

def node_2(state: State) -> dict:
    return {"data": "resultado_2", "log": ["n2"]}

def node_3(state: State) -> dict:
    return {"data": "resultado_3", "log": ["n3"]}

graph_builder = StateGraph(State)
graph_builder.add_node("n1", node_1)
graph_builder.add_node("n2", node_2)
graph_builder.add_node("n3", node_3)

graph_builder.add_edge(START, "n1")
graph_builder.add_edge("n1", "n2")
graph_builder.add_edge("n2", "n3")
graph_builder.add_edge("n3", END)

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

config = {"configurable": {"thread_id": "history-exercise"}}
graph.invoke({"data": "", "log": []}, config)

history = list(graph.get_state_history(config))
print(f"Total checkpoints: {len(history)}\n")

for cp in reversed(history):
    node = cp.metadata.get("source", "?")
    step = cp.metadata.get("step", "?")
    print(f"Step {step} | Nodo: {node} | Next: {cp.next}")
# Output esperado:
# Total checkpoints: 4
#
# Step 0 | Nodo: __start__ | Next: ('n1',)
# Step 1 | Nodo: n1 | Next: ('n2',)
# Step 2 | Nodo: n2 | Next: ('n3',)
# Step 3 | Nodo: n3 | Next: ()

Ejercicio 2: Inspeccionar writes de cada checkpoint (Fácil)

Usando el mismo grafo del ejercicio 1, imprime los writes de cada checkpoint (lo que cada nodo escribió en el estado). Esto te muestra el "diff" de cada paso.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    counter: int
    log: Annotated[list[str], operator.add]

def increment_a(state: State) -> dict:
    return {"data": "from_a", "counter": 10, "log": ["a"]}

def increment_b(state: State) -> dict:
    return {"data": "from_b", "counter": state["counter"] + 5, "log": ["b"]}

def increment_c(state: State) -> dict:
    return {"data": "from_c", "counter": state["counter"] + 3, "log": ["c"]}

graph_builder = StateGraph(State)
graph_builder.add_node("a", increment_a)
graph_builder.add_node("b", increment_b)
graph_builder.add_node("c", increment_c)

graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", "c")
graph_builder.add_edge("c", END)

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

config = {"configurable": {"thread_id": "writes-exercise"}}
graph.invoke({"data": "", "counter": 0, "log": []}, config)

for cp in reversed(list(graph.get_state_history(config))):
    node = cp.metadata.get("source", "?")
    writes = cp.metadata.get("writes", {})
    print(f"[{node}] writes:")
    for node_name, node_writes in writes.items():
        if isinstance(node_writes, dict):
            for key, value in node_writes.items():
                print(f"  {key} = {value}")
    print()
# Output esperado:
# [__start__] writes:
#   data =
#   counter = 0
#   log = []
#
# [a] writes:
#   data = from_a
#   counter = 10
#   log = ['a']
#
# [b] writes:
#   data = from_b
#   counter = 15
#   log = ['b']
#
# [c] writes:
#   data = from_c
#   counter = 18
#   log = ['c']

Ejercicio 3: Replay desde un checkpoint específico (Medio)

Crea un grafo de 4 nodos. Ejecuta completamente. Luego obtén el checkpoint después del nodo 2 y haz replay desde ahí. Verifica que el replay solo ejecuta nodos 3 y 4 (no 1 y 2).

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

EXECUTION_LOG = []

class State(TypedDict):
    data: str
    steps: Annotated[list[str], operator.add]

def make_node(name: str):
    def node(state: State) -> dict:
        EXECUTION_LOG.append(name)
        return {"data": f"output_{name}", "steps": [name]}
    return node

graph_builder = StateGraph(State)
for name in ["n1", "n2", "n3", "n4"]:
    graph_builder.add_node(name, make_node(name))

graph_builder.add_edge(START, "n1")
graph_builder.add_edge("n1", "n2")
graph_builder.add_edge("n2", "n3")
graph_builder.add_edge("n3", "n4")
graph_builder.add_edge("n4", END)

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

config = {"configurable": {"thread_id": "replay-exercise"}}

EXECUTION_LOG = []
graph.invoke({"data": "", "steps": []}, config)
print(f"Ejecución original: {EXECUTION_LOG}")

checkpoint_after_n2 = None
for cp in graph.get_state_history(config):
    if cp.metadata.get("source") == "n2":
        checkpoint_after_n2 = cp
        break

EXECUTION_LOG = []
replay_config = {
    "configurable": {
        "thread_id": "replay-exercise",
        "checkpoint_id": checkpoint_after_n2.config["configurable"]["checkpoint_id"],
    }
}
replay_result = graph.invoke(None, replay_config)
print(f"Replay (desde n2): {EXECUTION_LOG}")
print(f"Steps en resultado: {replay_result['steps']}")
# Output esperado:
# Ejecución original: ['n1', 'n2', 'n3', 'n4']
# Replay (desde n2): ['n3', 'n4']
# Steps en resultado: ['n1', 'n2', 'n3', 'n4']

Ejercicio 4: Fork con estado modificado (Medio)

Crea un grafo que calcula un precio: nodo 1 establece el precio base ($100), nodo 2 aplica un descuento (20%), nodo 3 calcula el total. Ejecuta normalmente. Luego haz fork desde después del nodo 1, cambia el precio base a $200, y compara el resultado del fork con el original.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    base_price: float
    discount: float
    final_price: float

def set_price(state: State) -> dict:
    return {"base_price": 100.0}

def apply_discount(state: State) -> dict:
    discount = 0.20
    return {"discount": discount}

def calculate_total(state: State) -> dict:
    total = state["base_price"] * (1 - state["discount"])
    return {"final_price": total}

graph_builder = StateGraph(State)
graph_builder.add_node("set_price", set_price)
graph_builder.add_node("discount", apply_discount)
graph_builder.add_node("total", calculate_total)

graph_builder.add_edge(START, "set_price")
graph_builder.add_edge("set_price", "discount")
graph_builder.add_edge("discount", "total")
graph_builder.add_edge("total", END)

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

config_original = {"configurable": {"thread_id": "price-original"}}
original = graph.invoke(
    {"base_price": 0, "discount": 0, "final_price": 0},
    config_original,
)
print(f"Original: base=${original['base_price']:.0f}, "
      f"descuento={original['discount']:.0%}, "
      f"final=${original['final_price']:.2f}")

price_checkpoint = None
for cp in graph.get_state_history(config_original):
    if cp.metadata.get("source") == "set_price":
        price_checkpoint = cp
        break

config_fork = {
    "configurable": {
        "thread_id": "price-fork",
        "checkpoint_id": price_checkpoint.config["configurable"]["checkpoint_id"],
    }
}
graph.update_state(config_fork, {"base_price": 200.0})

fork_result = graph.invoke(None, config_fork)
print(f"Fork:     base=${fork_result['base_price']:.0f}, "
      f"descuento={fork_result['discount']:.0%}, "
      f"final=${fork_result['final_price']:.2f}")

print(f"\nDiferencia: ${fork_result['final_price'] - original['final_price']:.2f}")
# Output esperado:
# Original: base=$100, descuento=20%, final=$80.00
# Fork:     base=$200, descuento=20%, final=$160.00
#
# Diferencia: $80.00

Ejercicio 5: Debugging workflow completo (Medio)

Crea un grafo de investigación con un bug intencional: el nodo de filtrado usa un threshold incorrecto (0.01 en vez de 0.5), dejando pasar fuentes irrelevantes. Ejecuta el grafo, luego usa time-travel debugging para: (1) encontrar el checkpoint después del filtrado, (2) inspeccionar qué fuentes pasaron, (3) identificar el bug. Imprime un reporte de debugging.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    raw_sources: Annotated[list[dict], operator.add]
    filtered_sources: list[dict]
    report: str

def gather(state: State) -> dict:
    return {"raw_sources": [
        {"name": "arXiv", "data": f"Papers sobre {state['topic']}", "score": 0.95},
        {"name": "Wikipedia", "data": f"Artículo sobre {state['topic']}", "score": 0.80},
        {"name": "Random Blog", "data": "Receta de cocina", "score": 0.02},
        {"name": "Spam Site", "data": "Compra ahora!!!", "score": 0.01},
    ]}

def filter_sources(state: State) -> dict:
    threshold = 0.01
    filtered = [s for s in state["raw_sources"] if s["score"] >= threshold]
    return {"filtered_sources": filtered}

def generate_report(state: State) -> dict:
    names = [s["name"] for s in state["filtered_sources"]]
    return {"report": f"Reporte sobre '{state['topic']}' usando {len(names)} fuentes: {', '.join(names)}"}

graph_builder = StateGraph(State)
graph_builder.add_node("gather", gather)
graph_builder.add_node("filter", filter_sources)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "gather")
graph_builder.add_edge("gather", "filter")
graph_builder.add_edge("filter", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "buggy-research"}}
result = graph.invoke(
    {"topic": "AI agents", "raw_sources": [], "filtered_sources": [], "report": ""},
    config,
)

print("=== REPORTE DE DEBUGGING ===\n")
print(f"Output final: {result['report']}")
print(f"\n⚠️  El reporte incluye 'Random Blog' y 'Spam Site' — esto es un bug.\n")

print("--- Análisis del historial ---\n")
for cp in graph.get_state_history(config):
    node = cp.metadata.get("source", "?")

    if node == "gather":
        print(f"[gather] Fuentes raw: {len(cp.values['raw_sources'])}")
        for s in cp.values["raw_sources"]:
            print(f"  {s['name']}: score={s['score']}")

    elif node == "filter":
        print(f"\n[filter] Fuentes filtradas: {len(cp.values['filtered_sources'])}")
        for s in cp.values["filtered_sources"]:
            flag = "⚠️ SOSPECHOSO" if s["score"] < 0.5 else "✅"
            print(f"  {flag} {s['name']}: score={s['score']}")

        passed_low = [s for s in cp.values["filtered_sources"] if s["score"] < 0.5]
        if passed_low:
            print(f"\n🐛 BUG ENCONTRADO: {len(passed_low)} fuentes con score < 0.5 pasaron el filtro.")
            print(f"   Causa probable: threshold demasiado bajo (debería ser 0.5, no 0.01)")
# Output esperado:
# === REPORTE DE DEBUGGING ===
#
# Output final: Reporte sobre 'AI agents' usando 4 fuentes: arXiv, Wikipedia, Random Blog, Spam Site
#
# ⚠️  El reporte incluye 'Random Blog' y 'Spam Site' — esto es un bug.
#
# --- Análisis del historial ---
#
# [gather] Fuentes raw: 4
#   arXiv: score=0.95
#   Wikipedia: score=0.8
#   Random Blog: score=0.02
#   Spam Site: score=0.01
#
# [filter] Fuentes filtradas: 4
#   ✅ arXiv: score=0.95
#   ✅ Wikipedia: score=0.8
#   ⚠️ SOSPECHOSO Random Blog: score=0.02
#   ⚠️ SOSPECHOSO Spam Site: score=0.01
#
# 🐛 BUG ENCONTRADO: 2 fuentes con score < 0.5 pasaron el filtro.
#    Causa probable: threshold demasiado bajo (debería ser 0.5, no 0.01)

Ejercicio 6: Comparación A/B con forks (Avanzado)

Crea un grafo que procesa texto en 3 pasos: limpieza → análisis → resumen. Ejecuta con un texto original. Luego crea dos forks desde el checkpoint después de "limpieza": uno con el texto original y otro con un texto enriquecido (agregando más contexto). Compara los resúmenes de las tres ejecuciones (original, fork A, fork B) e imprime una tabla comparativa.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    raw_text: str
    clean_text: str
    analysis: str
    summary: str
    variant: str

def clean(state: State) -> dict:
    cleaned = state["raw_text"].strip().replace("  ", " ")
    return {"clean_text": cleaned}

def analyze(state: State) -> dict:
    word_count = len(state["clean_text"].split())
    return {"analysis": f"{word_count} palabras, tema: AI"}

def summarize(state: State) -> dict:
    text_preview = state["clean_text"][:50]
    return {"summary": f"Resumen ({state['analysis']}): {text_preview}..."}

graph_builder = StateGraph(State)
graph_builder.add_node("clean", clean)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "clean")
graph_builder.add_edge("clean", "analyze")
graph_builder.add_edge("analyze", "summarize")
graph_builder.add_edge("summarize", END)

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

config_original = {"configurable": {"thread_id": "ab-original"}}
original = graph.invoke(
    {
        "raw_text": "  AI agents son programas que  usan LLMs para tomar decisiones  ",
        "clean_text": "", "analysis": "", "summary": "", "variant": "original",
    },
    config_original,
)

clean_checkpoint = None
for cp in graph.get_state_history(config_original):
    if cp.metadata.get("source") == "clean":
        clean_checkpoint = cp
        break

checkpoint_id = clean_checkpoint.config["configurable"]["checkpoint_id"]

config_a = {"configurable": {"thread_id": "ab-fork-a", "checkpoint_id": checkpoint_id}}
graph.update_state(config_a, {
    "clean_text": "AI agents son programas que usan LLMs para tomar decisiones",
    "variant": "fork-a-sin-cambios",
})
fork_a = graph.invoke(None, config_a)

config_b = {"configurable": {"thread_id": "ab-fork-b", "checkpoint_id": checkpoint_id}}
graph.update_state(config_b, {
    "clean_text": "AI agents son programas autónomos que usan LLMs para razonar, planificar y tomar decisiones complejas en entornos dinámicos",
    "variant": "fork-b-enriquecido",
})
fork_b = graph.invoke(None, config_b)

print("=== COMPARACIÓN A/B ===\n")
print(f"{'Variante':<25} {'Análisis':<30} {'Resumen'}")
print(f"{'-'*25} {'-'*30} {'-'*50}")
print(f"{'Original':<25} {original['analysis']:<30} {original['summary'][:50]}")
print(f"{'Fork A (sin cambios)':<25} {fork_a['analysis']:<30} {fork_a['summary'][:50]}")
print(f"{'Fork B (enriquecido)':<25} {fork_b['analysis']:<30} {fork_b['summary'][:50]}")
# Output esperado:
# === COMPARACIÓN A/B ===
#
# Variante                  Análisis                       Resumen
# ------------------------- ------------------------------ --------------------------------------------------
# Original                  9 palabras, tema: AI           Resumen (9 palabras, tema: AI): AI agents son prog
# Fork A (sin cambios)      9 palabras, tema: AI           Resumen (9 palabras, tema: AI): AI agents son prog
# Fork B (enriquecido)      16 palabras, tema: AI          Resumen (16 palabras, tema: AI): AI agents son pro

Resumen

En esta cápsula aprendiste:

  • Time-travel debugging es la herramienta principal para diagnosticar agentes — no es un feature de demo. Te permite navegar el historial completo de estados sin re-ejecutar nada, sin gastar dinero en API calls, y con la garantía de ver el estado exacto que tenía el agente
  • get_state_history(config) retorna todos los checkpoints en orden inverso. Cada checkpoint contiene: el estado completo (values), qué nodo lo generó (metadata.source), qué escribió (metadata.writes), y qué nodo seguía (next)
  • Replay te permite re-ejecutar desde cualquier checkpoint con graph.invoke(None, config_con_checkpoint_id). Útil para verificar que un fix funciona sin re-ejecutar todo el pipeline
  • Forking crea una rama alternativa: modificas el estado en un punto histórico con update_state() y ejecutas desde ahí. La ejecución original no se modifica
  • El workflow de debugging es sistemático: output inesperado → get_state_history → inspeccionar cada paso → encontrar dónde apareció el error → entender la causa raíz → fix → replay para verificar
  • metadata.writes muestra el "diff" de cada paso — exactamente qué cambió el nodo en el estado. Es la herramienta más precisa para localizar bugs
  • La función debug_execution() es un patrón reutilizable que puedes aplicar a cualquier grafo para ver el historial completo de manera legible

Próxima cápsula: Long-term Memory — cómo hacer que tu agente recuerde información entre sesiones diferentes, almacenando preferencias, contexto acumulado y conocimiento del usuario.

Recursos adicionales

  1. LangGraph Time Travel — Conceptos oficiales de time-travel en LangGraph
  2. How to view and update past graph state — Guía práctica de time-travel debugging
  3. LangGraph State History — Cómo navegar el historial de checkpoints
  4. LangGraph Checkpointer Concepts — Fundamentos de persistencia y checkpointing
  5. Replay and Fork — Replay y forking desde checkpoints históricos

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

← AnteriorDurable ExecutionSiguiente →Long-term Memory: Store y Cross-Session