从零构建Model Eon：轻量级模型版本管理系统的实践指南

在实际机器学习项目中，模型文件的管理往往比训练本身更容易埋下隐患。训练好的模型散落在各台机器的磁盘目录里，文件名可能是model_v2_final_v3_really_final.pkl，实验记录写在聊天记录或本地 Excel 里，线上服务用的模型版本靠人工确认。Model Eon 就是为了解决这类问题而设计的一套轻量模型版本管理系统：它把模型文件、元数据、版本关系、阶段状态集中管理起来，让训练、评估、上线、回滚每一步都有据可查。这篇文章会从零构建一个最小可运行的 Model Eon 服务，涵盖核心概念、数据表设计、接口实现、验证流程以及生产环境注意事项，适合正在做 MLOps 建设、或想把实验管理规范化的算法工程师和平台开发工程师。

1. Model Eon 要解决什么问题：模型版本失控是 ML 项目的隐形技术债

1.1 训练出的模型文件为什么需要“仓库化”管理

先看一个常见场景。算法工程师本地跑通一个模型，把model.pkl发给后端同学部署，后端同学把它放到生产目录，结果第二天算法又更新了权重，文件名还叫model.pkl，覆盖之后线上效果波动，想回滚发现上一版文件已经没了。

这类问题的本质不是“文件被覆盖”这么简单，而是缺少一个模型注册中心来回答三个问题：

• 当前有哪些模型，每个模型有哪些版本？
• 每个版本对应什么框架、什么训练数据、多少指标？
• 哪个版本处于 Staging（预发）状态，哪个版本已经 Production（生产）状态？

Model Eon 的设计目标就是把这三种信息统一管理起来。它不替代训练框架，也不替代模型推理服务，而是作为训练与部署之间的中性地带，负责登记、存储、校验、流转模型资产。

1.2 模型注册表与普通文件存储的关键差异

普通文件存储只保存二进制内容，模型注册表在二进制之上增加了一层“元数据约束”。两者差异如下表所示：

能力维度	普通文件存储	Model Eon 模型注册表
文件内容	只保存文件	保存文件副本，并提供哈希校验
版本维度	通过文件名区分	通过 registry 中的 version 字段管理
指标关联	需要外部记录	指标、数据集、训练参数统一落库
阶段状态	无	支持 Staging / Production / Archived 流转
回滚操作	手动找旧文件	按版本号一键切换状态

模型注册表的价值不在于“多存了一份文件”，而在于把“这份模型为什么可信”这件事变成可查询的记录。在负责事故定级、模型审计和多人协作的环境里，这一步是刚需。

2. 整体设计：Model Eon 的模块划分和数据模型

2.1 核心模块划分

为了让文章后面的代码有落点，Model Eon 采用 Python 技术栈实现，包含四个模块：

• registry：模型注册表 API，负责处理模型注册、版本查询、阶段流转。
• storage：模型文件存储层，负责将权重文件持久化到磁盘目录，并计算 SHA256 指纹。
• metadata：元数据管理，将模型描述、指标、数据集、框架信息写入数据库。
• cli：命令行客户端，方便训练脚本在结束训练后直接通过命令注册模型。

这种模块划分参考了成熟模型仓库的思路，但实现上做了尽量简化，方便本地跑通。

2.2 数据表设计

Model Eon 的最小数据模型只需要三张表：

• models：模型基础信息，包括模型名、任务类型、创建时间。
• model_versions：模型的版本信息，包括版本号、框架、模型文件路径、SHA256、指标 JSON、阶段状态。
• model_events：阶段流转记录，记录谁在什么时间把模型从哪个阶段切到了哪个阶段。

建表 SQL 如下：

CREATE TABLE IF NOT EXISTS models ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, task_type TEXT NOT NULL DEFAULT 'classification', description TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS model_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, model_id INTEGER NOT NULL REFERENCES models(id), version INTEGER NOT NULL, framework TEXT NOT NULL, artifact_path TEXT NOT NULL, sha256 TEXT NOT NULL, metrics TEXT NOT NULL DEFAULT '{}', stage TEXT NOT NULL DEFAULT 'None', created_at TEXT NOT NULL, UNIQUE(model_id, version) ); CREATE TABLE IF NOT EXISTS model_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, model_id INTEGER NOT NULL REFERENCES models(id), version INTEGER NOT NULL, from_stage TEXT DEFAULT '', to_stage TEXT NOT NULL, operator TEXT NOT NULL DEFAULT 'unknown', created_at TEXT NOT NULL );

这里有一个容易被忽视的设计点：model_versions表里加上了sha256字段。模型文件内容一旦被篡改或者传输损坏，哈希校验能第一时间发现，避免加载一个不完整的权重文件导致推理结果异常。

