AI开发之LangGraph教程7~时间旅行 (Time Travel)

一、什么是时间旅行？

想象你在玩游戏，有"存档"功能：

存档1：刚进入游戏 存档2：打完第一关 存档3：打完第二关 存档4：通关

如果你想重新打第二关，只需要读取"存档2"就行。

LangGraph 的时间旅行就是这个概念——你可以回溯到任意历史状态，从那里重新开始执行。

二、为什么需要时间旅行？

2.1 实际场景

2.2 一个接地气的例子

假设你在做一个"自动订餐"的 AI 助手：

步骤1: 用户说"我想订餐" 步骤2: AI 问"您想吃什么？" 步骤3: 用户说"汉堡" 步骤4: AI 问"要加芝士吗？" 步骤5: 用户说"要" 步骤6: AI 下单成功

如果用户在第6步说"等等，我不要芝士了"，你可以：

1. 回溯到步骤4（"要加芝士吗？"）
2. 修改答案为"不要"
3. 从那里重新执行

这就是时间旅行的威力！

三、核心概念

3.1 Checkpoint（检查点）

每个步骤执行后，LangGraph 会自动保存一个"快照"：

执行流程: START -> step_1 -> step_2 -> step_3 -> END 检查点: checkpoint_0: 初始状态 checkpoint_1: step_1 完成后的状态 checkpoint_2: step_2 完成后的状态 checkpoint_3: step_3 完成后的状态（最终）

3.2 checkpoint_id

每个检查点都有一个唯一标识符：

checkpoint_id: 1f14b599-6ce1-6255-8003-865b83a1cfcd

这就像"存档文件名"，用来精确定位某个历史状态。

3.3 状态历史

通过get_state_history()可以获取所有历史状态：

for state in graph.get_state_history(config): print(state.values) # 当时的状态值 print(state.next) # 下一个要执行的节点 print(state.config) # 包含 checkpoint_id

四、完整示例

4.1 场景设置

我们创建一个简单的三步流程：

START -> step_1 -> step_2 -> step_3 -> END

每一步都会修改状态，方便观察时间旅行效果。

4.2 代码实现

from typing import Annotated from typing_extensions import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages # 定义状态 class State(TypedDict): messages: Annotated[list, add_messages] step: int data: str history: list # 定义节点 def step_1(state: State) -> dict: return { "step": 1, "data": "Hello", "history": state.get("history", []) + ["步骤1完成"] } def step_2(state: State) -> dict: return { "step": 2, "data": state["data"] + " World", "history": state["history"] + ["步骤2完成"] } def step_3(state: State) -> dict: return { "step": 3, "data": state["data"] + "!", "history": state["history"] + ["步骤3完成"] } # 构建图 graph_builder = StateGraph(State) graph_builder.add_node("step_1", step_1) graph_builder.add_node("step_2", step_2) graph_builder.add_node("step_3", step_3) graph_builder.add_edge(START, "step_1") graph_builder.add_edge("step_1", "step_2") graph_builder.add_edge("step_2", "step_3") graph_builder.add_edge("step_3", END) # 关键：必须使用 checkpointer memory = MemorySaver() graph = graph_builder.compile(checkpointer=memory)

4.3 正常执行

config = {"configurable": {"thread_id": "demo"}} result = graph.invoke({ "messages": [], "step": 0, "data": "", "history": [] }, config) print(result) # {'step': 3, 'data': 'Hello World!', 'history': ['步骤1完成', '步骤2完成', '步骤3完成']}

五、时间旅行操作

5.1 查看历史

for state in graph.get_state_history(config): print(f"step: {state.values.get('step')}") print(f"data: {state.values.get('data')}") print(f"next: {state.next}") print(f"checkpoint_id: {state.config['configurable']['checkpoint_id']}") print("-" * 50)

输出：

