Module 8: Memory and Persistence

Time-Travel Debugging

Next →

Capsule overview

Time-travel debugging lets you navigate your agent's complete execution history — every state at every step. You can go back to any point and understand exactly what the agent knew and why it made a specific decision. You can replay from any step, fork the execution to explore alternative paths, and diagnose problems without re-running anything.

This is not a curiosity or a demo feature. It's the most powerful debugging tool that exists for agents. In traditional programming, when something breaks, you add print statements, re-run, and try to guess what happened. With agents, that strategy doesn't work — a single run can take minutes, cost money in API calls, and depend on non-deterministic LLM responses. Re-running doesn't guarantee you'll see the same bug.

With time-travel debugging: you navigate straight to the step where things went wrong, inspect the full state, understand the root cause, fix it, and replay to verify. Without spending extra time or money.

The scenario that makes it indispensable

Your Research Assistant produced a report on "AI Safety" and the summary section says something odd — it mixes data from two different topics. What do you do?

Without time-travel debugging:

1. You look at the final output → "the summary is wrong"
2. Where did it break? No idea → you add print statements
3. Re-run → $0.50 in API calls, 3 minutes of waiting
4. The LLM answers differently this time → you can't reproduce the bug
5. You repeat 3-4 times → $2.00 and 12 minutes gone
6. You still don't understand what happened

With time-travel debugging:

1. You look at the final output → "the summary is wrong"
2. get_state_history() → you see the 8 steps of the run
3. You navigate to step 6 (synthesis) → you inspect which sources it had
4. You discover: the search node returned data from the wrong topic at step 3
5. You navigate to step 3 → you see the exact query that was sent and the response
6. Root cause: the query was built wrong. Fixed in 2 minutes, $0.00 extra

The difference isn't just efficiency — it's that you can diagnose bugs that are impossible to reproduce because they depend on the specific state the agent had at that moment.

Checkpoint history: your agent's timeline

Every time a node completes, LangGraph saves a checkpoint. The checkpoint history is an ordered list of every state the agent passed through:

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 is a research field..."],
        "step_count": state.get("step_count", 0) + 1,
    }

def analyze(state: State) -> dict:
    return {
        "sources": ["Analysis: 3 main trends identified..."],
        "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"Synthesize briefly (2 sentences):\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="Research AI safety")], "sources": [], "step_count": 0},
    config,
)

print("=== Checkpoint history ===\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"  Node: {node}")
    print(f"  Step: {step}")
    print(f"  Sources: {len(state.values.get('sources', []))}")
    print(f"  Messages: {len(state.values.get('messages', []))}")
    print(f"  Next: {state.next}")
    print()
# Expected output:
# === Checkpoint history ===
#
# Checkpoint 0:
#   ID: 1ef8a1b2c3d4e5f6...
#   Node: synthesize
#   Step: 3
#   Sources: 2
#   Messages: 2
#   Next: ()
#
# Checkpoint 1:
#   ID: 1ef8a1b2c3d4e5f5...
#   Node: analyze
#   Step: 2
#   Sources: 2
#   Messages: 1
#   Next: ('synthesize',)
#
# Checkpoint 2:
#   ID: 1ef8a1b2c3d4e5f4...
#   Node: search
#   Step: 1
#   Sources: 1
#   Messages: 1
#   Next: ('analyze',)
#
# Checkpoint 3:
#   ID: 1ef8a1b2c3d4e5f3...
#   Node: __start__
#   Step: 0
#   Sources: 0
#   Messages: 1
#   Next: ('search',)

get_state_history returns checkpoints in reverse order (most recent first). Each checkpoint contains:

  • ✅ values: the graph's complete state at that point
  • ✅ metadata: which node generated this checkpoint, the step number, timestamp
  • ✅ config: includes the unique checkpoint_id so you can navigate to this point
  • ✅ next: which node was going to run next (an empty tuple if the run finished)

Inspecting a specific checkpoint

When you find a suspicious checkpoint in the history, you can inspect its full state:

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"Source 1: Data on '{state['query']}' from Wikipedia",
        f"Source 2: Paper on '{state['query']}' from arXiv",
        f"Source 3: Article on '{state['query']}' from TechCrunch",
    ]
    return {"search_results": results}

