1. 从零搭建AI工程体系为什么我劝你别一上来就啃论文“ai-engineering-from-scratch”这个标题乍一看像是又一份“从入门到放弃”的学习路线图。但我第一次看到它的时候脑子里蹦出来的不是教程而是一个很现实的问题为什么很多人学了一堆理论连一个能跑起来的小系统都搭不出来我自己带过不少刚入行的朋友也见过太多人卡在同一个地方——他们能背出反向传播的公式能说清楚Transformer的注意力机制但真让他们从零搭一个能处理真实数据的AI工程流水线立刻就懵了。数据怎么清洗特征怎么存模型怎么部署推理延迟怎么压这些问题论文里不会告诉你网课里也只是一笔带过。所以这篇东西我想聊的不是“AI理论从零到一”而是AI工程能力从零到一。这两件事的差别就像“懂发动机原理”和“能造一台能上路的车”之间的差别。前者是科学后者是工程。而“ai-engineering-from-scratch”这个项目标题背后真正有价值的东西恰恰是工程侧的那套脏活累活。这篇文章适合谁看如果你是刚转行做AI的开发者或者已经会调库但想搞清楚底层怎么串起来的工程师再或者你是带团队的技术负责人想给新人设计一条靠谱的成长路径那接下来的内容应该能帮你省下不少试错时间。我会把整个从零搭建AI工程体系的过程拆成几个核心模块每个模块都讲清楚为什么这么设计、具体怎么操作、以及我踩过哪些坑。2. 整体设计思路为什么我选择“先跑通再优化”的路线2.1 从零开始的真正含义不是重造轮子而是理解轮子怎么转很多人对“from scratch”有误解觉得必须从矩阵乘法开始手写不能用任何现成框架。我一开始也这么想过后来发现这条路效率极低而且容易陷入“造了个还不如现成库好用的轮子”的尴尬。我的理解是从零开始的核心是理解每一层的输入输出和依赖关系而不是拒绝使用工具。你可以用PyTorch但你得知道张量在内存里怎么排布你可以用sklearn但你得清楚fit和transform之间到底发生了什么你可以用Docker但你得明白镜像分层是怎么影响构建速度的。所以我的整体设计思路是分三层推进第一层数据管道。这是最容易被忽视但最重要的一层。我见过太多项目死在数据上而不是模型上。这一层的目标是让数据从原始状态变成模型能吃的格式并且这个过程要可复现、可监控。第二层模型训练与实验管理。这一层不是调参而是建立一套能追踪每次实验的机制。你改了哪个超参数、用了哪版数据、结果指标是多少这些信息必须自动记录否则一周后你根本记不清哪个模型是最好的。第三层推理服务与监控。模型训练完只是开始怎么把它变成别人能调用的接口怎么保证它在生产环境里不崩怎么发现数据漂移这些才是工程能力的试金石。这三层的顺序不能乱。我试过先搞模型再补数据管道结果就是反复返工因为数据格式一变训练代码就得重写。所以“先跑通再优化”的意思是先用最简单的方式把三层串起来哪怕数据只有100条、模型只是个逻辑回归、服务只是个Flask接口只要端到端能跑你就有了一个可以迭代的骨架。2.2 技术选型的取舍逻辑为什么我没用最火的那些工具说到具体工具市面上选择太多了。我最终定下来的组合是Pandas PyTorch FastAPI MLflow Docker。这个组合不是最潮的但我觉得是最适合“从零搭建”这个场景的。先说数据处理。有人推荐用Spark理由是“大数据”。但如果你手头的数据还没到TB级别Spark的运维成本远大于收益。Pandas单机处理千万级数据完全够用而且调试方便。我试过用Dask做分布式结果发现大部分时间花在解决分区和序列化问题上真正算数据的时间反而少了。所以我的建议是数据量没到内存装不下的程度就老老实实用Pandas。模型框架选PyTorch而不是TensorFlow理由很简单调试体验好。PyTorch的动态图让你可以在任意位置打断点看张量的值。TensorFlow 2.x虽然也支持动态图但生态里很多老代码还是静态图风格新手容易混淆。而且PyTorch的社区现在更活跃遇到问题更容易找到答案。服务框架用FastAPI而不是Flask主要是因为异步支持和自动文档。FastAPI的async/await能让推理服务在等待IO时处理更多请求而且它自动生成的Swagger文档省去了写接口文档的时间。Flask虽然更简单但当你需要处理并发请求时就得自己加gunicorn和gevent配置起来反而更麻烦。实验管理用MLflow而不是Weights Biases是因为MLflow可以本地部署数据留在自己手里。WB的云端界面确实好看但免费版有项目数量限制而且网络不稳定的时候同步会失败。MLflow虽然UI朴素但该有的功能都有而且和PyTorch集成很简单。Docker不用多说保证环境一致性。但我建议新手不要一上来就搞Kubernetes那东西的学习曲线太陡。先用Docker Compose把几个服务串起来等真正需要弹性伸缩的时候再考虑K8s。2.3 目录结构设计让项目自己告诉你怎么跑我见过很多项目代码全堆在一个目录里文件名从test1.py到test_final_v2.py。这种项目别说别人自己过两周都看不懂。所以从零搭建的时候目录结构必须一开始就定好。我的习惯是这样的project/ ├── data/ │ ├── raw/ # 原始数据只读 │ ├── processed/ # 清洗后的数据 │ └── features/ # 特征工程后的数据 ├── src/ │ ├── data/ # 数据加载和清洗脚本 │ ├── features/ # 特征工程脚本 │ ├── models/ # 模型定义和训练脚本 │ ├── serving/ # 推理服务代码 │ └── utils/ # 通用工具函数 ├── configs/ # 配置文件 ├── experiments/ # 实验记录 ├── tests/ # 单元测试 ├── docker/ # Dockerfile和compose文件 ├── requirements.txt └── README.md这个结构的关键在于数据分层和代码分模块。data/raw永远不动所有清洗逻辑写在src/data里输出到data/processed。这样当原始数据更新时你只需要重新跑清洗脚本而不是手动改文件。configs目录放YAML文件把超参数、路径、模型版本都抽出来代码里不写硬编码。experiments目录让MLflow自动写入每次实验一个子目录里面存模型文件和指标。注意不要把所有东西都塞进Jupyter Notebook。Notebook适合探索不适合工程。我的做法是先在Notebook里试试通了立刻把代码抽成.py文件Notebook只保留探索记录。3. 核心细节解析数据管道、训练循环和推理服务的实操要点3.1 数据管道从CSV到Tensor的完整链路数据管道是整个项目的地基。我见过太多人直接pd.read_csv然后train_test_split就开跑结果上线后发现线上数据的分布和训练数据完全不一样。问题出在哪儿出在没有把数据处理的每一步都固化下来。我的做法是写一个DataProcessor类把清洗、特征工程、划分都封装成方法。关键点是fit和transform必须分开。比如标准化你必须在训练集上计算均值和方差然后应用到验证集和测试集。如果对整个数据集做标准化就会造成数据泄露。class DataProcessor: def __init__(self): self.scaler StandardScaler() self.encoders {} def fit(self, df): # 只在训练集上计算统计量 self.scaler.fit(df[numeric_cols]) for col in categorical_cols: self.encoders[col] LabelEncoder().fit(df[col]) return self def transform(self, df): df df.copy() df[numeric_cols] self.scaler.transform(df[numeric_cols]) for col in categorical_cols: df[col] self.encoders[col].transform(df[col]) return df这个模式的好处是你可以把fit好的processor用joblib.dump存下来推理服务加载同一个processor保证线上线下处理逻辑一致。我踩过的坑是有一次忘了存encoder线上遇到训练时没见过的类别直接报错。后来我在transform里加了handle_unknownignore遇到新类别就映射到“未知”类别。另一个重点是数据版本控制。每次清洗后的数据我都会在文件名里加上时间戳和哈希值比如processed_20240501_a3f2c1.parquet。这样当模型效果下降时我可以回溯到具体是哪版数据训练的。用Parquet而不是CSV是因为Parquet有schema、压缩率高、读取快。实测下来同样数据量Parquet比CSV小60%左右读取速度快3到5倍。3.2 训练循环别让实验变成玄学训练循环看起来简单不就是forward、backward、step吗但真正做工程的时候你需要考虑的东西多得多学习率调度、梯度裁剪、早停、检查点保存、指标记录。这些如果每次都手写不仅容易出错而且不同实验之间没法对比。我的做法是写一个Trainer类把训练逻辑封装起来通过配置文件控制行为。关键设计是回调机制每个epoch结束时依次调用注册的回调函数比如EarlyStopping、ModelCheckpoint、MLflowLogger。这样加新功能只需要写一个新回调不用改训练循环。class Trainer: def __init__(self, model, optimizer, scheduler, callbacks): self.model model self.optimizer optimizer self.scheduler scheduler self.callbacks callbacks def train(self, train_loader, val_loader, epochs): for epoch in range(epochs): self.model.train() for batch in train_loader: loss self._train_step(batch) val_loss self._validate(val_loader) for cb in self.callbacks: cb.on_epoch_end(epoch, val_loss, self.model)这里有个细节验证集的loss计算必须用torch.no_grad()包起来否则会构建计算图显存直接爆掉。我一开始忘了加结果训练到一半OOM排查了半天才发现是验证阶段的问题。另一个经验是梯度裁剪。当使用RNN或Transformer时梯度爆炸很常见。我习惯在backward之后、step之前加torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)。这个max_norm的值不是拍脑袋定的我一般会先跑几个batch打印梯度范数然后取中位数的2到3倍作为阈值。学习率调度也很关键。我试过固定学习率结果要么收敛慢要么在最优解附近震荡。后来改用CosineAnnealingWarmRestarts效果稳定很多。它的逻辑是学习率按余弦曲线下降然后周期性重启每次重启后的峰值学习率略低于上一次。这样既能快速下降又能跳出局部最优。3.3 推理服务从模型文件到API接口模型训练完下一步是让别人能用。最直接的方式是写个Flask接口但生产环境要考虑的东西更多并发、超时、批处理、版本管理。我用FastAPI搭服务核心代码大概长这样from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model None processor None class PredictRequest(BaseModel): features: list app.on_event(startup) def load_model(): global model, processor model torch.load(experiments/best_model.pt, map_locationcpu) model.eval() processor joblib.load(experiments/processor.pkl) app.post(/predict) async def predict(request: PredictRequest): df pd.DataFrame([request.features], columnsfeature_names) X processor.transform(df) with torch.no_grad(): tensor torch.tensor(X.values, dtypetorch.float32) output model(tensor) return {prediction: output.item()}这里有几个关键点。第一模型加载放在startup事件里而不是每次请求都加载。我见过有人在接口里torch.load结果每个请求都要读几百MB的文件延迟直接上秒级。第二用model.eval()切换到推理模式这会影响Dropout和BatchNorm的行为。第三用torch.no_grad()关闭梯度计算减少显存占用和计算量。批处理是另一个优化点。如果请求量很大逐个推理效率很低。我的做法是加一个缓冲队列攒够一定数量或者等待超过10毫秒就批量推理。实测下来批量大小为32时吞吐量比单条推理高8到10倍。但批处理会增加延迟所以要根据业务场景权衡。如果是实时性要求高的场景批量大小就设小一点比如8。提示推理服务的输入验证不能省。我遇到过请求里传了字符串结果torch.tensor直接报错。后来用Pydantic的validator做了类型检查把错误在入口就拦住了。4. 实操过程从空目录到可运行系统的完整记录4.1 环境准备与依赖管理第一步是建目录、装依赖。我习惯用conda创建虚拟环境因为可以指定Python版本而且科学计算库的依赖解析比pip更稳。conda create -n ai-eng python3.10 conda activate ai-eng pip install torch pandas scikit-learn fastapi uvicorn mlflow joblib pyarrow这里有个细节PyTorch的版本要和CUDA版本匹配。如果你有GPU先去PyTorch官网查对应命令。我试过直接pip install torch结果装的是CPU版训练慢得想哭。后来用conda install pytorch torchvision torchaudio pytorch-cuda11.8 -c pytorch -c nvidia才搞定。依赖管理我推荐用pip-tools。先写requirements.in只写直接依赖然后pip-compile生成requirements.txt把所有间接依赖的版本都锁死。这样别人克隆你的项目pip install -r requirements.txt能装出一模一样的环境。我踩过的坑是有一次没锁版本过了一个月重新装numpy从1.24升到1.26结果某个API变了代码直接跑不起来。4.2 数据清洗与特征工程的具体操作假设我们有一个电商用户行为数据集目标是预测用户是否会复购。原始数据大概长这样user_idagecitylast_purchase_daystotal_ordersavg_amountrepurchase100125北京35120.51100234上海45280.00100328广州128200.31清洗步骤我一般分四步走缺失值处理。先统计每列缺失比例。如果缺失超过50%直接删列。如果缺失在5%到50%之间数值列用中位数填充类别列用众数填充。如果缺失小于5%可以考虑用模型预测填充但大多数时候中位数就够了。异常值处理。对数值列计算IQR超出Q1 - 1.5*IQR和Q3 1.5*IQR范围的值我一般用截断而不是删除。因为删除会损失信息截断保留了“这个值很大”的信号。类别编码。城市这种低基数类别用One-Hot用户ID这种高基数类别用Target Encoding。Target Encoding要注意用交叉验证的方式计算否则会泄露标签。特征交叉。比如total_orders / last_purchase_days可以表示购买频率avg_amount * total_orders可以表示总消费额。这些交叉特征往往比原始特征更有预测力。特征工程做完后我会用sklearn.feature_selection.SelectKBest做一轮筛选保留F值最高的20个特征。实测下来特征从50个降到20个模型AUC只掉了0.002但训练速度快了一倍。4.3 模型训练与超参数搜索模型我选的是XGBoost因为它在表格数据上表现稳定而且训练速度快。虽然标题是AI工程但工程的核心是解决问题不是非要用深度学习。XGBoost调参相对直观而且特征重要性可以直接看。超参数搜索我用Optuna比GridSearchCV高效得多。GridSearch是穷举Optuna是贝叶斯优化能在更少的尝试次数内找到更好的参数组合。import optuna def objective(trial): params { n_estimators: trial.suggest_int(n_estimators, 100, 1000), max_depth: trial.suggest_int(max_depth, 3, 10), learning_rate: trial.suggest_float(learning_rate, 0.01, 0.3, logTrue), subsample: trial.suggest_float(subsample, 0.6, 1.0), colsample_bytree: trial.suggest_float(colsample_bytree, 0.6, 1.0), } model XGBClassifier(**params) scores cross_val_score(model, X_train, y_train, cv5, scoringroc_auc) return scores.mean() study optuna.create_study(directionmaximize) study.optimize(objective, n_trials50)50次试验大概跑了20分钟最佳AUC是0.87。这里有个经验学习率用log均匀分布因为0.01和0.02的差别比0.2和0.21的差别重要得多。另外n_estimators和learning_rate要一起调高学习率配少树低学习率配多树Optuna会自动找到这个平衡。训练完成后我用MLflow记录参数和指标import mlflow with mlflow.start_run(): mlflow.log_params(study.best_params) mlflow.log_metric(auc, study.best_value) mlflow.sklearn.log_model(model, model)MLflow会自动在experiments/目录下创建运行记录包含模型文件、参数、指标。以后想对比不同实验直接mlflow ui打开界面就能看。4.4 服务部署与性能压测服务用Docker打包Dockerfile大概这样FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY experiments/best_model.pkl ./experiments/ COPY experiments/processor.pkl ./experiments/ EXPOSE 8000 CMD [uvicorn, src.serving.main:app, --host, 0.0.0.0, --port, 8000]构建镜像时有个技巧把requirements.txt的复制和安装放在最前面。因为Docker是分层构建只要requirements.txt没变这一层就会用缓存不用重新装依赖。我一开始把COPY . .放在最前面结果每次改代码都要重装一遍依赖构建时间从10秒变成5分钟。压测用locust模拟100个并发用户每个用户每秒发10个请求。结果发现P99延迟是120毫秒但P50只有15毫秒。说明大部分请求很快但少数请求很慢。排查后发现是批处理队列的等待时间导致的。把批量大小从32降到8P99降到45毫秒吞吐量只下降了15%。这个权衡是值得的因为用户体验主要受尾部延迟影响。注意压测时不要用本机跑服务又跑压测工具两者会抢CPU。我试过在同一台机器上跑结果数据完全不可信。后来用两台机器一台跑服务一台跑locust数据才稳定。5. 常见问题与排查技巧实录5.1 数据管道常见报错与解决问题一SettingWithCopyWarning。这是Pandas最常见的警告原因是你在切片上做修改Pandas不确定你是想改原数据还是副本。解决方法是用.copy()显式创建副本或者用.loc做赋值。问题二内存不足。当数据量超过内存时pd.read_csv会直接崩。我的做法是分块读取pd.read_csv(file.csv, chunksize100000)每次处理一块处理完就释放。或者用dtype参数指定更小的类型比如把int64改成int32内存直接减半。问题三类别编码不一致。训练时用LabelEncoder把“北京”编成0“上海”编成1。推理时遇到“广州”LabelEncoder没见过直接报错。解决方法是用handle_unknownignore或者用One-Hot新类别全编0。5.2 训练过程中的典型故障故障一Loss变成NaN。原因通常是学习率太大或者数据里有异常值。我的排查步骤是先打印每个batch的loss找到变NaN的位置然后检查那个batch的数据看有没有inf或超大值最后降低学习率加梯度裁剪。故障二验证集指标比训练集好。这听起来是好事但通常意味着数据泄露。检查一下是不是在划分数据集之前就做了标准化或者用了未来数据。我踩过的坑是用时间序列数据时随机划分导致未来数据泄露到训练集。后来改成按时间划分前80%训练后20%验证。故障三GPU利用率低。nvidia-smi显示GPU利用率只有20%说明数据加载是瓶颈。解决方法是把num_workers调大用pin_memoryTrue或者把数据预处理提前做好存成二进制格式。5.3 推理服务的性能瓶颈定位推理服务变慢排查顺序一般是网络、CPU、内存、模型。先看网络用curl -w curl-format.txt测一下请求的各个阶段耗时。如果DNS解析或TCP连接耗时长那是网络问题。如果服务端处理耗时长继续往下查。然后看CPU用top或htop看CPU使用率。如果CPU跑满说明计算是瓶颈。这时候可以考虑量化模型把float32转成int8推理速度能提升2到4倍精度损失通常在1%以内。再看内存用free -h看内存使用。如果内存持续增长说明有内存泄漏。常见原因是全局变量累积比如把每次请求的数据都append到一个列表里。解决方法是定期重启服务或者用gc.collect()手动回收。最后看模型本身。如果模型太大加载和推理都慢。可以考虑用ONNX Runtime它比原生PyTorch推理快20%到30%。转换方法是用torch.onnx.export导出ONNX模型然后用onnxruntime.InferenceSession加载。5.4 常见问题速查表问题现象可能原因排查方法解决方案训练loss不下降学习率太小、数据未归一化打印梯度范数、检查数据分布调大学习率、加BatchNorm验证loss上升过拟合对比训练和验证曲线加Dropout、早停、数据增强推理延迟高模型太大、批处理等待用profiler看各阶段耗时量化模型、调小批量服务OOM内存泄漏、并发太高监控内存曲线限制并发、定期重启预测结果全一样模型未训练好、输入未处理检查模型输出、对比训练输入重新训练、检查processor6. 一些让我少走弯路的实操心得第一个心得是日志比print好用。我一开始到处用print结果服务跑起来后日志和输出混在一起根本看不清。后来改用logging模块设置不同的日志级别INFO记录正常流程ERROR记录异常DEBUG记录详细数据。而且日志可以输出到文件方便事后排查。第二个心得是配置文件用YAML不用JSON。YAML支持注释可以写# 这是学习率JSON不行。而且YAML的层级结构更直观不用写一堆大括号。我用hydra管理配置可以在命令行覆盖参数比如python train.py model.lr0.01不用改文件。第三个心得是单元测试要覆盖数据管道。模型效果不好很多时候是数据问题。我写了一个test_data_processor.py测试fit和transform的输出形状、数据类型、缺失值处理。每次改数据代码先跑测试通过了再训练。这样能避免“改了数据代码模型效果下降但不知道是模型问题还是数据问题”的情况。第四个心得是模型版本要和数据版本绑定。我在MLflow里记录模型时会把数据文件的哈希值也记进去。这样当模型效果下降时我可以查到这个模型是用哪版数据训练的然后对比当前数据分布判断是不是数据漂移。第五个心得是不要追求一步到位。我见过有人一开始就想搞微服务、Kubernetes、特征存储结果三个月过去了连个能跑的demo都没有。我的建议是先用最简单的方式跑通然后根据实际瓶颈逐步优化。瓶颈没出现之前不要提前优化。这个项目后续还可以这样扩展加一个特征存储层把特征工程的结果缓存起来避免每次训练都重新计算加一个A/B测试框架对比不同模型在线上流量的表现加一个监控面板实时展示推理延迟、QPS、错误率。但这些都是后话先把基础的三层跑通比什么都重要。