1. 从零手搓AI工程:为什么我不建议你直接调包
第一次看到ai-engineering-from-scratch这个项目名,我脑子里蹦出来的画面是:一个刚入行的算法工程师,对着满屏的pip install和import torch发呆,然后决定把所有这些黑盒拆开,从矩阵乘法开始,一行一行地把整个AI工程栈重新搭一遍。这个项目标题本身就带着一股“不破不立”的劲儿——它不是教你调参,不是教你用现成框架跑个demo,而是让你从最底层开始,理解每一个环节到底在干什么。
我做了十多年一线开发,带过不少新人,也见过太多“调包侠”。他们能熟练地写出model.fit(),能背出各种预训练模型的名称,但一旦遇到数据管道崩了、梯度爆炸了、显存不够了,就完全不知道从哪下手。ai-engineering-from-scratch这个项目要解决的,恰恰就是这个断层——它把AI工程拆解成一条完整的链路:从数据采集与清洗、特征工程、模型定义、训练循环、评估指标、到部署推理,每一个环节都要求你亲手实现一遍。适合谁来学?我认为有三类人最应该关注:第一类是刚转行做AI、只会调库但不懂原理的开发者;第二类是有传统后端经验、想补上AI工程能力的工程师;第三类是在校学生,想通过一个完整项目把课本上的零散知识串起来。
这个项目的核心价值不在于“造轮子”,而在于“拆轮子”。你亲手实现一遍反向传播,和你看十遍公式推导,理解深度完全不一样。我试过让团队里的新人先花两周时间,不借助任何高级框架,只用NumPy把一个小型神经网络从零训练到收敛,之后再让他们用PyTorch,效率直接翻倍——因为他们知道每一行代码背后发生了什么。ai-engineering-from-scratch就是把这个过程系统化了,它不只是一个代码仓库,更是一套完整的工程思维训练。
2. 整体架构设计:从数据到部署的完整链路拆解
2.1 为什么选择“全链路手写”而不是“模块化调包”
市面上大多数AI教程都是模块化的:数据加载用DataLoader,模型定义用nn.Module,训练用Trainer,部署用FastAPI。这种模式上手快,但有个致命问题——你永远不知道中间层发生了什么。ai-engineering-from-scratch的设计思路完全相反:它要求你把每个模块都手写一遍,哪怕写得丑、写得慢,也要先跑通。
我分析下来,这种设计背后有三个核心考量。第一,建立直觉。当你亲手实现一个卷积层的前向传播和反向传播后,你会对“感受野”“步长”“填充”这些概念产生肌肉记忆,而不是死记硬背。第二,暴露问题。调包时很多问题被框架隐藏了,比如数值稳定性、内存对齐、梯度累积,手写时这些问题会直接暴露出来,逼着你去解决。第三,培养调试能力。当你的手写实现和框架实现结果不一致时,你需要逐层对比、打印中间变量、检查梯度,这个过程本身就是最好的工程训练。
当然,全链路手写不等于完全不用工具。项目里允许使用NumPy做基础矩阵运算,允许用Matplotlib做可视化,但核心的模型定义、损失函数、优化器、训练循环必须自己实现。这个边界划得很清楚:工具是用来加速的,不是用来替代理解的。
2.2 核心模块划分与依赖关系
整个项目可以拆成六个核心模块,它们之间的依赖关系是线性的,但每个模块都可以独立测试。我按照实际实现顺序来梳理:
- 数据管道模块:负责数据加载、清洗、分词、向量化、批处理。这个模块的输出是标准化的张量,供后续模型使用。
- 基础算子模块:实现矩阵乘法、卷积、池化、激活函数等底层运算。这个模块是纯数学实现,不依赖任何深度学习框架。
- 自动微分模块:实现计算图的构建、前向传播、反向传播、梯度计算。这是整个项目最核心也最难的部分。
- 模型组装模块:基于基础算子和自动微分,组装出全连接网络、卷积网络、循环网络等。
- 训练循环模块:实现损失计算、梯度更新、学习率调度、模型保存与加载。
- 推理与部署模块:实现模型导出、量化、服务化封装。
每个模块都有对应的单元测试,确保输入输出符合预期。我建议你在实现时也遵循这个顺序,不要跳步。特别是自动微分模块,如果没吃透,后面的模型组装会非常痛苦。
2.3 技术选型背后的逻辑
项目选择NumPy作为基础运算库,而不是直接用PyTorch的Tensor,这个决策很关键。NumPy的API足够底层,能让你看清每一步运算,同时它的广播机制、索引操作和PyTorch高度相似,学到的知识可以直接迁移。另一个选择是纯Python实现,但那样性能太差,训练一个小模型都要跑半天,会严重打击积极性。
在自动微分方面,项目采用的是基于计算图的动态图方案,而不是静态图。动态图更直观,调试更方便,和PyTorch的设计理念一致。具体实现上,每个张量对象都维护一个grad属性和一个_backward函数,前向传播时构建计算图,反向传播时按拓扑逆序调用_backward。这个方案虽然性能不如静态图,但胜在清晰易懂。
注意:手写自动微分时,最容易出错的地方是梯度累积和广播机制。当一个标量和一个矩阵相加时,标量的梯度应该是矩阵梯度的和,这个细节如果处理不好,训练结果会完全错误。
3. 核心细节解析:自动微分与训练循环的实操要点
3.1 自动微分的实现原理与关键代码
自动微分是整个项目的灵魂。我把它拆成三个核心概念:计算图、链式法则、梯度累积。计算图是一个有向无环图,每个节点是一个张量,每条边是一个运算。前向传播时,我们按顺序执行运算并记录依赖关系;反向传播时,我们从损失函数出发,沿着计算图反向遍历,利用链式法则计算每个张量的梯度。
具体实现上,我定义了一个Tensor类,它包含data、grad、_children、_op和_backward五个属性。data存储实际数值,grad存储梯度,_children记录生成该张量的子张量,_op记录运算类型,_backward是一个函数,负责将当前张量的梯度传播给子张量。每次进行运算时,比如加法,我们会创建一个新的Tensor,并定义它的_backward函数:
class Tensor: def __init__(self, data, children=(), op=''): self.data = np.array(data) self.grad = np.zeros_like(self.data) self._children = set(children) self._op = op self._backward = lambda: None def __add__(self, other): other = other if isinstance(other, Tensor) else Tensor(other) out = Tensor(self.data + other.data, (self, other), '+') def _backward(): self.grad += out.grad other.grad += out.grad out._backward = _backward return out这段代码看起来简单,但有几个坑点。第一,self.grad += out.grad必须是累加而不是赋值,因为一个张量可能被多个下游节点使用。第二,广播机制需要特殊处理,如果self.data的形状是(3, 1),other.data的形状是(3, 4),那么反向传播时self.grad需要沿广播维度求和。第三,计算图的遍历顺序必须是拓扑逆序,否则梯度会传播错误。
我实测下来,最稳妥的做法是维护一个拓扑排序列表,每次反向传播时按逆序执行_backward。这个列表可以在前向传播时动态构建,也可以每次反向传播时重新计算。对于小规模模型,重新计算完全没问题;对于大规模模型,建议缓存拓扑排序结果。
3.2 训练循环的完整实现与参数选择
训练循环看起来简单,但细节非常多。一个完整的训练循环包含:数据批处理、前向传播、损失计算、反向传播、梯度裁剪、参数更新、学习率调度、日志记录。我逐个拆解。
数据批处理方面,我建议实现一个简单的DataLoader,支持随机打乱和分批。批大小(batch size)的选择很关键:太小会导致梯度噪声大、训练不稳定;太大会导致显存不足、泛化能力下降。根据我的经验,对于手写实现的小模型,批大小设在32到128之间比较合适。你可以从64开始试,如果损失震荡厉害就降到32,如果训练太慢就升到128。
损失函数方面,分类任务用交叉熵,回归任务用均方误差。交叉熵的实现要注意数值稳定性,直接计算log(softmax(x))容易溢出,正确做法是先用x - max(x)做平移,再计算log_sum_exp。这个技巧在框架里是默认处理的,手写时很容易忽略。
梯度裁剪是防止梯度爆炸的重要手段。我通常用全局范数裁剪,设定一个阈值(比如1.0或5.0),如果梯度的L2范数超过阈值,就按比例缩放。这个操作在RNN和Transformer里尤其重要。
参数更新方面,我建议先实现最基础的随机梯度下降(SGD),然后再实现动量法、Adam等优化器。SGD的更新公式是param -= lr * grad,动量法引入一个速度变量v = beta * v + (1 - beta) * grad,Adam则进一步引入二阶矩估计。我实测下来,对于小模型,Adam的收敛速度明显快于SGD,但SGD的泛化能力有时更好。你可以两个都实现,对比一下效果。
学习率调度方面,我常用的是余弦退火和阶梯衰减。余弦退火让学习率从初始值平滑降到0,训练后期更稳定;阶梯衰减每隔几个epoch降一次,简单粗暴但有效。初始学习率的选择,我一般从0.001开始试,如果损失下降太慢就调到0.01,如果损失震荡就降到0.0001。
3.3 数据管道的构建与预处理技巧
数据管道是AI工程里最容易被低估的环节。很多人觉得数据加载就是read_csv加train_test_split,但实际上,数据质量直接决定模型上限。ai-engineering-from-scratch要求你手写数据管道,目的就是让你重视这个环节。
我通常把数据管道分成四步:采集、清洗、转换、加载。采集阶段,你需要处理各种格式的数据源,比如CSV、JSON、图片、文本。清洗阶段,你需要处理缺失值、异常值、重复值。转换阶段,你需要做归一化、标准化、编码、分词。加载阶段,你需要实现批处理、打乱、预取。
以文本分类任务为例,分词是第一步。我建议先实现一个简单的空格分词,再实现基于词表的分词。词表的构建需要统计词频,过滤低频词,保留高频词。这里有个经验:词表大小控制在10000到50000之间比较合适,太小会导致信息丢失,太大会导致嵌入矩阵过大、训练变慢。
数值特征方面,归一化是必须的。我常用的是最小-最大归一化,把特征缩放到[0, 1]区间。如果数据分布接近高斯分布,标准化(减均值除标准差)效果更好。注意,归一化的参数(最小值、最大值、均值、标准差)必须从训练集计算,然后应用到验证集和测试集,否则会造成数据泄露。
提示:数据泄露是新手最容易犯的错误之一。任何使用到验证集或测试集信息的预处理操作,都会导致评估结果虚高。我见过有人在做归一化时用了全量数据,结果模型在测试集上表现很好,一上线就崩了。
4. 实操过程:从零训练一个手写数字识别模型
4.1 环境准备与项目初始化
动手之前,先把环境搭好。我建议用Python 3.9以上版本,安装NumPy、Matplotlib、Pillow这三个库就够了。不需要PyTorch、TensorFlow,这个项目的目的就是不用它们。你可以用虚拟环境隔离依赖:
python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy matplotlib pillow项目目录结构我建议这样组织:
ai-engineering-from-scratch/ ├── data/ │ ├── raw/ │ └── processed/ ├── src/ │ ├── tensor.py │ ├── ops.py │ ├── nn.py │ ├── optim.py │ ├── data.py │ └── train.py ├── tests/ │ ├── test_tensor.py │ ├── test_ops.py │ └── test_nn.py └── README.mdtensor.py放自动微分核心,ops.py放基础算子,nn.py放网络层,optim.py放优化器,data.py放数据管道,train.py放训练脚本。每个模块都要有对应的测试文件,确保功能正确。
4.2 数据加载与预处理实操
我们以MNIST手写数字数据集为例。虽然这个数据集很经典,但自己从头处理一遍,收获远比直接调torchvision.datasets.MNIST大得多。
第一步,下载数据。MNIST的原始格式是二进制文件,你需要解析文件头,提取图像和标签。图像是28x28的灰度图,标签是0到9的数字。解析代码如下:
import struct import numpy as np def load_mnist_images(filename): with open(filename, 'rb') as f: magic, num, rows, cols = struct.unpack('>IIII', f.read(16)) images = np.frombuffer(f.read(), dtype=np.uint8) images = images.reshape(num, rows, cols) return images def load_mnist_labels(filename): with open(filename, 'rb') as f: magic, num = struct.unpack('>II', f.read(8)) labels = np.frombuffer(f.read(), dtype=np.uint8) return labels第二步,预处理。把图像像素值从[0, 255]归一化到[0, 1],把标签转成独热编码。独热编码的实现很简单:创建一个全零向量,把对应位置设为1。
def normalize(images): return images.astype(np.float32) / 255.0 def one_hot(labels, num_classes=10): return np.eye(num_classes)[labels]第三步,划分数据集。我通常按7:1:2的比例划分训练集、验证集、测试集。划分前要随机打乱,避免数据顺序影响训练。
第四步,实现DataLoader。核心功能是分批和打乱。我实现了一个简单的版本:
class DataLoader: def __init__(self, data, labels, batch_size=64, shuffle=True): self.data = data self.labels = labels self.batch_size = batch_size self.shuffle = shuffle self.indices = np.arange(len(data)) def __iter__(self): if self.shuffle: np.random.shuffle(self.indices) for start in range(0, len(self.data), self.batch_size): end = min(start + self.batch_size, len(self.data)) batch_idx = self.indices[start:end] yield self.data[batch_idx], self.labels[batch_idx] def __len__(self): return (len(self.data) + self.batch_size - 1) // self.batch_size这个实现虽然简单,但足够用。注意__len__的计算要用向上取整,否则最后一个不完整的批次会被漏掉。
4.3 模型定义与训练过程
我们定义一个简单的三层全连接网络:输入层784维,隐藏层128维,输出层10维。激活函数用ReLU,输出层用Softmax。
class Linear: def __init__(self, in_features, out_features): self.weight = Tensor(np.random.randn(in_features, out_features) * 0.01) self.bias = Tensor(np.zeros(out_features)) def __call__(self, x): return x @ self.weight + self.bias class ReLU: def __call__(self, x): out = Tensor(np.maximum(0, x.data), (x,), 'relu') def _backward(): x.grad += (x.data > 0) * out.grad out._backward = _backward return out class SoftmaxCrossEntropy: def __call__(self, logits, labels): # 数值稳定处理 shifted = logits.data - np.max(logits.data, axis=1, keepdims=True) exp = np.exp(shifted) probs = exp / np.sum(exp, axis=1, keepdims=True) loss = -np.mean(np.sum(labels * np.log(probs + 1e-8), axis=1)) # 反向传播 grad = (probs - labels) / len(labels) logits.grad += grad return loss训练循环的实现:
def train(model, train_loader, val_loader, epochs=10, lr=0.001): optimizer = SGD(model.parameters(), lr=lr) for epoch in range(epochs): model.train() train_loss = 0 for x, y in train_loader: x = Tensor(x.reshape(-1, 784)) y = one_hot(y) logits = model(x) loss = SoftmaxCrossEntropy()(logits, y) optimizer.zero_grad() loss.backward() optimizer.step() train_loss += loss.item() # 验证 val_acc = evaluate(model, val_loader) print(f'Epoch {epoch+1}, Loss: {train_loss/len(train_loader):.4f}, Val Acc: {val_acc:.4f}')这里有几个关键点。第一,zero_grad必须在反向传播前调用,否则梯度会累积。第二,loss.backward()需要实现拓扑排序,确保梯度传播顺序正确。第三,optimizer.step()更新参数后,需要清空计算图,避免内存泄漏。
我实测下来,这个简单模型在MNIST上训练10个epoch,验证集准确率能到97%左右。虽然比不上CNN,但作为手写实现的第一个模型,这个结果已经很有成就感了。
4.4 模型评估与推理部署
训练完成后,需要在测试集上评估最终性能。评估时要注意切换到推理模式,关闭Dropout和BatchNorm的训练行为。虽然我们这个简单模型没有这些层,但养成这个习惯很重要。
推理部署方面,我建议实现一个简单的predict函数,接收单张图片,返回预测类别。这个函数可以封装成HTTP服务,用Flask或FastAPI暴露接口。虽然这个模型很小,但整个流程和部署大模型是一样的:加载权重、预处理输入、前向传播、后处理输出。
def predict(model, image): model.eval() x = Tensor(image.reshape(1, 784)) logits = model(x) return np.argmax(logits.data, axis=1)[0]部署时要注意输入验证和异常处理。用户可能传入任意尺寸的图片,你需要先缩放到28x28,再归一化。这些预处理步骤必须和训练时完全一致,否则预测结果会完全错误。
5. 常见问题与排查技巧实录
5.1 梯度相关问题的排查思路
梯度问题是手写自动微分时最常见的坑。我整理了一个速查表:
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 损失不下降 | 学习率太小 | 打印梯度范数 | 增大学习率 |
| 损失震荡 | 学习率太大 | 观察损失曲线 | 减小学习率或加动量 |
| 损失变NaN | 梯度爆炸 | 检查梯度值 | 梯度裁剪、降低学习率 |
| 梯度为0 | 激活函数饱和 | 打印激活值 | 换ReLU、调整初始化 |
| 梯度形状错误 | 广播处理不当 | 逐层对比形状 | 修正反向传播的求和维度 |
我踩过最深的坑是广播机制。当时实现一个加法操作,self.data形状是(64, 128),other.data形状是(128,),前向传播没问题,但反向传播时other.grad的形状变成了(64, 128),导致参数更新时形状不匹配。正确的做法是对other.grad沿第0维求和,得到(128,)的形状。这个细节在框架里是自动处理的,手写时必须自己实现。
另一个坑是梯度累积。如果一个张量被多个下游节点使用,它的梯度应该是所有下游梯度的和。我一开始写成了赋值,结果只有最后一个下游节点的梯度被保留,训练完全不对。后来改成+=才解决。
5.2 数值稳定性问题的处理经验
数值稳定性是手写实现绕不开的问题。最典型的是Softmax和交叉熵。直接计算exp(x)当x很大时会溢出,log(0)会得到负无穷。解决方案是减去最大值:
shifted = x - np.max(x, axis=1, keepdims=True) exp = np.exp(shifted) probs = exp / np.sum(exp, axis=1, keepdims=True)这个操作在数学上是等价的,因为softmax(x) = softmax(x - c),但数值上稳定得多。交叉熵计算时也要加一个极小值1e-8,防止log(0)。
另一个数值问题是权重初始化。如果权重初始值太大,前向传播时激活值会爆炸;如果太小,激活值会趋近于0,梯度消失。我常用的初始化方法是Xavier初始化:weight = np.random.randn(in_features, out_features) * np.sqrt(2.0 / in_features)。这个公式保证前向传播时每层的方差大致相同。
5.3 训练效率优化的实操技巧
手写实现的训练速度肯定比不上框架,但通过一些技巧可以缩小差距。第一,向量化操作。尽量避免Python循环,用NumPy的广播和矩阵运算。我试过用循环实现卷积,训练一个epoch要半小时,改成im2col加矩阵乘法后,只要两分钟。第二,批处理。一次处理多个样本,充分利用CPU的并行能力。第三,数据类型。用float32而不是float64,内存占用减半,速度提升明显。第四,避免不必要的拷贝。NumPy的切片操作返回视图,但某些操作会触发拷贝,要注意区分。
提示:如果你发现训练速度异常慢,先检查是不是在Python循环里做了大量小规模运算。把循环改成矩阵运算,通常能有10倍以上的提速。
5.4 模型评估中的常见陷阱
评估环节有几个容易忽略的陷阱。第一,数据泄露。前面提过,预处理参数必须从训练集计算。第二,评估模式。Dropout和BatchNorm在训练和推理时的行为不同,评估时必须切换到推理模式。第三,指标选择。分类任务不能只看准确率,如果类别不平衡,准确率会虚高。我通常同时看精确率、召回率和F1分数。第四,随机性。多次评估取平均,避免单次评估的偶然性。
我见过一个案例:有人在验证集上调整超参数,调了上百次,最后选了一个验证集准确率最高的模型,结果测试集准确率低了5个百分点。这就是典型的过拟合验证集。正确的做法是划分出独立的测试集,只在最后评估一次。
6. 从手写实现到工程落地的扩展思路
6.1 性能优化与框架对比
手写实现跑通之后,下一步就是和框架对比。我建议你用同样的模型结构,分别用NumPy和PyTorch实现一遍,对比训练速度、内存占用、最终精度。你会发现,PyTorch的速度可能是NumPy的10到50倍,但精度基本一致。这个对比能让你直观感受到框架的价值——它帮你处理了底层优化,让你专注于模型设计。
如果你想进一步优化手写实现的性能,可以尝试几个方向。第一,用Cython或Numba加速热点代码。第二,用多进程做数据加载,避免IO成为瓶颈。第三,用BLAS库加速矩阵乘法,NumPy默认已经链接了BLAS,但你可以手动配置更高效的版本。第四,实现混合精度训练,用float16做前向传播,float32做参数更新。
6.2 扩展到更复杂的模型结构
掌握了全连接网络后,可以逐步扩展到卷积网络、循环网络、注意力机制。卷积网络的核心是im2col操作,把图像块展开成矩阵,然后用矩阵乘法实现卷积。循环网络的核心是时间步展开和梯度截断。注意力机制的核心是查询、键、值的矩阵运算和Softmax归一化。
每扩展一种结构,你都会遇到新的挑战。比如卷积的反向传播需要处理填充和步长的逆操作,循环网络的梯度容易爆炸或消失,注意力机制的计算复杂度是序列长度的平方。这些挑战正是这个项目的价值所在——它们逼着你去理解每个结构的本质。
6.3 工程化部署的注意事项
模型训练好之后,部署是最后一公里。我总结了几条经验。第一,模型序列化。把权重保存成文件,推理时加载。我常用np.savez保存权重,简单可靠。第二,输入预处理一致性。推理时的预处理必须和训练时完全一致,包括归一化参数、图像尺寸、通道顺序。第三,异常处理。用户输入可能千奇百怪,要做好输入验证和降级处理。第四,性能监控。记录推理延迟、吞吐量、错误率,及时发现线上问题。
如果你要把模型部署成服务,我建议先用Flask写一个简单接口,跑通之后再考虑用FastAPI或gRPC优化性能。不要一上来就追求高并发、低延迟,先把功能跑通,再逐步优化。
6.4 持续学习与项目扩展建议
ai-engineering-from-scratch不是一个一次性的项目,而是一个可以持续扩展的学习平台。我建议你每学到一个新概念,就尝试在手写框架里实现一遍。比如学了BatchNorm,就手写一个BatchNorm层;学了Dropout,就手写一个Dropout层;学了Adam,就手写一个Adam优化器。每实现一个,你对这个概念的理解就会深一层。
另外,我建议你把实现过程中的踩坑经历记录下来,写成笔记或博客。教是最好的学,当你试图向别人解释一个概念时,你会发现自己哪里没理解透。这个项目本身就是一个很好的分享素材,你可以把代码开源,把经验写成文档,和其他学习者交流。
我个人在实际操作中的体会是,手写实现最大的收获不是代码本身,而是那种“原来如此”的顿悟时刻。当你亲手实现反向传播,看到梯度一步步传播回去,你会对深度学习产生完全不同的认知。这种认知,是调包永远给不了的。