def analyze(state: State) -> dict:
    analysis = f"Analysis of {len(state['search_results'])} sources: "
    analysis += "convergence on 3 main trends."
    return {"analysis": analysis}

def report(state: State) -> dict:
    report_text = f"Report: {state['analysis']} "
    report_text += f"Based on {len(state['search_results'])} sources."
    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="Research quantum computing")],
        "query": "quantum computing",
        "search_results": [],
        "analysis": "",
        "final_report": "",
    },
    config,
)

history = list(graph.get_state_history(config))

print("=== Inspecting the checkpoint after 'search' ===\n")
for checkpoint in history:
    if checkpoint.metadata.get("source") == "search":
        print(f"Checkpoint ID: {checkpoint.config['configurable']['checkpoint_id'][:20]}...")
        print(f"Node that generated it: {checkpoint.metadata.get('source')}")
        print(f"Next node: {checkpoint.next}")
        print(f"\nFull state at this point:")
        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
# Expected output:
# === Inspecting the checkpoint after 'search' ===
#
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...
# Node that generated it: search
# Next node: ('analyze',)
#
# Full state at this point:
#   query: quantum computing
#   search_results (3):
#     - Source 1: Data on 'quantum computing' from Wikipedia
#     - Source 2: Paper on 'quantum computing' from arXiv
#     - Source 3: Article on 'quantum computing' from TechCrunch
#   analysis: ''
#   final_report: ''

Notice that after search:

  • search_results has 3 sources (the search node produced them)
  • analysis is empty (the analyze node hasn't run yet)
  • final_report is empty (the report node hasn't run yet)

You can see exactly what the agent knew at every point of the run. If the final report has an error, you walk backwards checkpoint by checkpoint until you find where the incorrect data appeared.

Replay from a checkpoint: "what would have happened if...?"

Replay is the ability to re-run the graph from a historical checkpoint. This has two main uses:

  1. Verify a fix: you changed a node's logic and you want to see whether the same input produces a correct result
  2. Explore alternatives: "what would have happened if the agent had this data instead of that?"
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 processed: '{state['input']}'",
        "path_taken": ["step_a"],
    }

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

def step_final(state: State) -> dict:
    return {
        "final": f"Result: {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": "original data", "step_a_result": "", "step_b_result": "", "final": "", "path_taken": []},
    config,
)
print(f"Original run: {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 from the checkpoint after step_a ---")
    print(f"Checkpoint ID: {checkpoint_id[:20]}...")
    print(f"State at this point: 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 result: {replay_result['final']}")
    print(f"Replay path: {replay_result['path_taken']}")
# Expected output:
# Original run: Result: B analyzed: 'A processed: 'original data''
# Path: ['step_a', 'step_b', 'final']
#
# --- Replay from the checkpoint after step_a ---
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...
# State at this point: step_a_result = 'A processed: 'original data''
#
# Replay result: Result: B analyzed: 'A processed: 'original data''
# Replay path: ['step_a', 'step_b', 'final']

How replay works:

  1. You get the checkpoint_id of the point you want to go back to
  2. You invoke the graph with None as the input (the state comes from the checkpoint) and the checkpoint_id in the config
  3. The graph runs forward from that point

You pass None because you're not providing new input — the complete state already exists in the checkpoint.

Forking: branching from a historical point

Forking goes one step beyond replay. Instead of re-running with the same state, you modify the state at a historical point and run from there. This creates an alternative branch of the execution:

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"Basic results on '{state['query']}'",
        "path": ["search"],
    }

def analyze(state: State) -> dict:
    return {
        "analysis": f"Analysis of: {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": "PREMIUM results with 50 papers and exclusive data on 'LangGraph'"},
    )

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

    original_state = graph.get_state(config)
    print(f"\nOriginal unchanged: {original_state.values['analysis']}")
