从零开始学LangGraph（13）：Checkpoints——Agent短期记忆的基石 - 文章 - 开发者社区

• 并行运行的节点属于同一个super-step ：如果多个节点可以同时执行（比如它们都收到了消息，且彼此之间没有依赖），它们会在同一个super-step中并行运行
• 顺序运行的节点属于不同的super-step ：如果节点A执行完后，节点B才能执行，那么它们属于不同的super-step

举个例子，假设我们有一个Graph：START → node_a → node_b → END。那么：

• 第一个super-step：从START节点到node_a执行完成
• 第二个super-step：从node_a到node_b执行完成
• 第三个super-step：从node_b到END，Graph执行完成

Checkpoints的创建

Checkpoint是Graph在每个super-step执行后自动保存的State快照，它由StateSnapshot对象表示，包含了State的值（values）、下一个要执行的节点（next）、配置信息（config）和元数据（metadata）等关键信息。

换言之，Graph每完成一个super-step，就会自动创建一个Checkpoint来保存当前State的快照，这样我们就可以追踪State在整个执行过程中的变化。

让我们用一个简单的例子来看看这个过程：

  
from langgraph.graph import StateGraph, START, END  
from langgraph.checkpoint.memory import InMemorySaver  
from typing import TypedDict  
from operator import add  
from typing import Annotated  
  
class State(TypedDict):  
    foo: str  
    bar: Annotated[list[str], add]  
  
def node\_a(state: State):  
    return {"foo": "a", "bar": ["a"]}  
  
def node\_b(state: State):  
    return {"foo": "b", "bar": ["b"]}  
  
# 创建Graph  
workflow = StateGraph(State)  
workflow.add\_node("node\_a", node\_a)  
workflow.add\_node("node\_b", node\_b)  
workflow.add\_edge(START, "node\_a")  
workflow.add\_edge("node\_a", "node\_b")  
workflow.add\_edge("node\_b", END)  
  
# 添加checkpointer并编译  
checkpointer = InMemorySaver()  
graph = workflow.compile(checkpointer=checkpointer)  
  
# 执行Graph  
config = {"configurable": {"thread\_id": "1"}}  
graph.invoke({"foo": "", "bar": []}, config=config)

执行完这段代码后，Graph会创建4个checkpoints，对应四种state：

初始checkpoint ：空的state， START 节点准备执行

输入checkpoint ：用户输入 {'foo': '', 'bar': []} ， node\_a 准备执行

node_a执行后 ：state变为 {'foo': 'a', 'bar': ['a']} ， node\_b 准备执行

node_b执行后 ：state变为 {'foo': 'b', 'bar': ['a', 'b']} ，没有下一个节点了

在上面的例子中，我们看到了两个关键概念：InMemorySaver()和thread\_id。这两个概念是Checkpoints功能得以实现的基础，下面我们来详细了解一下。

Checkpointer和Threads：实现Checkpoints的基础

Checkpointer：保存checkpoints的工具

Checkpointer是负责保存和管理checkpoints的工具。LangGraph提供了多种checkpointer实现，最简单的是InMemorySaver，它会把checkpoints保存在内存中。

  
from langgraph.checkpoint.memory import InMemorySaver  
  
checkpointer = InMemorySaver()

这行代码创建了一个内存checkpointer。需要注意的是，由于checkpoints保存在内存中，程序重启后这些数据就会丢失，所以不适合生产环境使用。

当然，LangGraph还支持其他checkpointer，比如可以保存到SQLite数据库的SqliteSaver，以及适合生产环境使用的PostgresSaver等高级工具，我们这里先用最简单的InMemorySaver来学习。

Threads：checkpoints的容器

想象一下，如果你同时与多个用户进行对话，每个对话都有自己独立的会话记录。在LangGraph中，Thread 就是这样的"独立对话会话"。