3. 环境准备与项目初始化

3.1 开发环境要求

实现 Model Eon 不需要重型基础设施，本地开发建议按下面清单准备：

依赖项	推荐版本或说明
Python	3.10 及以上
FastAPI	0.104 及以上
Uvicorn	0.23 及以上
SQLAlchemy	2.0 及以上
Pydantic	2.x 版本
数据库	开发环境使用 SQLite，生产建议 PostgreSQL

如果原始项目没有指定版本，落地前要确认依赖版本是否匹配，尤其是 SQLAlchemy 2.x 与旧版 1.x 的会话写法差异较大。

3.2 初始化项目结构

建议按下面的目录组织项目：

model_eon/ ├── app/ │ ├── __init__.py │ ├── database.py # 数据库连接和会话 │ ├── models.py # ORM 模型 │ ├── schemas.py # Pydantic 请求/响应模型 │ ├── storage.py # 模型文件存储与哈希校验 │ └── main.py # FastAPI 路由 ├── storage/ # 模型文件存储目录 ├── cli.py # 命令行工具 ├── requirements.txt └── README.md

storage/目录在启动服务前必须存在，否则上传模型文件时会因为目录不存在而报错。后续可以改成配置项，由启动脚本自动创建。

3.3 安装依赖

创建虚拟环境并安装依赖：

python -m venv .venv source .venv/bin/activate # Windows 下为 .venv\Scripts\activate pip install fastapi "uvicorn[standard]" sqlalchemy pydantic

安装完成后，先运行python -c "import fastapi; print(fastapi.__version__)"确认版本。如果出现 Pydantic 版本冲突，优先检查项目中是否同时安装了 pydantic 1.x 和 2.x。

4. 核心实现：注册、查询、阶段流转

4.1 数据库连接与会话管理

database.py负责创建 SQLAlchemy 引擎和会话。开发环境使用 SQLite，连接参数相对简单：

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL = "sqlite:///./model_eon.db" engine = create_engine( DATABASE_URL, connect_args={"check_same_thread": False} ) SessionLocal = sessionmaker(bind=engine, autoflush=False) Base = declarative_base() def get_db(): db = SessionLocal() try: yield db finally: db.close()

关键点在于check_same_thread: False。SQLite 默认不允许跨线程使用同一个连接，而 FastAPI 的异步模型可能会在不同线程中访问数据库，不加这个配置容易出现SQLite objects created in a thread can only be used in that same thread的报错。

ORM 模型对应前面设计的三张表，核心是ModelVersion：

class ModelVersion(Base): __tablename__ = "model_versions" id = Column(Integer, primary_key=True, index=True) model_id = Column(Integer, ForeignKey("models.id"), nullable=False) version = Column(Integer, nullable=False) framework = Column(String, nullable=False) artifact_path = Column(String, nullable=False) sha256 = Column(String, nullable=False) metrics = Column(Text, nullable=False, default="{}") stage = Column(String, nullable=False, default="None") created_at = Column(String, nullable=False)

该表通过唯一约束(model_id, version)保证一个模型下版本号不会重复。新增版本前一定要查询当前最大版本号，并在事务内完成插入，否则并发训练任务可能注册出相同版本号。

4.2 模型注册接口

模型注册分两步：持久化模型文件，再写入元数据。storage.py中实现文件保存和哈希计算：

import hashlib import os import shutil UPLOAD_DIR = "./storage" def save_artifact(file_stream, model_name: str, version: int) -> str: os.makedirs(UPLOAD_DIR, exist_ok=True) safe_name = model_name.replace("/", "_").replace("..", "_") relative_path = os.path.join(safe_name, f"v{version}.model") full_path = os.path.join(UPLOAD_DIR, relative_path) os.makedirs(os.path.dirname(full_path), exist_ok=True) sha256 = hashlib.sha256() with open(full_path, "wb") as f: while True: chunk = file_stream.read(1024 * 1024) if not chunk: break sha256.update(chunk) f.write(chunk) return relative_path, sha256.hexdigest()

注意不要在模型名里直接拼接路径，防止model_name传入../../xxx导致目录穿越。示例里使用replace做简单过滤，生产环境建议用正则白名单校验，只允许字母、数字、下划线和短横线。

注册接口在main.py中实现：

