VuePress 站点

🎯 课程主题

使用 LangGraph 的 AsyncPostgresSaver / PostgresSaver 将 Agent 的短期记忆（state）持久化到 PostgreSQL 数据库，实现跨进程、跨会话的状态保持。

📝 核心知识点

从 InMemorySaver 到 PostgresSaver：从内存存储升级到外部数据库持久化
Checkpointer（持久化器）：LangGraph 中负责保存和加载 Agent 状态的组件
from_connection_string：通过连接字符串构建 Checkpointer 的工厂方法
.setup() 方法：初始化数据库，自动创建必要的表结构（仅在表不存在时创建）
thread_id：通过 configurable 中的 thread_id 标识不同会话，同一 thread_id 可恢复历史状态
PostgreSQL 存储结构：Database → Schema (public) → Tables，核心表为 checkpoints

🏗️ 架构与工作流

┌────────────────────────────────────────────────┐
│               Python / PyCharm                  │
│                                                 │
│  DB_URL = "postgresql://user:pwd@host:5432/db" │
│       │                                         │
│       ▼                                         │
│  checkpointer = AsyncPostgresSaver              │
│                 .from_connection_string(DB_URL)  │
│       │                                         │
│       ▼                                         │
│  checkpointer.setup()  ← 初始化表结构            │
│       │                                         │
│       ▼                                         │
│  agent = create_react_agent(                    │
│      model, tools,                              │
│      checkpointer=checkpointer                  │
│  )                                              │
│       │                                         │
│       ▼                                         │
│  agent.invoke(                                  │
│      {"messages": [...]},                       │
│      config={"configurable": {"thread_id": "1"}}│
│  )                                              │
└────────────────────────────────────────────────┘
       │
       ▼ 读取/写入
┌────────────────────────────────────────────────┐
│           PostgreSQL 远程服务器                  │
│  Database: langgraph_db                        │
│    └─ Schema: public                            │
│         ├─ checkpoints  (主表，存快照)           │
│         ├─ checkpoint_writes                    │
│         ├─ checkpoint_blobs                     │
│         └─ checkpoint_migrations                │
└────────────────────────────────────────────────┘

💻 代码实战

完整示例：PostgreSQL 持久化存储

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage

# 1. 数据库连接字符串
DB_URL = "postgresql://langchain:password@<服务器公网IP>:5432/langgraph_db"

# 2. 构建 Checkpointer
checkpointer = PostgresSaver.from_connection_string(DB_URL)

# 3. 初始化（首次自动建表，后续跳过）
checkpointer.setup()

# 4. 创建 Agent（传入 checkpointer）
agent = create_react_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer
)

# 5. 配置 thread_id 用于会话隔离
config = {"configurable": {"thread_id": "1"}}

# 6. 第一次对话
response1 = agent.invoke(
    {"messages": [HumanMessage(content="你好，我是康师傅")]},
    config=config
)

# 打印回复
for msg in response1["messages"]:
    msg.pretty_print()

# 7. 第二次对话 — 同一 thread_id，可记忆上下文
response2 = agent.invoke(
    {"messages": [HumanMessage(content="你知道我是谁吗？")]},
    config=config
)

for msg in response2["messages"]:
    msg.pretty_print()
# 输出会显示 AI 知道用户是"康师傅"

查看持久化数据（在 PostgreSQL 中）

-- 查看所有数据库
\l

-- 进入 langgraph_db
\c langgraph_db

-- 查看所有 Schema
\dn

-- 查看当前 Schema
SELECT current_schema();

-- 查看所有表（共 4 张）
\dt

-- 核心表：checkpoints
-- 存放每个 thread 在每个时刻的 Checkpointer 快照

⚠️ 常见问题与避坑指南

连接失败：先在 XShell 中用 psql 命令行测试数据库连接，确认公网 IP 和端口可用
首次运行必须 .setup()：否则数据库中没有对应的表，后续操作会失败
.setup() 幂等性：只在表不存在时创建，已有表则跳过，可放心多次调用
thread_id 不能混用：不同会话使用不同 thread_id，同一 thread_id 可累积历史消息
核心代码就两行：from_connection_string + setup()，其余与 InMemorySaver 用法完全一致
表的数量：共创建 4 张表，checkpoints 为主表，存放每个 thread 的状态快照

💡 个人总结与延伸

Checkpointer 是 LangGraph 短期记忆的核心抽象，InMemorySaver 与 PostgresSaver 实现同一接口，切换成本极低
checkpoints 表存储的不仅是 messages，还包含 Agent 的完整 state（如工具调用中间状态、structured response 等）
生产环境推荐使用 PostgreSQL 持久化方案，支持多实例共享、故障恢复
连接字符串支持环境变量注入，避免硬编码敏感信息