首页 / 文章 / 深入了解 LangGraph 的 InMemorySaver：检查点、写入操作与二进制数据如何协同工作

LangGraphLangChainAI AgentsPythonLLM

深入了解 LangGraph 的 InMemorySaver：检查点、写入操作与二进制数据如何协同工作

遍历存储中的字典和二进制数据，将其写入LangGraph的InMemorySaver中，并追踪一个小型图运行过程如何转化为三个相互关联的检查点。

1834 词

LangGraph的InMemorySaver通常只需一行代码即可配置完成：将其传递给compile()后，对话就能记住状态，之后就无需再操心。然而，它处理数据的方式恰恰揭示了LangGraph的许多核心机制，包括如何实现恢复、时间回溯和容错功能，以及为什么持久化检查点工具会采用现有的设计。通过追踪经过该保存器内部字典处理的极简图结构，你便能读懂检查点文件的内容，并准确理解每个条目的含义。

为何图结构需要检查点

检查点相当于图的短期记忆：它在执行过程中记录下图状态的快照。可以将其想象成剧情模式游戏中的存档点：如果没有存档点，想要重新玩第二关就必须先重玩第一关。存档记录了玩家的进度，这样即便游戏已经结束，也能从该节点继续游戏。LangGraph在每一步之后都会执行类似操作，因此线程可以从更早的节点继续或重新播放。

用于测试的最小图结构

下面的示例构建了最简单的实用图结构：一个包含name和address字段的类型化状态，一个通过Command来设置这两个字段的确定性节点，以及依次为START、get_address和END的边。该示例使用InMemorySaver和InMemoryStore来编译图结构，在线程“12345”上执行它，最后输出检查指针的属性。InMemoryStore是用于线程间共享长期数据的独立组件，在后续操作中不起作用。虽然这段代码被标记为JavaScript，但实际上它是Python：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph
from typing import TypedDict, Literal
from langgraph.types import Command
from langgraph.graph.state import START, END

# we create a checkpointer, for now testing purposes we use inmemory
checkpointer = InMemorySaver()

# we will talk about this in our next blog
store = InMemoryStore()


# how you want to store your graph state which is persisted across chats
class GraphState(TypedDict):
    name: str
    address: str

# this is a determinsitic node that is present as a node
def get_address(state: GraphState) -> Command[Literal[END]]:
    return Command(update={
        "name": "pavaneeshwar",
        "address": "Hyderabad residency"
    })

# intialize graph
graph = StateGraph(GraphState)

# add this node to the graph
graph.add_node("get_address", get_address)

# by default START and END defines the START execution and end execution
graph.add_edge(START, "get_address")
graph.add_edge("get_address", END)

# the above graph we created is START => get_address => END

# we load the entire graph, this returns an object which we can run
app = graph.compile(checkpointer=checkpointer, store=store)

app.invoke({}, config={"configurable": {"thread_id": "12345"}})

# we are interested here how langgraph stores checkpointer
app.checkpointer.__dict__

InMemorySaver的属性

列出检查指针字典中的键后可见有五个属性：

app.checkpointer.__dict__.keys()
# dict_keys(['serde', 'storage', 'writes', 'blobs', 'stack'])

serde：序列化与反序列化

检查点数据无法以实时 Python 对象的形式存储在数据库中，即便在内存中，保存器也会以序列化形式来存储它。serde 是用于将值转换为字节及从字节转换回值的序列化工具，它会为每个值添加类似 msgpack 这样的类型标签。

存储：每个线程的检查点

storage 用于存储检查点本身。每段对话都有一个线程 ID，LangGraph 就通过该 ID 来获取特定线程的历史记录。其结构为嵌套字典：首先是线程 ID，接着是检查点命名空间（顶层图使用空字符串；子图则有各自的命名空间），最后才是检查点 ID：

{
    "thread_id": {
         "namespace" : {
            "checkpoint_uuid_0": (msgpack, <binary_data>),
            "checkpoint_uuid_1": (msgpack,<binary_data>, checkpoint_uuid_0),
            "checkpoint_uuid_2": (msgpack,<binary_data>, checkpoint_uuid_1),
         }
    }
}

每个条目包含序列化后的检查点数据、其序列化元数据以及父检查点的ID。这个父指针将线程的各个检查点连接成一条链式历史记录，从而实现了回退和分支功能。

写入操作：每个检查点的待处理写入数