import json from datetime import datetime, timezone from fastapi import FastAPI, Depends, HTTPException, UploadFile, File, Form from sqlalchemy.orm import Session from app import models as orm_models from app.database import Base, engine, get_db from app import storage app = FastAPI(title="Model Eon") Base.metadata.create_all(bind=engine) def now_str(): return datetime.now(timezone.utc).isoformat() @app.post("/models/{model_name}/versions") async def register_version( model_name: str, framework: str = Form(...), metrics: str = Form("{}"), file: UploadFile = File(...), db: Session = Depends(get_db) ): model = db.query(orm_models.Model).filter( orm_models.Model.name == model_name ).first() if model is None: model = orm_models.Model( name=model_name, task_type="classification", created_at=now_str() ) db.add(model) db.commit() db.refresh(model) max_version = db.query( db.func.max(orm_models.ModelVersion.version) ).filter( orm_models.ModelVersion.model_id == model.id ).scalar() or 0 new_version = max_version + 1 relative_path, sha256 = storage.save_artifact(file.file, model_name, new_version) db_version = orm_models.ModelVersion( model_id=model.id, version=new_version, framework=framework, artifact_path=relative_path, sha256=sha256, metrics=metrics, stage="None", created_at=now_str() ) db.add(db_version) db.commit() db.refresh(db_version) return { "model_name": model_name, "version": new_version, "sha256": sha256, "stage": db_version.stage }

第一次调用接口时，如果模型不存在会自动创建模型记录，这个行为方便训练脚本直接上报模型，不用预先在系统里建模型。

4.3 阶段流转接口

阶段流转是模型注册表里最重要的操作。所谓阶段，是指模型当前处于哪个生命周期：

• None：刚注册，未进入任何流程。
• Staging：进入预发验证阶段，可以跑离线测评、影子流量。
• Production：已经在生产环境提供服务。
• Archived：已下线归档，不再提供新流量。

实现阶段流转时要做状态校验，禁止从Archived直接切到Production，也禁止跳过Staging直接上线。

ALLOWED_STAGES = {"None", "Staging", "Production", "Archived"} TRANSITION_RULES = { "None": {"Staging"}, "Staging": {"Production", "Archived", "None"}, "Production": {"Archived", "None"}, "Archived": {"None"} } @app.post("/models/{model_name}/versions/{version}/transition") def transition_model( model_name: str, version: int, to_stage: str, operator: str = "unknown", db: Session = Depends(get_db) ): if to_stage not in ALLOWED_STAGES: raise HTTPException(status_code=400, detail=f"Unknown stage: {to_stage}") model = db.query(orm_models.Model).filter( orm_models.Model.name == model_name ).first() if model is None: raise HTTPException(status_code=404, detail="Model not found") db_version = db.query(orm_models.ModelVersion).filter( orm_models.ModelVersion.model_id == model.id, orm_models.ModelVersion.version == version ).first() if db_version is None: raise HTTPException(status_code=404, detail="Version not found") if to_stage not in TRANSITION_RULES.get(db_version.stage, set()): raise HTTPException( status_code=400, detail=f"Cannot transition from {db_version.stage} to {to_stage}" ) old_stage = db_version.stage db_version.stage = to_stage db.add(orm_models.ModelEvent( model_id=model.id, version=version, from_stage=old_stage, to_stage=to_stage, operator=operator, created_at=now_str() )) db.commit() return { "model_name": model_name, "version": version, "from_stage": old_stage, "to_stage": to_stage }

model_events表在这里不只是日志，而是审计记录。每次状态变更都保留操作人和前后状态，后面排查“谁把模型切到了 Production”时直接查这张表即可。

5. 运行验证与结果分析

5.1 启动服务

在项目根目录执行：

uvicorn app.main:app --host 0.0.0.0 --port 8000

启动后访问http://127.0.0.1:8000/docs可以看到 FastAPI 自动生成的接口文档。这里先不要急着点操作，先准备一个测试用的模型文件，任何二进制文件都可以，比如：

dd if=/dev/urandom of=test_model.bin bs=1024 count=512

5.2 通过命令行验证注册流程

用 curl 注册一个名为fraud_detection的模型：

curl -X POST "http://127.0.0.1:8000/models/fraud_detection/versions" \ -F "framework=pytorch" \ -F "metrics={\"auc\": 0.92, \"accuracy\": 0.97}" \ -F "file=@test_model.bin"

预期响应类似：

{ "model_name": "fraud_detection", "version": 1, "sha256": "d3b07384d3a3c5f1c1f1c1f1c1f1c1f1c1f1c1f1c1f1c1f1c1f1c1f1c1f1c1f", "stage": "None" }

再注册一次，版本号应自动变成 2。这个自动化递增过程验证的是max_version + 1逻辑。

5.3 验证阶段流转与查询

执行阶段流转：

