从零搭建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 python3.10 -y conda activate ai-engineering # 先装PyTorch或TensorFlow根据你的显卡选合适的CUDA版本 conda install pytorch torchvision pytorch-cuda11.8 -c pytorch -c nvidia pip install mlflow fastapi uvicorn scikit-learn pandas numpy这里有几个容易忽略的细节。第一conda环境不等于安全因为你依然可能通过pip把包装进全局site-packages最好在项目根目录放一个.condarc或者用.env管理PYTHONNOUSERSITE1避免加载用户级包。第二不要图省事直接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-filerequirements.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_features5000, ngram_range(1, 2), sublinear_tfTrue, min_df2, max_df0.95 ) X_train vectorizer.fit_transform(train_texts)为什么要设min_df和max_dfmin_df2表示至少出现在2个文档里的词才保留过滤只出现一次的生僻词max_df0.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_size0.1, stratifydf[label], random_state42) # 再从剩余部分划分训练/验证 train, val train_test_split(train_val, test_size0.11, stratifytrain_val[label], random_state42)最重要的一点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_namelogistic_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_locationcpu)) 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_modelPredictResponse) def predict(req: PredictRequest): label, confidence model.predict(req.text) return PredictResponse(labellabel, confidenceconfidence) 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项目靠谱了。