mongodb_persistence · MongoDB 断点持久化

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

说明与运行

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

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

完整源码

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

依赖: pip install agentframework[mongo]  (即 langgraph-checkpoint-mongodb)

关键点:
  - 网关异步, 用 MongoDBSaver(连接为 Motor 异步客户端)
  - 必须注入 PersistentSerializer(剥离含闭包的 _ctx)
  - await saver.setup() 自动建表(checkpoints + checkpoint_writes)

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

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

import asyncio

from langgraph.checkpoint.mongodb import MongoDBSaver

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

MONGO_URI = "mongodb://localhost:27017"
DB_NAME = "agentframework"


async def main() -> None:
    from motor.motor_asyncio import AsyncIOMotorClient

    client = AsyncIOMotorClient(MONGO_URI)
    saver = MongoDBSaver(client, db_name=DB_NAME, serde=PersistentSerializer())
    await saver.setup()

    wf = Workflow(name="mongo_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("断点已落 MongoDB; chatId → thread_id, 重启后交互可恢复")


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

---

回到示例索引。