每个Thread都有一个唯一的ID（thread\_id），用来标识不同的执行会话。当我们使用checkpointer时，必须指定一个thread\_id，这样checkpointer才知道要把checkpoints保存到哪里，以及从哪里读取。

  
config = {"configurable": {"thread\_id": "1"}}  
graph.invoke(input\_data, config=config)

比如这里我们指定了thread\_id为"1"，那么这次执行产生的所有checkpoints都会保存在这个thread下。

Checkpoints的应用场景

接下来看看我们能用Checkpoints做些什么。

1. 查看当前状态（Get State）

有时候，我们想知道Graph执行到哪一步了，当前的状态是什么。这时候就可以使用graph.get\_state(config)方法。

  
# 获取最新的checkpoint状态  
config = {"configurable": {"thread\_id": "1"}}  
state\_snapshot = graph.get\_state(config)  
print(state\_snapshot)

这会返回一个StateSnapshot对象，包含了当前checkpoint的所有信息：

  
StateSnapshot(  
    values={'foo': 'b', 'bar': ['a', 'b']},  # 当前state的值  
    next=(),  # 下一个要执行的节点（空表示执行完成）  
    config={'configurable': {'thread\_id': '1', 'checkpoint\_id': '...'}},  # checkpoint的配置  
    metadata={'source': 'loop', 'writes': {...}, 'step': 2},  # 元数据  
    created\_at='2024-08-29T19:19:38.821749+00:00',  # 创建时间  
    ...  
)

我们也可以查看指定checkpoint的状态，只需要在config中加上checkpoint\_id：

  
# 获取指定checkpoint的状态  
config = {  
    "configurable": {  
        "thread\_id": "1",  
        "checkpoint\_id": "1ef663ba-28fe-6528-8002-5a559208592c"  
    }  
}  
state\_snapshot = graph.get\_state(config)

2. 查看执行历史（Get State History）

如果我们想看看Graph的完整执行过程，可以使用graph.get\_state\_history(config)方法。这会返回所有checkpoints的列表，按照时间倒序排列（最新的在前）。

  
config = {"configurable": {"thread\_id": "1"}}  
history = list(graph.get\_state\_history(config))  
  
# 遍历所有checkpoints  
for i, checkpoint in enumerate(history):  
    print(f"Checkpoint {i}:")  
    print(f"  Values: {checkpoint.values}")  
    print(f"  Next: {checkpoint.next}")  
    print(f"  Step: {checkpoint.metadata.get('step', 'N/A')}")  
    print()

输出可能是这样的：

  
Checkpoint 0:  
  Values: {'foo': 'b', 'bar': ['a', 'b']}  
  Next: ()  
  Step: 2  
  
Checkpoint 1:  
  Values: {'foo': 'a', 'bar': ['a']}  
  Next: ('node\_b',)  
  Step: 1  
  
Checkpoint 2:  
  Values: {'foo': '', 'bar': []}  
  Next: ('node\_a',)  
  Step: 0  
  
Checkpoint 3:  
  Values: {'bar': []}  
  Next: ('\_\_start\_\_',)  
  Step: -1

这样，我们就可以清楚地看到Graph执行的完整时间线了。

3. 时间旅行：从checkpoint重放（Replay）

这是checkpoints最酷的功能之一：时间旅行 ！

想象一下，你正在看一个视频，突然想回到某个时间点重新看。在LangGraph中，我们可以从任意一个checkpoint重新开始执行，这就是Replay（重放）。

  
# 从指定checkpoint重新执行  
config = {  
    "configurable": {  
        "thread\_id": "1",  
        "checkpoint\_id": "1ef663ba-28f9-6ec4-8001-31981c2c39f8"  # node\_a执行后的checkpoint  
    }  
}  
graph.invoke(None, config=config)

这里有几个要点：

已执行的步骤不会重新执行 ：LangGraph会智能地判断哪些步骤已经执行过了，不会重复执行。只有checkpoint之后的步骤才会执行。

可以创建新的分支 ：从某个checkpoint重新执行，实际上是在创建一个新的执行分支。原来的执行历史还在，新的执行会从指定的checkpoint开始。

