一、什么是时间旅行想象你在玩游戏有存档功能存档1刚进入游戏 存档2打完第一关 存档3打完第二关 存档4通关如果你想重新打第二关只需要读取存档2就行。LangGraph 的时间旅行就是这个概念——你可以回溯到任意历史状态从那里重新开始执行。二、为什么需要时间旅行2.1 实际场景场景描述调试发现问题后回溯到某个步骤查看当时的状态重试某个步骤失败了从那里重新执行分支探索从历史点尝试不同的路径审计查看完整的执行历史了解发生了什么2.2 一个接地气的例子假设你在做一个自动订餐的 AI 助手步骤1: 用户说我想订餐 步骤2: AI 问您想吃什么 步骤3: 用户说汉堡 步骤4: AI 问要加芝士吗 步骤5: 用户说要 步骤6: AI 下单成功如果用户在第6步说等等我不要芝士了你可以回溯到步骤4要加芝士吗修改答案为不要从那里重新执行这就是时间旅行的威力三、核心概念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(checkpointermemory)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(fstep: {state.values.get(step)}) print(fdata: {state.values.get(data)}) print(fnext: {state.next}) print(fcheckpoint_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 从历史恢复找到step1的检查点从那里恢复执行# 找到目标检查点 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(fstep{state.values.get(step)}, next{state.next})6.2 历史记录[1] step3, next() - 第二次执行完成 [2] step2, next(step_3,) - 第二次执行中间 [3] step1, next(step_2,) [4] step3, next(step_1,) - 第一次执行完成 [5] step2, next(step_3,) - 第一次执行中间 [6] step1, next(step_2,) [7] step0, 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) break8.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(checkpointermemory) # 错误 - 没有历史记录 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(checkpointermemory) 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找到目标检查点 (step1):) 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 [{i1}] step{state.values.get(step)}, fdata{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] 从历史检查点恢复 找到目标检查点 (step1): 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] 完成处理 结果: step3, dataHello World! [第二次执行 - 继续] [步骤1] 初始化数据 [步骤2] 处理数据 [步骤3] 完成处理 结果: step3, dataHello World! 完整历史: [1] step3, dataHello World!, next() [2] step2, dataHello World, next(step_3,) [3] step1, dataHello, next(step_2,) [4] step3, dataHello World!, next(step_1,) [5] step3, dataHello World!, next(__start__,) [6] step3, dataHello World!, next() [7] step2, dataHello World, next(step_3,) [8] step1, dataHello, next(step_2,) [9] step0, dataA, next(step_1,) [10] stepNone, dataNone, next(__start__,) [场景5] 时间旅行到特定点 选择历史状态 #3: checkpoint_id: 1f14b5e2-328a-68c6-8006-21d2b606df23 step: 1 data: Hello 从该点重新执行... [步骤2] 处理数据 [步骤3] 完成处理 新结果: step3, dataHello 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] 从历史检查点恢复 找到目标检查点 (step1): 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记住状态时间旅行的基础Human-in-the-Loop人工干预可以在干预点回溯Custom State自定义状态状态历史包含自定义字段Time Travel回溯历史基于以上所有功能简单说时间旅行是 Memory 功能的延伸让你可以读取任意存档。