step: 3 data: Hello World! next: () checkpoint_id: 1f14b599-6ce1-6255-8003-865b83a1cfcd -------------------------------------------------- step: 2 data: Hello World next: ('step_3',) checkpoint_id: 1f14b599-6cbc-69f4-8002-f6b316827cc1 -------------------------------------------------- step: 1 data: Hello next: ('step_2',) checkpoint_id: 1f14b599-6c7e-67ca-8001-cc604f84cd68 -------------------------------------------------- step: 0 data: next: ('step_1',) checkpoint_id: 1f14b599-6c50-63ae-8000-166bb9c9bb46 --------------------------------------------------

5.2 从历史恢复

找到step=1的检查点，从那里恢复执行：

# 找到目标检查点 target_state = None for state in graph.get_state_history(config): if state.values.get('step') == 1: target_state = state break # 从该检查点恢复 result = graph.invoke(None, target_state.config) print(result) # {'step': 3, 'data': 'Hello World!', ...}

5.3 流程图

正常执行: START -> step_1 -> step_2 -> step_3 -> END (保存) (保存) (保存) 时间旅行（从 step_1 恢复）: 读取 step_1 的检查点 | v step_1(跳过) -> step_2 -> step_3 -> END (重新执行)

六、多次执行的历史

6.1 场景

同一个thread_id可以多次执行，每次执行都会记录在历史中：

config = {"configurable": {"thread_id": "multi_demo"}} # 第一次执行 result1 = graph.invoke({"step": 0, ...}, config) # 第二次执行（继续） result2 = graph.invoke({}, config) # 查看完整历史 for state in graph.get_state_history(config): print(f"step={state.values.get('step')}, next={state.next}")

6.2 历史记录

[1] step=3, next=() <- 第二次执行完成 [2] step=2, next=('step_3',) <- 第二次执行中间 [3] step=1, next=('step_2',) [4] step=3, next=('step_1',) <- 第一次执行完成 [5] step=2, next=('step_3',) <- 第一次执行中间 [6] step=1, next=('step_2',) [7] step=0, next=('step_1',) <- 初始状态

七、核心 API 速查

7.1 获取历史

# 遍历所有历史状态 for state in graph.get_state_history(config): state.values # 状态值 state.next # 下一个节点 state.config # 配置（含 checkpoint_id） state.metadata # 元数据

7.2 从历史恢复

# 方式1：使用历史状态的 config graph.invoke(None, historical_state.config) # 方式2：手动指定 checkpoint_id config_with_checkpoint = { "configurable": { "thread_id": "xxx", "checkpoint_id": "specific_checkpoint_id" } } graph.invoke(None, config_with_checkpoint)

7.3 查看当前状态

snapshot = graph.get_state(config) print(snapshot.values) # 当前状态 print(snapshot.next) # 下一个节点（空表示执行完毕）

八、使用场景详解

8.1 调试场景

# 发现问题，回溯查看 for state in graph.get_state_history(config): if "错误" in str(state.values): print("发现问题在这个检查点:") print(state.values) break

8.2 重试场景

# 某个步骤失败，从上一个成功的步骤重试 last_success = None for state in graph.get_state_history(config): if state.values.get("status") == "success": last_success = state break if last_success: # 从上一个成功的状态重新执行 graph.invoke(None, last_success.config)

8.3 分支探索

# 从某个历史点尝试不同的路径 branch_point = list(graph.get_state_history(config))[5] # 修改状态后重新执行 modified_state = branch_point.values.copy() modified_state["choice"] = "option_b" graph.invoke(modified_state, branch_point.config)

九、注意事项

9.1 必须使用 Checkpointer

# 正确 memory = MemorySaver() graph = builder.compile(checkpointer=memory) # 错误 - 没有历史记录 graph = builder.compile() # 无法使用时间旅行

9.2 恢复会重新执行

从历史检查点恢复时，会重新执行后续节点：

原始执行: step_1 -> step_2 -> step_3 从 step_1 恢复: step_2 -> step_3（重新执行）

