拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从零搭建AI工程:模型训练到部署的完整闭环

从零搭建AI工程:模型训练到部署的完整闭环

AI工程这个词,听起来像“训练模型”,但真正上手后你会发现,模型训练只是整条链路里很小的一段。我从零搭建过一个文本分类项目,项目代号就叫 ai-engineering-from-scratch——不依赖任何现成的AI平台,从环境、数据、训练到部署全部自己搭。这篇文章是我对这个项目的完整复盘,包含我踩过的坑、重新调整过的方案、以及最后稳定上线的整个路径。适合刚入门想系统地做AI工程的同学,也适合已经在做模型但总觉得流程不规范的工程师。你会发现,from scratch难的不是写代码,而是把每一个看似“能用就行”的环节,变成可控、可复现的工程。

1. 先想清楚“从零”是从哪里零

1.1 从零不是“自己造轮子”,而是掌握完整闭环

很多人在“from scratch”上会走极端:要么觉得自己必须从反向传播开始写神经网络,要么觉得干脆用现成库堆一切。我的理解完全不同:从零的意思是,你要能完整搭建并掌控数据、训练、评估、部署、监控这几个环节,而不是依赖一个黑盒平台把数据丢进去看结果。自己搭的目的是当线上出问题时,你能顺着日志、参数、数据流一步步定位到具体是哪一环出了问题。这个能力是AI工程师和“调包侠”最核心的区别。

拿我这个项目来说,任务是给客服工单自动分到“登录”“支付”“退款”“物流”几个类别。听起来简单,但如果直接调用某个现成的分类API,价格不说,你没法解释为什么某条工单被分到错误类别,也没法针对自己的业务调优。而自己做from scratch,你可以控制文本清洗规则、控制特征、控制模型结构,最后还能把模型放进自己的服务里。

1.2 先定义“做完了”的标准,再动手写代码

AI工程最容易出现的问题是没有验收标准就开始训练。“准确率高”是很空洞的,你需要先把业务指标翻译成可计算的机器学习指标。比如对工单分类,我最终和运营确认的标准是:整体准确率不低于92%,其中“退款”这个高风险类别的召回率不低于95%,因为漏掉退款工单会造成用户投诉升级。同时,单条请求延迟不能超过200毫秒。这些指标先定下来,后面每次迭代才有对比依据。

所以from scratch的第一步不是import torch,而是写下一句“什么样的产出才算成功”。我会把指标记录在项目的README里,同时标记出哪些指标是硬性的、哪些是软性的。没有这个前提,后面做得再花哨也无法判断是否该上线。

1.3 目录结构:把代码、数据、配置、实验结果彻底分开

一个AI项目,代码往往只占很小一部分。大量文件是数据、记录、配置和模型权重。如果全塞在一个文件夹里,两周后你大概率找不到之前跑实验用的那份数据。我建议目录一开始就按功能分割,参考这个骨架:

ai-engineering-from-scratch/ ├── configs/ # YAML配置文件,存放所有可调参数 ├── data/ │ ├── raw/ # 原始数据,永远不改动 │ ├── processed/ # 清洗和特征处理之后的样本 │ └── splits/ # 训练集、验证集、测试集 ├── src/ │ ├── data/ # 数据加载、清洗、特征工程 │ ├── models/ # 模型定义、训练循环 │ ├── evaluate/ # 评估脚本与指标计算 │ └── serve/ # 部署相关代码 ├── experiments/ # 每个实验的输出、日志、权重 ├── notebooks/ # 探索性分析,不进入主流程 ├── scripts/ # 一次性脚本 ├── requirements.txt └── README.md

这个结构的好处是:你想找什么,不需要翻半天聊天记录,看目录名就能定位。尤其是experiments目录,每一个实验我都建一个带时间戳的子文件夹,里面放配置、日志和最佳权重,避免模型文件覆盖后找不到。

1.4 为什么不用AutoML平台:可维护性和Debug主动权

当时我对比过AutoML方案:把数据传上去,平台自动选模型、调参数。效果其实不错,但我还是选择自己搭。原因有两个。第一,业务方会不断调整规则,比如新加一个工单类别“账号安全”,AutoML平台再训练一轮周期很长,而自己的管道我可以在十分钟内改完新类别重新跑通。第二,线上如果出现误分类,平台的黑盒模型很难给出可操作的修复建议;而自己用Logistic回归或BERT变体,我能清楚知道是数据问题还是特征问题,手动调整特征权重也简单得多。

