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_norm=1.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 python=3.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_size=0.2, random_state=42, stratify=y ) # 标准化,只用训练集的统计量 scaler = StandardScaler() X_train = scaler.fit_transform(X_train) X_val = scaler.transform(X_val) # 保存scaler,推理时要用 import joblib joblib.dump(scaler, 'experiments/scaler.pkl')这里的关键点是stratify=y,保证训练集和验证集的类别比例一致。还有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(), lr=1e-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_size=16, shuffle=True) 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(f'Epoch {epoch+1}, 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_model=PredictResponse) 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, dim=1) confidence, class_id = torch.max(probabilities, dim=1) return PredictResponse( class_id=int(class_id), class_name=class_names[int(class_id)], confidence=float(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,每一步都踩过坑,但每一步都值得。