Skip to main content
LangGraph 具有内置的持久化层,可将图状态保存为检查点。当您使用检查点保存器编译图时,图状态的快照会在执行的每一步保存,并组织成线程。这支持人机协同工作流、对话记忆、时间旅行调试和容错执行。 检查点
Agent Server 自动处理检查点 使用 Agent Server 时,您无需手动实现或配置检查点保存器。服务器会在后台为您处理所有持久化基础设施。

为何使用持久化

持久化是以下功能所必需的:
  • 人机协同:检查点保存器通过允许人类检查、中断和批准图步骤,促进人机协同工作流。这些工作流需要检查点保存器,因为人类必须能够随时查看图的状态,并且图必须能够在人类对状态进行任何更新后恢复执行。有关示例,请参阅中断
  • 记忆:检查点保存器允许在交互之间保留”记忆”。在重复的人类交互(如对话)情况下,任何后续消息都可以发送到该线程,该线程将保留其对先前交互的记忆。有关如何使用检查点保存器添加和管理对话记忆的信息,请参阅添加记忆
  • 时间旅行:检查点保存器允许进行”时间旅行”,使用户能够重放先前的图执行以审查和/或调试特定的图步骤。此外,检查点保存器使得可以在任意检查点分叉图状态以探索替代路径。
  • 容错性:检查点提供容错性和错误恢复:如果一个或多个节点在给定的超级步骤中失败,您可以从最后一个成功步骤重新启动图。
  • 待处理写入:当图节点在给定的超级步骤中执行失败时,LangGraph 会存储在该超级步骤中成功完成的任何其他节点的待处理检查点写入。当您从该超级步骤恢复图执行时,不会重新运行成功的节点。

核心概念

线程

线程是检查点保存器保存的每个检查点分配的唯一 ID 或线程标识符。它包含一系列运行的累积状态。当运行执行时,助手底层图的状态将持久化到线程中。 使用检查点保存器调用图时,您必须在配置的 configurable 部分指定 thread_id
{"configurable": {"thread_id": "1"}}
可以检索线程的当前和历史状态。要持久化状态,必须在执行运行之前创建线程。LangSmith API 提供了多个端点来创建和管理线程及线程状态。有关更多详细信息,请参阅 API 参考 检查点保存器使用 thread_id 作为存储和检索检查点的主键。没有它,检查点保存器无法保存状态或在中断后恢复执行,因为检查点保存器使用 thread_id 来加载保存的状态。

检查点

线程在特定时间点的状态称为检查点。检查点是在每个超级步骤保存的图状态快照,由 StateSnapshot 对象表示(有关完整字段参考,请参阅 StateSnapshot 字段)。

超级步骤

LangGraph 在每个超级步骤边界创建检查点。超级步骤是图的单个“滴答”,其中为该步骤计划的所有节点都会执行(可能并行)。对于像 START -> A -> B -> END 这样的顺序图,输入、节点 A 和节点 B 分别有单独的超级步骤——每个步骤后都会生成一个检查点。理解超级步骤边界对于时间旅行很重要,因为您只能从检查点(即超级步骤边界)恢复执行。 检查点会被持久化,并可用于在以后恢复线程的状态。 让我们看看在调用一个简单图时保存了哪些检查点:
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

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"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar":[]}, config)
运行图后,我们期望看到恰好 4 个检查点:
  • 空检查点,下一个要执行的节点是 START
  • 包含用户输入 {'foo': '', 'bar': []} 的检查点,下一个要执行的节点是 node_a
  • 包含 node_a 输出 {'foo': 'a', 'bar': ['a']} 的检查点,下一个要执行的节点是 node_b
  • 包含 node_b 输出 {'foo': 'b', 'bar': ['a', 'b']} 的检查点,没有下一个要执行的节点
请注意,bar 通道值包含来自两个节点的输出,因为我们为 bar 通道设置了归约器。

检查点命名空间

