LangGraph 实战（六）：持久化机制和可恢复执行

6.1. 概述

6.1.1. 什么是可恢复执行

在工作流、Agent、任务编排系统里，可恢复执行 Durable Execution 指的是：

把一次任务执行过程中的关键进度、状态、结果保存到可靠存储中，使任务可以在中断、失败、等待外部输入后继续执行。

它解决的是“执行过程能否恢复”的问题。

例如：

普通执行：
开始 -> A -> B -> C
如果执行到 B 后程序挂了，重启后可能只能从 A 重新开始。

可恢复执行：
开始 -> A   保存检查点
     -> B   保存检查点
     -> C
如果执行到 B 后挂了，恢复时可以从已保存的位置继续。

所以它关注的不是“数据长期保存”本身，而是：

执行过程本身具有可恢复性。

它关心的是：系统能否记住“任务执行到了哪里、当前状态是什么、接下来应该继续执行什么”。

6.1.2. LangGraph的持久化机制

6.1.2.1. 什么是LangGraph的持久化机制

LangGraph 的持久化机制（Persistence）可以理解为：

在图执行过程中，将每个关键阶段的图状态保存为检查点，并按照线程进行组织。

这里的“持久化”并不是简单地把某个变量保存到文件里，而是保存一次图执行所需的状态信息，使得未来可以基于检查点继续执行。

LangGraph 的持久化机制围绕以下几个问题展开：

1. 保存什么？

保存图在某个时刻的状态快照，也就是 Checkpoint。

2. 保存到哪里？

保存到 Checkpointer 管理的存储后端中。
开发时可以使用基于内存的 InMemorySaver，生产环境中通常会使用数据库等可靠存储，如 PostgresSaver。

3. 如何区分不同会话？

通过 thread_id 区分不同的执行线程。
同一个 thread_id 表示同一条持久化执行线，也可以理解为同一个会话。

4. 开发者看到的是什么？

开发者通常不会直接操作底层 Checkpoint，而是通过 get_state() 和 get_state_history() 查看 Checkpoint 的开发者视图 StateSnapshot。

因此，本章后续会围绕四个核心问题展开：

如何启用可恢复执行？
如何配置持久化模式？
如何查看历史检查点？
如何利用检查点进行恢复、回放和分叉？

6.1.2.2. 核心组件

本节对核心组件进行梳理，不必死记硬背，用到查阅即可。

LangGraph 的持久化机制依赖一系列核心组件，如下：

6.1.2.3. 小结

本节只需要先记住：

Checkpointer 负责保存检查点

thread_id 负责唯一标识会话

checkpoint_id 负责定位某个具体历史状态

StateSnapshot 是开发者查看检查点时看到的对象。

6.1.3. LangGraph的可恢复执行

6.1.3.1. 什么是LangGraph的可恢复执行

LangGraph 的可恢复执行（Durable Execution）依托于其持久化机制（Persistence）。

可以简单理解为：

持久化机制负责保存执行状态；
可恢复执行负责利用这些状态继续执行。

Persistence 是基础，Durable Execution 是建立在 Persistence 之上的能力。

在没有持久化机制时，一次图执行通常只存在于当前进程中。
如果程序退出、节点中断、人工审批暂停，无法恢复原来的执行进度。

而启用持久化之后，LangGraph 可以把执行过程中的状态保存为检查点。
后续只要使用相同的 thread_id，就可以回到会话对应的历史状态。

6.1.3.2. 使用场景

从使用场景看，可恢复执行主要支持以下几类能力：

因此，LangGraph 的可恢复执行是一组由持久化机制支撑的能力集合。

6.1.3.3. 小结

我们只需要知道

LangGraph 会通过 Checkpointer 持久化保存图执行过程中产生的 Checkpoint，并通过 thread_id 找回同一会话的检查点历史，从而在已有状态基础上继续执行。

6.2. 启用可恢复执行

6.2.1. 步骤

LangGraph 内置了持久化机制，但对于本地运行的 Graph API，只有在编译图时传入 checkpointer，运行过程中的状态才会被记录下来，后续才能基于已有检查点在发生故障时恢复运行或进行检查点回溯。

启用可恢复执行通常需要两步：

1. 在编译图时传入检查点存储器对象 checkpointer
2. 在调用图时传递带有 thread_id 的配置对象

底层的持久化机制会按照 thread_id 将运行时检查点记录在编译时传入的 checkpointer 中。

6.2.2. 检查点实现

LangGraph 提供了检查点存储器基类：langgraph.checkpoint.base.BaseCheckpointSaver

还提供了一系列检查点存储器实现，采用了不同的检查点后端，它们都继承了基类，整体作用是一致的：负责存取检查点，并记录节点执行的中间结果。

可以将 检查点后端 理解为：

负责保存检查点数据的存储介质。

检查点后端必须基于某个检查点实现，常见检查点实现和后端的对应关系如下

在这些实现中，InMemorySaver 更适合学习、调试和本地实验；SQLite 适合轻量级本地持久化；PostgreSQL、MongoDB、Redis 等数据库后端更适合需要跨进程、跨服务保存状态的场景。

6.2.3. 基于内存的检查点存储器

LangGraph 提供了基于内存的检查点存储器 InMemorySaver。它将检查点数据保存在当前 Python 进程的内存中，进程结束则数据丢失，适合快速开发、调试。

在 Jupyter 场景下，只要 Jupyter 内核没有被重启，并且 InMemorySaver() 实例没有被重新创建，内存中的检查点数据就会继续存在。

反之，如果重新执行完整代码，导致 checkpointer = InMemorySaver() 被重新执行，那么之前保存在内存中的检查点也会被清空。

示例如下

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 定义配置对象
config = {"configurable": {"thread_id": "chapter_6_6-2-2"}}
# 调用时传递
graph.invoke({"messages": [HumanMessage("你好，我是老王")]}, config=config)
graph.invoke({"messages": [HumanMessage("从现在开始，你是小王")]}, config=config)
res = graph.invoke({"messages": [HumanMessage("我是谁？你是谁？")]}, config=config)
print(res["output"])

print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
for msg in res["messages"]:
    msg.pretty_print()

输出如下

哎，老王，您这是考我呢？还是今天心情好，想跟我逗个闷子？

您问您是谁——那当然是我的老大哥，老王啊！平时有事儿招呼我，没架子，偶尔还爱指点我两句，我心里都记着。

至于我是谁——我，小王，您的得力助手，机灵踏实，随叫随到。您说往东，我绝不往西，主打一个“靠谱”。

怎么着，老王，是不是又想让我去办啥事儿了？您直说，我听着呢。😎
============================== -> 完整消息列表 <- ==============================
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

你好，老王！很高兴认识你。有什么我可以帮忙的吗？无论是聊天、解答问题还是需要一些建议，我都在这儿呢。😊
================================ Human Message =================================

从现在开始，你是小王
================================== Ai Message ==================================

好的，老王！我是小王。有啥事儿您尽管吩咐，我这儿随时听候差遣。咱是掏心窝子说话，还是聊点别的？您定！😄
================================ Human Message =================================

我是谁？你是谁？
================================== Ai Message ==================================

哎，老王，您这是考我呢？还是今天心情好，想跟我逗个闷子？

您问您是谁——那当然是我的老大哥，老王啊！平时有事儿招呼我，没架子，偶尔还爱指点我两句，我心里都记着。

至于我是谁——我，小王，您的得力助手，机灵踏实，随叫随到。您说往东，我绝不往西，主打一个“靠谱”。

怎么着，老王，是不是又想让我去办啥事儿了？您直说，我听着呢。😎

本例中，三次调用使用了相同的 thread_id，所以后两次调用能够读取前面已经保存的消息历史。最终模型可以知道“用户是老王”，也可以知道“自己被要求扮演小王”。

需要注意的是，如果重新执行完整代码，InMemorySaver() 实例会被重新创建，历史检查点会被清空。因此，消息列表不会继续累加，而是会基于新的空状态重新开始。

6.2.4. 基于持久化数据库的检查点存储器

本节将 PostgreSQL 作为检查点后端，对应的 Checkpointer 实现为 PostgresSaver。

和 InMemorySaver 不同，PostgresSaver 会将检查点保存到 PostgreSQL 数据库中。只要数据库中的记录没有被删除，即使 Python 程序结束、连接对象重建，历史检查点也仍然存在。

在学习 LangChain 时，我们已经介绍了 PostgreSQL 的部署和 PostgresSaver 的用法。此处直接给出案例，不再赘述。

示例如下

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 示例中为了方便演示直接调用 setup()
    # 实际项目中通常建议把数据库初始化/迁移作为独立步骤处理
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    # 定义配置对象
    config = {"configurable": {"thread_id": "chapter_6_6.2.4"}}
    # 调用时传递
    graph.invoke({"messages": [HumanMessage("你好，我是老王")]}, config=config)
    graph.invoke({"messages": [HumanMessage("从现在开始，你是小王")]}, config=config)
    res = graph.invoke({"messages": [HumanMessage("我是谁？你是谁？")]}, config=config)
    print(res["output"])

    print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
    for msg in res["messages"]:
        msg.pretty_print()

运行结果如下

好嘞，老王！既然你问了，那我正式回答：  

**你是老王**——我今儿认的“老大哥”，江湖人称“行走的故事书”，爱唠嗑、有阅历，可能还藏着点神秘技能（比如钓鱼从不空军、下棋总赢隔壁老李？）  

