Agno PostgreSQL 持久化实践指南：为 Agent、Team 与 Workflow 接入会话存储

Agno PostgreSQL 持久化实践指南：为 Agent、Team 与 Workflow 接入会话存储

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

Agno（agno）提供PostgresDb与AsyncPostgresDb两个数据库接口，可将 Agent、Team（团队）与 Workflow（工作流）的运行会话、消息历史和记忆状态持久化到 PostgreSQL 中，从而实现跨进程、跨重启的连续性对话。本文以仓库中的 cookbook/06_storage/postgres/README.md 为骨架，结合源码与完整示例，讲解同步/异步两种连接方式、三种调用主体的接入方法以及底层实现原理。读完你可以将任意 Agno 应用一键切换到 PostgreSQL 存储后端，并在需要高并发时无缝升级到异步方案。

为什么用 PostgreSQL 存储 Agno 会话

Agno 应用默认的会话数据保存在内存中，进程结束即丢失。给 Agent、Team 或 Workflow 挂载一个数据库后端后：

• 会话（Session）持久化：每次运行（Run）的消息历史、运行状态与结果被写入数据库，重启进程后可用原有session_id恢复对话；
• 历史上下文注入：配合add_history_to_context=True，模型在每轮生成前可读取历史消息，实现多轮连续对话；
• 状态与记忆管理：session、runs、memory 等表为 会话状态管理、记忆持久化 等高级能力提供落盘基础。

相比同为 Agno 支持的 SQLite、MySQL、MongoDB 等后端（见 cookbook/06_storage 目录），PostgreSQL 提供成熟的表结构管理、事务保障与并发控制，是生产环境最常见的选型之一。

环境安装与依赖

在同步场景中，Agno 通过 SQLAlchemy 连接 PostgreSQL，驱动使用psycopg2-binary。原文档给出的安装命令是：

uv pip install psycopg2-binary

仓库中的示例脚本注释补充了运行所需的其他依赖，例如 postgres_for_agent.py 头部注明：

uv pip install ddgs sqlalchemy openai

其中ddgs与openai分别服务于示例中使用的WebSearchTools与 OpenAI 模型；sqlalchemy是PostgresDb的核心依赖——在 libs/agno/agno/db/postgres/postgres.py 的源码中，如果检测不到 SQLAlchemy，会直接抛出ImportError并提示先安装。Team 示例 postgres_for_team.py 还需要newspaper4k与lxml_html_clean。

异步方案则改用纯异步驱动psycopg，依赖声明在 async_postgres/README.md 中：

uv pip install sqlalchemy psycopg

注意：同步与异步所用驱动与连接串协议不同（psycopgvspsycopg_async），请按需安装，不要混用。

同步接入：PostgresDb 基本配置

原文档给出的最小配置如下：

from agno.agent import Agent from agno.db.postgres import PostgresDb db = PostgresDb(db_url="postgresql+psycopg://username:password@localhost:5432/database") agent = Agent( db=db, add_history_to_context=True, )

db_url遵循 SQLAlchemy 标准 URL 格式，各段含义为：

将数据库对象传入db参数后，Agent 在每次运行时都会自动把本轮会话写入数据库。

关键参数与底层行为

对照 PostgresDb.init的源码，构造函数还支持以下高频参数：

源码对连接建立顺序有明确约定：优先使用传入的db_engine，其次使用db_url，二者皆缺则抛出ValueError。因此二选一传入即可。此外，PostgresDb还支持将 session、runs、memory、metrics、eval、knowledge、traces、spans、learnings、schedules 等运行数据分表存储，并内置了建表（create_schema）与 schema 管理能力，具体表结构定义可查阅 libs/agno/agno/db/postgres/schemas.py。

三份同步示例：Agent、Team、Workflow

原文档将示例分为三个文件，分别覆盖 Agno 的三类执行主体。

1. Agent：多轮对话自动存档

完整代码见 postgres_for_agent.py：

from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.tools.websearch import WebSearchTools db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" db = PostgresDb(db_url=db_url) agent = Agent( db=db, tools=[WebSearchTools()], add_history_to_context=True, ) if __name__ == "__main__": agent.print_response("How many people live in Canada?") agent.print_response("What is their national anthem called?")

关键点：

• 连续两次提问，第二次依赖历史上下文才能回答“their national anthem”指代的内容——这正是add_history_to_context=True配合db生效的体现；
• 示例使用localhost:5532端口，本地搭建时需保证该端口的 PostgreSQL 实例可达，并预先创建好ai数据库与账号。

2. Team：多智能体团队共享存储

postgres_for_team.py 演示了将db挂载到Team上。示例构建了一个由 HackerNews 研究者与 Web 搜索者组成的团队，并使用 pydantic 定义结构化输出Article：