每个检查点都有一个 checkpoint_ns(检查点命名空间)字段,用于标识它所属的图或子图:
  • ""(空字符串):检查点属于父(根)图。
  • "node_name:uuid":检查点属于作为给定节点调用的子图。对于嵌套子图,命名空间使用 | 分隔符连接(例如,"outer_node:uuid|inner_node:uuid")。
您可以通过配置从节点内部访问检查点命名空间:
from langchain_core.runnables import RunnableConfig

def my_node(state: State, config: RunnableConfig):
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    # "" 表示父图,"node_name:uuid" 表示子图
有关处理子图状态和检查点的更多详细信息,请参阅子图

获取和更新状态

获取状态

与保存的图状态交互时,您必须指定线程标识符。您可以通过调用 graph.get_state(config) 查看图的_最新_状态。这将返回一个 StateSnapshot 对象,该对象对应于与配置中提供的线程 ID 关联的最新检查点,或者如果提供了检查点 ID,则对应于与该线程的检查点 ID 关联的检查点。
# 获取最新状态快照
config = {"configurable": {"thread_id": "1"}}
graph.get_state(config)

# 获取特定 checkpoint_id 的状态快照
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
graph.get_state(config)
在我们的示例中,get_state 的输出将如下所示:
StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
    metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
    created_at='2024-08-29T19:19:38.821749+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}}, tasks=()
)

StateSnapshot 字段

字段类型描述
valuesdict此检查点的通道状态值。
nexttuple[str, ...]下一个要执行的节点名称。空 () 表示图已完成。
configdict包含 thread_idcheckpoint_nscheckpoint_id
metadatadict执行元数据。包含 source"input""loop""update")、writes(节点输出)和 step(超级步骤计数器)。
created_atstr此检查点创建时的 ISO 8601 时间戳。
parent_configdict | None前一个检查点的配置。第一个检查点为 None
taskstuple[PregelTask, ...]此步骤要执行的任务。每个任务都有 idnameerrorinterrupts,以及可选的 state(当使用 subgraphs=True 时的子图快照)。

获取状态历史

