mysql_persistence · MySQL 断点持久化

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

说明与运行

示例: MySQL Checkpoint 持久化(官方生态 saver 直连). 依赖: pip install agentframework[mysql] (即 langgraph-checkpoint-mysql) 要求: MySQL ≥ 8.0.19 / MariaDB ≥ 10.7.1 关键点: - 网关异步, 用 AIOMySQLSaver(asyncmy 异步驱动) - 必须注入 PersistentSerializer(剥离含闭包的 _ctx) - await saver.setup() 自动建表(checkpoints + checkpoint_writes) 运行前提: 本地有可连接的 MySQL 实例(修改下方连接参数)。 参考: https://github.com/langchain-ai/langgraph-checkpoint-mysql

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

完整源码

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

依赖: pip install agentframework[mysql]  (即 langgraph-checkpoint-mysql)
要求: MySQL ≥ 8.0.19 / MariaDB ≥ 10.7.1

关键点:
  - 网关异步, 用 AIOMySQLSaver(asyncmy 异步驱动)
  - 必须注入 PersistentSerializer(剥离含闭包的 _ctx)
  - await saver.setup() 自动建表(checkpoints + checkpoint_writes)

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

参考: https://github.com/langchain-ai/langgraph-checkpoint-mysql
"""

import asyncio

from langgraph.checkpoint.mysql import AIOMySQLSaver

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

MYSQL_PARAMS = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "password",
    "database": "agentframework",
    "autocommit": True,
}


async def main() -> None:
    from asyncmy import create_pool

    pool = await create_pool(**MYSQL_PARAMS)
    saver = AIOMySQLSaver(pool, serde=PersistentSerializer())
    await saver.setup()

    wf = Workflow(name="mysql_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")

    app = wf.compile(store=saver)
    print(f"已编译工作流 {wf.name}, checkpointer: {type(app.checkpointer).__name__}")
    print("断点已落 MySQL; chatId → thread_id, 重启后交互可恢复")


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

---

回到示例索引。