9.3 thread_id 很重要

不同的thread_id对应不同的历史记录：

# 会话A的历史 config_a = {"configurable": {"thread_id": "session_a"}} # 会话B的历史（完全独立） config_b = {"configurable": {"thread_id": "session_b"}}

十、完整代码

10.1 示例代码

""" LangGraph 时间旅行示例 - 演示状态历史回溯和恢复 本示例演示： - 使用 get_state_history() 获取历史状态 - 从历史检查点恢复执行 - 重播完整状态历史 - 理解 checkpoint_id 的作用 图结构: START -> step_1 -> step_2 -> step_3 -> END """ from typing import Annotated from typing_extensions import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] step: int data: str history: list def step_1(state: State) -> dict: print("\n [步骤1] 初始化数据") return { "step": 1, "data": "Hello", "history": state.get("history", []) + ["步骤1完成"] } def step_2(state: State) -> dict: print("\n [步骤2] 处理数据") new_data = state["data"] + " World" return { "step": 2, "data": new_data, "history": state["history"] + ["步骤2完成"] } def step_3(state: State) -> dict: print("\n [步骤3] 完成处理") return { "step": 3, "data": state["data"] + "!", "history": state["history"] + ["步骤3完成"] } graph_builder = StateGraph(State) graph_builder.add_node("step_1", step_1) graph_builder.add_node("step_2", step_2) graph_builder.add_node("step_3", step_3) graph_builder.add_edge(START, "step_1") graph_builder.add_edge("step_1", "step_2") graph_builder.add_edge("step_2", "step_3") graph_builder.add_edge("step_3", END) memory = MemorySaver() graph = graph_builder.compile(checkpointer=memory) def print_state_history(config): print("\n" + "=" * 70) print("[状态历史记录]") print("=" * 70) history_list = [] for i, state in enumerate(graph.get_state_history(config)): history_list.append(state) print(f"\n[检查点 {len(history_list)}]") print(f" checkpoint_id: {state.config['configurable'].get('checkpoint_id', 'N/A')}") print(f" step: {state.values.get('step', 0)}") print(f" data: {state.values.get('data', '')}") print(f" next: {state.next}") print(f" history: {state.values.get('history', [])}") return history_list def run_demo(): print("=" * 70) print("[LangGraph 时间旅行示例]") print("=" * 70) print("\n图结构: START -> step_1 -> step_2 -> step_3 -> END") print("\n功能演示:") print(" 1. 执行图并记录历史") print(" 2. 使用 get_state_history() 回溯历史") print(" 3. 从历史检查点恢复执行") print(" 4. 重播状态历史") print("=" * 70) config = {"configurable": {"thread_id": "time_travel_demo"}} # 场景1: 正常执行 print("\n" + "=" * 70) print("[场景1] 正常执行图") print("=" * 70) initial_state = { "messages": [], "step": 0, "data": "", "history": [] } print("\n执行中...") result = graph.invoke(initial_state, config) print("\n[OK] 执行完成!") print(f" 最终 step: {result['step']}") print(f" 最终 data: {result['data']}") print(f" 历史: {result['history']}") # 场景2: 查看状态历史 print("\n" + "=" * 70) print("[场景2] 查看状态历史") print("=" * 70) history = print_state_history(config) # 场景3: 从历史检查点恢复 print("\n" + "=" * 70) print("[场景3] 从历史检查点恢复") print("=" * 70) target_state = None for state in history: if state.values.get('step') == 1: target_state = state break if target_state: print(f"\n找到目标检查点 (step=1):") print(f" checkpoint_id: {target_state.config['configurable']['checkpoint_id']}") print(f" data: {target_state.values['data']}") print(f" next: {target_state.next}") print("\n从该检查点恢复执行...") print("（将重新执行 step_2 和 step_3）") replay_config = target_state.config replay_result = graph.invoke(None, replay_config) print("\n[OK] 恢复执行完成!") print(f" 最终 step: {replay_result['step']}") print(f" 最终 data: {replay_result['data']}") # 场景4: 多次执行并回溯 print("\n" + "=" * 70) print("[场景4] 多次执行并回溯") print("=" * 70) config2 = {"configurable": {"thread_id": "multi_step_demo"}} print("\n[第一次执行]") result1 = graph.invoke({ "messages": [], "step": 0, "data": "A", "history": [] }, config2) print(f" 结果: step={result1['step']}, data={result1['data']}") print("\n[第二次执行 - 继续]") result2 = graph.invoke({}, config2) print(f" 结果: step={result2['step']}, data={result2['data']}") print("\n完整历史:") for i, state in enumerate(graph.get_state_history(config2)): print(f" [{i+1}] step={state.values.get('step')}, " f"data={state.values.get('data')}, next={state.next}") # 场景5: 时间旅行到特定点 print("\n" + "=" * 70) print("[场景5] 时间旅行到特定点") print("=" * 70) all_states = list(graph.get_state_history(config2)) if len(all_states) >= 3: target = all_states[2] print(f"\n选择历史状态 #3:") print(f" checkpoint_id: {target.config['configurable']['checkpoint_id']}") print(f" step: {target.values.get('step')}") print(f" data: {target.values.get('data')}") print("\n从该点重新执行...") new_result = graph.invoke(None, target.config) print(f" 新结果: step={new_result['step']}, data={new_result['data']}") # 总结 print("\n" + "=" * 70) print("[总结] 时间旅行关键点") print("=" * 70) print(""" 核心概念: - checkpoint: 每个步骤的状态快照 - checkpoint_id: 唯一标识每个检查点 - get_state_history(): 获取所有历史状态 - 从历史恢复: 使用历史 config 重新执行 使用场景: 1. 调试: 回溯到某个步骤查看状态 2. 重试: 从某个点重新执行 3. 分支: 从历史点探索不同路径 4. 审计: 查看完整执行历史 核心 API: # 获取历史 for state in graph.get_state_history(config): print(state.values) print(state.next) print(state.config) # 从历史恢复 graph.invoke(None, historical_state.config) 注意事项: - 必须使用 MemorySaver 或其他 checkpointer - 恢复时会重新执行后续节点 - checkpoint_id 是时间戳，唯一标识每个状态 """) if __name__ == "__main__": run_demo()

