Time Travel: Checkpoint Replay and Execution Branching
Time Travel in LangGraph

1What is "time travel"? ——From history replay to execution branch

In the first few lessons of Module 3, we have learned about the mechanisms of breakpoints and human-in-the-loop.Time Travelis a natural extension of these capabilities: it allows you to not only "pause" the execution of the graph, but also operate it like a time machine,Return to any historical moment that has been executed, and start again from there.

LangGraph After each graph execution step, it will becheckpointerAutomatically save the current complete state as a snapshot. These snapshots form a timeline, and each snapshot has a uniquecheckpoint_id. Time travel is realized based on this timeline.

This mechanism is highly similar to Git version control:Checkpoints are like commits, replaying is like checking out a historical commit and viewing it again, and forking is like creating a new branch from a historical commit and committing different changes.

2Building an example Agent - a computing assistant with tool calls

The notebook uses a simple "arithmetic helper" Agent to demonstrate time travel. This Agent can call three tools: addition, multiplication, and division. Its graph structure is exactly the same as Module 1's ReAct Agent. The core is that it enablesMemorySaverAs a checkpointer, this is a prerequisite for time travel.

Tool definition

from langchain_deepseek import ChatDeepSeek

def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a by b."""
    return a / b

tools = [add, multiply, divide]
llm = ChatDeepSeek(model="deepseek-v4-pro")
llm_with_tools = llm.bind_tools(tools)

Build graph (key: pass in MemorySaver)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_core.messages import SystemMessage

# 系统提示，告诉 LLM 它是一个算术助手
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")

# ★ 关键：编译时传入 checkpointer，这样每步执行都会被持久化为快照
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

Run the graph and generate checkpoint history

from langchain_core.messages import HumanMessage

# 指定线程 ID，同一 thread_id 下的所有执行步骤都会被关联到同一时间线
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}
thread = {"configurable": {"thread_id": "1"}}

# 流式执行，打印每步最后一条消息
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

After execution, it prints in sequence: Human message → AI decides to call multiply → tool returns 6 → AI finally answers "2 times 3 equals 6".every step of the processare automatically saved as a checkpoint snapshot.

3Browsing execution history: get_state_history and checkpoint structures

After the graph is executed, you can view the historical status through two APIs:

• graph.get_state(config): Get the current (latest) status snapshot
• graph.get_state_history(config): Get the thread'sallHistorical checkpoints, sorted in reverse chronological order

# 获取当前状态
current = graph.get_state({'configurable': {'thread_id': '1'}})

# 获取所有历史检查点（按时间倒序，索引 0 = 最新）
all_states = [s for s in graph.get_state_history(thread)]
print(len(all_states))  # 输出: 5  ← 5 个步骤产生 5 个检查点

"Multiply 2 and 3" produced a total of5 checkpoints：

Structural analysis of StateSnapshot

Each checkpoint is aStateSnapshotObject, containing the following key fields:

# 查看倒数第二个检查点（最早的实际执行步骤）
snapshot = all_states[-2]

# ① values：该时刻图的完整状态（所有消息）
snapshot.values
# → {'messages': [HumanMessage(content='Multiply 2 and 3', id='4ee8c440...')]}

# ② next：从该快照继续执行时，下一步要去哪个节点
snapshot.next
# → ('assistant',)

# ③ config：该检查点的标识信息，包含 thread_id 和 checkpoint_id
snapshot.config
# → {'configurable': {'thread_id': '1',
#                     'checkpoint_ns': '',
#                     'checkpoint_id': '1ef6a440-a003-6c74-8000-8a2d82b0d126'}}

# ④ metadata：额外信息，如步骤编号、来源
snapshot.metadata
# → {'source': 'loop', 'writes': None, 'step': 0, 'parents': {}}

# ⑤ created_at：该检查点的创建时间
snapshot.created_at
# → '2024-09-03T22:29:52.988265+00:00'

# ⑥ parent_config：上一个检查点的标识（形成链式结构）
snapshot.parent_config

4Replay: re-execute from a certain checkpoint in the past

Replaying is the simplest operation in time travel: find a historical checkpoint, move itconfig(includescheckpoint_id) passed tograph.stream(), the graph will continue running from that point in time, and the execution path will be exactly the same as before.

Practical steps

• 1

  Find the checkpoint to replay

  fromall_statesSelect the target time. For example, chooseall_states[-2], that is, the moment after the graph is started and just receives the user message and is about to execute the assistant node.

• 2

  Confirm the status of the checkpoint and the next node

  Viewto_replay.valuesConfirm the content of the message at that time and viewto_replay.nextConfirm which node will be executed next.

• 3

  Pass the checkpoint's config into stream()

  The first parameter is passedNone(No new input is passed), the second parameter is passedto_replay.config, LangGraph will recognize that the checkpoint has been executed and automatically continue running from there.

# 步骤 1：选定要重放的历史检查点
to_replay = all_states[-2]

# 确认该时刻的状态（只有用户的输入消息）
to_replay.values
# → {'messages': [HumanMessage(content='Multiply 2 and 3')]}

# 确认下一步节点
to_replay.next
# → ('assistant',)

# 确认 checkpoint_id
to_replay.config
# → {'configurable': {'thread_id': '1',
#                     'checkpoint_id': '1ef6a440-a003-6c74-8000-8a2d82b0d126'}}

# 步骤 2：从该检查点重放
# input=None 表示不注入新输入，让图从检查点状态继续
for event in graph.stream(None, to_replay.config, stream_mode="values"):
    event['messages'][-1].pretty_print()

Output after replay:

================================ Human Message =================================
Multiply 2 and 3
================================== Ai Message ==================================
Tool Calls: multiply(a=2, b=3)
================================= Tool Message =================================
6
================================== Ai Message ==================================
The result of multiplying 2 and 3 is 6.

The meaning of replay

The most direct use of replay isDebug and reproduce: When your Agent produces strange output during a certain run, you can accurately restore the execution environment at that time through replay and step by step check the input and output of each node without re-running the entire workflow.

5Fork: modify the past state and open a new execution path

Forking is more powerful than replaying: you don’t just go back in time;Modify the status at that time, and then starting from the modified state, let the graph take a new path. Modification operation passedgraph.update_state()Once completed, it will create a new checkpoint (new branch) based on the original checkpoint without affecting the original timeline.

Key details of message reducers: overwriting vs appending

When modifying the status,messagesFields have special behavior:MessagesState uses add_messages reducer, the default is to append new messages instead of replacing them. To overwrite (modify) an existing message, you mustProvide the same message ID. In this way, the reducer knows "this is a modification of an existing message" rather than "this is a new message".

# 步骤 1：选定要分叉的历史检查点（与重放相同）
to_fork = all_states[-2]

# 查看当时的消息和其 ID
to_fork.values["messages"]
# → [HumanMessage(content='Multiply 2 and 3', id='4ee8c440-0e4a-47d7-852f-06e2a6c4f84d')]

to_fork.config["configurable"]["checkpoint_id"]
# → '1ef6a440-a003-6c74-8000-8a2d82b0d126'

# 步骤 2：使用 update_state 修改该检查点的状态
# 关键：传入相同的消息 ID → 触发"覆盖"而非"追加"
fork_config = graph.update_state(
    to_fork.config,          # 指定在哪个检查点上做修改
    {"messages": [HumanMessage(
        content='Multiply 5 and 3',          # ← 把"2 and 3"改成"5 and 3"
        id=to_fork.values["messages"][0].id  # ← 同一个消息 ID！
    )]},
)

# update_state 返回新创建的分叉检查点的 config
fork_config
# → {'configurable': {'thread_id': '1',
#                     'checkpoint_id': '1ef6a442-3661-62f6-8001-d3c01b96f98b'}}  ← 全新 ID！

# 步骤 3：确认分叉后的状态
graph.get_state({'configurable': {'thread_id': '1'}}).values["messages"]
# → [HumanMessage(content='Multiply 5 and 3', id='4ee8c440...')]  ← 内容已更新

# 步骤 4：从分叉检查点运行图（这次 LangGraph 会真实执行，而非重放）
for event in graph.stream(None, fork_config, stream_mode="values"):
    event['messages'][-1].pretty_print()

Execution output after forking - notice that the problem has changed to "multiplying 5 and 3":

================================ Human Message =================================
Multiply 5 and 3
================================== Ai Message ==================================
Tool Calls: multiply(a=5, b=3)
================================= Tool Message =================================
15
================================== Ai Message ==================================
The result of multiplying 5 and 3 is 15.

Checkpoint chain structure after fork

The original timeline remains intact, and the bifurcated timeline is generated from the bifurcation pointindependent chain, both inthread_idSame butcheckpoint_idThe links are different. current status (get_state) will point to the latest written fork end.

6Time travel via LangGraph API

In addition to native Python code, LangGraph also provides an HTTP API (viaLangGraph SDK), allowing you to also use time travel in LangSmith Studio/Production Services. The way this API works is highly symmetrical to Python code and the concepts are exactly the same.

Replay via API

from langgraph_sdk import get_client

# 连接本地 LangGraph 服务
client = get_client(url="http://127.0.0.1:2024")

# 1. 先运行一次，产生历史
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}
thread = await client.threads.create()
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=initial_input,
    stream_mode="updates",
):
    # 处理流式输出...
    pass

# 2. 获取历史检查点
states = await client.threads.get_history(thread['thread_id'])
to_replay = states[-2]  # 取最早的实际执行检查点
to_replay['checkpoint_id']
# → '1ef6a449-817f-6b55-8000-07c18fbdf7c8'

# 3. 从该检查点重放（传入 checkpoint_id 参数）
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=None,                              # 不传新输入
    stream_mode="values",
    checkpoint_id=to_replay['checkpoint_id']  # ← 关键：传入检查点 ID
):
    print(f"Receiving event type: {chunk.event}")
    print(chunk.data)

Forking via API

# 获取要分叉的检查点
states = await client.threads.get_history(thread['thread_id'])
to_fork = states[-2]
msg_id = to_fork['values']['messages'][0]['id']  # 原始消息的 ID

# 用 update_state 修改历史状态（需提供消息 ID 以触发覆盖）
forked_input = {"messages": HumanMessage(
    content="Multiply 3 and 3",
    id=msg_id  # ← 同一消息 ID
)}
forked_config = await client.threads.update_state(
    thread["thread_id"],
    forked_input,
    checkpoint_id=to_fork['checkpoint_id']  # ← 在哪个检查点上分叉
)
# forked_config 包含新创建的分叉检查点 ID

# 从分叉检查点运行（图会真实执行，计算 3×3=9）
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=None,
    stream_mode="updates",
    checkpoint_id=forked_config['checkpoint_id']  # ← 分叉检查点 ID
):
    # 输出：Multiply 3 and 3 → 9
    pass

7Application scenarios of time travel: debugging, experimentation, error recovery

Time travel is not only an academic concept, it has extremely practical value when building production-level AI agents. The following are three core application scenarios:

Scenario 1: Debugging - Accurately reproduce historical errors

# 找到问题线程
problem_thread = {"configurable": {"thread_id": "problem-thread-id"}}

# 列出所有检查点，找到问题节点前的那一刻
all_states = [s for s in graph.get_state_history(problem_thread)]

# 找到某个可疑的检查点
suspicious = all_states[3]
print(suspicious.next)   # 确认接下来会执行什么
print(suspicious.values) # 查看当时的完整状态

# 从该点重放，观察后续执行
for event in graph.stream(None, suspicious.config, stream_mode="values"):
    event['messages'][-1].pretty_print()

Scenario 2: Experiment – try multiple strategies with different inputs

Scenario 3: Error recovery - bypassing a failed step

# 假设工具节点失败了，失败前的检查点是 all_states[2]
before_failure = all_states[2]

# 手动注入工具应当返回的结果（跳过真实调用）
from langchain_core.messages import ToolMessage

recovery_config = graph.update_state(
    before_failure.config,
    {"messages": [ToolMessage(
        content="42",           # 手动提供工具结果
        tool_call_id="..."     # 对应原始工具调用的 ID
    )]},
    as_node="tools"           # 以 tools 节点的身份写入，保持元数据正确
)

# 从恢复检查点继续执行（LLM 会看到工具结果并继续推理）
for event in graph.stream(None, recovery_config, stream_mode="values"):
    event['messages'][-1].pretty_print()

8Summary of core concepts and frequently asked questions

FAQ

Module 3 Knowledge Review