curl -X POST "http://127.0.0.1:8000/models/fraud_detection/versions/1/transition" \ -H "Content-Type: application/json" \ -d '{"to_stage": "Staging", "operator": "alice"}'

返回结果为：

{ "model_name": "fraud_detection", "version": 1, "from_stage": "None", "to_stage": "Staging" }

此时再尝试从None直接切到Production，会看到接口拒绝操作。这正是预期行为：模型必须经过 Staging 才能进入生产，拦截了因误操作直接上线的风险。

验证完成后，检查目录结构：

storage/ └── fraud_detection/ ├── v1.model └── v2.model

再查数据库里的model_events表，确认阶段流转记录已写入。这一步一定要做，因为很多人验证接口只看了返回码，没有核对审计表数据是否完整。

6. 常见问题排查

6.1 典型问题对照表

把开发 Model Eon 过程中最容易遇到的问题整理成如下表格：

问题现象	常见原因	检查方式	处理建议
上传文件后接口返回 500	storage 目录不存在或权限不足	检查目录和日志堆栈	启动前创建目录，或用配置项自动创建
版本号突然跳到 3 而不是从 1 开始	删过旧版本记录，但文件残留	查询 model_versions 表最大版本号	注册前清理旧数据，文件与元数据保持一致
阶段流转接口提示 Cannot transition	当前状态不在规则表中	查询当前 stage	按规则先切换到 Staging 再上线
切换环境后数据库文件连接失败	SQLite 路径使用了相对路径	检查启动目录	生产环境改为绝对路径或 PostgreSQL
两个训练任务同时注册出相同版本号	缺少事务锁或唯一约束	查看 model_versions 唯一约束	依赖数据库唯一索引并捕获冲突异常

6.2 排查链路

如果注册接口报错，建议按下面顺序排查：

1. 先看 FastAPI 返回的 HTTP 状态码和 detail 文本。
2. 再看 Uvicorn 控制台日志，找到异常堆栈的根因行。
3. 用sqlite3 model_eon.db或 GUI 工具查看 model_versions 表结构是否未变化。
4. 检查上传的文件是否真的写入 storage 目录，文件大小是否与上传前一致。
5. 检查 sha256 是否匹配。可以在处理逻辑里单独打印哈希值，与本地sha256sum test_model.bin对比。

值得特别提醒的是，SQLite 开发库与 PostgreSQL 生产库的行为并不完全一致。SQLite 对并发写入支持较弱，如果多个训练任务同时在同一个模型下注册版本，很容易出现database is locked错误。生产环境建议切换到 PostgreSQL，并为model_versions(model_id, version)增加唯一约束，在代码层捕获IntegrityError后重试或返回明确错误。

7. 最佳实践和生产环境建议

7.1 上线前检查清单

Model Eon 从本地 demo 走向团队使用时，建议逐项核对：

• [ ] 模型文件名使用白名单规则校验，避免路径穿越
• [ ] 模型文件上传使用对象存储或共享存储，不使用本地磁盘
• [ ] 数据库从 SQLite 切换为 PostgreSQL，并配置连接池
• [ ] 接口增加认证和权限控制，阶段流转操作需要权限审计
• [ ] 指标字段使用结构化 JSON Schema 校验，防止格式错误
• [ ] 定期清理 Archived 状态的历史文件，保留元数据
• [ ] 上传过程增加大小限制和格式限制
• [ ] 阶段流转增加人工审批或二次确认机制

这个清单可以直接作为发布前的评审依据，不用再临时从文档里翻找。

7.2 从最小系统到完整 MLOps 的扩展方向

Model Eon 目前是一个最小可运行版本。实际工程里可以在它之上扩展三个能力：

第一，与训练框架集成。PyTorch 训练脚本结束前调用 Model Eon 的 CLI 工具注册模型，把torch.save的产物直接上报，训练、注册、验证形成一条完整链路。

第二，与推理服务联动。推理服务启动时从 Model Eon 拉取指定模型的 Production 版本，而不是从本地路径读取。这样模型升级就是一个“改注册状态”的操作，而不是重启容器、改配置。

第三，增加模型血缘记录。在元数据中加入训练代码版本、数据集版本、超参数配置的引用，让每一次模型产出都能回溯到对应的实验。这一步对模型审计和合规要求严格的业务尤其重要。

对于刚开始做模型管理的团队，建议不要一上来就追求完整 MLOps 平台。先跑通“注册、查询、阶段流转、审计”这四个核心动作，把模型版本信息从聊天记录和人脑记忆中迁到系统里，再逐步扩展自动化和发布流程。模型管理的价值不在于系统多复杂，而在于每一次模型上线和回滚都能被准确、高效地完成。