检查节点的参数 - 运行结果 - res: {} {final_output: 关于 猫咪 的七言绝句\n《戏猫》\n狸奴酣卧小窗南醉眼惺忪戏玉簪。\n忽跃雕檐追粉蝶却衔明月入花龛。\n笑话\n小猫问妈妈“为什么人总说‘猫有九条命’” \n妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” \n小猫追问“那为什么狗只有一条命” \n妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。”} - 历史检查点列表 - [ StateSnapshot( values{ topic: 猫咪, poem: 《戏猫》 狸奴酣卧小窗南醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶却衔明月入花龛。 , joke: 小猫问妈妈“为什么人总说‘猫有九条命’” 妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” 小猫追问“那为什么狗只有一条命” 妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。” , final_output: 关于 猫咪 的七言绝句 《戏猫》 狸奴酣卧小窗南醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶却衔明月入花龛。 笑话 小猫问妈妈“为什么人总说‘猫有九条命’” 妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” 小猫追问“那为什么狗只有一条命” 妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。” }, next(), config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-4cc7-61bd-8002-bcd7f63c3238 } }, metadata{ source: loop, step: 2, parents: {} }, created_at2026-06-09T09:36:08.36857500:00, parent_config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-4cc3-6d96-8001-8314b5dbbfe3 } }, tasks(), interrupts() ), StateSnapshot( values{ topic: 猫咪, poem: 《戏猫》 狸奴酣卧小窗南醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶却衔明月入花龛。 , joke: 小猫问妈妈“为什么人总说‘猫有九条命’” 妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” 小猫追问“那为什么狗只有一条命” 妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。” }, next(node_output,), config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-4cc3-6d96-8001-8314b5dbbfe3 } }, metadata{ source: loop, step: 1, parents: {} }, created_at2026-06-09T09:36:08.36723600:00, parent_config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-393f-6082-8000-b9c0708bf03d } }, tasks( PregelTask( id91d81ad6-d2fb-ebd7-2dd1-8e08c2be5a07, namenode_output, path(__pregel_pull, node_output), errorNone, interrupts(), stateNone, result{ final_output: 关于 猫咪 的七言绝句 《戏猫》 狸奴酣卧小窗南醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶却衔明月入花龛。 笑话 小猫问妈妈“为什么人总说‘猫有九条命’” 妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” 小猫追问“那为什么狗只有一条命” 妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。” } ), ), interrupts() ), StateSnapshot( values{ topic: 猫咪 }, next(node_poem, node_joke), config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-393f-6082-8000-b9c0708bf03d } }, metadata{ source: loop, step: 0, parents: {} }, created_at2026-06-09T09:36:06.32054200:00, parent_config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-393d-6b9a-bfff-180e30c1d3bc } }, tasks( PregelTask( id80d8974c-3450-6faa-8b06-dcf8787d74fc, namenode_poem, path(__pregel_pull, node_poem), errorNone, interrupts(), stateNone, result{ poem: 《戏猫》 狸奴酣卧小窗南醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶却衔明月入花龛。 } ), PregelTask( id2ca46996-dfc6-6de3-3662-cf1fe41bf987, namenode_joke, path(__pregel_pull, node_joke), errorNone, interrupts(), stateNone, result{ joke: 小猫问妈妈“为什么人总说‘猫有九条命’” 妈妈答“因为人类怕我们死一次就够他们内疚一辈子。” 小猫追问“那为什么狗只有一条命” 妈妈叹气“因为狗犯错靠装可怜而我们靠记仇。” } ), ), interrupts() ), StateSnapshot( values{}, next(__start__,), config{ configurable: { thread_id: 123, checkpoint_ns: , checkpoint_id: 1f163e6a-393d-6b9a-bfff-180e30c1d3bc } }, metadata{ source: input, step: -1, parents: {} }, created_at2026-06-09T09:36:06.32000900:00, parent_configNone, tasks( PregelTask( id3aa3bc58-3567-5832-d823-4373a9ab94e5, name__start__, path(__pregel_pull, __start__), errorNone, interrupts(), stateNone, result{ topic: 猫咪 } ), ), interrupts() ) ]首先这里的StateSnapshot 就是每一个超步的快照及每完成一个超步就会记录的快照这个快照从上到下是按照时间倒序排序的。1.valuesvalues是记录一个快照执行完后的数据的比如刚开始只有topic猫咪后面执行完步骤2就多出了两个数据2.nextnext是记录下一个节点要执行哪一个的当有并行执行时是有两个的。3. configconfig里面记录的是一个存档的唯一标识thread_idcheckpoint_nscheckpoint_id再持久化数据库中这三个字段共同构建了唯一约束thread_id:是用来标识这个langgraph线程的当下次做可恢复执行用这个来判断是否是同一个图checkpoint_ns是一个命名空间命名空间的作用是区分不同的持久化数据的checkpoint_id这个是一个超步的id用来定位一个超步的4.metadata存档的“标签”这个参数只需要知道step这个参数代表的执行到了第几步但需要注意 的是由于并行执行是一起执行的故而并行执行算成一步5.created_at存档创建的时间这个及字面意思创建这个超步的时间6.parent_config上一个存档的身份证这个参数是记录上一个超步的身份标识的参数有thread_idcheckpoint_nscheckpoint_id7.tasks这个参数呢主要是记录下一个超步执行完的结果至于怎么保存的是当下一个超步执行完成langgraph会自动保存到里面作用就是减少token的消耗和执行代码的效率后续如果执行可恢复执行有了这个就不用重新跑了直接看到这个直接执行下一个langgraph工具调用的进阶玩法之重试机制from typing import Literal from langgraph.graph import StateGraph, START, END from langgraph.prebuilt.tool_node import ToolNode from langgraph.graph.message import MessagesState from langchain.messages import HumanMessage, ToolMessage from langchain.tools import tool from langchain_deepseek import ChatDeepSeek from dataclasses import dataclass from loguru import logger from dotenv import load_dotenv load_dotenv(overrideTrue) import random tool(parse_docstringTrue) def get_weather(city: str) - str: 查询指定城市的当日天气 Args: city: 城市名称 # 70% 概率因网络波动而调用失败 rand_int random.randint(1,10) if rand_int 8: raise ConnectionError(网络波动失败) return f{city} 今天天气不错 tools [get_weather] model ChatDeepSeek( modeldeepseek-v4-flash, extra_body{ thinking: { type: disabled } } ) model_with_tools model.bind_tools(toolstools) dataclass class UserContext: max_attempts: int def llm_node(state: MessagesState) - MessagesState: messages state[messages] response model_with_tools.invoke(messages) return { messages: [response] } def router(state: MessagesState) - Literal[tool_node, END]: if state[messages][-1].tool_calls: return tool_node return END def wrap_tool_call(request, execute): max_attempts request.runtime.context.max_attempts tool_call_id request.runtime.tool_call_id tool_msg for i in range(max_attempts): try: tool_msg execute(request) break except ConnectionError as e: logger.info(工具调用失败当前调用次数: {}, 调用次数上限: {}, 异常信息: {}, i 1, max_attempts, e) if not tool_msg: tool_msg ToolMessage( tool_call_id tool_call_id, content 调用次数到达上限调用失败 ) return tool_msg builder StateGraph(state_schemaMessagesState, context_schemaUserContext) builder.add_node(llm_node, llm_node) builder.add_node(tool_node, ToolNode(toolstools, wrap_tool_callwrap_tool_call)) builder.add_edge(START, llm_node) builder.add_conditional_edges(llm_node, router, path_map[tool_node, END]) builder.add_edge(tool_node, llm_node) graph builder.compile() from IPython.display import display display(graph) res graph.invoke( {messages: [HumanMessage(今天北京天气如何)]}, context UserContext(max_attempts3)) for msg in res[messages]: msg.pretty_print()这串代码中定义了一个wrap_tool_call重试函数先说说这个函数里面的两个参数吧requestToolCallRequest这个函数保存的是一个工具调用节点的参数包含工具的 id、工具的名称工具调用的参数以及上下问的参数execute这是一个可执行的回调函数用于实际调用对应的工具并返回执行结果。这个重试机制的执行流程是这样的当执行到路由函数时工具路由函数会跳转到工具执行的ToolNode但在执行这个发之前会触发这个wrap_tool_call函数执行完这个函数会拿着这个函数返回的结果对state写入从而实现工具的调用于重试当时如果有多个工具调用时ToolNode会对返回的多个tool_call独立调用wrap_tool_call从而实现多工具调用的重试。如何查看子图检查点首先查看父图的检查点会发现父图调用了子图的检查点会多出一下部分state{ configurable: { thread_id: observe_checkpoint_1, checkpoint_ns: call_subgraph:9923d4be-9845-576d-785f-5b1d6da9a94f } }父节点调用的子节点检查点就是按照这个checkpoint_ns命名空间区分的。首先我们可以通过parent_config histories[-2].config获取到子图所在的节点的config配置通过这个配置parent_graph.get_state(configparent_config, subgraphsTrue)subgraphsTrue改为True就能查看子图最新快照了这个快照是子图运行完节点next为空的快照查看完整的检查点subgraph_config histories[-2].tasks[0].state这个能拿到需要查看检查点的所需要的参数{ configurable: { thread_id: observe_checkpoint_1, checkpoint_ns: call_subgraph:9923d4be-9845-576d-785f-5b1d6da9a94f } }通过线程id和命名空间字符串能list(parent_graph.get_state_history(configsubgraph_config))就能查看子图的所有检查点信息了