10.2 运行结果

====================================================================== [LangGraph 时间旅行示例] ====================================================================== 图结构: START -> step_1 -> step_2 -> step_3 -> END 功能演示: 1. 执行图并记录历史 2. 使用 get_state_history() 回溯历史 3. 从历史检查点恢复执行 4. 重播状态历史 ====================================================================== ====================================================================== [场景1] 正常执行图 ====================================================================== 执行中... [步骤1] 初始化数据 [步骤2] 处理数据 [步骤3] 完成处理 [OK] 执行完成! 最终 step: 3 最终 data: Hello World! 历史: ['步骤1完成', '步骤2完成', '步骤3完成'] ====================================================================== [场景2] 查看状态历史 ====================================================================== ====================================================================== [状态历史记录] ====================================================================== [检查点 1] checkpoint_id: 1f14b5e2-2f9b-6e5b-8003-105345650ff2 step: 3 data: Hello World! next: () history: ['步骤1完成', '步骤2完成', '步骤3完成'] [检查点 2] checkpoint_id: 1f14b5e2-2f97-6070-8002-79c27734a3f1 step: 2 data: Hello World next: ('step_3',) history: ['步骤1完成', '步骤2完成'] [检查点 3] checkpoint_id: 1f14b5e2-2f50-66b3-8001-95639e1d6bc1 step: 1 data: Hello next: ('step_2',) history: ['步骤1完成'] [检查点 4] checkpoint_id: 1f14b5e2-2f3f-65f5-8000-991c03e1e1a5 step: 0 data: next: ('step_1',) history: [] [检查点 5] checkpoint_id: 1f14b5e2-2ecd-6086-bfff-d79a920cf54e step: 0 data: next: ('__start__',) history: [] ====================================================================== [场景3] 从历史检查点恢复 ====================================================================== 找到目标检查点 (step=1): checkpoint_id: 1f14b5e2-2f50-66b3-8001-95639e1d6bc1 data: Hello next: ('step_2',) 从该检查点恢复执行... （将重新执行 step_2 和 step_3） [步骤2] 处理数据 [步骤3] 完成处理 [OK] 恢复执行完成! 最终 step: 3 最终 data: Hello World! ====================================================================== [场景4] 多次执行并回溯 ====================================================================== [第一次执行] [步骤1] 初始化数据 [步骤2] 处理数据 [步骤3] 完成处理 结果: step=3, data=Hello World! [第二次执行 - 继续] [步骤1] 初始化数据 [步骤2] 处理数据 [步骤3] 完成处理 结果: step=3, data=Hello World! 完整历史: [1] step=3, data=Hello World!, next=() [2] step=2, data=Hello World, next=('step_3',) [3] step=1, data=Hello, next=('step_2',) [4] step=3, data=Hello World!, next=('step_1',) [5] step=3, data=Hello World!, next=('__start__',) [6] step=3, data=Hello World!, next=() [7] step=2, data=Hello World, next=('step_3',) [8] step=1, data=Hello, next=('step_2',) [9] step=0, data=A, next=('step_1',) [10] step=None, data=None, next=('__start__',) ====================================================================== [场景5] 时间旅行到特定点 ====================================================================== 选择历史状态 #3: checkpoint_id: 1f14b5e2-328a-68c6-8006-21d2b606df23 step: 1 data: Hello 从该点重新执行... [步骤2] 处理数据 [步骤3] 完成处理 新结果: step=3, data=Hello World! ====================================================================== [总结] 时间旅行关键点 ====================================================================== 核心概念: - checkpoint: 每个步骤的状态快照 - checkpoint_id: 唯一标识每个检查点 - get_state_history(): 获取所有历史状态 - 从历史恢复: 使用历史 config 重新执行 使用场景: 1. 调试: 回溯到某个步骤查看状态 2. 重试: 从某个点重新执行 3. 分支: 从历史点探索不同路径 4. 审计: 查看完整执行历史 核心 API: # 获取历史 for state in graph.get_state_history(config): print(state.values) print(state.next) print(state.config) # 从历史恢复 graph.invoke(None, historical_state.config) 注意事项: - 必须使用 MemorySaver 或其他 checkpointer - 恢复时会重新执行后续节点 - checkpoint_id 是时间戳，唯一标识每个状态

10.3 结果说明

====================================================================== [LangGraph 时间旅行示例] ====================================================================== [场景1] 正常执行图 [OK] 执行完成! 最终 step: 3 最终 data: Hello World! 历史: ['步骤1完成', '步骤2完成', '步骤3完成'] [场景2] 查看状态历史 [检查点 1] checkpoint_id: 1f14b599-6ce1-6255-8003-865b83a1cfcd step: 3 data: Hello World! next: () [检查点 2] checkpoint_id: 1f14b599-6cbc-69f4-8002-f6b316827cc1 step: 2 data: Hello World next: ('step_3',) [场景3] 从历史检查点恢复 找到目标检查点 (step=1): checkpoint_id: 1f14b599-6c7e-67ca-8001-cc604f84cd68 data: Hello next: ('step_2',) 从该检查点恢复执行... [OK] 恢复执行完成! 最终 step: 3 最终 data: Hello World!

十一、延伸阅读

• LangGraph 官方文档 - 时间旅行
• LangGraph 教程 - 时间旅行
• LangGraph API 参考 - get_state_history

附录：时间旅行 vs 其他功能

简单说：时间旅行是 Memory 功能的延伸，让你可以"读取任意存档"。