Agno SQLite 持久化存储完整指南：为 Agent / Team / Workflow 落地会话与运行记录

本篇技术指南围绕 Agno 官方 SQLite 存储集成（对应 cookbook/06_storage/sqlite/README.md）展开，系统讲解如何用 SqliteDb 与 AsyncSqliteDb 两个数据库适配器，把 Agent、Team、Workflow 三类组件的会话（session）与运行记录（run）持久化到本地 SQLite 文件。读完你将掌握：连接参数与建表原理、同步/异步两套 API 的使用差异、多组件统一共享与自定义表名隔离两种用法，以及结合源码的实现细节，可直接复用到你自己的会话持久化项目中。

一、为什么需要 SQLite 持久化存储

在 Agno 中，Agent（单智能体）、Team（多智能体团队）与 Workflow（工作流）默认的生命周期状态只在进程内存中存活。一旦程序退出，会话历史、记忆与运行记录都会丢失。要构建"多轮可追溯"的智能体应用（例如客服机器人、自动化研究管线），需要把每次运行的输入输出、会话元数据落到磁盘。

Agno 的存储体系通过统一的 BaseDb 抽象（位于 libs/agno/agno/db/base.py）定义了一组接口，并提供 SQLite、Postgres、MySQL、MongoDB 等十余种适配器，其中 SQLite 是最轻量的选择：零外部服务、单文件即可运行，非常适合本地开发、Demo 演示与中小规模落地。本文对应的 cookbook/06_storage/sqlite 目录即为官方配套示例集。

目录提示：本文主题相关示例还包括 06_storage/sqlite/sqlite_for_agent.py、sqlite_for_team.py、sqlite_for_workflow.py 以及 async_sqlite 子目录 下对应异步版本。

二、快速接入：给 Agent 挂上 SqliteDb

关联文档给出的最小配置非常简洁，只差三步：

from agno.agent import Agent
from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="path/to/database.db")

agent = Agent(
    db=db,
    add_history_to_context=True,
)

要点解析：

• db_file：SQLite 数据库文件路径。相对路径会被解析为基于当前工作目录的绝对路径（源码在 sqlite.py 中通过 Path(db_file).resolve() 处理，并自动创建父目录），因此 "tmp/data.db" 这样的写法无需预先 mkdir tmp。
• add_history_to_context=True：把历史会话消息注入新一轮的模型上下文，是实现"多轮连贯对话"的关键开关；存储本身由 db=db 触发。
• 只要把 db 传给 Agent（Team、Workflow 同理），会话与运行记录便会自动写入 SQLite，业务调用方式完全不变——print_response、aprint_response 照常使用。

完整可运行示例见 sqlite_for_agent.py：它额外启用了 WebSearchTools 与 add_datetime_to_context=True，并连续问了三个问题来验证会话连续性：

if __name__ == "__main__":
    agent.print_response("How many people live in Canada?")
    agent.print_response("What is their national anthem?")
    agent.print_response("List my messages one by one")  # 依赖历史回放

第三问 "List my messages one by one" 正是对持久化历史的实测——只有运行记录被完整保存并回读，Agent 才能逐条复述历史消息。示例运行依赖 uv pip install ddgs sqlalchemy openai。

三、连接方式与默认建表行为（源码级原理）

SqliteDb 的构造（sqlite.py）支持三种显式连接参数，源码注释明确了优先级：

1. 传入 db_engine  → 直接使用该 SQLAlchemy Engine
2. 传入 db_url     → create_engine(db_url, json_serializer=...)
3. 传入 db_file    → create_engine("sqlite:///<绝对路径>")
4. 都不传          → 在当前目录创建 ./agno.db

即：db_engine > db_url > db_file > 默认 ./agno.db。所有行数据在写入时经过自定义 json_serializer 序列化，因此运行记录可以保存任意 JSON 化的结构化内容。

关于数据表的实现事实（来源于 sqlite.py 的 _create_all_tables 与 base.py 的默认命名）：

• 默认表名：会话表 agno_sessions、运行表 agno_runs；当自定义了 session_table 但未指定 runs_table 时，运行表自动命名为 {session_table}_runs。
• 自动建表：适配器按需创建多张业务表（sessions / runs / memories / metrics / evals / knowledge / learnings / schedules 等），首次使用时自动完成建表并记录 schema 版本。
• 外键与 WAL：每个新连接都会执行 PRAGMA foreign_keys = ON（让 runs 表对 sessions 的外键级联生效）与 PRAGMA journal_mode=WAL（以 WAL 日志替代默认 DELETE journal 以改善并发读写）；在不支持 WAL 的环境（如某些网络挂载盘）会退化为实际生效的模式并打印调试日志。

参数速查

构造器大部分参数用于重命名各类数据表，常用项汇总如下（完整清单见 sqlite.py 与 async_sqlite.py）：

