很多项目一开始只有一个数据库。于是 AI 很自然地写出：

async def get_user(user_id: int):
    async with sqlite_session() as session:
        ...

或者更直接：

conn = sqlite3.connect("app.db")

功能当然能跑,但问题会在以后出现。

今天是 SQLite ，明天可能要换成 PostgreSQL …

随着代码规模的增加，项目很快就会变成：

业务代码和具体存储技术绑死。

这正是 Multi-Data-Source Architecture（多数据源架构） 要解决的问题。

📌 技术名片

Multi-Data-Source Architecture / 多数据源架构

指一个系统同时使用多个不同的数据源，并通过统一的架构边界管理它们，而不是让业务代码直接依赖具体数据库实现。

常见数据源包括：

SQLite
PostgreSQL
MySQL
DuckDB

这里最重要的不是：

“支持多少种数据库。”

而是：

业务代码是否需要知道底层到底是什么数据库。

💡 一句话理解

可以把多数据源架构理解成：

业务层只负责“我要什么数据”，基础设施层负责“去哪里、怎么拿”。

通俗理解：数据库就像物流公司

假设你经营一家电商。业务部门只关心：

把商品送到客户手里。

至于运输方式：

顺丰
京东物流
DHL
海运
空运

不应该写进订单业务本身。错误设计可能是：

订单是国内：
    调顺丰 API

订单是欧洲：
    调 DHL API

订单是大件：
    调海运 API

更加合理的是，订单系统只说：

“帮我发货。”

物流层再决定具体怎么做。

数据库也是一样。业务层应该说：

查询用户
保存 Agent
读取知识库
执行分析 SQL

而不是：

SQLite 用这个语法
PostgreSQL 用那个语法
DuckDB 再单独判断

一、跨数据源架构

1. 业务层不能直接依赖数据库

一个健康的调用关系应该是：

Router
  ↓
Service
  ↓
Repository / Data Access
  ↓
Database Driver / ORM
  ↓
Database

其中：

Repository（仓储层）

负责封装数据访问。

业务 Service 不应该知道：

SQLAlchemy
sqlite3
aiosqlite
psycopg
DuckDB connection

它只应该知道：

user = await user_repository.get_by_id(user_id)

而不是：

async with session.execute(...):

2. ORM 的意义不只是“少写 SQL”

这里需要引出一个重要术语：

ORM / Object-Relational Mapping / 对象关系映射

ORM 可以把 Python Object 和 Relational Database Table 建立映射；比如 User 对应 users 表 。

很多人理解 ORM 时，只想到：

少写 SQL。

其实在多数据库架构里，它还有一个更重要的价值：

隔离不同数据库之间的大量语法和驱动差异。

例如 SQLAlchemy 可以通过不同：

Dialect（方言）

适配不同数据库。

应用层写：

select(User).where(User.id == user_id)

底层可以根据数据库转换为相应 SQL。

3. 但 ORM 也不是“万能兼容层”

这里需要特别说明，不同数据库之间仍然存在大量差异，例如：

数据类型
JSON 支持
全文检索
分页语法
UPSERT
自增主键
日期函数
锁机制
事务隔离
索引
窗口函数

如果 AI 大量写：

text("某数据库专属 SQL")

那么虽然用了 ORM：

项目依然被绑在某个数据库上。

因此规范应该是：

优先使用 ORM 的通用表达能力，数据库专属语法只能留在基础设施层。

二、miniagent 的主业务数据库

1. 创建

miniagent 当前主业务数据库采用：

SQLite
+
SQLAlchemy Async

应用启动时，在 ServiceContainer 中统一创建：

database_url = f"sqlite+aiosqlite:///{db_path}"

self.engine = create_async_engine(
    database_url,
    echo=False,
    future=True,
)

self.session_factory = async_sessionmaker(
    bind=self.engine,
    ...
)

然后再统一把：

engine
session_factory

注入不同的数据访问对象：

self.user_db = AsyncUserDatabase(
    self.engine,
    self.session_factory
)