实际应用 ：

• 调试：如果Graph执行出错了，可以从出错前的checkpoint重新执行，看看问题出在哪里
• 实验：想试试不同的参数或逻辑？从某个checkpoint重新执行，尝试不同的路径
• A/B测试 ：从同一个checkpoint开始，执行不同的逻辑，比较结果

4. 修改历史：更新checkpoint状态（Update State）

除了重放，我们还可以直接修改checkpoint的状态，然后重新执行。这就像是修改游戏存档，然后从修改后的存档继续游戏。

update\_state方法接受三个主要参数：

（1）config参数：指定要更新的checkpoint

config参数中必须包含thread\_id，用来指定要更新哪个thread的状态。如果只传thread\_id，会更新最新的checkpoint；如果还传了checkpoint\_id，会更新指定的checkpoint（这会创建一个新的分支，即从该checkpoint开始一个新的执行路径，原有的执行历史保持不变）。

  
# 更新最新checkpoint  
graph.update\_state(  
    config={"configurable": {"thread\_id": "1"}},  
    values={"foo": "modified"}  
)  
  
# 更新指定checkpoint（创建分支）  
graph.update\_state(  
    config={  
        "configurable": {  
            "thread\_id": "1",  
            "checkpoint\_id": "1ef663ba-28f9-6ec4-8001-31981c2c39f8"  
        }  
    },  
    values={"foo": "modified"}  
)

（2）values参数：要更新的状态值

values参数是要更新的状态值。这个更新会遵循State中定义的Reducer规则，而不是简单地覆盖所有值。

举个例子，假设我们的State定义是：

  
class State(TypedDict):  
    foo: int  # 没有reducer，会被覆盖  
    bar: Annotated[list[str], add]  # 有reducer，会累积

当前状态是{"foo": 1, "bar": ["a"]}，如果我们执行：

  
graph.update\_state(  
    config={"configurable": {"thread\_id": "1"}},  
    values={"foo": 2, "bar": ["b"]}  
)

结果会是：

• foo ：被完全替换为 2 （因为没有reducer）
• bar ：会变成 ["a", "b"] （因为有 add reducer，会累积）

（3）as_node参数：指定更新的来源节点

as\_node参数可以指定这次更新"假装"是来自哪个节点的。这会影响下一步执行哪个节点，因为Graph会根据最后一个更新State的节点来决定下一步的执行路径。

  
graph.update\_state(  
    config={"configurable": {"thread\_id": "1"}},  
    values={"foo": "modified"},  
    as\_node="node\_a"  # 指定更新来自node\_a  
)

如果不指定as\_node，LangGraph会尝试自动推断，但有时候可能会不明确，所以最好明确指定。

完整实战示例

好了，理论讲得差不多了，让我们用一个完整的例子来演示一下checkpoints的各种用法。这个例子模拟了一个简单的游戏流程，我们会展示如何构建带checkpointer的Graph，以及如何使用checkpoints的各种功能。

1. 构建Graph

首先，我们需要定义State、创建Node函数，然后构建Graph并添加checkpointer。这段代码定义了一个游戏状态的State，包含玩家名称、等级、血量和动作历史，然后创建了三个Node函数分别处理游戏开始、升级和受到伤害的逻辑。

  
from langgraph.graph import StateGraph, START, END  
from langgraph.checkpoint.memory import InMemorySaver  
from typing import TypedDict  
from typing import Annotated  
from operator import add  
  
# 定义State  
class GameState(TypedDict):  
    player\_name: str  
    level: int  
    hp: int  
    actions: Annotated[list[str], add]  
  
# 定义Node函数  
def start\_game(state: GameState):  
    """开始游戏"""  
    print(f"欢迎，{state['player\_name']}！游戏开始！")  
    return {"level": 1, "hp": 100, "actions": ["游戏开始"]}  
  
