pgsql实现持久化

发表于 更新于 分类于 阅读次数：

前言

常使用pgsql与redis配合langgraph的checkpoint与store进行持久化储存，完成长期记忆与短期记忆的实现

pgsql实现持久化

什么是pgsql

PostgreSQL（常简称pgsql或Postgres）是一个功能强大的开源对象-关系型数据库管理系统（ORDBMS），以其稳定性、扩展性和符合SQL标准著称。

docker拉取

docker-compose

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
version: '3.8'

services:
  postgres:
    image: postgres:15        # 指定具体版本
    container_name: postgres_db
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: postgres
      TZ: Asia/Shanghai       # 设置时区
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    restart: unless-stopped
    healthcheck:             # 健康检查
      test: ["CMD", "pg_isready", "-U", "nange"]
      interval: 10s
      timeout: 5s
      retries: 5
    command: ["postgres", "-c", "max_connections=200"]  # 自定义配置

volumes:
  pgdata:

Docker Compose 是一个 定义和运行多容器 Docker 应用 的 声明式工具。 它通过一个 YAML 文件（通常叫 docker-compose.yml） 描述整个应用的服务、网络、存储等配置，然后用一条命令即可 启动/停止/管理 所有容器，无需手动逐个 docker run。

1
2
#拉取并运行
docker-compose up -d

在pgsqsl安装依赖

因为LangGraph的PostgresStore需要使用到pgvector，因此需要在容器中按照如下步骤进行操作，直接使用Docker Desktop软件中进行操作

  • 为什么需要 pgvector？
  • PostgresStore 支持将 向量嵌入（embedding） 存储在 PostgreSQL 中，并基于它们进行 语义搜索。
  • 该功能依赖 pgvector 扩展提供的 vector 类型和索引机制（如 HNSW）。
1
2
3
安装依赖           
apt update    #刷新本地软件包索引           
apt install -y git build-essential postgresql-server-dev-15

apt install -y git build-essential postgresql-server-dev-15一次性安装 3 类依赖。

  • git —— 用来克隆 pgvector 源码。
  • build-essential —— Debian/Ubuntu 的“编译工具链”元包，包含 gcc、make 等。
  • postgresql-server-dev-15 —— 与当前 Postgres 主版本一致 的开发头文件和 pg_config。
1
2
3
4
5
6
7
8
9
10
编译并安装 pgvector      
#把 pgvector 的 v0.7.0 稳定版 源码克隆到本地目录 ./pgvector。
git clone --branch v0.7.0 https://github.com/pgvector/pgvector.git                
cd pgvector  
#调用 Makefile 根据当前操作系统 + PostgreSQL 版本编译出二进制文件（.so 共享库）。
make          
#把刚刚编好的 .so 文件和 .sql/.control 文件复制到 PostgreSQL 的扩展目录
make install                
验证安装，检查扩展文件是否安装成功                    
ls -l /usr/share/postgresql/15/extension/vector*

pgvector/pgvector: Open-source vector similarity search for Postgres

接下来，若要在脚本中进行使用，首先在系统环境中需要安装PostgreSQL 的开发库（libpq），因为 psycopg 需要它来编译或运行

1
2
sudo apt update
sudo apt install libpq-dev postgresql-server-dev-all

psycopg（Python 操作 PostgreSQL 的库） 在 Linux/macOS 上运行时，底层依赖于 PostgreSQL 的 C 语言开发库 libpq。 如果系统里 没有 libpq，psycopg 会出现以下两种问题：

  1. 编译安装失败（源码/旧版本） 当 pip install psycopg2 需要现场编译时，会找不到头文件 libpq-fe.h 或动态库 libpq.so，导致报错：

    1
    error: libpq-fe.h: No such file or directory

  2. 运行时崩溃 即使通过预编译的 wheel 包安装成功，运行时也可能提示：

    1
    ImportError: libpq.so.5: cannot open shared object file

最后，再安装相关依赖包

1
2
pip install langgraph-checkpoint-postgres                    
pip install psycopg psycopg-pool

psycopg官方 PostgreSQL 驱动，在 psycopg 之上再包一层“连接池”，让并发访问更快、更稳定。

连接pgsql

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
# 1) 连接字符串（URI 语法）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
#   协议://      用户   : 密码   @ 主机:端口 / 数据库名  ? 额外参数

# 2) 连接级参数
connection_kwargs = {
    "autocommit": True,     # 每条 SQL 执行完立即提交，无需手动 commit
    "prepare_threshold": 0, # 禁用服务器端 prepared statement，可减少一次往返
}

# 3) 创建池
connection_pool = ConnectionPool(
    conninfo=DB_URI,
    max_size=20,            # 最多 20 条物理连接
    kwargs=connection_kwargs,
)

# 4) 显式打开池（psycopg 3 的 ConnectionPool 默认懒启动，调 open() 会立即建 min_size 条连接）
connection_pool.open()
print("数据库连接池初始化成功")

