第 11 章 PostgresSaver 与生产级持久化

11.1 为什么生产需要持久化 Checkpointer

生产级 Agent 应用通常面临三类 MemorySaver 无法满足的核心需求：

进程重启不丢状态：服务滚动发布重启、容器 OOM 被调度重建、机器故障迁移后，正在执行的长流程（如多级审批链、多轮研究任务、异步数据处理管线）必须从中断点继续执行，而不是要求用户从头开始。对于可能执行数小时甚至数天的工作流（等待人工审批、等待外部 webhook 回调），进程内存无法承担状态保管责任。
多实例/多 worker 共享状态：生产环境普遍水平扩展多副本部署（K8s Deployment 多 Pod、Fargate 多任务、ECS 多实例），请求可能被负载均衡到任意 worker。Checkpointer 必须基于共享存储，任一 worker 都能根据 thread_id 加载最新状态快照，保证 resume 请求可以被任意实例接管。
长流程异步执行：HITL 场景下 interrupt 挂起后执行线程释放，人类可能在数分钟到数天后才 resume。这段时间内进程可能已重启、实例可能已迁移，只有持久化存储能保证状态长期存活。

PostgreSQL 因其成熟稳定、JSONB 原生支持、事务 ACID 保证、连接池生态成熟、运维工具完善，成为 LangGraph 官方推荐的生产 Checkpointer 后端。

11.2 PostgresSaver 安装与初始化

PostgresSaver 位于独立发布的 langgraph-checkpoint-postgres 包，底层基于 psycopg（psycopg3，PostgreSQL 官方推荐的新一代 Python 驱动，性能与异步支持显著优于 psycopg2）。

bash

# 基础安装（同步路径）
pip install langgraph-checkpoint-postgres psycopg psycopg-pool
# 异步路径追加二进制驱动（部分环境需要以避免编译 libpq）
pip install "psycopg[binary]"

初始化有两种方式：通过连接字符串便捷构造（适合开发与简单场景），或传入已配置好的连接池实例（生产推荐，连接池复用、参数可控）。

python

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# 方式一：连接字符串（开发期便捷方式）
# with 语句会自动管理连接池的打开与关闭
DB_URI = "postgresql://langgraph:langgraph@localhost:5432/langgraph?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()           # 幂等建表
    graph = builder.compile(checkpointer=checkpointer)
    # 在 with 块内使用 graph ...

# 方式二：连接池（生产推荐，应用生命周期复用）
pool = ConnectionPool(
    conninfo=DB_URI,
    min_size=5,                     # 最小预热连接数
    max_size=20,                    # 最大连接数
    timeout=5,                      # 从池获取连接的超时（秒）
    kwargs={"autocommit": True},    # 必须 autocommit=True，LangGraph 自行管理事务边界
)
pool.open()                         # 启动时打开连接池
checkpointer = PostgresSaver(pool)
checkpointer.setup()                # 幂等建表（首次启动）
graph = builder.compile(checkpointer=checkpointer)
# 应用关闭时调用 pool.close()

setup() 方法幂等地执行 DDL 创建所需表与索引，应用首次启动时调用一次即可；若表已存在且版本匹配则直接跳过，若 schema 落后会执行迁移脚本升级。

11.3 核心表结构

PostgresSaver 依赖三张核心数据表与一张迁移元数据表：

表名	作用	核心字段
checkpoints	checkpoint 元数据与 JSONB 可检索字段	thread_id(text), checkpoint_id(text, PK), parent_id(text, 自引用 FK), namespace(text), metadata(JSONB), type(text), created_at(timestamptz)
checkpoint_blobs	序列化的 channel 值二进制（按 thread+checkpoint+channel 粒度存储）	thread_id, checkpoint_id(FK→checkpoints), channel(text), type(text), blob(bytea), 主键 (thread_id, checkpoint_id, channel)
checkpoint_writes	节点待提交写入（pending writes），按 task_id+idx 分组，包含 interrupt/resume 事件	thread_id, checkpoint_id(FK), task_id(uuid), idx(int), channel(text), type(text), blob(bytea), task_path(text), 主键 (thread_id, checkpoint_id, task_id, idx)
checkpoint_migrations	schema 迁移版本表，setup() 据此判断是否需要升级	v(int, PK)，执行过的迁移版本号集合