您可以通过调用 graph.get_state_history(config) 获取给定线程的完整图执行历史。这将返回与配置中提供的线程 ID 关联的 StateSnapshot 对象列表。重要的是,检查点将按时间顺序排列,最近的检查点 / StateSnapshot 列表中的第一个。
config = {"configurable": {"thread_id": "1"}}
list(graph.get_state_history(config))
在我们的示例中,get_state_history 的输出将如下所示:
[
    StateSnapshot(
        values={'foo': 'b', 'bar': ['a', 'b']},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
        metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
        created_at='2024-08-29T19:19:38.821749+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        tasks=(),
    ),
    StateSnapshot(
        values={'foo': 'a', 'bar': ['a']},
        next=('node_b',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        metadata={'source': 'loop', 'writes': {'node_a': {'foo': 'a', 'bar': ['a']}}, 'step': 1},
        created_at='2024-08-29T19:19:38.819946+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        tasks=(PregelTask(id='6fb7314f-f114-5413-a1f3-d37dfe98ff44', name='node_b', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'foo': '', 'bar': []},
        next=('node_a',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        metadata={'source': 'loop', 'writes': None, 'step': 0},
        created_at='2024-08-29T19:19:38.817813+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        tasks=(PregelTask(id='f1b14528-5ee5-579c-949b-23ef9bfbed58', name='node_a', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'bar': []},
        next=('__start__',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        metadata={'source': 'input', 'writes': {'foo': ''}, 'step': -1},
        created_at='2024-08-29T19:19:38.816205+00:00',
        parent_config=None,
        tasks=(PregelTask(id='6d27aa2e-d72b-5504-a36f-8620e54a76dd', name='__start__', error=None, interrupts=()),),
    )
]
状态

查找特定检查点

您可以过滤状态历史以查找符合特定条件的检查点:
history = list(graph.get_state_history(config))

# 查找在特定节点执行之前的检查点
before_node_b = next(s for s in history if s.next == ("node_b",))

# 按步骤号查找检查点
step_2 = next(s for s in history if s.metadata["step"] == 2)

# 查找由 update_state 创建的检查点
forks = [s for s in history if s.metadata["source"] == "update"]

# 查找发生中断的检查点
interrupted = next(
    s for s in history
    if s.tasks and any(t.interrupts for t in s.tasks)
)

重放

重放从先前的检查点重新执行步骤。使用先前的 checkpoint_id 调用图以重新运行该检查点之后的节点。检查点之前的节点会被跳过(其结果已保存)。检查点之后的节点会重新执行,包括任何 LLM 调用、API 请求或中断——这些在重放期间总是会重新触发。 有关重放过去执行的完整详细信息和代码示例,请参阅时间旅行 重放

更新状态

您可以使用 update_state 编辑图状态。这会创建一个具有更新值的新检查点——它不会修改原始检查点。更新被视为与节点更新相同:如果定义了值,则通过归约器函数传递值,因此带有归约器的通道会_累积_值而不是覆盖它们。 您可以选择指定 as_node 来控制更新被视为来自哪个节点,这会影响下一个执行哪个节点。有关详细信息,请参阅时间旅行:as_node 更新

内存存储

共享状态模型 状态模式指定了一组在图执行期间填充的键。如上所述,状态可以由检查点保存器在每个图步骤写入线程,从而实现状态持久化。 如果我们想跨线程保留某些信息怎么办?考虑一个聊天机器人的情况,我们希望在与该用户的所有聊天对话(即线程)中保留有关用户的特定信息! 仅使用检查点保存器,我们无法跨线程共享信息。这促使了对 Store 接口的需求。作为说明,我们可以定义一个 InMemoryStore 来跨线程存储有关用户的信息。我们只需像以前一样使用检查点保存器编译图,并传递存储。
LangGraph API 自动处理存储 使用 LangGraph API 时,您无需手动实现或配置存储。API 会在后台为您处理所有存储基础设施。
InMemoryStore 适用于开发和测试。对于生产环境,请使用持久存储,如 PostgresStoreRedisStore。所有实现都扩展了 BaseStore,这是在节点函数签名中使用的类型注解。

基本用法

首先,让我们在不使用 LangGraph 的情况下单独展示这一点。
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
记忆按 tuple 命名空间,在此特定示例中将是 (<user_id>, "memories")。命名空间可以是任意长度并表示任何内容,不一定是用户特定的。
user_id = "1"
namespace_for_memory = (user_id, "memories")
我们使用 store.put 方法将记忆保存到存储中的命名空间。执行此操作时,我们指定如上定义的命名空间,以及记忆的键值对:键只是记忆的唯一标识符(memory_id),值(字典)是记忆本身。
memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
store.put(namespace_for_memory, memory_id, memory)
我们可以使用 store.search 方法读取命名空间中的记忆,这将返回给定用户的所有记忆作为列表。最近的记忆是列表中的最后一个。
memories = store.search(namespace_for_memory)
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}
每种记忆类型都是一个 Python 类(Item),具有某些属性。我们可以通过如上所述的 .dict 将其转换为字典来访问它。 它具有的属性是:
  • value:此记忆的值(本身是一个字典)
  • key:此命名空间中此记忆的唯一键
  • namespace:字符串元组,此记忆类型的命名空间
    虽然类型是 tuple[str, ...],但在转换为 JSON 时可能会序列化为列表(例如,['1', 'memories'])。
  • created_at:此记忆创建时的时间戳
  • updated_at:此记忆更新时的时间戳

语义搜索

除了简单检索外,存储还支持语义搜索,允许您根据含义而不是精确匹配查找记忆。为此,请使用嵌入模型配置存储:
from langchain.embeddings import init_embeddings

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),  # 嵌入提供程序
        "dims": 1536,                              # 嵌入维度
        "fields": ["food_preference", "$"]              # 要嵌入的字段
    }
)
现在搜索时,您可以使用自然语言查询来查找相关记忆:
# 查找有关食物偏好的记忆
# (这可以在将记忆放入存储后完成)
memories = store.search(
    namespace_for_memory,
    query="用户喜欢吃什么?",
    limit=3  # 返回前 3 个匹配项
)
您可以通过配置 fields 参数或在存储记忆时指定 index 参数来控制记忆的哪些部分被嵌入:
# 存储时指定要嵌入的特定字段
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {
        "food_preference": "我喜欢意大利美食",
        "context": "讨论晚餐计划"
    },
    index=["food_preference"]  # 仅嵌入 "food_preferences" 字段
)

