1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都在教你调API、跑现成框架、微调个模型就完事。真正愿意从零开始把AI工程当成一门手艺来拆解的内容少得可怜。我自己在这个行当里摸爬滚打了十来年从最早用脚本跑机器学习任务到后来带团队做推理服务、搞特征平台、搭训练流水线踩过的坑能写满三个笔记本。所以看到这个标题我特别有共鸣——它想做的事情是把AI工程从“魔法”还原成“工程”。这个项目适合谁如果你已经会写Python能看懂基本的矩阵运算但对“一个模型从数据到上线到底经历了什么”没有完整概念那它非常适合你。如果你已经会调model.fit()和model.predict()但说不清楚梯度是怎么算的、显存是怎么分配的、推理延迟为什么忽高忽低那你也应该看看。它解决的核心问题是把AI系统拆开让你看到每一个齿轮是怎么咬合的而不是只给你一辆能跑的车却不告诉你发动机在哪。我打算按照一个真实的AI工程项目从零到一的过程来展开把“ai-engineering-from-scratch”这个标题背后的东西掰开揉碎。不堆公式不抄文档就讲一个从业者真正会关心的事情每一步为什么这么做不这么做会怎样以及我踩过哪些坑。2. 整体设计思路为什么“从零”不等于“重复造轮子”2.1 先想清楚你要的到底是“懂原理”还是“能干活”很多人对“from scratch”有误解觉得必须手写一个Transformer才算数。我见过不少朋友花了两周用NumPy实现了一个多层感知机跑通了MNIST然后呢然后就没有然后了。因为真实项目里没人会让你用NumPy去训一个十亿参数的模型。所以第一步要明确从零搭建AI工程体系核心目标不是替代PyTorch而是理解PyTorch在替你做什么以及当它出问题的时候你知道去哪里找原因。我的建议是分两条线走。一条线是“原理线”用最小的代码量实现核心算法比如用几十行代码写一个反向传播用几百行写一个简单的训练循环。另一条线是“工程线”用主流框架搭建可复现、可扩展、可监控的训练和推理流程。两条线交替推进原理线帮你建立直觉工程线帮你积累手感。只走原理线容易变成学院派只走工程线遇到诡异bug就只能靠玄学。2.2 技术选型的底层逻辑为什么是Python PyTorch FastAPI这个组合不是拍脑袋定的。Python是AI领域的通用语言生态最全没有之一。PyTorch的动态图机制对调试极其友好你可以在任意位置打断点看张量的形状和数值这对“从零”阶段至关重要。FastAPI则是目前把模型包装成服务最顺手的选择异步支持好自动生成文档性能也够用。但我要强调一点选型不是一成不变的。如果你做的是边缘端部署可能要考虑ONNX Runtime或者TensorRT如果你做的是大规模分布式训练可能要上DeepSpeed或者Megatron。但作为“从零”的起点PyTorch FastAPI是最小可行组合能让你把注意力集中在工程逻辑上而不是被框架的复杂性淹没。2.3 项目目录结构一开始就养成好习惯我见过太多项目所有代码堆在一个main.py里到后面连自己都找不到东西。从零搭建的时候就要把目录结构定好。下面是我用了很多年的一个模板你可以直接抄ai-project/ ├── configs/ # 配置文件YAML或JSON ├── data/ # 数据相关原始数据、处理脚本、缓存 │ ├── raw/ │ ├── processed/ │ └── datasets.py ├── models/ # 模型定义 │ ├── __init__.py │ └── mlp.py ├── training/ # 训练相关 │ ├── trainer.py │ └── losses.py ├── serving/ # 推理服务 │ ├── app.py │ └── schemas.py ├── utils/ # 工具函数 │ ├── logger.py │ └── metrics.py ├── experiments/ # 实验记录每次训练的配置和结果 ├── requirements.txt └── README.md这个结构的好处是数据、模型、训练、服务各司其职。当你需要改数据预处理的时候不会误触模型代码当你需要换推理框架的时候不会影响训练逻辑。而且experiments/目录特别重要每次训练把配置和关键指标存下来后面复现结果的时候能省你无数时间。3. 核心细节解析数据、模型、训练、推理的四个关键环节3.1 数据管道别让脏数据毁了你的一切数据是AI工程的基石但也是最容易被忽视的环节。我见过太多项目模型结构调了又调最后发现是数据里有重复样本或者标签错误。从零搭建的时候数据管道要解决三个问题加载、预处理、版本管理。加载方面小数据直接用pandas或numpy读进来就行但一旦数据超过内存就要考虑流式加载或者内存映射。PyTorch的Dataset和DataLoader是标准做法但要注意num_workers的设置。我实测下来在Linux环境下num_workers设为CPU核心数的70%左右比较稳设太高反而会因为进程切换开销导致速度下降。预处理方面最关键的是把预处理逻辑固化下来。什么意思就是训练时用的归一化参数、分词器、图像增强策略在推理时必须完全一致。我踩过的坑是训练时用了某个版本的torchvision.transforms推理时环境里装的是另一个版本结果预处理行为有细微差异导致线上效果掉了一大截。解决办法是把预处理逻辑封装成独立的模块训练和推理共用同一份代码并且把关键参数比如均值、方差存到配置文件里。版本管理方面数据不像代码改一行能看出diff。我的做法是给每个数据集算一个哈希值存在experiments/里。如果数据变了哈希值就变了训练结果不可复现的时候就能快速定位是不是数据的问题。注意数据泄露是新手最容易犯的错误。比如在做归一化的时候用了整个数据集的均值和方差而不是只用训练集的。这会导致模型在验证集上表现虚高上线后原形毕露。正确做法是只用训练集计算统计量然后应用到验证集和测试集。3.2 模型定义从线性层到Transformer关键是理解形状变化模型定义这部分我建议从最简单的线性回归开始然后逐步加层、加激活函数、加正则化。每加一个东西都要问自己这个操作改变了张量的什么形状怎么变参数量增加了多少举个例子一个全连接层nn.Linear(in_features, out_features)输入形状是(batch_size, in_features)输出形状是(batch_size, out_features)。参数量是in_features * out_features out_features。这些看起来很简单但当你堆叠多层、加入残差连接、加入注意力机制的时候形状变化就会变得复杂。我见过不少人写模型前向传播跑不通报错说维度不匹配然后就开始瞎试试了半天才试对。其实只要在纸上画一画形状变化五分钟就能解决。对于“ai-engineering-from-scratch”这个主题我特别推荐手写一次反向传播。不用写完整的自动微分就写一个两层网络的梯度计算。你会深刻理解为什么需要计算图为什么PyTorch要设计backward()以及为什么梯度会消失或爆炸。这个经验对你后面调参和排查问题有巨大帮助。3.3 训练循环损失函数、优化器、学习率调度训练循环是AI工程的心脏。一个标准的训练循环包含前向传播、计算损失、反向传播、更新参数。但真正写好一个训练循环要考虑的东西远不止这些。损失函数的选择取决于任务。分类用交叉熵回归用均方误差但实际场景中往往需要自定义损失。比如样本不平衡的时候要给少数类更高的权重比如多任务学习的时候要把多个损失加权求和。这些权重怎么定我的经验是先用默认值跑一遍看各个损失的量级然后调整权重让它们在同一量级附近。优化器方面Adam是默认选择但学习率需要调。我一般从1e-3开始如果损失震荡就降到1e-4如果下降太慢就升到3e-3。学习率调度也很重要CosineAnnealingLR和ReduceLROnPlateau是我用得最多的两种。前者适合训练轮数固定的场景后者适合不确定什么时候收敛的场景。训练循环里还要加的东西梯度裁剪、混合精度训练、检查点保存、日志记录。梯度裁剪防止梯度爆炸混合精度训练能省显存加速检查点保存让你能从断点恢复日志记录让你能分析训练过程。这些不是可选项是必选项。# 一个简化的训练循环示例 for epoch in range(num_epochs): model.train() for batch in train_loader: inputs, targets batch inputs, targets inputs.to(device), targets.to(device) optimizer.zero_grad() outputs model(inputs) loss criterion(outputs, targets) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() # 验证和保存检查点 validate_and_save(model, val_loader, epoch)3.4 推理服务从模型文件到API接口训练好的模型要变成服务才算真正产生价值。推理服务要考虑的事情和训练完全不同延迟、吞吐、并发、容错。延迟方面模型推理时间取决于模型大小、输入尺寸、硬件。我实测下来一个10M参数的小模型在CPU上单次推理大概几十毫秒在GPU上能降到几毫秒。但如果输入尺寸很大比如高分辨率图像预处理的时间可能比推理还长。所以优化推理延迟往往要先优化预处理。吞吐方面批处理是提高吞吐的关键。但批处理会增加延迟因为要等齐一个批次。所以要在延迟和吞吐之间找平衡。我的做法是设置一个最大批次和最大等待时间比如最多等10毫秒或者凑够32个请求就发车。并发方面FastAPI默认是单进程的要用uvicorn的workers参数启动多个进程。但GPU推理要注意多个进程同时访问同一块GPU会竞争显存和计算资源。这时候要么用模型并行要么用请求队列串行化。容错方面推理服务要能处理各种异常输入空输入、超长输入、格式错误的输入。我一般会在API层做严格的输入校验用Pydantic定义请求和响应的schema把不合法的请求挡在模型之外。4. 实操过程从零到一搭建一个完整的AI工程4.1 环境准备Python版本、CUDA、依赖管理环境准备是第一步也是最容易出问题的一步。Python版本我推荐3.9或3.10太新的版本有些库还没适配太旧的版本缺少一些特性。CUDA版本要和PyTorch版本匹配去PyTorch官网查对应关系别凭感觉装。依赖管理我强烈建议用conda或者venv创建独立环境别在系统Python里瞎装。requirements.txt要写清楚版本号别只写包名。我踩过的坑是本地开发环境是numpy 1.24服务器上是numpy 1.19结果某个函数的行为不一样排查了半天。# 创建环境 conda create -n ai-eng python3.10 conda activate ai-eng # 安装PyTorch根据CUDA版本选择 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118 # 安装其他依赖 pip install fastapi uvicorn pandas scikit-learn matplotlib4.2 数据准备与预处理以一个小型分类任务为例为了演示完整流程我拿一个经典的分类任务来举例用鸢尾花数据集训练一个分类器。虽然简单但麻雀虽小五脏俱全。首先加载数据划分训练集和验证集。注意划分之前要打乱数据否则如果数据是按类别排序的训练集和验证集的分布会不一致。from sklearn.datasets import load_iris from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler import numpy as np iris load_iris() X, y iris.data, iris.target # 划分数据集 X_train, X_val, y_train, y_val train_test_split( X, y, test_size0.2, random_state42, stratifyy ) # 标准化只用训练集的统计量 scaler StandardScaler() X_train scaler.fit_transform(X_train) X_val scaler.transform(X_val) # 保存scaler推理时要用 import joblib joblib.dump(scaler, experiments/scaler.pkl)这里的关键点是stratifyy保证训练集和验证集的类别比例一致。还有scaler的保存推理时必须用同一个scaler否则输入分布不一致模型效果会崩。4.3 模型搭建与训练手写一个两层网络接下来用PyTorch搭一个两层网络。输入是4维特征隐藏层16个神经元输出3个类别。import torch import torch.nn as nn class IrisNet(nn.Module): def __init__(self): super().__init__() self.fc1 nn.Linear(4, 16) self.relu nn.ReLU() self.fc2 nn.Linear(16, 3) def forward(self, x): x self.fc1(x) x self.relu(x) x self.fc2(x) return x model IrisNet() criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(model.parameters(), lr1e-3)训练循环里把数据转成Tensor用DataLoader分批。注意CrossEntropyLoss内部已经包含了Softmax所以模型输出不需要加Softmax。from torch.utils.data import TensorDataset, DataLoader train_dataset TensorDataset( torch.FloatTensor(X_train), torch.LongTensor(y_train) ) train_loader DataLoader(train_dataset, batch_size16, shuffleTrue) for epoch in range(100): model.train() total_loss 0 for inputs, targets in train_loader: optimizer.zero_grad() outputs model(inputs) loss criterion(outputs, targets) loss.backward() optimizer.step() total_loss loss.item() if (epoch 1) % 20 0: print(fEpoch {epoch1}, Loss: {total_loss/len(train_loader):.4f})训练完成后在验证集上评估保存模型权重和scaler。4.4 推理服务搭建FastAPI Uvicorn把训练好的模型包装成API。定义请求和响应的schema加载模型和scaler提供预测接口。from fastapi import FastAPI from pydantic import BaseModel import torch import joblib import numpy as np app FastAPI() class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): class_id: int class_name: str confidence: float # 加载模型和scaler model IrisNet() model.load_state_dict(torch.load(experiments/model.pth)) model.eval() scaler joblib.load(experiments/scaler.pkl) class_names [setosa, versicolor, virginica] app.post(/predict, response_modelPredictResponse) def predict(request: PredictRequest): features np.array(request.features).reshape(1, -1) features scaler.transform(features) tensor torch.FloatTensor(features) with torch.no_grad(): outputs model(tensor) probabilities torch.softmax(outputs, dim1) confidence, class_id torch.max(probabilities, dim1) return PredictResponse( class_idint(class_id), class_nameclass_names[int(class_id)], confidencefloat(confidence) )启动服务uvicorn serving.app:app --host 0.0.0.0 --port 8000。然后用curl或者Postman测试。curl -X POST http://localhost:8000/predict \ -H Content-Type: application/json \ -d {features: [5.1, 3.5, 1.4, 0.2]}4.5 实验记录与版本管理让每次训练都可追溯每次训练都要记录数据版本、代码版本、超参数、训练指标、验证指标。我一般用experiments/目录每次训练建一个子目录里面放config.yaml、metrics.json、model.pth。# experiments/exp_001/config.yaml data: path: data/raw/iris.csv hash: abc123 model: hidden_size: 16 training: batch_size: 16 learning_rate: 0.001 epochs: 100这样后面要复现某个结果直接看配置就行。如果发现某个实验效果特别好也能快速定位是哪个超参数起了作用。5. 常见问题与排查技巧实录5.1 训练不收敛从数据、模型、超参三个方向排查训练不收敛是最常见的问题。我的排查顺序是先看数据再看模型最后看超参。数据方面检查有没有NaN或Inf检查标签有没有越界检查输入分布是否合理。我遇到过一次数据里有个特征的方差特别大导致梯度爆炸训练直接发散。解决办法是做归一化。模型方面检查初始化。PyTorch默认的初始化一般没问题但如果你自定义了层可能初始化得不好。可以试试kaiming_normal_或者xavier_normal_。超参方面学习率是最关键的。学习率太大损失震荡甚至发散学习率太小收敛太慢。我一般会做一个学习率扫描从1e-5到1e-1每个跑几十步看损失下降情况。5.2 显存不够用梯度累积、混合精度、模型并行显存不够是训练大模型时的常见问题。解决办法有几个梯度累积用小的batch size累积多次梯度再更新。比如batch size设为8累积4次等效于batch size 32。混合精度用torch.cuda.amp前向传播用float16反向传播用float32。能省一半显存速度还能提升。模型并行把模型的不同层放到不同的GPU上。这个比较复杂一般用DataParallel或DistributedDataParallel。# 混合精度训练示例 from torch.cuda.amp import autocast, GradScaler scaler GradScaler() for inputs, targets in train_loader: optimizer.zero_grad() with autocast(): outputs model(inputs) loss criterion(outputs, targets) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()5.3 推理延迟高从预处理、模型、后处理三段优化推理延迟高要分段测量预处理花了多少时间模型推理花了多少时间后处理花了多少时间。预处理往往是瓶颈。比如图像推理解码JPEG、缩放、归一化可能比模型推理还慢。优化方法是把预处理放到GPU上或者用更快的库比如opencv比PIL快。模型推理本身可以用torch.jit.trace或者torch.jit.script把模型编译成静态图减少Python开销。还可以用ONNX Runtime或者TensorRT进一步加速。后处理如果涉及复杂的逻辑比如NMS非极大值抑制也可能成为瓶颈。优化方法是把后处理也放到GPU上或者用C实现。5.4 常见问题速查表问题现象可能原因排查方法解决方案损失不下降学习率太小、数据未归一化、模型初始化差打印梯度范数、检查数据分布调大学习率、归一化数据、换初始化损失震荡学习率太大、batch size太小观察损失曲线调小学习率、增大batch size验证集效果差过拟合、数据泄露对比训练和验证损失加正则化、检查数据划分显存溢出batch size太大、模型太大打印显存占用减小batch size、混合精度、梯度累积推理延迟高预处理慢、模型未优化分段计时优化预处理、模型编译、GPU加速服务崩溃输入异常、并发过高看日志、压测输入校验、限流、多进程提示遇到问题先别急着改代码先复现问题再定位问题最后才解决问题。我见过太多人一上来就瞎改结果问题没解决还引入了新bug。6. 我踩过的坑和给你的建议6.1 别过早优化先跑通再优化我刚开始做AI工程的时候总想着一步到位分布式训练、混合精度、模型量化全都安排上。结果光是环境配置就花了一周模型还没跑起来。后来我学乖了先用最简单的配置跑通全流程确认数据、模型、训练、推理都没问题再逐步加优化。这样每一步都有基线出了问题也知道是哪个改动导致的。6.2 日志和监控不是可选项训练的时候不记日志等模型效果不好的时候就抓瞎了。我现在的习惯是每个epoch记录损失、学习率、梯度范数、验证指标每次推理记录延迟、输入分布、输出分布。这些数据平时看着没用出问题的时候就是救命稻草。6.3 版本管理要贯穿数据和模型代码用Git管理这个大家都知道。但数据和模型也要版本管理。数据用哈希模型用实验编号。这样当你发现线上模型效果下降的时候能快速回滚到之前的版本也能对比两个版本的数据和模型差异。6.4 测试要覆盖边界情况推理服务的测试不能只测正常输入。要测空输入、超长输入、格式错误的输入、极端值输入。我见过一个服务正常请求都没问题结果有人传了一个空列表直接导致服务崩溃。后来加了输入校验才解决了问题。6.5 文档和注释是给未来的自己看的我现在看三个月前写的代码如果没有注释很多地方都要想半天。所以我的习惯是每个函数写清楚输入输出和关键逻辑每个配置项写清楚作用和默认值每个实验记录写清楚目的和结论。这些文档在交接和复现的时候价值巨大。这个内容后续还可以这样扩展把训练部分换成分布式训练把推理部分换成ONNX Runtime把数据部分换成流式加载。每换一个组件你都会对AI工程有更深的理解。我自己就是这么一步步走过来的从单机单卡到多机多卡从手动部署到CI/CD每一步都踩过坑但每一步都值得。