# Expected output:
# Original: Analysis of: Basic results on 'LangGraph'
# Fork:     Analysis of: PREMIUM results with 50 papers and exclusive data on 'LangGraph'
#
# Original unchanged: Analysis of: Basic results on 'LangGraph'

Forking lets you answer questions like:

  • "What would have happened if the search had returned different data?"
  • "What happens if I change the analysis node's prompt?"
  • "What result do I get if I give the agent more context at step 3?"

And most importantly: the original run is not modified. The fork creates an independent branch with a different thread_id.

A practical debugging workflow

Here's the complete workflow you'll use every time your agent produces an unexpected result:

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"[DATA] Information about '{state['query']}': AI agents use LLMs to reason."}

def process_data(state: State) -> dict:
    processed = state["raw_data"].replace("[DATA]", "[PROCESSED]")
    processed += " NOTE: data verified."
    return {"processed_data": processed}

def summarize(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Summarize in one sentence: {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="Research AI agents")],
        "query": "AI agents",
        "raw_data": "",
        "processed_data": "",
        "summary": "",
    },
    config,
)

print("STEP 1: The output isn't what we expected")
print(f"  Summary: {result['summary']}\n")

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

print("STEP 3: Inspect each step")
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("\nSTEP 4: Find the suspicious checkpoint")
for cp in history:
    if cp.metadata.get("source") == "process":
        print(f"  After 'process':")
        print(f"    processed_data = '{cp.values['processed_data']}'")
        print(f"    Was the data processed correctly? → Inspect the node's logic")
        break

print("\nSTEP 5: Verify the fix with replay (after correcting the node)")
print("  You would use: graph.invoke(None, config_with_checkpoint_id)")
# Expected output:
# STEP 1: The output isn't what we expected
#   Summary: AI agents use LLMs to reason and make decisions...
#
# STEP 2: Get the full history
#   Total checkpoints: 4
#
# STEP 3: Inspect each step
#   [__start__]
#   [fetch]
#     raw_data: [DATA] Information about 'AI agents': AI agents use LL...
#   [process]
#     raw_data: [DATA] Information about 'AI agents': AI agents use LL...
#     processed_data: [PROCESSED] Information about 'AI agents': AI agents ...
#   [summarize]
#     raw_data: [DATA] Information about 'AI agents': AI agents use LL...
#     processed_data: [PROCESSED] Information about 'AI agents': AI agents ...
#     summary: AI agents use large language models (LLMs) for reasonin...
#
# STEP 4: Find the suspicious checkpoint
#   After 'process':
#     processed_data = '[PROCESSED] Information about 'AI agents'...'
#     Was the data processed correctly? → Inspect the node's logic
#
# STEP 5: Verify the fix with replay (after correcting the node)
#   You would use: graph.invoke(None, config_with_checkpoint_id)

The workflow in summary

1. Agent produces an unexpected output
     ↓
2. get_state_history(config) → list of every checkpoint
     ↓
3. Iterate the checkpoints: inspect the state at each step
     ↓
4. Find the step where the incorrect data appeared
     ↓
5. Compare the state BEFORE and AFTER the suspicious node
     ↓
6. Root cause identified → fix the node's logic
     ↓
7. Replay from the checkpoint before the error → verify the fix

Comparison: traditional debugging vs time-travel

To make it clear why this is transformative:

AspectTraditional debuggingTime-travel debugging
Seeing intermediate stateprint() + re-runNavigate to the checkpoint
Cost of debuggingRe-run = API calls + time$0.00 (data already saved)
ReproducibilityNot guaranteed (LLMs are non-deterministic)Exact state preserved
"What happened at step 5?"Add logging, re-run, waitget_state_history() → inspect
"What if I change X?"Modify code, re-run everythingFork + update_state + replay
Debugging in productionLogs, metrics, guessworkNavigate the exact history
Time to diagnosisMinutes to hoursSeconds to minutes

Checkpoint metadata: extra context

Every checkpoint includes metadata that helps you understand the context of the run:

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": "result from A", "log": ["A executed"]}

