# https://github.com/langchain-ai/langgraph 项目说明书

生成时间：2026-06-01 00:33:45 UTC

## 目录

- [LangGraph 简介](#page-introduction)
- [安装与配置](#page-installation)
- [核心架构](#page-core-architecture)
- [图构建与状态定义](#page-graph-construction)
- [Pregel 执行引擎](#page-pregel-execution)
- [通道与状态管理](#page-channels)
- [检查点与持久化](#page-checkpointing)
- [状态管理与内存](#page-state-management)
- [CLI 使用指南](#page-cli-usage)
- [SDK 与 API 集成](#page-sdk-integration)

<a id='page-introduction'></a>

## LangGraph 简介

### 相关页面

相关主题：[核心架构](#page-core-architecture), [安装与配置](#page-installation)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [README.md](https://github.com/langchain-ai/langgraph/blob/main/README.md)
- [libs/prebuilt/langgraph/prebuilt/tool_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_node.py)
- [libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py)
- [libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)
- [libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
- [libs/cli/examples/graphs/storm.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/examples/graphs/storm.py)
</details>

# LangGraph 简介

## 项目概述

LangGraph 是一个用于构建有状态、多角色 AI 应用的框架，基于 LangChain 构建。LangGraph 的核心价值在于将复杂的 AI 工作流建模为**有向图（Directed Graph）**，其中节点代表计算步骤或代理角色，边代表状态流转逻辑。

### 核心设计理念

LangGraph 的设计灵感来源于三个重要项目：

- **Pregel**：Google 的批量同步并行图处理系统
- **Apache Beam**：统一的大规模数据处理编程模型
- **NetworkX**：Python 图论库的公共接口设计

资料来源：[README.md:1]()

### 定位与生态

LangGraph 由 LangChain Inc 创建和维护，但可以**独立于 LangChain 使用**。它提供了：

| 组件 | 说明 |
|------|------|
| `libs/langgraph` | 核心图执行引擎 |
| `libs/cli` | LangGraph Server CLI 工具 |
| `libs/checkpoint` | 状态持久化与检查点机制 |
| `libs/checkpoint-postgres` | PostgreSQL 检查点后端 |
| `libs/prebuilt` | 预构建的 Agent 和工具组件 |
| `libs/sdk-py` | Python SDK |

资料来源：[AGENTS.md:1]()

## 核心概念

### 图（Graph）结构

LangGraph 使用**状态图（StateGraph）**作为核心抽象。图由以下基本元素组成：

```mermaid
graph LR
    A[节点 Node] -->|边 Edge| B[节点 Node]
    A --> C[条件边 Conditional Edge]
    C -->|路由| D[节点 1]
    C -->|路由| E[节点 2]
```

| 元素 | 说明 |
|------|------|
| **节点 (Node)** | 图中的计算单元，通常是 Python 函数或 Runnable |
| **边 (Edge)** | 定义节点之间的固定流转路径 |
| **条件边 (Conditional Edge)** | 根据状态动态选择下一个节点 |
| **状态 (State)** | 在图执行过程中流转的共享数据 |

### 状态管理

状态通过 `TypedDict` 定义，支持**注解式状态更新**：

```python
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    remaining_steps: NotRequired[int]
```

资料来源：[libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:35-40]()

LangGraph 支持**消息累积模式**，通过 `Annotated` 和 reducer 函数实现状态的增量更新：

```python
# 消息合并：左边的消息列表与右边的消息列表连接
def add_messages(left, right):
    if not isinstance(left, list):
        left = [left]
    if not isinstance(right, list):
        right = [right]
    return left + right
```

资料来源：[libs/cli/examples/graphs/storm.py:1-6]()

## 工具节点（ToolNode）

### 功能概述

`ToolNode` 是 LangGraph 提供的预构建组件，用于在图中执行工具调用。它支持：

- **并行执行**：多个工具调用可以同时执行，提高效率
- **错误处理**：可定制的错误消息和异常处理机制
- **状态注入**：工具可以访问和修改图状态
- **存储注入**：工具可以访问持久化存储

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_node.py:1-40]()

### 典型用法

```python
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def my_tool(x: int) -> str:
    return f"Result: {x}"

tool_node = ToolNode([my_tool])
```

### 工具执行上下文

`ToolRuntime` 类封装了工具执行所需的全部上下文信息：

```python
@dataclass
class ToolRuntime:
    state: Any  # 当前图状态
    context: Any  # 不可变上下文
    config: RunnableConfig  # 运行配置
    store: BaseStore  # 持久化存储
```

## 检查点机制（Checkpointing）

### 核心功能

检查点机制允许图在任意时刻暂停和恢复，这是实现**人机协作（Human-in-the-Loop）**和**长时间任务**的基础。

### 支持的类型

LangGraph 的消息序列化支持以下核心类型：

| 类型 | 模块路径 |
|------|----------|
| Send | `langgraph.types.Send` |
| Interrupt | `langgraph.types.Interrupt` |
| Command | `langgraph.types.Command` |
| TimeoutPolicy | `langgraph.types.TimeoutPolicy` |
| StateSnapshot | `langgraph.types.StateSnapshot` |
| PregelTask | `langgraph.types.PregelTask` |

资料来源：[libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py:100-120]()

### PostgreSQL 检查点后端

`langgraph-checkpoint-postgres` 提供了基于 PostgreSQL 的检查点持久化：

```python
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(conn_string)
```

数据库迁移通过 `MIGRATIONS` 列表管理，每个条目代表一个版本：

资料来源：[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:30-50]()

## 执行模型

### 工作流执行

LangGraph 的图执行遵循以下流程：

```mermaid
graph TD
    A[编译图] --> B[初始状态]
    B --> C[执行节点]
    C --> D{是否有检查点?}
    D -->|是| E[保存状态]
    D -->|否| F{是否中断?}
    E --> F
    F -->|是| G[暂停执行]
    F -->|否| H{是否有下一节点?}
    H -->|是| C
    H -->|否| I[完成]
    G --> J[恢复执行]
    J --> C
```

### 命令与路由

`Command` 类型用于从节点返回时指定下一步路由：

```python
Command(
    resume={"key": "value"},
    goto="next_node"
)
```

这使得节点可以：
- 指定下一个要执行的节点
- 向状态中添加数据
- 触发中断等待人工输入

## 高级特性

### 子图（Subgraphs）

LangGraph 支持**嵌套图**，允许将复杂逻辑封装为子图：

```python
 subgraph = StateGraph(SubGraphState)
 subgraph.add_node("process", process_node)
 subgraph.add_edge("start", "process")
 subgraph.add_edge("process", "end")
 sub_graph = subgraph.compile()
 
 parent_graph.add_node("subgraph_node", sub_graph)
```

### 条件分支

条件边根据状态动态选择目标节点：

```python
def route_based_on_confidence(state):
    if state["confidence"] > 0.8:
        return "high_confidence_path"
    return "low_confidence_path"

parent_graph.add_conditional_edges(
    "classifier",
    route_based_on_confidence,
    {
        "high_confidence_path": "approve",
        "low_confidence_path": "review"
    }
)
```

## 部署与集成

### LangGraph Server

通过 `langgraph.json` 配置文件部署：

```json
{
  "$schema": "https://langgra.ph/schema.json",
  "python_version": "3.12",
  "dependencies": [
    "langchain_community",
    "langchain_anthropic",
    "langchain_openai"
  ],
  "graphs": {
    "agent": "./agent.py:graph"
  }
}
```

资料来源：[libs/cli/examples/graphs/langgraph.json:1-12]()

### LangGraph CLI

`langgraph dev` 命令启动本地开发服务器，支持热重载和流式输出。

### 与 LangChain 集成

LangGraph 与 LangChain 深度集成，可以直接使用：

- LangChain 的 Chat Model 封装
- LangChain 的 Tool 接口
- LangChain 的 Memory 组件
- LangChain 的 Vector Store 集成

## 常见问题与社区热点

### 用户反馈的典型问题

| 问题类型 | 相关 Issue | 说明 |
|----------|------------|------|
| 状态序列化错误 | #4956, #5891 | `Type is not msgpack serializable` 错误 |
| Agent 无限循环 | #6731 | 递归限制错误 |
| 检查点持久化 | #5790, #3716 | PostgreSQL SSL 错误、配置被忽略 |
| 端口占用 | #7688 | TIME-WAIT 导致端口误报占用 |

### v1 路线图讨论

社区正在讨论以下重要改进方向：

- **Context API**：更直观的状态传递机制
- **错误处理增强**：更健壮的节点异常处理
- **合规性支持**：符合监管要求的人机协作示例

资料来源：[#4973: LangGraph v1 roadmap](https://github.com/langchain-ai/langgraph/issues/4973)

## 学习资源

| 资源 | 链接 |
|------|------|
| 官方文档 | [LangGraph Docs](https://docs.langchain.com/oss/python/langgraph/quickstart) |
| 快速入门 | [LangGraph Quickstart](https://docs.langchain.com/oss/python/langgraph/quickstart) |
| LangChain Academy | [Intro to LangGraph](https://academy.langchain.com/courses/intro-to-langgraph) |
| 社区论坛 | [LangChain Forum](https://forum.langchain.com) |
| JS 版本 | [langgraphjs](https://github.com/langchain-ai/langgraphjs) |

## 总结

LangGraph 是构建复杂 AI 应用的强大框架，其核心优势包括：

1. **图模型抽象**：将工作流自然地建模为有向图
2. **状态持久化**：通过检查点机制支持暂停和恢复
3. **灵活的路由**：支持固定边、条件边和命令式控制流
4. **丰富的组件**：ToolNode、预构建 Agent 等开箱即用
5. **多语言支持**：Python 和 JavaScript 两种实现

作为 LangChain 生态的核心组成部分，LangGraph 特别适合需要**长期运行**、**人机协作**或**复杂多代理交互**的应用场景。

---

<a id='page-installation'></a>

## 安装与配置

### 相关页面

相关主题：[LangGraph 简介](#page-introduction), [CLI 使用指南](#page-cli-usage)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/langgraph/pyproject.toml](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/pyproject.toml)
- [libs/cli/pyproject.toml](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/pyproject.toml)
- [libs/cli/langgraph_cli/schemas.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/schemas.py)
- [libs/cli/generate_schema.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/generate_schema.py)
- [libs/sdk-py/pyproject.toml](https://github.com/langchain-ai/langgraph/blob/main/libs/sdk-py/pyproject.toml)
- [libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
</details>

# 安装与配置

## 概述

LangGraph 是一个用于构建有状态、多角色应用程序的库，基于 LangChain 构建。其安装与配置体系涵盖多个组件库，包括核心库 (`langgraph`)、命令行工具 (`langgraph-cli`)、Python SDK (`sdk-py`) 以及多种检查点存储后端。本章节详细介绍各组件的安装方法、配置选项及其依赖关系。

## 核心依赖架构

LangGraph 的各组件之间存在明确的依赖关系，理解这些关系对于正确安装和配置至关重要。

```mermaid
graph TD
    A[checkpoint] --> B[checkpoint-postgres]
    A --> C[checkpoint-sqlite]
    A --> D[prebuilt]
    A --> E[langgraph]
    D --> E
    F[sdk-py] --> E
    F --> G[cli]
    H[sdk-js 独立]
```

根据项目文档，依赖关系声明如下：

- `checkpoint` 是基础检查点模块，被以下库依赖：
  - `checkpoint-postgres`
  - `checkpoint-sqlite`
  - `prebuilt`
  - `langgraph`
- `prebuilt` 依赖 `langgraph`
- `sdk-py` 依赖 `langgraph` 和 `cli`
- `sdk-js` 为独立包，无 Python 依赖

## 环境要求

### Python 版本支持

LangGraph 核心库对 Python 版本有明确要求。根据源码中的版本检查逻辑，版本低于 0.5 的 langgraph 与 checkpoint-postgres 存在兼容性问题。

| 组件 | 最低 Python 版本 | 备注 |
|------|-----------------|------|
| langgraph | 3.10+ | 推荐使用最新稳定版本 |
| langgraph-cli | 3.10+ | 与核心库版本一致 |
| sdk-py | 3.10+ | 依赖核心库 |

版本兼容性检查代码位于 `libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:19-26`：

```python
try:
    major, minor = get_version("langgraph").split(".")[:2]
    if int(major) == 0 and int(minor) < 5:
        warnings.warn(
            "You're using incompatible versions of langgraph and checkpoint-postgres. Please upgrade langgraph to avoid unexpected behavior.",
            DeprecationWarning,
            stacklevel=2,
        )
except Exception:
    # skip version check if running from source
    pass
```

### 运行时依赖

#### 核心库依赖

根据 `libs/langgraph/pyproject.toml`，核心库的主要依赖包括：

- `langchain-core` - LangChain 核心抽象
- `pydantic` - 数据验证
- `uuid` - 唯一标识生成

#### CLI 依赖

根据 `libs/cli/pyproject.toml`，CLI 组件额外依赖：

- `msgspec` - 高性能 JSON schema 处理
- `psycopg` (可选) - PostgreSQL 支持

## 安装方法

### 通过 pip 安装

#### 安装核心库

```bash
pip install langgraph
```

#### 安装 CLI 工具

```bash
pip install langgraph-cli
```

#### 安装检查点存储后端

```bash
# SQLite 后端
pip install langgraph-checkpoint-sqlite

# PostgreSQL 后端
pip install langgraph-checkpoint-postgres
```

#### 安装预构建组件

```bash
pip install langgraph-prebuilt
```

### 从源码安装

对于开发或最新功能，可以从源码安装：

```bash
# 克隆仓库
git clone https://github.com/langchain-ai/langgraph.git

# 安装核心库
cd libs/langgraph
pip install -e .

# 安装 CLI
cd ../cli
pip install -e .

# 安装检查点后端
cd ../checkpoint-postgres
pip install -e .
```

## LangGraph CLI 配置

### 配置文件结构

LangGraph CLI 使用 `langgraph.json` 配置文件来定义应用程序的运行方式。配置文件支持 JSON Schema 自动补全和验证。

配置文件结构定义在 `libs/cli/langgraph_cli/schemas.py`，主要包含以下配置类：

| 配置类 | 用途 |
|--------|------|
| `Config` | 主配置入口 |
| `GraphDef` | 图定义 |
| `StoreConfig` | 状态存储配置 |
| `IndexConfig` | 索引配置 |
| `AuthConfig` | 认证配置 |
| `SecurityConfig` | 安全配置 |
| `HttpConfig` | HTTP 服务器配置 |
| `CorsConfig` | CORS 跨域配置 |
| `CacheConfig` | 缓存配置 |
| `ThreadTTLConfig` | 线程 TTL 配置 |
| `CheckpointerConfig` | 检查点配置 |
| `SerdeConfig` | 序列化/反序列化配置 |
| `TTLConfig` | TTL 配置 |
| `ConfigurableHeaderConfig` | 可配置 Header 配置 |
| `WebhooksConfig` | Webhook 配置 |
| `WebhookUrlPolicy` | Webhook URL 策略 |

### 基础配置示例

一个典型的 `langgraph.json` 配置文件：

```json
{
  "python_version": "3.11",
  "pip_config_file": "./requirements.txt",
  "graphs": {
    "my_agent": "./agent.py:graph"
  },
  "env": [
    "OPENAI_API_KEY"
  ]
}
```

### 配置约束

配置文件中的 Python 版本与 Node.js 版本互斥，不能同时指定：

```python
# 来自 libs/cli/generate_schema.py:86-100
# Create two subschemas: one with python_version and one with node_version
# Define properties specific to Python projects
python_specific_props = ["python_version", "pip_config_file"]
# Define properties specific to Node.js projects
node_specific_props = ["node_version"]
```

## 检查点配置

### 检查点器基础配置

检查点 (Checkpoint) 是 LangGraph 实现状态持久化和时间旅行功能的核心机制。基础检查点器接口定义在 checkpoint 模块中。

### PostgreSQL 检查点配置

使用 PostgreSQL 作为检查点存储需要额外配置：

```python
from langgraph.checkpoint.postgres import PostgresSaver

# 使用连接字符串
saver = PostgresSaver.from_conn_string("postgresql://user:pass@host:5432/db")

# 或使用单独配置
saver = PostgresSaver(
    host="localhost",
    port=5432,
    database="langgraph",
    user="user",
    password="pass",
    ssl=True  # 启用 SSL
)
```

### 数据库迁移

PostgreSQL 检查点器会自动执行必要的数据库迁移。根据 `libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py`，迁移脚本存储在 `MIGRATIONS` 列表中：

```python
MIGRATIONS = [
    """CREATE TABLE IF NOT EXISTS checkpoint_migrations (
    v INTEGER PRIMARY KEY
);""",
    """CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    ...
```
```

### SQLite 检查点配置

SQLite 适用于开发环境和轻量级部署：

```python
from langgraph.checkpoint.sqlite import SqliteSaver

saver = SqliteSaver.from_conn_string("sqlite:///langgraph.db")
```

## 环境变量配置

### 必需环境变量

根据项目配置，通常需要设置以下环境变量：

| 变量名 | 描述 | 示例 |
|--------|------|------|
| `OPENAI_API_KEY` | OpenAI API 密钥 | `sk-...` |
| `LANGCHAIN_API_KEY` | LangChain API 密钥 | 可选 |
| `LANGCHAIN_TRACING_V2` | 启用追踪 | `true` |

### 配置注入方式

LangGraph CLI 支持通过 `env` 字段注入环境变量：

```json
{
  "env": [
    "OPENAI_API_KEY",
    "DATABASE_URL",
    "CUSTOM_VAR=default_value"
  ]
}
```

## SDK 配置

### Python SDK

根据 `libs/sdk-py/pyproject.toml`，Python SDK 提供远程 LangGraph 部署的接口：

- 支持 WebSocket 流式传输
- 支持线程管理
- 支持子图管理

### JavaScript SDK

JavaScript SDK (`sdk-js`) 为独立包，可在 Node.js 和浏览器环境中使用，无需 Python 环境。

## 常见配置问题与解决

### 版本不兼容问题

**问题**：使用 langgraph < 0.5 与 checkpoint-postgres 组合时出现警告。

**解决方案**：升级 langgraph 到 0.5 或更高版本：

```bash
pip install --upgrade langgraph langgraph-checkpoint-postgres
```

### 模块导入错误

**问题**：`ModuleNotFoundError: No module named 'langgraph.graph.graph'`

**原因**：这是已知的版本兼容性问题。**解决方案**：确保使用匹配的库版本，从源码安装时使用同一仓库的版本。

### 检查点存储问题

**问题**：psycopg SSL 连接错误

**背景**：社区报告了多个版本的 PostgreSQL 检查点 SSL 错误 (`psycopg.OperationalError: sending query and params failed: SSL error: bad length`)。

**解决方案**：
1. 升级 `langgraph-checkpoint-postgres` 到最新版本
2. 检查 PostgreSQL 服务器的 SSL 配置
3. 考虑使用 SQLite 后端进行开发环境测试

## 配置验证

### 使用 JSON Schema 验证

LangGraph CLI 提供 JSON Schema 用于 IDE 自动补全和配置验证：

```bash
# 生成 schema（通过 generate_schema.py 脚本）
python libs/cli/generate_schema.py
```

生成的 `schema.json` 文件可用于验证 `langgraph.json` 配置。

### 运行时配置检查

在启动服务前验证配置：

```bash
langgraph studio --config ./langgraph.json --dry-run
```

## 最佳实践

### 开发环境

- 使用 SQLite 检查点器简化本地开发
- 启用详细日志便于调试
- 使用 `.env` 文件管理环境变量

### 生产环境

- 使用 PostgreSQL 检查点器确保数据持久化
- 配置 SSL/TLS 加密连接
- 设置适当的线程 TTL 和检查点间隔
- 使用外部存储管理大状态对象

### 安全配置

- 不要在配置文件中硬编码密钥
- 使用环境变量注入敏感信息
- 启用 CORS 限制访问来源
- 配置适当的认证机制

## 相关资源

- [LangGraph 快速入门](https://docs.langchain.com/oss/python/langgraph/quickstart)
- [贡献指南](https://docs.langchain.com/oss/python/contributing/overview)
- [LangChain 论坛](https://forum.langchain.com) - 社区支持

---

<a id='page-core-architecture'></a>

## 核心架构

### 相关页面

相关主题：[Pregel 执行引擎](#page-pregel-execution), [图构建与状态定义](#page-graph-construction)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/langgraph/langgraph/pregel/main.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/main.py)
- [libs/langgraph/langgraph/pregel/_loop.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_loop.py)
- [libs/langgraph/langgraph/pregel/_algo.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_algo.py)
- [libs/langgraph/langgraph/config.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/config.py)
- [libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)
- [libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
- [libs/prebuilt/langgraph/prebuilt/tool_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_node.py)
- [libs/langgraph/langgraph/_internal/_pydantic.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/_internal/_pydantic.py)
</details>

# 核心架构

## 概述

LangGraph 是一个基于** Pregel 算法**和 **Apache Beam** 编程模型构建的图状态机框架。其核心架构受 [Google Pregel 论文](https://research.google/pubs/pub37252/) 启发，提供了有向状态图执行引擎，支持状态持久化、条件分支、子图嵌套和中断恢复等能力。

LangGraph 的核心设计目标：

- **状态持久化**：通过 Checkpoint 机制支持图执行的暂停、恢复和重放
- **确定性执行**：基于 checkpoint 的状态管理确保执行可重现
- **图可视化**：支持从代码生成可绘制的图结构（类似 NetworkX）
- **Human-in-the-Loop**：支持在执行过程中插入人工干预点

## 核心组件

### Pregel 内核

LangGraph 的执行引擎位于 `langgraph.pregel` 模块，核心文件包括：

| 文件 | 职责 |
|------|------|
| `main.py` | `Pregel` 主类，对外暴露 `invoke`、`ainvoke`、`stream` 等 API |
| `_loop.py` | 图执行循环逻辑，处理状态更新和任务调度 |
| `_algo.py` | 核心算法实现，包括 Send、Wait、Interrupt 等指令处理 |

### 核心类型系统

```mermaid
classDiagram
    class Interrupt {
        +value: Any
        +when: str
    }
    class Command {
        +goto: str | Send
        +update: dict | None
    }
    class Send {
        +node: str
        +arg: Any
    }
    class StateSnapshot {
        +config: RunnableConfig
        +values: dict
        +next_nodes: tuple
        + tasks: tuple
    }
    class PregelTask {
        +id: str
        +name: str
        +input: Any
        +interrupts: tuple
    }
```

LangGraph 定义了以下核心类型（定义于 `langgraph.types`）：

| 类型 | 说明 | 序列化支持 |
|------|------|-----------|
| `Interrupt` | 中断信号，包含中断值和触发时机 | Msgpack 可序列化 |
| `Command` | 控制指令，用于跳转节点和状态更新 | Msgpack 可序列化 |
| `Send` | 并发发送任务到目标节点 | ⚠️ 序列化问题（参见 #5891）|
| `StateSnapshot` | 图执行快照，包含完整状态 | Msgpack 可序列化 |
| `PregelTask` | 单个执行任务单元 | Msgpack 可序列化 |
| `Overwrite` | 状态覆盖指令 | Msgpack 可序列化 |

## 状态图（StateGraph）

### 图结构定义

StateGraph 是 LangGraph 的核心 API，用于定义有向状态机：

```mermaid
graph LR
    START([START]) -->|边| Node1
    Node1 -->|条件边| Node2
    Node1 -->|条件边| Node3
    Node2 --> END
    Node3 --> END
```

关键概念：

- **节点（Node）**：Python 函数或 Runnable，执行状态转换
- **边（Edge）**：定义节点间的数据流向
- **条件边（Conditional Edge）**：基于状态动态选择目标节点
- **入口点（Entry Point）**：图的起始节点（默认 `START`）
- **终止点（End Point）**：图的结束节点（默认 `END`）

### 状态定义

状态通过 `TypedDict` 定义，支持多种数据类型：

```python
class ResearchState(TypedDict):
    topic: str
    outline: Outline
    editors: list[Editor]
    interview_results: list[InterviewState]
    sections: list[WikiSection]
    article: str
```

## Checkpoint 机制

### 设计目的

Checkpoint 是 LangGraph 实现**状态持久化**和**恢复执行**的核心机制：

1. **故障恢复**：图执行中断后可从最近 checkpoint 恢复
2. **多轮对话**：支持跨请求维护对话状态
3. **人工介入**：支持在 checkpoint 处暂停等待人工审批

### 检查点数据结构

```python
# 来自 checkpoint/postgres/base.py
MIGRATIONS = [
    """CREATE TABLE IF NOT EXISTS checkpoint_migrations (
    v INTEGER PRIMARY KEY
);""",
    """CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    ...
)"""
]
```

### Checkpointer 层级

| Checkpointer | 存储后端 | 适用场景 |
|-------------|---------|---------|
| `MemorySaver` | 内存 | 开发测试 |
| `PostgresSaver` | PostgreSQL | 生产环境持久化 |
| `RedisSaver` | Redis | 高性能缓存 |
| `SqliteSaver` | SQLite | 轻量级部署 |

### 已知问题

**Issue #3716**：跨版本使用 `langgraph-checkpoint-postgres` 时可能出现 SSL 连接错误：

```
psycopg.OperationalError: sending query and params failed: SSL error: bad length
```

**Issue #7417**：长时间运行的任务（>180秒）可能从 checkpoint 被静默重新执行，导致重复工作。

## 消息序列化（Msgpack）

### 序列化类型白名单

LangGraph 使用 Msgpack 进行状态序列化，仅允许以下类型：

```python
# 来自 libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py
SAFE_MSGPACK_TYPES: dict[str, tuple[str, ...]] = {
    "langchain_core.messages.human": ("HumanMessage", "HumanMessageChunk"),
    "langchain_core.messages.ai": ("AIMessage", "AIMessageChunk"),
    "langchain_core.messages.tool": ("ToolMessage", "ToolMessageChunk"),
    "langgraph.types": ("Send", "Interrupt", "Command", "StateSnapshot", "PregelTask"),
    "langgraph.store.base": ("Item", "GetOp"),
}
```

### 已知序列化问题

**Issue #4956**：`TypeError: Type is not msgpack serializable: AIMessage`

**Issue #5891**：`TypeError: Type is not msgpack serializable: Send` - 在使用 `create_react_agent` 配合 `BaseStore` 时触发

## 执行流程

### 主循环（Pregel Loop）

```mermaid
sequenceDiagram
    participant Client
    participant Pregel
    participant Checkpointer
    participant Nodes

    Client->>Pregel: invoke(input, config)
    Pregel->>Checkpointer: get_checkpoint(config)
    alt 有历史checkpoint
        Checkpointer-->>Pregel: 恢复状态
    else 无checkpoint
        Pregel->>Pregel: 初始化空状态
    end
    Pregel->>Nodes: 执行当前节点
    loop 直到结束
        Nodes->>Pregel: 返回更新
        alt 遇到Interrupt
            Pregel->>Checkpointer: 保存checkpoint
            Pregel-->>Client: 中断返回
        else 正常完成
            Pregel->>Nodes: 执行下一节点
        end
    end
    Pregel-->>Client: 返回最终结果
    Pregel->>Checkpointer: 保存最终checkpoint
```

### 任务调度算法

在 `_algo.py` 中实现的核心调度逻辑：

1. **Send 指令**：将任务发送到指定节点
2. **Wait 指令**：等待所有并发任务完成
3. **Interrupt 检查**：每次迭代前检查是否需要中断
4. **递归限制**：通过 `recursion_limit` 配置防止无限循环

**Issue #6731**：某些情况下 Agent 会无限循环直到达到递归限制。这是由于 checkpoint 恢复后的状态与当前执行状态不一致导致的。

## 配置系统

### Config 结构

```python
# 来自 libs/langgraph/langgraph/config.py
@dataclass
class Config:
    tags: Sequence[str]
    metadata: Mapping[str, Any]
    callbacks: Callbacks
    recursion_limit: int
    configurable: dict[str, Any]
```

### Context API（v1 改进方向）

**Issue #5023** 中提到的 `config.configurable` → `context` API 迁移旨在提供更直观的上下文管理方式，减少开发者的困惑。

## 依赖注入与错误处理

### ToolInvocationError

当工具调用失败时抛出此异常：

```python
# 来自 libs/prebuilt/langgraph/prebuilt/tool_node.py
class ToolInvocationError(ToolException):
    def __init__(
        self,
        tool_name: str,
        source: ValidationError,
        tool_kwargs: dict[str, Any],
        filtered_errors: list[ErrorDetails] | None = None,
    ) -> None:
```

### 工具输出格式化

ToolNode 自动处理多种输出格式：

| 输出类型 | 处理方式 |
|---------|---------|
| `str` | 直接返回 |
| `list[dict]` (content blocks) | 直接返回 |
| 其他对象 | JSON 序列化或 `str()` 转换 |

## CLI 工具

### schema.json 生成

`libs/cli/generate_schema.py` 脚本为 `langgraph.json` 配置文件生成 JSON Schema，支持 IDE 自动补全和验证：

```python
# 生成命令
python libs/cli/generate_schema.py
```

生成的 Schema 包含以下配置项：

| 配置项 | 说明 |
|-------|------|
| `GraphDef` | 图定义 |
| `HttpConfig` | HTTP 服务器配置 |
| `SecurityConfig` | 安全配置 |
| `CheckpointerConfig` | Checkpointer 配置 |
| `StoreConfig` | 存储配置 |

**Issue #5790**：`langgraph dev` 忽略 Checkpointer 配置，强制使用内存存储。

**Issue #7688**：`langgraph dev` 错误报告端口占用，可能由 TIME-WAIT 状态引起。

## 技术债务与未来方向

### 待改进领域

1. **错误处理增强**（Issue #6170）：需要更健壮的节点错误处理机制，可能通过 Hooks 或 Middleware 实现
2. **序列化兼容性**：解决 Send、BaseStore 等类型的序列化问题
3. **并发安全**：修复长时间任务静默重执行问题

### v1 路线图重点

- 简化 `StateGraph` API
- 改进 Context 管理（Issue #5023）
- 更完善的中断和恢复机制
- 合规性支持（Issue #7687）

## 参考资料

- [LangGraph Pregel 核心实现](libs/langgraph/langgraph/pregel/main.py)
- [Pregel 执行循环](libs/langgraph/langgraph/pregel/_loop.py)
- [算法与指令处理](libs/langgraph/langgraph/pregel/_algo.py)
- [Checkpoint 序列化](libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)
- [PostgreSQL Checkpoint 实现](libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
- [工具节点实现](libs/prebuilt/langgraph/prebuilt/tool_node.py)

---

<a id='page-graph-construction'></a>

## 图构建与状态定义

### 相关页面

相关主题：[核心架构](#page-core-architecture), [通道与状态管理](#page-channels)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/langgraph/langgraph/graph/state.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/graph/state.py)
- [libs/langgraph/langgraph/graph/_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/graph/_node.py)
- [libs/langgraph/langgraph/graph/_branch.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/graph/_branch.py)
- [libs/langgraph/langgraph/graph/message.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/graph/message.py)
- [libs/langgraph/langgraph/pregel/main.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/main.py)
- [libs/prebuilt/langgraph/prebuilt/tool_validator.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_validator.py)
- [libs/cli/examples/graphs/storm.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/examples/graphs/storm.py)
- [libs/langgraph/langgraph/_internal/_pydantic.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/_internal/_pydantic.py)
</details>

# 图构建与状态定义

## 概述

LangGraph 的核心能力之一是通过 **StateGraph** API 构建有状态的工作流图。在 LangGraph 中，图（Graph）是工作流的抽象表示，状态（State）是贯穿整个图执行过程的共享数据结构。

图构建系统允许开发者：

- 定义工作流的拓扑结构（节点和边）
- 声明状态模式（State Schema）来约束数据流
- 使用条件逻辑动态控制执行路径
- 通过 checkpoint 实现状态持久化和恢复

LangGraph 的图构建采用声明式 API，参考了 Google 的 [Pregel](https://research.google/pubs/pub37252/) 和 [Apache Beam](https://beam.apache.org/) 的设计理念，同时借鉴了 [NetworkX](https://networkx.org/documentation/latest/) 的公共接口风格。资料来源：[README.md]()

## 核心概念

### 状态模式（State Schema）

状态模式定义了图的全局数据结构，通常使用 Python 的 `TypedDict` 或 Pydantic 的 `BaseModel` 来声明。

```python
from typing import TypedDict, Annotated, Sequence
from typing_extensions import NotRequired

class AgentState(TypedDict):
    """代理状态定义"""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    remaining_steps: NotRequired[int]
```

状态可以是：
- **TypedDict**：推荐用于简单状态场景
- **BaseModel**：适用于需要验证、默认值等高级功能的场景

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_validator.py:58-67]()

### 状态更新机制

LangGraph 使用注释式语法（Annotated）来实现状态更新策略。最常见的更新策略是 `add_messages`，它将新消息追加到现有消息列表中：

```python
from typing import Annotated, Sequence
from langgraph.graph.message import add_messages

class GraphState(TypedDict):
    messages: Annotated[list, add_messages]  # 新消息追加到列表
```

内置的更新策略包括：

| 更新策略 | 说明 | 源码位置 |
|---------|------|----------|
| `add_messages` | 将新消息追加到列表末尾 | [libs/langgraph/langgraph/graph/message.py]() |
| `operator.add` | 执行字典/值的加法合并 | 内置 |
| 自定义函数 | 支持自定义状态合并逻辑 | 用户定义 |

资料来源：[libs/langgraph/langgraph/graph/message.py]()

## StateGraph API

### 基本构建流程

```mermaid
graph TD
    A[创建 StateGraph] --> B[定义状态模式]
    B --> C[添加节点]
    C --> D[添加边]
    D --> E[设置入口和结束点]
    E --> F[编译图]
```

StateGraph 的基本使用流程：

```python
from langgraph.graph import StateGraph, END, START

# 1. 创建图构建器，指定状态模式
builder = StateGraph(AgentState)

# 2. 添加节点
builder.add_node("node_name", my_function)

# 3. 添加边
builder.add_edge(START, "first_node")
builder.add_edge("node_name", END)

# 4. 编译图
graph = builder.compile()
```

资料来源：[libs/langgraph/langgraph/graph/state.py]()

### 节点定义

节点是图中的计算单元，可以是：

- **普通函数**：同步或异步的 Python 函数
- **Runnable 对象**：实现 LangChain Runnable 接口的对象
- **子图**：嵌套的已编译图

```python
# 同步节点函数
def process_node(state: AgentState) -> AgentState:
    return {"messages": [...]}  # 返回状态更新

# 异步节点函数
async def async_process_node(state: AgentState) -> AgentState:
    result = await some_async_operation()
    return {"messages": result}
```

资料来源：[libs/langgraph/langgraph/graph/_node.py]()

### 边定义

边连接图中的节点，定义了执行流程。

#### 普通边

```python
# 从源节点到目标节点的固定转换
builder.add_edge("node_a", "node_b")
```

#### 条件边

条件边允许根据状态动态选择下一个节点：

```python
def route_function(state: AgentState) -> str:
    if some_condition:
        return "node_a"
    return "node_b"

builder.add_conditional_edges(
    "source_node",
    route_function,
    {"node_a": "node_a", "node_b": "node_b"}
)
```

条件边支持三种路由模式：

| 模式 | 说明 | 返回值示例 |
|------|------|-----------|
| 字符串路由 | 返回目标节点名称 | `"node_a"` |
| 字典型路由 | 返回节点名称到目标节点的映射 | `{"path1": "node_a", "path2": "node_b"}` |
| 字面量路由 | 使用 `Literal` 类型精确指定 | `Literal["node_a", "node_b"]` |

资料来源：[libs/langgraph/langgraph/graph/_branch.py]()

### 特殊节点标识

LangGraph 提供了两个特殊的起始/结束标识：

- `START`：图的入口节点
- `END`：图的终止节点

```python
builder.add_edge(START, "first_node")
builder.add_edge("final_node", END)
```

资料来源：[libs/langgraph/langgraph/graph/state.py]()

## 高级状态模式

### Pydantic 模型状态

对于需要复杂验证的场景，可以使用 Pydantic BaseModel：

```python
from pydantic import BaseModel, Field
from typing import Optional

class UserProfile(BaseModel):
    name: str = Field(description="用户名")
    age: int = Field(ge=0, le=150)
    email: Optional[str] = None
    tags: list[str] = Field(default_factory=list)
```

LangGraph 内部通过 `_create_model_cached` 函数优化 Pydantic 模型的创建：

```python
@lru_cache(maxsize=256)
def _create_model_cached(
    model_name: str,
    /,
    **field_definitions: Any,
) -> type[BaseModel]:
    return _create_model_base(
        model_name,
        __config__=_SchemaConfig,
        **_remap_field_definitions(field_definitions),
    )
```

资料来源：[libs/langgraph/langgraph/_internal/_pydantic.py]()

### 嵌套状态与子图

复杂应用可能需要嵌套状态。子图可以作为节点嵌入到父图中：

```python
# 创建子图
subgraph_builder = StateGraph(SubGraphState)
subgraph_builder.add_node("sub_node", sub_node_function)
subgraph_builder.add_edge(START, "sub_node")
subgraph_builder.add_edge("sub_node", END)
subgraph = subgraph_builder.compile()

# 在父图中添加子图作为节点
parent_builder = StateGraph(ParentState)
parent_builder.add_node("subgraph_node", subgraph)  # 传入编译后的子图
```

## 消息状态处理

LangGraph 提供了专门用于消息处理的 `add_messages` 注释，这是构建对话代理时的核心工具。

### add_messages 详解

```python
from langgraph.graph.message import add_messages, add_messages

class ConversationState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
```

`add_messages` 的合并逻辑：

1. 如果左值为空，创建新列表
2. 如果右值为空，返回左值
3. 否则，将左右列表连接

```python
def add_messages(left, right):
    if not isinstance(left, list):
        left = [left]
    if not isinstance(right, list):
        right = [right]
    return left + right
```

资料来源：[libs/cli/examples/graphs/storm.py]()

### 消息类型的序列化

消息必须可序列化才能进行 checkpoint。以下是常见的问题：

> ⚠️ **已知问题**：`AIMessage` 等 LangChain 消息类型默认不可通过 msgpack 序列化，可能导致 `TypeError: Type is not msgpack serializable: AIMessage` 错误。

资料来源：[社区 Issue #4956](https://github.com/langchain-ai/langgraph/issues/4956)

## 配置与上下文

### Configurable 字段

`configurable` 字段允许在运行时传递配置参数：

```python
class MyState(TypedDict):
    config_param: str
```

使用运行时配置：

```python
config = {"configurable": {"config_param": "value"}}
result = graph.invoke(initial_state, config=config)
```

> 💡 **设计讨论**：社区正在讨论将 `config.configurable` API 改进为更直观的 `context` API，以减少开发者的困惑。

资料来源：[社区 Issue #5023](https://github.com/langchain-ai/langgraph/issues/5023)

### 状态验证

#### ValidationNode（已弃用）

`ValidationNode` 用于验证工具调用的参数结构：

```python
from langgraph.prebuilt import ValidationNode

validation_node = ValidationNode(tools=[...])
```

> ⚠️ **弃用警告**：`ValidationNode` 已弃用，推荐使用 `create_agent` from `langchain.agents` 配合自定义工具错误处理。

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_validator.py:19-21]()

#### 自定义状态验证

可以创建自定义的验证节点：

```python
class ValidatedState(TypedDict):
    validated_field: str

def validate_and_transform(state: InputState) -> ValidatedState:
    # 执行验证和转换逻辑
    return {"validated_field": validated_value}
```

### InvalidUpdateError 处理

状态更新必须写入至少一个字段，否则会抛出 `InvalidUpdateError`：

> ⚠️ **常见错误**：`langgraph.errors.InvalidUpdateError: Must write to at least one of ['input', 'plan', 'past_steps', 'response']`

资料来源：[社区 Issue #740](https://github.com/langchain-ai/langgraph/issues/740)

## 图编译与执行

### compile() 方法

编译后的图是一个 `Pregel` 对象：

```python
from langgraph.graph import StateGraph, END, START

builder = StateGraph(AgentState)
# ... 添加节点和边 ...
compiled_graph = builder.compile(
    checkpointer=None,  # 可选：checkpoint 持久化
    interrupt_before=None,  # 可选：中断点
    interrupt_after=None,  # 可选：中断点
)
```

编译参数：

| 参数 | 类型 | 说明 |
|------|------|------|
| `checkpointer` | `BaseCheckpointSaver | None` | 状态持久化检查器 |
| `interrupt_before` | `Sequence[str] | None` | 执行前中断的节点列表 |
| `interrupt_after` | `Sequence[str] | None` | 执行后中断的节点列表 |
| `debug` | `bool` | 启用调试日志 |

资料来源：[libs/langgraph/langgraph/pregel/main.py]()

### 运行时状态访问

已编译的图提供多种状态访问方法：

```python
# 获取 JSON Schema
schema = compiled_graph.get_context_jsonschema()

# 旧版 API（已弃用）
deprecated_schema = compiled_graph.config_schema()
```

> ⚠️ **弃用提示**：`config_schema()` 和 `get_config_jsonschema()` 已弃用，请使用 `get_context_jsonschema()`。

资料来源：[libs/langgraph/langgraph/pregel/main.py:280-310]()

## 常见模式

### 代理循环检测

代理可能陷入无限循环直到达到递归限制：

> ⚠️ **已知 Bug**：在某些情况下，代理会无限循环直到触发递归限制错误。

资料来源：[社区 Issue #6731](https://github.com/langchain-ai/langgraph/issues/6731)

### 带剩余步数限制的循环

```python
from typing import Annotated, Literal
from langgraph.constants import END
from langgraph.pregel import RetryPolicy

MAX_TURNS = 10

def should_continue(state: AgentState) -> Literal["continue", "end"]:
    if len(state["messages"]) > MAX_TURNS * 2:
        return "end"
    return "continue"

builder.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "action", "end": END}
)
```

### 条件边中的 None 值处理

> ⚠️ **已知 Bug**：当节点返回 `Command[Literal[...]]` 且该节点同时使用 `add_conditional_edges` 时，调用 `graph.get_graph()` 可能失败并抛出 `TypeError: '<' not supported between instances of 'NoneType' and 'str'`。

资料来源：[社区 Issue #7691](https://github.com/langchain-ai/langgraph/issues/7691)

## 最佳实践

### 1. 状态设计原则

- 保持状态简洁，只存储必要的数据
- 使用 TypedDict 声明状态以获得类型检查支持
- 对于复杂验证场景使用 Pydantic BaseModel

### 2. 节点设计原则

- 节点函数应该幂等（相同输入产生相同输出）
- 避免在节点中修改外部状态
- 使用类型注解明确输入输出状态

### 3. 边设计原则

- 使用条件边处理分支逻辑
- 避免过深的嵌套条件边
- 为所有可能的执行路径定义终点

### 4. Checkpoint 配置

对于生产环境，使用持久化 checkpointer：

```python
from langgraph.checkpoint.postgres import PostgresSaver

# 开发环境：内存 checkpointer
in_memory_checkpointer = MemorySaver()

# 生产环境：PostgreSQL checkpointer
postgres_checkpointer = PostgresSaver.from_conn_string(connection_string)

graph = builder.compile(checkpointer=postgres_checkpointer)
```

> ⚠️ **已知问题**：PostgreSQL checkpointer 在某些版本中存在 SSL 错误问题。

资料来源：[社区 Issue #3716](https://github.com/langchain-ai/langgraph/issues/3716)

## 源码结构

### 核心文件

```
libs/langgraph/langgraph/graph/
├── state.py          # StateGraph 主类实现
├── _node.py          # 节点定义和处理
├── _branch.py        # 条件边和分支逻辑
├── message.py        # 消息状态处理
└── __init__.py       # 公共 API 导出
```

### 状态管理核心

```
libs/langgraph/langgraph/pregel/
└── main.py           # Pregel 运行时实现，包含图的编译和执行逻辑
```

## 总结

LangGraph 的图构建与状态定义系统提供了：

1. **声明式 API**：通过 TypedDict 或 BaseModel 定义状态
2. **灵活的节点类型**：支持函数、Runnable 和子图
3. **动态路由**：通过条件边实现复杂的执行流程
4. **状态持久化**：通过 checkpointer 支持断点续执
5. **类型安全**：完整的类型注解支持

这套系统使得构建复杂的多步骤 AI 工作流变得简单且可靠，同时保持了足够的灵活性来应对各种生产场景需求。

---

<a id='page-pregel-execution'></a>

## Pregel 执行引擎

### 相关页面

相关主题：[核心架构](#page-core-architecture), [检查点与持久化](#page-checkpointing)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/langgraph/langgraph/pregel/_runner.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_runner.py)
- [libs/langgraph/langgraph/pregel/_executor.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_executor.py)
- [libs/langgraph/langgraph/pregel/_retry.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_retry.py)
- [libs/langgraph/langgraph/pregel/_call.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/_call.py)
- [libs/langgraph/langgraph/types.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/types.py)
- [libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)
- [libs/prebuilt/langgraph/prebuilt/tool_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_node.py)
- [libs/langgraph/langgraph/_internal/_pydantic.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/_internal/_pydantic.py)
</details>

# Pregel 执行引擎

## 概述

Pregel 执行引擎是 LangGraph 的核心执行组件，源自 Google 的 Pregel 图处理论文灵感，用于协调 LangGraph 中的状态流和任务执行。它负责管理图中节点之间的消息传递、状态持久化、任务调度和并发执行。

```mermaid
graph TB
    subgraph "Pregel 执行引擎"
        Runner["_Runner<br/>任务编排器"]
        Executor["_Executor<br/>并发执行器"]
        Retry["_Retry<br/>重试策略"]
        Call["_Call<br/>任务调用"]
    end
    
    Runner --> Executor
    Executor --> Retry
    Retry --> Call
    Call -->|"状态更新"| Checkpoint["Checkpointer<br/>状态持久化"]
    
    subgraph "核心类型"
        Send["Send<br/>任务发送"]
        Command["Command<br/>控制命令"]
        Interrupt["Interrupt<br/>中断信号"]
    end
```

## 核心类型定义

### 状态与类型变量

```python
StateT = TypeVar("StateT")
OutputT = TypeVar("OutputT")
```

### 持久性模式

LangGraph 支持三种持久性模式，用于控制状态何时被持久化到检查点存储：

```python
Durability = Literal["sync", "async", "exit"]
```

| 模式 | 说明 |
|------|------|
| `sync` | 同步模式：变更在下一步开始前同步持久化 |
| `async` | 异步模式：变更在下一步执行时异步持久化 |
| `exit` | 退出模式：变更仅在图退出时持久化 |

资料来源：[libs/langgraph/langgraph/types.py:38-46]()

### 流输出类型

Pregel 引擎支持多种流输出模式，用于实时观测执行状态：

```python
class StreamMode(Enum):
    values = "values"      # 状态值
    updates = "updates"    # 增量更新
    messages = "messages"  # 消息流
    custom = "custom"      # 自定义流
    checkpoint = "checkpoint"  # 检查点
    tasks = "tasks"        # 任务信息
    debug = "debug"        # 调试信息
```

| 流模式 | 说明 |
|--------|------|
| `values` | 输出完整的状态值 |
| `updates` | 输出节点的增量更新 |
| `messages` | 输出消息（ChatMessage 流） |
| `custom` | 自定义数据流 |
| `checkpoint` | 输出检查点快照 |
| `tasks` | 输出任务执行信息 |
| `debug` | 输出调试信息 |

## 执行流程

### 任务编排

```mermaid
sequenceDiagram
    participant Client as 客户端
    participant Runner as _Runner
    participant Executor as _Executor
    participant Call as _Call
    participant Checkpointer as Checkpointer
    
    Client->>Runner: invoke/ainvoke
    Runner->>Runner: 初始化状态
    Runner->>Checkpointer: 加载检查点
    Runner->>Executor: 提交并发任务
    
    loop 执行循环
        Executor->>Call: 执行任务
        Call->>Call: 调用节点函数
        alt 需要重试
            Call->>Retry: 应用重试策略
        end
        Call-->>Executor: 任务结果
        Executor->>Checkpointer: 保存检查点
        Executor-->>Runner: 步骤完成
    end
    
    Runner-->>Client: 返回最终状态
```

### 重试策略

重试机制由 `_retry.py` 模块实现，支持灵活的错误处理：

```python
# 重试装饰器配置
@retry(
    stops=stops.after_attempt(3),
    wait=waits.exponential(multiplier=1, max=10),
    retry=retry_if_exception_type(ValueError)
)
async def task_with_retry():
    pass
```

资料来源：[libs/langgraph/langgraph/pregel/_retry.py]()

| 参数 | 说明 |
|------|------|
| `stops` | 停止条件（如最大重试次数） |
| `wait` | 等待策略（指数退避、固定延迟等） |
| `retry` | 判断条件（异常类型、返回值等） |

## 任务发送与命令控制

### Send 类型

`Send` 用于从当前节点并行发送到多个目标节点：

```python
class Send:
    """发送到图中的其他节点"""
    node: str
    arg: Any
```

使用示例：

```python
def route_node(state):
    # 并行发送到多个处理节点
    return [
        Send("processor_a", {"data": state["data"]}),
        Send("processor_b", {"data": state["data"]}),
    ]
```

### Command 类型

`Command` 提供显式的控制流控制，可中断、路由或更新状态：

```python
class Command:
    """控制命令"""
    goto: Optional[str] = None
    update: Optional[dict] = None
    resume: Optional[Any] = None
```

| 字段 | 说明 |
|------|------|
| `goto` | 指定下一个要执行的节点 |
| `update` | 更新状态值 |
| `resume` | 从中断点恢复执行 |

## 检查点与状态持久化

### 检查点机制

LangGraph 的检查点系统是实现状态持久化和恢复的核心。执行引擎在每个步骤后保存状态快照，允许从中断点恢复执行。

```mermaid
graph LR
    subgraph "检查点生命周期"
        Save1["步骤 1 完成<br/>保存检查点"]
        Save2["步骤 2 完成<br/>保存检查点"]
        Save3["步骤 3 完成<br/>保存检查点"]
        Int["中断发生"]
        Resume["恢复执行"]
    end
    
    Save1 --> Save2 --> Save3
    Save3 --> Int
    Int --> Resume
```

### Msgpack 序列化

检查点数据通过 msgpack 序列化以支持高效存储。`SAFE_MSGPACK_TYPES` 定义了允许序列化的类型：

```python
SAFE_MSGPACK_TYPES: dict = {
    # 消息类型
    ("langchain_core.messages", "AIMessage"),
    ("langchain_core.messages", "HumanMessage"),
    ("langchain_core.messages", "SystemMessage"),
    ("langchain_core.messages", "ToolMessage"),
    # LangGraph 类型
    ("langgraph.types", "Send"),
    ("langgraph.types", "TimeoutPolicy"),
    ("langgraph.types", "Interrupt"),
    ("langgraph.types", "Command"),
    ("langgraph.types", "StateSnapshot"),
}
```

资料来源：[libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py:1-50]()

> **注意**：社区问题 #4956 和 #5891 报告了 msgpack 序列化相关错误，如 `TypeError: Type is not msgpack serializable: AIMessage`。确保返回的数据类型在允许列表中。

## 并发执行模型

### 多节点并行

执行引擎支持从单个节点并行发送到多个目标节点：

```python
# 伪代码示例
async def parallel_execution():
    tasks = [
        execute_node("node_a", state),
        execute_node("node_b", state),
        execute_node("node_c", state),
    ]
    results = await asyncio.gather(*tasks)
    return merge_results(results)
```

### 工具节点执行

`ToolNode` 是预构建的工具执行组件，支持：

- **并行工具调用**：从单个 AIMessage 批量执行多个工具
- **错误处理**：可定制的错误消息格式化
- **状态注入**：通过 `InjectedState` 访问图状态
- **存储注入**：通过 `InjectedStore` 访问持久化存储

```python
from langgraph.prebuilt import ToolNode

tool_node = ToolNode([my_tool])
```

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_node.py:1-50]()

## 常见问题与最佳实践

### 无限循环问题

社区问题 #6731 报告了 Agent 无限循环直到达到递归限制的错误。解决建议：

1. 设置合理的 `recursion_limit`
2. 使用 `interrupt()` 在适当位置中断执行
3. 实现节点访问计数器防止重复执行

### 长工具调用处理

社区问题 #7417 描述了超过 180 秒的工具调用被静默重新执行的问题。建议：

- 使用长超时配置
- 实现幂等性检查
- 使用 `Durability.SYNC` 模式确保状态同步

### 错误处理

社区问题 #6170 提出了更健壮的节点错误处理需求。建议：

```python
# 使用重试策略包装节点
@retry(stops=stops.after_attempt(3))
async def robust_node(state):
    try:
        return await risky_operation(state)
    except Exception as e:
        # 返回错误状态而非抛出
        return {"error": str(e)}
```

## 配置选项

### 核心配置参数

| 参数 | 类型 | 说明 |
|------|------|------|
| `checkpointer` | Checkpointer | 状态持久化器 |
| `interrupt_before` | list[str] | 中断前执行的节点 |
| `interrupt_after` | list[str] | 中断后执行的节点 |
| `recursion_limit` | int | 最大递归深度（默认 25） |
| `timeout` | float | 单步超时时间 |
| `durability` | Durability | 持久性模式 |

## 总结

Pregel 执行引擎是 LangGraph 的核心，它通过协调任务执行、状态管理和持久化，为构建复杂的 Agent 工作流提供了坚实的基础。理解其执行模型对于有效使用 LangGraph 至关重要。

关键要点：

- **状态驱动**：所有执行围绕状态更新展开
- **检查点持久化**：支持中断和恢复
- **并发执行**：支持多节点并行
- **灵活控制**：通过 `Send` 和 `Command` 实现复杂流程控制

---

<a id='page-channels'></a>

## 通道与状态管理

### 相关页面

相关主题：[图构建与状态定义](#page-graph-construction), [状态管理与内存](#page-state-management)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/langgraph/langgraph/channels/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/channels/base.py)
- [libs/langgraph/langgraph/channels/last_value.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/channels/last_value.py)
- [libs/langgraph/langgraph/channels/any_value.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/channels/any_value.py)
- [libs/langgraph/langgraph/channels/topic.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/channels/topic.py)
- [libs/langgraph/langgraph/channels/binop.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/channels/binop.py)
- [libs/langgraph/langgraph/pregel/main.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/main.py)
- [libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
- [libs/prebuilt/langgraph/prebuilt/tool_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_node.py)
</details>

# 通道与状态管理

LangGraph 的通道（Channels）与状态管理是其核心执行模型——Pregel 算法——的关键组成部分。通道定义了如何在图的节点之间传递和存储数据，而状态管理则负责协调整个图的执行流程、版本控制和持久化。

## 核心概念

### Pregel 计算模型

LangGraph 基于 Google 的 Pregel 模型构建，采用"像图一样思考"（think like a graph）的编程范式。在 Pregel 中：

- 图由节点（Nodes）和边（Edges）组成
- 每个节点维护自己的状态（Value）
- 节点通过通道（Channels）进行通信
- 计算以"超步"（Superstep）为单位进行

```mermaid
graph TD
    A[开始] --> B[节点 A 执行]
    B --> C[通道更新]
    C --> D[节点 B 读取]
    D --> E[节点 B 执行]
    E --> F[通道更新]
    F --> G{还有更多节点?}
    G -->|是| D
    G -->|否| H[结束]
```

### 通道的类型体系

LangGraph 提供了多种通道类型，每种类型定义了不同的值更新语义：

| 通道类型 | 更新语义 | 使用场景 |
|---------|---------|---------|
| `LastValue` | 覆盖为最新值 | 默认状态类型 |
| `AnyValue` | 保留任意非空值 | 只关心最终结果 |
| `Topic` | 按主题累积值 | 消息队列模式 |
| `BinaryOperator` | 二元运算符合并 | 累加、拼接等 |

## 通道类型详解

### BaseChannel 抽象基类

所有通道类型都继承自 `BaseChannel`：

```python
class BaseChannel:
    """通道基类，定义了通道的通用接口"""
    
    @abstractmethod
    def checkpoint(self) -> Any: ...
    
    @abstractmethod
    def update(self, values: Sequence[Any]) -> None: ...
```

关键方法包括：

- `checkpoint()`: 创建通道状态的快照
- `update(values)`: 根据新值更新通道状态
- `resume(values)`: 从检查点恢复通道状态

资料来源：[libs/langgraph/langgraph/channels/base.py:1-50]()

### LastValue 通道

`LastValue` 是 LangGraph 的默认通道类型，它始终保留最近一次更新的值。

```python
class LastValue(TypedDict):
    """LastValue 通道状态"""
    idx: int  # 值的索引/版本号
    value: Any  # 实际值
```

特点：

- 每次 `update` 调用会覆盖之前存储的值
- 保持一个递增的版本索引
- 适用于大多数状态字段

```python
# LastValue 的更新逻辑示例
def update(self, values: Sequence[Any]) -> None:
    if values:
        self.value = values[-1]  # 始终保留最后一个值
        self.idx += 1
```

资料来源：[libs/langgraph/langgraph/channels/last_value.py:1-30]()

### AnyValue 通道

`AnyValue` 通道在接收到第一个非空值后锁定，之后的更新将被忽略：

```python
class AnyValue(BaseChannel):
    """保留第一个非空值，忽略后续更新"""
    
    def __init__(self, typ: type[Any]):
        self.typ = typ
        self.value: Any | None = None
        self.version = 0
```

适用场景：

- 只关心某个操作的第一次成功结果
- 防止重复计算覆盖有效结果
- 实现"一旦设置，永不改变"的语义

资料来源：[libs/langgraph/langgraph/channels/any_value.py:1-25]()

### Topic 通道

`Topic` 通道允许按主题分类累积值：

```python
class Topic(BaseChannel):
    """按主题累积值的通道"""
    
    def update(
        self, 
        values: dict[str, list[Any]] | Sequence[dict[str, Any]]
    ) -> None:
        # 按主题分组累积值
```

特点：

- 支持多主题并行累积
- 每个主题维护独立的值列表
- 适用于消息队列和事件流场景

资料来源：[libs/langgraph/langgraph/channels/topic.py:1-40]()

### BinaryOperator 通道

`BinaryOperator` 通道使用二元运算符合并新值：

```python
class BinaryOperator(BaseChannel):
    """使用二元运算符合并值的通道"""
    
    def __init__(
        self, 
        typ: type[Any], 
        operator: Callable[[Any, Any], Any]
    ):
        self.operator = operator
```

常用操作符示例：

| 操作符 | 函数 | 效果 |
|-------|------|-----|
| 加法 | `operator.add` | 数值累加 |
| 列表拼接 | `lambda a, b: a + b` | 列表合并 |
| 字典更新 | `dict.update` | 字典合并 |

资料来源：[libs/langgraph/langgraph/channels/binop.py:1-35]()

## 状态管理架构

### StateGraph 与状态模式

`StateGraph` 是 LangGraph 中用于定义有状态图的核心类：

```python
class StateGraph(Chain["StateSchema"]):
    """有状态图，支持通道和状态管理"""
    
    def __init__(self, schema: type[dict | TypedDict]):
        self.schema = schema
        self.channels: dict[str, BaseChannel] = {}
```

状态模式通过 TypedDict 定义：

```python
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    remaining_steps: int
    memory: Annotated[dict, operator.or_]
```

注意：某些类型（如 `AgentState`）已从 `langgraph.prebuilt` 迁移到 `langchain.agents`：

> `AgentState has been moved to langchain.agents. Please update your import.`

资料来源：[libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:30-50]()

### Annotated 与通道注解

使用 `Annotated` 类型可以将通道类型绑定到状态字段：

```python
class State(TypedDict):
    # LastValue 通道（默认）
    counter: int
    
    # 自定义通道注解
    messages: Annotated[Sequence[BaseMessage], add_messages]
    
    # BinaryOperator 通道
    memory: Annotated[dict, operator.or_]
```

这允许精细控制每个状态字段的更新语义。

### 状态版本控制

LangGraph 为每个通道维护版本号，用于检测并发修改和确保一致性：

```python
class ChannelVersions(dict[str, int]):
    """通道版本映射"""
    pass
```

版本递增规则：

1. 通道被更新时，版本号 +1
2. 检查点保存时记录当前版本
3. 从检查点恢复时还原版本

## 检查点与持久化

### 检查点机制

检查点（Checkpoint）用于保存和恢复图的执行状态：

```mermaid
graph LR
    A[执行中] -->|中断| B[创建检查点]
    B --> C[保存到存储]
    C --> D[恢复执行]
    D -->|从检查点| E[继续执行]
```

关键方法：

- `get_checkpoint()`: 获取当前检查点
- `put_checkpoint()`: 保存检查点
- `list_checkpoints()`: 列出历史检查点

资料来源：[libs/langgraph/langgraph/pregel/main.py:100-150]()

### 检查点存储后端

LangGraph 支持多种检查点存储后端：

| 后端 | 模块 | 适用场景 |
|-----|------|---------|
| Memory | `langgraph.checkpoint.memory` | 开发/测试 |
| SQLite | `langgraph.checkpoint.sqlite` | 小规模生产 |
| PostgreSQL | `langgraph.checkpoint.postgres` | 大规模生产 |

### PostgreSQL 检查点

生产环境推荐使用 PostgreSQL 后端：

```python
from langgraph.checkpoint.postgres import PostgresSaver

saver = PostgresSaver.from_conn_string(conn_string)
checkpointer = saver
```

PostgreSQL 后端特点：

- 支持多线程并发访问
- 内置迁移管理
- 支持 SSL 连接

资料来源：[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:1-80]()

## 常见问题与社区反馈

### Agent 无限循环问题

社区反馈中常见的问题是 Agent 陷入无限循环直到达到递归限制：

> **#6731**: LangGraph [1.0.6] Bug - Agent infinite looping until recursion limit error is hit

这通常与状态更新逻辑相关，需要检查：

- 状态字段是否正确更新
- 通道的更新语义是否符合预期
- 是否存在状态未正确持久化导致的重试

### 序列化错误

消息序列化是另一个常见问题来源：

> **#4956**: TypeError: Type is not msgpack serializable: AIMessage

解决方案：

1. 确保自定义对象实现了 `__msgpack__` 方法
2. 使用 LangGraph 内置的可序列化消息类型
3. 配置适当的序列化器

### InvalidUpdateError

状态更新时必须至少写入一个字段：

> **#740**: langgraph.errors.InvalidUpdateError: Must write to at least one of ['input', 'plan', 'past_steps', 'response']

这确保图的执行不会静默失败，每个节点都应该产生有意义的输出。

### 检查点配置问题

`langgraph dev` 命令可能忽略检查点配置：

> **#5790**: `langgraph dev` Ignores Checkpointer Configuration, Forcing In-Memory Storage

在生产环境中显式配置检查点以确保状态持久化。

### 长工具调用重复执行

当工具调用耗时较长（>3分钟）时，可能出现重复执行：

> **#7417**: Long tool calls (~180s+) silently re-executed from checkpoint

这与检查点保存时机和并发控制有关，需要在架构设计中考虑超时和幂等性。

## 最佳实践

### 状态设计原则

1. **最小化状态**：只保存必要的数据
2. **选择合适的通道类型**：根据更新语义选择
3. **避免状态膨胀**：定期清理过期数据
4. **确保可序列化**：所有状态字段必须可序列化

### 检查点策略

```python
# 推荐：显式配置检查点
graph = StateGraph(schema)
# ... 添加节点和边
compiled = graph.compile(checkpointer=PostgresSaver.from_conn_string(conn_string))
```

### 错误处理

```python
from langgraph.errors import InvalidUpdateError

try:
    result = await graph.ainvoke(initial_state, config)
except InvalidUpdateError as e:
    # 处理状态更新错误
    logger.error(f"Invalid state update: {e}")
```

## 总结

LangGraph 的通道与状态管理系统提供了：

- **灵活的通道类型**：支持多种值更新语义
- **类型安全的状态**：基于 TypedDict 的强类型定义
- **可靠的持久化**：多后端支持的检查点机制
- **版本控制**：完整的并发控制和冲突检测

正确理解和运用这些机制，是构建可靠 LangGraph 应用的基础。

---

<a id='page-checkpointing'></a>

## 检查点与持久化

### 相关页面

相关主题：[Pregel 执行引擎](#page-pregel-execution), [状态管理与内存](#page-state-management)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/checkpoint/langgraph/checkpoint/base/__init__.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/base/__init__.py)
- [libs/checkpoint/langgraph/checkpoint/memory/__init__.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/memory/__init__.py)
- [libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)
- [libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py)
- [libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)
</details>

# 检查点与持久化

## 概述

检查点（Checkpoint）是 LangGraph 中实现状态持久化和图执行中断恢复的核心机制。通过检查点系统，LangGraph 能够在图执行过程中保存当前状态，支持人机交互中断、错误恢复、以及多线程/多会话状态管理。

LangGraph 的检查点系统基于类似 Google Pregel 和 Apache Beam 的设计理念，提供统一的持久化抽象层，支持多种后端存储实现。资料来源：[README.md](README.md)

## 核心架构

### 检查点系统组件

```mermaid
graph TB
    subgraph "应用层"
        A[StateGraph] --> B[Pregel Executor]
    end
    
    subgraph "检查点抽象层"
        B --> C[BaseCheckpointSaver]
        C --> D[MemorySaver]
        C --> E[SqliteSaver]
        C --> F[PostgresSaver]
    end
    
    subgraph "存储后端"
        D --> G[内存]
        E --> H[SQLite 文件]
        F --> I[PostgreSQL 数据库]
    end
```

### 关键接口

`BaseCheckpointSaver` 是所有检查点实现的基础抽象类，定义了以下核心方法：

| 方法 | 描述 | 返回类型 |
|------|------|----------|
| `aget` | 异步获取指定检查点 | `AsyncIterator[Checkpoint]` |
| `aput` | 异步保存检查点 | `None` |
| `aget_versions` | 获取检查点版本历史 | `AsyncIterator[Checkpoint]` |
| `aget_channel_versions` | 获取通道版本信息 | `ChannelVersions` |

资料来源：[libs/checkpoint/langgraph/checkpoint/base/__init__.py](libs/checkpoint/langgraph/checkpoint/base/__init__.py)

## 检查点数据结构

### Checkpoint 结构

```python
class Checkpoint(TypedDict):
    """检查点数据结构"""
    id: str                          # 检查点唯一标识
    checkpoint_ns: str               # 命名空间
    checkpoint_id: str               # 检查点序列号
    parent_checkpoint_id: str | None # 父检查点ID（用于版本追踪）
    channels: dict[str, Any]         # 通道状态快照
    channel_versions: ChannelVersions # 通道版本映射
    pending_writes: list[PendingWrite] # 待写入操作
```

### ChannelVersions

`ChannelVersions` 用于追踪图中各通道的版本状态，支持增量更新和差异同步：

```python
ChannelVersions = dict[str, int]  # channel_name -> version_number
```

资料来源：[libs/checkpoint/langgraph/checkpoint/base/__init__.py](libs/checkpoint/langgraph/checkpoint/base/__init__.py)

## 存储后端实现

### 内存存储 (MemorySaver)

`MemorySaver` 适用于开发和测试环境，数据存储在内存中，重启后数据丢失。

```python
from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点存储器
memory = MemorySaver()

# 在图中使用
graph = StateGraph(...)
compiled = graph.compile(checkpointer=memory)
```

特点：
- 快速，无需序列化开销
- 适合单元测试和原型开发
- 不支持多进程/多机器共享

资料来源：[libs/checkpoint/langgraph/checkpoint/memory/__init__.py](libs/checkpoint/langgraph/checkpoint/memory/__init__.py)

### SQLite 存储 (SqliteSaver)

`SqliteSaver` 适用于单机部署和轻量级生产环境，使用 SQLite 数据库存储检查点。

```python
from langgraph.checkpoint.sqlite import SqliteSaver

# 创建 SQLite 检查点存储器
sqlite = SqliteSaver.from_conn_string("./checkpoints.db")

# 在图中使用
compiled = graph.compile(checkpointer=sqlite)
```

特点：
- 单文件存储，便于部署
- 支持基本的并发读取
- 适合中小规模应用

资料来源：[libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py](libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py)

### PostgreSQL 存储 (PostgresSaver)

`PostgresSaver` 适用于高并发、多实例部署的生产环境。

```python
from langgraph.checkpoint.postgres import PostgresSaver

# 创建 PostgreSQL 检查点存储器
postgres = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/langgraph")

# 在图中使用
compiled = graph.compile(checkpointer=postgres)
```

数据库 Schema：

```sql
CREATE TABLE IF NOT EXISTS checkpoint_migrations (
    v INTEGER PRIMARY KEY
);

CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    -- ... 其他字段
);
```

特点：
- 支持高并发读写
- 适合分布式部署
- 可通过连接池优化性能

资料来源：[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:1-45](libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)

## 序列化机制

### MsgPack 序列化

LangGraph 使用 MsgPack 作为默认的检查点序列化格式：

```python
from langgraph.checkpoint.serde._msgpack import MsgPackSerializer

serde = MsgPackSerializer()
```

序列化配置通过 `SerdeConfig` 进行管理，支持自定义序列化器：

```python
class SerdeConfig(BaseModel):
    """序列化配置"""
    bigint: Literal["error", "string", "int"] = "error"
    datetime: Literal["error", "string", "iso"] = "iso"
    uuid: Literal["error", "string"] = "error"
```

注意：某些类型（如 `AIMessage`、`Send`）可能不支持 MsgPack 序列化，社区中存在相关问题报告。资料来源：[#4956](https://github.com/langchain-ai/langgraph/issues/4956)、[#5891](https://github.com/langchain-ai/langgraph/issues/5891)

资料来源：[libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py](libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py)

## 在图中使用检查点

### 基本配置

```python
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    count: int

# 创建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

# 编译时配置检查点
checkpointer = MemorySaver()
compiled = graph.compile(checkpointer=checkpointer)
```

### 带配置的图执行

```python
# 创建新线程
config = {"configurable": {"thread_id": "session-123"}}

# 首次执行
result = compiled.invoke({"messages": [HumanMessage(content="Hello")]}, config)

# 恢复执行（从上次中断处继续）
result = compiled.invoke({"messages": [HumanMessage(content="Continue")]}, config)
```

## 人机交互与中断恢复

检查点系统支持人机交互（Human-in-the-Loop）模式：

```mermaid
sequenceDiagram
    participant User as 用户
    participant Graph as LangGraph
    participant Checkpoint as 检查点存储
    participant LLM as LLM/工具

    Graph->>LLM: 执行节点
    LLM-->>Graph: 需要人工确认
    Graph->>Checkpoint: 保存检查点
    Graph-->>User: 返回待确认状态
    User->>Graph: 确认/修改
    Graph->>Checkpoint: 恢复检查点
    Graph->>LLM: 继续执行
```

### 中断与恢复示例

```python
# 使用 interrupt() 中断执行
from langgraph.constants import Interrupt

def should_interrupt(state: AgentState) -> bool:
    if len(state["messages"]) > 5:
        return True
    return False

# 编译图时添加中断条件
graph = StateGraph(AgentState)
graph.add_conditional_edges("agent", should_interrupt, {...})
compiled = graph.compile(checkpointer=MemorySaver())

# 执行 - 会自动保存检查点并中断
result = compiled.invoke(initial_state)

# 用户确认后恢复执行
resume_result = compiled.invoke(None, config)
```

## Delta Channel（增量通道）

Delta Channel 是 LangGraph 用于高效存储增量状态变化的机制：

```python
from langgraph.checkpoint.base import DeltaChannelHistory

# _DELTA_PAGE_SIZE 控制分页扫描大小
_DELTA_PAGE_SIZE = 1024  # 内部常量
```

特点：
- 减少存储空间占用
- 支持快速增量同步
- 适用于长时间运行的任务

资料来源：[libs/checkpoint/langgraph/checkpoint/base/__init__.py](libs/checkpoint/langgraph/checkpoint/base/__init__.py)、[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py](libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)

## 已知问题与限制

### 社区报告的问题

| 问题 | 描述 | 严重程度 |
|------|------|----------|
| [#6731](https://github.com/langchain-ai/langgraph/issues/6731) | Agent 无限循环直到达到递归限制错误 | 高 |
| [#3716](https://github.com/langchain-ai/langgraph/issues/3716) | PostgreSQL 检查点 SSL 连接错误 | 中 |
| [#7417](https://github.com/langchain-ai/langgraph/issues/7417) | 长时间工具调用（>180秒）被静默重新执行 | 高 |
| [#5790](https://github.com/langchain-ai/langgraph/issues/5790) | `langgraph dev` 忽略检查点配置 | 中 |

### 版本兼容性

检查点库与核心 langgraph 库之间存在版本依赖关系：

```python
try:
    major, minor = get_version("langgraph").split(".")[:2]
    if int(major) == 0 and int(minor) < 5:
        warnings.warn(
            "You're using incompatible versions of langgraph and checkpoint-postgres.",
            DeprecationWarning,
            stacklevel=2
        )
except Exception:
    pass
```

资料来源：[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:22-31](libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py)

## 最佳实践

### 生产环境建议

1. **选择合适的存储后端**
   - 开发/测试：使用 `MemorySaver`
   - 小规模部署：使用 `SqliteSaver`
   - 生产环境：使用 `PostgresSaver` 并配置连接池

2. **配置序列化选项**
   ```python
   serde = SerdeConfig(
       bigint="string",
       datetime="iso",
       uuid="string"
   )
   ```

3. **定期清理旧检查点**
   ```python
   # 删除指定线程的旧检查点
   await checkpointer.adelete_thread("thread_id")
   ```

4. **监控长时间运行的任务**
   - 设置合理的 `recursion_limit`
   - 为长时间工具调用配置超时

### 性能优化

```python
# 使用批量操作减少检查点开销
await graph.abatch(inputs, configs)  # 批量执行

# 优化检查点频率
# 在状态更新较少时减少检查点保存频率
```

## 相关资源

- [LangGraph 文档 - 检查点](https://langchain-ai.github.io/langgraph/how-to/persistence/)
- [人机交互示例](https://langchain-ai.github.io/langgraph/how-to/human-in-the-loop/)
- [PostgreSQL 检查点配置指南](https://langchain-ai.github.io/langgraph/how-to/persistence_postgres/)

## 更新日志

| 版本 | 更新内容 |
|------|----------|
| 0.5+ | 添加 Delta Channel 增量通道支持 |
| 0.4+ | 改进 PostgreSQL 检查点性能 |
| 0.3+ | 添加 `get_context_jsonschema` 替代废弃的 `config_schema` |

---

<a id='page-state-management'></a>

## 状态管理与内存

### 相关页面

相关主题：[检查点与持久化](#page-checkpointing)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/checkpoint/langgraph/store/base/__init__.py](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint/langgraph/store/base/__init__.py)
- [libs/prebuilt/langgraph/prebuilt/tool_validator.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_validator.py)
- [libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py)
- [libs/langgraph/langgraph/pregel/main.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/pregel/main.py)
- [libs/langgraph/langgraph/_internal/_pydantic.py](https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/_internal/_pydantic.py)
</details>

# 状态管理与内存

LangGraph 的状态管理与内存系统是构建可靠 AI Agent 的核心基础设施。该系统负责在图执行过程中持久化和恢复状态，支持多轮对话、容错恢复、以及复杂的工作流管理。

## 核心概念

### 状态图（StateGraph）

状态图是 LangGraph 的核心抽象，它定义了一个有向图结构，其中每个节点代表一个处理步骤，每条边代表状态转换。状态图通过 `TypedDict` 定义状态模式，确保类型安全的状态管理。

```mermaid
graph TD
    A[开始] --> B[节点1: 状态更新]
    B --> C[节点2: 工具调用]
    C --> D{条件判断}
    D -->|继续| E[节点3: 处理结果]
    D -->|中断| F[人机交互]
    E --> G[结束]
    F --> E
```

状态管理采用不可变性原则，每次状态更新都会生成新的状态快照，而非原地修改。这种设计确保了状态的完整性和可追溯性。

### Checkpoint 检查点机制

检查点系统允许在任意时刻保存图执行的完整状态，并在需要时恢复。检查点存储包括：

| 组件 | 说明 |
|------|------|
| 状态快照 | 完整的图状态数据 |
| 通道版本 | 每个通道的版本号 |
| 元数据 | 检查点时间戳、配置信息 |
| 任务队列 | 待处理的子任务 |

资料来源：[libs/langgraph/langgraph/pregel/_checkpoint.py:1-50]()

## 状态类定义

### AgentState 结构

`AgentState` 是预构建 Agent 使用的标准状态结构，定义于 `chat_agent_executor.py`：

```python
class AgentState(TypedDict):
    """The state of the agent."""

    messages: Annotated[Sequence[BaseMessage], add_messages]
    remaining_steps: NotRequired[RemainingSteps]
```

| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `messages` | `Annotated[Sequence[BaseMessage], add_messages]` | 是 | 消息历史，自动合并新消息 |
| `remaining_steps` | `NotRequired[int]` | 否 | 剩余执行步数限制 |

资料来源：[libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:25-30]()

### 弃用状态类

LangGraph v1.0 中部分状态类已被弃用：

| 弃用类名 | 替代方案 | 弃用消息 |
|----------|----------|----------|
| `AgentStatePydantic` | `AgentState` (from `langchain.agents`) | Pydantic 状态已迁移 |
| `AgentStateWithStructuredResponse` | `AgentState` | 结构化响应已内置 |

资料来源：[libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py:32-45]()

```python
@deprecated(
    "AgentState has been moved to `langchain.agents`. Please update your import to `from langchain.agents import AgentState`.",
    category=LangGraphDeprecatedSinceV10,
)
```

## Store 存储系统

### 存储操作类型

LangGraph 提供四种核心存储操作类型，统一封装在 `Op` 联合类型中：

```python
Op = GetOp | SearchOp | PutOp | ListNamespacesOp
Result = Item | list[Item] | list[SearchItem] | list[tuple[str, ...]] | None
```

资料来源：[libs/checkpoint/langgraph/store/base/__init__.py:180-181]()

| 操作类型 | 功能 | 返回类型 |
|----------|------|----------|
| `GetOp` | 根据命名空间和键获取单个条目 | `Item` |
| `SearchOp` | 搜索匹配条件的条目 | `list[SearchItem]` |
| `PutOp` | 存储或更新条目 | `None` |
| `ListNamespacesOp` | 列出所有命名空间 | `list[tuple[str, ...]]` |

### 索引配置

存储条目支持灵活的索引配置，用于向量搜索优化：

```python
index: list[str] | None = None
"""
配置需要索引的字段路径。

支持的路径语法：
- 简单字段: "title"
- 嵌套字段: "metadata.title"
- 数组索引: "authors[0].name"
- 数组通配符: "context[*].content"
- 末元素: "array[-1]"
"""
```

资料来源：[libs/checkpoint/langgraph/store/base/__init__.py:140-155]()

### TTL 配置

存储条目支持 TTL（生存时间）机制：

```python
ttl: float | None = None
"""
TTL 配置，单位为分钟。

特性：
- 支持自动过期删除
- 访问操作会刷新 TTL
- 过期后条目会被调度删除
- 默认为 None（永不过期）
"""
```

资料来源：[libs/checkpoint/langgraph/store/base/__init__.py:165-173]()

## 状态序列化与检查点

### msgpack 序列化问题

社区反馈的常见问题之一是序列化错误。当状态中包含不可序列化的对象时，会触发 `TypeError: Type is not msgpack serializable` 错误。

常见原因及解决方案：

| 错误类型 | 原因 | 解决方案 |
|----------|------|----------|
| `AIMessage` | LangChain 消息对象未序列化 | 使用 `ToolMessage` 包装 |
| `Send` 对象 | 与 `BaseStore` 结合使用时的类型 | 参见 Issue #5891 |
| 自定义对象 | 缺少序列化支持 | 实现 `__msgpack__` 方法 |

资料来源：[Issue #4956](https://github.com/langchain-ai/langgraph/issues/4956) | [Issue #5891](https://github.com/langchain-ai/langgraph/issues/5891)

### 检查点配置

图编译时可指定检查点配置：

```python
graph = StateGraph(AgentState)
graph = graph.compile(checkpointer=MemorySaver())
# 或使用 PostgreSQL 检查点
graph = graph.compile(checkpointer=PostgresSaver(conn_string))
```

注意：`langgraph dev` 命令可能忽略检查点配置，使用内存存储。参见 Issue #5790。

资料来源：[Issue #5790](https://github.com/langchain-ai/langgraph/issues/5790)

## ValidationNode 验证节点

`ValidationNode` 是用于验证工具调用参数的内置节点，现已被弃用：

```python
@deprecated(
    "ValidationNode is deprecated. Please use `create_agent` from `langchain.agents` with custom tool error handling.",
    category=LangGraphDeprecatedSinceV10,
)
class ValidationNode(RunnableCallable):
```

### 使用示例

```python
from langgraph.prebuilt import ValidationNode

class SelectNumber(BaseModel):
    number: int = Field(..., ge=0, le=100)

validation_node = ValidationNode(schema=SelectNumber)
```

### 验证流程

```mermaid
graph LR
    A[AI生成工具调用] --> B[ValidationNode接收]
    B --> C{参数验证通过?}
    C -->|是| D[返回ToolMessage]
    C -->|否| E[返回错误消息]
    E --> F[重新生成]
    F --> A
```

ValidationNode 的验证逻辑：
1. 接收 `AIMessage` 中的工具调用
2. 使用 Pydantic 模型验证参数
3. 返回验证后的 `ToolMessage` 或错误消息

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_validator.py:20-65]()

## 状态更新与图执行

### 状态更新规则

状态更新必须遵循以下规则：

1. **至少写入一个字段**：状态更新必须修改至少一个状态字段
2. **类型一致性**：更新值必须与状态定义的类型兼容
3. **不可变性**：节点返回的更新会被合并到当前状态

```python
# 错误示例
def bad_node(state):
    # 抛出 InvalidUpdateError
    return {}  # 必须写入至少一个字段

# 正确示例
def good_node(state):
    return {"messages": [HumanMessage(content="hello")]}
```

错误消息：`langgraph.errors.InvalidUpdateError: Must write to at least one of ['input', 'plan', 'past_steps', 'response']`

资料来源：[Issue #740](https://github.com/langchain-ai/langgraph/issues/740)

### 图执行流程

```mermaid
sequenceDiagram
    participant User as 用户
    participant Graph as StateGraph
    participant Checkpoint as Checkpoint
    participant Node as 节点
    
    User->>Graph: invoke(input, config)
    Graph->>Checkpoint: 获取最新检查点
    Checkpoint-->>Graph: 恢复状态
    loop 执行节点
        Graph->>Node: 调用节点函数
        Node-->>Graph: 返回状态更新
        Graph->>Checkpoint: 保存检查点
    end
    Graph-->>User: 返回最终状态
```

## 配置模式

### context_schema

LangGraph 支持通过 `config_schema` 定义可配置参数：

```python
def get_context_jsonschema(self) -> dict[str, Any] | None:
    """获取上下文配置的 JSON Schema"""
```

配置参数通过 `config.configurable` 传递：

```python
config = {"configurable": {"thread_id": "123"}}
result = graph.invoke(input, config=config)
```

这解决了社区反馈中最常见的困惑来源之一。参见 RFC #5023。

资料来源：[libs/langgraph/langgraph/pregel/main.py:1-50]() | [Issue #5023](https://github.com/langchain-ai/langgraph/issues/5023)

### 弃用的配置 API

| 弃用 API | 替代方案 | 弃用原因 |
|----------|----------|----------|
| `config_schema()` | `get_context_jsonschema()` | 更清晰的命名 |
| `get_config_jsonschema()` | `get_context_jsonschema()` | 统一入口 |

```python
@deprecated(
    "`config_schema` is deprecated. Use `get_context_jsonschema` for the relevant schema instead.",
    category=None,
)
def config_schema(self, *, include: Sequence[str] | None = None) -> type[BaseModel]:
```

资料来源：[libs/langgraph/langgraph/pregel/main.py:200-215]()

## 常见问题与解决方案

### 无限循环问题

Agent 可能陷入无限循环直到达到递归限制。常见原因：

| 原因 | 症状 | 解决方案 |
|------|------|----------|
| 缺少终止条件 | 重复相同动作 | 添加条件边 |
| 状态更新无效 | 状态不变 | 检查更新逻辑 |
| 工具返回错误 | 反复调用同一工具 | 处理错误情况 |

参见 Issue #6731。

资料来源：[Issue #6731](https://github.com/langchain-ai/langgraph/issues/6731)

### 长任务静默重执行

当工具调用耗时超过约 3 分钟（180秒）时，可能出现静默重执行：

| 场景 | 行为 | 影响 |
|------|------|------|
| 原始任务运行中 | 检查点恢复并重新执行 | 重复工作 |
| 并发执行 | 多个实例同时运行 | 3x 成本增加 |

参见 Issue #7417。

资料来源：[Issue #7417](https://github.com/langchain-ai/langgraph/issues/7417)

### Pydantic 模型创建

LangGraph 内部使用 `_pydantic.py` 中的工具函数创建 Pydantic 模型：

```python
def _create_root_model_cached(
    model_name: str,
    type_: Any,
    *,
    module_name: str | None = None,
    default_: object = NO_DEFAULT,
) -> type[BaseModel]:
    """创建根模型的缓存版本"""
```

这些工具支持任意类型作为根类型，用于序列化复杂的状态结构。

资料来源：[libs/langgraph/langgraph/_internal/_pydantic.py:80-100]()

## 最佳实践

### 状态设计原则

1. **最小化状态**：只存储必要的状态字段
2. **类型安全**：使用 `TypedDict` 明确定义状态结构
3. **不可变操作**：返回新值而非修改现有值
4. **合理 TTL**：为临时数据设置过期时间

### 检查点策略

| 场景 | 推荐检查点 |
|------|------------|
| 开发调试 | `MemorySaver` |
| 生产部署 | `PostgresSaver` / `SqliteSaver` |
| 分布式 | `PostgresSaver` |

### 序列化注意事项

确保状态中所有对象都可序列化：

```python
# 正确：将消息内容提取为字符串
{"messages": [{"role": "user", "content": state["query"]}]}

# 避免：直接存储 LangChain 对象
{"messages": [HumanMessage(content="...")]}  # 可能导致序列化错误
```

## 相关资源

- [Issue #4956: msgpack 序列化错误](https://github.com/langchain-ai/langgraph/issues/4956)
- [Issue #740: InvalidUpdateError](https://github.com/langchain-ai/langgraph/issues/740)
- [Issue #5790: 检查点配置被忽略](https://github.com/langchain-ai/langgraph/issues/5790)
- [Issue #6731: 无限循环问题](https://github.com/langchain-ai/langgraph/issues/6731)
- [Issue #7417: 长任务重执行](https://github.com/langchain-ai/langgraph/issues/7417)
- [RFC #5023: context API 反馈](https://github.com/langchain-ai/langgraph/issues/5023)

---

<a id='page-cli-usage'></a>

## CLI 使用指南

### 相关页面

相关主题：[安装与配置](#page-installation), [SDK 与 API 集成](#page-sdk-integration)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/cli/langgraph_cli/cli.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/cli.py)
- [libs/cli/langgraph_cli/deploy.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/deploy.py)
- [libs/cli/langgraph_cli/docker.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/docker.py)
- [libs/cli/langgraph_cli/config.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/config.py)
- [libs/cli/langgraph_cli/schemas.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/langgraph_cli/schemas.py)
</details>

# CLI 使用指南

LangGraph CLI 是 LangGraph 项目的命令行工具，提供本地开发、部署和容器化管理功能。通过该 CLI，开发者可以快速启动本地开发服务器、构建 Docker 镜像、将应用部署到 LangGraph Platform，以及管理图配置和依赖项。

## 功能概述

LangGraph CLI 主要包含以下功能模块：

| 模块 | 功能说明 |
|------|----------|
| `dev` | 启动本地开发服务器，热重载支持 |
| `deploy` | 构建并推送 Docker 镜像到远程仓库 |
| `docker` | 管理本地 Docker 容器生命周期 |
| `config` | 验证和查看 `langgraph.json` 配置文件 |
| `print-schema` | 输出配置 JSON Schema 用于 IDE 自动完成 |

## 核心命令详解

### langgraph dev

`langgraph dev` 命令用于启动本地开发服务器，适合在开发阶段快速迭代图定义。

```bash
langgraph dev [OPTIONS]
```

**主要参数：**

| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `--help` | - | - | 显示帮助信息 |
| `-c, --config` | PATH | `langgraph.json` | 配置文件路径 |
| `-p, --port` | INT | 2024 | 服务器监听端口 |

**已知限制：**

根据社区 Issue [#5790](https://github.com/langchain-ai/langgraph/issues/5790)，当前版本的 `langgraph dev` 会忽略检查点配置，强制使用内存存储，导致状态无法持久化。这是开发者需要特别注意的问题。

另一个相关问题是 Issue [#7688](https://github.com/langchain-ai/langgraph/issues/7688)：由于 TIME-WAIT 条目存在，`langgraph dev` 可能错误地报告 "Port 2024 already in use"。

### langgraph deploy

`langgraph deploy` 命令用于构建 Docker 镜像并推送到远程仓库。

```bash
langgraph deploy [OPTIONS]
```

**主要参数：**

| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `--help` | - | - | 显示帮助信息 |
| `-c, --config` | PATH | `langgraph.json` | 配置文件路径 |
| `-i, --image-name` | TEXT | 必需 | Docker 镜像名称（格式：registry/repo:tag） |

**工作流程：**

1. 读取 `langgraph.json` 配置文件
2. 根据配置选择 Python 或 Node.js 环境
3. 构建 Docker 镜像
4. 推送到指定镜像仓库

### langgraph docker

`langgraph docker` 命令提供本地 Docker 容器管理功能。

```bash
langgraph docker [OPTIONS] COMMAND [ARGS]...
```

**子命令：**

| 命令 | 说明 |
|------|------|
| `up` | 启动 Docker Compose 服务 |
| `down` | 停止 Docker Compose 服务 |
| `logs` | 查看容器日志 |

### langgraph config

`langgraph config` 命令用于验证 `langgraph.json` 配置文件的正确性。

```bash
langgraph config [OPTIONS]
```

**主要参数：**

| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `-c, --config` | PATH | `langgraph.json` | 配置文件路径 |
| `--print-schema` | - | False | 输出 JSON Schema |

### langgraph print-schema

`langgraph print-schema` 输出完整的 JSON Schema，支持 IDE 自动完成和配置验证。

```bash
langgraph print-schema [OPTIONS]
```

该功能基于 `libs/cli/generate_schema.py` 中的 `generate_schema()` 函数实现，使用 `msgspec` 库生成结构化配置 schema。

## 配置文件格式

LangGraph CLI 使用 `langgraph.json` 作为主配置文件。

### 基本配置结构

```json
{
  "$schema": "https://langgra.ph/schema.json",
  "python_version": "3.12",
  "dependencies": [
    "langchain_community",
    "langchain_anthropic",
    "langchain_openai"
  ],
  "graphs": {
    "agent": "./agent.py:graph",
    "storm": "./storm.py:graph"
  },
  "env": "../.env"
}
```

### 配置字段说明

| 字段 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `$schema` | string | 是 | JSON Schema URL，用于 IDE 验证 |
| `python_version` | string | 是* | Python 版本（与 `node_version` 二选一） |
| `node_version` | string | 是* | Node.js 版本（与 `python_version` 二选一） |
| `dependencies` | array | 是 | 外部依赖包列表 |
| `graphs` | object | 是 | 图定义映射（名称 -> 模块路径） |
| `env` | string | 否 | 环境变量文件路径 |

> *注意：根据配置验证逻辑，`python_version` 和 `node_version` 必须指定其中一个。

### 高级配置选项

`libs/cli/langgraph_cli/schemas.py` 定义了完整的配置模型：

| 配置类 | 用途 |
|--------|------|
| `Config` | 主配置类 |
| `GraphDef` | 图定义 |
| `StoreConfig` | 持久化存储配置 |
| `AuthConfig` | 认证配置 |
| `SecurityConfig` | 安全配置 |
| `HttpConfig` | HTTP 服务配置 |
| `CorsConfig` | CORS 跨域配置 |
| `CacheConfig` | 缓存配置 |
| `CheckpointerConfig` | 检查点配置 |
| `SerdeConfig` | 序列化配置 |

## 依赖管理

### Python 项目

```json
{
  "python_version": "3.12",
  "pip_config_file": "requirements.txt"
}
```

### Node.js 项目

```json
{
  "node_version": "18"
}
```

### 依赖图关系

根据 `CLAUDE.md` 和 `AGENTS.md` 中的定义：

```text
checkpoint
├── checkpoint-postgres
├── checkpoint-sqlite
├── prebuilt
└── langgraph

prebuilt
└── langgraph

sdk-py
├── langgraph
└── cli

sdk-js (standalone)
```

## 常见问题与解决方案

### 端口占用问题

**问题：** 启动时报 "Port 2024 already in use"  
**原因：** TIME-WAIT 连接未被完全清理（Issue #7688）  
**解决：** 等待片刻后重试，或使用 `--port` 指定其他端口

### 检查点配置不生效

**问题：** `langgraph dev` 忽略检查点配置  
**原因：** 这是已知的 bug（Issue #5790）  
**解决：** 目前无官方解决方案，可考虑使用完整部署而非开发服务器

### 模块导入错误

**问题：** `ModuleNotFoundError: No module named 'langgraph.graph.graph'`  
**说明：** 这是版本兼容性问题，请确保安装的 langgraph 版本与依赖库版本匹配

## 安装与升级

### 前置要求

- Python 3.12+ 或 Node.js 18+
- Docker 和 Docker Compose
- pip 或 npm 包管理器

### 安装方式

```bash
# 通过 pip 安装
pip install langgraph-cli

# 通过 npm 安装
npm install -g @langchain/langgraph-cli
```

### 验证安装

```bash
langgraph --version
```

## 架构图

### CLI 命令处理流程

```mermaid
graph TD
    A[用户输入 langgraph 命令] --> B[解析命令行参数]
    B --> C{命令类型}
    C -->|dev| D[启动开发服务器]
    C -->|deploy| E[构建 Docker 镜像]
    C -->|docker| F[Docker 容器管理]
    C -->|config| G[配置文件验证]
    C -->|print-schema| H[输出 Schema]
    D --> I[监听文件变化]
    E --> J[推送镜像到仓库]
    F --> K[执行 docker-compose]
    G --> L[msgspec 验证]
    H --> M[输出 JSON Schema]
```

### 配置加载流程

```mermaid
graph LR
    A[langgraph.json] --> B[Config 模型验证]
    B --> C{依赖类型}
    C -->|Python| D[安装 pip 依赖]
    C -->|Node.js| E[安装 npm 依赖]
    D --> F[启动服务]
    E --> F
```

## 相关资源

- 官方文档：[LangGraph Quickstart](https://docs.langchain.com/oss/python/langgraph/quickstart)
- 社区论坛：[LangChain Forum](https://forum.langchain.com)
- 问题反馈：[GitHub Issues](https://github.com/langchain-ai/langgraph/issues)
- 贡献指南：[Contributing Guide](https://docs.langchain.com/oss/python/contributing/overview)

---

<a id='page-sdk-integration'></a>

## SDK 与 API 集成

### 相关页面

相关主题：[CLI 使用指南](#page-cli-usage), [Pregel 执行引擎](#page-pregel-execution)

<details>
<summary>相关源码文件</summary>

以下源码文件用于生成本页说明：

- [libs/sdk-py/langgraph_sdk/auth/__init__.py](https://github.com/langchain-ai/langgraph/blob/main/libs/sdk-py/langgraph_sdk/auth/__init__.py)
- [libs/cli/examples/graphs/langgraph.json](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/examples/graphs/langgraph.json)
- [libs/sdk-py/langgraph_sdk/client.py](https://github.com/langchain-ai/langgraph/blob/main/libs/sdk-py/langgraph_sdk/client.py)
- [libs/prebuilt/langgraph/prebuilt/tool_node.py](https://github.com/langchain-ai/langgraph/blob/main/libs/prebuilt/langgraph/prebuilt/tool_node.py)
- [libs/cli/generate_schema.py](https://github.com/langchain-ai/langgraph/blob/main/libs/cli/generate_schema.py)
</details>

# SDK 与 API 集成

LangGraph SDK 是 LangGraph 项目的核心组件之一，提供了与 LangGraph Server 进行交互的客户端接口、认证授权机制以及远程执行能力。SDK 采用 Python 实现，遵循异步优先的设计理念，同时提供同步和异步两种使用方式。

## SDK 架构概述

LangGraph SDK 主要由以下模块组成：

| 模块路径 | 功能描述 |
|---------|---------|
| `langgraph_sdk.auth` | 认证与授权管理系统 |
| `langgraph_sdk.client` | 同步客户端主入口 |
| `langgraph_sdk._async.client` | 异步客户端实现 |
| `langgraph_sdk._sync.client` | 同步客户端实现 |
| `langgraph_sdk._async.runs` | 异步运行管理 |
| `langgraph_sdk.auth.types` | 认证类型定义 |
| `langgraph_sdk.auth.exceptions` | 认证异常定义 |

依赖关系如下：

```text
sdk-py
├── langgraph
└── cli
```

资料来源：[AGENTS.md]()

## 认证与授权系统

### Auth 类概述

`Auth` 类是 LangGraph SDK 中用于管理认证和授权的核心组件。它提供了统一的方式来处理用户身份验证和细粒度的资源权限控制。

主要功能包括：
- 自定义用户认证协议支持
- 细粒度授权规则定义
- 认证和授权处理器注册

资料来源：[libs/sdk-py/langgraph_sdk/auth/__init__.py]()

### 认证配置

在 `langgraph.json` 配置文件中，通过 `auth` 字段配置认证模块：

```json
{
  "dependencies": ["."],
  "graphs": {
    "agent": "./my_agent/agent.py:graph"
  },
  "env": ".env",
  "auth": {
    "path": "./auth.py:my_auth"
  }
}
```

配置项说明：

| 配置项 | 类型 | 说明 |
|--------|------|------|
| `path` | string | 认证模块路径，格式为 `文件路径:函数名` |
| `dependencies` | array | Python 依赖包列表 |
| `graphs` | object | 定义的图及其源文件路径 |
| `env` | string | 环境变量文件路径 |

资料来源：[libs/sdk-py/langgraph_sdk/auth/__init__.py:1-50]()

### 认证处理器

通过装饰器模式注册认证和授权处理器：

```python
from langgraph_sdk import Auth

my_auth = Auth()

@my_auth.authenticate
async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:
    user = await verify_token(authorization)
    if not user:
        raise Auth.exceptions.HTTPException(
            status_code=401, detail="Unauthorized"
        )
    return user
```

### 认证类型定义

`Auth.types` 模块定义了认证相关的类型：

| 类型名 | 说明 |
|--------|------|
| `MinimalUserDict` | 最小用户信息字典 |
| `Handler` | 认证处理器基类 |
| `Authenticator` | 授权认证器基类 |

资料来源：[libs/sdk-py/langgraph_sdk/auth/__init__.py]()

### 认证异常处理

`Auth.exceptions` 模块提供标准 HTTP 异常类，支持设置状态码和错误详情：

```python
Auth.exceptions.HTTPException(status_code=401, detail="Unauthorized")
```

## 客户端架构

### 客户端层级结构

LangGraph SDK 的客户端采用分层架构设计：

```text
┌─────────────────┐
│  client.py      │  ← 同步主入口 (导出接口)
├─────────────────┤
│ _async/client   │  ← 异步客户端实现
├─────────────────┤
│ _sync/client    │  ← 同步客户端实现
├─────────────────┤
│ _async/runs.py  │  ← 运行管理模块
└─────────────────┘
```

这种设计允许同步和异步实现共享底层逻辑，同时保持各自接口的简洁性。

### 认证模块集成

认证系统与客户端深度集成，通过以下流程处理请求：

```mermaid
sequenceDiagram
    participant Client as 客户端
    participant Auth as Auth 模块
    participant Server as LangGraph Server
    
    Client->>Auth: 发送请求 (携带 Authorization header)
    Auth->>Auth: 验证令牌
    Auth->>Server: 认证通过，转发请求
    Server-->>Client: 返回响应
```

## 工具节点集成

### ToolNode 架构

`ToolNode` 是 LangGraph SDK 提供的预置组件，用于在 LangGraph 工作流中执行工具：

```python
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def my_tool(x: int) -> str:
    return f"Result: {x}"

tool_node = ToolNode([my_tool])
```

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_node.py:1-50]()

### 工具执行设计模式

`ToolNode` 实现以下关键设计模式：

| 模式 | 描述 |
|------|------|
| 并行执行 | 多个工具调用可并行执行以提高效率 |
| 错误处理 | 可定制的错误消息格式 |
| 状态注入 | 支持向工具注入图状态 |
| Store 注入 | 支持向工具注入持久化存储 |

资料来源：[libs/prebuilt/langgraph/prebuilt/tool_node.py]()

### 工具运行时信息

`ToolRuntime` 数据类封装了工具执行所需的运行时信息：

```python
@dataclass
class ToolRuntime:
    state: Any           # 图状态
    context: Any         # 执行上下文
    config: RunnableConfig  # 运行配置
    stream_writer: Callable  # 流写入器
    tool_call_id: str    # 工具调用 ID
    store: BaseStore     # 持久化存储
```

### InjectedState 和 InjectedStore

用于标注需要注入的依赖：

| 类型 | 用途 |
|------|------|
| `InjectedState` | 标注注入图状态 |
| `InjectedStore` | 标注注入持久化存储 |

## CLI 配置与 Schema 生成

### 配置文件结构

LangGraph CLI 使用 `langgraph.json` 作为主配置文件：

```json
{
  "$schema": "https://langgra.ph/schema.json",
  "python_version": "3.12",
  "dependencies": [
    "langchain_community",
    "langchain_anthropic",
    "langchain_openai",
    "wikipedia",
    "scikit-learn",
    "."
  ],
  "graphs": {
    "agent": "./agent.py:graph",
    "storm": "./storm.py:graph"
  },
  "env": "../.env"
}
```

资料来源：[libs/cli/examples/graphs/langgraph.json]()

### Schema 生成机制

`generate_schema.py` 脚本用于从 `Config` 类生成 JSON Schema，以支持 IDE 自动完成和验证：

```python
from langgraph_cli.schemas import (
    Config,
    AuthConfig,
    CacheConfig,
    CheckpointerConfig,
    # ...
)

def add_descriptions_to_schema(schema, cls):
    """Add docstring descriptions to the schema properties."""
    # 从类文档字符串和属性注解中提取描述
```

资料来源：[libs/cli/generate_schema.py]()

### 配置类型

| 配置类型 | 说明 |
|---------|------|
| `Config` | 主配置类 |
| `AuthConfig` | 认证配置 |
| `CacheConfig` | 缓存配置 |
| `CheckpointerConfig` | 检查点配置 |
| `GraphDef` | 图定义 |
| `HttpConfig` | HTTP 配置 |
| `SecurityConfig` | 安全配置 |
| `SerdeConfig` | 序列化配置 |
| `StoreConfig` | 存储配置 |
| `WebhookUrlPolicy` | Webhook URL 策略 |

## 社区问题与已知限制

根据社区反馈，以下问题与 SDK 和 API 集成相关：

### Msgpack 序列化问题

**问题 #4956**: `TypeError: Type is not msgpack serializable: AIMessage`

当使用某些 LangChain 消息类型时，SDK 的检查点机制可能出现序列化错误。SDK 已添加对以下类型的 msgpack 支持：

| 模块 | 类型 |
|------|------|
| `langchain_core.messages` | AIMessage, AIMessageChunk, HumanMessage, etc. |
| `langgraph.types` | Send, Interrupt, Command, StateSnapshot |
| `langgraph.store.base` | Item, GetOp |

资料来源：[libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py]()

### 认证集成建议

社区讨论中提出了对更强大错误处理机制的需求（Issue #6170）。建议在 SDK 层面增加：

- 节点错误处理的钩子机制
- 中间件支持
- 更完善的文档

## 最佳实践

### 1. 使用异步客户端

优先使用异步客户端以获得更好的性能：

```python
from langgraph_sdk import get_client

async with get_client(url="http://localhost:2024") as client:
    # 异步操作
    pass
```

### 2. 正确配置认证

确保认证模块正确导出，并在 `langgraph.json` 中正确引用：

```python
# auth.py
from langgraph_sdk import Auth
my_auth = Auth()
# ... 配置处理器
```

### 3. 依赖版本匹配

SDK 版本与 LangGraph 核心库版本需匹配。SDK 会进行版本检查：

```python
try:
    major, minor = get_version("langgraph").split(".")[:2]
    if int(major) == 0 and int(minor) < 5:
        warnings.warn(
            "版本不兼容警告...",
            DeprecationWarning
        )
except Exception:
    # 跳过版本检查（如从源码运行）
    pass
```

资料来源：[libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py]()

## 相关资源

- [LangGraph SDK 文档](https://docs.langchain.com/oss/python/langgraph)
- [LangChain 论坛](https://forum.langchain.com) - SDK 相关讨论区
- [GitHub Issues](https://github.com/langchain-ai/langgraph/issues) - SDK 问题反馈
- [LangGraph Studio](https://docs.langchain.com/oss/python/langgraph/studio) - 可视化原型设计

---

<!-- evidence_pipeline_checked: true -->
<!-- evidence_injected: true -->

---

## Doramagic 踩坑日志

项目：langchain-ai/langgraph

摘要：发现 17 个潜在踩坑项，其中 11 个为 high/blocking；最高优先级：安装坑 - 来源证据：AsyncSqliteSaver.put() and put_writes() deadlock instead of raising when called synchronously from within the event loop。

## 1. 安装坑 · 来源证据：AsyncSqliteSaver.put() and put_writes() deadlock instead of raising when called synchronously from within the event loop

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：AsyncSqliteSaver.put() and put_writes() deadlock instead of raising when called synchronously from within the event loop
- 对用户的影响：可能阻塞安装或首次运行。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_ec8f888eb036440b88ed7e31328b3046 | https://github.com/langchain-ai/langgraph/issues/7857 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 2. 安装坑 · 来源证据：Suggest: generate random Postgres password instead of hardcoded default in Docker Compose

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Suggest: generate random Postgres password instead of hardcoded default in Docker Compose
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_1a1a157468f4446fbeac473755397fa9 | https://github.com/langchain-ai/langgraph/issues/7276 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 3. 安装坑 · 来源证据：[BUG] Interrupt() in a loop will cause extra resumes

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：[BUG] Interrupt() in a loop will cause extra resumes
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_69df436706ce4f59962e3bd145649e86 | https://github.com/langchain-ai/langgraph/issues/7780 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 4. 安装坑 · 来源证据：fix(langgraph): add missing stacklevel to warnings.warn() calls

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：fix(langgraph): add missing stacklevel to warnings.warn() calls
- 对用户的影响：可能影响升级、迁移或版本选择。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_4c8845f5dfba41d0affb4194e5bc7f14 | https://github.com/langchain-ai/langgraph/issues/7776 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 5. 配置坑 · 来源证据：Add support for Python 3.14

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：Add support for Python 3.14
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_ca917407ea67474d8da666afac039e8c | https://github.com/langchain-ai/langgraph/issues/5253 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 6. 配置坑 · 来源证据：v3 stream.subgraphs doesn't detect sub-agents invoked inside tool functions

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：v3 stream.subgraphs doesn't detect sub-agents invoked inside tool functions
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_56545a2106724f298aded17776940248 | https://github.com/langchain-ai/langgraph/issues/7910 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 7. 运行坑 · 来源证据：checkpoint: add stable tie-breaks for equal timestamp ordering

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：checkpoint: add stable tie-breaks for equal timestamp ordering
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_a7ae8f2ae349477ea5ebe1b1691dcd5a | https://github.com/langchain-ai/langgraph/issues/7179 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 8. 维护坑 · 来源证据：AsyncGraphRunStream lacks interleave for v3 stream projections

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：AsyncGraphRunStream lacks interleave for v3 stream projections
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_a14d67e6025641f2a4a5bb7a36e4fc58 | https://github.com/langchain-ai/langgraph/issues/7793 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 9. 安全/权限坑 · 来源证据：Default value of state variable not working with reducer function

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Default value of state variable not working with reducer function
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_2a10ebd879a14b72ba63bdd62beb6ebe | https://github.com/langchain-ai/langgraph/issues/5225 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 10. 安全/权限坑 · 来源证据：Docs safety guidance: auditable final-state receipts for agent completion claims?

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Docs safety guidance: auditable final-state receipts for agent completion claims?
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_cc5ca96d7d24420ca0bd3de5c32028ce | https://github.com/langchain-ai/langgraph/issues/7844 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 11. 安全/权限坑 · 来源证据：Nested astream_events(v3) inside a tool yields empty messages due to parent callback leaking via contextvar

- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Nested astream_events(v3) inside a tool yields empty messages due to parent callback leaking via contextvar
- 对用户的影响：可能阻塞安装或首次运行。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_ce3a3de1be5f4bb482c552dd578af732 | https://github.com/langchain-ai/langgraph/issues/7948 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

## 12. 能力坑 · 能力判断依赖假设

- 严重度：medium
- 证据强度：source_linked
- 发现：README/documentation is current enough for a first validation pass.
- 对用户的影响：假设不成立时，用户拿不到承诺的能力。
- 建议检查：将假设转成下游验证清单。
- 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
- 证据：capability.assumptions | github_repo:676672661 | https://github.com/langchain-ai/langgraph | README/documentation is current enough for a first validation pass.

## 13. 维护坑 · 维护活跃度未知

- 严重度：medium
- 证据强度：source_linked
- 发现：未记录 last_activity_observed。
- 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
- 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
- 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
- 证据：evidence.maintainer_signals | github_repo:676672661 | https://github.com/langchain-ai/langgraph | last_activity_observed missing

## 14. 安全/权限坑 · 下游验证发现风险项

- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：下游已经要求复核，不能在页面中弱化。
- 建议检查：进入安全/权限治理复核队列。
- 防护动作：下游风险存在时必须保持 review/recommendation 降级。
- 证据：downstream_validation.risk_items | github_repo:676672661 | https://github.com/langchain-ai/langgraph | no_demo; severity=medium

## 15. 安全/权限坑 · 存在评分风险

- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：风险会影响是否适合普通用户安装。
- 建议检查：把风险写入边界卡，并确认是否需要人工复核。
- 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
- 证据：risks.scoring_risks | github_repo:676672661 | https://github.com/langchain-ai/langgraph | no_demo; severity=medium

## 16. 维护坑 · issue/PR 响应质量未知

- 严重度：low
- 证据强度：source_linked
- 发现：issue_or_pr_quality=unknown。
- 对用户的影响：用户无法判断遇到问题后是否有人维护。
- 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
- 防护动作：issue/PR 响应未知时，必须提示维护风险。
- 证据：evidence.maintainer_signals | github_repo:676672661 | https://github.com/langchain-ai/langgraph | issue_or_pr_quality=unknown

## 17. 维护坑 · 发布节奏不明确

- 严重度：low
- 证据强度：source_linked
- 发现：release_recency=unknown。
- 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
- 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
- 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
- 证据：evidence.maintainer_signals | github_repo:676672661 | https://github.com/langchain-ai/langgraph | release_recency=unknown

<!-- canonical_name: langchain-ai/langgraph; human_manual_source: deepwiki_human_wiki -->