检查节点的参数

============================== -> 运行结果 <- ============================== 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_at="2026-06-09T09:36:08.368575+00: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_at="2026-06-09T09:36:08.367236+00:00", parent_config={ "configurable": { "thread_id": "123", "checkpoint_ns": "", "checkpoint_id": "1f163e6a-393f-6082-8000-b9c0708bf03d" } }, tasks=( PregelTask( id="91d81ad6-d2fb-ebd7-2dd1-8e08c2be5a07", name="node_output", path=("__pregel_pull", "node_output"), error=None, interrupts=(), state=None, 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_at="2026-06-09T09:36:06.320542+00:00", parent_config={ "configurable": { "thread_id": "123", "checkpoint_ns": "", "checkpoint_id": "1f163e6a-393d-6b9a-bfff-180e30c1d3bc" } }, tasks=( PregelTask( id="80d8974c-3450-6faa-8b06-dcf8787d74fc", name="node_poem", path=("__pregel_pull", "node_poem"), error=None, interrupts=(), state=None, result={ "poem": """ 《戏猫》 狸奴酣卧小窗南，醉眼惺忪戏玉簪。 忽跃雕檐追粉蝶，却衔明月入花龛。 """ } ), PregelTask( id="2ca46996-dfc6-6de3-3662-cf1fe41bf987", name="node_joke", path=("__pregel_pull", "node_joke"), error=None, interrupts=(), state=None, 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_at="2026-06-09T09:36:06.320009+00:00", parent_config=None, tasks=( PregelTask( id="3aa3bc58-3567-5832-d823-4373a9ab94e5", name="__start__", path=("__pregel_pull", "__start__"), error=None, interrupts=(), state=None, result={ "topic": "猫咪" } ), ), interrupts=() ) ]

首先这里的StateSnapshot 就是每一个超步的快照，及每完成一个超步就会记录的快照，这个快照从上到下是按照时间倒序排序的。

1.values

values是记录一个快照执行完后的数据的，比如刚开始只有topic：猫咪，后面执行完步骤2就多出了两个数据，

2.next

next是记录下一个节点要执行哪一个的，当有并行执行时是有两个的。

3. config

config里面记录的是一个存档的唯一标识（thread_id，checkpoint_ns，checkpoint_id），再持久化数据库中这三个字段共同构建了唯一约束，thread_id:是用来标识这个langgraph线程的，当下次做可恢复执行用这个来判断是否是同一个图；checkpoint_ns：是一个命名空间，命名空间的作用是区分不同的持久化数据的；checkpoint_id：这个是一个超步的id，用来定位一个超步的

4.metadata：存档的“标签”

这个参数只需要知道step，这个参数代表的执行到了第几步，但需要注意 的是由于并行执行是一起执行的故而并行执行算成一步

5.created_at：存档创建的时间

这个及字面意思，创建这个超步的时间

6.parent_config：上一个存档的身份证

这个参数是记录上一个超步的身份标识的，参数有（thread_id，checkpoint_ns，checkpoint_id）；

7.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(override=True) import random @tool(parse_docstring=True) 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( model="deepseek-v4-flash", extra_body={ "thinking": { "type": "disabled" } } ) model_with_tools = model.bind_tools(tools=tools) @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_schema=MessagesState, context_schema=UserContext) builder.add_node("llm_node", llm_node) builder.add_node("tool_node", ToolNode(tools=tools, wrap_tool_call=wrap_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_attempts=3)) for msg in res['messages']: msg.pretty_print()

这串代码中定义了一个wrap_tool_call重试函数，先说说这个函数里面的两个参数吧：

request：ToolCallRequest；这个函数保存的是一个工具调用节点的参数，包含工具的 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(config=parent_config, subgraphs=True)subgraphs=True改为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(config=subgraph_config))就能查看子图的所有检查点信息了