第 11 章 · 存储 Storage：会话持久化

本章目标：理解 Agno 2.x 的统一数据库层，学会用 SqliteDb 做开发存储、用 PostgresDb 做生产存储，并验证"重启进程后对话依然接得上"。

11.1 存储解决什么问题

上一章的聊天历史和会话状态都依赖一个前提：有数据库。没有存储的 Agent 是"金鱼"——进程重启即失忆；有了存储，同一个 session_id 就能跨进程、跨机器地续上对话。官方文档的定义非常直白：

存储让 Agent 拥有跨会话的记忆：只要使用相同的 session_id，即使进程重启，也能从上次中断的地方继续对话。（官方文档原话的中文译意）

Agno 2.x 的一个重要架构演进是统一了存储层：不再区分旧版的 Storage 与记忆库等多套配置，而是把会话 run、会话状态、用户记忆、会话摘要全部交给同一个 db 对象管理（对应表如 agno_sessions、agno_memories 等）。你只需要选对数据库驱动。

11.2 开发环境：SqliteDb

SQLite 零部署、单文件，是本地开发的默认选择：

python

# storage_basic.py —— 会话持久化最小示例
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.tools.yfinance import YFinanceTools

db = SqliteDb(db_file="tmp/agents.db")   # 一个文件就是一整个库

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    tools=[YFinanceTools(enable_stock_price=True)],
    db=db,
    add_history_to_context=True,
    num_history_runs=5,
    markdown=True,
)

session_id = "finance-session"

agent.print_response("简单分析一下英伟达的股价走势", session_id=session_id)
agent.print_response("把它和 AMD 对比一下", session_id=session_id)      # "它"=英伟达
agent.print_response("两者哪个更值得投资？", session_id=session_id)      # 完整上下文

第三次提问时模型能同时理解"两者"指什么——因为前两轮的完整消息（含工具调用结果）都在库里。

11.3 重启验证：持久化的试金石

存储是否生效，标准只有一个：杀掉进程再跑，对话还能不能接上。

python

# resume_check.py —— 第一次运行：写入
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    db=SqliteDb(db_file="tmp/agent.db"),
    add_history_to_context=True,
)

agent.print_response(
    "记住这个暗号：菠萝披萨。",
    session_id="resume-demo",
)
print("第一段运行结束，现在请结束进程后运行 resume_check_2.py")

python

# resume_check_2.py —— 第二次运行（新进程）：读取
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    db=SqliteDb(db_file="tmp/agent.db"),   # 指向同一个文件
    add_history_to_context=True,
)

agent.print_response(
    "暗号是什么？我们之前聊到哪了？",
    session_id="resume-demo",
)   # 模型应能复述暗号与上文 —— 持久化生效

两次运行之间无论间隔多久（甚至换一台装着同一份 agent.db 文件的机器），对话都能无缝衔接。

库里到底存了什么

打开 tmp/agents.db（sqlite3 tmp/agents.db ".tables"）可以看到 agno_sessions、agno_runs 等表：每轮 run 的消息列表、工具调用、metrics、session_state 快照都以 JSON 形式存放在行记录里。理解这一点有助于排查"为什么模型知道了不该知道的事"。

11.4 生产环境：PostgresDb

SQLite 不支持并发写与水平扩展，上生产应切换 PostgresDb。得益于统一接口，迁移通常只需改一行：

python

# storage_postgres.py —— 生产级存储
import os
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIChat

db = PostgresDb(
    db_url=os.getenv(
        "DB_URL",
        "postgresql+psycopg://ai:ai@localhost:5432/ai",
    ),
)

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    db=db,
    add_history_to_context=True,
    update_memory_on_run=True,   # 记忆也走同一个 db（第 12 章）
)

# 多实例部署下，只要共享同一个 Postgres，
# 任意实例都能服务同一 session_id 的请求。
agent.print_response("继续我们上次的话题", session_id="user-42-thread-1")