设计要点：checkpoint_blobs 与 checkpoint_writes 将 channel 值和 pending 写入序列化为二进制 blob（由 serde 处理），checkpoints 表只存元数据和 JSONB 可索引字段（如 metadata），实现"热元数据"与"冷 blob"的分离，便于按 metadata 做快速检索（如按 user_id 过滤、按 created_at 批量清理），又避免 JSONB 列存储大数据影响性能。

11.4 JsonPlusSerializer 与自定义 serde

默认序列化器为 JsonPlusSerializer（langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer），在标准 JSON 基础上扩展了对以下 Python/LangChain 类型的序列化支持：datetime、date、time、timedelta、UUID、Enum、set/frozenset、bytes、Decimal、Pydantic BaseModel、LangChain BaseMessage 及其子类（AIMessage/HumanMessage/ToolMessage/SystemMessage）。扩展类型通过类型标记字段（"lc" 为 LangChain 对象、"type" 为扩展类型码）在反序列化时自动还原正确的 Python 类型。

当 State 中含有 JsonPlusSerializer 不支持的类型（如自定义业务类、ORM 实体、numpy 数组、pandas DataFrame）时，会在 put 阶段抛序列化错误。有两种解决路径：

推荐路径：State 中只放可序列化的简单类型（str/int/float/bool/list/dict/None）与 JsonPlusSerializer 原生支持的类型（datetime、UUID、BaseMessage、Pydantic BaseModel），自定义对象在节点内部做转换，State 只存 JSON 友好的 DTO。该方式跨版本兼容性好、schema 演进灵活、PostgresSaver 与其他后端（SqliteSaver/RedisSaver）可无缝切换。
自定义 serde 路径：继承 langgraph.checkpoint.serde.base.BaseSerializer（或 SerializerProtocol），实现 dumps_typed(obj) -> (str, bytes) 与 loads_typed(typ: str, data: bytes) -> Any 方法，构造 PostgresSaver 时通过 serde= 参数传入。自定义 serde 必须保证跨版本反序列化稳定（老数据用新代码能读），是较高级的扩展点。

11.5 连接池配置

连接池是高并发场景稳定性的关键配置项。psycopg-pool 的核心参数与生产推荐值如下（基于 PostgreSQL 默认 max_connections=100）：

参数	推荐值	说明
min_size	5	预热连接数，避免首批请求冷启动延迟
max_size	20	单实例最大连接数；多实例总连接 = 实例数 × max_size，不能超过 DB max_connections（默认 100）
timeout	5–10 秒	从池获取连接超时，超过即抛 PoolTimeout 快速失败，避免无限等待堆积
max_waiting	100	等待队列长度，超过即对新请求抛 PoolTimeout，防止连接耗尽拖垮数据库
kwargs.autocommit	True	必须开启。LangGraph 在 put/put_writes 中自行管理事务边界，autocommit=True 让每条 SQL 独立提交，避免连接跨请求残留事务状态
max_idle	300 秒	空闲连接最大存活时间，到期回收，防止连接被 DB 或中间 LB 强制断开（idle_in_transaction_session_timeout）
max_lifetime	1800 秒	连接最大寿命，定期重连避免僵尸连接、内存泄漏
check	ConnectionPool.check_connection	连接健康检查回调，取连接前检测连接是否存活

多实例部署时务必核算总连接数：若 10 个 K8s Pod 每个 max_size=20，则总连接数为 200，超过 PostgreSQL 默认 max_connections=100，要么调高 max_connections、要么在前端加 PgBouncer 做连接池复用、要么降低每个 Pod 的 max_size。

11.6 异步使用：AsyncConnectionPool + AsyncPostgresSaver

FastAPI、Starlette、Sanic 等异步 Web 框架必须使用异步 Checkpointer。异步路径下若错误地使用同步 PostgresSaver，ainvoke 会通过线程池调用同步数据库操作，在高并发下会阻塞事件循环导致严重性能下降，某些场景还会因 ContextVar 在跨线程传播问题产生死锁。

python

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool

async_pool = AsyncConnectionPool(
    conninfo=DB_URI,
    min_size=5,
    max_size=20,
    timeout=5,
    kwargs={"autocommit": True},
)
await async_pool.open()
checkpointer = AsyncPostgresSaver(async_pool)
await checkpointer.asetup()
graph = builder.compile(checkpointer=checkpointer)
# 关闭时 await async_pool.close()

