mini-rag 是一个手把手教你构建生产级 RAG（检索增强生成）应用的开源教学项目。它的数据库架构经历了一次教科书式的演进：从早期的MongoDB + Motor，平滑迁移到PostgreSQL + SQLAlchemy + Alembic。本文将带你完整走一遍这条迁移路线——为什么换库、如何配置连接、如何用 SQLAlchemy 定义 ORM 模型、如何用 Alembic 管理表结构版本，以及最终如何通过 Docker Compose 一键部署。

一、为什么要从 MongoDB 迁移到 PostgreSQL？

早期教程中，mini-rag 使用 MongoDB 存储项目（projects）、资产（assets）和数据块（chunks），数据以文档形式存放在集合中。但随着功能完善，MongoDB 的"自由"开始暴露短板：

• 缺乏强关系约束：chunks依赖assets，assets依赖projects，文档数据库没有外键，孤儿数据只能靠代码自觉维护；
• 事务与一致性弱：RAG 管道中一次入库要同时写资产表和数据块表，关系型事务更可靠；
• 没有内置的 Schema 版本管理：表结构变更全凭手改，难以追溯；
• 向量一体化趋势：PostgreSQL 的 pgvector 扩展让向量检索和业务数据共存于同一数据库，少维护一套存储。

迁移后的技术栈对比一目了然：

💡 依赖文件 src/requirements.txt 中你能同时看到motor、pydantic-mongo（旧栈残留）和SQLAlchemy==2.0.36、asyncpg、alembic、psycopg2、pgvector新栈组件，这正是迁移过程留下的"化石证据"。

二、连接配置：环境变量驱动的连接串

迁移的第一步是把数据库凭据收拢到统一的配置类中。在 src/helpers/config.py 里，PostgreSQL 的连接参数由五个环境变量组成：

POSTGRES_USERNAME: str POSTGRES_PASSWORD: str POSTGRES_HOST: str POSTGRES_PORT: int POSTGRES_MAIN_DATABASE: str

FastAPI 应用在启动阶段拼接连接串并创建异步引擎。关键代码位于 src/main.py：

postgres_conn = f"postgresql+asyncpg://{username}:{password}@{host}:{port}/{database}" app.db_engine = create_async_engine(postgres_conn) app.db_client = sessionmaker(app.db_engine, class_=AsyncSession, expire_on_commit=False)

三个细节值得注意：

1. postgresql+asyncpg://：指定 asyncpg 驱动，这是 SQLAlchemy 对 PostgreSQL 最流行的异步方案；
2. expire_on_commit=False：提交后对象属性不被标记过期，避免在请求结束后再次查库，是异步 ORM 的高频优化项；
3. 会话工厂挂在 app 上：路由层通过依赖注入拿到 session，Model 层统一复用。

Linux 环境下别忘了先装底层库（来自 README.md）：

sudo apt install libpq-dev gcc python3-dev

否则psycopg2编译安装会失败——这是新手踩坑率最高的问题之一。

三、SQLAlchemy 模型：从集合到表结构的映射

mini-rag 的模型统一继承一个声明式基类，定义在 src/models/db_schemes/minirag/schemes/minirag_base.py：

from sqlalchemy.ext.declarative import declarative_base SQLAlchemyBase = declarative_base()

所有表模型通过 schemes/init.py 集中导出。以核心三张表为例，展示 Mongo 文档字段是如何翻译成 SQL 列的：

1️⃣ 项目表—— project.py

class Project(SQLAlchemyBase): __tablename__ = "projects" project_id = Column(Integer, primary_key=True, autoincrement=True) project_uuid = Column(UUID(as_uuid=True), default=uuid.uuid4, unique=True, nullable=False) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now())

2️⃣ 资产表—— asset.py

class Asset(SQLAlchemyBase): __tablename__ = "assets" asset_id = Column(Integer, primary_key=True, autoincrement=True) asset_uuid = Column(UUID(as_uuid=True), default=uuid.uuid4, unique=True, nullable=False) asset_config = Column(JSONB, nullable=True) asset_project_id = Column(Integer, ForeignKey("projects.project_id"), nullable=False) project = relationship("Project", back_populates="assets")

3️⃣ 数据块表—— datachunk.py

class DataChunk(SQLAlchemyBase): __tablename__ = "chunks" chunk_uuid = Column(UUID(as_uuid=True), default=uuid.uuid4, unique=True, nullable=False) chunk_metadata = Column(JSONB, nullable=True) chunk_project_id = Column(Integer, ForeignKey("projects.project_id"), nullable=False) chunk_asset_id = Column(Integer, ForeignKey("assets.asset_id"), nullable=False)

可以看到几个典型的"PostgreSQL 化"设计：