当然,这并不意味着所有项目都该“from scratch”。如果你只是快速验证某个想法、数据量极小、指标要求不敏感,用AutoML或现成API是合理的。但如果你要长期维护这个AI能力,自己掌控核心链路会省掉大量返工时间。

2. 从零搭建开发环境:用最稳的技术栈,别给未来埋雷

2.1 用conda把环境彻底隔离,锁死版本

AI项目的依赖是整个项目里最容易“爆雷”的部分。我见过太多同学把包装到全局环境,跑完A项目再跑B项目,版本冲突后只能重装系统。这里我强烈建议用conda为每个项目建独立环境,尤其注意Python版本和CUDA版本的匹配。我的做法是:

conda create -n ai-engineering python=3.10 -y conda activate ai-engineering # 先装PyTorch或TensorFlow,根据你的显卡选合适的CUDA版本 conda install pytorch torchvision pytorch-cuda=11.8 -c pytorch -c nvidia pip install mlflow fastapi uvicorn scikit-learn pandas numpy

这里有几个容易忽略的细节。第一,conda环境不等于安全,因为你依然可能通过pip把包装进全局site-packages,最好在项目根目录放一个.condarc或者用.env管理PYTHONNOUSERSITE=1,避免加载用户级包。第二,不要图省事直接pip install最新版本包,AI库之间经常有版本联动,比如某个版本的transformers可能不兼容某个版本的torch。这个坑我后面会单独说。

2.2 数据也要做版本管理

很多项目代码用Git管理得很好,但数据完全没版本。原始数据如果是手动下载或修改的csv,一旦被覆盖,实验结果就无法复现。我的方案是:原始数据统一放在data/raw/下,文件名带日期;任何清洗逻辑的输出都落在data/processed/,并且在Git LFS或者一个单独的data_versions.csv里记录每一次数据的hash值。这样做最大的价值就是,有人报告某个线上问题,你可以通过hash精确定位线上模型当时训练用的是哪一份数据。

也可以使用DVC这个工具来做数据版本控制。但我个人在实际项目中用更轻量的方案:每次跑数据管道之前,计算数据文件的MD5写到artifacts/manifest.json,和实验记录绑定。不需要额外工具,几行代码就够:

import hashlib def file_sha256(path): h = hashlib.sha256() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(8192), b''): h.update(chunk) return h.hexdigest()

2.3 配置即代码:用YAML管理所有超参数

不要做“凌晨改代码里的数值跑实验”这种操作。超参数一旦散落在代码里,实验对比就全乱了。我习惯把所有可调参数写进configs/config.yaml,然后在训练入口用一行代码加载。

data: train_path: data/splits/train.csv val_path: data/splits/val.csv test_path: data/splits/test.csv features: max_features: 5000 ngram_range: [1, 2] model: name: "logistic_regression" penalty: "l2" C: 1.0 training: seed: 42 max_epochs: 20 batch_size: 64 learning_rate: 3e-4 patience: 3 save_dir: experiments/run_001

这样做的好处非常明显:跑实验时只需要复制一份配置文件,然后修改你觉得要测的参数,实验记录文件里明确写着“我用的是哪个配置”。不用靠记忆,也经得起回溯。后期做超参数搜索时,可以直接循环修改这个config对象,而不是去改源码。

2.4 依赖锁定:不要用pip freeze裸奔

项目跑到第3个月,你会发现当初装了个什么包都记不清。pip freeze > requirements.txt虽然常见,但它会把项目无关的包也锁进来,而且不能真正固定依赖树。更好的办法是使用pip-tools或者conda-lock。

我的做法:维护一个requirements.in,只写顶层依赖,然后:

pip-compile requirements.in --output-file=requirements.txt pip install -r requirements.txt

这样锁出来的文件会把所有传递依赖钉死,且会标注每个包来自哪个顶层包。改依赖时只改requirements.in再重新compile,干净很多。配合conda环境,基本能把“换台机器跑不起来”的概率降到最低。

3. 数据管线:把脏数据变成可计算的样本

3.1 先探查数据,别急着做特征工程

拿到工单数据的第一个下午,我没有写一行训练代码,全在翻数据。这个环节不可跳过。我需要知道:每个类别的样本量是否均衡?文本是否是中文?有没有大量脏字段?重复工单占多少?日期分布是否受节假日影响?举一个实际发现:数据里约15%的工单是同一用户重复提交的“催促单”,比如“怎么还没退款”“退款怎么这么慢”,这些文本和原始工单高度重复,如果不做去重,模型会严重偏向“退款”类,而且线上同样用户反复提类似工单会导致预测结果分布扭曲。

