项目概述

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

核心设计理念

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

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

资料来源：README.md:1

定位与生态

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

资料来源：AGENTS.md:1

核心概念

图（Graph）结构

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

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

状态管理

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

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 函数实现状态的增量更新：

# 消息合并：左边的消息列表与右边的消息列表连接
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

典型用法

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 类封装了工具执行所需的全部上下文信息：

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

检查点机制（Checkpointing）

核心功能

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

支持的类型

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

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

PostgreSQL 检查点后端

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

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 的图执行遵循以下流程：

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 类型用于从节点返回时指定下一步路由：

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

这使得节点可以：

• 指定下一个要执行的节点
• 向状态中添加数据
• 触发中断等待人工输入

高级特性

子图（Subgraphs）

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

 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)

条件分支

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

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 配置文件部署：

{
  "$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 集成

常见问题与社区热点

用户反馈的典型问题

v1 路线图讨论

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

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

资料来源：#4973: LangGraph v1 roadmap

学习资源

总结

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

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

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

概述

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

核心依赖架构

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

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 存在兼容性问题。

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

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 安装

#### 安装核心库

pip install langgraph

#### 安装 CLI 工具

pip install langgraph-cli

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

# SQLite 后端
pip install langgraph-checkpoint-sqlite

# PostgreSQL 后端
pip install langgraph-checkpoint-postgres

#### 安装预构建组件

pip install langgraph-prebuilt

从源码安装

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

# 克隆仓库
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，主要包含以下配置类：

基础配置示例

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

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

配置约束

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

# 来自 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 作为检查点存储需要额外配置：

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 列表中：

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 适用于开发环境和轻量级部署：

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` 字段注入环境变量：

{ "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 或更高版本：

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 自动补全和配置验证：

生成 schema（通过 generate_schema.py 脚本）

python libs/cli/generate_schema.py

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

### 运行时配置检查

在启动服务前验证配置：

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) - 社区支持

概述

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

LangGraph 的核心设计目标：

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

核心组件

Pregel 内核

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

核心类型系统

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）：

状态图（StateGraph）

图结构定义

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

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 定义，支持多种数据类型：

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 处暂停等待人工审批

检查点数据结构

# 来自 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 层级

已知问题

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

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

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

消息序列化（Msgpack）

序列化类型白名单

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

# 来自 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）

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 结构

# 来自 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

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

# 来自 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 自动处理多种输出格式：

CLI 工具

schema.json 生成

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

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

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

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 核心实现
• Pregel 执行循环
• 算法与指令处理
• Checkpoint 序列化
• PostgreSQL Checkpoint 实现
• 工具节点实现

概述

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

图构建系统允许开发者：

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

LangGraph 的图构建采用声明式 API，参考了 Google 的 Pregel 和 Apache Beam 的设计理念，同时借鉴了 NetworkX 的公共接口风格。资料来源：README.md

核心概念

状态模式（State Schema）

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

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，它将新消息追加到现有消息列表中：

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

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

内置的更新策略包括：

资料来源：libs/langgraph/langgraph/graph/message.py

StateGraph API

基本构建流程

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

StateGraph 的基本使用流程：

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 接口的对象
• 子图：嵌套的已编译图

# 同步节点函数
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

边定义

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

#### 普通边

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

#### 条件边

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

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

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

资料来源：libs/langgraph/langgraph/graph/_branch.py

特殊节点标识

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

• START：图的入口节点
• END：图的终止节点

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

资料来源：libs/langgraph/langgraph/graph/state.py

高级状态模式

Pydantic 模型状态

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

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 模型的创建：

@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

嵌套状态与子图

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

# 创建子图
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 详解

from langgraph.graph.message import add_messages, add_messages

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

add_messages 的合并逻辑：

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

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

配置与上下文

Configurable 字段

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

class MyState(TypedDict):
    config_param: str

使用运行时配置：

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

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

资料来源：社区 Issue #5023

状态验证

#### ValidationNode（已弃用）

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

from langgraph.prebuilt import ValidationNode

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

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

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

#### 自定义状态验证

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

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

图编译与执行

compile() 方法

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

from langgraph.graph import StateGraph, END, START

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

编译参数：

资料来源：libs/langgraph/langgraph/pregel/main.py

运行时状态访问

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

# 获取 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

带剩余步数限制的循环

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

最佳实践

1. 状态设计原则

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

2. 节点设计原则

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

3. 边设计原则

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

4. Checkpoint 配置

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

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

源码结构

核心文件

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 工作流变得简单且可靠，同时保持了足够的灵活性来应对各种生产场景需求。

概述

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

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

核心类型定义

状态与类型变量

StateT = TypeVar("StateT")
OutputT = TypeVar("OutputT")

持久性模式

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

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

资料来源：libs/langgraph/langgraph/types.py:38-46

流输出类型

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

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

执行流程

任务编排

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 模块实现，支持灵活的错误处理：

# 重试装饰器配置
@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

任务发送与命令控制

Send 类型

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

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

使用示例：

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

Command 类型

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

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

检查点与状态持久化

检查点机制

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

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 定义了允许序列化的类型：

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。确保返回的数据类型在允许列表中。

并发执行模型

多节点并行

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

# 伪代码示例
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 访问持久化存储

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 提出了更健壮的节点错误处理需求。建议：

# 使用重试策略包装节点
@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)}

配置选项

核心配置参数

总结

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

关键要点：

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

核心概念

Pregel 计算模型

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

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

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

通道的类型体系

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

通道类型详解

BaseChannel 抽象基类

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

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 的默认通道类型，它始终保留最近一次更新的值。

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

特点：

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

# 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 通道在接收到第一个非空值后锁定，之后的更新将被忽略：

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 通道允许按主题分类累积值：

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 通道使用二元运算符合并新值：

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

常用操作符示例：

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

状态管理架构

StateGraph 与状态模式

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

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

状态模式通过 TypedDict 定义：

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 类型可以将通道类型绑定到状态字段：

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

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

状态版本控制

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

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

版本递增规则：

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

检查点与持久化

检查点机制

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

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 支持多种检查点存储后端：

PostgreSQL 检查点

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

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. 确保可序列化：所有状态字段必须可序列化

检查点策略

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

错误处理

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 应用的基础。

概述

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

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

核心架构

检查点系统组件

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 是所有检查点实现的基础抽象类，定义了以下核心方法：

资料来源：libs/checkpoint/langgraph/checkpoint/base/__init__.py

检查点数据结构

Checkpoint 结构

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 用于追踪图中各通道的版本状态，支持增量更新和差异同步：

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

资料来源：libs/checkpoint/langgraph/checkpoint/base/__init__.py

存储后端实现

内存存储 (MemorySaver)

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

from langgraph.checkpoint.memory import MemorySaver

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

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

特点：

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

资料来源：libs/checkpoint/langgraph/checkpoint/memory/__init__.py

SQLite 存储 (SqliteSaver)

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

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

PostgreSQL 存储 (PostgresSaver)

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

from langgraph.checkpoint.postgres import PostgresSaver

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

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

数据库 Schema：

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

序列化机制

MsgPack 序列化

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

from langgraph.checkpoint.serde._msgpack import MsgPackSerializer

serde = MsgPackSerializer()

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

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

注意：某些类型（如 AIMessage、Send）可能不支持 MsgPack 序列化，社区中存在相关问题报告。资料来源：#4956、#5891

资料来源：libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py

在图中使用检查点

基本配置

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)

带配置的图执行

# 创建新线程
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）模式：

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: 继续执行

中断与恢复示例

# 使用 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 用于高效存储增量状态变化的机制：

from langgraph.checkpoint.base import DeltaChannelHistory

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

特点：

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

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

已知问题与限制

社区报告的问题

版本兼容性

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

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

最佳实践

生产环境建议

1. 选择合适的存储后端

• 开发/测试：使用 MemorySaver
• 小规模部署：使用 SqliteSaver
• 生产环境：使用 PostgresSaver 并配置连接池

``python serde = SerdeConfig( bigint="string", datetime="iso", uuid="string" ) ``

1. 配置序列化选项

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

1. 定期清理旧检查点

1. 监控长时间运行的任务

• 设置合理的 recursion_limit
• 为长时间工具调用配置超时

性能优化

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

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

相关资源

• LangGraph 文档 - 检查点
• 人机交互示例
• PostgreSQL 检查点配置指南

更新日志

核心概念

状态图（StateGraph）

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

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：

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

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

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

弃用状态类

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

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

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

Store 存储系统

存储操作类型

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

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

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

索引配置

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

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（生存时间）机制：

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

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

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

状态序列化与检查点

msgpack 序列化问题

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

常见原因及解决方案：

资料来源：Issue #4956 | Issue #5891

检查点配置

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

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

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

资料来源：Issue #5790

ValidationNode 验证节点

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

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

使用示例

from langgraph.prebuilt import ValidationNode

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

validation_node = ValidationNode(schema=SelectNumber)

验证流程

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. 不可变性：节点返回的更新会被合并到当前状态

# 错误示例
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

图执行流程

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 定义可配置参数：

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

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

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

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

资料来源：libs/langgraph/langgraph/pregel/main.py:1-50 | Issue #5023

弃用的配置 API

@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

长任务静默重执行

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

参见 Issue #7417。

资料来源：Issue #7417

Pydantic 模型创建

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

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：为临时数据设置过期时间

检查点策略

序列化注意事项

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

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

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

相关资源

• Issue #4956: msgpack 序列化错误
• Issue #740: InvalidUpdateError
• Issue #5790: 检查点配置被忽略
• Issue #6731: 无限循环问题
• Issue #7417: 长任务重执行
• RFC #5023: context API 反馈

功能概述

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

核心命令详解

langgraph dev

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

langgraph dev [OPTIONS]

主要参数：

已知限制：

根据社区 Issue #5790，当前版本的 langgraph dev 会忽略检查点配置，强制使用内存存储，导致状态无法持久化。这是开发者需要特别注意的问题。

另一个相关问题是 Issue #7688：由于 TIME-WAIT 条目存在，langgraph dev 可能错误地报告 "Port 2024 already in use"。

langgraph deploy

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

langgraph deploy [OPTIONS]

主要参数：

工作流程：

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

langgraph docker

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

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

子命令：

langgraph config

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

langgraph config [OPTIONS]

主要参数：

langgraph print-schema

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

langgraph print-schema [OPTIONS]

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

配置文件格式

LangGraph CLI 使用 langgraph.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"
}

配置字段说明

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

高级配置选项

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

依赖管理

Python 项目

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

Node.js 项目

{
  "node_version": "18"
}

依赖图关系

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

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 包管理器

安装方式

# 通过 pip 安装
pip install langgraph-cli

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

验证安装

langgraph --version

架构图

CLI 命令处理流程

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]

配置加载流程

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

相关资源

• 官方文档：LangGraph Quickstart
• 社区论坛：LangChain Forum
• 问题反馈：GitHub Issues
• 贡献指南：Contributing Guide

SDK 架构概述

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

依赖关系如下：

sdk-py
├── langgraph
└── cli

资料来源：AGENTS.md

认证与授权系统

Auth 类概述

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

主要功能包括：

• 自定义用户认证协议支持
• 细粒度授权规则定义
• 认证和授权处理器注册

资料来源：libs/sdk-py/langgraph_sdk/auth/__init__.py

认证配置

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

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

配置项说明：

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

认证处理器

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

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 模块定义了认证相关的类型：

资料来源：libs/sdk-py/langgraph_sdk/auth/__init__.py

认证异常处理

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

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

客户端架构

客户端层级结构

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

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

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

认证模块集成

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

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 工作流中执行工具：

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 实现以下关键设计模式：

资料来源：libs/prebuilt/langgraph/prebuilt/tool_node.py

工具运行时信息

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

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

InjectedState 和 InjectedStore

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

CLI 配置与 Schema 生成

配置文件结构

LangGraph CLI 使用 langgraph.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 自动完成和验证：

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

配置类型

社区问题与已知限制

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

Msgpack 序列化问题

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

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

资料来源：libs/checkpoint/langgraph/checkpoint/serde/_msgpack.py

认证集成建议

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

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

最佳实践

1. 使用异步客户端

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

from langgraph_sdk import get_client

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

2. 正确配置认证

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

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

3. 依赖版本匹配

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

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 文档
• LangChain 论坛 - SDK 相关讨论区
• GitHub Issues - SDK 问题反馈
• LangGraph Studio - 可视化原型设计

Pitfall Log / 踩坑日志

项目：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