def level\_up(state: GameState):  
    """升级"""  
    new\_level = state["level"] + 1  
    print(f"恭喜！你升级到了 {new\_level} 级！")  
    return {"level": new\_level, "actions": [f"升级到{new\_level}级"]}  
  
def take\_damage(state: GameState):  
    """受到伤害"""  
    damage = 20  
    new\_hp = state["hp"] - damage  
    print(f"你受到了 {damage} 点伤害！当前HP: {new\_hp}")  
    return {"hp": new\_hp, "actions": [f"受到{damage}点伤害"]}  
  
# 创建Graph  
workflow = StateGraph(GameState)  
workflow.add\_node("start\_game", start\_game)  
workflow.add\_node("level\_up", level\_up)  
workflow.add\_node("take\_damage", take\_damage)  
  
workflow.add\_edge(START, "start\_game")  
workflow.add\_edge("start\_game", "level\_up")  
workflow.add\_edge("level\_up", "take\_damage")  
workflow.add\_edge("take\_damage", END)  
  
# 添加checkpointer并编译  
checkpointer = InMemorySaver()  
graph = workflow.compile(checkpointer=checkpointer)

这段代码的整体逻辑如下：

首先，我们定义了一个GameState，包含四个字段：player\_name（玩家名称）、level（等级）、hp（血量）和actions（动作历史）。其中actions字段使用了Annotated[list[str], add]，这意味着它使用了add作为Reducer，多个节点返回的actions列表会被合并而不是覆盖。

然后，我们创建了三个Node函数：

• start\_game ：初始化游戏状态，将等级设为1，HP设为100，并记录"游戏开始"动作
• level\_up ：处理升级逻辑，将等级加1，并记录升级动作。由于 actions 字段有Reducer，这个动作会被追加到现有的动作列表中
• take\_damage ：处理受到伤害的逻辑，将HP减去20点伤害，并记录受到伤害的动作

接着，我们创建了Graph，将三个Node按顺序连接：START → start_game → level_up → take_damage → END，形成一个线性的执行流程。

最后，我们创建了一个InMemorySaver作为checkpointer，并在编译Graph时传入它。这样，Graph就会在每个super-step执行后自动保存checkpoint，记录State的完整信息。

2. 执行Graph并查看当前状态

执行Graph后，我们可以使用get\_state()方法查看最新的checkpoint状态。

  
# 执行Graph  
config = {"configurable": {"thread\_id": "game-001"}}  
result = graph.invoke(  
    {"player\_name": "勇者", "level": 0, "hp": 0, "actions": []},  
    config=config  
)

这段代码执行Graph：

• config ：配置字典，必须包含 thread\_id ，这里我们使用 "game-001" 作为thread标识
• invoke() ：执行Graph，第一个参数是初始State，第二个参数是config。执行完成后，Graph会在每个super-step自动创建checkpoint

执行效果 ：Graph会依次执行start\_game、level\_up、take\_damage三个节点，控制台会输出：

  
欢迎，勇者！游戏开始！  
恭喜！你升级到了 2 级！  
你受到了 20 点伤害！当前HP: 80

最终返回的result包含了执行完成后的State值。

  
print("\n=== 执行完成 ===")  
print(f"最终状态: {result}")  
  
# 查看当前状态  
print("\n=== 当前状态 ===")  
current\_state = graph.get\_state(config)  
print(f"当前HP: {current\_state.values['hp']}")  
print(f"当前等级: {current\_state.values['level']}")  
print(f"执行步骤: {current\_state.metadata.get('step', 'N/A')}")

这段代码查看当前状态：

• graph.get\_state(config) ：获取最新的checkpoint状态，返回一个 StateSnapshot 对象
• current\_state.values ：访问State的值，这是一个字典，包含所有State字段的值
• current\_state.metadata ：访问checkpoint的元数据， step 字段表示执行到第几步
• 打印出当前HP、等级和执行步骤，让我们了解Graph执行后的最终状态

执行效果 ：控制台会输出类似以下内容：

  
=== 执行完成 ===  
最终状态: {'player\_name': '勇者', 'level': 2, 'hp': 80, 'actions': ['游戏开始', '升级到2级', '受到20点伤害']}  
  
