1. 为什么我要从零手搓一套AI工程化流程第一次看到ai-engineering-from-scratch这个项目名的时候我正被一堆散落在各处的实验脚本折磨得够呛。Jupyter Notebook 里躺着十几个版本的模型训练代码文件名从train_final.py一路排到train_final_v3_really_final.py每次想复现两周前跑出来的那个还不错的结果都得花半天时间回忆当时到底改了哪个参数。这种状态持续了大概三个月直到我下定决心把整个AI工程化的链路从头到尾梳理一遍用最朴素的方式重新搭一套能跑通、能复现、能交接的流程。这个项目标题里的 from scratch 其实有两层意思。一层是字面上的从零开始不依赖那些开箱即用的大平台自己动手把数据、训练、评估、部署这几块拼起来另一层是认知上的从零开始把那些平时被框架封装掉的细节重新摊开来看清楚搞明白每一步到底在干什么。我见过太多人包括我自己调包调得很熟练但一旦遇到框架不支持的场景就抓瞎根本原因就是中间那层黑盒从来没打开过。这套流程适合谁呢如果你是一个刚入行不久、想搞清楚AI项目从数据到上线完整链路的工程师或者是一个在中小团队里需要一个人扛起整个模型迭代流程的开发者再或者你只是单纯好奇那些大厂里的MLOps到底在做什么那这篇内容应该能给你一些可以直接抄作业的东西。我不打算讲太玄乎的理论重点放在每个环节为什么这么设计、实际踩过哪些坑、以及怎么用最少的工具把事办成。整个流程我大概花了六周时间打磨中间推翻重来过两次最终稳定下来的版本支撑了我后续半年的模型迭代工作。下面我把这套东西拆开来讲从整体设计思路到每个模块的具体实现再到实际运行中遇到的各种幺蛾子尽量讲透。2. 整体架构设计与技术选型思路2.1 为什么选择轻量级组合而不是一体化平台市面上做AI工程化的方案大致分两派。一派是重量级的一体化平台功能大而全从数据标注到模型监控全给你包圆了另一派是轻量级的工具组合每个环节用最合适的工具通过约定和脚本来串联。我最终选了后者原因很实际我手上的项目规模不大数据量在几十万条这个级别模型也是中等体量的微调任务用一体化平台属于杀鸡用牛刀光是平台本身的运维成本就够我喝一壶的。轻量级组合的另一个好处是可控。每个环节的输入输出都是明确定义的文件或接口出了问题能快速定位到具体是哪个环节挂了。一体化平台虽然省事但一旦出问题排查起来往往要翻平台自己的日志有时候还得等社区回复节奏完全不在自己手里。我印象特别深的一次某个平台的训练任务莫名其妙卡住查了两天最后发现是平台内部的一个资源调度bug这种问题在自建流程里根本不会出现。当然轻量级组合也有代价就是需要自己写不少胶水代码。但这些胶水代码写一次就能复用很久而且写的过程中你会对整个流程的理解越来越深这笔账算下来是划算的。2.2 核心模块的划分与职责边界我把整个流程拆成了五个核心模块每个模块只干一件事模块之间通过文件系统来传递数据。这种设计看起来有点土但实际用下来非常稳。模块名称核心职责输入输出数据准备数据清洗、切分、版本化原始数据文件标准化数据集训练引擎模型训练、超参管理数据集、配置文件模型权重、训练日志评估模块指标计算、结果对比模型权重、测试集评估报告实验追踪记录每次实验的完整信息各模块的输出实验数据库部署服务模型加载、推理接口模型权重HTTP接口这个划分的关键在于职责边界要清晰。比如数据准备模块只负责把原始数据变成模型能吃的格式它不关心模型是什么训练引擎只负责根据配置训练它不关心数据是怎么来的。这种解耦带来的好处是我想换一个模型架构只需要改训练引擎和配置文件数据准备那块完全不用动。2.3 目录结构的设计与约定目录结构这个东西看起来不起眼但设计不好后期会非常痛苦。我最终采用的方案是按功能划分顶层目录每个目录内部再按时间或版本分子目录。project/ ├── data/ │ ├── raw/ # 原始数据只读 │ ├── processed/ # 清洗后的数据 │ └── splits/ # 训练/验证/测试切分 ├── configs/ │ ├── base.yaml # 基础配置 │ └── experiments/ # 各实验的覆盖配置 ├── src/ │ ├── data/ # 数据处理代码 │ ├── train/ # 训练代码 │ ├── eval/ # 评估代码 │ └── serve/ # 服务代码 ├── experiments/ # 实验输出 │ └── exp_20240101_001/ │ ├── config.yaml # 本次实验的完整配置 │ ├── checkpoints/ # 模型权重 │ ├── logs/ # 训练日志 │ └── metrics.json # 评估指标 └── scripts/ # 各种入口脚本这个结构里有个关键约定experiments 目录下的每个子目录都是一次完整的实验记录包含复现这次实验所需的一切信息。只要拿到这个目录就能知道当时用了什么配置、什么数据、跑出了什么结果。这个约定后来救了我很多次尤其是当老板问上个月那个效果不错的版本是怎么跑出来的的时候我直接翻出对应的目录就行。2.4 配置管理的取舍配置管理这块我纠结了很久试过三种方案。第一种是纯命令行参数简单直接但参数一多就记不住第二种是Python字典写在代码里灵活但容易和代码耦合太深第三种是YAML配置文件加命令行覆盖最终选了这个。YAML的好处是可读性好非技术人员也能看懂大概。我采用的模式是有一个base.yaml存放所有默认配置然后每个实验有一个小的覆盖文件只写和默认值不一样的部分。运行时把两个文件合并生成最终的配置。这样既避免了重复又能清楚地看到每次实验改了什么。# base.yaml 片段 model: name: bert-base max_length: 128 dropout: 0.1 training: batch_size: 32 learning_rate: 2e-5 epochs: 3 warmup_ratio: 0.1 data: train_file: data/splits/train.jsonl valid_file: data/splits/valid.jsonl# experiments/exp_001.yaml 覆盖配置 training: learning_rate: 1e-5 epochs: 5合并逻辑我用了一个简单的递归字典合并函数大概二十行代码比引入第三方库更可控。这里有个细节要注意列表类型的配置不要做合并直接覆盖。我一开始想当然地做了列表合并结果有次改数据路径的时候新旧路径被合并成了一个列表训练脚本读到一个不存在的路径直接崩了排查了半天才发现是合并逻辑的问题。3. 数据准备环节的核心细节与实操3.1 数据清洗的标准化流程数据清洗这块看起来简单实际上是最容易埋雷的地方。我总结了一套标准化的流程每次拿到新数据都按这个流程走一遍。第一步是格式统一。不管原始数据是CSV、JSON还是数据库导出的先统一转成JSONL格式每行一个样本。JSONL的好处是流式读取方便不会因为文件太大把内存撑爆而且出错了能快速定位到具体哪一行。第二步是字段规范化。我遇到过原始数据里同一个含义的字段有五六种命名的情况比如文本这个字段有的叫text有的叫content有的叫sentence。这时候要统一成一个标准字段名我一般用text和label这两个最通用的名字。第三步是异常样本过滤。常见的异常包括空文本、超长文本、标签缺失、编码错误等。这里有个经验过滤规则要记录日志每次过滤掉了多少条、分别是什么原因都要写清楚。我有次发现清洗后数据少了一大半查日志才发现是编码检测太严格把一堆正常的中文文本误判成了乱码。def clean_sample(sample): 单条样本的清洗逻辑 # 去除首尾空白 text sample.get(text, ).strip() # 空文本过滤 if not text: return None, empty_text # 超长文本截断不是过滤是截断 if len(text) 10000: text text[:10000] # 标签检查 label sample.get(label) if label is None: return None, missing_label return {text: text, label: label}, None3.2 数据切分的坑与正确姿势数据切分看起来就是随机分三份但实际操作里有几个坑我踩过。第一个坑是切分前没有打乱。如果原始数据是按类别排序的直接按比例切分会导致训练集和验证集的类别分布差异巨大。我一开始就犯过这个错训练集里正样本占80%验证集里正样本只占20%结果验证指标一直上不去还以为是模型有问题。第二个坑是数据泄漏。有些数据集里同一个样本会以不同形式出现多次如果随机切分同一个样本的变体可能同时出现在训练集和验证集里导致验证指标虚高。解决办法是在切分前先做去重或者按样本的来源ID来切分保证同一个来源的样本只出现在一个集合里。第三个坑是切分比例固定不变。小数据集上这个问题特别明显不同的随机种子切出来的验证集指标能差好几个点。我的做法是对于小数据集做多次切分取平均或者用交叉验证。数据量大的话就固定一个种子保证可复现。import random from collections import defaultdict def stratified_split(samples, ratios(0.8, 0.1, 0.1), seed42): 按标签分层切分保证各集合类别分布一致 random.seed(seed) # 按标签分组 by_label defaultdict(list) for s in samples: by_label[s[label]].append(s) train, valid, test [], [], [] for label, items in by_label.items(): random.shuffle(items) n len(items) n_train int(n * ratios[0]) n_valid int(n * ratios[1]) train.extend(items[:n_train]) valid.extend(items[n_train:n_train n_valid]) test.extend(items[n_train n_valid:]) # 最后再打乱一次 random.shuffle(train) random.shuffle(valid) random.shuffle(test) return train, valid, test3.3 数据版本化的轻量方案数据版本化这块我用了一个非常土但有效的方案每次数据处理生成一个版本号版本号由处理脚本的哈希和输入数据的哈希组合而成。处理脚本变了或者输入数据变了版本号就会变这样就能保证同样的版本号对应的数据是完全一样的。具体实现上我在数据目录下放一个manifest.json记录每个版本的数据来源、处理脚本的哈希、样本数量、生成时间等信息。训练的时候配置里指定数据版本号训练脚本会校验实际数据的版本号是否匹配不匹配就直接报错。这个校验机制帮我避免了好几次用错数据训练的事故。提示数据版本号不要用时间戳时间戳虽然唯一但不可复现。用内容哈希虽然计算稍慢但能保证相同内容得到相同版本号这对复现实验至关重要。3.4 实操心得数据质量检查清单在数据进入训练之前我一定会跑一遍质量检查清单如下样本总数是否和预期一致差异超过5%就要查原因各标签的样本数量分布是否存在严重不平衡文本长度的分布P99长度是多少决定模型的最大长度设置是否有重复样本重复率是多少随机抽20条人工看一眼确认格式和内容正常这个清单看起来简单但每次都能发现点问题。有次抽检发现一批数据的标签全是0查下来是上游导出的时候字段映射错了如果没检查直接训练模型会学成一个只会输出0的废物。4. 训练引擎的搭建与关键配置4.1 训练脚本的骨架设计训练脚本我改过很多版最终稳定下来的骨架包含这几个部分配置加载、数据加载、模型初始化、优化器和调度器设置、训练循环、验证循环、检查点保存。每个部分都独立成函数主函数只负责串联。这种设计的好处是每个部分都能单独测试。比如我想验证数据加载是否正确直接调用数据加载函数打印几条样本就行不用跑整个训练。模型初始化也是可以先初始化完打印一下参数量确认没问题再开始训练。def train(config): # 1. 设置随机种子 set_seed(config[seed]) # 2. 加载数据 train_loader build_dataloader(config, splittrain) valid_loader build_dataloader(config, splitvalid) # 3. 初始化模型 model build_model(config) model.to(config[device]) # 4. 优化器和调度器 optimizer build_optimizer(model, config) scheduler build_scheduler(optimizer, config, len(train_loader)) # 5. 训练循环 best_metric 0 for epoch in range(config[training][epochs]): train_one_epoch(model, train_loader, optimizer, scheduler, epoch) metrics evaluate(model, valid_loader) # 6. 保存最佳模型 if metrics[f1] best_metric: best_metric metrics[f1] save_checkpoint(model, config, epoch, metrics) log_metrics(epoch, metrics)4.2 随机种子与可复现性可复现性这块我踩的坑最多。理论上设置一个随机种子就完事了但实际上影响结果的因素远不止一个种子。首先是框架层面的随机性。PyTorch里除了torch.manual_seed还有CUDA的随机性、cuDNN的算法选择等。要完全固定需要设置好几个地方import torch import numpy as np import random def set_seed(seed): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) # 关闭cuDNN的自动算法选择 torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False其次是数据加载的随机性。DataLoader的shuffle、worker的初始化都有随机性需要给worker设置种子。还有就是如果用了数据增强增强本身的随机性也要控制。最后是浮点运算的非确定性。即使设置了所有种子某些CUDA操作的结果仍然可能有微小差异这是硬件层面的问题没法完全消除。我的做法是接受这个现实在评估的时候用足够的样本量让这点微小差异不影响最终判断。注意cudnn.deterministic True会降低训练速度大概慢10%到20%。如果对速度敏感可以只在需要精确复现的时候开启日常实验可以关掉。4.3 超参数配置的经验值超参数这块没有万能公式但有一些经验值可以参考。我整理了一个表格是我在文本分类任务上常用的配置范围。超参数常用范围经验值说明学习率1e-5 ~ 5e-52e-5微调预训练模型的标准范围batch_size16 ~ 6432受显存限制尽量大epochs3 ~ 105看验证集指标早停warmup_ratio0.05 ~ 0.20.1防止训练初期震荡weight_decay0.01 ~ 0.10.01正则化防止过拟合max_grad_norm0.5 ~ 2.01.0梯度裁剪防止梯度爆炸学习率是最关键的参数我的经验是先用一个中等值跑一遍看loss曲线。如果loss下降太慢就调大如果loss震荡或者变成nan就调小。warmup的作用是在训练初期用较小的学习率等模型稳定了再升到设定值这个对微调任务特别重要能明显提升稳定性。4.4 检查点保存策略检查点保存看起来简单但策略不对会浪费大量磁盘空间或者丢失最佳模型。我的策略是保存两个检查点最新的和最佳的。最新的检查点用于断点续训每次epoch结束覆盖保存。最佳的检查点用于最终部署只在验证指标超过历史最佳时保存。这样既保证了能续训又不会因为保存太多检查点把磁盘撑爆。检查点里要保存的东西也有讲究除了模型权重还要保存优化器状态、调度器状态、当前epoch、最佳指标值。这些信息在续训的时候都需要。我一开始只保存了模型权重结果续训的时候优化器状态丢了学习率调度从头开始训练效果明显变差。def save_checkpoint(model, optimizer, scheduler, epoch, metric, path): torch.save({ model_state: model.state_dict(), optimizer_state: optimizer.state_dict(), scheduler_state: scheduler.state_dict(), epoch: epoch, best_metric: metric, }, path)4.5 实操心得训练过程中的监控要点训练过程中我主要盯三个东西loss曲线、学习率曲线、验证指标。loss曲线看的是训练是否正常。正常的loss应该是先快速下降然后逐渐平缓如果loss一直不降说明学习率太小或者模型有问题如果loss震荡剧烈说明学习率太大如果loss突然变成nan说明梯度爆炸了。学习率曲线看的是调度器是否正常工作。warmup阶段学习率应该从0线性升到设定值然后逐渐衰减。如果学习率曲线不对说明调度器配置有问题。验证指标看的是模型是否过拟合。如果训练loss持续下降但验证指标开始下降说明过拟合了应该早停或者加正则化。我一般每100个step打印一次训练loss每个epoch结束打印一次验证指标。日志用简单的文本格式方便后续用脚本解析。5. 评估模块与实验追踪的实现5.1 评估指标的选择与计算评估指标的选择取决于任务类型。分类任务我一般看准确率、精确率、召回率、F1值这四个。准确率在类别不平衡的时候会失真所以主要看F1。精确率和召回率能看出模型是偏保守还是偏激进根据业务需求来权衡。指标计算我踩过一个坑多分类的F1要区分macro和micro。macro是每个类别算F1再平均micro是全局算F1。类别不平衡的时候这两个差异很大macro会被小类别拉低micro会被大类别主导。我一般两个都报让看的人自己判断。from sklearn.metrics import accuracy_score, precision_recall_fscore_support def compute_metrics(preds, labels): acc accuracy_score(labels, preds) p_macro, r_macro, f_macro, _ precision_recall_fscore_support( labels, preds, averagemacro ) p_micro, r_micro, f_micro, _ precision_recall_fscore_support( labels, preds, averagemicro ) return { accuracy: acc, precision_macro: p_macro, recall_macro: r_macro, f1_macro: f_macro, f1_micro: f_micro, }5.2 实验追踪的极简方案实验追踪这块我用了一个极简方案每次实验生成一个目录目录里放一个metrics.json记录所有指标再放一个config.yaml记录完整配置。然后写一个脚本扫描所有实验目录把指标汇总成一个表格。这个方案的好处是不依赖任何外部服务所有数据都在本地文件系统里随时可以查看和备份。缺点是没法做复杂的查询和可视化但对于我这种实验数量在几百次这个级别的场景完全够用。汇总脚本大概长这样import json from pathlib import Path import pandas as pd def collect_experiments(exp_dir): records [] for exp_path in Path(exp_dir).iterdir(): if not exp_path.is_dir(): continue metrics_file exp_path / metrics.json config_file exp_path / config.yaml if not metrics_file.exists(): continue with open(metrics_file) as f: metrics json.load(f) record {exp_name: exp_path.name} record.update(metrics) records.append(record) df pd.DataFrame(records) return df.sort_values(f1_macro, ascendingFalse)5.3 结果对比与版本管理有了实验汇总表之后对比不同实验的结果就方便了。我一般会关注几个维度哪个配置组合效果最好、不同随机种子的方差有多大、哪些改动带来了提升。这里有个经验单次实验的结果不要全信要看多次实验的稳定性。我有次调了一个参数单次实验F1提升了2个点高兴得不行结果换了三个种子重跑平均下来反而降了0.5个点。后来我养成了习惯重要的改动至少跑三个种子看平均值和方差。版本管理这块代码用git管理数据和模型用版本号管理配置跟着实验目录走。这三者通过实验目录里的config.yaml关联起来里面记录了代码的commit hash、数据版本号、模型版本号。这样任何时候都能追溯到一次实验的完整信息。5.4 实操心得评估阶段的常见陷阱评估阶段有几个陷阱我踩过这里列出来提醒一下。第一个是测试集不能反复用。如果根据测试集的结果来调参测试集就变成了验证集最终报告的指标会虚高。我的做法是测试集只在最终确定模型后跑一次中间调参只看验证集。第二个是评估脚本要和训练脚本用同一套数据预处理。我有次评估的时候忘了应用训练时的文本截断导致评估结果和训练时的验证结果对不上查了半天才发现是预处理不一致。第三个是指标的计算方式要和业务对齐。比如业务上更关心高精确率那评估的时候就要重点看精确率而不是只看F1。这个需要和业务方提前沟通清楚。6. 部署服务与常见问题排查6.1 推理服务的轻量实现部署这块我用的是FastAPI加Uvicorn的组合简单够用。核心逻辑就是加载模型提供一个HTTP接口接收文本返回预测结果。from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model None tokenizer None class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int confidence: float app.on_event(startup) def load_model(): global model, tokenizer model build_model(config) model.load_state_dict(torch.load(best_model.pt)) model.eval() tokenizer build_tokenizer(config) app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length128) with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) label int(probs.argmax()) confidence float(probs.max()) return PredictResponse(labellabel, confidenceconfidence)这个服务启动后监听一个端口用curl就能测试。生产环境的话前面再加个Nginx做反向代理和负载均衡基本就够用了。6.2 性能优化的几个手段推理性能优化我做了三件事。第一件是模型量化把FP32的权重转成INT8模型体积缩小到四分之一推理速度提升大概两倍精度损失在可接受范围内。第二件是批处理把多个请求攒一批一起推理吞吐量能提升好几倍。第三件是缓存对于重复的输入直接返回缓存结果这个在有些场景下效果很明显。量化用PyTorch自带的动态量化就行几行代码的事quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )批处理需要在服务层做维护一个请求队列攒够一批或者超时了就触发推理。这个实现起来稍复杂但收益很大尤其是QPS高的时候。6.3 常见问题速查表实际运行中遇到的问题我整理成了一个速查表方便快速定位。问题现象可能原因排查方法解决方案服务启动报OOM模型太大或显存不足看日志确认在哪一步OOM量化模型或换小模型推理结果和训练时不一致预处理不一致对比两边的预处理代码统一预处理逻辑响应时间波动大批处理超时设置不合理看请求延迟分布调整批处理参数服务运行一段时间后变慢内存泄漏监控内存使用曲线检查是否有未释放的缓存并发请求时结果错乱全局变量竞争检查是否有共享状态加锁或改成无状态6.4 实操心得上线前的检查清单服务上线前我一定会跑一遍检查清单这里分享出来。模型文件是否正确加载参数量是否和预期一致单条请求的延迟是否在可接受范围内并发请求下结果是否正确有没有出现错乱异常输入空文本、超长文本、特殊字符是否能正常处理服务重启后是否能自动恢复日志是否完整出问题能否快速定位这个清单帮我避免了好几次线上事故。有次就是忘了测异常输入结果上线后遇到一个超长文本直接把服务打挂了后来加了长度限制才解决。7. 我在这套流程上踩过的坑和最终体会这套流程从最初的想法到最终稳定运行中间踩的坑比我预想的多得多。最大的一个坑是过度设计。一开始我想把每个环节都做得尽善尽美数据版本化要支持回滚实验追踪要支持可视化部署要支持灰度发布。结果花了两周时间搭架子真正跑实验的时间反而没多少。后来我砍掉了大部分花哨功能只保留最核心的反而效率高了很多。第二个坑是忽视文档。有段时间我改代码改得很勤但懒得更新文档结果两周后自己都忘了某个参数是干什么的。后来我强制自己每次改完代码花五分钟更新一下对应的文档这个习惯坚持下来受益很大。第三个坑是不重视日志。早期我的日志就是简单的print出了问题根本查不到原因。后来改成了结构化的日志每条日志带时间戳、模块名、日志级别排查问题的效率提升了好几倍。如果让我给刚开始搭这套流程的人一个建议那就是先跑通再优化。不要一上来就追求完美先用最土的办法把整个链路跑通然后再逐个环节优化。跑通的过程中你会对每个环节有更实际的理解这时候再优化才知道该往哪个方向优化。这套流程后来我又扩展了一些东西比如加了简单的A/B测试框架加了模型监控的指标采集但核心的骨架一直没变。我觉得一个好的工程化流程就应该是这样骨架稳定细节可以不断迭代。