正常情况每次 SQL 都新建一条 TCP 连接、做 SSL 握手、验证密码、分配内存，如果不复用连接，这些动作就要 每次都重新来一遍，成本非常高。

连接池（Connection Pool）是一种 数据库访问层资源管理组件，其核心目标是在 高并发、短事务 场景下，通过 复用已建立的数据库物理连接 来降低系统整体延迟、减少资源消耗，并防止数据库因瞬时连接风暴而崩溃。

初始化pgsql

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 初始化PostgresStore
in_postgres_store = PostgresStore(
    pool,
    index={
        "dims": 1536,
        "embed": embedding
    }
)
in_postgres_store.setup()

#初始化checkpoint
# 使用传入的连接池创建 PostgresSaver
checkpointer = PostgresSaver(pool)
checkpointer.setup()

#最后编译时添加
graph_builder.compile(checkpointer=checkpointer, store=in_postgres_store)

in_postgres_store.setup() 的角色一句话就能说清：

把数据库里所有为了让向量存储正常工作的“一次性基建”全部建好——只建一次，后面再跑就不会重复执行。

具体而言，它通常干下面三件事：1.建表 / 建扩展；2.建向量索引；3.元数据初始化

fastapi实现生命周期的管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# 定义了一个异步函数lifespan，它接收一个FastAPI应用实例app作为参数。这个函数将管理应用的生命周期，包括启动和关闭时的操作
# 函数在应用启动时执行一些初始化操作，如加载上下文数据、以及初始化问题生成器
# 函数在应用关闭时执行一些清理操作
# @asynccontextmanager 装饰器用于创建一个异步上下文管理器，它允许你在 yield 之前和之后执行特定的代码块，分别表示启动和关闭时的操作
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    # 申明引用全局变量，在函数中被初始化，并在整个应用中使用
    global graph, connection_pool
    # 启动时执行
    try:
        logger.info("正在初始化模型、定义 Graph...")
        # 初始化 LLM
        llm, embedding = get_llm(llm_type)
        # 创建数据库连接池
        DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
        connection_kwargs = {
            "autocommit": True,
            "prepare_threshold": 0,
        }
        connection_pool = ConnectionPool(
            conninfo=DB_URI,
            max_size=20,
            kwargs=connection_kwargs,
        )
        connection_pool.open()  # 显式打开连接池
        logger.info("数据库连接池初始化成功")
        # 短期记忆 初始化checkpointer
        checkpointer = PostgresSaver(connection_pool)
        checkpointer.setup()
        # 长期记忆 初始化PostgresStore
        in_postgres_store = PostgresStore(
            connection_pool,
            index={
                "dims": 1536,
                "embed": embedding
            }
        )
        in_postgres_store.setup()
        # 定义 Graph
        graph = create_graph(llm, checkpointer, in_postgres_store )
        # 保存 Graph 可视化图
        save_graph_visualization(graph)
        logger.info("初始化完成")
    except Exception as e:
        logger.error(f"初始化过程中出错: {str(e)}")
        raise

    yield  # 应用运行期间

    # 关闭时执行
    logger.info("正在关闭...")
    if connection_pool:
        connection_pool.close()  # 关闭连接池
        logger.info("数据库连接池已关闭")


# lifespan参数用于在应用程序生命周期的开始和结束时执行一些初始化或清理工作
app = FastAPI(lifespan=lifespan)

pgsql的存储结构

一、Checkpoints 系列

作用：让 LangGraph Runtime 能在 分布式/长流程 场景下 断点续跑、重放、并发控制。

表名 存什么 典型字段（示意） 何时写入
checkpoints 每个「图实例」的 最新快照（state 的完整 JSONB） thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, state, created_at 每次节点执行成功后覆盖更新
checkpoint_blobs checkpoints 里 大字段的拆分（避免行过大） thread_id, checkpoint_ns, channel, type, blob 当 state 过大，自动拆分
checkpoint_migrations 记录 schema 版本/迁移脚本 version, name, applied_at 只在 setup() 时写一次
checkpoint_writes 写放大日志（每个节点写 state 的增量 diff） thread_id, checkpoint_id, task_id, idx, channel, type, value 每次节点完成时追加

关系：
checkpoints = 最新完整快照
checkpoint_writes = 所有增量历史（用于重放/审计）
checkpoint_blobs = checkpoints 里超大型 value 的切片

二、Store 系列

作用：给 业务代码（开发者） 提供 持久化 KV / 向量存储，与图运行状态无关。

表名 存什么 典型字段（示意） 何时写入
store 任意 KV 文档（LangChain Document → JSONB） uuid, namespace, key, value, created_at, updated_at 你调用 store.amput / amset 等 API
store_migrations 同 checkpoint_migrations，记录 store schema 版本 version, name, applied_at 只在第一次 setup()
store_vectors 向量索引表（embedding → vector 类型） uuid, collection_id, embedding, document, metadata 你调用 add_documents(..., embeddings=...)
vector_migrations 记录 pgvector 扩展及索引迁移版本 version, applied_at setup() 时若第一次装 pgvector

