Session Storage

使用 Session Storage 将 Agent 会话和状态持久化到数据库或文件中。

为什么我们需要 Session Storage？

Agent 是短暂的，内置的内存仅能维持单个执行周期。

在生产环境中，我们通过 API 提供（或触发）Agent，需要在多次请求中延续同一会话。Storage 能将会话历史记录和状态持久化到数据库中，让我们能够接续之前的进度。

Storage 还允许我们检查和评估 Agent 会话，提取 few-shot 示例，并构建内部监控工具。它让我们能够查看数据，这有助于我们构建更好的 Agent。

将存储添加到 Agent、Team 或 Workflow 中，只需提供一个 Storage 驱动程序，Agno 会处理其余部分。你可以使用 Sqlite、Postgres、Mongo 或任何你想要的数据库。

下面是一个演示跨执行周期持久化的简单示例：

storage.py

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.sqlite import SqliteStorage
from rich.pretty import pprint

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # 固定 session id 以在执行周期之间延续相同的会话
    session_id="fixed_id_for_demo",
    storage=SqliteStorage(table_name="agent_sessions", db_file="tmp/data.db"),
    add_history_to_messages=True,
    num_history_runs=3,
)
agent.print_response("What was my last question?")
agent.print_response("What is the capital of France?")
agent.print_response("What was my last question?")
pprint(agent.get_messages_for_session())

首次运行此代码时，“What was my last question?” 的答案将不可用。但再次运行后，Agent 将能够正确回答。由于我们固定了会话 ID，Agent 每次运行时都将从相同的会话继续。

Storage 的优势

Storage 通常是 Agent Engineering 中一个讨论较少的环节——但我们认为它是生产环境中 Agent 应用的幕后英雄。

在生产环境中，你需要 storage 来：

• 延续会话：检索会话历史记录并接续之前的进度。
• 获取会话列表：要延续之前的会话，你需要维护一个可用于该 Agent 的会话列表。
• 保存运行之间的状态：将 Agent 的状态保存到数据库或文件中，以便稍后进行检查。

但还有更多：

• Storage 保存我们的 Agent 会话数据以供检查和评估。
• Storage 帮助我们提取 few-shot 示例，可用于改进 Agent。
• Storage 使我们能够构建内部监控工具和仪表板。

Storage 是 Agentic 基础设施的关键组成部分，绝不应外包给第三方。你应该几乎始终为你自己的 Agent 使用自己的存储层。

示例：使用 Postgres 进行存储

1. 运行 Postgres

   安装 docker desktop 并使用以下命令在端口 5532 上运行 Postgres：

   docker run -d \
     -e POSTGRES_DB=ai \
     -e POSTGRES_USER=ai \
     -e POSTGRES_PASSWORD=ai \
     -e PGDATA=/var/lib/postgresql/data/pgdata \
     -v pgvolume:/var/lib/postgresql/data \
     -p 5532:5432 \
     --name pgvector \
     agno/pgvector:16
   
2. 创建带有 Storage 的 Agent

   创建一个名为 agent_with_storage.py 的文件，内容如下：

   import typer
   from typing import Optional, List
   from agno.agent import Agent
   from agno.storage.postgres import PostgresStorage
   from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
   from agno.vectordb.pgvector import PgVector, SearchType
   
   db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
   knowledge_base = PDFUrlKnowledgeBase(
       urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
       vector_db=PgVector(table_name="recipes", db_url=db_url, search_type=SearchType.hybrid),
   )
   storage = PostgresStorage(table_name="pdf_agent", db_url=db_url)
   
   def pdf_agent(new: bool = False, user: str = "user"):
       session_id: Optional[str] = None
   
       if not new:
           existing_sessions: List[str] = storage.get_all_session_ids(user)
           if len(existing_sessions) > 0:
               session_id = existing_sessions[0]
   
       agent = Agent(
           session_id=session_id,
           user_id=user,
           knowledge=knowledge_base,
           storage=storage,
           # 在响应中显示工具调用
           show_tool_calls=True,
           # 使 Agent 能够读取聊天历史记录
           read_chat_history=True,
           # 我们也可以自动将聊天历史记录添加到发送给模型的消息中
           # 但向模型提供聊天历史记录并非总是有用的，因此我们提供一个工具给它
           # 仅在需要时使用。
           # add_history_to_messages=True,
           # 要添加到消息中的历史响应数量。
           # num_history_responses=3,
       )
       if session_id is None:
           session_id = agent.session_id
           print(f"Started Session: {session_id}\n")
       else:
           print(f"Continuing Session: {session_id}\n")
   
       # 将 Agent 作为 CLI 应用运行
       agent.cli_app(markdown=True)
   
   
   if __name__ == "__main__":
       # 加载知识库：首次运行后注释掉
       knowledge_base.load(upsert=True)
   
       typer.run(pdf_agent)
   
3. 运行 Agent

   安装库

   Mac

   pip install -U agno openai pgvector pypdf "psycopg[binary]" sqlalchemy
   

   运行 Agent

   python agent_with_storage.py
   

   现在 Agent 可以跨会话延续了。问一个问题：

   How do I make pad thai?
   

   然后输入 bye 退出，重新启动应用程序并提问：

   What was my last message?
   
4. 开始新的运行

   使用 --new 标志运行 agent_with_storage.py 文件以开始新的运行。

   python agent_with_storage.py --new
   

Schema Upgrades

在使用 AgentStorage 时，基于 SQL 的存储类具有固定的模式。随着新 Agno 功能的发布，模式可能需要更新。

升级可以手动或自动完成。

自动升级

自动升级是在存储类的构造函数中将 auto_upgrade_schema 参数设置为 True 时完成的。 你只需要为 Agent 运行设置一次，模式就会被升级。

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
storage = PostgresStorage(table_name="agent_sessions", db_url=db_url, auto_upgrade_schema=True)

手动升级

可以通过在存储类上调用 upgrade_schema 方法来手动升级模式。

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
storage = PostgresStorage(table_name="agent_sessions", db_url=db_url)
storage.upgrade_schema()

Params

Parameter	Type	Default	Description
storage	Optional[AgentStorage]	None	Agent 的存储机制。

开发者资源

• 查看 Cookbook