FastAPI 集成模式：在 lifespan 异步上下文管理器中打开/关闭连接池，将编译好的图存入 app.state.graph，请求处理函数通过 request.app.state.graph 获取并用 ainvoke/astream 调用。

11.7 thread_id 设计实践与过期清理

thread_id 命名实践：使用业务主键组合命名，如 "{tenant_id}:{user_id}:{session_id}" 或工单场景的 "{ticket_id}"。避免使用纯 UUID 作为 thread_id：出现线上问题时，从日志/告警中的 thread_id 无法直接追溯业务对象（哪个用户、哪个工单），排查效率低。可以在 config["metadata"] 中额外存入 user_id、tenant_id、业务类型、来源渠道等字段——metadata 会写入 checkpoints 表的 JSONB 列，可用于后续按维度索引查询、批量清理、数据统计。

过期清理策略：checkpoint_blobs 是 append-only 表，无自动清理机制，随调用持续增长。三种常见策略：

定期 DELETE 任务：通过定时任务（CronJob、Celery Beat、Airflow）按 checkpoints.created_at < NOW() - INTERVAL '30 days' 批量删除，先 DELETE checkpoint_writes、checkpoint_blobs，再 DELETE checkpoints（注意外键顺序）；
PostgreSQL 原生分区：按 created_at 做 RANGE 分区（周/月粒度），过期分区直接 DROP TABLE，效率远高于 DELETE，适合数据量大的场景；
TTL 字段驱动：在 metadata 中写入业务 TTL（如短对话 7 天、工单流程 90 天），清理任务按 metadata.ttl 时间删除不同类别的数据。

状态膨胀问题：messages 字段使用 add_messages reducer 时默认无限追加，长对话会导致 channel_values 体积指数增长，每次 checkpoint 序列化/写入时间增大、数据库 IO 上升、恢复速度变慢。对策：

定期调用 trim_messages（langchain_core.messages.trim_messages）保留最近 N 轮（如最近 20 条消息），旧消息做摘要替换；
引入消息摘要节点：每经过若干轮对话触发一个 summarization 节点，将旧消息压缩为一条 SystemMessage 摘要替换原历史，控制 messages 总长度在 token 上限内；
分层存储：短期上下文保留在 messages（进 Checkpointer），长期事实通过 Store 保存（不进 checkpoint），参考第 12 章。

11.8 其他 Checkpointer 选型对比

后端	包名	适用场景	优点	局限
MemorySaver	langgraph（内置）	开发调试、单元测试、Notebook	零依赖、速度最快	无持久化、不支持多 worker 共享、内存无限增长
SqliteSaver	langgraph-checkpoint-sqlite	嵌入式应用、桌面端、单实例部署、原型	单文件零运维、SQLite 库成熟	单写者限制、不适合高并发多实例、运维工具不如 PG 丰富
PostgresSaver	langgraph-checkpoint-postgres	生产首选（Web 服务、多实例、长流程）	事务 ACID、连接池、JSONB 索引、多实例共享、pgvector 可与 Store 共用	需要运维 PostgreSQL 实例
RedisSaver	社区包（langgraph-checkpoint-redis）	超低延迟短时缓存、消息队列场景	读写延迟极低（亚毫秒）、适合短 TTL 会话	持久化弱于 PG、TTL 需配置、集群模式一致性问题
DynamoDBSaver	社区包（langgraph-checkpoint-dynamodb）	AWS Serverless 生态（Lambda/DynamoDB）	弹性伸缩、按需付费、免运维	一致性延迟、冷启动、成本在高写入时不低、本地调试不便

11.9 生产端到端示例：docker-compose + FastAPI + PostgresSaver

docker-compose.yml 骨架：

yaml

version: "3.9"
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: langgraph
      POSTGRES_PASSWORD: ${DB_PASSWORD:-langgraph}
      POSTGRES_DB: langgraph
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langgraph -d langgraph"]
      interval: 5s
      retries: 10
      start_period: 10s
    restart: unless-stopped

  app:
    build: .
    environment:
      DB_URI: postgresql://langgraph:${DB_PASSWORD:-langgraph}@postgres:5432/langgraph?sslmode=disable
      OPENAI_API_KEY: ${OPENAI_API_KEY}
    ports:
      - "8000:8000"
    depends_on:
      postgres:
        condition: service_healthy
    restart: unless-stopped
    deploy:
      replicas: 2  # 多实例水平扩展，PostgresSaver 天然支持