三、调用链脑图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
┌──────────────┐
│ LangGraph    │ 运行图实例
└────┬─────────┘
     │1. 写 checkpoints
     │2. 写 checkpoint_writes
     │3. 拆大字段到 checkpoint_blobs
     ▼
┌──────────────┐
│ 业务代码     │ 读写 KV/向量
└────┬─────────┘
     │1. 写 store
     │2. 写 store_vectors
     ▼
 PostgreSQL (pgsql)

在linux安装postgresql

查看linux发行版本

1
2
# 查看发行版名称和版本
cat /etc/os-release

输出

1
2
3
4
5
6
7
8
9
10
11
12
13
PRETTY_NAME="Ubuntu 24.04.2 LTS"
NAME="Ubuntu"
VERSION_ID="24.04"
VERSION="24.04.2 LTS (Noble Numbat)"
VERSION_CODENAME=noble
ID=ubuntu
ID_LIKE=debian
HOME_URL="https://www.ubuntu.com/"
SUPPORT_URL="https://help.ubuntu.com/"
BUG_REPORT_URL="https://bugs.launchpad.net/ubuntu/"
PRIVACY_POLICY_URL="https://www.ubuntu.com/legal/terms-and-policies/privacy-policy"
UBUNTU_CODENAME=noble
LOGO=ubuntu-logo

安装postgresql

1
2
3
4
5
6
7
8
9
10
11
12
13
# 更新软件包索引
sudo apt update

# 安装 PostgreSQL 和常用扩展
sudo apt install -y postgresql postgresql-contrib

# 检查服务状态
#Linux 容器/子系统（ Docker、WSL 或 LXC）没有使用 systemd ，因此 systemctl 无法工作
sudo systemctl status postgresql

#确认 PostgreSQL 已安装
which psql            # 应该输出 /usr/bin/psql
pg_ctl --version      # 显示版本号即已安装
  • postgresql 是主程序
  • postgresql-contrib 提供额外扩展（如 uuid-ossp、pgcrypto 等）

systemctl一般用于服务器上，完整的linux系统上

Linux 容器/子系统（ Docker、WSL 或 LXC）使用 service

pgsql常用指令

启动pgsql服务

1
sudo service postgresql start

停止pgsql服务

1
sudo service postgresql stop

查看状态

1
sudo service postgresql status

连接PostgreSQL 默认的系统用户，可以执行pgsql的相关指令

1
sudo -u postgres psql

psql为PostgreSQL 的终端客户端，默认会连接与当前操作系统用户名同名的数据库用户和数据库。

postgres为PostgreSQL 自带的系统数据库，postgres 默认 数据库密码为空，可以通过以下指令进行设置

1
2
-- 设置postgres用户密码
ALTER USER postgres PASSWORD 'postgres';

数据库关于user的作用

数据库里的“用户”是 PostgreSQL 内部用来做“访问控制”的一把钥匙。一句话：“谁能连哪个库、谁能读哪张表、谁能改哪些行”——全靠这些数据库用户（角色）来判定。

作用 举例
1. 认证（Authentication） 告诉 PostgreSQL “我连库时提供的用户名+密码是否合法”。
2. 授权（Authorization） 决定 “这个用户连进来后，对哪些库、哪些表、哪些行有何种权限（SELECT/INSERT/UPDATE/DELETE…）”。
3. 资源隔离（Isolation） 不同业务/团队用不同用户，方便审计、限流、回收权限，互不干扰。

pgsql通用 URL 模板

1
postgresql://<用户名>:<密码>@127.0.0.1:5432/<数据库名>[?参数=值&...]

连接pgsql

1
2
3
4
5
6
7
8
9
10
11
12
# 基于数据库持久化存储的short-term
db_uri = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# short-term短期记忆 实例化PostgresSaver对象 并初始化checkpointer
# long-term长期记忆 实例化PostgresStore对象 并初始化store
async with (
    AsyncPostgresSaver.from_conn_string(db_uri) as checkpointer,
    AsyncPostgresStore.from_conn_string(db_uri) as store

):
    await store.setup()
    await checkpointer.setup()

更换apt镜像源

1
2
3
4
5
6
7
8
cat > /etc/apt/sources.list <<'EOF'
deb http://mirrors.aliyun.com/debian/ bookworm main contrib non-free
deb http://mirrors.aliyun.com/debian/ bookworm-updates main contrib non-free
deb http://mirrors.aliyun.com/debian/ bookworm-backports main contrib non-free
deb http://mirrors.aliyun.com/debian-security bookworm-security main contrib non-free
EOF

apt update

参考资料

postgresql向量扩展pgvector的安装与入门本文简答的介绍了 rag 的架构，引申出向量数据库的作用，介绍了 - 掘金

langchain支持向量存储向量存储 | 🦜️🔗 LangChain — Vector stores | 🦜️🔗 LangChain