=== 当前状态 ===  
当前HP: 80  
当前等级: 2  
执行步骤: 2

3. 查看执行历史

使用get\_state\_history()方法可以获取Graph执行的完整历史，查看所有checkpoints。

  
# 查看执行历史  
print("\n=== 执行历史 ===")  
history = list(graph.get\_state\_history(config))

get\_state\_history(config)返回一个迭代器，包含该thread下的所有checkpoints，按照时间倒序排列（最新的在前）。我们使用list()将其转换为列表，方便遍历。

  
for i, checkpoint in enumerate(history):  
    step = checkpoint.metadata.get('step', 'N/A')  
    hp = checkpoint.values.get('hp', 'N/A')  
    level = checkpoint.values.get('level', 'N/A')  
    print(f"步骤 {step}: HP={hp}, 等级={level}")

这段代码遍历所有checkpoints，打印出每一步的State值：

• checkpoint.metadata.get('step', 'N/A') ：获取执行步骤，如果不存在则返回'N/A'
• checkpoint.values.get('hp', 'N/A') ：获取该checkpoint的HP值，如果不存在则返回'N/A'
• 打印出每一步的步骤号、HP和等级，让我们能够清楚地看到State的变化过程

执行效果 ：控制台会输出类似以下内容，展示Graph执行的完整时间线：

  
=== 执行历史 ===  
步骤 2: HP=80, 等级=2  
步骤 1: HP=100, 等级=2  
步骤 0: HP=100, 等级=1  
步骤 -1: HP=N/A, 等级=N/A

可以看到State从初始状态到最终状态的完整变化过程。

4. 从checkpoint重放

我们可以从任意一个checkpoint重新开始执行，实现State的"时间旅行"。

  
# 找到level\_up执行后的checkpoint  
level\_up\_checkpoint = None  
for checkpoint in history:  
    if checkpoint.values.get('level') == 2:  # 升级后的等级  
        level\_up\_checkpoint = checkpoint  
        break

这段代码在历史记录中查找升级后的checkpoint：

• 遍历 history 列表，查找 level 等于2的checkpoint（因为 start\_game 将等级设为1， level\_up 将其加1变成2）
• 找到后保存到 level\_up\_checkpoint 变量中

执行效果 ：如果找到了符合条件的checkpoint，level\_up\_checkpoint会被赋值；否则为None。

  
# 从某个checkpoint重放  
print("\n=== 从升级后的checkpoint重放 ===")  
  
if level\_up\_checkpoint:  
    replay\_config = {  
        "configurable": {  
            "thread\_id": "game-001",  
            "checkpoint\_id": level\_up\_checkpoint.config["configurable"]["checkpoint\_id"]  
        }  
    }  
  
graph.invoke(None, config=replay\_config)

这段代码从找到的checkpoint重新执行：

• replay\_config ：创建重放配置，包含 thread\_id 和 checkpoint\_id 。 checkpoint\_id 从找到的checkpoint的config中获取
• graph.invoke(None, config=replay\_config) ：从指定的checkpoint重新执行Graph。第一个参数传入 None ，因为State已经保存在checkpoint中了
• 这样Graph会从升级后的checkpoint继续执行，而不是从头开始

执行效果 ：控制台会输出：

  
=== 从升级后的checkpoint重放 ===  
  
{'player\_name': '勇者', 'level': 2, 'hp': 80, 'actions': ['游戏开始', '升级到2级', '受到20点伤害']}

从最后输出的state可以看到，actions包含了前两个节点的内容，这意味着Graph跳过了start\_game和level\_up节点（因为它们已经在checkpoint之前执行过了），直接从take\_damage节点开始执行。

5. 修改状态

使用update\_state()方法可以修改checkpoint的State值，然后继续执行。

  
# 修改状态  
print("\n=== 修改状态 ===")  
graph.update\_state(  
    config={"configurable": {"thread\_id": "game-001"}},  
    values={"hp": 150},  # 恢复HP到150  
    as\_node="start\_game"  
)