writes用于记录任务产生的各项更新。系统不会直接覆盖现有状态，而是将每次更新作为一条新条目进行记录，这些条目的键由线程、命名空间以及任务所运行的检查点决定。在每条记录中，每次写入都会通过任务ID和索引来标识：

{
    ('thread_id', 'namespace', 'checkpoint_uuid_1') : {
        ('operation_uuid_1', 0) : ('operation_uuid_1', 'channel_name', ('msgpack', '<binary data>')),
        ('operation_uuid_2', 1) : ('operation_uuid_2', 'channel_name', ('msgpack', '<binary data>'))
    }
}

此示例中的channel_name仅为占位符。当节点更新name时，通道名称即为name；当更新address时，通道名称则为address。如果节点同时更新这两个值，则会在同一个检查点下生成两条记录。由于写入操作会在任务完成后立即存储，因此在执行过程中出错的运行无需重新执行那些已经成功的任务。

blobs：带版本号的通道值

blobs会存储每个版本下各个通道的实际值。其键由线程、命名空间、通道及版本组合而成，因此检查点可以通过版本来引用通道值，而无需嵌入副本：

{
    ('thread_id', 'namespace', 'channel_name', 'version') : ('mssgpack', '<binary data>')
}

stack：上下文管理

stack属性有时被描述为待处理任务的队列，但在保存器的实现中，它实际上是一个上下文管理器栈（即ExitStack），用于在以上下文管理器形式进入和退出保存器时管理资源。它并不存储图执行状态。这些属于内部实现细节，因此请根据您安装的版本进行核对。

逐步追踪运行过程

一次图的计算会生成三个检查点。

检查点1：输入数据到达

第一个检查点的ID为1f1b054e-b2a5-660a-bfff-7484776ebce0，其中包含两个msgpack格式的数据：检查点本身及其元数据。

// First Message pack
{
  "v": 4,
  "ts": "2026-09-14T15:56:59.435773+00:00",
  "id": "1f1b054e-b2a5-660a-bfff-7484776ebce0",
  "channel_versions": {
    "__start__": "00000000000000000000000000000001.0.267464090313665"
  },
  "versions_seen": {
    "__input__": {}
  },
  "updated_channels": [
    "__start__"
  ]
}

// Second Message Pack, this is just meta data

{
  "source": "input",
  "step": -1,
  "parents": {}
}

此时仅存在 __start__ 通道。它已收到首个版本，updated_channels 列出了该版本，元数据将来源标记为 input，且 step 值设为 -1，表示这是任何图处理步骤执行之前的状态。版本字符串遵循简单的格式：一个补零后的、单调递增的计数器，后跟一个用于确保版本唯一性的随机小数。

检查点通过版本的标识来引用通道的值，而对应的二进制数据块则存储实际数据。此处的输入是一个空字典，msgpack 将其编码为单个字节 \x80：

// this msgpack basically {}
('12345', '', '__start__', '00000000000000000000000000000001.0.267464090313665'): ('msgpack', b'\x80')

检查点 2：路由至节点

第二个检查点 1f1b054e-b2a6-6294-8000-96e3a3cb81ac 记录了从 START 到 get_address 的路径。这涉及路由规划，尚未开始执行节点功能：

// first message pack
{
  "v": 4,
  "ts": "2026-09-14T15:56:59.436094+00:00",
  "id": "1f1b054e-b2a6-6294-8000-96e3a3cb81ac",
  "channel_versions": {
    "__start__": "00000000000000000000000000000002.0.27282425125643517",
    "branch:to:get_address": "00000000000000000000000000000002.0.27282425125643517"
  },
  "versions_seen": {
    "__input__": {},
    "__start__": {
      "__start__": "00000000000000000000000000000001.0.267464090313665"
    }
  },
  "updated_channels": [
    "branch:to:get_address"
  ]
}

// second message pack
{
  "source": "loop",
  "step": 0,
  "parents": {}
}

目前有两个通道用于传输版本2的数据。由于 __start__ 的输入已被使用，它便切换到新版本，同时一个新的通道 branch:to:get_address 会指示接下来应执行 get_address。versions_seen 显示 __start__ 任务已经处理过 __start__ 通道的版本1；LangGraph正是通过这种记录机制来判断哪些节点仍需执行。元数据会切换到源 loop，其 step 值为0。

触发此次切换的写入操作会被存储在之前的检查点ID下，因为它是由从该检查点开始运行的任务生成的：