def node_b(state: State) -> dict:
    return {"data": "result from B", "log": ["B executed"]}

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 of each checkpoint ===\n")
for cp in graph.get_state_history(config):
    print(f"Node: {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()
# Expected output:
# === Metadata of each checkpoint ===
#
# Node: b
#   step: 2
#   writes: {'b': {'data': 'result from B', 'log': ['B executed']}}
#   checkpoint_id: 1ef8a1b2c3d4e5f6ab...
#   parent_checkpoint_id: {'configurable': {'thread_id': 'metadata-demo', 'checkpoint_id': '...'}}
#
# Node: a
#   step: 1
#   writes: {'a': {'data': 'result from A', 'log': ['A executed']}}
#   checkpoint_id: 1ef8a1b2c3d4e5f5ab...
#   parent_checkpoint_id: {'configurable': {'thread_id': 'metadata-demo', 'checkpoint_id': '...'}}
#
# Node: __start__
#   step: 0
#   writes: {'__start__': {'data': '', 'log': []}}
#   checkpoint_id: 1ef8a1b2c3d4e5f4ab...
#   parent_checkpoint_id: None

The most useful fields for debugging:

  • ✅ source: which node generated this checkpoint
  • ✅ step: the sequential step number
  • ✅ writes: exactly what the node wrote into the state (the "diff")
  • ✅ parent_config: the checkpoint of the previous step (for navigating backwards)

The writes field is especially powerful — it shows you not just the complete state, but what changed in this specific step.

Debugging with real agents: a complete example

Let's look at a more realistic case where time-travel debugging solves a concrete problem:

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"Encyclopedic info on {state['topic']}", "relevance": 0.9},
        {"name": "arXiv", "data": f"Papers on {state['topic']}", "relevance": 0.8},
        {"name": "Reddit", "data": "Cat memes", "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"Analysis based on {source_names}: {source_data}"}

def generate_report(state: ResearchState) -> dict:
    return {"report": f"REPORT: {state['analysis']}. Total sources: {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("=== The final report looks fine, but was the filtering correct? ===")
print(f"Report: {result['report']}\n")

print("=== Debugging: check what happened at each step ===\n")

history = list(graph.get_state_history(config))

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

    if node == "gather":
        print(f"[gather] Sources collected: {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] Sources after filtering: {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']}) — FILTERED OUT")

    elif node == "analyze":
        print(f"\n[analyze] Analysis generated:")
        print(f"  {cp.values['analysis'][:80]}...")

    elif node == "report":
        print(f"\n[report] Final report:")
        print(f"  {cp.values['report'][:80]}...")
# Expected output:
# === The final report looks fine, but was the filtering correct? ===
# Report: REPORT: Analysis based on ['Wikipedia', 'arXiv']: Encyclopedic info on transformer architectures | Papers on transformer architectures. Total sources: 2.
#
# === Debugging: check what happened at each step ===
#
# [gather] Sources collected: 3
#   - Wikipedia (relevance: 0.9): Encyclopedic info on transformer archite...
#   - arXiv (relevance: 0.8): Papers on transformer architectures...
#   - Reddit (relevance: 0.1): Cat memes...
#
# [filter] Sources after filtering: 2
#   ✅ Wikipedia (relevance: 0.9)
#   ✅ arXiv (relevance: 0.8)
#   ❌ Reddit (relevance: 0.1) — FILTERED OUT
#
# [analyze] Analysis generated:
#   Analysis based on ['Wikipedia', 'arXiv']: Encyclopedic info on transform...
#
# [report] Final report:
#   REPORT: Analysis based on ['Wikipedia', 'arXiv']: Encyclopedic info on tr...

With this debugging you can verify:

  • ✅ Were the right sources collected? → Yes, 3 sources
  • ✅ Was the filtering correct? → Yes, Reddit (relevance 0.1) was filtered out
  • ✅ Did the analysis use the right sources? → Yes, Wikipedia and arXiv
  • ✅ Does the report reflect the analysis? → Yes

If something were wrong (say, Reddit passed the filter), you'd know exactly at which step the error happened.

Advanced forking: comparing alternative decisions

A powerful use case: your agent filtered sources with a threshold of 0.5. What happens if you change the threshold to 0.3? Instead of re-running everything, you fork from just before the filter:

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"Summary based on {len(names)} sources: {', '.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 unchanged:      {original_check.values['summary']}")
# Expected output:
# Original (threshold=0.5): Summary based on 2 sources: Source A, Source C
# Fork (threshold=0.3):    Summary based on 3 sources: Source A, Source B, Source C
#
# Original unchanged:      Summary based on 2 sources: Source A, Source C

With threshold 0.5, 2 sources made it in. With threshold 0.3, 3 did (Source B now passes). You can compare both results without having re-run the gather step (which could have cost API calls).

Building a reusable debugging function

To make debugging practical in your day-to-day, wrap the logic in a function:

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):
    """Shows the complete history of a run for 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} | Node: {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"  next: {cp.next}")
        else:
            print(f"  [RUN COMPLETE]")
        print()

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

def node_b(state: State) -> dict:
    return {"step_b": f"B analyzed: {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": "test data", "step_a": "", "step_b": "", "output": "", "log": []}, config)

debug_execution(graph, config, fields_to_show=["step_a", "step_b", "output"])
# Expected output:
# ============================================================
# Thread: debug-util
# Total checkpoints: 4
# ============================================================
#
# Step 0 | Node: __start__ | ID: 1ef8a1b2c3d4...
#   next: ('a',)
#
# Step 1 | Node: a | ID: 1ef8a1b2c3d5...
#   → step_a = A processed: test data
#   next: ('b',)
#
# Step 2 | Node: b | ID: 1ef8a1b2c3d6...
#   → step_b = B analyzed: A processed: test data
#   next: ('output',)
#
# Step 3 | Node: output | ID: 1ef8a1b2c3d7...
#   → output = Final: B analyzed: A processed: test data
#   [RUN COMPLETE]

You can use debug_execution() on any graph. All you need is the compiled graph and the config. The fields_to_show parameter lets you filter which state fields you want to see (handy when the state has many fields).

Troubleshooting

Problem 1: "get_state_history returns an empty list"

Symptom: You call graph.get_state_history(config) and get no checkpoints.

Cause: The graph has no checkpointer, or the thread_id doesn't match.

Solution: Check that you compiled with a checkpointer and that the thread_id is correct:

# ❌ No checkpointer — there's no history
graph = graph_builder.compile()

# ✅ With a checkpointer
graph = graph_builder.compile(checkpointer=MemorySaver())

# Verify the thread_id
config = {"configurable": {"thread_id": "my-thread"}}  # same ID as the run

Problem 2: "Replay produces a different result than the original"

Symptom: You replay from a checkpoint and the result is different.

Cause: The nodes contain non-deterministic logic (LLM calls, timestamps, random) that produces different results every time.

Solution: This is expected with LLMs. Replay re-runs the nodes from the checkpoint, and an LLM can answer differently. For debugging, what matters is that the input state is the same — you can compare the input and understand why the output changed.

Problem 3: "I can't find the checkpoint where the error happened"

Symptom: You have many checkpoints and don't know which one to inspect.

Cause: You're not using the metadata to filter.

Solution: Use metadata['source'] to filter by node:

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

Problem 4: "update_state has no effect on the fork"

Symptom: You call graph.update_state() but the fork runs with the original state.

Cause: The checkpoint_id or thread_id don't match, or the update was made on the wrong config.

Solution: Check that the fork's config has the right checkpoint_id:

fork_config = {
    "configurable": {
        "thread_id": "new-thread-for-fork",
        "checkpoint_id": checkpoint_id_from_history,
    }
}
graph.update_state(fork_config, {"field": "new_value"})
result = graph.invoke(None, fork_config)

Exercises

Exercise 1: Explore the checkpoint history (Easy)

Create a graph with 3 sequential nodes. Run one invocation and then use get_state_history() to print: the node name, the step number, and the next of each checkpoint. Verify that the history has 4 checkpoints (start + 3 nodes).

See solution 
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": "result_1", "log": ["n1"]}

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

def node_3(state: State) -> dict:
    return {"data": "result_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} | Node: {node} | Next: {cp.next}")
# Expected output:
# Total checkpoints: 4
#
# Step 0 | Node: __start__ | Next: ('n1',)
# Step 1 | Node: n1 | Next: ('n2',)
# Step 2 | Node: n2 | Next: ('n3',)
# Step 3 | Node: n3 | Next: ()

Exercise 2: Inspect the writes of each checkpoint (Easy)

Using the same graph from exercise 1, print the writes of each checkpoint (what each node wrote into the state). This shows you the "diff" of each step.

See solution 
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()
# Expected output:
# [__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']

Exercise 3: Replay from a specific checkpoint (Medium)

Create a graph with 4 nodes. Run it end to end. Then get the checkpoint after node 2 and replay from there. Verify that the replay only runs nodes 3 and 4 (not 1 and 2).

See solution 
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"Original run: {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 (from n2): {EXECUTION_LOG}")
print(f"Steps in the result: {replay_result['steps']}")
# Expected output:
# Original run: ['n1', 'n2', 'n3', 'n4']
# Replay (from n2): ['n3', 'n4']
# Steps in the result: ['n1', 'n2', 'n3', 'n4']

Exercise 4: Fork with a modified state (Medium)

Create a graph that computes a price: node 1 sets the base price ($100), node 2 applies a discount (20%), node 3 computes the total. Run it normally. Then fork from after node 1, change the base price to $200, and compare the fork's result with the original.

See solution 
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"discount={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"discount={fork_result['discount']:.0%}, "
      f"final=${fork_result['final_price']:.2f}")

print(f"\nDifference: ${fork_result['final_price'] - original['final_price']:.2f}")
# Expected output:
# Original: base=$100, discount=20%, final=$80.00
# Fork:     base=$200, discount=20%, final=$160.00
#
# Difference: $80.00

Exercise 5: A complete debugging workflow (Medium)

Create a research graph with an intentional bug: the filtering node uses the wrong threshold (0.01 instead of 0.5), letting irrelevant sources through. Run the graph, then use time-travel debugging to: (1) find the checkpoint after the filtering, (2) inspect which sources passed, (3) identify the bug. Print a debugging report.

See solution 
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 on {state['topic']}", "score": 0.95},
        {"name": "Wikipedia", "data": f"Article on {state['topic']}", "score": 0.80},
        {"name": "Random Blog", "data": "Cooking recipe", "score": 0.02},
        {"name": "Spam Site", "data": "Buy now!!!", "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"Report on '{state['topic']}' using {len(names)} sources: {', '.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("=== DEBUGGING REPORT ===\n")
print(f"Final output: {result['report']}")
print(f"\n⚠️  The report includes 'Random Blog' and 'Spam Site' — that's a bug.\n")

print("--- History analysis ---\n")
for cp in graph.get_state_history(config):
    node = cp.metadata.get("source", "?")

    if node == "gather":
        print(f"[gather] Raw sources: {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] Filtered sources: {len(cp.values['filtered_sources'])}")
        for s in cp.values["filtered_sources"]:
            flag = "⚠️ SUSPICIOUS" 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 FOUND: {len(passed_low)} sources with score < 0.5 passed the filter.")
            print(f"   Likely cause: threshold too low (should be 0.5, not 0.01)")
# Expected output:
# === DEBUGGING REPORT ===
#
# Final output: Report on 'AI agents' using 4 sources: arXiv, Wikipedia, Random Blog, Spam Site
#
# ⚠️  The report includes 'Random Blog' and 'Spam Site' — that's a bug.
#
# --- History analysis ---
#
# [gather] Raw sources: 4
#   arXiv: score=0.95
#   Wikipedia: score=0.8
#   Random Blog: score=0.02
#   Spam Site: score=0.01
#
# [filter] Filtered sources: 4
#   ✅ arXiv: score=0.95
#   ✅ Wikipedia: score=0.8
#   ⚠️ SUSPICIOUS Random Blog: score=0.02
#   ⚠️ SUSPICIOUS Spam Site: score=0.01
#
# 🐛 BUG FOUND: 2 sources with score < 0.5 passed the filter.
#    Likely cause: threshold too low (should be 0.5, not 0.01)

Exercise 6: A/B comparison with forks (Advanced)

Create a graph that processes text in 3 steps: cleanup → analysis → summary. Run it with an original text. Then create two forks from the checkpoint after "cleanup": one with the original text and another with an enriched text (adding more context). Compare the summaries of the three runs (original, fork A, fork B) and print a comparison table.

See solution 
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} words, topic: AI"}

def summarize(state: State) -> dict:
    text_preview = state["clean_text"][:50]
    return {"summary": f"Summary ({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 are programs that  use LLMs to make decisions  ",
        "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 are programs that use LLMs to make decisions",
    "variant": "fork-a-unchanged",
})
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 are autonomous programs that use LLMs to reason, plan and make complex decisions in dynamic environments",
    "variant": "fork-b-enriched",
})
fork_b = graph.invoke(None, config_b)

print("=== A/B COMPARISON ===\n")
print(f"{'Variant':<25} {'Analysis':<30} {'Summary'}")
print(f"{'-'*25} {'-'*30} {'-'*50}")
print(f"{'Original':<25} {original['analysis']:<30} {original['summary'][:50]}")
print(f"{'Fork A (unchanged)':<25} {fork_a['analysis']:<30} {fork_a['summary'][:50]}")
print(f"{'Fork B (enriched)':<25} {fork_b['analysis']:<30} {fork_b['summary'][:50]}")
# Expected output:
# === A/B COMPARISON ===
#
# Variant                   Analysis                       Summary
# ------------------------- ------------------------------ --------------------------------------------------
# Original                  10 words, topic: AI            Summary (10 words, topic: AI): AI agents are progr
# Fork A (unchanged)        10 words, topic: AI            Summary (10 words, topic: AI): AI agents are progr
# Fork B (enriched)         18 words, topic: AI            Summary (18 words, topic: AI): AI agents are auton

Summary

In this capsule you learned:

  • Time-travel debugging is the primary tool for diagnosing agents — it's not a demo feature. It lets you navigate the complete state history without re-running anything, without spending money on API calls, and with the guarantee of seeing the exact state the agent had
  • get_state_history(config) returns every checkpoint in reverse order. Each checkpoint contains: the full state (values), which node generated it (metadata.source), what it wrote (metadata.writes), and which node came next (next)
  • Replay lets you re-run from any checkpoint with graph.invoke(None, config_with_checkpoint_id). Useful for verifying that a fix works without re-running the whole pipeline
  • Forking creates an alternative branch: you modify the state at a historical point with update_state() and run from there. The original run is not modified
  • The debugging workflow is systematic: unexpected output → get_state_history → inspect each step → find where the error appeared → understand the root cause → fix → replay to verify
  • metadata.writes shows the "diff" of each step — exactly what the node changed in the state. It's the most precise tool for pinpointing bugs
  • The debug_execution() function is a reusable pattern you can apply to any graph to see the complete history in a readable form

Next capsule: Long-term Memory — how to make your agent remember information across different sessions, storing preferences, accumulated context and knowledge about the user.

Additional resources

  1. LangGraph Time Travel — Official time-travel concepts in LangGraph
  2. How to view and update past graph state — Practical guide to time-travel debugging
  3. LangGraph State History — How to navigate the checkpoint history
  4. LangGraph Checkpointer Concepts — Fundamentals of persistence and checkpointing
  5. Replay and Fork — Replaying and forking from historical checkpoints

Module 8 — LangChain & LangGraph: From Chains to Agents

← PreviousDurable ExecutionNext →Long-term Memory: Store and Cross-Session