Take To

Appearance

6. 持久化机制和可恢复执行（上）：检查点基础 ​

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。

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

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

6.1.2.2. 核心组件 ​

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

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

概念作用
State用户定义的图状态结构，负责节点间信息传递。LangGraph 会将 State 转换为底层 Channel；运行时真正负责节点间通信的是 Channel。
ChannelLangGraph 用于节点间通信的底层机制，节点从 Channel 读数据、向 Channel 写更新；既包括承载 State 字段更新的状态通道，也包括用于分支、任务调度等行为的内部通道。
Checkpoint检查点，超步边界上的底层状态快照，保存 Channel 的值、版本等信息。
CheckpointMetadata与 Checkpoint 关联的元数据，如超步编号 step、父检查点 ID parents 等。
Checkpointer检查点存储器，负责存储检查点、元数据、配置等信息。
thread此处的线程不同于操作系统的线程，是指 LangGraph 中一条逻辑上的、可持久化的执行线，也可以理解为会话。一个会话可以包含多次调用，并在执行过程中生成多个检查点。因此，要将多次调用组织在一个会话中，必须为该 thread 配置 checkpointer。
thread_idthread 的唯一标识，用于区分不同的会话。复用同一个 thread_id，就表示多次调用共享同一条持久化执行线，也就是共享同一个会话历史。此处的 thread/thread_id 和 LangChain Agent 中提到的 thread/thread_id 是同一概念。
checkpoint_ns检查点命名空间 namespace，用于区分父图和子图的 checkpoint。根图通常是空字符串 ""，子图会有自己的 namespace。这个字段在子图持久化时很重要，下文会有专门的篇幅讲解子图相关用法。
checkpoint_idCheckpoint 的唯一标识。
StateSnapshot开发者通过 get_state() / get_state_history() 看到的状态快照对象。它不是原始 Checkpoint，而是 LangGraph 基于检查点和运行时信息封装出来的开发者视图，包含当前状态值、下一步待执行节点、任务、config、metadata、中断信息等内容。

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. 使用场景 ​

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

场景说明
多轮会话同一个 thread_id 下的多次调用可以共享历史状态。
中断恢复图执行到人工确认等节点时可以中断，之后再继续。
失败恢复程序异常退出后，可以基于已经保存的检查点继续执行，避免重复计算。
Time Travel可以回到某个历史检查点，重放或分叉执行。

因此，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

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

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

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

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

类名包名检查点后端
InMemorySaverlanggraph-checkpoint内存（In-memory）
SqliteSaverlanggraph-checkpoint-sqliteSQLite
PostgresSaverlanggraph-checkpoint-postgresPostgreSQL
MongoDBSaverlanggraph-checkpoint-mongodbMongoDB
RedisSaverlanggraph-checkpoint-redisRedis

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

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

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

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

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

示例如下

python
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 的用法。此处直接给出案例，不再赘述。

示例如下

python
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 模式的基础上增加了响应延迟。

总结

模式写入时机性能开销响应延迟容灾能力
exit运行退出时写入低最低最弱
async每个超步末尾写入主检查点，任务完成后写入中间结果，后台异步写入高较高较强
sync和 async 区别在于，进入下一个超步之前等待当前主检查点的写入任务完成高最高最强

6.3.2. 示例 ​

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

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

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

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

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

示例如下

python
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 参数切换持久化模式：

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

运行结果如下

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

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

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

image-20260609151649967

6.4. 查看历史检查点 ​

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

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

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

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

6.4.1.1. 示例 ​

python
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. 运行日志
json
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。

  1. 拓扑结构

image-20260609173103640

  1. 最终结果和检查点列表
json
============================== -> 运行结果 <- ==============================
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. 检查点主要信息说明 ​

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

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

即

python
最新检查点 → 最旧检查点

其中：

  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 的检查点中：

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

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

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

python
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. 查看最新检查点 ​

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

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

完整示例如下

python
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. 拓扑结构

    image-20260610112506158

  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：

python
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. 获取历史检查点列表 ​

示例如下

python
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)

运行结果如下

json
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 已执行

image-20260611183125275

json
============================== -> 运行结果 <- ==============================
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() 查看该检查点。

示例如下

python
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)

运行结果如下

json
============================== -> 根据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()，从而便捷而精确地查看该检查点的快照。