企业官网建设流程全解析

agno AgentOS 数据库与媒体存储配置指南：从 SQLite 到生产级 Postgres 与 S3/GCS

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

本篇技术指南围绕 agno AgentOS 的持久化配置展开，核心讲解如何通过AgentOS(db=...)注入默认数据库、通过AgentOS(media_storage=...)将媒体文件字节外置到对象存储，以及如何启动服务、执行数据库迁移与删除会话媒体。读完本篇，你将掌握 AgentOS 默认数据库继承机制、SQLite/Postgres/SurrealDB 等十余种存储后端的接入方式、外部媒体存储的完整读写与删除链路，以及生产环境下的 schema 迁移命令。文中所有配置与命令均可在当前仓库的 cookbook/05_agent_os/02_databases 目录中直接运行验证。

AgentOS 数据库设计：一次注入，全局继承

AgentOS 的持久化设计遵循"默认值 + 组件级覆盖"原则。将数据库传给AgentOS(db=...)后，该数据库会成为所有未显式指定自身数据库的 agent、team 和 workflow 的默认存储后端；任何组件级db配置始终优先于默认值。

这一行为在示例 basic.py 中有直接体现：database_agent在构造时刻意省略db参数，由AgentOS将注入的SqliteDb自动分配给它。

from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS db = SqliteDb( id="agent-os-default-db", db_file="tmp/databases.db", ) # 该 agent 故意不设置 db；AgentOS 会注入默认数据库 database_agent = Agent( id="database-agent", name="Database Agent", model=OpenAIResponses(id="gpt-5.5"), instructions="Answer questions concisely.", markdown=True, ) agent_os = AgentOS( id="database-basics-os", description="AgentOS default-database inheritance with SQLite.", db=db, agents=[database_agent], auto_provision_dbs=True, ) app = agent_os.get_app()

从源码看，auto_provision_dbs是AgentOS构造函数的显式参数，默认值为True（见 libs/agno/agno/os/app.py）。当服务启动时，若该参数为真，FastAPI 的 lifespan 钩子会依次调用_initialize_sync_databases()与_initialize_async_databases()，在事件循环中创建所需的数据表；只有当你使用外部迁移流程接管 schema 管理时，才应将其关闭。

@asynccontextmanager async def db_lifespan(app: FastAPI, agent_os: "AgentOS"): """Initializes databases in the event loop and closes them on shutdown.""" if agent_os.auto_provision_dbs: agent_os._initialize_sync_databases() await agent_os._initialize_async_databases()

一个实用的验证技巧：服务启动后访问http://localhost:7777/config，对比OS 数据库 ID与各 agent 的数据库 ID，即可确认默认数据库是否正确继承到了所有组件。

文件清单：本目录六个示例的职责划分

本目录下的示例文件按照"数据库后端"与"媒体存储"两个主题组织：

后端参考：支持的存储适配器一览

README 提供了一张完整的后端参考表，涵盖导入路径、连接方式与所需服务。以下按类别整理并补充必要的使用说明：

几条值得注意的边界说明（均来自原文档）：

• Neon 与 Supabase使用 Postgres wire 协议，因此应把它们的连接串直接传给PostgresDb，无需独立适配器。
• 异步 Postgres需导入AsyncPostgresDb并使用postgresql+psycopg_async://...形式的 URL。
• ClickHouse只实现了 trace 与 span 数据面。会话、记忆、知识、评估与组件等业务数据应使用行存储（如 Postgres）。

生产持久化示例：同步与异步 Postgres

postgres.py 展示了如何在同一份代码中按环境变量切换同步与异步适配器：

from os import getenv from agno.agent import Agent from agno.db.postgres import AsyncPostgresDb, PostgresDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS sync_db = PostgresDb( id="agent-os-postgres-sync", db_url="postgresql+psycopg://ai:ai@localhost:5532/ai", ) async_db = AsyncPostgresDb( id="agent-os-postgres-async", db_url="postgresql+psycopg_async://ai:ai@localhost:5532/ai", ) use_async = getenv("AGENTOS_USE_ASYNC_POSTGRES", "false").lower() == "true" db = async_db if use_async else sync_db postgres_agent = Agent( id="postgres-agent", name="Postgres Agent", model=OpenAIResponses(id="gpt-5.5"), instructions="Answer questions concisely.", markdown=True, ) agent_os = AgentOS( id="postgres-agent-os", description="AgentOS backed by sync or async Postgres.", db=db, agents=[postgres_agent], ) app = agent_os.get_app()

同步适配器是默认且最简单的方式；当数据库操作必须在异步应用中保持非阻塞时，再通过AGENTOS_USE_ASYNC_POSTGRES=true切换到异步变体。示例中的连接串默认指向本地 pgvector 容器（localhost:5532），对应仓库脚本 scripts/run_pgvector.sh。