数据探查我推荐用pandas-profiling或ydata-profiling直接生成报告,然后重点看类别分布和缺失值。看到不平衡需要提前想好对策:是过采样,还是降低阈值,还是换评估指标。这个决定会影响后面的所有实验。

3.2 清洗规则:能自动化的不要手搓

工单文本的清洗比较琐碎,我总结下来几条实用规则:统一小写或保留中英文原样,去掉URL和邮箱,把连续标点缩成一个,去除HTML标签,按业务词表做归一化(比如“忘了密码”和“密码记不住”都归到“登录问题”)。对中文文本,我一般用jieba分词,但不会盲目全量分词,会保留一些业务短语的完整形式,比如“退货退款”“验证码”“银行卡”等。

清洗代码尽量写成函数管道,每一步只做一件事,并且输出一份清洗报告,统计每一步删除了多少样本。比如URL移除影响了几条、空文本清掉了几条。这些统计很重要,因为业务方事后会问“为什么训练数据比原始数据少了一万条”,你总得答得清楚。

3.3 先做一道特征基线,别急着上深度学习

模型上,不一定一开始就选BERT。我在这个项目里打算先用TF-IDF加Logistic回归跑出一个baseline。原因很简单:它快、透明、能解释,且在很多文本分类场景下已经够用。特征工程这一步用的是TfidfVectorizer,参数先保守设置:

from sklearn.feature_extraction.text import TfidfVectorizer vectorizer = TfidfVectorizer( max_features=5000, ngram_range=(1, 2), sublinear_tf=True, min_df=2, max_df=0.95 ) X_train = vectorizer.fit_transform(train_texts)

为什么要设min_df和max_df?min_df=2表示至少出现在2个文档里的词才保留,过滤只出现一次的生僻词;max_df=0.95过滤几乎每个文档都出现的停用词比如“的”“是”“了”这类。这一步能用很低的成本砍掉大量无效特征。后面如果发现这个baseline已经接近指标要求,我甚至可以不换模型,只调C和ngram参数。

3.4 划分数据:分层抽样和时间切片都要做对

数据集划分是我最警惕的工程点,一旦分错,前面的所有努力都白搭。工单数据天然有先后顺序,如果随机划分,容易把同一个用户相近时间段的投诉同时放到训练集和验证集,形成隐性泄漏。稳妥的做法是:按时间排序,前80%作为训练集,后10%作为验证集,最后10%作为测试集。同时要保证每种类别在三个集合里的分布接近。可以先用StratifiedShuffleSplit做类别分层,对于时间型数据再检查时间分布。

下面是我实际使用的划分逻辑:

from sklearn.model_selection import train_test_split # 先按时间排序 df = df.sort_values("create_time") # 先留出测试集(最后10%) train_val, test = train_test_split(df, test_size=0.1, stratify=df["label"], random_state=42) # 再从剩余部分划分训练/验证 train, val = train_test_split(train_val, test_size=0.11, stratify=train_val["label"], random_state=42)

最重要的一点:random_state固定。如果每次划分随机性不一样,后续模型对比就无法分辨效果的差异是数据变化还是模型变化。另外,再把划分之后的文件保存到splits/目录,作为本次版本的基准。

3.5 数据校验:上线前发现脏数据是最伤的

在模型上线后,你还需要继续接新的工单推进预测。此时新数据的格式往往和训练时不一致,比如业务方有一天新增了一列“渠道来源”,或者把某个字段改成了英文。如果数据读取代码不够健壮,训练时跑得好好的,一接新数据就报错。

我的做法是在数据管道入口加一道校验,用pydantic或普通断言检查字段名、类型、缺失率、类别范围。校验不通过就中断管道并报警,而不是让脏数据流进模型生成奇怪的结果。下面是一个简单示例:

required_cols = {"id", "content", "label", "create_time"} if not required_cols.issubset(df.columns): raise ValueError(f"缺少字段: {required_cols - set(df.columns)}") if df["content"].isna().mean() > 0.01: raise ValueError(f"正文缺失率过高: {df['content'].isna().mean():.2%}")

4. 训练与实验管理:精度不是第一位的,可复现才是

4.1 先跑通baseline,别急着上BERT

很多项目最怕的不是效果差,而是第一个版本就跑出一个不可复现的“惊喜值”。所以我的原则是先用基模型把整条pipeline跑通。用Logistic回归在这个工单分类任务上,预计准确率能有85%左右,离目标92%还有距离,但至少我们知道数据管道没问题,评估代码没问题,然后才允许自己上更复杂的模型。

