恢复 SQLite 历史记录

重新打开一个真实的临时数据库，恢复历史记录，对消息分页，更新标题，然后删除会话。

运行示例

在仓库根目录下：

uv run python docs/_examples/sessions.py

输出包含：

Raw roles: system, user, assistant, system, user, assistant

这个程序使用确定性的离线模型，并在需要时使用临时存储。它不需要 API 密钥，由文档测试套件执行。

可以带入你应用的做法

为一个对话保留一个稳定的 session_id，并在进程重启时针对同一个数据库打开持久化扩展。示例在创建第二个客户端之前先关闭了第一个连接，因此恢复的历史记录来自 SQLite，而不是内存缓存。

async def main() -> None:
    # No demo touches the default database under the user's home directory.
    with TemporaryDirectory(prefix="zett-docs-") as directory:
        path = Path(directory) / "sessions.sqlite"
        session_id = "project-notes"
        config = AgentRunConfig(session_id=session_id)
        first_storage = SQLiteSessionExtension(path)
        try:
            first = await create_agent(TurnModel(), config=config, extensions=[first_storage])
            print((await first.run("Remember Python", metadata={"origin": "tutorial"})).content)
        finally:
            await first_storage.close()

        restored_storage = SQLiteSessionExtension(path)
        try:
            restored = await create_agent(TurnModel(), config=config, extensions=[restored_storage])
            print((await restored.run("Continue")).content)
            rows = await restored_storage.list_raw_messages(session_id)
            assert [row.message.role.value for row in rows] == [
                "system",
                "user",
                "assistant",
                "system",
                "user",
                "assistant",
            ]
            assert rows[0].metadata == {"origin": "tutorial"}
            assert len(await restored_storage.list_raw_messages(session_id, limit=2, offset=2)) == 2
            view = await restored_storage.storage.load(session_id)
            assert view.snapshot is None and len(view.raw_tail) == 6
            await restored_storage.update_session(session_id, title="Project notes")
            renamed = await restored_storage.get_session(session_id)
            assert renamed is not None and renamed.title == "Project notes"
            assert await restored_storage.delete_session(session_id)
            assert await restored_storage.list_raw_messages(session_id) == []
            print("Raw roles: system, user, assistant, system, user, assistant")
            print("Session restored, renamed, and deleted in a temporary database")
        finally:
            await restored_storage.close()

示例还会对消息分页、重命名对话并删除它。它只删除临时数据库中的会话。在你的应用中，删除会话会移除历史记录；应把它置于明确的用户操作之后。

完整源码

下载 sessions.py。

"""Restore a conversation through a new SQLite connection and a new Agent."""

import asyncio
from collections.abc import AsyncIterator
from pathlib import Path
from tempfile import TemporaryDirectory

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.messages import (
    AssistantMessage,
    UserMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)


class TurnModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        turn = sum(isinstance(message, UserMessage) for message in request.messages)
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=f"Stored turn {turn}")))


async def main() -> None:
    # No demo touches the default database under the user's home directory.
    with TemporaryDirectory(prefix="zett-docs-") as directory:
        path = Path(directory) / "sessions.sqlite"
        session_id = "project-notes"
        config = AgentRunConfig(session_id=session_id)
        first_storage = SQLiteSessionExtension(path)
        try:
            first = await create_agent(TurnModel(), config=config, extensions=[first_storage])
            print((await first.run("Remember Python", metadata={"origin": "tutorial"})).content)
        finally:
            await first_storage.close()

        restored_storage = SQLiteSessionExtension(path)
        try:
            restored = await create_agent(TurnModel(), config=config, extensions=[restored_storage])
            print((await restored.run("Continue")).content)
            rows = await restored_storage.list_raw_messages(session_id)
            assert [row.message.role.value for row in rows] == [
                "system",
                "user",
                "assistant",
                "system",
                "user",
                "assistant",
            ]
            assert rows[0].metadata == {"origin": "tutorial"}
            assert len(await restored_storage.list_raw_messages(session_id, limit=2, offset=2)) == 2
            view = await restored_storage.storage.load(session_id)
            assert view.snapshot is None and len(view.raw_tail) == 6
            await restored_storage.update_session(session_id, title="Project notes")
            renamed = await restored_storage.get_session(session_id)
            assert renamed is not None and renamed.title == "Project notes"
            assert await restored_storage.delete_session(session_id)
            assert await restored_storage.list_raw_messages(session_id) == []
            print("Raw roles: system, user, assistant, system, user, assistant")
            print("Session restored, renamed, and deleted in a temporary database")
        finally:
            await restored_storage.close()


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

下一步：阅读保存并恢复对话了解存储选择和资源归属，或阅读压缩而不丢失历史记录了解长对话。