SurrealDB：以构造参数代替连接串

surreal.py 展示了 SurrealDB 与众不同的接入方式——不使用 SQL 连接串，而是显式传入 client、URL、凭据、namespace 与 database，且所有值均可通过环境变量覆盖：

from os import getenv from agno.agent import Agent from agno.db.surrealdb import SurrealDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS db = SurrealDb( client=None, db_url=getenv("SURREALDB_URL", "ws://localhost:8000"), db_creds={ "username": getenv("SURREALDB_USER", "root"), "password": getenv("SURREALDB_PASSWORD", "root"), }, db_ns=getenv("SURREALDB_NAMESPACE", "agno"), db_db=getenv("SURREALDB_DATABASE", "agent_os"), id="agent-os-surreal", )

默认值ws://localhost:8000与root/root与仓库脚本 scripts/run_surrealdb.sh 中的本地启动配置保持一致。

外部媒体存储：数据库只存引用，字节交给对象存储

数据库负责存储会话文本，而media_storage决定文件字节的去向。将后端传给AgentOS(media_storage=...)后，上传与生成的媒体文件会被写入对象存储，会话行中只保留一个MediaReference，而非 base64 内联数据。媒体随后通过GET /sessions/{session_id}/media/{storage_key}路由对外提供（该路由定义见 libs/agno/agno/os/routers/media/media.py）。

每个后端都有对应的Async变体，供异步应用使用。

S3 媒体存储示例

s3_media_storage.py 将媒体字节送往 S3，并同时把数据库、媒体存储、文件生成工具绑定到同一 Agent：

