> ## Documentation Index > Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt > Use this file to discover all available pages before exploring further. # 持久化 LangGraph 内置了一个持久化层，它将图状态保存为检查点。当您使用检查点保存器编译图时，执行过程中的每个步骤都会保存图状态的快照，并组织成线程。这支持了人机交互工作流、对话记忆、时间旅行调试和容错执行。 检查点 **Agent Server 自动处理检查点保存** 当使用 [Agent Server](/langsmith/agent-server) 时，您无需手动实现或配置检查点保存器。服务器会在后台为您处理所有持久化基础设施。 ## 为什么使用持久化 以下功能需要持久化： * **人机交互**：检查点保存器通过允许人类检查、中断和批准图步骤来促进[人机交互工作流](/oss/python/langgraph/interrupts)。这些工作流需要检查点保存器，因为人员必须能够在任何时间点查看图的状态，并且图必须能够在人员对状态进行任何更新后恢复执行。有关示例，请参阅[中断](/oss/python/langgraph/interrupts)。 * **记忆**：检查点保存器允许在交互之间存在["记忆"](/oss/python/concepts/memory)。在重复的人类交互（如对话）的情况下，任何后续消息都可以发送到该线程，该线程将保留其对先前交互的记忆。有关如何使用检查点保存器添加和管理对话记忆的信息，请参阅[添加记忆](/oss/python/langgraph/add-memory)。 * **时间旅行**：检查点保存器允许["时间旅行"](/oss/python/langgraph/use-time-travel)，允许用户重放先前的图执行以审查和/或调试特定的图步骤。此外，检查点保存器使得可以在任意检查点处分叉图状态，以探索替代轨迹。 * **容错性**：检查点保存提供了容错性和错误恢复：如果一个或多个节点在给定的超级步骤中失败，您可以从最后一个成功的步骤重新启动图。 * **待处理写入**：当图节点在给定的[超级步骤](#super-steps)中执行失败时，LangGraph 会存储在该超级步骤中成功完成的任何其他节点的待处理检查点写入。当您从该超级步骤恢复图执行时，您不会重新运行成功的节点。 ## 核心概念 ### 线程 线程是分配给检查点保存器保存的每个检查点的唯一 ID 或线程标识符。它包含一系列[运行](/langsmith/runs)的累积状态。当运行执行时，助手底层图的[状态](/oss/python/langgraph/graph-api#state)将被持久化到线程中。 当使用检查点保存器调用图时，您**必须**在配置的 `configurable` 部分指定 `thread_id`： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} {"configurable": {"thread_id": "1"}} ``` 可以检索线程的当前和历史状态。要持久化状态，必须在执行运行之前创建线程。LangSmith API 提供了多个端点来创建和管理线程及线程状态。有关更多详细信息，请参阅 [API 参考](https://reference.langchain.com/python/langsmith/)。 检查点保存器使用 `thread_id` 作为存储和检索检查点的主键。没有它，检查点保存器无法保存状态或在[中断](/oss/python/langgraph/interrupts)后恢复执行，因为检查点保存器使用 `thread_id` 来加载保存的状态。 ### 检查点 线程在特定时间点的状态称为检查点。检查点是在每个[超级步骤](#super-steps)保存的图状态快照，由 `StateSnapshot` 对象表示（有关完整字段参考，请参阅 [StateSnapshot 字段](#statesnapshot-fields)）。 #### 超级步骤 LangGraph 在每个**超级步骤**边界创建一个检查点。超级步骤是图的一个单独“滴答”，其中为该步骤调度的所有节点执行（可能并行执行）。对于像 `START -> A -> B -> END` 这样的顺序图，输入、节点 A 和节点 B 有单独的超级步骤——在每个之后产生一个检查点。理解超级步骤边界对于[时间旅行](/oss/python/langgraph/use-time-travel)很重要，因为您只能从检查点（即超级步骤边界）恢复执行。 检查点被持久化，可用于在以后恢复线程的状态。 让我们看看当以如下方式调用简单图时保存了哪些检查点： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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`](https://reference.langchain.com/python/langgraph/constants/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"`）。 您可以通过配置从节点内部访问检查点命名空间： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain_core.runnables import RunnableConfig def my_node(state: State, config: RunnableConfig): checkpoint_ns = config["configurable"]["checkpoint_ns"] # "" 表示父图，"node_name:uuid" 表示子图 ``` 有关使用子图状态和检查点的更多详细信息，请参阅[子图](/oss/python/langgraph/use-subgraphs)。 ## 获取和更新状态 ### 获取状态 与保存的图状态交互时，您**必须**指定[线程标识符](#threads)。您可以通过调用 `graph.get_state(config)` 来查看图的\_最新\_状态。这将返回一个 `StateSnapshot` 对象，该对象对应于与配置中提供的线程 ID 关联的最新检查点，或者如果提供了检查点 ID，则对应于与该线程关联的检查点 ID 的检查点。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 获取最新状态快照 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 字段 | 字段 | 类型 | 描述 | | --------------- | ------------------------ | --------------------------------------------------------------------------------------------- | | `values` | `dict` | 此检查点处的状态通道值。 | | `next` | `tuple[str, ...]` | 下一个要执行的节点名称。空的 `()` 表示图已完成。 | | `config` | `dict` | 包含 `thread_id`、`checkpoint_ns` 和 `checkpoint_id`。 | | `metadata` | `dict` | 执行元数据。包含 `source`（`"input"`、`"loop"` 或 `"update"`）、`writes`（节点输出）和 `step`（超级步骤计数器）。 | | `created_at` | `str` | 此检查点创建时的 ISO 8601 时间戳。 | | `parent_config` | `dict \| None` | 前一个检查点的配置。第一个检查点为 `None`。 | | `tasks` | `tuple[PregelTask, ...]` | 此步骤要执行的任务。每个任务都有 `id`、`name`、`error`、`interrupts`，以及可选的 `state`（子图快照，当使用 `subgraphs=True` 时）。 | ### 获取状态历史记录 您可以通过调用 [`graph.get_state_history(config)`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.get_state_history) 来获取给定线程的完整图执行历史记录。这将返回与配置中提供的线程 ID 关联的 `StateSnapshot` 对象列表。重要的是，检查点将按时间顺序排列，最新的检查点/`StateSnapshot` 是列表中的第一个。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} config = {"configurable": {"thread_id": "1"}} list(graph.get_state_history(config)) ``` 在我们的示例中，[`get_state_history`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.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=()),), ) ] ``` 状态 #### 查找特定检查点 您可以过滤状态历史记录以查找符合特定条件的检查点： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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 请求或[中断](/oss/python/langgraph/interrupts)——在重放期间总是会重新触发。 有关重放过去执行的完整详细信息和代码示例，请参阅[时间旅行](/oss/python/langgraph/use-time-travel)。 重放 ### 更新状态 您可以使用 [`update_state`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.update_state) 编辑图状态。这会创建一个具有更新值的新检查点——它不会修改原始检查点。更新的处理方式与节点更新相同：当定义了[归约器](/oss/python/langgraph/graph-api#reducers)函数时，值会通过归约器函数传递，因此带有归约器的通道会\_累积\_值而不是覆盖它们。 您可以选择指定 `as_node` 来控制更新被视为来自哪个节点，这会影响接下来执行哪个节点。有关详细信息，请参阅[时间旅行：`as_node`](/oss/python/langgraph/use-time-travel#from-a-specific-node)。 更新 ## 存储 共享状态模型 [状态模式](/oss/python/langgraph/graph-api#schema)指定了一组在图执行时填充的键。如上所述，状态可以由检查点保存器在每个图步骤写入线程，从而实现状态持久化。 如果我们想\_跨线程\_保留一些信息怎么办？考虑一个聊天机器人的场景，我们希望在与该用户的所有聊天对话（即线程）中保留关于该用户的特定信息！ 仅使用检查点保存器，我们无法跨线程共享信息。这就需要 [`Store`](https://reference.langchain.com/python/langgraph/store/) 接口。作为说明，我们可以定义一个 `InMemoryStore` 来跨线程存储关于用户的信息。我们只需像以前一样使用检查点保存器编译图，并传递存储。 **LangGraph API 自动处理存储** 当使用 LangGraph API 时，您无需手动实现或配置存储。API 会在后台为您处理所有存储基础设施。 [InMemoryStore](https://reference.langchain.com/python/langchain-core/stores/InMemoryStore) 适用于开发和测试。对于生产环境，请使用持久化存储，如 `PostgresStore`、`MongoDBStore` 或 `RedisStore`。所有实现都扩展了 [BaseStore](https://reference.langchain.com/python/langchain-core/stores/BaseStore)，这是在节点函数签名中使用的类型注解。 ### 基本用法 首先，让我们在不使用 LangGraph 的情况下单独展示这一点。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.store.memory import InMemoryStore store = InMemoryStore() ``` 记忆通过 `tuple` 进行命名空间划分，在此特定示例中将是 `(, "memories")`。命名空间可以是任意长度并表示任何内容，不一定是特定于用户的。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} user_id = "1" namespace_for_memory = (user_id, "memories") ``` 我们使用 `store.put` 方法将记忆保存到存储中的命名空间。执行此操作时，我们指定如上定义的命名空间，以及记忆的键值对：键只是记忆的唯一标识符（`memory_id`），值（字典）是记忆本身。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} memory_id = str(uuid.uuid4()) memory = {"food_preference" : "I like pizza"} store.put(namespace_for_memory, memory_id, memory) ``` 我们可以使用 `store.search` 方法读取命名空间中的记忆，该方法将返回给定用户的所有记忆作为列表。最近的记忆是列表中的最后一个。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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`](https://langchain-ai.github.io/langgraph/reference/store/#langgraph.store.base.Item)），具有某些属性。我们可以通过如上所述的 `.dict` 转换将其作为字典访问。 它具有的属性有： * `value`：此记忆的值（本身是一个字典） * `key`：此记忆在此命名空间中的唯一键 * `namespace`：字符串元组，此记忆类型的命名空间 虽然类型是 `tuple[str, ...]`，但在转换为 JSON 时可能会序列化为列表（例如，`['1', 'memories']`）。 * `created_at`：此记忆创建时的时间戳 * `updated_at`：此记忆更新时的时间戳 ### 语义搜索 除了简单的检索，存储还支持语义搜索，允许您根据含义而不是精确匹配来查找记忆。为此，请使用嵌入模型配置存储： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.embeddings import init_embeddings store = InMemoryStore( index={ "embed": init_embeddings("openai:text-embedding-3-small"), # 嵌入提供者 "dims": 1536, # 嵌入维度 "fields": ["food_preference", "$"] # 要嵌入的字段 } ) ``` 现在搜索时，您可以使用自然语言查询来查找相关记忆： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 查找关于食物偏好的记忆 # （这可以在将记忆放入存储后完成） memories = store.search( namespace_for_memory, query="What does the user like to eat?", limit=3 # 返回前 3 个匹配项 ) ``` 您可以通过配置 `fields` 参数或在存储记忆时指定 `index` 参数来控制记忆的哪些部分被嵌入： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 存储时指定要嵌入的字段 store.put( namespace_for_memory, str(uuid.uuid4()), { "food_preference": "I love Italian cuisine", "context": "Discussing dinner plans" }, index=["food_preference"] # 仅嵌入 "food_preferences" 字段 ) # 存储时不嵌入（仍然可检索，但不可搜索） store.put( namespace_for_memory, str(uuid.uuid4()), {"system_info": "Last updated: 2024-01-01"}, index=False ) ``` ### 在 LangGraph 中使用 所有这些就绪后，我们在 LangGraph 中使用存储。存储与检查点保存器协同工作：检查点保存器将状态保存到线程，如上所述，而存储允许我们存储任意信息以供\_跨\_线程访问。我们如下使用检查点保存器和存储编译图。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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`，我们将使用它将记忆命名空间划分到这个特定用户，如上所示。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 调用图 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 会自动注入它。以下是如何使用它来保存记忆的示例： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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` 方法获取记忆。请记住，记忆作为对象列表返回，可以转换为字典。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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` 相同，我们仍然可以访问相同的记忆。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 在新线程上调用图 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/studio) 中）还是[使用 LangSmith 托管](/langsmith/platform-setup)，基础存储默认可用，无需在图编译期间指定。但是，要启用语义搜索，您**确实**需要在 `langgraph.json` 文件中配置索引设置。例如： ```json theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} { ... "store": { "index": { "embed": "openai:text-embeddings-3-small", "dims": 1536, "fields": ["$"] } } } ``` 有关更多详细信息和配置选项，请参阅[部署指南](/langsmith/semantic-search)。 ## 优化检查点存储 默认情况下，LangGraph 检查点在每个超级步骤写入每个状态通道的完整值。对于具有大量累积的长时间运行的线程——例如多轮对话——这可能会随时间产生显著的存储增长。 [`DeltaChannel`](https://reference.langchain.com/python/langgraph/channels/delta/DeltaChannel) 仅存储增量差异，而不是完整的累积值，从而显著减少追加密集型通道的检查点大小。有关用法和存储与延迟的权衡，请参阅 [DeltaChannel](/oss/python/langgraph/pregel#deltachannel-beta)。 `DeltaChannel` 需要 `langgraph>=1.2`，目前处于测试阶段。API 可能在未来版本中更改。 ## 检查点保存器库 在底层，检查点保存由符合 [`BaseCheckpointSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) 接口的检查点保存器对象提供支持。LangGraph 提供了多种检查点保存器实现，所有实现都通过独立的、可安装的库实现。 有关可用提供者，请参阅[检查点保存器集成](/oss/python/integrations/checkpointers/index)。 * `langgraph-checkpoint`：检查点保存器保存器的基础接口（[`BaseCheckpointSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver)）和序列化/反序列化接口（[`SerializerProtocol`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol)）。包括用于实验的内存检查点保存器实现（[`InMemorySaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.memory.InMemorySaver)）。LangGraph 自带 `langgraph-checkpoint`。 * `langgraph-checkpoint-sqlite`：使用 SQLite 数据库的 LangGraph 检查点保存器实现（[`SqliteSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.sqlite.SqliteSaver) / [`AsyncSqliteSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver)）。适用于实验和本地工作流。需要单独安装。 * `langgraph-checkpoint-postgres`：使用 Postgres 数据库的高级检查点保存器（[`PostgresSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.postgres.PostgresSaver) / [`AsyncPostgresSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver)），在 LangSmith 中使用。适用于生产环境。需要单独安装。 * `langchain-azure-cosmosdb`：使用 Azure Cosmos DB for NoSQL 的 LangGraph 检查点保存器实现（[`CosmosDBSaverSync`](https://reference.langchain.com/python/langchain-azure-cosmosdb/) / [`CosmosDBSaver`](https://reference.langchain.com/python/langchain-azure-cosmosdb/)）。适用于在 Azure 上生产使用。支持同步和异步操作，使用 Microsoft Entra ID 身份验证。需要单独安装。 ### 检查点保存器接口 每个检查点保存器都符合 [`BaseCheckpointSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) 接口，并实现以下方法： * `.put` - 存储检查点及其配置和元数据。 * `.put_writes` - 存储链接到检查点的中间写入（即[待处理写入](#pending-writes)）。 * `.get_tuple` - 使用给定配置（`thread_id` 和 `checkpoint_id`）获取检查点元组。这用于填充 `graph.get_state()` 中的 `StateSnapshot`。 * `.list` - 列出与给定配置和过滤条件匹配的检查点。这用于填充 `graph.get_state_history()` 中的状态历史记录。 如果检查点保存器与异步图执行一起使用（即通过 `.ainvoke`、`.astream`、`.abatch` 执行图），则将使用上述方法的异步版本（`.aput`、`.aput_writes`、`.aget_tuple`、`.alist`）。 要异步运行图，您可以使用 [`InMemorySaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.memory.InMemorySaver)，或 Sqlite/Postgres 检查点保存器的异步版本——[`AsyncSqliteSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver) / [`AsyncPostgresSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver) 检查点保存器。 ### 序列化器 当检查点保存器保存图状态时，它们需要序列化状态中的通道值。这是使用序列化器对象完成的。 `langgraph_checkpoint` 定义了实现序列化器的[协议](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol)，并提供了一个默认实现（[`JsonPlusSerializer`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer)），可以处理多种类型，包括 LangChain 和 LangGraph 原语、日期时间、枚举等。 #### 使用 `pickle` 进行序列化 默认序列化器 [`JsonPlusSerializer`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer) 在底层使用 ormsgpack 和 JSON，这并不适合所有类型的对象。 如果您希望对我们 msgpack 编码器当前不支持的对象（如 Pandas 数据帧）回退到 pickle，您可以使用 [`JsonPlusSerializer`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer) 的 `pickle_fallback` 参数： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer # ... 定义图 ... graph.compile( checkpointer=InMemorySaver(serde=JsonPlusSerializer(pickle_fallback=True)) ) ``` #### 加密 检查点保存器可以选择加密所有持久化的状态。要启用此功能，请将 [`EncryptedSerializer`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer) 的实例传递给任何 [`BaseCheckpointSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) 实现的 `serde` 参数。创建加密序列化器的最简单方法是通过 [`from_pycryptodome_aes`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer.from_pycryptodome_aes)，它从 `LANGGRAPH_AES_KEY` 环境变量读取 AES 密钥（或接受 `key` 参数）： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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) ``` ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.base.CipherProtocol) 并将其提供给 [`EncryptedSerializer`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer) 来使用其他加密方案。 ***
[将这些文档连接](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获取实时答案。 [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/persistence.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。