# 存储时不嵌入(仍可检索,但不可搜索)
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {"system_info": "最后更新:2024-01-01"},
    index=False
)

在 LangGraph 中使用

准备好这一切后,我们在 LangGraph 中使用存储。存储与检查点保存器协同工作:检查点保存器将状态保存到线程,如上所述,而存储允许我们存储任意信息以供跨线程访问。我们按如下方式使用检查点保存器和存储编译图。
from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver

@dataclass
class Context:
    user_id: str

# 我们需要这个,因为我们想要启用线程(对话)
checkpointer = InMemorySaver()

# ... 定义图 ...

# 使用检查点保存器和存储编译图
builder = StateGraph(MessagesState, context_schema=Context)
# ... 添加节点和边 ...
graph = builder.compile(checkpointer=checkpointer, store=store)
我们像以前一样使用 thread_id 调用图,并使用 user_id,我们将使用它将记忆命名空间限定到此特定用户,如上所示。
# 调用图
config = {"configurable": {"thread_id": "1"}}

# 首先,我们只是向 AI 问好
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi"}]},
    config,
    stream_mode="updates",
    context=Context(user_id="1"),
):
    print(update)
您可以通过使用 Runtime 对象在_任何节点_中访问存储和 user_id。当您将 Runtime 作为参数添加到节点函数时,LangGraph 会自动注入它。以下是如何使用它来保存记忆:
from langgraph.runtime import Runtime
from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

async def update_memory(state: MessagesState, runtime: Runtime[Context]):

    # 从运行时上下文中获取用户 ID
    user_id = runtime.context.user_id

    # 为记忆命名空间
    namespace = (user_id, "memories")

    # ... 分析对话并创建新记忆

    # 创建新的记忆 ID
    memory_id = str(uuid.uuid4())

    # 我们创建一个新记忆
    await runtime.store.aput(namespace, memory_id, {"memory": memory})

如上所示,我们也可以在任何节点中访问存储,并使用 store.search 方法获取记忆。回想一下,记忆作为对象列表返回,可以转换为字典。
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}
我们可以访问记忆并在模型调用中使用它们。
from dataclasses import dataclass
from langgraph.runtime import Runtime

@dataclass
class Context:
    user_id: str

async def call_model(state: MessagesState, runtime: Runtime[Context]):
    # 从运行时上下文中获取用户 ID
    user_id = runtime.context.user_id

    # 为记忆命名空间
    namespace = (user_id, "memories")

    # 基于最新消息搜索
    memories = await runtime.store.asearch(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    info = "\n".join([d.value["memory"] for d in memories])

    # ... 在模型调用中使用记忆
如果我们创建一个新线程,只要 user_id 相同,我们仍然可以访问相同的记忆。
# 在新线程上调用图
config = {"configurable": {"thread_id": "2"}}

# 再次问好
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]},
    config,
    stream_mode="updates",
    context=Context(user_id="1"),
):
    print(update)
当我们使用 LangSmith 时,无论是本地(例如在 Studio 中)还是由 LangSmith 托管,基础存储默认可用,无需在图编译期间指定。但是,要启用语义搜索,您确实需要在 langgraph.json 文件中配置索引设置。例如:
{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}
有关更多详细信息和配置选项,请参阅部署指南

检查点保存器库

