sqlite_persistence · SQLite 断点持久化¶
本页收录可运行示例，配套本文档「指南 06」(指南 06)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。
说明与运行¶
示例: Checkpoint 持久化(SQLite 官方 saver 直连).
演示 S1 断点持久化的两条核心能力:
1. interactive 暂停 → 模拟进程重启(新连接 + 新 saver + 同一 db 文件)
→ 挂起交互不丢失, 恢复值续跑成功
2. 会话续接: 同一 chatId 的新请求从上次断点状态继续
关键点:
- 网关是异步的, 必须用 AsyncSqliteSaver(同步 SqliteSaver 不支持 async 方法)
- 必须注入 PersistentSerializer: 框架 _ctx 含 stream_callback 闭包,
默认 msgpack 序列化器无法落库, 该序列化器落库前剥离 _ctx、恢复时网关注入新 ctx
- 官方 saver 经 await saver.setup() 自动建表, 无需手写建表语句
运行:
cd agentframework
PYTHONPATH=src python examples/sqlite_persistence.py
运行前需在已安装
agentframework的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如chromadb/ 数据库驱动）会在说明中注明。
完整源码¶
"""示例: Checkpoint 持久化(SQLite 官方 saver 直连).
演示 S1 断点持久化的两条核心能力:
1. interactive 暂停 → 模拟进程重启(新连接 + 新 saver + 同一 db 文件)
→ 挂起交互不丢失, 恢复值续跑成功
2. 会话续接: 同一 chatId 的新请求从上次断点状态继续
关键点:
- 网关是异步的, 必须用 `AsyncSqliteSaver`(同步 SqliteSaver 不支持 async 方法)
- 必须注入 `PersistentSerializer`: 框架 `_ctx` 含 stream_callback 闭包,
默认 msgpack 序列化器无法落库, 该序列化器落库前剥离 _ctx、恢复时网关注入新 ctx
- 官方 saver 经 `await saver.setup()` 自动建表, 无需手写建表语句
运行:
cd agentframework
PYTHONPATH=src python examples/sqlite_persistence.py
"""
import asyncio
import tempfile
import aiosqlite
import httpx
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from agentframework import Workflow
from agentframework.gateway.app import create_app
from agentframework.gateway.registry import AppRegistry
from agentframework.nodes import Answer, UserSelect
from agentframework.runtime.checkpointer import PersistentSerializer
async def build_app(db_path: str):
"""构建带 SQLite 持久化断点的网关应用.
saver 持有到 db 文件的连接; setup() 自动建表(checkpoints + checkpoint_writes).
返回 (FastAPI 应用, aiosqlite 连接) —— 连接由调用方在使用后关闭(否则进程不退出).
"""
conn = await aiosqlite.connect(db_path)
saver = AsyncSqliteSaver(conn, serde=PersistentSerializer())
await saver.setup()
wf = Workflow(name="sqlite_persistence")
wf.add_node(
UserSelect(
name="sel",
description="请选择操作",
options=[
{"key": "query", "value": "查询"},
{"key": "exit", "value": "退出"},
],
)
)
wf.add_node(Answer(name="ans", text=wf.ref("sel.selected")))
wf.add_edge("sel", "ans")
compiled = wf.compile(store=saver) # 官方 saver 直传: 框架注入 PersistentSerializer
registry = AppRegistry()
registry.register("app-sqlite", compiled, name="SQLite 持久化演示")
return create_app(registry), conn
async def main() -> None:
db_path = tempfile.mktemp(suffix=".db")
# ---------- 第一次请求(进程 A): 执行到 UserSelect 暂停 ----------
app_a, conn_a = await build_app(db_path)
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app_a), base_url="http://test"
) as client:
r1 = await client.post(
"/api/v2/chat/completions",
json={
"appId": "app-sqlite",
"chatId": "session-1",
"messages": [{"role": "user", "content": "你好"}],
"stream": True,
},
)
await conn_a.close()
assert "event: interactive" in r1.text, "应输出 interactive 暂停事件"
print("① 首次请求 → 暂停, 输出 interactive 事件")
# ---------- 模拟进程重启: 新连接 + 新 saver + 同一 db 文件 ----------
app_b, conn_b = await build_app(db_path)
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app_b), base_url="http://test"
) as client:
r2 = await client.post(
"/api/v2/chat/completions",
json={
"appId": "app-sqlite",
"chatId": "session-1",
"messages": [{"role": "user", "content": "查询"}],
"stream": True,
},
)
await conn_b.close()
assert "event: answer" in r2.text and "查询" in r2.text, "重启后应恢复并输出 answer"
print("② 模拟进程重启后恢复 → 挂起交互不丢失, 恢复值续跑成功")
# ---------- 会话隔离: 新 chatId = 独立断点, 互不干扰 ----------
app_c, conn_c = await build_app(db_path)
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app_c), base_url="http://test"
) as client:
r3 = await client.post(
"/api/v2/chat/completions",
json={
"appId": "app-sqlite",
"chatId": "session-2",
"messages": [{"role": "user", "content": "另一会话"}],
"stream": True,
},
)
await conn_c.close()
assert "event: interactive" in r3.text
print("③ 新会话(独立 chatId) → 独立断点, 互不干扰")
if __name__ == "__main__":
asyncio.run(main())
print("\nSQLite 断点持久化演示完成 ✅(断点已落库, 重启可恢复)")
回到示例索引。