如果你已经受够了那种“打开Notebook、加载一个预训练模型、跑通就算学会”的教程那ai-engineering-from-scratch这个项目名应该能让你眼睛一亮。它要做的就是“从零开始做AI工程”注意这里说的零不是让你从手写矩阵乘法开始而是把一条从原始数据到线上推理的完整链路亲手搭起来数据管道、训练脚本、评估逻辑、推理服务、监控闭环每一个环节都不靠现成的MLOps平台偷懒。简单说它解决的是一个很常见但很少被系统讲透的问题为什么代码在本地能跑、指标也好看一上真实场景就崩为什么换个数据、换台机器、流量大一点模型就撑不住如果你是一个刚转AI工程方向的开发者或者已经会用现成框架调模型、但一到部署环节就卡壳的工程师这套实践路径应该能帮你省下大量试错时间。1. 项目定位与设计思路为什么从零开始更值得1.1 从零的真正含义掌握全链路的控制力很多教程教你三句话加载一个模型from transformers import pipeline。这没问题但问题在于当模型的输出不符合预期、线上延迟突然爆表时你根本不知道应该去哪一步查。从零做一遍的本质是把黑盒拆成白盒。训练代码是你写的切分数据是你写的部署脚本是你写的那么任何环节出问题你至少知道回到哪一段代码去找原因。这不是炫技是建立一种“出了问题我不慌”的掌控感。我见过太多新人fine-tune一个模型拿到的准确率很高结果一查是数据泄漏导致的就是因为他对切分逻辑没有控制力。用做饭来类比现在很多AI课程是教你拆一包料理包加热三分钟上桌但这个项目是逼你从洗菜、切菜、备料、掌握火候都走一遍。过程确实啰嗦但只有这么走一次你才会真正理解盐放多了会咸、火太大会糊。成品不好吃的时候你就能迅速定位是哪个环节出了问题而不是对着黑盒干瞪眼。1.2 项目覆盖的范围与刻意划出的边界做这个项目时我给自己列了一个模块清单这是整个闭环的骨架数据层采集、清洗、去重、按业务场景切分训练/验证/测试集记录数据版本训练层选一个足够小但完整的模型手写训练循环管理好随机种子、学习率、早停评估层不只打印准确率还要能根据业务目标选择合适的指标并解释模型坏在哪类样本上服务层用FastAPI把训练好的模型包成HTTP接口处理输入校验、并发、缓存、健康检查监控层记录线上预测结果和输入分布能够及时发现数据变化和效果衰减。我刻意没有做的事情是不做分布式训练、不上K8s、不接外部AutoML平台。整个项目要求在一台普通开发机上把闭环跑通8G显存就够纯CPU也能跑。原因很简单如果一个小而完整的系统你都控制不住引入更多组件只会让问题更难定位。先把小系统做成闭环再谈扩展这是我认为最稳妥的工程路径。2. 核心原理拆解数据、训练、评估到底在解决什么问题2.1 数据管道命门往往不在模型在输入我发现大多数“模型效果不行”的case最后都查到了数据头上。数据管道被轻视是新手翻车的第一大原因。首先要做的是把原始数据变成干净的、可复现的数据集。我习惯每跑一次训练前给数据文件算一个hash并写进实验日志否则等到模型指标突然变化时你连是不是换了数据都说不清。第二个关键点是切分方式。很多任务不能随机打乱切分。比如用过去60天数据预测未来如果随机切分模型等于提前看到了“未来”的样本特征测试集分数虚高。正确的做法是按时间边界切分前N天训练后M天验证。我做过一个排队时长预测项目随机切分时AUC有0.87改成按门店和时间切分后掉到0.71这才是模型真实水平。第三个关键点是特征缩放。min-max或z-score的均值、标准差只能从训练集上拟合再把同一个scaler应用到验证集和测试集。新手常犯的错是拿着全部数据去fit scaler这已经造成了特征泄漏。下面的写法才是安全的import pandas as pd from sklearn.preprocessing import MinMaxScaler train_df pd.read_csv(train.csv) val_df pd.read_csv(val.csv) scaler MinMaxScaler() # 只拟合训练集 scaler.fit(train_df[[age, amount, session_len]]) train_df[[age, amount, session_len]] scaler.transform( train_df[[age, amount, session_len]] ) val_df[[age, amount, session_len]] scaler.transform( val_df[[age, amount, session_len]] )用一句话解释fit相当于考试前划重点transform才是真正做题。如果你让验证集也参与划重点就等于提前把答案漏给了验证集离线分数再高都不能信。2.2 训练循环亲手写完一遍才知道哪里会炸很多人只用model.fit()但工程发布之后你需要精确控制梯度怎么清零、什么时候更新参数、什么时候关掉梯度计算、怎么提前停止。下面是一个标准的PyTorch训练循环骨架我建议你亲手敲一遍哪怕最后依然用高级封装也要能一眼看懂它import torch import torch.nn as nn def train_one_epoch(model, dataloader, optimizer, criterion, device): model.train() total_loss 0.0 for x_batch, y_batch in dataloader: x_batch x_batch.to(device) y_batch y_batch.to(device) optimizer.zero_grad() logits model(x_batch) loss criterion(logits, y_batch) loss.backward() optimizer.step() total_loss loss.item() return total_loss / len(dataloader) def validate(model, dataloader, criterion, device): model.eval() total_loss 0.0 correct 0 total 0 with torch.no_grad(): for x_batch, y_batch in dataloader: x_batch x_batch.to(device) y_batch y_batch.to(device) logits model(x_batch) total_loss criterion(logits, y_batch).item() preds logits.argmax(dim1) correct (preds y_batch).sum().item() total y_batch.size(0) return total_loss / len(dataloader), correct / total这里每一行都有讲究。model.train()和model.eval()影响Dropout和BatchNorm的行为忘了切换会出非常隐蔽的问题torch.no_grad()能省显存和计算optimizer.zero_grad()如果漏了梯度会在多个batch之间累加loss曲线会出现莫名其妙的大幅波动。超参方面我用Adam时习惯从3e-4起步batch size设为32或64。如果loss出现NaN先把学习率降一个数量级试试如果验证loss连续多轮不降就加载之前保存的最佳checkpoint。2.3 评估指标离线分数漂亮不等于业务可用准确率是最直观也最容易骗人的指标。类别不平衡时比如正负样本比例1:99全预测成多数类准确率都有99%。工程上要把指标选择跟业务成本绑定下面这个表是我常用的一张速查表场景关心的问题推荐指标搜索排序排序质量NDCG、MRR广告点击正样本稀少AUC、LogLoss欺诈识别漏过一个代价大召回率、F1邮件过滤误杀正常邮件代价大精确率时长预测偏移绝对值很重要MAE、MAPE选好指标之后还要学会看混淆矩阵而不是只盯一个数。比如垃圾邮件分类如果你更怕误杀正常邮件那么就不该无脑选AUC最高的模型而是要看在阈值调整后正常邮件被拦截的比例能不能接受。评估这件事本质是在回答一个问题模型上线后业务方会怎么骂你。3. 实操落地从数据到接口一步步把服务搭起来3.1 环境准备与依赖锁版这个项目我用Python 3.11核心依赖就下面这些全部锁死版本fastapi0.110.0 uvicorn0.29.0 pydantic2.7.0 torch2.2.0 pandas2.2.0 numpy1.26.4 scikit-learn1.4.0为什么必须锁版本因为AI工程的复现难度已经够高了Python依赖哪怕稍微漂移一点可能都会跑出完全不同的结果。踩过这个坑的人会明白pip install一时爽复现的时候火葬场。创建虚拟环境python3.11 -m venv .venv source .venv/bin/activate pip install -r requirements.txt这里提醒一句项目里所有代码在纯CPU环境都能跑完因为示例模型非常小。别让“没有GPU”成为你不动手的原因。3.2 数据管道与训练脚本从CSV到checkpoint我选了一个很小的文本情感分类任务来演示目标是让“中文影评是正面还是负面”。不需要加载大型预训练模型手工就能跑通全流程。先做数据预处理构建词表、编码序列、做padding。这里刻意绕开huggingface的tokenizer就是让你看清底层的逻辑import pandas as pd from collections import Counter from sklearn.model_selection import train_test_split df pd.read_csv(reviews.csv) # 建立词表 counter Counter() for text in df[review]: counter.update(text.lower().split()) # 留两个位置给 pad 和 unk vocab {word: i 2 for i, (word, _) in enumerate(counter.items()) if i 9999} vocab[pad] 0 vocab[unk] 1 def encode(text, max_len64): tokens [vocab.get(w, 1) for w in text.lower().split()[:max_len]] tokens [0] * (max_len - len(tokens)) return tokens df[encoded] df[review].apply(encode) X_train, X_val, y_train, y_val train_test_split( df[encoded].tolist(), df[label].tolist(), test_size0.2, random_state42 )接下来是Dataset和DataLoader。这层封装主要解决“怎么从数据到模型”的对接问题import torch from torch.utils.data import Dataset, DataLoader class ReviewDataset(Dataset): def __init__(self, texts, labels): self.texts texts self.labels labels def __len__(self): return len(self.texts) def __getitem__(self, idx): return torch.tensor(self.texts[idx], dtypetorch.long), torch.tensor(self.labels[idx], dtypetorch.long) train_loader DataLoader(ReviewDataset(X_train, y_train), batch_size64, shuffleTrue) val_loader DataLoader(ReviewDataset(X_val, y_val), batch_size64, shuffleFalse)模型我选了一个非常朴素的方案词向量取平均再接一个线性层。简单、快、对演示足够import torch.nn as nn class SentimentClassifier(nn.Module): def __init__(self, vocab_size, embed_dim64, num_classes2): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) self.fc nn.Linear(embed_dim, num_classes) def forward(self, x): embedded self.embedding(x).mean(dim1) return self.fc(embedded)训练循环直接复用前面写的train_one_epoch和validate跑8轮然后保存checkpoint。注意把vocab也一起存起来否则推理时没法做编码model SentimentClassifier(len(vocab)) optimizer torch.optim.Adam(model.parameters(), lr3e-4) criterion nn.CrossEntropyLoss() device torch.device(cpu) for epoch in range(8): train_loss train_one_epoch(model, train_loader, optimizer, criterion, device) val_loss, val_acc validate(model, val_loader, criterion, device) print(fepoch {epoch}: train_loss{train_loss:.4f}, val_loss{val_loss:.4f}, val_acc{val_acc:.4f}) torch.save({model: model.state_dict(), vocab: vocab}, sentiment_model.pt)到这里你已经完成了一条最简训练链路。整个过程不需要外部平台一个脚本跑到底。3.3 用FastAPI包成线上服务模型本身不产生价值能被业务调用才有价值。我直接用FastAPI写了一个推理服务import torch from fastapi import FastAPI from pydantic import BaseModel app FastAPI() ckpt torch.load(sentiment_model.pt, map_locationcpu) model SentimentClassifier(len(ckpt[vocab])) model.load_state_dict(ckpt[model]) model.eval() vocab ckpt[vocab] class ReviewIn(BaseModel): text: str app.get(/health) def health(): return {status: ok} app.post(/predict) def predict(payload: ReviewIn): tokens encode(payload.text, vocab) with torch.no_grad(): logits model(torch.tensor([tokens])) prob torch.softmax(logits, dim1).squeeze().tolist() return {negative: prob[0], positive: prob[1]}这里有两个容易被忽略的点。第一接口返回的是概率而不是硬标签这样下游业务可以根据场景调整阈值而不是被模型固定的阈值绑死。第二服务启动后一定要保持model.eval()状态否则Dropout在推理时依然随机同一条文本可能返回不同结果这种bug极其隐蔽。启动并调用uvicorn app:app --host 0.0.0.0 --port 8000curl -X POST http://127.0.0.1:8000/predict \ -H Content-Type: application/json \ -d {text: the movie is really bad}看到返回的概率分布说明你已经完成了从模型到接口的最后一跳。3.4 并发与性能接口能跑还不够很多第一次写推理服务的人只要curl通一次就觉得完工了。真正的考验是并发。我写了一个简单的压测脚本import time import requests import concurrent.futures def call(text): r requests.post(http://127.0.0.1:8000/predict, json{text: text}) return r.status_code texts [good movie] * 200 start time.time() with concurrent.futures.ThreadPoolExecutor(max_workers20) as ex: list(ex.map(call, texts)) print(QPS:, 200 / (time.time() - start))在这个小模型场景下单机跑出几百QPS很正常。但你真正要关心的是P99延迟也就是最慢的那一部分请求有多慢。FastAPI对于同步def接口会丢到线程池处理放在这个规模完全够用。我的建议是先测出基线再决定要不要上缓存、消息队列、批处理。不要一上来就堆Redis和Kafka那是拿架构复杂度掩盖对问题本质的不理解。4. 上线后的实战常见问题与排错技巧4.1 数据泄漏高指标的幻觉我做推荐系统时踩过最深的坑就是数据泄漏。当时用随机切分离线AUC能到0.92结果上线一测真实用户转化率几乎没变。后来排查发现训练数据里包含了一个“是否点击过这个商品”的特征而这个特征是实时生成的放到历史样本里本身就是信息泄漏。重新按时间切分把这类实时特征剔除后AUC掉到0.78但线上效果反而变好了。排错方法其实不复杂。第一重新按业务规则切分数据如果分数大幅下降说明原来的切分很可能有问题。第二检查特征重要度排行榜如果出现明显和未来相关的字段比如“本单是否已被退款”基本可以确定泄漏。第三做一个时间维度的盲测用旧数据训练、新数据验证如果分数崩了说明模型学到的是历史惯性而不是泛化规律。注意数据泄漏不会让模型“不能用”它只会让你误以为模型很好。它骗的是你的判断而不是线上流量。4.2 训练不收敛与显存崩掉的修复顺序训练过程中最常见的两个报错一个是loss为NaN一个是CUDA out of memory。我修复的顺序永远是固定的先打印每一步的loss找到loss变成NaN的起点大多发生在第二步或第三步把学习率降一个数量级比如从3e-4降到3e-5八成问题能解决如果还是不稳定把batch size减半同时可以考虑梯度裁剪如果是显存OOM先看nvidia-smi确认是显存被打满还是代码里缓存没释放如果确实需要更大batch但显存不够可以用梯度累积模拟大batchfor i, (x, y) in enumerate(train_loader): loss compute_loss(model, x, y) loss loss / accum_steps loss.backward() if (i 1) % accum_steps 0: optimizer.step() optimizer.zero_grad()梯度累积的思路是小步多次攒够了再更新参数本质上等价于更大的batch。它解决的是“单次显存放不下”的问题但要注意如果显存是因为中间激活值爆炸减输入长度或输入尺寸往往更有效。4.3 线上效果衰减从训练到服务的全链路排查模型上线后效果变差是每一个AI工程师迟早要面对的事。这时候不要慌按下面这个表逐项排查症状可能原因排查动作线上精度比离线低1-3个点线上和训练的预处理不一致取一条线上请求重放训练管道比对特征值效果随时间缓慢下降数据分布漂移按天聚合预测概率分布看是否偏移某个时段突然崩上游字段变更、数据缺失检查日志对比请求缺失字段新版本回滚后恢复新模型本身有问题对模型A/B对比不要全量切排查的前提是有日志。我每次给predict接口加日志时都会记录时间戳、输入文本的hash、预测概率。这样出了问题至少能回到某一天、某一条请求去复现import logging import json import time logger logging.getLogger(prediction_log) # predict接口内 logger.info(json.dumps({ ts: time.time(), text_hash: hash(payload.text), prob_positive: prob[1], }))别小看这几行代码。很多线上问题靠肉眼和记忆是永远定位不到的但有了分布日志你可以在半小时内画出趋势图找到拐点对应的发布事件。这是工程上性价比最高的监控手段。5. 越做越顺的进阶建议和个人复盘5.1 闭环之后先别急着上大集群一个人或者一个小团队跑通闭环后最容易犯的错是迅速引入大量基础设施。我见过有人刚跑通一个demo就开始搭K8s、搭Airflow、搭Feature Store结果光运维就把精力吃光了。我的建议是先加实验追踪比如用MLflow或者自己写一个CSV记录把每次实验的数据版本、模型参数、指标存下来然后再加自动重训任务让模型能定期基于新数据重新训练。只有当单机确实撑不住时再考虑分布式训练。这是需求倒推架构而不是为了显得专业而堆组件。5.2 给后来者的实在建议如果你问我这个项目从头到尾跟一遍到底值不值我会先说这样一句话如果你只想快速交差调现成API就够了如果你想在问题发生时不用两眼一抹黑亲手搭一次是值得的。我自己做这个项目最大的收获其实不是模型精度而是把“AI工程”从一个模糊的名词变成了手里一条条可见的、可测的、可回滚的链路。最后再分享一个小技巧每次往系统里加新组件之前先问自己一句——没有它我现在这个系统会死吗不会就先别加。这句话帮我挡住了无数过度设计也是我从这个项目里带走的、最值钱的一条经验。