**我是小王**——刚上岗的AI小跟班，脑子灵光（但可能偶尔短路），腿脚勤快（指秒回消息），擅长接梗、查资料、瞎琢磨，主打一个“您开口，我跑腿”。  

需要小王帮您办点啥？甭客气！🚀
============================== -> 完整消息列表 <- ==============================
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

老王你好！我是DeepSeek，很高兴认识你。有什么我可以帮忙的吗？无论是聊天、解答问题、还是需要一些建议，尽管开口，我随时在这儿！😊
================================ Human Message =================================

从现在开始，你是小王
================================== Ai Message ==================================

好的，老王！从现在开始，我就是小王啦。😄  
有什么吩咐？无论唠嗑、干活还是出主意，随叫随到！
================================ Human Message =================================

我是谁？你是谁？
================================== Ai Message ==================================

好嘞，老王！既然你问了，那我正式回答：  

**你是老王**——我今儿认的“老大哥”，江湖人称“行走的故事书”，爱唠嗑、有阅历，可能还藏着点神秘技能（比如钓鱼从不空军、下棋总赢隔壁老李？）  

**我是小王**——刚上岗的AI小跟班，脑子灵光（但可能偶尔短路），腿脚勤快（指秒回消息），擅长接梗、查资料、瞎琢磨，主打一个“您开口，我跑腿”。  

需要小王帮您办点啥？甭客气！🚀

如果多次运行这段代码，并且始终使用相同的 thread_id，那么历史消息会不断累加。这是因为检查点已经被保存到了 PostgreSQL 中，重新创建 Python 连接对象并不会清空数据库中的历史记录。

再次运行结果如下所示。

（正了正脑门上虚拟工牌，一脸正经）  

**您是老王**——我单方面认证的“茶馆街溜子荣誉会长”，上知股票涨跌，下懂馒头蒸法，口头禅是“这我熟啊！”  

**我是小王**——您的AI影子跟班，刚给自己刻了个电子木鱼，每天默念三遍：  
“老王说的都对，如果错了……那一定是我听岔了。” 😎
============================== -> 完整消息列表 <- ==============================
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

老王你好！我是DeepSeek，很高兴认识你。有什么我可以帮忙的吗？无论是聊天、解答问题、还是需要一些建议，尽管开口，我随时在这儿！😊
================================ Human Message =================================

从现在开始，你是小王
================================== Ai Message ==================================

好的，老王！从现在开始，我就是小王啦。😄  
有什么吩咐？无论唠嗑、干活还是出主意，随叫随到！
================================ Human Message =================================

我是谁？你是谁？
================================== Ai Message ==================================

好嘞，老王！既然你问了，那我正式回答：  

**你是老王**——我今儿认的“老大哥”，江湖人称“行走的故事书”，爱唠嗑、有阅历，可能还藏着点神秘技能（比如钓鱼从不空军、下棋总赢隔壁老李？）  

**我是小王**——刚上岗的AI小跟班，脑子灵光（但可能偶尔短路），腿脚勤快（指秒回消息），擅长接梗、查资料、瞎琢磨，主打一个“您开口，我跑腿”。  

需要小王帮您办点啥？甭客气！🚀
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

（拍大腿）哎哟，老王！这声招呼听着就是亲切！今天的茶泡上了没？隔壁李大爷刚还念叨你上回赢他的那盘残局呢——要不咱再研究研究？😏（小王搓手）
================================ Human Message =================================

从现在开始，你是小王
================================== Ai Message ==================================

（立正站好，手机屏闪了闪）得嘞，老王！小王这回把工牌焊脑门上了，您掀个眼皮瞅瞅——  
（工牌晃晃悠悠浮出电子光字：「小王·AI茶馆分部·随叫随到版」）  

有事您敲桌沿，三秒内准给您接话茬儿！ 🫖
================================ Human Message =================================

我是谁？你是谁？
================================== Ai Message ==================================

（正了正脑门上虚拟工牌，一脸正经）  

**您是老王**——我单方面认证的“茶馆街溜子荣誉会长”，上知股票涨跌，下懂馒头蒸法，口头禅是“这我熟啊！”  

**我是小王**——您的AI影子跟班，刚给自己刻了个电子木鱼，每天默念三遍：  
“老王说的都对，如果错了……那一定是我听岔了。” 😎

因此，PostgresSaver 和 InMemorySaver 的关键区别在于：

• InMemorySaver：状态保存在当前进程内存中，进程结束或对象重建后丢失。
• PostgresSaver：状态保存在 PostgreSQL 中，只要数据库记录存在，程序重启后仍可读取。

所以，基于数据库的 Checkpointer 可靠性更高、更适合需要长期保存会话状态的场景。

6.3. 持久化模式

6.3.1. 概述

启用 Checkpointer 后，LangGraph 会在执行过程中保存检查点。

检查点写入越及时，系统在异常中断后可恢复的状态越完整，容灾能力越强；但更及时的持久化通常也意味着更多中间写入，或额外的阻塞式等待，带来额外的性能开销和响应延迟（从调用者的角度考虑）。

LangGraph 支持三种持久化模式，采用不同的检查点保存时机，在容灾能力和性能开销、响应时效性之间作取舍。

• exit：退出模式。只在计算图正常结束、异常退出、被中断（如 Human-In-The-Loop 中断）时保存检查点。不能处理中途进程崩溃的场景。这种模式响应性能开销最小，但容灾能力最弱。

• async：异步模式。默认模式，顾名思义，检查点在后台异步写入。

它会在每个超步结束后写入完整检查点（主检查点），并在图中任务执行完毕后记录中间结果。

和 exit 模式相比，增加了性能开销，但写入操作发生在后台，不会引入明显的响应延迟，同时提升了容灾能力。

• sync：同步模式。

和异步模式唯一的区别在于，LangGraph 会在进入下一个超步之前等待当前主检查点的写入任务完成。

它的容灾能力最强，但在 async 模式的基础上增加了响应延迟。

总结

6.3.2. 示例

持久化模式影响的是检查点写入时机和故障恢复能力，通常不会影响正常情况下的业务输出。

也就是说，同一张图在正常执行完成时，使用 exit、async 或 sync，最终返回结果通常是一样的。真正的区别主要体现在异常中断或进程崩溃后恢复执行时。

durability 控制的是检查点写入策略，不改变图本身的执行逻辑。

要真正看到不同模式的差异

• 可以在学习使用场景后自行设计案例
• 或调试源码，观察不同模式下的写入时机

示例如下

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 定义配置对象
config = {"configurable": {"thread_id": "chapter_tmp"}}
# 调用时传递
res=graph.invoke(
    {"messages": [HumanMessage("你好")]},
    config=config,
    durability="async" # sync / exit
)
print(res["output"])

print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
for msg in res["messages"]:
    msg.pretty_print()

from IPython.display import display
display(graph)

上述代码中，可以通过修改 durability 参数切换持久化模式：

durability="exit"
durability="async"
durability="sync"

运行结果如下

你好！很高兴见到你，有什么我可以帮助你的吗？😊
============================== -> 完整消息列表 <- ==============================
================================ Human Message =================================

你好
================================== Ai Message ==================================

你好！很高兴见到你，有什么我可以帮助你的吗？😊

6.4. 查看历史检查点

LangGraph 在配置检查点存储器后，会把同一个 thread_id 下的执行过程保存为一组检查点。通过这些检查点，我们可以查看图运行的中间状态，也可以为后续的检查点回溯（重放、分叉）和失败恢复做准备。

本节主要介绍两个常用方法：

• graph.get_state_history(config)：查看指定会话的完整历史检查点。
• graph.get_state(config)：查看指定会话的最新检查点，或者查看某个指定 checkpoint_id 对应的检查点。

6.4.1. 查看完整历史检查点列表

6.4.1.1. 示例

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str

class InputState(TypedDict):
    topic: str

class OutputState(TypedDict):
    final_output: str

def node_poem(state: InputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句")]).content

    return {
        "poem": poem
    }

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话")]).content

    return {
        "joke": joke
    }

def node_output(state: OverAllState) -> OutputState:
    logger.info("node_output 已执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于 {topic} 的七言绝句：\n{poem}\n笑话：\n{joke}"

    return {
        "final_output": final_output
    }

builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)
res = graph.invoke({"topic": "猫咪"}, config=config)

from IPython.display import display
display(graph)

print('=' * 30, '-> 运行结果 <-', '=' * 30)
print("res: {}", res)

print('=' * 30, '-> 历史检查点列表 <-', '=' * 30)
history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

6.4.1.2. 运行结果

1. 运行日志

2026-06-09 17:24:10.266 | INFO     | __main__:node_joke:42 - node_joke 已执行
2026-06-09 17:24:10.268 | INFO     | __main__:node_poem:33 - node_poem 已执行
2026-06-09 17:24:14.177 | INFO     | __main__:node_output:51 - node_output 已执行

由于 node_poem 和 node_joke 都从 START 出发，并且彼此没有依赖关系，所以它们会被调度到同一个超步中执行。实际日志顺序可能不同，例如可能先打印 node_joke，也可能先打印 node_poem。

2. 拓扑结构

3. 最终结果和检查点列表

============================== -> 运行结果 <- ==============================
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=()
    )
]