参数	默认行为	说明
db_file	无	数据库文件路径，自动创建父目录
db_url / db_engine	无	更高优先级的连接方式，优先级见上文
session_table	agno_sessions	存储 Agent/Team/Workflow 会话的表名
runs_table	agno_runs 或 {session_table}_runs	每个会话的运行记录表
memory_table	agno_memories 一类默认名	用户记忆表
metrics_table / eval_table / knowledge_table	默认名	指标、评测、知识文档相关表
traces_table / spans_table	默认名	链路追踪相关表
id	根据连接串自动生成	数据库实例唯一标识，源码中基于 db_url or db_file or engine URL 生成稳定 ID

若要在多个逻辑隔离的会话空间之间做物理隔离，可仿照 Team 示例使用 session_table 参数（见下文）。

四、Team 场景：多个成员共用一个存储

Team（多智能体团队）把多个成员 Agent 组织起来协同完成一项任务，示例 sqlite_for_team.py 展示了"HackerNews 调研 + 网络搜索"双成员团队，并将整个团队的会话持久化到同一 SQLite 文件：

from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="tmp/data.db", session_table="new_sessions_five")

• db_file="tmp/data.db" 与 Agent 示例共用同一物理文件；
• session_table="new_sessions_five" 把 Team 的会话写入一张独立命名的表，与 Agent 会话表隔离开，避免混淆。这正是多组件共享一个数据库文件时最常用的"表级隔离"手法。

完整用法中，hn_researcher 与 web_searcher 作为 Team(members=[...]) 的成员，团队本身设置 db=db、output_schema=Article（结构化输出）与 markdown=True；运行 hn_team.print_response(...) 后，团队会话、成员运行与最终结构化产出均被持久化。运行前需 uv pip install openai ddgs newspaper4k lxml_html_clean agno。

五、Workflow 场景：把多步骤工作流跑批历史落盘

Workflow 是比 Team 更显式的流程编排结构。示例 sqlite_for_workflow.py 搭建了一个"研究 → 内容规划"的两步内容创作流水线，并把工作流本身挂到 SQLite 上：

from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="tmp/workflow.db")  # 独立文件存储工作流会话

content_creation_workflow = Workflow(
    name="Content Creation Workflow",
    description="Automated content creation from blog posts to social media",
    db=db,
    steps=[research_step, content_planning_step],
)
content_creation_workflow.print_response(input="AI trends in 2024", markdown=True)

架构说明（代码取自同一文件）：

1. hackernews_agent + web_agent 组成 research_team，负责抓取 Hackernews 帖子并从网络检索最新资讯；
2. content_planner 是一个独立 Agent，接收上一步研究结果并规划 4 周内容排期（每周 3 篇）；
3. 两个 Step（Step(name="Research Step", team=research_team) 与 Step(name="Content Planning Step", agent=content_planner)）串成工作流。

把 db=db 传给 Workflow 后，每次执行的运行记录（含各 Step 的中间产物）都会写入 tmp/workflow.db。多次运行即可形成完整的"研究-规划"轨迹档案，方便回溯与审计。

六、异步用法：AsyncSqliteDb

6.1 依赖与最小配置

对于基于 asyncio 的高并发场景，Agno 提供 AsyncSqliteDb（继承自 AsyncBaseDb，实现见 async_sqlite.py）。关联文档与 async_sqlite/README.md 都给出了一致的接入方式：

uv pip install sqlalchemy aiosqlite

from agno.agent import Agent
from agno.db.sqlite import AsyncSqliteDb

db = AsyncSqliteDb(db_file="path/to/database.db")

agent = Agent(
    db=db,
    add_history_to_context=True,
)

6.2 连接串与运行时差异

从源码 async_sqlite.py 可以看到，异步版通过 create_async_engine 创建引擎：

• 基于 db_file 时，实际驱动连接串为 sqlite+aiosqlite:///<绝对路径>；
• 都不传任何连接参数时，默认在 ./agno.db 建库；
• 同样在连接事件里执行 PRAGMA foreign_keys = ON 与 PRAGMA journal_mode=WAL（此处挂载在 db_engine.sync_engine 上）；
• 会话工厂采用 async_sessionmaker(..., expire_on_commit=False)。

注意调用差异：使用异步存储时，Agent 的交互方法也要切换到异步版本。官方示例 async_sqlite_for_agent.py 展示得非常清楚：

async def main():
    await agent.aprint_response("How many people live in Canada?")
    await agent.aprint_response("What is their national anthem called?")

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

同步示例中的 print_response(...) 在此处一律换成 await aprint_response(...)。

6.3 异步家族完整示例

async_sqlite 目录下与同步版本一一对应，覆盖三种组件：

示例文件	覆盖场景
async_sqlite_for_agent.py	Agent 使用 AsyncSqliteDb 存储
async_sqlite_for_team.py	Team 使用 AsyncSqliteDb 存储
async_sqlite_for_workflow.py	Workflow 使用 AsyncSqliteDb 存储

