astron-agent Memory 服务的 Alembic 数据库迁移机制：启动自动升级、双数据库适配与手动回滚实战

• 人工智能
• AI Agent
• Agent 工作流
• RPA
• 后端
• 前端
• 工具调用

本文围绕 astron-agent 仓库中core/memory/database/alembic/目录下的迁移文档与配套源码展开，讲清 Memory（mdb）服务如何借助 Alembic 实现“服务启动即自动完成 schema 迁移”：包括 Redis 分布式锁防并发、存量库 stamp 兜底、DB_TYPE驱动的 PostgreSQL/MySQL 双适配、autogenerate 工作流以及downgrade回滚的完整操作与风险边界。读完本文，你可以独立完成该模块的表结构变更、生成与审查迁移文件，并安全地执行版本回退。

一、文档定位与迁移体系总览

core/memory/database/alembic/README.md 的标题是 "Alembic Database Migration"，并明确这是一套 "Generic single-database configuration"（通用单库配置）。它约定了三层能力：

1. 启动自动迁移：服务启动时自动检测并补齐数据库结构；
2. 手动迁移命令：创建新迁移（autogenerate / 空模板）、按版本回滚；
3. 模型变更工作流：从修改 SQLModel 到提交迁移文件的标准化流程。

迁移体系由以下几部分文件协作构成：

二、启动自动迁移：Redis 锁 + upgrade head + 存量库 stamp

README 给出的自动迁移行为是三步：

1. Check if tables exist withoutalembic_version→ stamp to current version
2. Runalembic upgrade headto apply any pending migrations
3. Use Redis lock to ensure only one instance runs migrations at a time

2.1 入口：main.py 中在启动 Web 服务前执行迁移

从 core/memory/database/main.py 的__main__分支可以看到执行顺序：先做扩展初始化（rep_initialize_extensions()）、Kong 探活注册（_log_ready_after_delay()），然后才是数据库迁移与 uvicorn 启动（见 L299-L322）：

if __name__ == "__main__": # database init asyncio.run(rep_initialize_extensions()) # kong init asyncio.run(_log_ready_after_delay()) # alembic init run_database_migration() uvicorn.run( app="main:create_app", host="0.0.0.0", port=int(os.getenv("SERVICE_PORT", "7990")), ... )

注意迁移发生在 uvicorn 之前，且run_database_migration()是同步阻塞调用——这保证 Web 服务对外提供请求时表结构一定已就绪。

2.2 迁移执行器：锁、升级与兜底逻辑

database_migration.py 的run_database_migration()是自动迁移的核心实现，关键逻辑如下：

def run_database_migration() -> None: memory_dir = Path(__file__).parent.parent.parent.parent.parent alembic_dir = memory_dir / "database" / "alembic" alembic_ini = alembic_dir / "alembic.ini" if not alembic_ini.exists(): logging.error("alembic.ini not found: %s", alembic_ini) raise FileNotFoundError(f"alembic.ini not found: {alembic_ini}") config = Config(str(alembic_ini)) config.set_main_option("script_location", str(alembic_dir)) cache_service = get_cache_service() is_locked = cache_service.setnx( "memory_database_migration_lock", "locked", expire_time=60 ) if is_locked: try: command.upgrade(config, "head") except Exception as e: if "already exists" in str(e): try: command.stamp(config, "f2a4ce6e3198") command.upgrade(config, "head") ... logging.error("Database migration failed: %s", e)

逐点解读：

• alembic.ini存在性校验：路径通过源码相对位置推导（memory_dir/database/alembic/alembic.ini），找不到时直接抛FileNotFoundError阻止服务启动，避免“缺迁移目录还能跑”的隐患。
• script_location显式设置：虽然alembic.ini里写的是script_location = .，程序化调用时改为绝对路径alembic_dir，保证versions/解析与 CWD 无关。
• Redis 分布式锁：通过cache_service.setnx("memory_database_migration_lock", "locked", expire_time=60)抢占锁。setnx的实现在 core/common/service/cache/redis_cache.py（L106-L124），底层是SET key value NX EX 60，即“键不存在才写入且 60 秒过期”。这意味着：多实例同时启动时只有一实例真正执行upgrade head，其余实例直接跳过；即使持锁实例崩溃，锁也会在 60 秒后自动释放，不会形成死锁。这也正是 README 第 3 条 "Use Redis lock to ensure only one instance runs migrations at a time" 的落地方式。
• 存量库兜底（对应 README 第 1 条）：若数据库里表已存在但没有alembic_version记录（例如早期由 ORMcreate_all建表的库），upgrade head会因重复建表报 "already exists"。代码捕获该特征字符串后，执行command.stamp(config, "f2a4ce6e3198")—— 将库直接标记为初始迁移版本（即 versions/2026_02_11_1801-f2a4ce6e3198_init.py 的 revision ID），再重新upgrade head补齐后续版本。这是把“无版本记录的旧库”平滑纳入 Alembic 管理的典型手法。