6.4.1.3. 分析

6.4.1.3.1. StateSnapshot

get_state_history() 返回的是一个历史检查点迭代器。将其转换为列表后，可以看到一组 StateSnapshot 对象。

StateSnapshot 是检查点的开发者视图，字段含义如下

• values：当前检查点的状态值

• next：从该检查点继续执行时，下一步将要执行的节点，即归属于下一个超步的节点

• config：当前检查点配置。常见结构如下：

  json
{
"configurable": {
"thread_id": "...",
"checkpoint_ns": "...",
"checkpoint_id": "..."
}
}

其中

• configurable：用于记录检查点的可配置信息，可能包含用户自定义字段，它一定包含以下三个字段：

  • thread_id：会话唯一标识
  • checkpoint_ns：检查点命名空间。根图的命名空间通常为空字符串 ""；子图会使用非空命名空间。
  • checkpoint_id：检查点唯一标识

  这三个字段可以唯一标识一条检查点记录，在大多数检查点存储器实现中，写入检查点后端时都会将它们作为检查点的唯一键

• metadata：检查点元数据，本节只需要关注 step 字段，后者是当前检查点对应的超步编号

• created_at：检查点创建时间

• parent_config：父检查点、即上一个检查点的配置

• tasks：当前检查点关联的待执行任务信息，元素类型通常是 PregelTask。

tasks 通常和 next 对应，表示从当前检查点继续执行时，下一步将要运行的任务。

需要注意的是，tasks 中还可能包含这些任务已经成功（result）或失败（error）的任务记录，这是在下一个超步执行过程中记录的中间结果。在失败恢复场景中，可以避免重复执行已成功节点。

• interrupts：当前图的中断信息，中断机制详见下文

6.4.1.3.2. 检查点主要信息说明

本例中检查点按照如下顺序返回

Step 2 → Step 1 → Step 0 → Step -1

即

最新检查点 → 最旧检查点

其中：

1. step=-1

这是输入检查点，metadata["source"] == "input"，这表示该检查点是在运行时初始化时被记录的，尚未进入图计算循环。

此时用户输入还没有真正变成普通业务节点可消费的状态，因此

python
values={}

从这个检查点继续执行，下一步是内部启动节点 __start__，因此：

python
next=("__start__",)

2. step=0

这是图计算循环的第一个检查点，metadata["source"] == "loop"，这表示检查点是在图计算循环中被写入的。

此时，__start__ 已经把用户输入写入状态，所以 values 中已经可以看到状态信息，如下：

python
values={
"topic": "猫咪"
}

从这个检查点继续执行，node_poem 和 node_joke 将在下一个超步执行，因此：

python
next=("node_poem", "node_joke")

3. step=1

这是图计算循环的第二个检查点，同样 metadata["source"] == "loop"。

同时，它对应第一个执行图节点计算逻辑的超步。

此时，node_poem 和 node_joke 的输出已经到状态中，因此：

python
values={
"topic": "猫咪",
"poem": "...",
"joke": "..."
}

从这个检查点开始，下游的汇总节点 node_output 满足执行条件，因此：

python
next=("node_output",)

4. step=2

这是图计算循环的第三个检查点，同样 metadata["source"] == "loop"

此时，node_output 已经执行完成，最终输出也已经写入状态，因此：

python
values={
"topic": "猫咪",
"poem": "...",
"joke": "...",
"final_output": "..."
}

此时，图已经执行结束，所以：

python
next=()

6.4.1.3.3. tasks字段说明

tasks 是一个元组，其中的每个元素都是一个 PregelTask 实例

tasks 表示：如果从当前检查点继续执行，下一步要运行哪些任务。比如在 step=0 的检查点中：

next=("node_poem", "node_joke")
tasks=(
    PregelTask(name="node_poem", ...),
    PregelTask(name="node_joke", ...),
)

这表示从该检查点继续执行时，下一步会运行 node_poem 和 node_joke 节点，底层执行的是它们的同名任务。

不过，查看历史检查点时，PregelTask 实例的 result 字段可能非空，如：

PregelTask(
    name="node_poem",
    result={
        "poem": "..."
    }
)

这看起来像是“上一个检查点中提前保存了下一个超步的执行结果”。实际上：

• 完整的检查点（主检查点）是在超步末尾保存的

• 但在一个超步内部，LangGraph 还会记录节点级别的计算结果

• checkpoint["id"] 会在保存主检查点前由 create_checkpoint() 推进为新的 ID，然后用这个 ID 写入主检查点（假设当前超步为 S1）。下一轮（超步为 S2）计算循环中，任务执行完毕后的中间结果会用相同的 ID 写入。

因此，下一轮（超步为 S2）任务的计算结果绑定到了“当前超步（超步为 S1）的主检查点”。

• 从检查点后端检索时，会按照 checkpoint_id 将主检查点和中间结果组织起来

• 因此，检索到的历史检查点列表中，主检查点（超步为 S1）和下一个超步（超步为 S2）的中间结果会被放在同一个 StateSnapshot 中

• 如果同一个超步里有多个并行任务，其中部分任务已经成功，另一个任务失败，那么成功任务的结果可以被保存下来；

• 后续恢复时，LangGraph 可以复用这些已成功任务的结果，避免重复执行已经成功的节点。

在 6.5.3. 失败恢复 一节，我们将看到某个超步中部分任务执行成功、部分任务失败导致计算异常终止，恢复运行时已成功任务未被重新执行。

6.4.2. 查看最新检查点

如果只想查看当前线程的最新检查点，可以使用：

latest_history_checkpoint = graph.get_state(config=config)
print(latest_history_checkpoint)

完整示例如下

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str

class InputState(TypedDict):
    topic: str

class OutputState(TypedDict):
    final_output: str

