> ## Documentation Index
> Fetch the complete documentation index at: https://langchain.idochub.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# 持久化

<警告>
  **Alpha 版本提示：** 本文档涵盖的是 **v1-alpha** 版本。内容尚不完整，且可能随时更改。

  如需查阅最新稳定版本，请参阅当前的 [LangGraph Python 文档](https://langchain-ai.github.io/langgraph/) 或 [LangGraph JavaScript 文档](https://langchain-ai.github.io/langgraphjs/)。
</警告>

LangGraph 内置了持久化层，通过检查点器（checkpointer）实现。当你使用检查点器编译图时，检查点器会在每个超步（super-step）保存图状态的“检查点”。这些检查点保存到“线程”（thread）中，可在图执行后访问。由于“线程”允许在执行后访问图的状态，因此可实现包括人在回路（human-in-the-loop）、记忆、时间旅行和容错在内的多种强大功能。下文将详细讨论这些概念。

[检查点](/oss/images/checkpoints.jpg)

<Info>
  **LangGraph API 自动处理检查点**
  使用 LangGraph API 时，您无需手动实现或配置检查点器。API 会在后台自动为您处理所有持久化基础设施。
</Info>

## 线程（Threads）

线程是由检查点器为每个保存的检查点分配的唯一 ID 或线程标识符。它包含一系列[运行](/langgraph-platform/assistants#execution)的累积状态。当执行一次运行时，助手底层图的[状态](/oss/python/langgraph/graph-api#state)将被持久化到该线程中。

调用带有检查点器的图时，您**必须**在配置的 `configurable` 部分指定 `thread_id`。

```python theme={null}
{"configurable": {"thread_id": "1"}}
```

可以检索线程的当前和历史状态。要持久化状态，必须在执行运行前创建线程。LangGraph 平台 API 提供了多个用于创建和管理线程及线程状态的端点。详情请参阅[API 参考文档](https://langchain-ai.github.io/langgraph/cloud/reference/api/)。

## 检查点（Checkpoints）

线程在特定时间点的状态称为检查点。检查点是在每个超步保存的图状态快照，由 `StateSnapshot` 对象表示，包含以下关键属性：

* `config`：与此检查点关联的配置。
* `metadata`：与此检查点关联的元数据。
* `values`：此时各状态通道的值。
* `next`：图中接下来要执行的节点名称元组。
* `tasks`：包含待执行任务信息的 `PregelTask` 对象元组。如果该步骤之前已尝试过，则包含错误信息。如果图从节点内部被[动态中断](/oss/python/langgraph/add-human-in-the-loop#pause-using-interrupt)，任务将包含与中断相关的附加数据。

检查点会被持久化，可用于在稍后时间恢复线程状态。

让我们看看当一个简单图被如下调用时保存了哪些检查点：

```python theme={null}
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": ""}, config)
```

运行图后，我们预期会看到恰好 4 个检查点：

* 空检查点，`START` 作为下一个要执行的节点
* 包含用户输入 `{'foo': '', 'bar': []}` 且 `node_a` 为下一个要执行节点的检查点
* 包含 `node_a` 输出 `{'foo': 'a', 'bar': ['a']}` 且 `node_b` 为下一个要执行节点的检查点
* 包含 `node_b` 输出 `{'foo': 'b', 'bar': ['a', 'b']}` 且无后续节点的检查点

注意，由于我们为 `bar` 通道定义了 reducer，因此 `bar` 通道的值包含两个节点的输出。

### 获取状态

与保存的图状态交互时，您**必须**指定[线程标识符](#threads)。通过调用 `graph.get_state(config)` 可查看图的\_最新\_状态。这将返回一个 `StateSnapshot` 对象，对应于配置中提供的线程 ID 的最新检查点，或如果提供了检查点 ID，则返回该线程对应的检查点。

```python theme={null}
# 获取最新状态快照
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=()
)
```

### 获取状态历史

通过调用 `graph.get_state_history(config)`，您可以获取给定线程的完整图执行历史。这将返回与配置中提供的线程 ID 关联的 `StateSnapshot` 对象列表。重要的是，检查点将按时间顺序排列，最近的检查点 / `StateSnapshot` 位于列表首位。

```python theme={null}
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=()),),
    )
]
```

[状态](/oss/images/get_state.jpg)

### 重播（Replay）

还可以回放之前的图执行。如果我们使用 `thread_id` 和 `checkpoint_id` 调用图，LangGraph 将在对应 `checkpoint_id` 的检查点\_之前\_重新播放已执行的步骤，并仅执行检查点\_之后\_的步骤。

* `thread_id` 是线程的 ID。
* `checkpoint_id` 是指代线程内特定检查点的标识符。

调用图时，必须将这些作为配置的 `configurable` 部分传递：

```python theme={null}
config = {"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}
graph.invoke(None, config=config)
```

重要的是，LangGraph 知道特定步骤是否之前已执行过。如果已执行，LangGraph 仅在图中\_重播\_该步骤，而不会重新执行，但仅适用于提供的 `checkpoint_id` \_之前\_的步骤。所有 `checkpoint_id` \_之后\_的步骤都将被执行（即新分支），即使它们之前已执行过。有关重播的更多信息，请参阅[时间旅行操作指南](/oss/python/langgraph/use-time-travel)。

[重播](/oss/images/re_play.png)

### 更新状态

除了从特定 `检查点` 重播图外，我们还可以\_编辑\_图状态。我们使用 `graph.update_state()` 方法实现。此方法接受三个参数：

#### `config`

配置应包含指定要更新的线程的 `thread_id`。仅传递 `thread_id` 时，我们将更新（或分叉）当前状态。可选地，如果包含 `checkpoint_id` 字段，则分叉该选定检查点。

#### `values`

这些值将用于更新状态。请注意，此更新的处理方式与任何节点更新完全相同。这意味着这些值将传递给[reducer](/oss/python/langgraph/graph-api#reducers) 函数（如果图状态中的某些通道已定义）。这意味着 `update_state` 不会自动覆盖每个通道的通道值，而仅覆盖没有 reducer 的通道。让我们通过一个示例说明。

假设您使用以下模式定义了图的状态（参见上面的完整示例）：

```python theme={null}
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]
```

现在假设图的当前状态为

```
{"foo": 1, "bar": ["a"]}
```

如果您如下更新状态：

```python theme={null}
graph.update_state(config, {"foo": 2, "bar": ["b"]})
```

则图的新状态将为：

```
{"foo": 2, "bar": ["a", "b"]}
```

`foo` 键（通道）被完全更改（因为该通道未指定 reducer，因此 `update_state` 会覆盖它）。然而，`bar` 键指定了 reducer，因此它将 `"b"` 附加到 `bar` 的状态。

#### `as_node`

调用 `update_state` 时，您还可以选择指定 `as_node`。如果提供了该参数，更新将被视为来自节点 `as_node`。如果未提供 `as_node`，则将其设置为上次更新状态的节点（如果不含糊）。这很重要，因为要执行的下一步取决于上次给出更新的节点，因此这可用于控制下一个执行的节点。有关分叉状态的更多信息，请参阅[时间旅行操作指南](/oss/python/langgraph/use-time-travel)。

[更新](/oss/images/checkpoints_full_story.jpg)

## 内存存储（Memory Store）

[共享状态模型](/oss/images/shared_state.png)

[状态模式](/oss/python/langgraph/graph-api#schema) 指定了图执行时填充的一组键。如上所述，状态可以在每个图步骤由检查点器写入线程，从而实现状态持久化。

但是，如果我们希望在\_线程之间\_保留某些信息呢？考虑一个聊天机器人的场景，我们希望在与该用户的\_所有\_聊天对话（即线程）中保留特定的用户信息！

仅使用检查点器，我们无法在线程之间共享信息。这促使了 [`Store`](https://python.langchain.com/api_reference/langgraph/index.html#module-langgraph.store) 接口的需求。作为示例，我们可以定义一个 `InMemoryStore` 来跨线程存储有关用户的信息。我们只需像以前一样使用检查点器编译图，并添加新的 `in_memory_store` 变量。

<Info>
  **LangGraph API 自动处理存储**
  使用 LangGraph API 时，您无需手动实现或配置存储。API 会在后台自动为您处理所有存储基础设施。
</Info>

### 基本用法

首先，让我们在不使用 LangGraph 的情况下单独展示这一点。

```python theme={null}
from langgraph.store.memory import InMemoryStore
in_memory_store = InMemoryStore()
```

内存按 `元组` 命名空间划分，在此特定示例中为 `(<user_id>, "memories")`。命名空间可以是任意长度，代表任何内容，不必特定于用户。

```python theme={null}
user_id = "1"
namespace_for_memory = (user_id, "memories")
```

我们使用 `store.put` 方法将内存保存到存储中的命名空间。执行此操作时，我们指定上述定义的命名空间，以及内存的键值对：键是内存的唯一标识符 (`memory_id`)，值（字典）是内存本身。

```python theme={null}
memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
in_memory_store.put(namespace_for_memory, memory_id, memory)
```

我们可以使用 `store.search` 方法读取命名空间中的内存，这将返回给定用户的所有内存列表。最近的内存位于列表末尾。

```python theme={null}
memories = in_memory_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`：字符串列表，此内存类型的命名空间
* `created_at`：此内存创建的时间戳
* `updated_at`：此内存更新的时间戳

### 语义搜索

除了简单检索，存储还支持语义搜索，允许您根据含义而非精确匹配查找内存。要启用此功能，请使用嵌入模型配置存储：

```python theme={null}
from langchain.embeddings import init_embeddings

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),  # 嵌入提供者
        "dims": 1536,                              # 嵌入维度
        "fields": ["food_preference", "$"]              # 要嵌入的字段
    }
)
```

现在搜索时，您可以使用自然语言查询查找相关内存：

```python theme={null}
# 查找有关食物偏好的内存
# （这可以在将内存放入存储后执行）
memories = store.search(
    namespace_for_memory,
    query="What does the user like to eat?",
    limit=3  # 返回前 3 个匹配项
)
```

您可以通过配置 `fields` 参数或在存储内存时指定 `index` 参数来控制内存的哪些部分被嵌入：

```python theme={null}
# 存储时指定要嵌入的特定字段
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 中使用 `in_memory_store`。`in_memory_store` 与检查点器协同工作：检查点器将状态保存到线程（如上所述），而 `in_memory_store` 允许我们存储可在\_线程之间\_访问的任意信息。我们如下编译同时包含检查点器和 `in_memory_store` 的图。

```python theme={null}
from langgraph.checkpoint.memory import InMemorySaver

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

# ... 定义图 ...

# 使用检查点器和存储编译图
graph = graph.compile(checkpointer=checkpointer, store=in_memory_store)
```

我们像以前一样使用 `thread_id` 调用图，同时还使用 `user_id`，我们将用它来为特定用户的内存命名空间，如上所示。

```python theme={null}
# 调用图
user_id = "1"
config = {"configurable": {"thread_id": "1", "user_id": user_id}}

# 首先，我们只需向 AI 问好
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi"}]}, config, stream_mode="updates"
):
    print(update)
```

我们可以在\_任何节点\_中通过传递 `store: BaseStore` 和 `config: RunnableConfig` 作为节点参数来访问 `in_memory_store` 和 `user_id`。以下是我们如何在节点中使用语义搜索查找相关内存的示例：

```python theme={null}
def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseStore):

    # 从配置中获取用户 ID
    user_id = config["configurable"]["user_id"]

    # 为内存命名空间
    namespace = (user_id, "memories")

    # ... 分析对话并创建新内存

    # 创建新内存 ID
    memory_id = str(uuid.uuid4())

    # 我们创建一个新内存
    store.put(namespace, memory_id, {"memory": memory})

```

如上所示，我们还可以在任何节点中访问存储并使用 `store.search` 方法获取内存。回想一下，内存作为对象列表返回，可以转换为字典。

```python theme={null}
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={null}
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    # 从配置中获取用户 ID
    user_id = config["configurable"]["user_id"]

    # 为内存命名空间
    namespace = (user_id, "memories")

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

    # ... 在模型调用中使用内存
```

如果我们创建一个新线程，只要 `user_id` 相同，我们仍然可以访问相同的内存。

```python theme={null}
# 调用图
config = {"configurable": {"thread_id": "2", "user_id": "1"}}

# 让我们再次问好
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]}, config, stream_mode="updates"
):
    print(update)
```

当我们在 LangGraph 平台（本地，例如在 LangGraph Studio 中，或使用 LangGraph 平台）使用时，基础存储默认可用，无需在图编译期间指定。然而，要启用语义搜索，您**必须**在 `langgraph.json` 文件中配置索引设置。例如：

```json theme={null}
{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}
```

有关更多详细信息和配置选项，请参阅[部署指南](/langgraph-platform/semantic-search)。

## 检查点器库

在底层，检查点由符合 [BaseCheckpointSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) 接口的检查点器对象提供支持。LangGraph 提供了多个检查点器实现，均通过独立的、可安装的库实现：

* `langgraph-checkpoint`：检查点器保存器的基础接口 ([BaseCheckpointSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver)) 和序列化/反序列化接口 ([SerializerProtocol](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol))。包括用于实验的内存检查点器实现 ([InMemorySaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.memory.InMemorySaver))。LangGraph 自带 `langgraph-checkpoint`。
* `langgraph-checkpoint-sqlite`：使用 SQLite 数据库的 LangGraph 检查点器实现 ([SqliteSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.sqlite.SqliteSaver) / [AsyncSqliteSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver))。适用于实验和本地工作流。需要单独安装。
* `langgraph-checkpoint-postgres`：使用 Postgres 数据库的高级检查点器 ([PostgresSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.postgres.PostgresSaver) / [AsyncPostgresSaver](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver))，用于 LangGraph 平台。适用于生产环境。需要单独安装。

### 检查点器接口

每个检查点器符合 [BaseCheckpointSaver](https://langchain-ai.github.io/langgraph/reference/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`）。

<Note>
  要异步运行您的图，您可以使用 `InMemorySaver`，或 Sqlite/Postgres 检查点器的异步版本 -- `AsyncSqliteSaver` / `AsyncPostgresSaver` 检查点器。
</Note>

### 序列化器

当检查点器保存图状态时，它们需要序列化状态中的通道值。这是通过序列化器对象完成的。

`langgraph_checkpoint` 定义了 [protocol](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol) 用于实现序列化器，并提供默认实现 ([JsonPlusSerializer](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer))，可处理多种类型，包括 LangChain 和 LangGraph 原语、日期时间、枚举等。

#### 使用 `pickle` 序列化

默认序列化器 [`JsonPlusSerializer`](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer) 在底层使用 ormsgpack 和 JSON，这不适用于所有类型的对象。

如果您想对当前 msgpack 编码器不支持的对象（如 Pandas 数据帧）回退到 pickle，可以使用 `JsonPlusSerializer` 的 `pickle_fallback` 参数：

```python theme={null}
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

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

#### 加密

检查点器可以选择加密所有持久化状态。要启用此功能，请将 [`EncryptedSerializer`](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer) 实例传递给任何 `BaseCheckpointSaver` 实现的 `serde` 参数。创建加密序列化器的最简单方法是通过 [`from_pycryptodome_aes`](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer.from_pycryptodome_aes)，它从 `LANGGRAPH_AES_KEY` 环境变量读取 AES 密钥（或接受 `key` 参数）：

```python theme={null}
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={null}
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()
```

在 LangGraph 平台上运行时，只要存在 `LANGGRAPH_AES_KEY`，就会自动启用加密，因此您只需提供环境变量。其他加密方案可以通过实现 [`CipherProtocol`](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.serde.base.CipherProtocol) 并将其提供给 `EncryptedSerializer` 来使用。

## 功能

### 人在回路（Human-in-the-loop）

首先，检查点器通过允许人类检查、中断和批准图步骤来促进[人在回路工作流](/oss/python/langgraph/add-human-in-the-loop)。这些工作流需要检查点器，因为人类必须能够在任何时间点查看图的状态，并且图必须能够在人类对状态进行任何更新后恢复执行。有关示例，请参阅[操作指南](/oss/python/langgraph/add-human-in-the-loop)。

### 记忆（Memory）

其次，检查点器允许交互之间的“[记忆](/oss/python/memory)”。在重复的人类交互（如对话）情况下，任何后续消息都可以发送到该线程，该线程将保留对之前消息的记忆。有关如何使用检查点器添加和管理对话记忆的信息，请参阅[添加记忆](/oss/python/langgraph/add-memory)。

### 时间旅行（Time Travel）

第三，检查点器允许“[时间旅行](/oss/python/langgraph/use-time-travel)”，允许用户重播之前的图执行以审查和/或调试特定图步骤。此外，检查点器使得在任意检查点分叉图状态以探索替代轨迹成为可能。

### 容错（Fault-tolerance）

最后，检查点还提供容错和错误恢复：如果一个或多个节点在给定超步失败，您可以从最后一个成功步骤重新启动图。此外，当图节点在给定超步执行中途失败时，LangGraph 会存储该超步中任何其他成功完成的节点的待处理检查点写入，以便当我们从该超步恢复图执行时，我们不会重新运行成功的节点。

#### 待处理写入（Pending writes）

此外，当图节点在给定超步执行中途失败时，LangGraph 会存储该超步中任何其他成功完成的节点的待处理检查点写入，以便当我们从该超步恢复图执行时，我们不会重新运行成功的节点。