在底层,检查点由符合 BaseCheckpointSaver 接口的检查点保存器对象提供支持。LangGraph 提供了多种检查点保存器实现,所有实现都通过独立的、可安装的库完成。
有关可用提供程序,请参阅检查点保存器集成
  • langgraph-checkpoint:检查点保存器的基本接口(BaseCheckpointSaver)和序列化/反序列化接口(SerializerProtocol)。包括用于实验的内存检查点保存器实现(InMemorySaver)。LangGraph 自带 langgraph-checkpoint
  • langgraph-checkpoint-sqlite:使用 SQLite 数据库的 LangGraph 检查点保存器实现(SqliteSaver / AsyncSqliteSaver)。适用于实验和本地工作流。需要单独安装。
  • langgraph-checkpoint-postgres:使用 Postgres 数据库的高级检查点保存器(PostgresSaver / AsyncPostgresSaver),在 LangSmith 中使用。适用于生产环境。需要单独安装。
  • langgraph-checkpoint-cosmosdb:使用 Azure Cosmos DB 的 LangGraph 检查点保存器实现(CosmosDBSaver / AsyncCosmosDBSaver)。适用于在 Azure 上的生产环境使用。支持同步和异步操作。需要单独安装。

检查点保存器接口

每个检查点保存器都符合 BaseCheckpointSaver 接口,并实现以下方法:
  • .put - 存储检查点及其配置和元数据。
  • .put_writes - 存储与检查点链接的中间写入(即待处理写入)。
  • .get_tuple - 使用给定配置(thread_idcheckpoint_id)获取检查点元组。这用于在 graph.get_state() 中填充 StateSnapshot
  • .list - 列出与给定配置和过滤条件匹配的检查点。这用于在 graph.get_state_history() 中填充状态历史。
如果检查点保存器与异步图执行一起使用(即通过 .ainvoke.astream.abatch 执行图),则将使用上述方法的异步版本(.aput.aput_writes.aget_tuple.alist)。
要异步运行图,您可以使用 InMemorySaver,或 Sqlite/Postgres 检查点保存器的异步版本——AsyncSqliteSaver / AsyncPostgresSaver 检查点保存器。

序列化器

当检查点保存器保存图状态时,它们需要序列化状态中的通道值。这是使用序列化器对象完成的。 langgraph_checkpoint 定义了实现序列化器的协议,并提供了默认实现(JsonPlusSerializer),该实现处理多种类型,包括 LangChain 和 LangGraph 原语、日期时间、枚举等。

使用 pickle 进行序列化

默认序列化器 JsonPlusSerializer 在底层使用 ormsgpack 和 JSON,这不适用于所有类型的对象。 如果您想回退到 pickle 来处理当前 msgpack 编码器不支持的对象(例如 Pandas 数据框),可以使用 JsonPlusSerializerpickle_fallback 参数:
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

# ... 定义图 ...
graph.compile(
    checkpointer=InMemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
)

加密

检查点保存器可以选择加密所有持久化状态。要启用此功能,请将 EncryptedSerializer 的实例传递给任何 BaseCheckpointSaver 实现的 serde 参数。创建加密序列化器的最简单方法是通过 from_pycryptodome_aes,它从 LANGGRAPH_AES_KEY 环境变量读取 AES 密钥(或接受 key 参数):
import sqlite3

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver

serde = EncryptedSerializer.from_pycryptodome_aes()  # 读取 LANGGRAPH_AES_KEY
checkpointer = SqliteSaver(sqlite3.connect("checkpoint.db"), serde=serde)
from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.postgres import PostgresSaver

serde = EncryptedSerializer.from_pycryptodome_aes()
checkpointer = PostgresSaver.from_conn_string("postgresql://...", serde=serde)
checkpointer.setup()
在 LangSmith 上运行时,只要存在 LANGGRAPH_AES_KEY,加密就会自动启用,因此您只需提供环境变量。可以通过实现 CipherProtocol 并将其提供给 EncryptedSerializer 来使用其他加密方案。