volumes:
  pgdata:

FastAPI 应用骨架（异步栈）：

python

# app.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

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

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)

def chatbot(state: AppState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def build_graph(checkpointer):
    b = StateGraph(AppState)
    b.add_node("chatbot", chatbot)
    b.add_edge(START, "chatbot")
    b.add_edge("chatbot", END)
    return b.compile(checkpointer=checkpointer)

DB_URI = os.environ["DB_URI"]

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动：打开连接池、初始化 checkpointer、编译图
    pool = AsyncConnectionPool(
        conninfo=DB_URI, min_size=5, max_size=20, timeout=5,
        kwargs={"autocommit": True},
    )
    await pool.open()
    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.asetup()  # 幂等建表
    app.state.graph = build_graph(checkpointer)
    app.state.pool = pool
    yield
    # 关闭：关闭连接池
    await pool.close()

app = FastAPI(lifespan=lifespan, title="LangGraph Chat API")

class ChatRequest(BaseModel):
    session_id: str
    message: str

class ChatResponse(BaseModel):
    reply: str
    session_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    try:
        result = await app.state.graph.ainvoke(
            {"messages": [HumanMessage(content=req.message)]},
            config={
                "configurable": {"thread_id": req.session_id},
                "tags": ["chat-api"],
                "metadata": {"session_id": req.session_id},
                "run_name": "chat_turn",
                "recursion_limit": 15,
            },
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return ChatResponse(
        reply=result["messages"][-1].content,
        session_id=req.session_id,
    )

Dockerfile 骨架：

dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

requirements.txt：

fastapi>=0.115
uvicorn>=0.30
langgraph>=1.2.0
langgraph-checkpoint-postgres>=2.0
langchain-openai>=0.2
psycopg[binary]>=3.2
psycopg-pool>=3.2
python-dotenv>=1.0

本章小结

生产级 Agent 必须使用持久化 Checkpointer 以满足进程重启恢复、多实例共享状态、长流程异步执行三类核心需求，PostgresSaver 是官方推荐的生产后端。安装 langgraph-checkpoint-postgres 与 psycopg/psycopg-pool 后，通过 from_conn_string 便捷构造或传入 ConnectionPool/AsyncConnectionPool 实例（生产推荐），调用 setup() 幂等建表。核心表为 checkpoints（元数据 JSONB）、checkpoint_blobs（通道值二进制 blob）、checkpoint_writes（pending writes），通过外键关联。默认 JsonPlusSerializer 支持 datetime/UUID/Pydantic/BaseMessage 等常见类型，复杂自定义类型需通过 State 层 DTO 转换或自定义 serde。连接池关键参数（min_size/max_size/timeout/max_waiting/autocommit）按实例规模与 DB max_connections 合理配置。异步应用（FastAPI 等）必须使用 AsyncConnectionPool+AsyncPostgresSaver，避免事件循环阻塞。thread_id 使用业务主键命名，checkpoint_blobs 需定期清理避免磁盘膨胀，messages 字段需配合 trim_messages 或摘要节点控制体积。其他可选后端包括 SqliteSaver（单实例嵌入式）、RedisSaver（低延迟短时缓存）、DynamoDBSaver（AWS Serverless）。

常见错误

未调用 setup() 导致表不存在。现象：首次调用 invoke 时抛 psycopg.errors.UndefinedTable: relation "checkpoints" does not exist。原因：创建 PostgresSaver 后忘记调用 setup() 或 asetup()，DDL 没有执行。修复：应用启动 lifespan 中显式调用 await checkpointer.asetup()；该方法幂等，重复调用安全。
同步连接池被异步图调用导致死锁/阻塞。现象：ainvoke 请求延迟极高或挂起不返回，CPU 使用率异常但数据库连接大量等待。原因：异步图调用 aget_tuple/aput 时通过同步 ConnectionPool 执行阻塞数据库 IO，阻塞 asyncio 事件循环。修复：异步图必须使用 AsyncConnectionPool + AsyncPostgresSaver，同步/异步栈严格不混用。
State 中放入不可序列化对象导致 put 失败。现象：执行到超步末 Update 阶段抛 TypeError: Object of type X is not JSON serializable 或自定义 Serde 编码异常。原因：State 字段中放入了文件句柄、数据库连接/游标、线程锁、socket、生成器、自定义类实例（未注册 serde）。修复：State 中只保留 JSON 基础类型、datetime/UUID/Enum/Decimal、Pydantic BaseModel、BaseMessage 等 JsonPlusSerializer 支持的类型；外部资源句柄通过节点内局部变量或 configurable 传递，不进入 State。
连接池大小不足导致高并发超时。现象：流量高峰时大量请求抛 psycopg_pool.PoolTimeout，监控显示数据库连接数已满。原因：max_size 设置过小，或多实例总连接数超过 DB max_connections 导致新连接被拒绝。修复：根据峰值 QPS 与每个请求平均 DB 调用次数估算所需连接数，调整 max_size；多实例场景核算总连接数 = 实例数 × max_size，必要时调高 PostgreSQL max_connections 或前置 PgBouncer 连接池。
未做 checkpoint 定期清理导致磁盘膨胀。现象：运行数周或数月后 PostgreSQL 数据目录持续增长，checkpoint_blobs 表体积达数 GB 甚至数十 GB，查询性能下降、备份耗时变长。原因：checkpointer 是 append-only 写入，checkpoint_blobs 无自动清理机制。修复：上线即部署定期清理任务（按 created_at 或业务 TTL）；数据量极大时使用 PostgreSQL 分区表按时间分区，过期分区 DROP；监控 checkpoint_blobs 总行数与表体积。

生产建议

连接池在应用 lifespan 启动时打开并保持到应用关闭，禁止在请求路径上新建/关闭连接；
默认使用异步栈（AsyncConnectionPool + AsyncPostgresSaver + ainvoke/astream）作为生产部署形态，保持与 FastAPI 异步中间件生态兼容；
PostgreSQL 连接开启 SSL（生产环境 sslmode=require 或 verify-full），密码通过 Secret Manager/K8s Secret 注入，不写入代码或镜像；
监控三类核心指标：checkpoint_blobs 表大小与行数（存储膨胀）、checkpoint_writes 堆积行数（pending writes 异常堆积说明有中断长期未 resume）、连接池 waiting_clients 队列长度（连接池容量是否充足）；
为 messages 字段设计消息裁剪策略：每 N 轮触发 trim_messages 或摘要节点，控制单次 checkpoint 体积在几十 KB 以内（理想状态 <10KB）；
数据库层面开启常规物理备份（pg_basebackup/WAL 归档），对关键业务 thread_id 定期做逻辑备份（pg_dump 按 thread_id 导出）；
新业务上线前用压测工具模拟 1000+ 并发会话持续运行 24 小时，观察 checkpoint_blobs 增长速率、连接池峰值、pg_stat_activity 连接分布，确认配置参数在峰值下稳定。

C 语言

语言点

深度

新标准

面试整理

第 11 章 PostgresSaver 与生产级持久化 ​

11.1 为什么生产需要持久化 Checkpointer ​

11.2 PostgresSaver 安装与初始化 ​

11.3 核心表结构 ​

11.4 JsonPlusSerializer 与自定义 serde ​

11.5 连接池配置 ​

11.6 异步使用：AsyncConnectionPool + AsyncPostgresSaver ​

11.7 thread_id 设计实践与过期清理 ​

11.8 其他 Checkpointer 选型对比 ​

11.9 生产端到端示例：docker-compose + FastAPI + PostgresSaver ​

本章小结 ​

常见错误 ​

生产建议 ​

第 11 章 PostgresSaver 与生产级持久化

11.1 为什么生产需要持久化 Checkpointer

11.2 PostgresSaver 安装与初始化

11.3 核心表结构

11.4 JsonPlusSerializer 与自定义 serde

11.5 连接池配置

11.6 异步使用：AsyncConnectionPool + AsyncPostgresSaver

11.7 thread_id 设计实践与过期清理

11.8 其他 Checkpointer 选型对比

11.9 生产端到端示例：docker-compose + FastAPI + PostgresSaver

本章小结

常见错误

生产建议