self.chat_db = AsyncChatDatabase(
    self.engine,
    self.session_factory
)

self.kb_db = AsyncKnowledgeBaseDatabase(
    self.engine,
    self.session_factory
)

这个结构非常关键。

因为业务 Service 并没有各自 create_engine() ，而是统一复用数据库基础设施。

2. 统一 Session 管理，避免 AI 到处 commit/rollback

数据库访问还有一个非常容易被 AI 写乱的问题：

Transaction（事务）

AI 很容易在不同地方分别写：

await session.commit()

另一个地方：

await session.rollback()

再一个地方忘记 close。

于是 Session 生命周期到处散落。

miniagent 提供统一的 AsyncBaseDatabase，其中：

@asynccontextmanager
async def get_session(self):

    async with self.AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()

        except SQLAlchemyError:
            await session.rollback()
            raise

        finally:
            await session.close()

这里的：

Transaction（事务）

可以理解为：

一组数据库操作要么全部成功，要么全部失败。

这样 Repository 层可以统一使用：

async with self.get_session() as session:
    ...

而不用让每个业务函数重新考虑：

什么时候 commit？
失败要不要 rollback？
什么时候 close？

3. 多数据源不等于“所有数据库强行统一”

这是这篇文章最重要的一点之一。

miniagent 本身就是一个很好的例子。

它的：

用户
Agent
知识库配置
权限
聊天记录
系统设置

属于典型业务数据。

适合：

SQLAlchemy
↓
SQLite

但 SQL Agent 面向的是另一种场景：

分析型数据查询。

所以 miniagent 还单独存在 DuckDBManager：

self.conn = duckdb.connect(db_path)

def execute(self, sql, params=None):
    return self.conn.execute(
        sql,
        params or []
    ).fetchall()

也就是说，miniagent 并没有强行要求：

所有数据必须通过同一种数据库访问方式。

而是：

Operational Data
业务数据
     ↓
SQLAlchemy / SQLite

Analytical Data
分析数据
     ↓
DuckDB

这是非常合理的。

4. 为什么分析型数据库可以单独存在？

因为：

SQLite
PostgreSQL
MySQL

这类数据库更擅长：

OLTP（Online Transaction Processing，联机事务处理）

例如：

创建用户
修改 Agent
保存聊天记录
权限管理

而 DuckDB 更偏向：

OLAP（Online Analytical Processing，联机分析处理）

例如：

分析 CSV
聚合百万行数据
GROUP BY
统计报表
临时分析

所以正确的多数据源架构并不是：

“找一种数据库解决所有问题。”

而是：

不同的数据源承担不同职责。

5. 一个更完整的多数据源架构

下面是 miniagent 的完整数据架构：

miniagent数据架构

从上图可见，miniagent 主要有两条明显不同的数据访问路线。

第一条是业务数据：

Router
   ↓
Service
   ↓
Async*Database / Repository
   ↓
AsyncBaseDatabase
   ↓
SQLAlchemy
   ↓
SQLite

ServiceContainer 统一创建 Engine 和 SessionFactory，再注入多个业务数据库对象，而不是让每个模块自行创建数据库连接。

AsyncBaseDatabase 则进一步统一管理 Session、Commit、Rollback 和 Close 生命周期。

第二条是分析数据：

SQL Agent
   ↓
DuckDBManager
   ↓
DuckDB

DuckDBManager 直接封装 DuckDB 连接与 SQL 执行。

因此 miniagent 当前真正体现的是：

flowchart TB A[miniagent] --> B[Business / Runtime] B --> C[Business Data] B --> D[Analytics Data] C --> E[Repository Layer] E --> F[AsyncBaseDatabase] F --> G[SQLAlchemy] G --> H[SQLite] D --> I[SQL Agent] I --> J[DuckDBManager] J --> K[DuckDB]

它不是：

“所有数据库必须统一使用 SQLAlchemy。”

而是：

不同数据源采用适合自己的访问方式，但具体数据库实现被限制在基础设施边界之后。

