做AI工程这段时间我最大的感受是市面上不缺模型教程不缺框架文档缺的是“把一堆组件按正确姿势组装起来让它稳定跑在生产环境”的经验。很多人拿着训练好的模型不知道怎么上线或者上了线之后一遇到数据波动就抓瞎。这个“ai-engineering-from-scratch”的项目本质上就是把一条AI应用链路从零开始搭出来需求澄清、数据准备、模型选型、服务封装、性能评估、监控迭代每一环都自己动手不依赖现成的平台套件。适合那些已经会写Python、懂一点机器学习基础但还没完整走过一个生产级项目的开发者也适合团队里刚接手AI落地任务、需要快速建立全局观的技术负责人。我踩过的坑不算少从GPU显存溢出到线上推理延迟抖动再到数据分布悄悄漂移导致指标下滑每一个问题都让项目差点翻车。这篇文章把整个过程拆开讲清楚包括每一步为什么这么选、参数怎么定、哪些环节最容易出幺蛾子尽量让读者看完就能照着搭一套属于自己的AI工程链路。1. 先想清楚AI工程到底在解决什么问题很多教程上来就讲模型结构、训练技巧但实际工作中模型只是整条链路里的一小部分。数据管道动不动就占掉一半工期模型服务化要考虑吞吐和延迟上线后还要面对监控告警、版本迭代、回滚机制这些传统软件工程的老问题。所谓AI工程就是把这堆事当成一个系统工程来对待而不是把一个notebook跑通就算完事。1.1 从“能跑”到“能生产”差距到底在哪用一个类比来说训练一个模型有点像在实验室里烤出一个完美的蛋糕。你按配方来温度时间都调好了蛋糕确实好吃。但AI工程要解决的是开一家蛋糕店每天要稳定做出几百个品质一致的蛋糕原料采购要有标准烤箱坏了要有备份员工要按SOP操作顾客投诉要有反馈渠道。整个链路任何一个环节出了问题蛋糕店都开不下去。对应到实际项目里这些环节分别是数据环节数据采集是否稳定清洗逻辑是否可复用标注质量怎么把控数据版本怎么管理。模型环节选什么算法特征怎么设计训练和评估流程是否可复现模型文件怎么保存。部署环节模型怎么被业务调用用离线批量预测还是在线实时推理性能能不能扛住峰值流量。运营环节模型上线后效果怎么监控数据分布变了怎么办模型多久重新训练一次出了badcase怎么定位。很多团队在“能跑”阶段浪费了大量时间因为代码写得太随意数据清洗散落在各种临时脚本里模型参数写死在配置文件里评估脚本和训练脚本互相矛盾。等项目要上线了发现线上数据和训练数据对不上推理服务的内存又撑不住才回头补课。这个项目取名“from-scratch”的意义就在这里不回避任何一个基础环节从最底层开始认真搭。1.2 和“调包”与“纯研究”的区别现在AI门槛被各种封装库拉得很低很多人调一下接口就能出一个demo这确实很容易。但这类做法的问题在于一旦线上环境出现异常你基本没有任何排查抓手因为中间的关键环节对你来说是个黑盒。纯研究导向的人则相反他们特别关注模型指标准确率、AUC这些但对工程交付没什么概念模型做得再花哨没法和业务系统对接起来也是白搭。AI工程恰恰站在两者中间既要知道模型怎么训、怎么调也要知道怎么把模型变成业务系统里一个稳定、可维护的服务。通俗地说就是“既懂算法也懂工程”。这个项目里的每一步都是在训练这种复合能力——数据质量不过关就回头改管道推理延迟太高就做模型蒸馏或者加缓存效果不达标就分析badcase反哺数据迭代。整个过程是一个循环不是一个一次性流程。2. 核心环节深度拆解每一步为什么这么选很多人做项目是一上来就写代码我建议先花时间把方案设计好。AI工程里的技术选型直接决定后面踩坑的数量和深度。2.1 数据管道数据不对后面全是白搭数据是整个链路里最不起眼、但风险最大的环节。我见过太多项目模型结构换了一版又一版准确率就是上不去最后查下来发现训练数据里有大量重复样本而且某些类别的label本身就标错了。做数据管道重点要解决三件事来源稳定、清洗可复现、版本可追溯。来源稳定指的是数据采集不能三天两头断流或者格式经常变。这块通常要建立一个统一的数据接入层不管是数据库同步、日志采集还是文件上传都走同一套接口进到数据湖或者数仓。清洗可复现指的是同一批原始数据任何时候重新跑清洗脚本产出的结果必须是一样的不能因为运行顺序或者随机因子导致结果漂移。这里有个很关键的习惯清洗脚本里用到随机数一定要固定种子所有变换操作要写成幂等的。版本可追溯指的是你知道当前模型是用哪一批数据训练出来的。建议给每一份训练数据打上版本号比如用日期批次同时把这个版本号和模型文件绑定记录。这样线上模型出了问题你可以快速定位是数据原因还是模型原因而不是靠猜。2.2 模型策略不要一上来就训大模型大多数业务场景根本不需要从零预训练一个大语言模型或者超深度的网络成本和收益完全不成正比。更务实的做法是先清楚自己的任务是分类、回归、排序还是生成然后从效果和成本两个维度做权衡。如果是文本分类、实体抽取这类常见任务基于预训练模型做微调通常比从零训练效果好得多而且训练成本可以控制在一张消费级显卡能承受的范围。如果是结构化表格数据XGBoost/LightGBM这些传统模型往往比深度学习更适合因为它们对小样本、特征稀疏的场景更鲁棒训练和调参也更快。做一个技术选型对比方便大家对着自己的项目参考场景推荐方案理由注意事项短文本分类预训练模型微调如BERT类语义理解能力强需关注推理延迟可考虑蒸馏表格数据预测XGBoost / LightGBM训练快、可解释性较好特征工程仍然很关键图像分类ResNet / EfficientNet微调收敛快、效果好数据增强直接影响泛化序列数据LSTM / Transformer时序依赖建模注意滑窗大小与采样频率这个项目里我把重点放在“怎么系统地做选型和评估”上而不是单纯追新模型。模型效果对比要有统一的评估脚本不能今天用accuracy明天用F1那样对比毫无意义。2.3 特征工程决定效果上限的隐形因素特征工程常常被AI工程化的教程忽略但实际业务里它往往决定模型效果的天花板。模型结构只是逼近这个上限的工具特征质量从根本上限定了能学到的信息量。我常用的特征设计套路是先做业务理解把已知的领域知识转化为特征比如用户ID的统计特征、时间窗口内的行为聚合然后再做数据探索用相关性分析和特征重要性排序找出有效特征。特别要注意“未来信息泄漏”这个坑——用训练时刻之后的数据构造特征看起来指标很高上线后立刻崩盘。典型例子是在预测用户次日留存时把用户当天的付费金额作为特征这在训练集里有效但在真实预测时根本拿不到当天数据。特征存储也需要重视。线上推理和离线训练的特征必须来自同一个特征平台否则线上缺特征或特征定义不一致模型表现会大打折扣。这个项目的“from-scratch”精神就体现在这里连特征都要自己搭一套可复用的存取方案而不是每个模型各自搞一套临时逻辑。2.4 部署策略离线跑批还是在线服务模型部署先问自己一个问题业务上需要实时拿到预测结果吗如果只是给用户打标签、批量生成推荐结果、晚上跑风控规则那离线批处理就够了。离线方案省心很多不用考虑延迟和并发每天定时跑一次任务把结果写回数据库业务方直接读结果就行。如果业务必须实时响应比如客服对话中的实时意图识别、下单时的实时风控那就需要在线推理服务。在线服务要考虑的事情多得多最核心的是两件延迟和吞吐。延迟就是接口从收到请求到返回结果的时间一般业务容忍度在200ms以内超过的话体验就很差。吞吐是服务单位时间能处理多少请求需要压测找出服务的上限。举个例子一个BERT微调模型的单次推理时间大约是20ms在GPU上单卡并发16路请求时延迟会涨到50ms左右再往上加并发延迟会非线性上升。所以上线前一定要压测找到延迟和吞吐的平衡点。技术选型方面FastAPI是轻量服务的不错选择简单直接TensorFlow Serving和TorchServe适合模型规模大、需要动态批处理和高性能的场景。我个人的习惯是团队规模小、定制逻辑多的项目用FastAPI自己封装如果只是标准模型推理直接用专门的推理服务运维成本更低。关于“动态批处理”多说一句在线推理场景里单条请求推理效率其实很低因为GPU算力没有充分打满。把多个请求攒在一起凑够一批再统一推理吞吐能提升好几倍。但批处理有个代价——延迟会增加因为第一批请求要等后面的请求到齐才开始推理。所以动态批处理的核心策略是设置一个最大等待时间比如20ms时间一到就算批次没凑满也立刻推理。2.5 观测与反馈不看监控等于在裸奔模型上线只是起点后面才是真正考验工程能力的地方。核心观测项包括推理服务的CPU/内存/延迟/错误率模型预测分布的变化线上效果指标点击率、转化率、准确率的波动。这里一个很多人忽略的关键点模型预测分布要和训练时的分布做对比。如果线上业务发生了变化比如新用户策略调整、外部环境变化模型预测结果可能会整体偏向某个类别而准确率指标还没有明显下跌。这种“缓慢漂移”是最危险的它会一点点降低业务效果等人察觉时往往已经过去好几周了。我建议每天对线上预测结果做一次分布统计和训练集的分布对比设置一个容忍阈值超了就报警。反馈闭环指的是模型效果数据回流到训练集里。用户行为数据、业务结果数据都需要定期同步到数据管道经过清洗和标注后再进入下一轮训练。没有这个闭环模型只会越来越跟不上业务变化。这个项目里我专门设计了一个“周级迭代”机制每周固定时间同步最新数据重新评估模型如果效果低于线上版本就继续用旧模型否则发布新版本。3. 实操过程从零搭建一个文本分类服务下面用一个具体场景串起整个流程。假设我们要给一个客服平台做“工单智能分拣”目的很简单根据用户提交的问题描述自动判断工单类别比如账号问题、支付问题、技术故障、退款申请等然后自动路由到对应处理组。这个场景在业务中非常普遍而且流程完整适合当作参考模板。3.1 需求澄清与评估指标定义第一件事不是写代码而是把问题和评估方式定义清楚。工单分类是一个典型的文本多分类任务但“类别”的粒度需要和业务方反复确认。分得太粗路由不精确分得太细标注成本高模型也容易混淆。我们最终定了8个类别覆盖95%以上的工单场景剩下的进入“人工兜底”。评估指标不能只看准确率因为类别不均衡“咨询”类往往最多而“投诉”类较少。我采用宏平均F1作为主要指标再辅以每个类别的精确率和召回率。为什么不用准确率一个极端的例子如果90%的工单都是咨询类模型全部预测成咨询类准确率也有90%但真正重要的投诉类全被漏掉了这对业务毫无意义。宏平均F1会平等对待每个类别更能反映模型的真实能力。3.2 数据准备清洗、切分、验证工单数据来自客服系统的历史记录提取近三个月的工单共约5万条。原始字段包括用户表述、客服备注等需要做这些清洗步骤去除HTML标签和URL链接全角符号转半角去除重复工单同一用户连续提交的相同内容视为一次统一简繁体根据业务场景决定这里统一转为简体清洗之后剩下约4.2万条有效数据。切分比例用8:1:1即训练集3.36万、验证集4200、测试集4200。切分时有个细节按用户ID做分组切分而不是按样本随机切——保证同一个用户的多条工单不会同时出现在训练集和测试集里这能更真实地模拟“新用户来了怎么办”的场景。之后做类别分布的检查。如果某个类别样本太少比如“退款申请”只有300条需要考虑数据增强或者选择更鲁棒的模型结构。3.3 模型实现从词向量到预训练模型我用了两步走的方式。第一步搭一个基于词向量TextCNN的基线模型不需要预训练模型几分钟就能训完作为效果参考基线。第二步用预训练模型微调做主力模型对比效果和推理延迟。模型代码部分为了控制篇幅我给出核心的训练和评估框架方便参考复现。import torch import torch.nn as nn from transformers import AutoTokenizer, AutoModel, AdamW NUM_CLASSES 8 DEVICE torch.device(cuda if torch.cuda.is_available() else cpu) class TextClassifier(nn.Module): def __init__(self, model_name: str, num_classes: int): super().__init__() self.backbone AutoModel.from_pretrained(model_name) hidden_size self.backbone.config.hidden_size self.classifier nn.Sequential( nn.Dropout(0.3), nn.Linear(hidden_size, num_classes) ) def forward(self, input_ids, attention_mask): outputs self.backbone( input_idsinput_ids, attention_maskattention_mask ) pooled outputs.last_hidden_state[:, 0, :] # [CLS] token return self.classifier(pooled)训练流程要注意几个点学习率不要直接用默认的5e-5对预训练模型微调最好从2e-5到3e-5之间开始batch size要看GPU显存来定显存不够就减小batch size同时配合梯度累积来保持训练稳定性训练轮数通过早停来控制当验证集F1连续两轮不提升就停止。def train_epoch(model, data_loader, optimizer, device): model.train() total_loss 0 for batch in data_loader: input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[labels].to(device) optimizer.zero_grad() outputs model(input_ids, attention_mask) loss nn.functional.cross_entropy(outputs, labels) loss.backward() optimizer.step() total_loss loss.item() return total_loss / len(data_loader)评估阶段不能只打印一个Loss要把每个类别的精确率、召回率、F1都输出出来。这里我写了一个简单的评估辅助函数from sklearn.metrics import classification_report def evaluate_model(model, data_loader, id_to_label: dict): model.eval() all_preds [] all_labels [] with torch.no_grad(): for batch in data_loader: input_ids batch[input_ids].to(DEVICE) attention_mask batch[attention_mask].to(DEVICE) logits model(input_ids, attention_mask) preds torch.argmax(logits, dim-1).cpu().numpy() labels batch[labels].numpy() all_preds.extend(preds) all_labels.extend(labels) target_names [id_to_label[i] for i in sorted(id_to_label) if i NUM_CLASSES] print(classification_report(all_labels, all_preds, target_namestarget_names, zero_division0))训练完成后在测试集上的宏平均F1大概是0.86类别准确率最低的是“退款申请”因为这类工单的表述变体比较多比如“我不想买了”“给我退钱”“订单取消”指向的其实是同一件事。这个badcase在后文会提到怎么处理。3.4 服务封装与上线模型效果满足要求后进入服务化环节。我用FastAPI封装在启动时加载模型和tokenizer推理接口接收文本返回类别ID和置信度。这个阶段最主要的工作不是写接口而是做好异常处理和性能优化。第一是输入长度控制。用户提交的工单长度参差不齐超过模型最大长度的部分要截断或分段处理。记住截断可能导致关键信息丢失所以在截断策略上要做取舍。我通常的做法是截断前256个字符和末尾128个字符拼起来作为输入尽量减少关键信息丢失。第二是结果置信度阈值。模型输出的softmax置信度不能直接当作绝对可信的标准。我们需要根据验证集结果设定一个阈值低于阈值就返回“待人工审核”而不是强行分类。这个机制能显著减少错误路由代价是一小部分工单需要人工处理但业务上是划算的。下面是服务封装的核心代码结构from fastapi import FastAPI from pydantic import BaseModel app FastAPI() model, tokenizer load_model() class InferenceRequest(BaseModel): text: str class InferenceResponse(BaseModel): label_id: int label_name: str confidence: float app.post(/predict, response_modelInferenceResponse) def predict(req: InferenceRequest): text clean_text(req.text) encoded tokenizer( text, truncationTrue, max_length512, paddingmax_length, return_tensorspt ) with torch.no_grad(): logits model(**encoded.to(DEVICE)) probabilities torch.softmax(logits, dim-1) confidence, label_id torch.max(probabilities, dim-1) label_id int(label_id.cpu().numpy()[0]) confidence float(confidence.cpu().numpy()[0]) # 低于阈值的返回人工兜底 if confidence CONFIDENCE_THRESHOLD: return InferenceResponse(label_id8, label_name需人工审核, confidenceconfidence) return InferenceResponse(label_idlabel_id, label_nameID_TO_LABEL[label_id], confidenceconfidence)3.5 稳定性保障压测与限流模型服务上线的第一道坎是压测。我用Locust做并发压测验证服务在预期峰值流量下的表现。压测结果对我们的优化方向影响很大我记录了一组典型数据。并发数平均延迟(ms)P99延迟(ms)每秒处理请求数1253235104068180207514026050160320310可以看到并发从20涨到50吞吐只提升了一点点延迟却翻倍了这说明服务已经进入过载区。于是我把产品侧的单机并发上限设为20然后通过水平扩容来支撑更大流量。限流方面我加了一个简单的并发信号量超过上限的请求直接返回503并提示稍后重试。大量用户的请求文本长度不一长文本会显著拉高延迟。我的优化方案是长度超过200的文本走一个单独的推理队列避免长文本阻塞短文本请求。这一招很有效P99延迟从320ms降到了180ms左右。4. 常见问题与排查技巧实录这部分记录我在这个项目以及过往类似项目里真实遇到的坑每一项都配了排查思路可以直接收藏备用。4.1 GPU显存溢出都改好了为什么还会炸显存溢出是微调阶段最常遇到的问题。排查步骤是先看batch size是否太大batch size等于1都炸的话再检查序列长度——有些样本长度是450padding到512后显存占用比预期高很多。另一个隐蔽原因是参数管理优化器里如果对冻结层也维护了梯度状态显存会白白多占。排查方法很简单用torch.cuda.memory_summary()看一下内存分配曲线基本能定位问题。nvidia-smi --query-gpumemory.used,memory.total --formatcsv如果batch size已经调到4了还是OOM就启用梯度累积accumulation_steps 4每4个小batch做一次参数更新效果等价于batch size16但显存占用控制在单batch的水平。4.2 训练集和线上数据分布不一致这是最隐蔽的问题没有之一。模型在验证集上表现很好上线后效果却差得离谱。原因往往是线上输入和训练数据存在一些细微差异“咨询”类工单在训练集里长度普遍100字以内线上却经常出现300字的长描述线上有大量“发票”“合同”等稀有词汇训练集里几乎没出现过。应对办法有两类一类是上线前做小流量灰度观察一小段时间的预测分布和badcase确认没问题再放量另一类是在持续运行中做数据分布监控用KS检验或简单的分布直方图对比发现问题后及时补充相关样本并重新训练。4.3 延迟与吞吐量的平衡关系很多第一次做在线推理的同学会陷入一个误区并发越高越好。实际上并发超过一定阈值后延迟会剧烈增长吞吐反而停滞不前。正确姿势是先定义好SLA比如P99延迟小于200ms再压测找到能支撑这个SLA的最大并发数最后按这个数字做限流和扩容规划。如果你用的是GPU推理一定要考虑动态批处理。实测下来把batch size从1调到8GPU利用率能提升好几倍。但注意动态批处理需要等待等待时间设置不当会拖垮延迟指标。我的经验是等待时间和单次推理时间保持在同一个量级比如单次推理20ms等待窗口就设20ms。4.4 冷启动与模型加载太慢怎么处理预训练模型文件动辄几百MB甚至上GB服务一启动就要几十秒。这对业务来说很难接受尤其当服务因为更新需要滚动重启的时候。优化方案是先把模型序列化成本地缓存文件启动时直接用from_pretrained(..., local_files_onlyTrue)加载再把模型加载逻辑放到异步初始化里接口先返回“模型加载中”的健康检查状态等模型ready后再标记服务为可用。容器化部署时还有一个细节镜像里预置模型文件而不是每次启动时从对象存储下载。这样能省掉大量启动时间。4.5 模型效果持续下降但没人发现模型效果下降往往不是突变而是渐变。业务数据一天天变模型预测能力一点点衰减单看某一天的指标根本看不出问题。我建议设置一个“新老模型对照”机制线上同时跑新模型和旧模型把预测结果做交叉验证持续观察差异比例。差异超过阈值就触发告警。这比定期看指标更灵敏因为预测差异往往先于业务指标变化出现。4.6 成本失控训练和推理开销比预期大得多很多人忽略AI项目的持续成本。训练是一次性的还好推理成本是持续性的每天调用量一大GPU账单就让人肉疼。省钱思路有几个用小模型替代大模型蒸馏或直接换更小的底座对简单样本走规则匹配只把复杂样本送进模型对推理密集的业务做批量优化。文本分类任务我们就做了个“长度关键词”的预筛规则约30%的工单压根不需要进模型直接匹配到正确类别推理量直接砍掉三分之一。5. 从零到一一天的完整搭建流程如果你也想从零抛出一个能用的项目下面这条时间线可以作为参考。这只是一个参考节奏具体耗时取决于你的数据和业务复杂度。上午用来搭数据和训练下午用来封装服务和压测晚上处理badcase和准备灰度发布。一天虽然紧但能完整跑通一个闭环建立整体感觉。之后就是反复迭代观察线上数据、补充badcase、重新训练、评估对比、灰度发布。这个循环转得越流畅AI工程的成熟度就越高。我个人的体会是这个项目最值钱的不是最终那个0.86的F1报告而是中间被迫解决的问题数据怎么切分才公平、置信度阈值定多少才合适、并发上限调到哪里才平衡、监控指标选哪些才有预警价值。这些事情在任何一个教程里都不会写得特别细但真实项目里每一天都要面对。把它们一个一个解决掉你就真正理解了什么叫AI工程。