Skip to Content
advanced-topicsTime Travel

Last Updated: 3/11/2026

Time Travel

Replay past executions and fork to explore alternative paths.

What Is Time Travel?

Time travel lets you:

  • Replay: Re-run from a prior checkpoint
  • Fork: Branch from a checkpoint with modified state

Both work by resuming from saved checkpoints. Nodes before the checkpoint use cached results. Nodes after re-execute.

<Warning> Replay re-executes nodes—LLM calls, API requests, and interrupts fire again and may return different results. </Warning>

Replay

Retry from a prior checkpoint:

Python:

from langgraph.checkpoint.memory import MemorySaver
 
app = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "1"}}
 
# Initial run
result = app.invoke(inputs, config)
 
# Find checkpoint to replay from
history = list(app.get_state_history(config))
before_node_b = next(s for s in history if s.next == ("node_b",))
 
# Replay from that checkpoint
replay_result = app.invoke(None, before_node_b.config)
# node_b re-executes, node_a does not

JavaScript:

const app = builder.compile({ checkpointer: new MemorySaver() });
const config = { configurable: { thread_id: "1" } };
 
await app.invoke(inputs, config);
 
const states = [];
for await (const state of app.getStateHistory(config)) {
  states.push(state);
}
 
const beforeNodeB = states.find(s => s.next.includes("nodeB"));
const replayResult = await app.invoke(null, beforeNodeB.config);

Fork

Branch from a checkpoint with modified state:

Python:

# Find checkpoint
history = list(app.get_state_history(config))
before_joke = next(s for s in history if s.next == ("write_joke",))
 
# Fork: update state
fork_config = app.update_state(
    before_joke.config,
    values={"topic": "chickens"}  # Change from original topic
)
 
# Resume from fork
fork_result = app.invoke(None, fork_config)
# write_joke re-executes with new topic

JavaScript:

const states = [];
for await (const state of app.getStateHistory(config)) {
  states.push(state);
}
 
const beforeJoke = states.find(s => s.next.includes("writeJoke"));
 
const forkConfig = await app.updateState(
  beforeJoke.config,
  { topic: "chickens" }
);
 
const forkResult = await app.invoke(null, forkConfig);

Control Next Node with as_node

Specify which node the update comes from:

Python:

fork_config = app.update_state(
    checkpoint_config,
    values={"topic": "chickens"},
    as_node="generate_topic"  # Treat as if this node produced the update
)
# Execution resumes at generate_topic's successors

JavaScript:

const forkConfig = await app.updateState(
  checkpointConfig,
  { topic: "chickens" },
  { asNode: "generateTopic" }
);

Use as_node when:

  • Parallel branches: Multiple nodes updated in same step
  • No history: Setting up state on fresh thread
  • Skipping nodes: Make graph think a node already ran

Time Travel with Interrupts

Interrupts re-trigger during time travel:

from langgraph.types import interrupt, Command
 
def ask_human(state):
    answer = interrupt("What is your name?")
    return {"value": [f"Hello, {answer}!"]}
 
# First run
app.invoke({"value": []}, config)
app.invoke(Command(resume="Alice"), config)
 
# Replay from before ask_human
history = list(app.get_state_history(config))
before_ask = [s for s in history if s.next == ("ask_human",)][-1]
 
replay_result = app.invoke(None, before_ask.config)
# Pauses at interrupt again - needs new Command(resume=...)
 
# Resume with different answer
app.invoke(Command(resume="Bob"), config)

Multiple Interrupts

Fork between interrupts to change later answers:

def ask_name(state):
    name = interrupt("What is your name?")
    return {"value": [f"name:{name}"]}
 
def ask_age(state):
    age = interrupt("How old are you?")
    return {"value": [f"age:{age}"]}
 
# After completing both interrupts, fork between them
history = list(app.get_state_history(config))
between = [s for s in history if s.next == ("ask_age",)][-1]
 
fork_config = app.update_state(between.config, {"value": ["modified"]})
result = app.invoke(None, fork_config)
# ask_name result preserved, ask_age pauses for new answer

Time Travel with Subgraphs

Default Subgraphs

Subgraphs without their own checkpointer are treated as single steps:

# Subgraph (inherits parent checkpointer)
subgraph = StateGraph(State).add_node("step_a", step_a).compile()
 
app = StateGraph(State).add_node("sub", subgraph).compile(checkpointer=MemorySaver())
 
# Time travel from before subgraph
history = list(app.get_state_history(config))
before_sub = [s for s in history if s.next == ("sub",)][-1]
 
fork_config = app.update_state(before_sub.config, {"value": ["forked"]})
result = app.invoke(None, fork_config)
# Entire subgraph re-executes from scratch

Subgraphs with Checkpointers

Give subgraph its own checkpointer for fine-grained time travel:

# Subgraph with own checkpointer
subgraph = StateGraph(State).add_node("step_a", step_a).compile(checkpointer=True)
 
app = StateGraph(State).add_node("sub", subgraph).compile(checkpointer=MemorySaver())
 
# Get subgraph's checkpoint
parent_state = app.get_state(config, subgraphs=True)
sub_config = parent_state.tasks[0].state.config
 
# Fork from within subgraph
fork_config = app.update_state(sub_config, {"value": ["forked"]})
result = app.invoke(None, fork_config)
# Only nodes after the fork point re-execute

Use Cases

Debugging

Replay from before a failing node:

history = list(app.get_state_history(config))
before_failure = next(s for s in history if s.next == ("failing_node",))
 
# Add logging
import logging
logging.basicConfig(level=logging.DEBUG)
 
replay_result = app.invoke(None, before_failure.config)

A/B Testing

Fork to test different approaches:

# Original run
result_a = app.invoke({"query": "test"}, config)
 
# Fork with different parameters
history = list(app.get_state_history(config))
before_llm = next(s for s in history if s.next == ("call_llm",))
 
fork_config = app.update_state(
    before_llm.config,
    {"model": "gpt-4o"}  # Try different model
)
result_b = app.invoke(None, fork_config)
 
# Compare results

Error Recovery

Fix state and resume:

# Run fails
try:
    app.invoke(inputs, config)
except Exception:
    pass
 
# Fix the problematic state
history = list(app.get_state_history(config))
last_good = history[1]  # Before the error
 
fork_config = app.update_state(
    last_good.config,
    {"corrected_field": "fixed_value"}
)
 
# Resume with fixed state
result = app.invoke(None, fork_config)

Best Practices

  1. Use descriptive thread IDs: Make it easy to find the right execution
  2. Filter history carefully: Use next, metadata, or step to find checkpoints
  3. Test forks: Ensure state modifications are valid
  4. Document replay points: Note which checkpoints are good replay targets
  5. Handle re-triggered interrupts: Plan for interrupts firing again

Next Steps

  • Subgraphs: Understand subgraph checkpointing
  • Human-in-the-Loop: Use interrupts effectively
  • Persistence & Memory: Deep dive into checkpointers
Error HandlingSubgraphs