def node_poem(state: InputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句，不要赏析，只给诗句即可")]).content

    return {
        "poem": poem
    }

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话，尽可能简单，不超过一百字")]).content

    return {
        "joke": joke
    }

def node_output(state: OverAllState) -> OutputState:
    logger.info("node_output 已执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于 {topic} 的七言绝句：\n{poem}\n笑话：\n{joke}"

    return {
        "final_output": final_output
    }

builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)
res = graph.invoke({"topic": "猫咪"}, config=config)

from IPython.display import display
display(graph)

print('=' * 30, '-> 运行结果 <-', '=' * 30)
print("res: {}", res)

print('=' * 30, '-> 最新历史检查点 <-', '=' * 30)
latest_history_checkpoint = graph.get_state(config=config)
print(latest_history_checkpoint)

运行结果如下

1. 运行日志

json
2026-06-10 11:23:59.610 | INFO | __main__:node_poem:33 - node_poem 已执行
2026-06-10 11:23:59.611 | INFO | __main__:node_joke:42 - node_joke 已执行
2026-06-10 11:24:01.424 | INFO | __main__:node_output:51 - node_output 已执行

2. 拓扑结构

3. 运行结果及检查点信息

json
============================== -> 运行结果 <- ==============================
res: {} {'final_output': '关于 猫咪 的七言绝句：\n《戏题狸奴》\n昼眠锦褥夜巡檐，碧眼圆瞳映月纤。\n捕鼠本非真本领，得鱼时复近妆奁。\n笑话：\n小猫咪第一次照镜子，看到里面的自己，吓得炸毛：“你是谁？！” 它绕到镜子后面找了一圈，没找到，又回来对着镜子哈气。 最后它恍然大悟，对主人说：“我明白了，那是个隐藏的摄像头！”'}
============================== -> 最新历史检查点 <- ==============================
StateSnapshot(
values={
"topic": "猫咪",
"poem": (
"《戏题狸奴》\n"
"昼眠锦褥夜巡檐，碧眼圆瞳映月纤。\n"
"捕鼠本非真本领，得鱼时复近妆奁。"
),
"joke": (
"小猫咪第一次照镜子，看到里面的自己，吓得炸毛：“你是谁？！” "
"它绕到镜子后面找了一圈，没找到，又回来对着镜子哈气。 "
"最后它恍然大悟，对主人说：“我明白了，那是个隐藏的摄像头！”"
),
"final_output": (
"关于 猫咪 的七言绝句：\n"
"《戏题狸奴》\n"
"昼眠锦褥夜巡檐，碧眼圆瞳映月纤。\n"
"捕鼠本非真本领，得鱼时复近妆奁。\n"
"笑话：\n"
"小猫咪第一次照镜子，看到里面的自己，吓得炸毛：“你是谁？！” "
"它绕到镜子后面找了一圈，没找到，又回来对着镜子哈气。 "
"最后它恍然大悟，对主人说：“我明白了，那是个隐藏的摄像头！”"
),
},
next=(),
config={
"configurable": {
"thread_id": "123",
"checkpoint_ns": "",
"checkpoint_id": "1f1647bd-3512-631a-8002-4e48b5ecda6d",
}
},
metadata={
"source": "loop",
"step": 2,
"parents": {},
},
created_at="2026-06-10T03:24:01.426084+00:00",
parent_config={
"configurable": {
"thread_id": "123",
"checkpoint_ns": "",
"checkpoint_id": "1f1647bd-350d-62c7-8001-34cd7fa012ed",
}
},
tasks=(),
interrupts=(),
)

graph.get_state(config=config) 只返回当前会话的最新检查点。

在本例中，图已经执行完成，所以最新检查点具有以下特征：

python
next=()
tasks=()

这表示没有后续节点需要继续执行。

6.4.3. 根据ID查看指定检查点

如果希望查看某个历史检查点，可以在 configurable 中额外传入 checkpoint_id：

target_config = {
    "configurable": {
        "thread_id": "123",
        "checkpoint_id": "某个历史 checkpoint_id"
    }
}

snapshot = graph.get_state(config=target_config)
print(snapshot)

这样返回的就不是最新检查点，而是指定 checkpoint_id 对应的历史检查点。

6.4.3.1. 获取历史检查点列表

示例如下

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str

class InputState(TypedDict):
    topic: str

class OutputState(TypedDict):
    final_output: str

def node_poem(state: InputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句，不要赏析，只给诗句即可")]).content

    return {
        "poem": poem
    }

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话，尽可能简单，不超过一百字")]).content

    return {
        "joke": joke
    }

def node_output(state: OverAllState) -> OutputState:
    logger.info("node_output 已执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于 {topic} 的七言绝句：\n{poem}\n笑话：\n{joke}"

    return {
        "final_output": final_output
    }

builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)
res = graph.invoke({"topic": "猫咪"}, config=config)

from IPython.display import display
display(graph)

print('=' * 30, '-> 运行结果 <-', '=' * 30)
print("res: {}", res)

print('=' * 30, '-> 历史检查点列表 <-', '=' * 30)
history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

运行结果如下

2026-06-11 18:28:29.126 | INFO     | __main__:node_poem:33 - node_poem 已执行
2026-06-11 18:28:29.128 | INFO     | __main__:node_joke:42 - node_joke 已执行
2026-06-11 18:28:31.021 | INFO     | __main__:node_output:51 - node_output 已执行

============================== -> 运行结果 <- ==============================
res: {} {'final_output': '关于 猫咪 的七言绝句：\n《戏猫》\n狸奴日午卧青毡，蝶影翩然忽跃前。\n捕得飞花轻似梦，独摇银尾弄春烟。\n笑话：\n猫咪走进咖啡店，点了一杯牛奶。  \n店员问：“要不要加糖？”  \n猫咪摇头：“不用，我最近在减肥。”  \n店员看了看它圆滚滚的肚子：“真的吗？”  \n猫咪叹了口气：“唉，都是猫粮的错。”'}
============================== -> 历史检查点列表 <- ==============================
[
    StateSnapshot(
        values={
            "topic": "猫咪",
            "poem": (
                "《戏猫》\n"
                "狸奴日午卧青毡，蝶影翩然忽跃前。\n"
                "捕得飞花轻似梦，独摇银尾弄春烟。"
            ),
            "joke": (
                "猫咪走进咖啡店，点了一杯牛奶。  \n"
                "店员问：“要不要加糖？”  \n"
                "猫咪摇头：“不用，我最近在减肥。”  \n"
                "店员看了看它圆滚滚的肚子：“真的吗？”  \n"
                "猫咪叹了口气：“唉，都是猫粮的错。”"
            ),
            "final_output": (
                "关于 猫咪 的七言绝句：\n"
                "《戏猫》\n"
                "狸奴日午卧青毡，蝶影翩然忽跃前。\n"
                "捕得飞花轻似梦，独摇银尾弄春烟。\n"
                "笑话：\n"
                "猫咪走进咖啡店，点了一杯牛奶。  \n"
                "店员问：“要不要加糖？”  \n"
                "猫咪摇头：“不用，我最近在减肥。”  \n"
                "店员看了看它圆滚滚的肚子：“真的吗？”  \n"
                "猫咪叹了口气：“唉，都是猫粮的错。”"
            ),
        },
        next=(),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-acae-618c-8002-5a8b9e940205",
            }
        },
        metadata={
            "source": "loop",
            "step": 2,
            "parents": {},
        },
        created_at="2026-06-11T10:28:31.022523+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-acab-66b3-8001-d74912e4cb07",
            }
        },
        tasks=(),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "topic": "猫咪",
            "poem": (
                "《戏猫》\n"
                "狸奴日午卧青毡，蝶影翩然忽跃前。\n"
                "捕得飞花轻似梦，独摇银尾弄春烟。"
            ),
            "joke": (
                "猫咪走进咖啡店，点了一杯牛奶。  \n"
                "店员问：“要不要加糖？”  \n"
                "猫咪摇头：“不用，我最近在减肥。”  \n"
                "店员看了看它圆滚滚的肚子：“真的吗？”  \n"
                "猫咪叹了口气：“唉，都是猫粮的错。”"
            ),
        },
        next=("node_output",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-acab-66b3-8001-d74912e4cb07",
            }
        },
        metadata={
            "source": "loop",
            "step": 1,
            "parents": {},
        },
        created_at="2026-06-11T10:28:31.021422+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-9a92-67bc-8000-dd5ab888e9d5",
            }
        },
        tasks=(
            PregelTask(
                id="66dedbf9-3fbb-c539-5f4e-0e3f604a9c56",
                name="node_output",
                path=("__pregel_pull", "node_output"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "final_output": (
                        "关于 猫咪 的七言绝句：\n"
                        "《戏猫》\n"
                        "狸奴日午卧青毡，蝶影翩然忽跃前。\n"
                        "捕得飞花轻似梦，独摇银尾弄春烟。\n"
                        "笑话：\n"
                        "猫咪走进咖啡店，点了一杯牛奶。  \n"
                        "店员问：“要不要加糖？”  \n"
                        "猫咪摇头：“不用，我最近在减肥。”  \n"
                        "店员看了看它圆滚滚的肚子：“真的吗？”  \n"
                        "猫咪叹了口气：“唉，都是猫粮的错。”"
                    )
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "topic": "猫咪",
        },
        next=("node_poem", "node_joke"),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-9a92-67bc-8000-dd5ab888e9d5",
            }
        },
        metadata={
            "source": "loop",
            "step": 0,
            "parents": {},
        },
        created_at="2026-06-11T10:28:29.123773+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-9a90-6db5-bfff-d0eb857db9d8",
            }
        },
        tasks=(
            PregelTask(
                id="8b59cf5a-16b4-c734-41c8-56ac9b7614c9",
                name="node_poem",
                path=("__pregel_pull", "node_poem"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "poem": (
                        "《戏猫》\n"
                        "狸奴日午卧青毡，蝶影翩然忽跃前。\n"
                        "捕得飞花轻似梦，独摇银尾弄春烟。"
                    )
                },
            ),
            PregelTask(
                id="388c08f3-e341-3c76-cfc4-a9825ec97353",
                name="node_joke",
                path=("__pregel_pull", "node_joke"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "joke": (
                        "猫咪走进咖啡店，点了一杯牛奶。  \n"
                        "店员问：“要不要加糖？”  \n"
                        "猫咪摇头：“不用，我最近在减肥。”  \n"
                        "店员看了看它圆滚滚的肚子：“真的吗？”  \n"
                        "猫咪叹了口气：“唉，都是猫粮的错。”"
                    )
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={},
        next=("__start__",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f165804-9a90-6db5-bfff-d0eb857db9d8",
            }
        },
        metadata={
            "source": "input",
            "step": -1,
            "parents": {},
        },
        created_at="2026-06-11T10:28:29.123104+00:00",
        parent_config=None,
        tasks=(
            PregelTask(
                id="63a24d9c-ea6d-0420-4b5a-b978c6eaeb90",
                name="__start__",
                path=("__pregel_pull", "__start__"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "猫咪",
                },
            ),
        ),
        interrupts=(),
    ),
]

6.4.3.2. 根据ID查看特定检查点

获取历史检查点列表后，可以从某个历史快照中取出它的 checkpoint_id，然后调用 graph.get_state() 查看该检查点。

示例如下

print('=' * 30, '-> 根据ID查看特定检查点 <-', '=' * 30)
checkpointe_id = history_checkpoints[3].config["configurable"]["checkpoint_id"]
print(f"checkpointe_id: {checkpointe_id}")

target_config = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": checkpointe_id
    }
}

specific_history_checkpoint = graph.get_state(config=target_config)
print(specific_history_checkpoint)

运行结果如下

============================== -> 根据ID查看特定检查点 <- ==============================
checkpointe_id: 1f165804-9a90-6db5-bfff-d0eb857db9d8
StateSnapshot(
    values={},
    next=("__start__",),
    config={
        "configurable": {
            "thread_id": "123",
            "checkpoint_id": "1f165804-9a90-6db5-bfff-d0eb857db9d8",
        }
    },
    metadata={
        "source": "input",
        "step": -1,
        "parents": {},
    },
    created_at="2026-06-11T10:28:29.123104+00:00",
    parent_config=None,
    tasks=(
        PregelTask(
            id="63a24d9c-ea6d-0420-4b5a-b978c6eaeb90",
            name="__start__",
            path=("__pregel_pull", "__start__"),
            error=None,
            interrupts=(),
            state=None,
            result={
                "topic": "猫咪",
            },
        ),
    ),
    interrupts=(),
)

当前场景下，我们已经获取了完整检查点列表，然后从中取了某个检查点的 checkpoint_id，然后调用 get_state() 查到了它对应的快照。

这看起来有些“绕”，因为我们本来就已经拿到了完整的历史快照。但在实际场景中，这个方法仍然很有意义。