官方还提供异步版 AsyncPostgresDb 配合 Agent.arun() 使用；除 Postgres 外，同一套 db 抽象还有 MySQL、MongoDB、Redis、DynamoDB 等驱动可选，选型原则与你团队现有的运维栈保持一致即可。

场景	推荐	理由
本地开发/测试/演示	`SqliteDb`	零部署，单文件易备份
生产 API 服务（多副本）	`PostgresDb` / `AsyncPostgresDb`	并发安全、可水平扩展
已有 Redis/Mongo 栈	对应 Db 驱动	复用现有运维能力

11.5 存储与状态、历史的协同

把前面所学串成一个完整的心智模型：

python

# full_stack.py —— 存储 + 历史 + 状态三者协作
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.run import RunContext


def advance_stage(run_context: RunContext) -> str:
    """Move the order workflow to its next stage.

    Args:
        (无模型可见参数)
    """
    stages = ["收集需求", "确认方案", "报价", "完成"]
    current = run_context.session_state.get("stage_idx", 0)
    if current >= len(stages) - 1:
        return "流程已完成"
    run_context.session_state["stage_idx"] = current + 1
    return f"当前阶段: {stages[current + 1]}"


agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    db=SqliteDb(db_file="tmp/workflow.db"),   # 一切持久化的根基
    session_state={"stage_idx": 0},           # 状态随会话落库
    tools=[advance_stage],
    instructions="订单阶段: {stage_idx} 对应 收集需求/确认方案/报价/完成",
    add_history_to_context=True,              # 历史随会话落库
)

agent.print_response("我要定制一批 T 恤", session_id="order-9001")   # 推进到确认方案
# ……进程重启后……
agent.print_response("我们进行到哪一步了？下一步是什么？", session_id="order-9001")

一次 run 结束后，落库的内容包括：消息历史（含工具调用）、session state 快照、token metrics。db 是这三者的共同载体——这就是为什么第 10 章说"历史注入的前提是配置数据库"。

本章小结

存储让 Agent 跨进程/跨机器保持记忆；判定标准是"重启后同 session 能续聊"；
Agno 2.x 统一存储层：会话、状态、记忆、摘要共用一个 db 对象；
开发用 SqliteDb(db_file=...)，生产用 PostgresDb（异步场景 AsyncPostgresDb），迁移成本约等于改一行；
数据库里能看到 agno_sessions / agno_runs 等表，run 的消息、状态快照、metrics 全在其中。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Agno 2.x 中负责会话持久化的核心配置是？

2. 验证存储是否真正生效的最可靠方法是？

3. 从 SQLite 切换到 Postgres，Agent 的代码需要怎么改？

4. 以下哪项不是 run 结束后随会话写入数据库的内容？

🛠️ 动手实践

运行 11.3 的两段脚本完成重启验证，然后用 sqlite3 tmp/agent.db ".tables" 和 .schema agno_runs 观察表结构，截图或摘录关键列名。
把 11.5 的订单流程扩展为 5 个阶段，并在 instructions 中加入"当前阶段该问用户什么问题"的引导语，测试跨重启推进。
用 Docker 启动一个 Postgres，将本章两个示例切换到 PostgresDb 并重复重启验证，记录两种存储在行为上有无差异。

完成动手实践后，进入第 12 章：记忆 Memory——用户画像与摘要。

第 11 章 · 存储 Storage：会话持久化 ​

11.1 存储解决什么问题 ​

11.2 开发环境：SqliteDb ​

11.3 重启验证：持久化的试金石 ​

11.4 生产环境：PostgresDb ​

11.5 存储与状态、历史的协同 ​

本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 11 章 · 存储 Storage：会话持久化

11.1 存储解决什么问题

11.2 开发环境：SqliteDb

11.3 重启验证：持久化的试金石

11.4 生产环境：PostgresDb

11.5 存储与状态、历史的协同

本章小结

🛠️ 动手实践