• 双 ID 策略：自增Integer主键负责内部关系引用（外键指向它），UUID列对外暴露（API 使用），兼顾性能与安全；
• JSONB 替代 Mongo 子文档：asset_config、chunk_metadata这类半结构化数据用JSONB类型，既保留了文档式灵活性，又能建 GIN 索引；
• 外键 + relationship：project_id → asset → chunk三层关系由数据库强制执行，relationship让 ORM 可以直接project.assets遍历；
• 时间戳自动化：server_default=func.now()保证创建时间由数据库生成，onupdate=func.now()处理更新时间；
• 显式索引：__table_args__中为asset_project_id、chunk_asset_id等高频查询列声明了Index。

对比迁移前的痕迹：src/models/enums/DataBaseEnum.py 中仍保留着COLLECTION_PROJECT_NAME = "projects"这类集合名枚举——MongoDB 时代的遗产，如今只是历史注脚。

四、异步 Model 层：统一的增删改查入口

模型层通过工厂模式包装数据库操作，BaseDataModel.py 定义基类，各 Model 只关心业务查询。以 AssetModel.py 为例：

async def create_asset(self, asset: Asset): async with self.db_client() as session: async with session.begin(): session.add(asset) await session.commit() await session.refresh(asset) return asset

查询侧则使用 SQLAlchemy 2.0 风格的select()表达式（见 ProjectModel.py 的分页查询）：

query = select(Project).offset((page - 1) * page_size).limit(page_size) projects = await session.execute(query).scalars().all()

这套"工厂创建 + 异步会话 + 声明式查询"的模式，让上层路由（如 routes/data.py、routes/nlp.py）完全不感知数据库细节——换库只影响 Model 层以下，这正是平滑迁移的核心价值。

五、Alembic：让表结构变更可追溯

手动DROP TABLE/ 改列字段是生产事故的重灾区。mini-rag 引入 Alembic 做 Schema 版本控制，目录位于 src/models/db_schemes/minirag/alembic/。

1. 环境配置—— alembic/env.py 将元数据指向 ORM 基类：

target_metadata = SQLAlchemyBase.metadata

有了这行，--autogenerate才能自动对比"模型"与"数据库"的差异生成迁移脚本。

2. 版本链——versions/目录下当前有三份迁移：

3. 日常操作—— 项目文档 db_schemes/README.md 给出了标准三步流程：

# 复制模板并填入 sqlalchemy.url cp alembic.ini.example alembic.ini # 模型有变更时自动生成迁移 alembic revision --autogenerate -m "Add ..." # 应用所有迁移 alembic upgrade head

配置模板见 alembic.ini.example，把sqlalchemy.url改成你的实际连接串即可。

六、Docker 部署：pgvector 镜像 + 自动迁移

生产部署时，mini-rag 直接选用官方pgvector镜像，见 docker-compose.yml：

pgvector: image: pgvector/pgvector:0.8.0-pg17 ports: - "5400:5432" volumes: - 'pgvector:/var/lib/postgresql/data' healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres"]

两个细节体现"生产级"考虑：

• 选择 pgvector 镜像而非裸 postgres：后续教程将向量索引入库，无需额外装扩展；
• healthcheck 保障启动顺序：fastapi与celery-worker服务声明depends_on: pgvector: condition: service_healthy，数据库就绪前应用不会抢跑。

容器入口脚本 docker/minirag/entrypoint.sh 则把迁移固化到每次启动：

cd /app/models/db_schemes/minirag/ alembic upgrade head

也就是说：每次容器启动都会自动把表结构升级到最新版本，迁移脚本与代码同仓库发布，真正做到"数据库随版本走"。

七、迁移经验清单：新手避坑指南

结合 mini-rag 的完整实践，总结 5 条可复用的经验：

1. 先迁结构，再迁数据：用 Alembic 按现有集合文档结构生成初始create_table迁移（参考 初始迁移脚本），再写一次性脚本从 Mongo 导出 JSON 批量导入 Postgres；
2. 保留 UUID 对外、Int 对内：对外 API 继续用 UUID 标识资源，避免迁移改变接口契约；
3. 半结构化字段用 JSONB：不必为每个 metadata 字段建列，兼顾灵活与可索引；
4. 异步链路保持一致：Motor → asyncpg 都是异步驱动，async with会话模式几乎无缝替换，路由层零改动；
5. 迁移进容器生命周期：entrypoint.sh里alembic upgrade head一步到位，杜绝"代码是新表、库是旧结构"的错配。

八、总结

mini-rag 的这次迁移示范了 Python 生态中数据库升级的完整范式：环境配置（pydantic-settings）→ 异步引擎（asyncpg）→ ORM 建模（SQLAlchemy 2.0）→ 版本迁移（Alembic）→ 容器化部署（docker-compose）。对 RAG 类应用而言，PostgreSQL 还额外送上了 pgvector 这张"向量数据库"门票，为后续在同一个库内完成语义检索打下基础。

如果你正在为自己的 RAG 项目选型，不妨直接参考这条被教学项目完整验证过的路线：关系约束、事务、版本化迁移，一个都不会少。🚀