postgres_persistence · PostgreSQL 断点持久化

本页收录可运行示例，配套本文档「指南 06」(指南 06)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。

说明与运行

示例: PostgreSQL Checkpoint 持久化(官方 saver 直连). 依赖: pip install agentframework[pg] (即 langgraph-checkpoint-postgres) 关键点: - 网关异步, 用 AsyncPostgresSaver(需 psycopg async 连接) - 必须注入 PersistentSerializer(剥离含闭包的 _ctx) - await saver.setup() 自动建表(checkpoints + checkpoint_writes) 运行前提: 本地有可连接的 PostgreSQL 实例(修改下方 DSN)。 参考: https://langchain-ai.github.io/langgraph/reference/checkpoints/

运行前需在已安装 agentframework 的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如 chromadb / 数据库驱动）会在说明中注明。

完整源码

"""示例: PostgreSQL Checkpoint 持久化(官方 saver 直连).

依赖: pip install agentframework[pg]  (即 langgraph-checkpoint-postgres)

关键点:
  - 网关异步, 用 AsyncPostgresSaver(需 psycopg async 连接)
  - 必须注入 PersistentSerializer(剥离含闭包的 _ctx)
  - await saver.setup() 自动建表(checkpoints + checkpoint_writes)

运行前提: 本地有可连接的 PostgreSQL 实例(修改下方 DSN)。

参考: https://langchain-ai.github.io/langgraph/reference/checkpoints/
"""

import asyncio

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

from agentframework import Workflow
from agentframework.nodes import Answer, UserSelect
from agentframework.runtime.checkpointer import PersistentSerializer

DSN = "postgresql://user:password@localhost:5432/agentframework"


async def main() -> None:
    from psycopg import AsyncConnection

    conn = await AsyncConnection.connect(DSN, autocommit=True)
    saver = AsyncPostgresSaver(conn, serde=PersistentSerializer())
    await saver.setup()

    wf = Workflow(name="pg_persistence")
    wf.add_node(
        UserSelect(
            name="sel",
            description="请选择操作",
            options=[{"key": "a", "value": "查询"}],
        )
    )
    wf.add_node(Answer(name="ans", text=wf.ref("sel.selected")))
    wf.add_edge("sel", "ans")

    # 官方 saver 直传: 框架识别为 BaseCheckpointSaver 并复用(注入剥离序列化器)
    app = wf.compile(store=saver)
    print(f"已编译工作流 {wf.name}, checkpointer: {type(app.checkpointer).__name__}")
    print("断点已落 PostgreSQL; chatId → thread_id, 重启后交互可恢复")


if __name__ == "__main__":
    asyncio.run(main())

---

回到示例索引。