import os from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.media.storage.s3 import AsyncS3MediaStorage from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.tools.file import FileGenerationTools from dotenv import load_dotenv load_dotenv() bucket = os.getenv("AGNO_FILE_OUTPUT_S3_BUCKET") if not bucket: raise ValueError( "AGNO_FILE_OUTPUT_S3_BUCKET must be set to the destination S3 bucket" ) db = SqliteDb(db_file="tmp/agentos_media_storage.db") storage = AsyncS3MediaStorage( bucket=bucket, region=os.getenv("AWS_REGION"), # 未设置时回退到 AWS_DEFAULT_REGION 或 ~/.aws/config prefix="agno/agentos/files/", presigned_url_expiry=3600, ) file_agent = Agent( id="media-storage-agent", name="Media Storage Agent", model=OpenAIResponses(id="gpt-5.5"), db=db, media_storage=storage, store_media=True, add_history_to_context=True, tools=[FileGenerationTools(all=True)], description="Analyze uploaded media and generate files stored in S3.", instructions=[ "Read and answer questions about attached media and files.", "Use the appropriate file-generation tool when the user requests an output file.", "Always use a descriptive filename with the correct extension.", "Briefly explain what you read or generated.", ], markdown=True, ) agent_os = AgentOS( id="agentos-media-storage", name="AgentOS Media Storage", agents=[file_agent], db=db, media_storage=storage, ) app = agent_os.get_app()

示例通过store_media=True开启媒体持久化，AsyncS3MediaStorage还支持prefix（对象前缀，便于统一管理目录）与presigned_url_expiry（预签名 URL 有效期，秒）两个实用参数。运行时可附加图片或 CSV 提问，再要求 Agent 生成 CSV，观察文件落桶而数据库仅存引用。

GCS 媒体存储示例

gcs_media_storage.py 与 S3 版本结构完全对称，仅更换存储后端与凭据方式：

db = SqliteDb(db_file="tmp/agentos_gcs_media_storage.db") storage = AsyncGCSMediaStorage( bucket=bucket, project=os.getenv("GCP_PROJECT"), credentials_path=os.getenv("GOOGLE_APPLICATION_CREDENTIALS"), prefix="agno/agentos/files/", presigned_url_expiry=3600, )

GCS 的认证使用 Google Cloud Application Default Credentials，也可通过GOOGLE_APPLICATION_CREDENTIALS指向服务账号 JSON 文件。若未设置AGNO_FILE_OUTPUT_GCS_BUCKET，示例会直接抛出ValueError提示。

region 参数：容易被忽略的"保存成功但加载失败"陷阱

原文档特别提醒：当桶不在默认区域时，务必传入region。上传操作会自行找到正确的区域，但媒体 URL 的签名中携带 region 信息——如果不传 region，媒体会成功保存，随后却因签名区域不匹配而加载失败。这是 S3/GCS 媒体存储最典型的隐蔽故障点。

读取与删除会话媒体：完整的生命周期管理

media_storage_delete.py 演示了媒体从写入到删除的完整链路。默认情况下，媒体对象的生命周期长于会话——run 上的引用是记录"哪个对象属于哪个会话"的唯一凭据，因此删除时必须先读取行中的 storage key，再清理底层对象。

db = SqliteDb(db_file="tmp/agentos_media_delete.db") storage = AsyncS3MediaStorage( bucket=bucket, region=os.getenv("AWS_REGION"), prefix="agno/agentos/files/", presigned_url_expiry=3600, ) file_agent = Agent( id="media-delete-agent", name="Media Delete Agent", model=OpenAIResponses(id="gpt-5.5"), db=db, media_storage=storage, store_media=True, description="Answer questions about attached files.", markdown=True, ) agent_os = AgentOS( id="agentos-media-delete", name="AgentOS Media Delete", agents=[file_agent], db=db, media_storage=storage, # 读与删路由都通过它解析 storage key ) app = agent_os.get_app()

从源码看（libs/agno/agno/os/routers/session/session.py），删除会话时传入delete_media=true，服务端会先从会话行中收集所有关联的 storage key，再调用adelete_media_keys批量清除对象；该清理是 best-effort 的——行已删除，存储失败不应导致整个删除请求失败。媒体 key 的收集还做了会话作用域限定：另一会话的引用不属于本会话可删除的范围。

运行方式与前置条件

环境准备

• 所有示例仅在 agent 运行并调用模型时需要OPENAI_API_KEY；
• Postgres 示例需先启动本地服务：./cookbook/scripts/run_pgvector.sh；
• SurrealDB 示例需安装agno[surrealdb]并运行./cookbook/scripts/run_surrealdb.sh；
• S3 相关示例需安装agno[s3]，设置AGNO_FILE_OUTPUT_S3_BUCKET与 AWS 凭据；
• GCS 示例需安装agno[gcs]，设置AGNO_FILE_OUTPUT_GCS_BUCKET，并使用 Google Cloud Application Default Credentials 认证。

启动命令

SQLite 默认数据库：

.venvs/demo/bin/python cookbook/05_agent_os/02_databases/basic.py

同步 Postgres：

.venvs/demo/bin/python cookbook/05_agent_os/02_databases/postgres.py

异步 Postgres：

AGENTOS_USE_ASYNC_POSTGRES=true \ .venvs/demo/bin/python cookbook/05_agent_os/02_databases/postgres.py

SurrealDB：

.venvs/demo/bin/python cookbook/05_agent_os/02_databases/surreal.py

S3 媒体存储：

AGNO_FILE_OUTPUT_S3_BUCKET=my-bucket \ .venvs/demo/bin/python cookbook/05_agent_os/02_databases/s3_media_storage.py

GCS 媒体存储：

AGNO_FILE_OUTPUT_GCS_BUCKET=my-bucket \ .venvs/demo/bin/python cookbook/05_agent_os/02_databases/gcs_media_storage.py

读取并删除会话媒体：

AGNO_FILE_OUTPUT_S3_BUCKET=my-bucket \ .venvs/demo/bin/python cookbook/05_agent_os/02_databases/media_storage_delete.py

每个示例服务都监听7777 端口。

运行时 API 验证与数据库迁移

服务启动后，可以按如下顺序做端到端验证（以媒体删除示例为例）：

1. 向一次 run 附加文件；
2. GET /sessions/{session_id}查看会话中的MediaReference；
3. GET /sessions/{session_id}/media/{storage_key}流式读回媒体；
4. DELETE /sessions/{session_id}?delete_media=true同时删除行记录与底层对象。

数据库迁移方面，AgentOS 暴露了迁移 API（实现见 libs/agno/agno/os/routers/database.py）。先从GET /config读取该服务的数据库 ID，然后触发迁移：

curl -X POST http://localhost:7777/databases/<db-id>/migrate

如需迁移到指定 schema 版本，在 URL 后追加?target_version=<version>即可。从源码可以看出，迁移管理器会根据目标版本与当前版本的高低自动选择升级（up）或降级（down）；远程数据库（RemoteDb）会被拒绝迁移——它归属另一个 AgentOS 实例，应由其所有者执行迁移。

小结

AgentOS 的持久化架构将"会话数据"与"媒体字节"清晰地分层：db决定结构化数据的落点（本地开发用 SQLite，生产用 Postgres），media_storage决定文件的去向（S3/GCS 等对象存储），MediaReference则作为二者之间的桥梁。配合auto_provision_dbs自动建表与/databases/{db_id}/migrate版本化迁移，开发者可以从本地 SQLite 无缝过渡到生产级 Postgres，同时将媒体负载安全地卸载到对象存储。相关示例与测试可直接在 cookbook/05_agent_os/02_databases 目录下运行验证，完整数据库适配器实现位于 libs/agno/agno/db，媒体存储实现位于 libs/agno/agno/media/storage。

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