【LangGraph】持久化实现的三大能力——时间旅行

【LangGraph】新篇章LangGraph 持久化的三大应用能力*重点*前言一、什么是时间旅行二、为什么需要时间旅行三、核心前提必须启用 Checkpointer四、时间旅行四步法详解4.1 初始执行工作流4.2 查看历史检查点4.3 修改状态创建分支4.4 从检查点恢复执行五、完整示例AI笑话生成器的时间旅行调试5.1 定义状态结构5.2 定义工作流节点5.3 构建工作流图5.4 第一次执行5.5 时间旅行调试案例六、核心API速查表七、总结7.1 核心能力7.2 工作流示意图7.3 小细节7.4 优势上一章- 【LangGraph】持久化实现的三大能力——人机交互前言AI 工作流具有不确定性同样的输入大模型可能产生不同的结果当复杂任务需要多个 AI 调用协同完成时错误可能出现在任何步骤定位问题变得非常困难LangGraph 的时间旅行Time Travel能力 让你可以重放过去的每一次执行检查和修改中间状态甚至尝试不同的决策路径本文带你掌握这项调试神器一、什么是时间旅行LangGraph 的工作方式每个节点执行后都会自动“存档”生成一个检查点checkpoint时间旅行允许你查看历史上任意一个检查点的状态从某个检查点恢复执行重放修改某个检查点的状态然后继续执行分支探索这相当于给你的 AI 工作流装上了一台“时光机”二、为什么需要时间旅行场景说明分析推理过程理解 AI 如何得出最终结果学习成功的决策路径定位和修复错误精确找到错误发生的节点测试修复方案而不影响原始流程探索替代方案尝试不同的输入或中间状态比较不同路径的效果举个例子假设你构建了一个三步 Agent先生成搜索关键词 → 调用搜索引擎 → 总结网页内容 最终答案不理想时你无法直接知道是关键词生成出了问题 还是搜索引擎返回了垃圾信息还是总结模型跑偏了有了时间旅行我们就可以回溯到第一步查看生成的关键词也可以回溯到第二步查看搜索引擎的原始返回甚至可以修改关键词后重新执行后两步快速定位问题所在三、核心前提必须启用 Checkpointer时间旅行依赖持久化机制。在编译图时必须传入checkpointer如InMemorySaver、PostgresSaverfromlanggraph.checkpoint.memoryimportInMemorySaver graphbuilder.compile(checkpointerInMemorySaver())为什么必须要有 checkpointer因为 LangGraph 需要将每个节点执行后的状态自动保存到检查点中没有 checkpointer图不会记录任何历史状态自然也就无法回溯或重放四、时间旅行四步法详解4.1 初始执行工作流importuuid# 创建唯一线程ID确保执行隔离config{configurable:{thread_id:str(uuid.uuid4())}}# 传入空字典作为初始状态config包含线程配置resultgraph.invoke({},config)4.2 查看历史检查点# 获取完整状态历史按时间倒序排列stateslist(graph.get_state_history(config))forstateinstates:print(f检查点ID:{state.config[configurable][checkpoint_id]})print(f下⼀步节点:{state.next})print(f当前状态:{state.values})print(-*50)举个例子【检查点 1/3】ID: 1f0d4d2b-bdc2-6f06-8002-9b67bb1d3867待执行节点: 流程结束状态值: {‘topic’: ‘程序员’, ‘joke’: ‘为什么程序员总是分不清万圣节和圣诞节因为 Oct 31 Dec 25’}【检查点 2/3】ID: 1f0d4d2b-9506-6bf8-8001-6af0cdc2fea0待执行节点: (‘get_joke’,)状态值: {‘topic’: ‘程序员’}4.3 修改状态创建分支# 选择生成主题后的检查点索引1target_statestates[1]# 修改主题字段创建新分支new_configgraph.update_state(target_state.config,# 原检查点配置values{topic:程序员调试代码时的笑话}# 新状态值)# 注意原始检查点仍保留在历史记录中4.4 从检查点恢复执行# 从修改后的配置继续执行传入None表示使用已有状态resultgraph.invoke(None,new_config)# 打印新生成的笑话print(新笑话,result[joke])五、完整示例AI笑话生成器的时间旅行调试5.1 定义状态结构fromtypingimportTypedDictclassState(TypedDict):工作流状态定义topic:str# 笑话主题joke:str# 生成的笑话内容5.2 定义工作流节点fromlangchain.chat_modelsimportinit_chat_model# 初始化模型示例使用简化版GPT-4modelinit_chat_model(gpt-4o-mini)defgenerate_topic(state:State):生成笑话主题节点responsemodel.invoke(生成一个程序员相关的搞笑主题要求六字以内)return{topic:response.content}# 去除可能的多余引号defget_joke(state:State):jokemodel.invoke(f说一个关于{state[topic]}的笑话)return{topic:joke.content}5.3 构建工作流图fromlanggraph.graphimportStateGraph,START,END builderStateGraph(State)builder.add_node(get_topic)builder.add_node(get_joke)builder.add_edge(START,get_topic)builder.add_edge(get_topic,get_joke)builder.add_edge(get_joke,END)# 编译工作流启用内存检查点graphbuilder.compile(checkpointerInMemorySaver())5.4 第一次执行config{configurable:{thread_id:123}}# 步骤一初始执⾏⼯作流print(graph.invoke({},config))5.5 时间旅行调试案例假设首次生成的主题过于宽泛# 获取历史状态stateslist(graph.get_state_history(config))# 查看生成主题后的状态和相关配置updatestates[1]print(update.values[topic])print(update.config)# 修改为更具体的主题new_configupdate.update_state(update[1].config,values{topic:讲一个程序员笑话})# 从新主题继续执行resultgraph.invoke(None,new_config)print(\n优化后结果)print(f新主题{result[topic]})print(f新笑话{result[joke]})六、核心API速查表方法参数说明返回值典型用途graph.get_state(config)thread_id必填最新状态对象获取当前进度graph.get_state_history(config)thread_id必填状态对象列表审计/调试历史graph.update_state(config, values)原配置新值字典新配置对象创建分支版本graph.invoke(None, config)检查点配置最终状态继续执行流程关键参数说明thread_id执行会话的唯一标识checkpoint_id通过历史查询或update_state获取values需要更新的状态字段字典七、总结7.1 核心能力功能实现方式典型应用场景状态追溯get_state_history()调试异常流程分支实验update_state()invoke()A/B测试不同参数流程修复修改中间状态后重放纠正模型偏差交互调试结合interrupt()机制人工审核节点7.2 工作流示意图7.3 小细节检查点自动包含完整上下文无需手动保存修改状态会创建新分支原始记录完整保留通过thread_id可管理多个独立会话互不干扰结合长期存储Store可实现跨会话记忆7.4 优势精准定位 AI 思维偏差节点快速尝试不同生成路径构建可解释的 AI 工作流实现人机协同调试流程掌握时间旅行能力我们就能把 AI 工作流从“黑盒”变成“白盒”——可回溯、可修改、可分支、可复用本期分享就到这里了下期再见~