跳转到内容
AI Agent 教程 AI Agent 教程

Agno 数据库可插拔适配器模式

Agno 数据库可插拔适配器模式

学习目标

本章将分析 Agno 的数据库适配器架构：

  • BaseDb / AsyncBaseDb 统一抽象接口
  • 10+ 数据库后端的实现概览
  • ComponentType 枚举的组件区分
  • 数据库迁移脚本管理

项目实践

统一抽象接口

Agno 的数据库层围绕两个抽象基类构建：

# 伪代码：BaseDb 抽象
class BaseDb(ABC):
    @abstractmethod
    def create_tables(self) -> None: ...


    @abstractmethod
    def session_exists(self, session_id: str, user_id: Optional[str] = None) -> bool: ...


    @abstractmethod
    def read_session(self, session_id: str, user_id: Optional[str] = None) -> Optional[Dict]: ...


    @abstractmethod
    def upsert_session(self, session: Dict) -> None: ...


    @abstractmethod
    def delete_session(self, session_id: str, user_id: Optional[str] = None) -> None: ...


    # ... 更多抽象方法


class AsyncBaseDb(BaseDb):
    @abstractmethod
    async def acreate_tables(self) -> None: ...


    @abstractmethod
    async def asession_exists(self, ...) -> bool: ...


    # ... 所有方法的 async 版本

关键设计：

  • AsyncBaseDb 继承 BaseDb，sync 方法仍然可用（调用同步版本）
  • 每个方法有明确的签名，实现类必须遵循

10+ 数据库后端

后端目录适用场景
PostgreSQLdb/postgres/生产环境，事务支持
AsyncPostgreSQLdb/async_postgres/异步生产环境
SQLitedb/sqlite/开发/测试环境
JSONdb/json/文件存储，无数据库
In-Memorydb/in_memory/测试，无持久化
MongoDBdb/mongo/文档数据库
Redisdb/redis/缓存 + 持久化
DynamoDBdb/dynamo/AWS 托管
Firestoredb/firestore/GCP 托管
GCS JSONdb/gcs_json/Google Cloud Storage
SingleStoredb/singlestore/分布式 SQL
SurrealDBdb/surrealdb/多模型数据库

ComponentType 枚举

Agno 使用 ComponentType 区分不同组件的数据存储：

# 伪代码
class ComponentType(str, Enum):
    AGENT = "agent"
    TEAM = "team"
    WORKFLOW = "workflow"

同一数据库后端为 Agent、Team、Workflow 分别维护独立的 session 表/集合：

PostgreSQL:
  agent_sessions 表
  team_sessions 表
  workflow_sessions 表


MongoDB:
  agent_sessions 集合
  team_sessions 集合
  workflow_sessions 集合

SessionType 区分

# 伪代码
class SessionType(str, Enum):
    AGENT = "agent"
    TEAM = "team"
    WORKFLOW = "workflow"

SessionType 与 ComponentType 配合，在数据库操作中标记会话类型。

数据库迁移

db/migrations/ 目录包含迁移脚本：

db/migrations/
  001_create_agent_sessions.py
  002_create_team_sessions.py
  003_create_workflow_sessions.py
  004_add_metrics_column.py
  ...

PostgreSQL 后端使用 Alembic 风格的迁移，其他后端可能有简化的迁移逻辑。

配置解析

db/utils.py 提供数据库配置解析：

# 伪代码
def resolve_db_from_config(db_url: str) -> BaseDb:
    if db_url.startswith("postgresql"):
        return PostgresDb(db_url=db_url)
    elif db_url.startswith("sqlite"):
        return SqliteDb(db_url=db_url)
    elif db_url.startswith("mongodb"):
        return MongoDb(db_url=db_url)
    # ... 更多后端

问题与规避

1. 抽象接口与具体实现的差距

BaseDb 定义的方法在某些数据库中无法完美映射（如 Redis 的键值存储不支持复杂查询）。

规避：适配器实现可能使用数据结构模拟（如在 Redis 中使用 Hash 结构模拟表）。对于不支持的操作，抛出 NotImplementedError。

2. 迁移脚本的后端差异

不同数据库的迁移语法不同（PostgreSQL 用 SQL，MongoDB 用 JavaScript）。

规避：每个后端实现自己的迁移逻辑。BaseDb.create_tables() 自动执行当前后端的迁移。

3. 性能差异

In-Memory 和 Redis 后端比 PostgreSQL 快很多，但功能受限。

规避：

  • 开发用 SQLite/In-Memory
  • 生产用 PostgreSQL/AsyncPostgreSQL
  • 缓存层用 Redis

设计取舍

为什么支持这么多数据库后端？

优势：用户可以选择已有的基础设施，无需额外部署数据库 代价：维护成本高，每个后端都需要测试 替代方案：只支持 PostgreSQL 和 SQLite——覆盖大多数场景，但失去灵活性

为什么 AsyncBaseDb 继承 BaseDb？

优势：sync 方法在异步后端中仍然可用（通过 asyncio.to_thread 包装同步调用） 代价：AsyncBaseDb 需要实现两组方法 替代方案：两个完全独立的接口——更干净但不能混用 sync/async

参考来源

版本信息

📅 2026-05-30 🏷️ 1444c9b 🔀 1444c9b

标签

databaseadapter-patternpluggable-backendabstraction