时间旅行：检查点回放与执行分支
Time Travel in LangGraph

1 什么是"时间旅行"？——从历史回放到执行分支

在学习 Module 3 的前几节课时，我们已经了解了断点（Breakpoints）和人工介入（Human-in-the-loop）的机制。时间旅行（Time Travel） 是这些能力的自然延伸：它让你不仅能"暂停"图的执行，还能像操作时光机一样，回到任意一个已执行过的历史时刻，然后从那里重新出发。

LangGraph 在每个图执行步骤之后，都会由 Checkpointer（检查点器） 自动将当前完整状态保存为一个快照。这些快照构成了一条时间线，每个快照都有唯一的 checkpoint_id。时间旅行正是基于这条时间线实现的。

这个机制与 Git 的版本控制有着高度的相似性：检查点就像 commit，重放就像 checkout 到某个历史 commit 再次查看，分叉就像从某个历史 commit 建立新分支并提交不同的更改。

2 构建示例 Agent——带工具调用的计算助手

notebook 使用一个简单的"算术助手" Agent 来演示时间旅行。这个 Agent 可以调用三个工具：加法、乘法、除法。它的图结构与 Module 1 的 ReAct Agent 完全相同，核心在于启用了 MemorySaver 作为检查点器，这是时间旅行的前提条件。

工具定义

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)

构建图（关键：传入 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)

运行图，产生检查点历史

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()

执行后，依次打印：Human 消息 → AI 决定调用 multiply → 工具返回 6 → AI 最终回答"2 乘以 3 等于 6"。整个过程的每一步都被自动保存为一个检查点快照。

3 浏览执行历史：get_state_history 与检查点结构

图执行完毕后，你可以通过两个 API 来查看历史状态：

• graph.get_state(config)：获取当前（最新）状态快照
• graph.get_state_history(config)：获取该线程的所有历史检查点，按时间倒序排列

# 获取当前状态
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"这次执行共产生 5 个检查点：

StateSnapshot 的结构解析

每个检查点都是一个 StateSnapshot 对象，包含以下关键字段：

# 查看倒数第二个检查点（最早的实际执行步骤）
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

4 重放（Replay）：从过去某个检查点重新执行

重放是时间旅行中最简单的操作：找到一个历史检查点，把它的 config（包含 checkpoint_id）传给 graph.stream()，图就会从那个时间点续跑，执行路径与原来完全相同。

实际操作步骤

• 1

  找到要重放的检查点

  从 all_states 中选取目标时刻。例如选 all_states[-2]，即图启动后刚收到用户消息、即将执行 assistant 节点的那一刻。

• 2

  确认该检查点的状态和下一步节点

  查看 to_replay.values 确认当时的消息内容，查看 to_replay.next 确认接下来会执行哪个节点。

• 3

  将检查点的 config 传入 stream()

  第一个参数传 None（不传新输入），第二个参数传 to_replay.config，LangGraph 会识别该检查点已执行过，自动从那里续跑。

# 步骤 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()

重放后的输出：

================================ 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.

重放的意义

重放最直接的用途是调试复现：当你的 Agent 在某次运行中产生了奇怪的输出，你可以通过重放精确还原当时的执行环境，逐步检查每个节点的输入输出，无需重新运行整个工作流。

5 分叉（Fork）：修改过去状态并开启新的执行路径

分叉比重放更强大：你不仅回到过去，还可以修改那时的状态，然后从修改后的状态出发，让图走一条全新的路径。修改操作通过 graph.update_state() 完成，它会在原有检查点的基础上创建一个新的检查点（新分支），而不影响原来的时间线。

消息 Reducer 的关键细节：覆盖 vs 追加

在修改状态时，messages 字段有特殊的行为：MessagesState 使用 add_messages Reducer，默认是追加新消息而非替换。要想覆盖（修改）一条已有消息，必须提供相同的消息 ID。这样 Reducer 才知道"这是对已有消息的修改"，而不是"这是一条新消息"。

# 步骤 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()

分叉后的执行输出——注意问题已经变成了"5 和 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.

分叉后的检查点链结构

原始时间线完好保留，分叉时间线是从分叉点产生的独立链条，两者在 thread_id 相同但 checkpoint_id 链路不同。当前状态（get_state）会指向最新写入的分叉末端。

6 通过 LangGraph API 实现时间旅行

除了本地 Python 代码，LangGraph 还提供了 HTTP API（通过 LangGraph SDK），让你在 LangSmith Studio / 生产服务中同样使用时间旅行。这个 API 的工作方式与 Python 代码高度对称，概念完全相同。

通过 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)

通过 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

7 时间旅行的应用场景：调试、实验、错误恢复

时间旅行不仅是一个学术概念，它在构建生产级 AI Agent 时有着极其实用的价值。以下是三个核心应用场景：

场景一：调试——精确复现历史错误

# 找到问题线程
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()

场景二：实验——用不同输入尝试多种策略

场景三：错误恢复——绕过某个失败步骤

# 假设工具节点失败了，失败前的检查点是 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()

8 核心概念总结与常见问题

常见问题解答

Module 3 知识回顾