How to replay a LangGraph run from a checkpoint

Checkpoint histories turn a completed LangGraph thread into a set of restartable execution boundaries. Replaying from the point before a suspect node isolates that portion of the graph, while a fork lets you change saved state and compare an alternate path without rerunning earlier nodes.

An in-memory checkpointer and deterministic nodes expose replay behavior without a model or credentials. Use a durable checkpointer instead when checkpoint history must survive process restarts.

LangGraph returns the newest checkpoint first from get_state_history(). A snapshot whose next tuple contains write_joke holds the state immediately before that node; invoke(None, checkpoint_config) re-executes downstream nodes, while update_state() creates a branch without changing the original checkpoint.

1. Create replay_checkpoint.py with the state schema and deterministic node functions.

   replay_checkpoint.py

   from typing_extensions import NotRequired, TypedDict
    
   from langgraph.checkpoint.memory import InMemorySaver
   from langgraph.graph import END, START, StateGraph
    
    
   class State(TypedDict):
       topic: NotRequired[str]
       joke: NotRequired[str]
    
    
   def choose_topic(state: State) -> State:
       return {"topic": state.get("topic", "socks")}
    
    
   def write_joke(state: State) -> State:
       return {"joke": f"A joke about {state['topic']}."}

2. Append the graph topology and in-memory checkpointer to replay_checkpoint.py.

   replay_checkpoint.py

   builder = StateGraph(State)
   builder.add_node("choose_topic", choose_topic)
   builder.add_node("write_joke", write_joke)
   builder.add_edge(START, "choose_topic")
   builder.add_edge("choose_topic", "write_joke")
   builder.add_edge("write_joke", END)
   graph = builder.compile(checkpointer=InMemorySaver())

3. Append the original invocation and checkpoint replay to replay_checkpoint.py.

   replay_checkpoint.py

   thread = {"configurable": {"thread_id": "checkpoint-demo"}}
   original = graph.invoke({}, thread)
    
   history = list(graph.get_state_history(thread))
   before_joke = next(
       snapshot for snapshot in history
       if snapshot.next == ("write_joke",)
   )
   replayed = graph.invoke(None, before_joke.config)

   The next nodes or metadata identify a checkpoint more safely than list position because the history is newest-first and grows as the thread branches.

4. Append the checkpoint fork and original-state lookup to replay_checkpoint.py.

   replay_checkpoint.py

   fork_config = graph.update_state(before_joke.config, {"topic": "chicken"})
   forked = graph.invoke(None, fork_config)
   original_checkpoint = graph.get_state(history[0].config)

5. Append the output and fail-capable assertions to replay_checkpoint.py.

   replay_checkpoint.py

   print(f"checkpoint_next={before_joke.next}")
   print(f"original={original}")
   print(f"replayed={replayed}")
   print(f"forked={forked}")
   print(f"original_unchanged={original_checkpoint.values}")
    
   assert replayed == original
   assert forked["topic"] == "chicken"
   assert forked["joke"] != original["joke"]
   assert original_checkpoint.values == original

6. Verify checkpoint replay and fork preservation with replay_checkpoint.py.

   $ python replay_checkpoint.py
   checkpoint_next=('write_joke',)
   original={'topic': 'socks', 'joke': 'A joke about socks.'}
   replayed={'topic': 'socks', 'joke': 'A joke about socks.'}
   forked={'topic': 'chicken', 'joke': 'A joke about chicken.'}
   original_unchanged={'topic': 'socks', 'joke': 'A joke about socks.'}

   The matching replayed value shows downstream execution resumed from the selected checkpoint. The changed forked value and unchanged original snapshot prove that update_state() created a separate branch.