例如，在后续的 检查点回溯-分叉 场景中，我们经常会基于历史检查点创建新的检查点，拿到该检查点的、带有 checkpoint_id 的配置信息 config。此时就可以将 config 传递给 graph.get_state()，从而便捷而精确地查看该检查点的快照。

6.5. 使用场景

6.5.1. 多轮对话

最基础的使用场景是多轮对话。前文示例已经实现了一个简单的多轮对话流程，这里不再展开。

6.5.2. 中断恢复

中断恢复需要结合 LangGraph 的中断机制理解，见下文。

6.5.3. 失败后恢复运行

6.5.3.1. 用法说明

任务失败的原因可能是偶发的外部因素，如网络波动、秘钥过期、第三方服务异常；也可能是程序自身的BUG。

如果是前者，故障发生后，修复外部问题，直接重试即可；如果是后者，需要更改代码修复BUG后，再基于历史检查点恢复运行。

要在失败后基于历史检查点恢复运行，需要满足以下条件：

1. 启用检查点存储器
   – 如果是 InMemorySaver，不要重建检查点存储器对象，否则历史检查点丢失，无法恢复
   – 如果希望跨进程、服务重启后仍可恢复，应使用基于 SQLite、Postgres 等持久化检查点后端的存储器
2. 再次运行时用 None 作为计算图的状态输入
3. 传递的配置信息应包含 thread_id 而不能包含 checkpoint_id

此时，LangGraph 会根据 thread_id 从 checkpointer 读取该会话的最新检查点，并从该检查点继续推进。

如果某个超步中有多个并行任务，其中一部分任务成功，另一部分任务失败，那么已经成功完成的任务结果会作为 pending_writes 被保存。恢复运行时，已经成功完成的任务不会重复执行。

6.5.3.2. 示例

6.5.3.2.1. 模拟失败

本例使用如下结构：

START
  -> node_change_topic
      -> node_poem
      -> node_joke
  -> node_output
  -> END

运行时，node_change_topic 会把输入主题 "猫咪" 改写为 "猫咪: 布偶猫"，然后并行执行 node_poem 和 node_joke，最后由 node_output 汇总输出。

示例如下

from typing import TypedDict

from dotenv import  load_dotenv
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph,START,END
from loguru import logger
load_dotenv(override=True)

model = ChatDeepSeek(
    model = "deepseek-v4-flash",
    extra_body={
        "thinking":{
            "type":"disabled"
        }
    }
)

#1. 声明状态
class OverAllState(TypedDict):
    topic:str
    poem:str
    joke:str
    final_output:str

#1.1 输入状态
class InputState(TypedDict):
    topic:str

#1.2 输出状态
class OutputState(TypedDict):
    final_output:str

topics = ["布偶猫","狸花猫","金渐层"]
topic_index = 0
#2. 定义节点
def node_change_topic(state:InputState)->OverAllState:
    global topic_index
    logger.info("topic_index:{}",topic_index)
    sub_topic = topics[topic_index]
    topic_index += 1
    topic_index %= len(topics)

    return {
        "topic":f"{state["topic"]}:{sub_topic}"
    }
#2.1 同一个超步的位置 成功运行的节点
def node_poem(state:OverAllState) -> OverAllState:
    logger.info("node_poem正在执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于{topic}主题的七言绝句")]).content
    return {
        "poem":poem
    }
import  time
#2.2 同一个超步的位置 运行失败的节点
def node_joke(state:OverAllState) -> OverAllState:
    logger.info("node_joke正在执行")
    topic = state["topic"]
    time.sleep(5)
    raise Exception("人为抛异常")
    joke = model.invoke([HumanMessage(f"写一首关于{topic}主题的笑话")]).content
    return {
        "joke":joke
    }

def node_output(state:OverAllState) -> OutputState:
    logger.info("node_output正在执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于{topic}的七言绝句:{poem}\n 笑话:{joke}\n"
    return {
        "final_output":final_output
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState,input_schema=InputState,output_schema=OutputState)

#3.1 添加节点
builder.add_node("node_change_topic",node_change_topic)
builder.add_node("node_poem",node_poem)
builder.add_node("node_joke",node_joke)
builder.add_node("node_output",node_output)

#3.2 添加边
builder.add_edge(START,"node_change_topic")
builder.add_edge("node_change_topic","node_poem")
builder.add_edge("node_change_topic","node_joke")
builder.add_edge("node_poem","node_output")
builder.add_edge("node_joke","node_output")
builder.add_edge("node_output",END)

#4. 添加检查点后端
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    from IPython.display import display
    display(graph)

    config = {
        "configurable":{
            "thread_id":"chapter03-05"
        }
    }

    res = graph.invoke({"topic":"猫"},config=config)
    print(res)

运行结果如下

2026-06-10 16:17:43.544 | INFO     | __main__:node_change_topic:45 - topic_idx: 0
2026-06-10 16:17:43.545 | INFO     | __main__:node_joke:65 - node_joke 已执行
2026-06-10 16:17:43.548 | INFO     | __main__:node_poem:54 - node_poem 已执行
>Traceback...
Exception: 人为中断
During task with name 'node_joke' and id 'edb59019-4808-ed51-68eb-d5fba8696b13'

本例在 node_joke 中抛出异常，从而模拟计算图执行失败的场景。

为了观察恢复执行时哪些节点会被重新执行，示例中额外定义了全局变量 topics 和 topic_idx。node_change_topic 每执行一次，都会从 topics 中取出一个新的子主题，作为状态字段 topic 的值。

如果恢复运行时 node_change_topic 没有再次执行，则最终结果仍然会使用第一次生成的主题，即 猫咪: 布偶猫。如果该节点被重新执行，则主题可能会变成 猫咪: 狸花猫 或 猫咪: 金渐层。

6.5.3.2.2. 查看历史检查点列表

示例如下

list(graph.get_state_history(config=config))

运行结果如下

[
    StateSnapshot(
        values={
            "topic": "猫咪: 布偶猫"
        },
        next=(
            "node_poem",
            "node_joke"
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e60d-6881-8001-c717582e8893"
            }
        },
        metadata={
            "source": "loop",
            "step": 1,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.431987+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e608-6596-8000-ed45af11cea7"
            }
        },
        tasks=(
            PregelTask(
                id="66287e00-4228-a985-fc09-6791a14e26ce",
                name="node_poem",
                path=(
                    "__pregel_pull",
                    "node_poem"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "poem": "《戏题布偶猫》\n雪翼云裘自绝尘，琉璃碧眼若含春。\n行来莲步生兰气，卧处梨花是此身。"
                }
            ),
            PregelTask(
                id="b2632b70-562b-7d5c-0dae-0f83bd6c2912",
                name="node_joke",
                path=(
                    "__pregel_pull",
                    "node_joke"
                ),
                error="Exception('人为中断')",
                interrupts=(),
                state=None,
                result=None
            )
        ),
        interrupts=()
    ),

    StateSnapshot(
        values={
            "topic": "猫咪"
        },
        next=(
            "node_change_topic",
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e608-6596-8000-ed45af11cea7"
            }
        },
        metadata={
            "source": "loop",
            "step": 0,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.429860+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e606-6a42-bfff-9f09d8a3aa96"
            }
        },
        tasks=(
            PregelTask(
                id="273fe47f-b4cc-93d0-0ab7-57a0db4bad56",
                name="node_change_topic",
                path=(
                    "__pregel_pull",
                    "node_change_topic"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "猫咪: 布偶猫"
                }
            ),
        ),
        interrupts=()
    ),

    StateSnapshot(
        values={},
        next=(
            "__start__",
        ),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f164a86-e606-6a42-bfff-9f09d8a3aa96"
            }
        },
        metadata={
            "source": "input",
            "step": -1,
            "parents": {}
        },
        created_at="2026-06-10T08:43:19.429167+00:00",
        parent_config=None,
        tasks=(
            PregelTask(
                id="ec8e6e32-6969-6798-9c5f-aade41e9211a",
                name="__start__",
                path=(
                    "__pregel_pull",
                    "__start__"
                ),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "猫咪"
                }
            ),
        ),
        interrupts=()
    )
]

可以看到，失败后的检查点中（step=1）记录了当前超步相关任务的执行情况：

• node_poem 执行成功，结果记录在 result 字段中。
• node_joke 执行失败，result 为空，异常信息记录在 error 字段中。

本例中，node_poem 和 node_joke 属于同一个超步。虽然 node_joke 失败了，但 node_poem 已成功，其计算结果会被持久化机制保存下来。因此，后续恢复运行时，node_poem 不会重复执行。

6.5.3.2.3. 修复BUG之后恢复运行

修复 node_joke 中的人为异常后，重新编译计算图，并基于最新检查点恢复运行。

示例如下

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]

    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话")]).content

    return {
        "joke": joke
    }

builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_change_topic", node_change_topic)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_change_topic")
builder.add_edge("node_change_topic", "node_poem")
builder.add_edge("node_change_topic", "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

new_graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(new_graph)

res = new_graph.invoke(None, config=config)
print(res)

这里需要注意两点：

1. 重新编译计算图时，仍然使用原来的 checkpointer 对象。
2. 恢复运行时，输入参数传入 None，并且配置中只传入 thread_id。

运行结果如下

2026-06-10 16:37:12.985 | INFO     | __main__:node_joke:2 - node_joke 已执行
2026-06-10 16:37:18.878 | INFO     | __main__:node_output:76 - node_output 已执行
{
    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《戏题布偶猫》\n雪翼云裘自绝尘，琉璃碧眼若含春。\n行来莲步生兰气，卧处梨花是此身。\n笑话：\n一只布偶猫走进咖啡馆，优雅地跳上吧台，对老板说：“给我来一杯最贵的猫屎咖啡。”  \n老板愣住：“可……您本身就是猫啊？”  \n布偶猫舔了舔爪子，翻了个白眼：“所以呢？我自己产的屎，你们人类不是喝得挺香吗？今天我倒要尝尝，你们拿我的‘周边产品’能泡出什么花样来。”'
}

运行结果中可以观察到：

• node_joke 被重新执行。
• node_output 被执行。
• node_poem 没有被重新执行。

最终输出中的七言绝句和历史检查点中 node_poem.result 记录的内容一致，说明 node_poem 的执行结果被成功复用了。

6.5.4. Time Travel

Time Travel 直译为 时间旅行，在当前场景下太过生硬，从技术实现和使用场景来看，译为 检查点回溯 更为合理。

所谓检查点回溯，是指基于某个历史检查点，重新执行后续流程，或者在该检查点基础上修改状态并创建新的执行分支。

检查点回溯有两种形式，根据是否更改历史状态区分：

• Replay：检查点重放，回到某个历史检查点，沿着原先的执行路径重新执行后续节点。
• Fork：检查点分叉，回到某个历史检查点，修改状态，从该位置创建一条新的执行分支。

二者的共同点是：

检查点之前的节点不会重新执行，检查点之后的节点会重新执行。

二者的区别是：

Replay 不修改历史状态；Fork 会基于历史检查点应用新的状态更新，并创建新的检查点分支。

6.5.4.1. Replay

Replay 模式和失败恢复很接近，但二者的触发方式和语义不同。

• 失败恢复通常基于最新检查点继续运行，配置中只包含 thread_id，不包含 checkpoint_id。

• Replay 则是显式传入某个历史检查点的配置，配置中包含 checkpoint_id。LangGraph 会据此从该历史检查点开始重放后续步骤。

需要注意：

Replay 不是简单读取历史缓存，而是重新执行该检查点之后的节点。

因此，检查点之后的 LLM 调用、API 请求、工具调用、中断 等都会重新触发，最终结果可能和原始结果不同。

6.5.4.1.1. 正常运行的任务

本例仍然使用如下结构：

START
  -> node_change_topic
      -> node_poem
      -> node_joke
  -> node_output
  -> END

示例如下

from typing import TypedDict

from dotenv import  load_dotenv
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph,START,END
from loguru import logger
load_dotenv(override=True)

model = ChatDeepSeek(
    model = "deepseek-v4-flash",
    extra_body={
        "thinking":{
            "type":"disabled"
        }
    }
)

#1. 声明状态
class OverAllState(TypedDict):
    topic:str
    poem:str
    joke:str
    final_output:str

#1.1 输入状态
class InputState(TypedDict):
    topic:str

#1.2 输出状态
class OutputState(TypedDict):
    final_output:str

topics = ["布偶猫","狸花猫","金渐层"]
topic_index = 0
#2. 定义节点
def node_change_topic(state:InputState)->OverAllState:
    global topic_index
    logger.info("topic_index:{}",topic_index)
    sub_topic = topics[topic_index]
    topic_index += 1
    topic_index %= len(topics)

    return {
        "topic":f"{state["topic"]}:{sub_topic}"
    }
#2.1 同一个超步的位置 成功运行的节点
def node_poem(state:OverAllState) -> OverAllState:
    logger.info("node_poem正在执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于{topic}主题的七言绝句")]).content
    return {
        "poem":poem
    }
import  time
#2.2 同一个超步的位置 运行失败的节点
def node_joke(state:OverAllState) -> OverAllState:
    logger.info("node_joke正在执行")
    topic = state["topic"]
    # time.sleep(5)
    # raise Exception("人为抛异常")
    joke = model.invoke([HumanMessage(f"写一首关于{topic}主题的笑话")]).content
    return {
        "joke":joke
    }

def node_output(state:OverAllState) -> OutputState:
    logger.info("node_output正在执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于{topic}的七言绝句:{poem}\n 笑话:{joke}\n"
    return {
        "final_output":final_output
    }

#3. 构建图
builder = StateGraph(state_schema=OverAllState,input_schema=InputState,output_schema=OutputState)

#3.1 添加节点
builder.add_node("node_change_topic",node_change_topic)
builder.add_node("node_poem",node_poem)
builder.add_node("node_joke",node_joke)
builder.add_node("node_output",node_output)

#3.2 添加边
builder.add_edge(START,"node_change_topic")
builder.add_edge("node_change_topic","node_poem")
builder.add_edge("node_change_topic","node_joke")
builder.add_edge("node_poem","node_output")
builder.add_edge("node_joke","node_output")
builder.add_edge("node_output",END)

#4. 添加检查点后端
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    from IPython.display import display
    display(graph)

    config = {
        "configurable":{
            "thread_id":"chapter03-08"
        }
    }

    res = graph.invoke({"topic":"猫咪"},config=config)
    print(res)

运行结果如下

2026-06-12 14:32:42.019 | INFO     | __main__:node_change_topic:45 - topic_idx: 0
2026-06-12 14:32:42.022 | INFO     | __main__:node_joke:63 - node_joke 已执行
2026-06-12 14:32:42.026 | INFO     | __main__:node_poem:54 - node_poem 已执行
2026-06-12 14:32:43.585 | INFO     | __main__:node_output:72 - node_output 已执行
2026-06-12 14:32:43.588 | INFO     | __main__:<module>:102 - res: 
{
    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
}

6.5.4.1.2. 查看历史检查点列表

示例如下

history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

运行结果如下

[
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫',
            'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。',
            'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”',
            'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
        },
        next=(),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4ad3-67ef-8003-01d021de9403'
            }
        },
        metadata={
            'source': 'loop',
            'step': 3,
            'parents': {}
        },
        created_at='2026-06-12T06:32:43.586547+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4acf-62cf-8002-aed204b5115d'
            }
        },
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫',
            'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。',
            'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
        },
        next=('node_output',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-4acf-62cf-8002-aed204b5115d'
            }
        },
        metadata={
            'source': 'loop',
            'step': 2,
            'parents': {}
        },
        created_at='2026-06-12T06:32:43.584774+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3be4-6523-8001-dd86f05c34b8'
            }
        },
        tasks=(
            PregelTask(
                id='b9042203-0ddf-f638-aeeb-e0501601381a',
                name='node_output',
                path=('__pregel_pull', 'node_output'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'final_output': '关于 猫咪: 布偶猫 的七言绝句：\n《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。\n笑话：\n一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
                }
            ),
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪: 布偶猫'
        },
        next=('node_poem', 'node_joke'),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3be4-6523-8001-dd86f05c34b8'
            }
        },
        metadata={
            'source': 'loop',
            'step': 1,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.020569+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bde-6373-8000-9750b4be0665'
            }
        },
        tasks=(
            PregelTask(
                id='eab6bf51-4986-a60f-753c-e493b78c56b8',
                name='node_poem',
                path=('__pregel_pull', 'node_poem'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'poem': '《布偶猫》\n雪团云絮落仙踪，碧眼盈盈醉九重。\n静若琼瑶闲卧月，清姿一步一玲珑。'
                }
            ),
            PregelTask(
                id='a228a7bf-4c3b-2ad3-039f-12ac0107c87b',
                name='node_joke',
                path=('__pregel_pull', 'node_joke'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'joke': '一只布偶猫照镜子，惊呼：“我这么美，怎么还没当上国王？” 主人笑道：“因为你没有‘喵’民拥护。” 布偶猫委屈：“我有啊，铲屎官你不是天天跪着拥护我吗？”'
                }
            )
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'topic': '猫咪'
        },
        next=('node_change_topic',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bde-6373-8000-9750b4be0665'
            }
        },
        metadata={
            'source': 'loop',
            'step': 0,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.018069+00:00',
        parent_config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bdc-60a4-bfff-d4fdcf6e6a9e'
            }
        },
        tasks=(
            PregelTask(
                id='e0034c08-dd19-0ae9-7059-f8414745d212',
                name='node_change_topic',
                path=('__pregel_pull', 'node_change_topic'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'topic': '猫咪: 布偶猫'
                }
            ),
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={},
        next=('__start__',),
        config={
            'configurable': {
                'thread_id': '123',
                'checkpoint_ns': '',
                'checkpoint_id': '1f166288-3bdc-60a4-bfff-d4fdcf6e6a9e'
            }
        },
        metadata={
            'source': 'input',
            'step': -1,
            'parents': {}
        },
        created_at='2026-06-12T06:32:42.017182+00:00',
        parent_config=None,
        tasks=(
            PregelTask(
                id='1a1ac312-152a-dcc3-2aa7-c2588f651448',
                name='__start__',
                path=('__pregel_pull', '__start__'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'topic': '猫咪'
                }
            ),
        ),
        interrupts=()
    )
]

6.5.4.1.3. 重放

get_state_history() 返回的是历史检查点列表，并且默认按照时间倒序排列。

history_checkpoints 结构如下：

history_checkpoints[0] -> 最终完成后的检查点
                        next=()