三、配置层解析：alembic.ini 与 env.py

3.1 alembic.ini：文件名模板与空 URL

alembic.ini 中有两处对迁移工程实践影响较大的配置：

[alembic] # path to migration scripts script_location = . # template used to generate migration file names # Use date-time prefix format file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s ... # Database URL will be read from config.env file via env.py # You can also override it by setting MYSQL_URL environment variable # Format: mysql+pymysql://user:password@host:port/database sqlalchemy.url =

• file_template带日期前缀：生成的迁移文件形如2026_02_11_1801-f2a4ce6e3198_init.py（年月日-时分 + revision 短哈希 + slug）。相比默认纯哈希命名，时间前缀让versions/目录按创建时间天然有序，排查“哪个版本何时引入”时更易读。
• sqlalchemy.url留空：连接串不落在配置文件里，而是运行时由env.py依据环境变量注入，避免明文密码入库。

3.2 env.py：由 DB_TYPE 适配器动态构建连接

env.py 在模块加载时做了两件事：

alembic_env_path = Path(__file__).resolve() alembic_dir = alembic_env_path.parent database_dir = alembic_dir.parent memory_dir = database_dir.parent project_root = memory_dir.parent sys.path.append(str(project_root)) try: # Import all models for SQLModel metadata registration from memory.database.domain.models.database_meta import DatabaseMeta # noqa: F401 from memory.database.domain.models.schema_meta import SchemaMeta # noqa: F401 print("SQLModel and models load success!") except ImportError as e: print(f"load failed: {e}") sys.exit(1)

• 注册 metadata 是 autogenerate 的前提：把DatabaseMeta、SchemaMeta全部导入，让SQLModel.metadata收集到目标表结构；导入失败直接sys.exit(1)，防止生成基于不完整 metadata 的错误迁移。
• 动态 URL：get_database_url()（L43-L52）通过get_adapter()拿到当前数据库适配器，用适配器给出的环境变量前缀（PGSQL/MYSQL）读取{prefix}_USER/_PASSWORD/_DATABASE/_HOST/_PORT，再调用build_sync_url()生成同步驱动的 URL，最后config.set_main_option("sqlalchemy.url", ...)注入 ini 配置。

各适配器的同步驱动差异（迁移场景使用同步连接）：

DB_TYPE的取值由 registry.py 中的get_adapter()判定：默认postgresql，支持postgresql/mysql，其他值抛ValueError。而DB_TYPE与连接参数本身来自服务配置 core/memory/database/config.env，例如：

# Database type selection: postgresql or mysql DB_TYPE=postgresql PGSQL_HOST=127.0.0.1 PGSQL_PORT=5432 PGSQL_USER=xxxx PGSQL_PASSWORD=xxxx PGSQL_DATABASE=xxxx # MySQL database configuration (used when DB_TYPE=mysql) MYSQL_HOST=127.0.0.1 MYSQL_PORT=3306 ...

main.py在模块级通过os.environ["CONFIG_ENV_PATH"] = "./memory/database/config.env"指定该配置路径（L29），这也是 README 中 "Database URL will be read from config.env file via env.py" 的完整链路。

3.3 include_object：按数据库类型裁剪迁移对象范围

env.py的include_object()（L62-L105）是双数据库适配的关键细节：