from agno.team import Team hn_team = Team( name="HackerNews Team", model=OpenAIChat("gpt-5.6-luna"), members=[hn_researcher, web_searcher], db=db, instructions=[ "First, search hackernews for what the user is asking about.", "Then, ask the web searcher to search for each story to get more information.", "Finally, provide a thoughtful and engaging summary.", ], output_schema=Article, markdown=True, show_members_responses=True, ) if __name__ == "__main__": hn_team.print_response("Write an article about the top 2 stories on hackernews")

这里的db复用方式与 Agent 完全一致，说明PostgresDb 实例可被 Agent、Team 等对象共享；团队内各成员的中间响应、团队会话记录都会通过同一存储后端落盘。该脚本的运行方式注释在文件头给出：

python cookbook/06_storage/postgres/postgres_for_team.py

3. Workflow：指定会话表的多步流程

postgres_for_workflow.py 演示了 Workflow 场景，且展示了构造函数中session_table参数的用法——Workflow 使用独立命名空间，避免与 Agent/Team 的会话表冲突：

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

示例流程包含两个步骤：research_step（由 HackerNews 与 Web 两个 Agent 组成的research_team执行研究）与content_planning_step（由content_planner制定四周内容计划）。由于 Workflow 本身不依赖轮询式对话而是逐步推进，把中间状态持久化到数据库，能让长流程具备断点恢复与可观测性。

异步方案：AsyncPostgresDb

在高并发或 IO 密集场景下，Agno 提供了AsyncPostgresDb，API 设计与PostgresDb对齐，只是驱动连接串换用异步协议。原文档最小配置：

from agno.agent import Agent from agno.db.postgres import AsyncPostgresDb db = AsyncPostgresDb(db_url="postgresql+psycopg_async://username:password@localhost:5432/database") agent = Agent( db=db, add_history_to_context=True, )

请留意连接串从postgresql+psycopg变为postgresql+psycopg_async，这是同步/异步方案最重要的差别。异步用例如 async_postgres_for_agent.py 所示，整个驱动链路由asyncio协调：

import asyncio from agno.agent import Agent from agno.db.postgres import AsyncPostgresDb db_url = "postgresql+psycopg_async://ai:ai@localhost:5532/ai" db = AsyncPostgresDb(db_url=db_url) agent = Agent( db=db, tools=[WebSearchTools()], add_history_to_context=True, add_datetime_to_context=True, ) 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())

配套的异步示例还有两个：

• async_postgres_for_team.py：结构上与同步 Team 示例一致，仅将hn_team.print_response(...)替换为asyncio.run(hn_team.aprint_response(...))；
• async_postgres_for_workflow.py：将同一个AsyncPostgresDb实例传入Workflow(db=db, ...)，并通过asyncio.run(content_creation_workflow.aprint_response(...))执行。

AsyncPostgresDb的实现位于 libs/agno/agno/db/postgres/async_postgres.py，与PostgresDb共享同一套 schema 与表结构设计，二者之间可以按需切换而不必改动业务逻辑（仅需更换连接串与调用前缀a）。

运行验证与排障要点

准备一个可用的 PostgreSQL 实例后，可按如下顺序验证：

# 1. 安装依赖（同步示例） uv pip install psycopg2-binary sqlalchemy openai ddgs # 2. 运行 Agent 示例，首次启动会自动建表 python cookbook/06_storage/postgres/postgres_for_agent.py # 3. 运行 Workflow 示例 python cookbook/06_storage/postgres/postgres_for_workflow.py # 4. 异步方案换用异步驱动 uv pip install psycopg sqlalchemy openai python cookbook/06_storage/postgres/async_postgres/async_postgres_for_agent.py

常见问题与排查方向：

• 连不上数据库：确认db_url中端口（默认5432，示例脚本用5532）、账号密码与库名正确，且目标库已存在；
• sqlalchemy not installed报错：PostgresDb强依赖 SQLAlchemy，需显式安装；异步示例另需psycopg（同步为psycopg2-binary）；
• 连接串协议错误：同步用postgresql+psycopg://，异步用postgresql+psycopg_async://，切勿混用；
• 表冲突：Agent/Team/Workflow 会话可分别通过session_table指定独立表（如workflow_session），避免不同主体的会话数据互相覆盖；
• 跨进程恢复会话：数据库存储后，配合 session_id 管理可在新进程中恢复既有会话，相关基础用法可对照 cookbook/06_storage/01_persistent_session_storage.py 验证。

小结

Agno 将 PostgreSQL 存储封装为一行式配置：同步场景使用PostgresDb+postgresql+psycopg://，异步场景使用AsyncPostgresDb+postgresql+psycopg_async://，二者均可直接挂载到 Agent、Team 与 Workflow 上，配合add_history_to_context实现会话级记忆与多轮连续性。仓库源码层面，连接策略（db_engine优先、其次db_url）、自动建表开关（create_schema）以及 session/runs/memory/metrics/traces 等分表参数都已内置，足以支撑从本地原型到生产部署的平滑演进。你可以在 cookbook/06_storage/postgres 目录中继续查阅全部同步与异步示例源码。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno