纲要时光旅行什么是时光旅行状态机的回溯与分叉get_state_history获取历史状态快照replay从任意状态重新执行fork修改历史状态后分叉执行流式输出为什么需要流式输出提升用户体验stream_mode模式对比values输出完整状态updates仅输出变更部分messages逐 token 打字机效果debug全量调试信息工具调用工具调用的四步生命周期创建工具tool绑定模型bind_tools工具选择tool_calling工具执行ToolNode手动执行与自动执行工具调用的链式与并行完整可运行示例时光旅行演示流式输出对比工具调用完整流程总结与相关度说明时光旅行状态机的回溯与分叉LangGraph 的时光旅行本质上是状态机的管理能力——你可以查看任意历史状态从某个状态点重新执行甚至修改历史状态后让流程走向不同分支。这在调试、审计和探索替代方案时极其有用。核心方法get_state_history(config)获取某一线程的所有历史状态快照。app.invoke(None, config_with_snapshot)从指定快照状态重新执行。app.update_state(config, new_values)修改当前状态然后继续执行实现分叉。以下示例展示一个简单的两步工作流演示重放和分叉。importosfromtypingimportTypedDictfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaver load_dotenv()# 定义状态classTravelState(TypedDict):text:str# 节点1添加前缀defadd_prefix(state:TravelState)-TravelState:return{text:[前缀]state[text]}# 节点2添加后缀defadd_suffix(state:TravelState)-TravelState:return{text:state[text][后缀]}# 构建图builderStateGraph(TravelState)builder.add_node(prefix,add_prefix)builder.add_node(suffix,add_suffix)builder.set_entry_point(prefix)builder.add_edge(prefix,suffix)builder.add_edge(suffix,END)memoryMemorySaver()appbuilder.compile(checkpointermemory)# 第一次执行config{configurable:{thread_id:travel-1}}input_data{text:原始文本}print( 原始执行 )resultapp.invoke(input_data,config)print(最终结果:,result[text])# 获取历史状态print(\n 状态历史 )historylist(app.get_state_history(config))fori,stateinenumerate(history):print(f快照{i}:{state.values.get(text,空)})# 时光旅行回到第一个快照prefix 之后的状态重新执行# 假设我们要回到 suffix 之前print(\n 从历史快照重放 )# 取倒数第二个快照prefix 执行后suffix 执行前replay_statehistory[-2]# 通常 snapshot-2 是 prefix 之后print(重放起点状态:,replay_state.values)# 使用该状态的 config 重新调用replay_resultapp.invoke(None,replay_state.config)print(重放结果:,replay_result[text])# 分叉修改状态后执行print(\n 分叉修改状态 )# 回到 prefix 之后修改 text 为其他内容app.update_state(config,{text:[前缀]被修改的文本})# 继续执行fork_resultapp.invoke(None,config)print(分叉结果:,fork_result[text])运行后可以看到原始结果、历史快照列表、重放结果以及修改状态后的分叉结果。流式输出不止是打字机效果LangGraph 大幅增强了流式输出能力通过stream_mode参数可以精确控制输出粒度适应不同场景。模式输出内容适用场景values每个节点执行后的完整状态需要查看完整状态时updates仅输出节点产生的变更只需知道变化时messages逐 token 输出 LLM 生成内容ChatBot 打字机效果debug包含所有调试信息开发调试阶段流式输出对比演示importosfromtypingimportTypedDictfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage load_dotenv()classJokeState(TypedDict):topic:strjoke:str# 节点1将主题改写为主题和小狗defrefine_topic(state:JokeState)-JokeState:new_topicstate[topic]和小狗return{topic:new_topic}# 节点2用LLM生成笑话展示messages模式llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7)defgenerate_joke(state:JokeState)-JokeState:promptf用一句话讲一个关于{state[topic]}的笑话responsellm.invoke([HumanMessage(contentprompt)])return{joke:response.content}builderStateGraph(JokeState)builder.add_node(refine,refine_topic)builder.add_node(joke,generate_joke)builder.set_entry_point(refine)builder.add_edge(refine,joke)builder.add_edge(joke,END)appbuilder.compile()input_data{topic:冰淇淋}# 模式1values完整状态print( stream_modevalues )forchunkinapp.stream(input_data,stream_modevalues):print(chunk)# 模式2updates仅变更print(\n stream_modeupdates )forchunkinapp.stream(input_data,stream_modeupdates):print(chunk)# 模式3messages逐token输出LLM生成内容print(\n stream_modemessages )# 仅在包含LLM的节点上使用messages模式此处简化为全图使用# 实际使用时可针对特定节点配置# 这里用一个更简单的图来演示classChatState(TypedDict):messages:listdefllm_node(state:ChatState)-ChatState:responsellm.invoke(state[messages])return{messages:[response]}chat_builderStateGraph(ChatState)chat_builder.add_node(llm,llm_node)chat_builder.set_entry_point(llm)chat_builder.add_edge(llm,END)chat_appchat_builder.compile()fortoken_chunk,metadatainchat_app.stream({messages:[HumanMessage(content用一句话讲一个笑话)]},stream_modemessages):ifisinstance(token_chunk,AIMessage):print(token_chunk.content,end,flushTrue)print()# 换行运行后可以看到三种模式的区别values输出完整状态updates仅输出变更messages实现打字机效果。工具调用四步拆解LangGraph 将工具调用拆分为更细粒度的步骤每一步都可以插入人工审核或自定义逻辑。创建工具 tool绑定模型 bind_tools工具选择 tool_calling工具执行 ToolNode返回结果 ToolMessage完整工具调用流程importosfromtypingimportTypedDict,Annotated,Sequenceimportoperatorfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportBaseMessage,HumanMessage,ToolMessagefromlanggraph.graphimportStateGraph,END,MessagesStatefromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.checkpoint.memoryimportMemorySaver load_dotenv()# 1. 创建工具defget_weather(city:str)-str:查询指定城市的天气ifcityin[北京,深圳]:returnf{city}天气晴朗25°Celifcityin[哈尔滨]:returnf{city}天气多云10°Creturnf{city}天气未知defget_coldest_cities()-str:返回当前最冷的城市列表return哈尔滨, 北京tools[get_weather,get_coldest_cities]# 2. 绑定模型llmChatOpenAI(modelgpt-3.5-turbo,temperature0)llm_with_toolsllm.bind_tools(tools)# 34. 定义节点工具选择由模型自动完成工具执行由ToolNode完成defcall_model(state:MessagesState)-MessagesState:responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 构建图builderStateGraph(MessagesState)builder.add_node(agent,call_model)builder.add_node(tools,ToolNode(tools))builder.set_entry_point(agent)builder.add_conditional_edges(agent,tools_condition)builder.add_edge(tools,agent)memoryMemorySaver()appbuilder.compile(checkpointermemory)# 测试单个工具调用config{configurable:{thread_id:tools-demo-1}}print( 单工具调用深圳天气 )foreventinapp.stream({messages:[HumanMessage(content深圳天气怎么样)]},config):fornode_name,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]ifisinstance(last_msg,ToolMessage):print(f工具执行 [{node_name}]:{last_msg.content})else:print(f节点{node_name}:{last_msg.content})# 测试链式工具调用print(\n 链式调用最冷城市天气 )config2{configurable:{thread_id:tools-demo-2}}foreventinapp.stream({messages:[HumanMessage(content最冷的城市天气怎么样)]},config2):fornode_name,valueinevent.items():ifmessagesinvalueandvalue[messages]:last_msgvalue[messages][-1]ifisinstance(last_msg,ToolMessage):print(f工具执行 [{node_name}]:{last_msg.content})else:print(f节点{node_name}:{last_msg.content})运行后可以看到第一个例子调用了get_weather第二个例子先调用get_coldest_cities然后自动将返回的城市列表中的每个城市分别调用get_weather展示了链式工具调用的能力。总结LangGraph 的高级特性——时光旅行、流式输出和细粒度工具调用——为构建生产级 AI 应用提供了坚实的基础。时光旅行让调试和探索替代路径变得简单流式输出确保了良好的用户体验而拆解后的工具调用流程则允许开发者在任何环节插入控制逻辑极大地提升了系统的可靠性和灵活性。本文覆盖了时光旅行重放与分叉、流式输出的多种模式values/updates/messages、工具调用的完整生命周期创建/绑定/选择/执行以及链式调用和手自动执行方式并提供了代码示例。