def include_object(object, name, type_, reflected, compare_to): if type_ == "foreign_key_constraint": return False adapter = _get_adapter() db_type = adapter.get_db_type() if db_type == "postgresql": # Only include objects from sparkdb_manager schema if type_ == "schema": return name == "sparkdb_manager" if type_ == "table" and hasattr(object, "schema"): return object.schema == "sparkdb_manager" else: # MySQL: include all tables (no schema qualifier) if type_ == "schema": return False return True

• 统一忽略外键约束：foreign_key_constraint一律不纳入 autogenerate，避免自动生成的迁移里塞入难以预测的 FK DDL；
• PostgreSQL：只跟踪sparkdb_manager这个管理 schema 下的对象（业务库中每个租户 schema 是动态创建的，不应进入 Alembic 版本链）；
• MySQL：不跟踪 schema 对象（MySQL 的 database 即 schema，但本服务通过USE直连库，不需要 Alembic 管理库级对象）。

run_migrations_online()（L138-L179）在context.configure()中同时启用了compare_type=True、compare_server_default=True，让 autogenerate 能发现列类型与默认值变更；并把适配器的get_alembic_version_table_schema()/get_alembic_include_schemas()传入 context——PostgreSQL 下alembic_version表落在sparkdb_managerschema 内（PostgreSQLAdapter.get_alembic_version_table_schema()返回"sparkdb_manager"），MySQL 下返回None（版本表直接建在连接库中，见 mysql_adapter.py L268-L273）。

另一个实用钩子是process_revision_directives（L148-L153）：当 autogenerate 检测到script.upgrade_ops.is_empty()（无变更）时，直接清空 directives 不生成文件，并记录 "No changes in schema detected."。这防止了“改了一下代码但没有 schema 变化”时产生空迁移文件污染版本链。

四、初始迁移文件：database_meta 与 schema_meta 两张表

versions/2026_02_11_1801-f2a4ce6e3198_init.py 是当前唯一的迁移（revision = "f2a4ce6e3198"，down_revision = None），其结构正是 3.2 节模型与 3.3 节过滤规则的产物：

def _get_schema() -> Union[str, None]: """Get the schema based on DB_TYPE. PostgreSQL uses sparkdb_manager schema, MySQL uses None.""" db_type = os.getenv("DB_TYPE", "postgresql").lower() if db_type == "postgresql": return "sparkdb_manager" return None

upgrade()创建两张元数据表及索引：

• database_meta：id（BigInteger 主键，雪花 ID）、uid、name、description、space_id、create_at/update_at、create_by/update_by；在name、space_id、uid上各建普通索引（PostgreSQL 下索引名带sparkdb_manager_前缀，MySQL 下为ix_database_meta_*，见 L53-L85 的条件命名）；
• schema_meta：id、database_id（BigInteger）、schema_name、时间戳与审计字段；在schema_name上建索引。

downgrade()按“先删索引、再删表”的逆序操作回滚。这与 domain/models/database_meta.py 和 domain/models/schema_meta.py 中的 SQLModel 定义一一对应——模型同样通过_get_table_args()委托适配器取得__table_args__（PostgreSQL 返回{"schema": "sparkdb_manager"}，MySQL 返回{}），保证“模型声明”与“迁移 DDL”使用同一套 schema 规则。

五、手动迁移命令（继承原文档）

README 提供的手动命令完整如下。所有命令均应在core/memory/database/alembic/目录（即alembic.ini所在处）执行，且执行前保证DB_TYPE、{PGSQL|MYSQL}_*等环境变量已按 config.env 约定配置。

创建新迁移

# Auto-generate migration from model changes alembic revision --autogenerate -m "description of changes" # Create empty migration file alembic revision -m "description of changes"

• --autogenerate依赖env.py中注册的SQLModel.metadata与include_object过滤规则，生成内容已被裁剪为“仅sparkdb_manager相关、不含外键”；
• 无 schema 变更时不会落盘文件（见 3.3 节process_revision_directives）；
• 不带--autogenerate则基于 script.py.mako 模板生成空骨架，适合纯数据迁移（data migration）场景。

回滚迁移

README 原文档特别强调了三条回滚注意事项，这里完整保留：

Important Notes:

1. Downgrade database first- Run the downgrade command in thenew version code(which contains the downgrade migration)
2. Then downgrade service- After database downgrade is complete, downgrade the workflow service to the compatible version and restart it
3. Data loss risk- Downgrading may cause data loss if the migration removes columns or tables. Always backup your database first