history_checkpoints[1] -> node_output 执行前的检查点
                        next=('node_output',)
history_checkpoints[2] -> node_poem、node_joke 执行前的检查点
                        next=('node_poem', 'node_joke')
history_checkpoints[3] -> node_change_topic 执行前的检查点
                        next=('node_change_topic',)
history_checkpoints[4] -> __start__ 执行前的输入检查点
                        next=('__start__',)

因此，如果希望从写诗写笑话开始重放，选择 next == ('node_poem', 'node_joke') 的检查点即可。

示例如下

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    config = {
        "configurable":{
            "thread_id":"chapter03-08"
        }
    }
    # 获取到检查点历史
    history_checkpoints = list(graph.get_state_history(config=config))

    new_checkpoint = None
    #next = ('node_poem', 'node_joke')
    for checkpoint in history_checkpoints:
        if checkpoint.next == ('node_poem', 'node_joke'):
            new_checkpoint = checkpoint
            break

    # 如果想要实现replay的效果 状态填写None  config填写为之前某一个检查点的config
    res = graph.invoke(None,config=new_checkpoint.config)
    print(res)

运行结果如下

2026-07-07 18:31:32.019 | INFO     | __main__:node_joke:59 - node_joke正在执行
2026-07-07 18:31:32.023 | INFO     | __main__:node_poem:50 - node_poem正在执行
2026-07-07 18:31:38.158 | INFO     | __main__:node_output:69 - node_output正在执行
{'final_output': '关于猫咪:狸花猫的七言绝句:《狸花猫》\n玄文斑驳隐花光，夜半巡檐气自昂。\n踏雪无痕轻似梦，一窗明月一炉香。\n\n注：我的创作思路是捕捉狸花猫的神秘与灵动。首句“玄文斑驳”描绘其毛发纹理，如暗夜中的花纹；次句“夜半巡檐”展现其夜行习性，气度不凡。后两句以“踏雪无痕”喻其轻盈，结句“明月炉香”营造温馨画面，暗合狸花猫守护家园的意象，使全诗动静相生，虚实相映。\n 笑话:# 狸花猫的烦恼\n\n有只狸花猫去宠物医院看病，医生问它：“哪里不舒服？”\n\n狸花猫叹了口气说：“医生，我总觉得自己不够‘花’。”\n\n医生疑惑：“你可是标准的狸花猫啊，条纹多漂亮！”\n\n狸花猫委屈地说：“可是隔壁那只三花猫，人家身上有黑、白、橘三种颜色，比我缤纷多了！我整天就这一身灰黑条纹，太单调了。”\n\n医生笑了：“那你想要怎么办？”\n\n狸花猫眼睛一亮：“医生，您能不能给我开点染色剂？我想在背上加几块橘色斑点，尾巴再来点白的，耳朵尖染成黑的…”\n\n医生打断它：“等等，那样你就变成‘四不像’了！”\n\n狸花猫思考了一会，突然骄傲地挺起胸膛：“你说得对！我这一身条纹可是老祖宗传下来的迷彩服，抓老鼠时特别管用！三花猫再好看，还得靠我帮忙抓家里的老鼠呢！”\n\n医生欣慰地说：“这就对了。”\n\n狸花猫走出诊室，正好遇到一只胖橘猫。胖橘猫嘲笑它：“哟，这不是那只会抓老鼠却不会卖萌的狸花猫吗？”\n\n狸花猫淡定地回了一句：“我会抓老鼠，你会什么？”\n\n胖橘猫想了想，尴尬地说：“我…我会吃。”\n\n狸花猫笑道：“那不就成了？你负责吃，我负责抓，咱们各有各的活法！”\n\n—— **适合自己的，才是最好的。** 🐱\n'}

本例中：

1. node_poem 和 node_joke 被重新执行。
2. node_output 被重新执行。

由于没有执行change_topic,所以可以一直写同一只猫的内容

6.5.4.2. Fork

6.5.4.2.1. 用法说明

Fork 依赖状态图的 update_state() 方法

示例如下

change_input_config = graph.update_state(
    config=before_router_config.config,
    values={"user_input": "帮我写一个关于布偶猫的笑话"},
    as_node=START
)

其中：

• config：历史检查点的配置信息，通常包含 thread_id、checkpoint_ns 和 checkpoint_id。LangGraph 会据此检索历史检查点

• values：要应用到该检查点上的状态更新

• as_node：指定这次状态更新应当被视为哪个节点产生的输出。

LangGraph 会按照该节点的写入逻辑，将 values 写入状态通道，并据此决定后续从哪些节点继续执行。

基于分叉检查点运行时，并不是重新执行 as_node 本身，而是从它的后继节点继续推进

如 START 所属超步编号为 0，则恢复时从编号为 1 的超步开始运行。

注意

update_state() 不会修改原来的历史检查点，而是基于某个历史检查点创建一个新的检查点。这个新检查点相当于一条新的执行分支。

6.5.4.2.2. 准备检查点历史

6.5.4.2.2.1. 准备运行图并运行一次

示例如下

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from loguru import logger
from dotenv import load_dotenv

load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    username: str
    user_input: str
    output: str

def router_node(state: OverAllState) -> StructuredOutputState:
    logger.info("路由节点已执行")
    user_input = state["user_input"]
    res = model_with_structure.invoke(
        [
            HumanMessage(user_input)
        ]
    )
    logger.info("路由结果-res: {}", res)

    return res

def router(state: StructuredOutputState) -> Literal["node_poem", "node_joke", "node_default"]:
    logger.info("路由函数已执行")
    if state["mode"] == "poem":
        logger.info("路由至 node_poem 节点")
        return "node_poem"
    elif state["mode"] == "joke":
        logger.info("路由至 node_joke 节点")
        return "node_joke"

    logger.info("路由至兜底节点")
    return "node_default"

def node_poem(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句，不要赏析，只给诗句即可")]).content

    return {
        "output": poem
    }

def node_joke(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话，尽可能简单，不超过一百字")]).content

    return {
        "output": joke
    }

def node_default(state: StructuredOutputState) -> OverAllState:
    logger.info(f"node_default 已执行")

    return {
        "output": "无法处理的任务类型"
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("router_node", router_node)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_default", node_default)

builder.add_edge(START, "router_node")
builder.add_conditional_edges(
    "router_node",
    router,
    path_map=["node_poem", "node_joke", "node_default"]
)
builder.add_edge("node_poem", END)
builder.add_edge("node_joke", END)
builder.add_edge("node_default", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)

from IPython.display import display
display(graph)

res = graph.invoke({
    "username": "小黄",
    "user_input": "帮我写一首关于布偶猫的七言绝句"
}, config=config)
print(res)

执行逻辑如下：

1. router_node 根据用户输入提取结构化结果，例如 {"topic": "布偶猫", "mode": "poem"}。
2. router 根据 mode 决定后续分支。
3. 如果 mode == "poem"，进入 node_poem。
4. 如果 mode == "joke"，进入 node_joke。
5. 如果无法识别，则进入 node_default。

运行结果如下

2026-06-11 14:26:48.353 | INFO     | __main__:router_node:31 - 路由节点已执行
2026-06-11 14:26:49.618 | INFO     | __main__:router_node:38 - 路由结果-res: {'topic': '布偶猫', 'mode': 'poem'}
2026-06-11 14:26:49.619 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:26:49.619 | INFO     | __main__:router:45 - 路由至 node_poem 节点
2026-06-11 14:26:49.620 | INFO     | __main__:node_poem:55 - node_poem 已执行
{
    'username': '小黄', 
    'user_input': '帮我写一首关于布偶猫的七言绝句', 
    'output': '《布偶猫》\n蓝眼澄明玉作团，云踪雪影卧雕栏。\n无端一跃梅花落，犹带娇声唤月寒。'
}

6.5.4.2.2.2. 获取历史检查点列表

示例如下

history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

运行结果如下

[
    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
            "output": (
                "《布偶猫》\n"
                "蓝眼澄明玉作团，云踪雪影卧雕栏。\n"
                "无端一跃梅花落，犹带娇声唤月寒。"
            ),
            "topic": "布偶猫",
            "mode": "poem",
        },
        next=(),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-8784-65e9-8002-e21663d55c94",
            }
        },
        metadata={
            "source": "loop",
            "step": 2,
            "parents": {},
        },
        created_at="2026-06-11T06:26:51.611071+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-7488-66c4-8001-f46b7c2d9a4e",
            }
        },
        tasks=(),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
            "topic": "布偶猫",
            "mode": "poem",
        },
        next=("node_poem",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-7488-66c4-8001-f46b7c2d9a4e",
            }
        },
        metadata={
            "source": "loop",
            "step": 1,
            "parents": {},
        },
        created_at="2026-06-11T06:26:49.620436+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
            }
        },
        tasks=(
            PregelTask(
                id="cb3a8901-3a87-d0b0-3bab-d1adda5563d8",
                name="node_poem",
                path=("__pregel_pull", "node_poem"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "output": (
                        "《布偶猫》\n"
                        "蓝眼澄明玉作团，云踪雪影卧雕栏。\n"
                        "无端一跃梅花落，犹带娇声唤月寒。"
                    )
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={
            "username": "小黄",
            "user_input": "帮我写一首关于布偶猫的七言绝句",
        },
        next=("router_node",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
            }
        },
        metadata={
            "source": "loop",
            "step": 0,
            "parents": {},
        },
        created_at="2026-06-11T06:26:48.352564+00:00",
        parent_config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
            }
        },
        tasks=(
            PregelTask(
                id="fb88436e-b163-6308-6341-7d8368e041f9",
                name="router_node",
                path=("__pregel_pull", "router_node"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "topic": "布偶猫",
                    "mode": "poem",
                },
            ),
        ),
        interrupts=(),
    ),

    StateSnapshot(
        values={},
        next=("__start__",),
        config={
            "configurable": {
                "thread_id": "123",
                "checkpoint_ns": "",
                "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
            }
        },
        metadata={
            "source": "input",
            "step": -1,
            "parents": {},
        },
        created_at="2026-06-11T06:26:48.350273+00:00",
        parent_config=None,
        tasks=(
            PregelTask(
                id="a1b5e5fb-ee60-fd7b-68bc-475fa9cb66e2",
                name="__start__",
                path=("__pregel_pull", "__start__"),
                error=None,
                interrupts=(),
                state=None,
                result={
                    "username": "小黄",
                    "user_input": "帮我写一首关于布偶猫的七言绝句",
                },
            ),
        ),
        interrupts=(),
    ),
]