这才是多数据源架构更实际的落地方式。

6. 跨数据源的核心不是“兼容”，而是“隔离”

很多人会把目标写成：

我要兼容 MySQL、SQLite、PostgreSQL。

但更好的目标应该是：

当数据库改变时，改动尽量限制在基础设施层。

也就是说：

Database Changed
      ↓
Infrastructure
      ↓
Repository

而不是：

Database Changed
 ├── Router 修改
 ├── Service 修改
 ├── Tool 修改
 ├── API 修改
 └── 前端修改

这就是：

Change Isolation（变化隔离）

架构真正想解决的就是变化成本。

7. Repository 层就是数据库的“防火墙”

可以把 Repository 理解成：

业务世界和数据库世界之间的一堵墙。

左边：

Business

说：

给我一个 User
保存 Agent
查询 KB

右边：

Database

考虑：

SQL
Join
Session
Transaction
Dialect
Index
Connection

中间 Repository 负责翻译。

例如：

Service
   ↓
user_repository.get_by_id(10)
   ↓
Repository
   ↓
SQLAlchemy
   ↓
SQLite / PostgreSQL

所以一个非常重要的 AI Rule 是：

Service 不允许跨过 Repository 直接操作数据库。

三、多数据源架构

1. 应该怎么分层？

可以采用这样的规则：

flowchart TB A[Application / Service] --> B[Data Access Abstraction] B --> C[Repository] B --> D[Adapter] B --> E[Client] C --> F[SQL DB] D --> G[DuckDB] E --> H[Vector DB]

其中：

Repository

适合 业务数据库 CRUD。

Adapter

适合 不同协议或特殊数据源适配。

Client / Manager

适合：

Vector Store
外部 Search Engine

重点是：

上层看到的是“能力”，而不是“产品名字”。

2. 什么时候应该抽统一接口？

例如业务层只需要：

await user_repository.get_by_id(id)

那么可以有：

UserRepository
      │
      ├── SQLAlchemyUserRepository
      └── FutureOtherRepository

但不要为了所谓“未来可能支持十个数据库”，一开始就创建：

AbstractDatabaseFactoryProviderManagerAdapter

这样的巨大体系。

架构需要留扩展点，但不要过度设计。

对于很多 Python 项目：

Service
↓
Repository
↓
SQLAlchemy

已经足够隔离 SQLite → PostgreSQL 这类变化。

3. 有什么好处？

数据库迁移成本更低

假设：

开发环境：SQLite

以后生产改成：

PostgreSQL

如果业务大量使用 SQLAlchemy 的通用能力，改动主要集中于：

Database URL
Driver
Migration
少量 Dialect 差异

而不是重新修改所有 Service。

可以让不同数据库发挥各自优势

例如：

业务事务
→ PostgreSQL / SQLite

分析查询
→ DuckDB

向量搜索
→ Vector Database

缓存
→ Redis

而不是用一把锤子处理所有问题。

更容易测试

Service 如果只依赖：

Repository

测试时可以替换成：

Fake Repository

而不需要真的启动数据库。

数据库能力不会污染业务语义

业务代码仍然是：

await agent_service.create(...)

而不是：

SQLite INSERT
PostgreSQL ON CONFLICT
MySQL UPSERT

业务代码更容易理解。

更适合 AI 编程

AI 最危险的一点是：

看见一个能工作的写法，就复制到整个项目。

如果项目规定：

Router → Service → Repository → Database

AI 就有明确路径。

否则它很容易：

哪里需要数据
→ 哪里直接连数据库

四、给 AI 建立多数据源架构规则

可以直接加入 Project Rules：

## 多数据源架构

业务代码禁止直接依赖具体数据库实现。

调用关系：

Router
→ Service
→ Repository / Data Adapter
→ Database Client / ORM
→ Data Source

规则：