如果一上来就跑BERT,训练时间长是一方面,另一方面,出问题时你很难判断到底是数据清洗的锅还是预训练模型的锅。baseline就像给项目上了一道保险:它的效果就是地板,后面每次迭代都必须比它高,否则说明你加了复杂模块但没带来收益。

4.2 训练循环的工程细节:固定种子、早停、保存最优权重

如果后面确实决定用深度学习模型,训练循环不能是notebook里的几行代码,要形成可复用的模块。我强烈建议在训练开始前做三件事:固定随机种子、记录环境信息、设置早停。

固定随机种子的代码很简单,但很多人不知道PyTorch的随机源不止一个:

import random import numpy as np import torch def set_seed(seed: int): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

早停则是对资源最大的保护。我会在每个epoch结束后计算验证集损失,如果连续几轮没有变好就停止训练,并保存验证集效果最好的那一版权重。千万不要抱着“再多跑10轮肯定能涨”的心态,很多时候模型已经开始过拟合,继续训练只会让验证集指标波动变大。

4.3 用MLflow把每次实验记录成一条可对比的曲线

实验记录工具我推荐MLflow,它能自动记录参数、指标、模型文件,而且本地部署很简单。最轻量的用法是:

import mlflow with mlflow.start_run(run_name="logistic_regression_baseline"): mlflow.log_params(config["model"]) mlflow.log_metrics({"accuracy": acc, "recall_refund": recall_refund}) mlflow.log_artifact("configs/config.yaml") mlflow.log_artifact(best_model_path)

每跑一次实验,MLflow UI里就会多一条记录。你可以按准确率排序、按配置筛选,还能看到哪个实验对应哪个指标。这就解决了“我昨天那个99%的效果是用的什么参数”这种经典难题。我甚至会记录每个实验的训练耗时、GPU型号,方便后面排查复现问题。

4.4 超参调优:一次只动一个变量,守住随机性

超参调优是一个容易把实验变成玄学的环节。我见过有人同时改学习率、batch size、dropout和预处理方式,然后发现效果好了,却不知道是谁的功劳。我的纪律是:一次只改一个变量,其他全部固定。比如先用默认参数跑基线,然后只把C从0.1调到1,看指标变化;再只把ngram_range从(1,1)调到(1,2),看变化。

如果要做系统性搜索,用GridSearchCV或Optuna,但同样要在搜索过程里记录每一个中间结果。等搜索结束后,我还会用最优参数在3个不同随机种子下各跑一次,看稳定的提升是否真的稳定。这样得出的结论才是可信的,后续业务方问你“为什么线上比测试集高/低”你才能给出有理有据的回答。

5. 部署与上线:模型写成服务才算真正落地

5.1 导出模型,别再把训练脚本搬到生产环境

训练脚本里那一堆预处理、数据增强逻辑不适合直接铺到生产环境。生产环境只需要三样东西:预处理配置、模型权重、推理代码。我把这部分的输出整理成serve/模块,里面只有predict.py和一个加载模型文件的函数。训练用的任何类都不能直接import到服务里,除非你的代码结构一开始就做了清晰分层,否则很容易把几十个依赖一起拖进生产环境。

对于sklearn模型,可以直接用joblib.dump保存;对于PyTorch模型,优先保存state_dict,而不是整个模型对象:

torch.save(model.state_dict(), "best_model.pt") # 推理时 model.load_state_dict(torch.load("best_model.pt", map_location="cpu")) model.eval()

导出之后先做一个单元测试:加载模型,对一个手工样本调用predict,确认输出合理。这个测试能挡掉80%的“打包时少了一个模型文件”事故。

5.2 用FastAPI封装一个最小可用的推理服务

部署我推荐用FastAPI,它轻量、自带数据校验、文档页面开箱即用。一个最小的服务长这样:

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str confidence: float @app.post("/predict", response_model=PredictResponse) def predict(req: PredictRequest): label, confidence = model.predict(req.text) return PredictResponse(label=label, confidence=confidence) @app.get("/health") def health(): return {"status": "ok"}

一个容易被忽略的点:模型服务要一次性加载到内存,不要在每次请求里去加载权重文件。所以model = Model()要写在全局作用域——也就是在app初始化之后加载。用from_pretrained或load_model都行,但一定确保它只加载一次,否则请求一多内存直接顶穿。

5.3 容器化与上线之前的自检清单

上线前我习惯把服务打进Docker镜像里,避免“在我电脑上是好的”这种问题。Dockerfile不需要很复杂:

FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY serve/ ./serve/ COPY artifacts/ ./artifacts/ EXPOSE 8000 CMD ["uvicorn", "serve.main:app", "--host", "0.0.0.0", "--port", "8000"]

