三周前我给自己定了一个目标把一个ai-engineering-from-scratch的项目从头到尾完整落地。所谓 from scratch不是说不用现成框架、从零写出一个神经网络而是抛开那些点一下就能出模型的黑盒平台数据管道自己搭训练循环自己写评估指标自己算部署服务自己包——每一步都清楚知道系统在干什么、出了问题该去哪一层查。这篇文章就是这次实践的过程记录。我选的实践载体是中文商品评论情感三分类输入一句评论输出正面、负面或中性。这个任务足够典型既涉及文本处理、模型训练也涉及服务端部署和线上监控一套流程跑下来基本就把 AI 工程化的主干摸了一遍。如果你也正在从调包调参往真正理解全流程的方向走这篇文章应该能帮你少踩几个坑。1. 先别急着写代码AI工程项目的定位与技术选型很多人拿到一个 AI 工程需求第一反应是克隆一个开源仓库跑通 demo然后就开始陷入这玩意儿到底怎么变成产品的迷茫。我的习惯相反先花一两天时间做需求拆解和技术选型把所有模糊的地方变成可执行的规格后面写代码才不会被反复推翻。1.1 需求拆解从一个模糊想法到可执行的项目规格AI 工程里最贵的不是 GPU而是返工。需求没拆清楚就动手最常见的结局是模型训练完了上线时发现接口形态不对或者评估指标根本不是业务方在乎的。所以第一步我用一个表格把核心问题固定下来。需求要素我这次落地选择选择的原因任务类型中文短文本三分类情感数据公开、评估直观、容易扩展到多分类输入数据公开商品评论数据集约 8000 条规模适中单卡 CPU/GPU 都能跑输出形式HTTP APIJSON 返回标签和置信度便于对接上层业务系统使用场景电商评论后台的辅助标注工具对延迟不敏感但要求可解释核心指标宏平均 F1 和置信度校准情况类别不平衡时准确率不可靠这里最关键的一点是明确输出形式和使用场景。很多人拿到任务只想到训练一个高精度模型却没有考虑接接口的人到底要什么。比如辅助标注工具需要的是稳定的批量预测能力那单次请求的延迟要求就不需要压到毫秒级反而更关注吞吐量如果是 C 端实时拦截那才需要上推理加速、模型剪枝这些手段。需求拆解的价值就是让后续所有的技术决策都围绕真实约束展开而不是为了炫技。另外我建议把不做什么也写进规格里。这次我就明确不碰多模态、不做实时流处理、不做增量训练先跑通单模型闭环。范围控制住项目才可能在可控时间内交付。1.2 技术选型为什么是PyTorch、Transformers和FastAPI技术选型没有银弹关键是匹配你的约束。我这次的约束是要能快速迭代要方便调试要能顺畅部署到一台小云主机上。最终选型是 PyTorch HuggingFace Transformers FastAPI Docker。PyTorch 对我来说是心智负担最低的框架。它的动态图机制让训练循环的每一步都可打印、可打断这对 debug 至关重要。我不反对 TensorFlow 或 JAX但如果你要自己写数据管道和训练逻辑PyTorch 的生态会让你少看很多为什么这里又有隐式行为的文档。Transformers 库提供的是预训练模型和分词器这部分没有必要 from scratch。我们常说的 AI 工程重点在工程而不是重新发明模型。自己从零实现 BERT 可能能学到注意力机制但对于一个业务项目来说成本极高、收益极低。合理的边界是模型结构用成熟的数据管道、训练循环、评估流程、服务封装全部自己控制。服务端我选了 FastAPI。它自带 OpenAPI 文档有 Pydantic 做请求校验异步支持也不错。和 Flask 相比FastAPI 在并发压力和请求体验上明显更好和 Django 相比它轻量得多适合只做模型推理服务。Docker 则是为了让部署环境一致避免在我机器上是好的这种经典问题。1.3 哪些部分必须自己写真正的from scratch边界经常有人问我你不是说 from scratch 吗为什么不自己写 Transformer我的回答是from scratch 不等于重复造轮子。AI 工程实践的核心能力在于把系统痛点识别出来然后亲手解决。在这次的链路里我坚持自己写的部分包括数据清洗和样本去重逻辑因为公开数据集的噪音往往比想象中大训练循环和早停机制自己要能控制每一步的梯度行为评估报告生成用混淆矩阵和分组指标判断失败模式API 服务的加载、预测、异常处理逻辑确保模型不是裸奔上线。直接复用的部分是预训练权重、分词器、模型基础类。这两者的边界在于复用部分属于通用能力自己写的部分属于项目特有逻辑。把边界划清楚既不会浪费时间重复造轮子也不会让整个工程变成没人敢动的黑盒。2. 数据是唯一的真相清洗、划分与评估指标实战2.1 拿到数据后第一件事肉眼检查与清洗我见过太多人拿到 CSV 第一件事就是df.head()看两行然后直接喂给模型结果训练到一半才发现标签里混着空值、文本里全是 URL 和乱码。AI 工程里有一句话叫垃圾进垃圾出但更隐蔽的问题是看起来干净其实埋着雷。这次我拿到评论数据后先在终端里用 Python 随机打印了 50 条原始样本边看边记录问题类型。结果发现有三种情况很典型标签列出现空值文本里有大量全角空格和 HTML 实体存在完全重复的评论。清洗逻辑就围绕这三个问题展开。import pandas as pd df pd.read_csv(reviews.csv) print(df.shape) print(df[label].value_counts(dropnaFalse)) # 1. 缺省标签直接删掉因为无法靠规则补全 df df.dropna(subset[label]) # 2. 统一文本格式去全角空格、去HTML实体、转小写中文不转写 df[text] df[text].str.replace(\u3000, , regexFalse) df[text] df[text].str.replace(nbsp;, , regexFalse) # 3. 去重以 文本标签 为准相同文本不同标签是脏数据直接删除 df df.drop_duplicates(subset[text], keepFalse) print(df.shape)这里有个容易被忽略的点drop_duplicates的keepFalse会把所有重复项全部删掉而不是保留其中一条。为什么要这么做因为同一条评论被标注成不同的标签说明标注本身就存在严重分歧这种样本会让模型学到矛盾的决策边界删掉比保留更安全。如果重复文本对应的标签一致那保留一条就够了。实际经验是宁可数据集小一点也不要留下自相矛盾的样本。2.2 数据划分为什么分层采样这么重要很多教程只是简单调用train_test_split从不告诉你为什么要设置stratify。真实业务数据的标签分布几乎都不是均匀的比如这次的数据里负面评论占 55%正面占 30%中性占 15%。如果随机切分验证集里的中性样本可能只剩几十条评估结果会非常不稳定。分层采样的意义就是让训练集、验证集、测试集的类别比例都尽量接近原始分布这样模型在验证集上的表现才能反映它在真实分布上的水平。实操如下from sklearn.model_selection import train_test_split train_val, test train_test_split( df, test_size0.15, random_state42, stratifydf[label] ) train, val train_test_split( train_val, test_size0.15, random_state42, stratifytrain_val[label] ) print(train[label].value_counts(normalizeTrue)) print(val[label].value_counts(normalizeTrue)) print(test[label].value_counts(normalizeTrue))random_state42是固定随机种子这看起来是小事却决定了项目的可复现性。如果没有固定种子每次运行生成的训练集都不一样同一个模型结果可能上下波动好几个点你就很难判断某个改动到底是真有效还是随机波动。还有一点要特别注意如果数据本身是按时间排序的就不能随机划分必须用时间切分否则模型会把未来信息漏到训练集里上线后立刻现出原形。2.3 评估指标准确率骗人F1和混淆矩阵才靠谱这个项目我一开始只盯着准确率95% 的准确率看着挺美直到我打开混淆矩阵才发现中性评论几乎全军覆没——因为模型把绝大多数中性都预测成了负面。这就是类别不平衡下准确率的典型陷阱如果你的训练集里负面占大头模型只要全部预测负面准确率就能轻松超过一半。但在业务上把所有中性评论都当负面处理用户体验是灾难性的。所以这次我把评估指标定为宏平均 F1也就是对每个类别单独算 F1再取算术平均。这样任何一个类别表现差整体指标都会被拖下来。混淆矩阵则能告诉你具体的失败模式是什么。from sklearn.metrics import classification_report, confusion_matrix, f1_score y_pred model.predict(val_loader) print(classification_report(val_labels, y_pred, target_names[negative, neutral, positive])) print(confusion_matrix(val_labels, y_pred))我在实践中还养成了一个习惯不只打印总指标还要分不同的文本长度、不同关键词分组的子集合再看一遍指标。比如太贵了这种短评和物流很快但包装很烂这种复合情感评论模型的失败模式完全不同。把评估结果拆细你才能找到真正要优化的方向而不是拿着一份漂漂亮亮的平均数自我感动。3. 从训练到上线手写全链路AI工程代码3.1 数据管道与特征表示把中文文本喂给模型前的最后一步清洗完数据之后接下来的任务是把原始文本转换成模型能吃的数值输入。我用的预训练模型是bert-base-chinese它自带的分词器负责把中文句子切成带词边界的 token 序列再映射成词典中的 id。这个过程看起来只是调一下库里面却有几个工程细节值得注意。首先是截断策略。我把单条评论的最大长度设为 128 个 token。太长会浪费显存太短则可能丢关键信息。对短文本分类任务来说128 通常够用了。第二个细节是padding策略训练和推理时我统一用max_length的方式补齐到固定长度这样 batch 内张量维度完全一致能省去动态 padding 的复杂度代价是多一点存储和计算对于这个量级的数据完全可接受。from transformers import BertTokenizer import torch from torch.utils.data import Dataset tokenizer BertTokenizer.from_pretrained(bert-base-chinese) class ReviewDataset(Dataset): def __init__(self, texts, labels, max_len128): self.texts texts self.labels labels self.max_len max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): text str(self.texts[idx]) label int(self.labels[idx]) inputs tokenizer( text, truncationTrue, paddingmax_length, max_lengthself.max_len, return_tensorspt ) return { input_ids: inputs[input_ids].squeeze(0), attention_mask: inputs[attention_mask].squeeze(0), labels: torch.tensor(label, dtypetorch.long) }构建好 Dataset 之后用 DataLoader 包一层设置batch_size16和shuffleTrue。shuffle 的作用是打乱样本顺序避免模型记住同一个类别的连续模式这在小数据集上尤其重要。数据管道写完后我建议单独跑一个 batch 出来打印 shape确认为[16, 128]再进模型省的后面训练时才发现维度问题。3.2 训练循环自己动手写才能真正控制训练过程很多库的 Trainer 封装帮我们隐藏了大量细节但也隐藏了出问题时的线索。这次我坚持自己写训练循环因为我想精确控制每一次参数更新的行为。先看核心代码from transformers import BertForSequenceClassification from torch.optim import AdamW from torch.utils.data import DataLoader model BertForSequenceClassification.from_pretrained( bert-base-chinese, num_labels3 ) device torch.device(cuda if torch.cuda.is_available() else cpu) model.to(device) optimizer AdamW(model.parameters(), lr3e-5) train_loader DataLoader(train_dataset, batch_size16, shuffleTrue) for epoch in range(3): model.train() total_loss 0 for step, batch in enumerate(train_loader): optimizer.zero_grad() input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[labels].to(device) outputs model( input_idsinput_ids, attention_maskattention_mask, labelslabels ) loss outputs.loss loss.backward() optimizer.step() total_loss loss.item() if step % 50 0: print(fepoch {epoch}, step {step}, loss {loss.item():.4f})一个小细节是optimizer.zero_grad()必须放在前向之前还是之后其实放在前向之前更稳妥。如果你放在loss.backward()之后每次迭代前清零没有问题但有些复用的代码可能忘了清梯度会不断累积。另一个关键点是梯度裁剪BERT 这类大模型训练在微调阶段很容易出现梯度爆炸我会在loss.backward()之后、optimizer.step()之前加一行torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)这行的作用是把梯度的 L2 范数限制在 1.0 以内防止参数一次性被更新到离谱的范围。配合学习率 3e-5这个组合在大多数中文文本分类任务上都能稳定收敛。训练中我还会做早停。每次 epoch 结束后在验证集上计算 F1如果连续两个 epoch 没有提升就停止训练并保存当前最优权重。模型保存不要只保存整个模型对象更好的方式是保存state_dict并同时保存 tokenizer 的配置文件这样部署时加载最干净。3.3 评估调参在验证集上做决策不盯着测试集训练完成后我会回到验证集上做评估。这里有一个纪律测试集只能使用一次所有调参决策都必须基于验证集否则你的模型在测试集上的分数就是被我亲自污染过的分数。在项目实践中这条纪律比你想得更重要——因为看到测试集分数不理想人很容易忍不住回头调两下参数再跑一次多来几轮测试集就等于训练集了。调参时我最先看的是混淆矩阵。如果模型把中性大量误判成负面我会考虑调整类别权重给中性类更高的损失权重或者对中性类做简单的上采样。如果验证集 F1 已经稳定但损失还在下降说明模型可能过度拟合了训练集这时我会减小学习率或增加正则化。一次只改一个变量是铁律。我见过有人同时改了学习率、batch size、max length结果模型涨了两个点根本不知道是哪个改动起了作用。我的习惯是维护一个实验记录表每次只改一个维度记录训练损失、验证 F1、显存占用、训练时间。项目结束后这个表格就是最值钱的交付物之一。3.4 模型部署用FastAPI封装一个生产可用的推理服务模型训练完不算结束真正让它产生价值的是部署成可供业务调用的服务。我选择 FastAPI核心原因是它自带异步支持和自动接口文档调试和对接都方便。部署逻辑分两层加载模型然后提供预测接口。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() model None tokenizer None class ReviewRequest(BaseModel): text: str class ReviewResponse(BaseModel): label: str confidence: float app.on_event(startup) def load_model(): global model, tokenizer model BertForSequenceClassification.from_pretrained(saved_model) model.eval() model.to(cuda if torch.cuda.is_available() else cpu) tokenizer BertTokenizer.from_pretrained(saved_model) app.post(/predict) def predict(req: ReviewRequest): inputs tokenizer( req.text, truncationTrue, paddingmax_length, max_length128, return_tensorspt ) inputs {k: v.to(model.device) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim1)[0] label_idx probs.argmax().item() labels [negative, neutral, positive] return ReviewResponse( labellabels[label_idx], confidenceprobs[label_idx].item() )这里有几个非常值得注意的工程细节。第一模型必须在startup事件里加载一次而不是在每个请求里重新加载否则单次推理会慢到怀疑人生。第二推理时要包在torch.no_grad()里告诉 PyTorch 不需要构建计算图这能显著降低内存占用和推理延迟。第三model.eval()必须显式调用否则 dropout 层在推理时依然生效结果会不稳定。为了让部署环境可复现我用 Dockerfile 把服务包装起来。基础镜像选择pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime安装依赖后把模型文件复制进去暴露 8000 端口。这里最重要的经验是把模型文件放到/models目录下并通过环境变量指定路径这样升级模型时就只需要替换镜像中的文件而不需要改代码。上线前我做了一个简单的压测并发 5 个请求平均延迟 80ms吞吐足够支撑一个小型标注平台。4. 踩坑实录训练异常、数据泄漏与部署隐患排查4.1 训练损失变成NaN或不收敛按这个顺序排查训练过程里最让人崩溃的就是 loss 变成 NaN或者怎么训练损失都不下降。我这次也中招了排查顺序如下现象可能原因排查动作Loss 直接 NaN输入数据存在 NaN/极大值检查数据清洗后的文本和标签Loss 稳定不降学习率不合适先跑一个 batch 过拟合测试梯度爆炸学习率过高、没有梯度裁剪加入clip_grad_norm_验证集 F1 波动大随机种子未固定固定random_state对比实验先跑一个 batch 过拟合是我最推荐的快速判断手段。具体做法是取训练集的一个 batch设置比较大的训练轮数正常情况模型应该能把这个 batch 的 loss 降到很低。如果连一个 batch 都过拟合不了说明是代码或数据的问题而不是模型结构的问题。这种排查思路能把模型不收敛这种模糊问题快速定位到代码 bug还是超参不佳。我还踩过一个很隐蔽的坑混合精度训练时某些算子不兼容导致 loss 变成 NaN。当时我在训练循环里加了torch.cuda.amp自动混合精度结果大概跑了 200 步之后损失直接消失。排查后的结论是数据里有一条评论触发了非正常路径关闭混合精度后一切正常。所以当你启用高级功能时先确认它在你的数据分布上是真的稳定。4.2 数据泄漏为什么模型在验证集完美上线就崩这是 AI 工程里的慢性毒药。我在验证集上 F1 达到了 0.92但上线后面对真实评论效果肉眼可见地差。后来复盘找到了两个泄漏来源。第一个泄漏来源是我在数据清洗阶段用了全量数据集做统计特征比如计算全量文本的平均长度然后对异常长文本做截断。这个统计值包含了测试集和验证集的数据等于在建模之前就把测试集的信息透传到了模型里。解决办法是把清洗逻辑拆成基于先验规则和基于数据统计两部分数据统计必须在划分训练、验证、测试之后再做而且只基于训练集。第二个泄漏来源更隐蔽去重的时机。我先对全量数据做了去重再去划分数据集结果训练集和测试集中出现了完全相同的评论。这等于测试集已经见过标准答案线上自然打回原形。正确的顺序是先按时间或随机方式划分再在每个集合内部去重这样能保证验证集和训练集没有重复样本。数据泄漏之外还有分布偏移问题。训练数据里的评论集中在价格、物流、质量这些关键词而线上真实评论会出现大量新表达比如xx 品牌 yyds。解决分布偏移的工程手段不是想办法让模型变得更聪明而是建立线上数据回流机制定期抽样线上请求人工标注后补进训练集形成持续迭代的闭环。4.3 API服务隐藏问题内存、并发、超时与模型漂移部署上线后真正的运维挑战才开始。我遇到的第一类问题就是内存模型常驻内存需要约 1.2GB但 Docker 容器默认能用完整个宿主机的内存一旦并发请求同时进来显存和内存都可能被撑爆。解决方法是给容器设置内存上限比如docker run --memory2g同时限制并发数。FastAPI 的异步特性在模型推理场景下并不完美因为 PyTorch 的 CPU 推理本质上是阻塞操作。如果多个请求同时进来模型权重会被频繁读取GPU/CPU 利用率波动很大。我的做法是在预测函数外面加一个全局锁把并发请求串行化先保证正确性再考虑吞吐。如果你确实需要高并发可以用批处理队列攒够 N 个请求一起推理但这会显著增加单请求延迟需要根据业务权衡。最后是模型漂移。模型上线后不是万事大吉我每周会抽 200 条真实请求人工打标签和模型预测结果对比计算最新 F1。如果发现下降趋势就把这些样本加入训练集重新微调。这条循环跑起来之后这个 AI 工程才算真正拥有了从零到一之后的从一到稳。做这个项目最大的体会是AI 工程从零开始落地的过程本质上是不断和不确定性做斗争。数据会脏、梯度会炸、接口会超时但只要你把每一个环节都变成自己亲手控制、可观测、可复现的模块那些让你深夜抓狂的问题就会慢慢变成你经验库里的普通条目。下次再遇到类似问题你可能只需要五分钟就能定位到根因。