即：先用新版代码（含对应 downgrade 迁移）回退数据库，再回退服务版本并重启；由于 downgrade 可能删列删表，务必先备份。对应命令：

# Downgrade one step alembic downgrade -1 # Downgrade to specific version alembic downgrade <revision> # Downgrade all migrations alembic downgrade base

以当前版本链为例，alembic downgrade base会从 head 一路执行到f2a4ce6e3198的downgrade()，最终删除database_meta、schema_meta两张表及其索引——这正是第 3 条“数据丢失风险”的具体体现。

六、模型变更的标准工作流

README 给出的五步流程如下，其中第 1 步的模型路径需按本模块实际位置理解：Memory 服务的 SQLModel 位于 core/memory/database/domain/models/（DatabaseMeta、SchemaMeta，基类为 base.py 中带 orjson 序列化的SQLModelSerializable）。

1. Modify your SQLModel classesindomain/models/；
2. Generate migration：

   alembic revision --autogenerate -m "add user table"

3. Review the generated fileinalembic/versions/；
4. Edit if needed- autogenerate may not catch everything：
   • Data migrations（数据迁移，autogenerate 只产生 DDL，不搬数据）
   • Index renames（索引重命名，autogenerate 常生成为“删旧建新”）
   • Complex constraint changes（复杂约束变更）
5. Committhe migration file to git。

实操补充（源自本模块源码约定）：

• 模型主键统一使用雪花 ID：id: int = Field(default_factory=get_id, sa_column=Column(BigInteger, primary_key=True))（见 database_meta.py L37-L39），因此新增列迁移时留意是否需要为历史数据补默认值——autogenerate 生成的server_default可配合env.py中的compare_server_default=True被正确比对；
• 新增字段尽量声明default=/Optional，并显式index=True或unique=True，与现有uid/name/space_id/schema_name的索引风格保持一致；
• 若同时支持 PostgreSQL 与 MySQL，修改upgrade()时沿用初始迁移里_get_schema()+ 条件索引名的写法，避免在两种方言间硬编码 schema 前缀。

七、验证与排错要点

结合本仓库实现，排查迁移问题可从以下入口入手：

1. 启动失败、报alembic.ini not found：run_database_migration()在找不到 ini 时抛FileNotFoundError，检查服务工作目录与core/memory/database/alembic/目录是否完整（见 database_migration.py L35-L40）。
2. 多实例重复迁移：确认 Redis 可用（setnx依赖 redis_cache.py 的SET NX EX）；锁键为memory_database_migration_lock，过期 60 秒。若迁移耗时可能超过 60 秒，注意当前实现未做续期（从源码结构看，长迁移理论上可能被第二实例重复执行，好在upgrade head对已应用的版本是幂等的）。
3. autogenerate 生成内容不符预期：优先检查env.py的include_object与模型导入是否完整；PostgreSQL 下表未出现在sparkdb_managerschema 时会被静默排除。
4. 回滚后服务异常：核对回滚是否遵循“先库后服务”的顺序（第五节三条注意事项），并确认目标版本与代码中模型定义兼容。

八、小结

这套迁移方案可以概括为三条设计主线：

• 自动化：服务启动即在 uvicorn 之前同步执行upgrade head，配合 RedisSETNX锁与“already exists → stamp 初始版本”的兜底，让新版本发布无需人工介入数据库；
• 多数据库：通过DatabaseAdapter策略接口把 URL 驱动、env 前缀、version 表位置、schema/include 规则收敛到两个适配器，env.py与script.py.mako保持稳定，切换DB_TYPE即可在同一套迁移脚本上运行；
• 可回滚、可审计：迁移文件带日期前缀入库，downgrade 顺序与数据风险在 README 中被显式约束，为后续多人协作维护版本链留下了明确的操作边界。

对于需要在 PostgreSQL 与 MySQL 之间双轨部署、且要求“发版即迁移、故障可回退”的中台服务，alembic.ini、env.py 与 database_migration.py 的组合实现提供了可直接参考的工程范式。

• 人工智能
• AI Agent
• Agent 工作流
• RPA
• 后端
• 前端
• 工具调用