('12345', '', '1f1b054e-b2a5-660a-bfff-7484776ebce0'): {
        ('4efa087d-283c-eb5c-478a-97c592eb3802', 0): ('4efa087d-283c-eb5c-478a-97c592eb3802', 'branch:to:get_address', ('null', b''), '~__pregel_pull, __start__')
 }

同时还会创建两个新的数据块。__start__数据块被标记为empty，表明该通道在内容被读取后被清空了；而分支通道则存储null值，因为它仅起到触发作用：

// one created for progressing start
('12345', '', '__start__', '00000000000000000000000000000002.0.27282425125643517'): ('empty', b''),

// one for creating branch
('12345', '', 'branch:to:get_address', '00000000000000000000000000000002.0.27282425125643517'): ('null', b'')

检查点3：节点状态更新

第三个检查点1f1b054e-b2a6-6d66-8001-d006da4d6d19记录了get_address函数的执行过程以及其对name和address字段的更新情况：

// first message pack
{
  "v": 4,
  "ts": "2026-09-14T15:56:59.436372+00:00",
  "id": "1f1b054e-b2a6-6d66-8001-d006da4d6d19",
  "channel_versions": {
    "__start__": "00000000000000000000000000000002.0.27282425125643517",
    "branch:to:get_address": "00000000000000000000000000000003.0.07103778333502464",
    "name": "00000000000000000000000000000003.0.07103778333502464",
    "address": "00000000000000000000000000000003.0.07103778333502464"
  },
  "versions_seen": {
    "__input__": {},
    "__start__": {
      "__start__": "00000000000000000000000000000001.0.267464090313665"
    },
    "get_address": {
      "branch:to:get_address": "00000000000000000000000000000002.0.27282425125643517"
    }
  },
  "updated_channels": [
    "address",
    "name"
  ]
}

// second message pack
{
  "source": "loop",
  "step": 1,
  "parents": {}
}

channel_versions始终保存着每个频道的最新版本，而versions_seen则记录了每个节点运行时所看到的版本信息。__start__的版本始终为2，因为之后再也没有任何操作会修改它。分支频道和两个状态频道则升级到版本3，updated_channels中列出了address和name，同时步进计数器变为1。

该节点写了两个值，因此在第二个检查点的ID下会显示两次写入记录，每个频道各一次，它们共享同一个任务ID：

('12345', '', '1f1b054e-b2a6-6294-8000-96e3a3cb81ac'): {
        ('a6b6f3e8-32e4-88a4-559d-cd6d409c7910', 0): ('a6b6f3e8-32e4-88a4-559d-cd6d409c7910', 'name', ('msgpack', b'\xacpavaneeshwar'), '~__pregel_pull, get_address'),
        ('a6b6f3e8-32e4-88a4-559d-cd6d409c7910', 1): ('a6b6f3e8-32e4-88a4-559d-cd6d409c7910', 'address', ('msgpack', b'\xb3Hyderabad residency'), '~__pregel_pull, get_address')
}

最后，新的数据块中存储了这两个状态字段的msgpack编码字符串：

('12345', '', 'name', '00000000000000000000000000000003.0.07103778333502464'): ('msgpack', b'\xacpavaneeshwar'),
('12345', '', 'address', '00000000000000000000000000000003.0.07103778333502464'): ('msgpack', b'\xb3Hyderabad residency')

为何采用这种布局设计

对于只有一个节点的图结构来说，使用三个字典似乎有些过度，但实际上每个部分都有其存在的必要：

  • 与父节点关联的检查点能为每个线程提供完整的历史记录。你可以查看任何过去的状态，从中恢复，或基于该状态创建新的分支。
  • 版本化的数据块会在每次变更时仅存储一次对应通道的值，因此即使状态很大且大部分内容未变，检查点体积依然保持较小。
  • 待处理的写入操作使得各步骤可恢复执行。如果某一步中的某个任务失败，那些已成功完成的任务的写入内容早已被保存，无需再次运行。

像 Postgres 这样的持久化检查点工具会将检查点、数据块及写入操作分别存储在反映这些结构的独立表中，因此同样的逻辑模型也适用于你的数据库。

核心要点

  • InMemorySaver适用于开发和测试环境；进程退出后其存储的数据会消失。
  • storage用于存储每个线程和命名空间下的检查点及元数据，这些数据通过父ID相互关联。
  • writes用于存储按任务生成的更新内容，这些更新以生成它们的检查点作为键值。
  • blobs按版本存储通道值，因此未被修改的通道不会被复制。

    • 相关阅读

    全部文章发送简报