这段代码修改State：

• config ：指定要更新哪个thread的状态，这里只传了 thread\_id ，表示更新最新的checkpoint
• values ：要更新的State值，这里将 hp 改为150
• as\_node ：指定这次更新"假装"是来自 start\_game 节点的，这会影响下一步执行哪个节点
• 注意：由于 hp 字段没有Reducer，所以会被直接覆盖为150

执行效果 ：State中的hp值会被更新为150，checkpoint状态被修改。

  
updated\_state = graph.get\_state(config)  
print(f"修改后的HP: {updated\_state.values['hp']}")

这段代码查看修改后的State：

• graph.get\_state(config) ：获取更新后的最新checkpoint状态
• 打印修改后的HP值，验证State是否成功更新

执行效果 ：控制台会输出：

  
=== 修改状态 ===  
修改后的HP: 150

可以看到HP已经从80恢复到了150，State更新成功。

这个例子完整展示了checkpoints的各种用法：如何创建带checkpointer的Graph，如何查看当前状态和执行历史，如何从checkpoint重放，以及如何修改状态。大家可以运行一下，看看效果！

总结

今天我们一起学习了LangGraph中的Checkpoints，它就像是Graph的"存档点"，让我们可以：

• 记录历史 ：保存Graph执行过程中的每一步状态
• 查看状态 ：随时查看当前或历史状态
• 时间旅行 ：从任意checkpoint重新执行
• 修改历史 ：修改checkpoint状态，创建新的执行分支

这些功能在实践中可以用来处理：

• 人机交互 ：保存对话历史，支持多轮对话
• 故障恢复 ：出错时从最近的checkpoint恢复
• 调试分析 ：查看执行历史，找出问题所在
• 实验测试 ：从同一个checkpoint尝试不同的执行路径

希望大家能够理解checkpoints的概念，并在实际项目中灵活运用。后面，我们会进一步探索一些更高级的功能，比如Interrupts和Human-in-the-loop，这些功能都依赖于checkpoints。

好了，以上就是本期的全部内容，如果大家觉得对自己有帮助的，还请多多点赞、收藏、转发、关注！祝大家玩得开心，我们下期再见！

—— END——

往期精华：

1.COZE教程

零基础搞定！萌新的 Coze 开源版保姆级本地部署指南

AI工作流编排手把手指南之一：Coze智能体的创建与基本设置

AI工作流编排手把手指南之二：Coze智能体的插件添加与调用

AI工作流编排手把手指南之三：Coze智能体的工作流

Agent | 工作流编排指南4：萌新友好的Coze选择器节点原理及配置教程

Agent | 工作流编排指南5：长文扩写自由 — Coze循环节点用法详解

Coze工作流编排指南6：聊天陪伴类智能体基本工作流详解-快来和玛奇玛小姐姐谈心吧~

PPT自由！Coze工作流 X iSlide插件-小白也能看懂的节点参数配置原理详解

2.MCP探索

Excel-MCP应用 | 自动提取图片数据到Excel的极简工作流手把手教程

markitdown-mcp联动Obsidian-mcp | 一个极简知识管理工作流

【15合1神器】不会代码也能做高级图表！这个MCP工具让我工作效率翻了不止三倍！

【效率翻倍】Obsidian自动待办清单实现：MCP联动Prompt保姆级教程（萌新3分钟上手）

萌新靠MCP实现RPA、爬虫自由？playwright-mcp实操案例分享！

高德、彩云MCP全体验：让Cherry Studio化身私人小助理的喂饭版指南！

3.Prompt设计

干货分享 | Prompt设计心法 - 如何3步做到清晰表达需求？

打工人看了流泪的Prompt设计原理，如何用老板思维让AI一次听懂需求？

不会Prompt还敢说自己会用DeepSeek？别怕！10分钟让你成为提示大神！