- Router 禁止创建数据库连接；
- Service 禁止出现数据库专属 SQL；
- 普通业务数据库访问统一放在 Repository；
- Connection、Session、Transaction 和 Driver 生命周期属于基础设施层；
- 普通 CRUD 优先使用 ORM 通用表达；
- 特定数据库 SQL 只能留在数据访问层；
- 禁止在业务层散落 database_type 判断；
- 不同数据源允许使用不同访问技术；
- 不强制 DuckDB、向量数据库、Redis 等套进关系型 ORM。

核心目标：

统一的是“访问边界”，
不是强迫所有数据源使用同一种实现。

提示词落地：不要只告诉 AI“支持 PostgreSQL”

错误提示词：

把这个项目改成同时支持 SQLite 和 PostgreSQL。

AI 很可能开始：

if db_type == "sqlite":
    ...
else:
    ...

然后到处改。

更加合理的提示词：

请为当前项目增加 PostgreSQL 兼容能力。

要求：

1. 先检查现有数据库访问边界；
2. Router 和 Service 不允许感知数据库类型；
3. 普通 CRUD 继续复用现有 Repository；
4. SQLAlchemy Engine 和 SessionFactory 由基础设施层统一创建；
5. 优先使用 SQLAlchemy 通用表达；
6. 找出 SQLite 专属 SQL 或数据类型；
7. 数据库差异只允许封装在 infrastructure / repository；
8. 禁止在业务代码新增 db_type if/else；
9. 保持现有 Service API 不变；
10. 给出 SQLite 与 PostgreSQL 的兼容差异清单。

这样 AI 的目标就从：

“哪里报错就加 if。”

变成：

“把数据库差异关进正确的架构层。”

添加新数据源时也先问 AI 几个问题

例如要引入 DuckDB、Redis 或向量数据库，可以要求 AI 先判断：

在实现之前，请回答：

1. 新数据源承担什么职责？
2. 它是业务事务数据还是分析数据？
3. 是否应该复用现有 Repository？
4. 是否需要独立 Adapter / Manager？
5. Service 应看到什么统一能力？
6. 哪些数据库专属逻辑必须被隔离？
7. Connection 生命周期由谁管理？
8. 是否需要事务？
9. 是否需要连接池？
10. 是否真的应该和现有数据库统一实现？

尤其是最后一个问题非常重要。

因为：

多数据源架构不意味着把所有数据源做成一样。

五、以后 miniagent 可以怎样自然演进？

如果以后业务数据库从 SQLite 切换：

SQLite
↓
PostgreSQL

只要继续保持：

Service
↓
Repository
↓
SQLAlchemy

多数业务逻辑不需要知道这个变化。

而系统又可以继续保留 DuckDB 用于分析，同时 Chroma / Vector Store 承担向量检索，未来还可能有 Redis 承担共享 Value Cache。

最终形成：

flowchart TB A[Application] --> B[Service Layer] B --> C[Data Access Boundaries] C --> D[Relational DB] C --> E[DuckDB] C --> F[Vector DB] C --> G[Redis] D --> H[PostgreSQL] E --> I[Analytics] F --> J[Retrieval] G --> K[Cache]

上层业务不需要直接处理这些差异。

总结

跨多数据源架构真正要解决的问题，并不是：

“怎样写代码同时连接五种数据库？”

而是：

“怎样让五种数据库存在时，业务代码仍然保持干净？”

真正重要的原则是：

业务定义需要什么数据
        ↓
数据访问层决定如何取得
        ↓
基础设施决定使用什么数据库

对于普通业务 CRUD：

Service
↓
Repository
↓
ORM
↓
Relational Database

对于特殊数据源：

Service / Tool
↓
Dedicated Adapter / Manager
↓
DuckDB / Vector DB / Redis

所以，多数据源架构最值得写进 AI 项目规则的一句话不是：

“项目要支持多数据库。”

而是：

禁止 AI 把具体数据库写进业务逻辑；统一的是访问边界，不是强迫所有数据源使用同一种实现。

这样以后不管底层是：

SQLite
PostgreSQL
DuckDB
Redis
Vector Database

真正变化的，应该主要是基础设施，而不是整个项目。

开源代码

🪐祝您好运🪐