生产部署与数据存储
生产环境 vs 开发环境
| 环境 | 命令 | 存储 | 适用场景 |
|---|---|---|---|
| 开发 | langgraph dev | 内存(重启丢失) | 本地调试 |
| 生产 | langgraph build | PostgreSQL/Redis | 生产部署 |
持久化存储配置
安装依赖
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
PostgreSQL 配置
# src/agent/graph.py
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from langgraph.prebuilt import create_react_agent
DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"
with (
PostgresStore.from_conn_string(DB_URI) as store,
PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup() # 初次创建数据库表
checkpointer.setup()
graph = create_react_agent(
llm,
tools=[...],
checkpointer=checkpointer,
store=store
)
Redis 配置
pip install -U langgraph langgraph-checkpoint-redis
from langgraph.checkpoint.redis import RedisSaver
DB_URI = "redis://:6379"
with RedisSaver.from_conn_string(DB_URI) as checkpointer:
graph = create_react_agent(llm, tools=[...], checkpointer=checkpointer)
存储选择:PostgreSQL vs Redis
| 维度 | PostgreSQL | Redis |
|---|---|---|
| 数据持久性 | ✅ ACID,磁盘存储 | ⚠️ 内存为主,AOF/RDB |
| 读取速度 | 稍慢 | ✅ 更快 |
| 写入速度 | 稍慢 | ✅ 更快 |
| 查询能力 | ✅ SQL 复杂查询 | ❌ 仅 KV |
| 数据量 | TB 级 | GB 级 |
| 成本 | 磁盘便宜 | 内存昂贵 |
| 企业接受度 | ✅ 主数据库 | 多用作缓存 |
推荐选择
| 场景 | 推荐存储 |
|---|---|
| 生产环境 | PostgreSQL(首选) |
| 开发/测试 | 内存 |
| 追求极致性能 + 数据量小 | Redis(可选) |
注意:Redis 虽支持 LangGraph 全部功能(checkpointer、store、threads ),但作为主数据库使用时,需接受 AOF 持久化模式。生产环境建议优先选择 PostgreSQL。
数据库表结构
LangGraph 使用的实际表结构如下:
| 表名 | 用途 |
|---|---|
checkpoints | 对话状态快照(检查点) |
checkpoint_blobs | 大型二进制数据(如图片、文件) |
checkpoint_writes | 状态变更记录 |
checkpoint_migrations | checkpointer 迁移记录 |
store | 长期记忆数据(用户自定义数据) |
store_migrations | store 迁移记录 |
表结构差异说明
LangGraph SDK 版本差异:
-
LangGraph SDK < 0.1.x(如 0.0.x):使用
threads和assistants表threads- 会话/线程元数据assistants- 智能体配置messages- 消息历史
-
LangGraph SDK >= 0.1.x:使用新的表结构
checkpoints/checkpoint_blobs/checkpoint_writes- 状态管理store- 长期记忆- 不再使用独立的
threads和assistants表
threads 和 assistants 数据存储位置
如果你使用的是 LangGraph Server(langgraph dev 或 langgraph build):
- 线程/会话数据:存储在
checkpoints表中,通过thread_id(或checkpoint_id)关联 - 智能体配置:通常在代码中定义,或通过
store表存储自定义配置
如果你使用的是 LangGraph Platform(云服务或自托管),可以在管理界面或通过 API 查看和管理 threads/assistants 数据。
存储架构
┌─────────────────────────────────────┐
│ API 请求 │
└─────────────┬── ─────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ LangGraph Server │
│ ┌──────────┐ ┌────────────┐ │
│ │ store │ │ checkpointer│ │
│ │ 长期记忆 │ │ 对话状态 │ │
│ └─────┬────┘ └──────┬─────┘ │
└────────┼─────────────┼─────────────┘
│ │
▼ ▼
┌─────────────────────────────────────┐
│ PostgreSQL / Redis │
│ ┌─────────┐ ┌────────────────┐ │
│ │ store │ │ checkpoints │ │
│ │ │ │ checkpoint_blobs│ │
│ │ │ │ checkpoint_writes│ │
│ └─────────┘ └────────────────┘ │
└─────────────────────────────────────┘
说明:上述架构适用于 LangGraph SDK >= 0.1.x。旧版本(< 0.1.x)的 threads/assistants 数据通过 LangGraph Server API 管理。
常用内置 API
注意:以下 API 仅在 LangGraph Server 或 LangGraph Platform(云服务/自托管)环境下可用。通过 SDK 直接使用 checkpointer 和 store 时,需自行实现状态管理逻辑。
| API 端点 | 功能 |
|---|---|
GET /threads | 列出所有会话 |
POST /threads | 创建新会话 |
GET /threads/{thread_id} | 获取会话详情 |
GET /threads/{thread_id}/history | 获取会话历史 |
POST /threads/{thread_id}/history/clear | 清除会话历史 |
GET /assistants | 列出所有智能体 |
Python SDK 示例
此 SDK 用于连接 LangGraph Server,非直接连接 PostgreSQL。
from langgraph_sdk import get_client
client = get_client(url="http://localhost:2024")
# 列出所有线程
threads = client.threads.list()
# 创建新线程
new_thread = client.threads.create(metadata={"user_id": "123"})
# 获取历史
history = client.threads.history.get("thread_id")
多 Agent API 配置
在 langgraph.json 中配置多个智能体:
{
"dependencies": ["."],
"graphs": {
"agent": "./src/agent/graph.py:graph",
"weather": "./src/weather/graph.py:graph",
"search": "./src/search/graph.py:graph"
}
}
构建生产镜像
# 构建 Docker 镜像
langgraph build -o my-agent.tar
# 推送到容器注册表
langgraph build --push-to-registry gcr.io/my-project/langgraph-agent