然后我还会跑一遍自检清单:镜像能正常启动吗?健康检查接口返回200吗?对一个真实请求预测一次,记录延迟是几十毫秒?单机并发20个请求时,内存占用是否正常?模型文件是否确实被打进镜像?这些检查花五分钟,但能避免上线后手忙脚乱。

5.4 线上监控:延迟、输入分布、预测置信度都不能少

模型上线不是终点,而是另一段维护旅程的起点。我至少会监控三个指标:请求延迟、输入文本长度、预测置信度。置信度这个指标特别有用,如果线上新来的工单很多都预测在0.5边缘摇摆,说明模型需要重新训练了。我还会定期抽样记录预测结果,存到一张表里,人工复核一次,用来看模型漂移。

一个简便的监控实现:在FastAPI内部写一个loguru日志,把每次请求的文本长度、预测类别、最高置信度打到结构化日志里,然后定时跑一个脚本统计当天的平均置信度和类别分布。这样即使没有专业监控平台,也能先顶住。

6. 我从这些坑里总结出的排查清单

6.1 训练时很好,验证/测试却崩:数据泄漏是第一嫌疑

表格里的排查路径,是我每次都走的:

表现优先排查常见原因
训练集准确率接近100%,测试集很差特征是否包含未来信息用了“订单号”“处理结果”这类字段,或者划分时没按时间切
验证集好但测试集差验证集分布和线上不一致验证集没有随机抽样,或者线上真实数据分布变了
两个batch之间指标跳变明显随机种子没固定或shuffle不规范多线程数据加载时随机状态没控制好

还有一个很经典的数据泄漏是:在划分数据集之前就对全量数据做了特征归一化或TF-IDF拟合。比如上面我写TF-IDF第一步在整份数据上fit,再切训练测试,这是不对的。正确做法是只用训练集fit,然后transform验证集和测试集。这一点很多初学者都会踩,我专门写在了代码注释里。

6.2 实验结果复现不了:环境一致性是根因

“我昨天跑的时候还有92%,今天变成90%了”这类问题,十有八九出在环境。可能昨天是PyTorch 1.13,今天不知道谁把环境升级到了2.0,反向传播的默认浮点精度变了,结果就不同。也可能是同一个随机种子,但GPU型号不一样,cudnn的算法选择不一样,最终结果也会有微小差异。

我现在的习惯是每个实验在MLflow里记录pip freeze、torch.__version__、platform.platform()。如果换机器跑,我会先用Docker锁定镜像。对精度要求特别高的场景,甚至可以把训练时的随机数序列记录下来。当然,这通常没必要,记录环境基本信息就够了。

6.3 服务上线后效果越来越差:数据漂移是常态

模型上线一个月后,运营反馈分得越来越不准。我查了一下,发现新工单里出现了一个旧训练数据里几乎没有的类别“账号异地登录”,而这个类别之前被归到“登录”类,导致“登录”类的置信度整体被拉低。这就是业务演化和数据漂移。

我的解决办法分两层:第一,在服务端记录输入特征的哈希或统计量,每天对比训练集分布,比如计算平均文本长度、类别预测比例,超过阈值就报警;第二,保持每周用最新标注数据做增量微调的小流程,让模型跟上业务变化。不要以为一次训练一劳永逸,AI工程里维护模型的比例远大于训练模型。

6.4 显存或内存不够时的几条活路

如果训练时爆显存,我的第一反应不是撸起袖子换大卡,而是先确认有没有无意义的浪费。比如batch_size是否过大、序列长度是否做了padding,数据加载时有几个worker在并行读取、每个worker是否会复制一份模型。最简单有效的招是:减小batch_size,配合梯度累积;将模型输入统一截断到合理长度,不要用全文本长度做max_length;推理时用半精度fp16,对显存占用降低非常明显。

如果内存不足,优先排查是不是在循环里保存了历史数据或历史梯度。有些新手会在每轮迭代把loss列表堆到list里,训练完再画曲线,轮数多的时候list很大。正确做法是每个epoch只保存一个均值,或者用deque限制长度。这些小优化往往比换机器更快见效。

最后分享一点个人体会。从零搭建AI工程,最大收获不是模型本身,而是对每个环节都有了掌控力。后来业务方提新需求、线上出问题,我都能很快定位到待优化的环节。这中间踩过很多坑,尤其是数据和环境相关的问题。如果你也在从零起步,希望这份拆解能让你少走一些弯路。有一点我反复强调:任何一个环节,只要做到“可复现、可监控、可说清楚为什么”,你就已经比大多数AI项目靠谱了。

返回列表