三者的组装模式与同步版本完全一致（Agent/Team/Workflow(db=db, ...)），仅在入口把 print_response 换成 await aprint_response，并把脚本主体包进 async def main() 后交给 asyncio.run()。

七、数据落在哪：一张图看懂读写链路

当一次 print_response 结束时，Agno 大致会经历如下路径（可从 sqlite.py 的各类 upsert_* / read_* 方法与 db/utils.py 的序列化辅助函数印证）：

用户输入
   │
   ▼
Agent/Team/Workflow 执行（print_response / aprint_response）
   │  将本次 session 与 run 序列化（json_serializer）
   ▼
SqliteDb / AsyncSqliteDb 适配器
   │  自动建表（缺失时）→ 写入 agno_sessions / agno_runs（可自定义）
   ▼
SQLite 单文件（如 tmp/data.db）
   │
   ▼
下次启动：add_history_to_context=True 时回读历史，注入新一轮模型上下文

从存储视角看，默认的核心落盘对象是两类行：

• 会话行（session）：一个 Agent/Team/Workflow 实例对应多个会话，每条会话包含会话 ID、元信息及关联运行；
• 运行行（run）：每次与模型的交互（含工具调用、消息、输出）作为一条 run 记录挂在会话下，run_index 保证严格的插入顺序，历史回放依赖它做有序读取。

add_history_to_context=True 之所以能工作，正是因为适配器实现了"按会话按序取回历史 run 并反序列化为消息上下文"的能力；从源码结构看，底层通过 SessionRunObjectCache 按行文本缓存反序列化对象，仅在行内容变化时重建，从而降低重复读取的开销（见 sqlite.py）。

八、运行示例与验证清单

8.1 运行方式

以仓库 cookbook 目录为工作区，依次运行：

# Agent（先安装依赖）
uv pip install ddgs sqlalchemy openai
python cookbook/06_storage/sqlite/sqlite_for_agent.py

# Team（依赖略多）
uv pip install openai ddgs newspaper4k lxml_html_clean agno
python cookbook/06_storage/sqlite/sqlite_for_team.py

# Workflow
python cookbook/06_storage/sqlite/sqlite_for_workflow.py

# 异步版本：先装 aiosqlite
uv pip install openai ddgs sqlalchemy aiosqlite
python cookbook/06_storage/sqlite/async_sqlite/async_sqlite_for_agent.py

8.2 验证持久化是否生效

1. 复述测试：运行 sqlite_for_agent.py，观察第三问能按顺序列出此前消息，说明历史已从库中回读；
2. 文件检查：查看是否生成 tmp/data.db、tmp/workflow.db 等数据库文件；
3. 表结构检查：用 sqlite3 tmp/data.db ".tables" 应能看到 agno_sessions、agno_runs 等表；自定义 session_table="new_sessions_five" 时会出现对应命名表；
4. 重启续聊：第二次运行同一脚本（或复用会话 ID 恢复会话），Agent 能回忆起上一进程的对话——这是"跨进程持久化"最直接的证据。

8.3 使用建议

• 开发 / Demo / 单机：直接用 SqliteDb(db_file=...)，无需任何外部服务，README 中的最小配置即可覆盖大多数演示需求。
• 生产化取舍：SQLite 天然适合轻量单机场景；若需要更高的并发写入、多实例共享或更强的隔离，可以横向迁移到同仓库 cookbook 下 06_storage/postgres、06_storage/mysql 等适配器——因为所有适配器都实现自同一套 BaseDb 接口，切换成本主要在替换 db 对象本身。
• 同库多组件：多个 Agent/Team/Workflow 可共享一个 db_file（如共用 tmp/data.db），需要隔离时用 session_table / runs_table 参数区分，官方 Team 示例即采用此模式。

九、小结

围绕 cookbook/06_storage/sqlite/README.md，本文把 SQLite 持久化从"最小配置"延伸到了完整的实战与源码层面：

• 接口层：SqliteDb 与 AsyncSqliteDb 只是把同一份 db 对象注入 Agent / Team / Workflow，组件行为不变、持久化自动生效；
• 参数层：掌握 db_file / db_url / db_engine 的优先级、session_table 等表名定制以及默认 agno_sessions / agno_runs 命名规则；
• 原理层：理解自动建表、外键级联、WAL 日志、JSON 序列化与运行顺序化存储，这些共同支撑了 add_history_to_context 的跨进程会话恢复能力；
• 异步层：AsyncSqliteDb + aiosqlite 驱动 + await aprint_response 三步即可获得 asyncio 原生体验。

把这几份示例代码跑通并对照源码观察表结构变化，你就能在自己的项目中举一反三，为任意 Agent 应用快速搭建可靠的会话与运行记录存档。