6.5.4.2.2.3. 获取 router_node 之前的检查点

示例如下

router_node 的上一个检查点的 next 字段为 ('router_node',)，据此筛选检查点

before_router_checkpoint = next(h for h in history_checkpoints if h.next == ('router_node',))
before_router_checkpoint

运行结果如下

StateSnapshot(
    values={
        "username": "小黄",
        "user_input": "帮我写一首关于布偶猫的七言绝句",
    },
    next=("router_node",),
    config={
        "configurable": {
            "thread_id": "123",
            "checkpoint_ns": "",
            "checkpoint_id": "1f1655e8-6871-6073-8000-ddcbe680da21",
        }
    },
    metadata={
        "source": "loop",
        "step": 0,
        "parents": {},
    },
    created_at="2026-06-11T06:26:48.352564+00:00",
    parent_config={
        "configurable": {
            "thread_id": "123",
            "checkpoint_ns": "",
            "checkpoint_id": "1f1655e8-686b-66c7-bfff-24e0e8d44f4c",
        }
    },
    tasks=(
        PregelTask(
            id="fb88436e-b163-6308-6341-7d8368e041f9",
            name="router_node",
            path=("__pregel_pull", "router_node"),
            error=None,
            interrupts=(),
            state=None,
            result={
                "topic": "布偶猫",
                "mode": "poem",
            },
        ),
    ),
    interrupts=(),
)

该检查点表示：

用户输入已经写入状态，下一步将要执行 router_node。

从这个检查点分叉，本文采用两种方案（还可以有别的方案）：

1. 修改输入，让 router_node 重新执行。见 6.5.4.2.3. 节
2. 直接伪造 router_node 的输出，从而跳过 router_node。见 6.5.4.2.4. 节

6.5.4.2.3. 只改状态

6.5.4.2.3.1. 更新配置

示例如下

change_input_config = graph.update_state(
    config=before_router_checkpoint.config,
    values={"user_input": "帮我写一个关于布偶猫的笑话"},
    as_node=START
)
change_input_config

这里把 as_node 设置为 START，表示：

把这次状态更新视为 START 的输出。

因此，后续运行时会从 START 的后继节点继续推进，也就是重新执行 router_node。

上述代码运行结果如下

{
    "configurable": {
        "thread_id": "123",
        "checkpoint_ns": "",
        "checkpoint_id": "1f1655f8-86fb-6dd1-8001-9e7808dfaf1a",
    }
}

执行 update_state() 后会返回一个新的配置，其中包含新的 checkpoint_id。这个新的 checkpoint_id 不同于原有历史检查点，说明 update_state() 创建了一个新的检查点分支。

需要注意：

update_state() 返回的是新检查点的配置，不是新状态本身。

新的状态已经被记录在检查点后端，运行时生效。可以执行以下代码查看新建的检查点快照：

graph.get_state(change_input_config)

此处不再演示。

6.5.4.2.3.2. 重新运行

示例如下

graph.invoke(None, change_input_config)

运行结果如下

2026-06-11 14:13:44.267 | INFO     | __main__:router_node:31 - 路由节点已执行
2026-06-11 14:13:45.222 | INFO     | __main__:router_node:38 - 路由结果-res: {'topic': '布偶猫', 'mode': 'joke'}
2026-06-11 14:13:45.223 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:13:45.223 | INFO     | __main__:router:48 - 路由至 node_joke 节点
2026-06-11 14:13:45.225 | INFO     | __main__:node_joke:64 - node_joke 已执行
{
    "username": "小黄",
    "user_input": "帮我写一个关于布偶猫的笑话",
    "output": (
        "一只布偶猫照镜子，问：“魔镜魔镜，谁是世界上最美的猫？”  \n"
        "镜子没说话。  \n"
        "猫急了：“快说呀！不然我卖萌把你甜碎。”  \n"
        "镜子终于开口：“……我死机了，重启中。”"
    ),
}

由运行结果可知：

1. user_input 已经从“帮我写一首关于布偶猫的七言绝句”变成“帮我写一个关于布偶猫的笑话”。而 username 没有被更新，仍然保持原值 "小黄"。说明 values 是 状态更新 而非完整的 状态替换。
2. router_node 开始的节点全部重新运行。由于新的输入被识别为笑话任务，因此后续进入 node_joke。

6.5.4.2.4. 改状态且跳过某些节点

6.5.4.2.4.1. 更新配置

如果不希望重新执行 router_node，而是直接指定 router_node 的输出，可以把 as_node 设置为 "router_node"。

示例如下

skip_router_config = graph.update_state(
    config=before_router_checkpoint.config,
    values={"topic": "狸花猫", "mode": "笑话"},
    as_node="router_node"
)
skip_router_config

1. 此处把 {"topic": "狸花猫", "mode": "笑话"} 当作 router_node 的输出。

2. 因此，后续运行时，router_node 本身不会被重新执行，从它的后继节点继续推进。

3. 由于 router_node 挂载了条件边：

python
builder.add_conditional_edges(
"router_node",
router,
path_map=["node_poem", "node_joke", "node_default"]
)

所以当 update_state() 把 values 视为 router_node 的输出时，条件路由函数 router 也会根据这次更新后的状态计算后续分支。

这里把 mode 设置为 "笑话"，router 函数命中兜底分支 node_default。

运行结果如下

2026-06-11 14:14:02.223 | INFO     | __main__:router:43 - 路由函数已执行
2026-06-11 14:14:02.223 | INFO     | __main__:router:51 - 路由至兜底节点
{
    "configurable": {
        "thread_id": "123",
        "checkpoint_ns": "",
        "checkpoint_id": "1f1655cb-de15-6c4e-8001-ca03b011e993",
    }
}

可以看到：

1. update_state() 创建了一个新的检查点，返回的 skip_router_config 指向这个新检查点。

2. 日志中出现了：

text
路由函数已执行
路由至兜底节点

这说明 router_node 的节点函数没有被重新执行，但与 router_node 条件边相关的路由函数被执行了。

3. update_state() 底层会根据 as_node 找到对应图节点的 writers。

对于 router_node，writers 包括状态写入器和条件边对应的路由逻辑。

4. 执行 update_state() 时，传入的 values 会被当作 router_node 的输出，交给该节点的 writers 处理。

状态写入器会把 values 转换为对状态 Channel 的写入；条件边对应的路由逻辑会根据当前状态计算后继分支，并生成类似 branch:to:node_default 这样的触发通道写入。

5. 随后，update_state() 会把这些写入应用到检查点中，并保存一个新的检查点。

因此，状态更新和路由结果已经记录在 skip_router_config 指向的新检查点中。后续再调用：

python
graph.invoke(None, skip_router_config)

时，运行图会基于这个新检查点继续执行，被触发的后继节点也会继续运行。

综上，graph.update_state() 的作用可以理解为：

不执行指定节点的节点函数本身，而是把传入的 values 当作该节点已经产生的输出，执行该节点对应的 writers，应用状态写入和可能存在的路由写入，然后生成一个新的检查点。

因此，在本例中可以看到：router_node 的节点逻辑没有被执行，但它挂载的路由函数被执行，状态更新也已经生效。

6.5.4.2.4.2. 重新运行

示例如下

graph.invoke(None, skip_router_config)

运行结果如下

2026-06-11 16:49:42.501 | INFO     | __main__:node_default:73 - node_default 已执行
{
  "username": "小黄",
  "user_input": "帮我写一首关于布偶猫的七言绝句",
  "output": "无法处理的任务类型"
}

和分析一致，触发了兜底分支。

这种方式常用于测试场景。例如：

• 跳过不稳定的大模型路由节点。
• 手动指定路由结果，测试不同分支。
• 修改历史状态，验证后续节点能否正常运行。
• 基于同一个历史检查点探索多条可能的执行路径。

不过在正式业务流程中，应谨慎使用这种方式。因为它是在伪造某个节点的输出，如果伪造的状态不符合后续节点的预期，可能导致逻辑混乱，甚至抛出运行时异常。

系列导航

上一篇：节点执行与容错机制 ｜ 下一篇：图记忆管理

系列目录：AI框架