1. 从零手搓AI工程:为什么我不建议你直接调包
第一次看到ai-engineering-from-scratch这个项目标题时,我脑子里蹦出来的画面是:一个人坐在终端前,从矩阵乘法开始,一行一行地把一个能跑起来的AI系统给搭出来。不是那种pip install transformers然后model.generate()的玩法,而是真正去碰张量、碰反向传播、碰推理引擎的调度逻辑。这个方向这两年越来越多人关注,原因很直接——调包调久了,遇到性能瓶颈、显存爆炸、推理延迟高的时候,你根本不知道问题出在哪一层。
这个项目要解决的核心问题,就是把AI工程从"黑盒调用"拉回到"白盒理解"。它适合谁?适合那些已经会用PyTorch或TensorFlow跑通模型,但说不清楚一次前向传播到底经历了哪些内存操作的人;适合想自己写推理框架、想搞懂CUDA kernel调度、想从底层优化模型性能的工程师;也适合刚入行、不想只当"调参侠"的AI方向学生。一句话,它帮你建立从数学原理到工程落地的完整链路认知。
我自己的经历是,早些年做模型部署,遇到一个算子融合的问题,查了三天文档都没搞明白为什么融合后反而变慢了。后来逼着自己用NumPy手写了一遍卷积和BN的实现,才真正理解内存访问模式对性能的影响。从那以后我就认定,AI工程能力的地基,必须自己亲手夯一遍。这个项目就是干这个的。
2. 整体设计思路:自底向上,但不必从晶体管开始
2.1 分层拆解:从数学到系统的四层架构
ai-engineering-from-scratch这类项目的设计哲学,我理解下来是四层递进:数学层、算法层、框架层、系统层。数学层解决"是什么",比如张量到底是什么、梯度怎么算;算法层解决"怎么学",比如反向传播、优化器更新;框架层解决"怎么组织",比如计算图、自动微分;系统层解决"怎么跑得快",比如内存管理、算子调度、并行策略。
为什么这么分?因为如果你一上来就啃CUDA编程,大概率会被各种内存对齐、线程束这些概念劝退。但如果你先把NumPy版本的神经网络跑通,理解了计算流程,再去优化,路径就顺了。这就像学做菜,先学会把菜炒熟,再去研究火候和锅具。
提示:不要跳过数学层直接进框架层。我见过太多人直接看PyTorch源码,结果卡在
autograd的拓扑排序上,就是因为没手推过链式法则。
2.2 技术选型:为什么是Python + NumPy + C++/CUDA
项目通常会用Python做胶水层和原型验证,NumPy做基础张量运算,C++/CUDA做性能关键路径。这个组合不是随便选的。Python的生态和可读性无可替代,NumPy的ndarray是理解张量操作的最佳教具,而C++/CUDA则是通往工业级性能的必经之路。
有人会问,为什么不直接用Rust或者Julia?我的看法是,Rust的学习曲线在AI工程这个场景下太陡,而且生态还在建设中;Julia性能好但调试工具链不够成熟。Python + C++的组合,是开发效率和运行效率的平衡点。你可以用Python快速验证想法,然后把热点代码用C++重写,通过pybind11暴露接口。这套流程在工业界已经被验证过无数次。
2.3 避坑设计:为什么不做"大而全"
这类项目最容易犯的错误,是想把Transformer、CNN、RNN全实现一遍,结果每个都浅尝辄止。我的建议是先深后广:选一个经典结构(比如MLP或者小型Transformer),从数据加载、前向传播、损失计算、反向传播、参数更新,完整走一遍。走通之后,再横向扩展。
为什么?因为AI工程的难点不在模型种类,而在通用基础设施。你把自动微分搞明白了,实现CNN和RNN只是算子不同;你把内存池搞明白了,所有模型的显存管理都受益。所以项目设计上,应该把80%的精力放在通用机制上,20%放在具体模型上。
3. 核心细节解析:张量、自动微分与计算图
3.1 张量实现:不只是多维数组
很多人以为张量就是NumPy的ndarray加个requires_grad,其实远不止。一个用于AI工程的张量,至少需要包含这些信息:数据指针、形状(shape)、步长(stride)、数据类型(dtype)、设备(device)、梯度存储、以及是否参与梯度计算的标志。
步长这个概念特别关键。为什么?因为它决定了你如何在不复制数据的情况下做切片、转置、广播。举个例子,一个形状为(3, 4)的二维张量,如果按行优先存储,步长是(4, 1);转置之后形状变成(4, 3),但数据没动,步长变成(1, 4)。这就是视图操作的本质——只改元信息,不碰数据。
import numpy as np a = np.arange(12).reshape(3, 4) print(a.strides) # (32, 8),单位是字节 b = a.T print(b.strides) # (8, 32),数据没变,步长变了自己实现张量时,我建议先把步长机制吃透。因为后续的自动微分、算子融合、内存优化,全都建立在这个基础上。踩过的坑是:早期我用连续内存假设写算子,结果遇到非连续输入时结果全错。后来强制自己在每个算子入口检查is_contiguous,不连续就先contiguous(),问题才解决。
3.2 自动微分:计算图的构建与反向遍历
自动微分的核心是计算图。前向传播时,每个操作记录下输入、输出和反向传播函数;反向传播时,从损失开始,按拓扑逆序调用每个节点的反向函数,把梯度累加到输入上。
这里有个关键设计决策:动态图还是静态图。PyTorch用动态图,每次前向都重建计算图,灵活但开销大;TensorFlow 1.x用静态图,先定义后运行,优化空间大但调试困难。ai-engineering-from-scratch通常从动态图入手,因为实现简单、调试直观。
实现时要注意几点。第一,梯度累加要用+=而不是=,因为一个张量可能被多个下游节点使用。第二,要及时释放不需要的中间变量,否则显存会爆。第三,要处理原地操作(in-place operation)对计算图的影响,这是很多bug的根源。
class Tensor: def __init__(self, data, requires_grad=False): self.data = data self.requires_grad = requires_grad self.grad = None self._backward = lambda: None self._prev = set() def __add__(self, other): out = Tensor(self.data + other.data, self.requires_grad or other.requires_grad) def _backward(): if self.requires_grad: self.grad = out.grad if self.grad is None else self.grad + out.grad if other.requires_grad: other.grad = out.grad if other.grad is None else other.grad + out.grad out._backward = _backward out._prev = {self, other} return out上面这段代码是极简版,但已经能看出自动微分的骨架。实际项目中还要处理广播、矩阵乘法、激活函数等,但模式是一样的。
3.3 内存管理:显存池与生命周期
AI工程和普通软件开发最大的区别之一,就是显存是稀缺资源。你不能像写Java那样随便new对象然后等GC。在GPU上,一次显存分配可能耗时几十微秒,频繁分配释放会严重拖慢训练速度。
解决方案是显存池。预先分配一大块显存,然后自己管理分配和回收。PyTorch的CUDACachingAllocator就是这个思路。实现时要注意碎片问题:如果频繁分配不同大小的块,池子会碎片化。常见做法是按大小分桶,比如小于1MB的走小对象池,大于1MB的走大对象池。
注意:显存池的回收策略要谨慎。我试过用引用计数,结果遇到循环引用时显存泄漏。后来改成基于计算图生命周期的标记清除,才稳定下来。
4. 实操过程:从零搭建一个可训练的MLP
4.1 环境准备与依赖安装
先明确环境。我推荐Python 3.10以上,NumPy 1.24以上。如果要跑GPU版本,需要CUDA 11.8或12.x,以及对应的驱动。C++编译器用g++ 11以上或clang 14以上。构建工具用CMake 3.20以上。
python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy pytest matplotlib为什么用venv而不是conda?因为这类项目依赖少,venv足够轻量,而且和系统Python隔离干净。pytest用于写单元测试,matplotlib用于可视化损失曲线,这两个是调试必备。
4.2 张量类的完整实现
我们来实现一个支持自动微分的张量类。核心方法包括:__init__、__add__、__mul__、__matmul__、relu、backward。每个方法都要正确处理梯度的反向传播。
class Tensor: def __init__(self, data, requires_grad=False, _children=()): self.data = np.asarray(data, dtype=np.float32) self.requires_grad = requires_grad self.grad = None self._backward = lambda: None self._prev = set(_children) def __add__(self, other): other = other if isinstance(other, Tensor) else Tensor(other) out = Tensor(self.data + other.data, self.requires_grad or other.requires_grad, (self, other)) def _backward(): if self.requires_grad: g = out.grad # 处理广播:把梯度求和回原形状 while g.ndim > self.data.ndim: g = g.sum(axis=0) for i, dim in enumerate(self.data.shape): if dim == 1: g = g.sum(axis=i, keepdims=True) self.grad = g if self.grad is None else self.grad + g if other.requires_grad: g = out.grad while g.ndim > other.data.ndim: g = g.sum(axis=0) for i, dim in enumerate(other.data.shape): if dim == 1: g = g.sum(axis=i, keepdims=True) other.grad = g if other.grad is None else other.grad + g out._backward = _backward return out def __matmul__(self, other): out = Tensor(self.data @ other.data, self.requires_grad or other.requires_grad, (self, other)) def _backward(): if self.requires_grad: g = out.grad @ other.data.T self.grad = g if self.grad is None else self.grad + g if other.requires_grad: g = self.data.T @ out.grad other.grad = g if other.grad is None else other.grad + g out._backward = _backward return out def relu(self): out = Tensor(np.maximum(0, self.data), self.requires_grad, (self,)) def _backward(): if self.requires_grad: g = out.grad * (self.data > 0) self.grad = g if self.grad is None else self.grad + g out._backward = _backward return out def backward(self): topo = [] visited = set() def build(v): if v not in visited: visited.add(v) for child in v._prev: build(child) topo.append(v) build(self) self.grad = np.ones_like(self.data) for v in reversed(topo): v._backward()这段代码有几个细节值得说。第一,广播的处理。当两个形状不同的张量相加时,梯度需要求和回原形状,否则后续更新参数时形状对不上。第二,拓扑排序用递归实现,简单但要注意Python的递归深度限制,大模型可能需要改成迭代。第三,backward的起点梯度是全1,因为损失对标量的导数就是1。
4.3 训练循环与参数更新
有了张量类,就可以搭一个两层MLP了。输入784维(28x28图像),隐藏层128维,输出10类。
class MLP: def __init__(self): self.W1 = Tensor(np.random.randn(784, 128) * 0.01, requires_grad=True) self.b1 = Tensor(np.zeros(128), requires_grad=True) self.W2 = Tensor(np.random.randn(128, 10) * 0.01, requires_grad=True) self.b2 = Tensor(np.zeros(10), requires_grad=True) def forward(self, x): h = (x @ self.W1 + self.b1).relu() return h @ self.W2 + self.b2 def parameters(self): return [self.W1, self.b1, self.W2, self.b2]训练循环里,每个batch做四件事:前向传播、计算损失、反向传播、更新参数。损失用交叉熵,更新用SGD。
def cross_entropy(logits, labels): # logits: (N, 10), labels: (N,) max_logits = logits.data.max(axis=1, keepdims=True) exp_logits = np.exp(logits.data - max_logits) probs = exp_logits / exp_logits.sum(axis=1, keepdims=True) loss = -np.log(probs[np.arange(len(labels)), labels]).mean() return Tensor(loss), probs # 训练循环 lr = 0.01 for epoch in range(10): for x_batch, y_batch in dataloader: logits = model.forward(Tensor(x_batch)) loss, probs = cross_entropy(logits, y_batch) for p in model.parameters(): p.grad = None loss.backward() for p in model.parameters(): p.data -= lr * p.grad print(f"Epoch {epoch}, Loss: {loss.data:.4f}")实测下来,这个极简实现在MNIST上跑10个epoch,准确率能到92%左右。虽然比不上PyTorch,但整个流程完全透明,每一行代码你都知道在干什么。
4.4 性能优化:从NumPy到C++的迁移路径
NumPy版本跑通后,你会发现瓶颈在矩阵乘法上。NumPy底层用的是BLAS,单线程下性能一般。优化路径有两条:一是用多线程BLAS(如OpenBLAS),二是自己写C++扩展。
我建议先做profiling,用cProfile找出热点函数。通常__matmul__和relu占大头。然后针对热点写C++版本,用pybind11暴露接口。
// matmul.cpp #include <pybind11/pybind11.h> #include <pybind11/numpy.h> namespace py = pybind11; py::array_t<float> matmul(py::array_t<float> a, py::array_t<float> b) { auto a_buf = a.request(); auto b_buf = b.request(); // 省略具体实现,核心是三重循环 + 分块优化 // ... }编译命令:
g++ -O3 -shared -std=c++17 -fPIC $(python -m pybind11 --includes) matmul.cpp -o matmul$(python-config --extension-suffix)提示:C++扩展的调试比Python麻烦得多。建议先用小规模数据验证正确性,再上大规模。我踩过的坑是:C++里用了
int索引,结果矩阵超过2^31元素时溢出,排查了半天。
5. 常见问题与排查技巧实录
5.1 梯度爆炸与消失:诊断与缓解
训练深层网络时,梯度爆炸或消失是家常便饭。诊断方法很简单:打印每层梯度的范数。如果某层梯度范数超过1e3,就是爆炸;如果小于1e-6,就是消失。
缓解手段有几个。第一,用Xavier或He初始化,根据输入输出维度调整初始权重尺度。第二,加BatchNorm,把每层输入归一化。第三,用梯度裁剪,把梯度范数限制在阈值内。
def clip_gradients(parameters, max_norm=1.0): total_norm = np.sqrt(sum(np.sum(p.grad ** 2) for p in parameters if p.grad is not None)) if total_norm > max_norm: scale = max_norm / (total_norm + 1e-6) for p in parameters: if p.grad is not None: p.grad *= scale我实测下来,梯度裁剪对RNN类模型效果特别明显,对MLP影响不大。所以要根据模型结构决定是否使用。
5.2 显存泄漏:定位与修复
显存泄漏的表现是:训练几个epoch后,显存占用持续上升,最后OOM。定位方法是记录每次分配和释放的调用栈,找出没有配对的分配。
常见原因有三个。一是计算图没有释放,中间变量一直被引用。二是循环引用,两个张量互相持有对方。三是缓存没有清理,比如保存了所有中间激活值用于反向传播,但反向传播结束后没释放。
修复方法:在backward结束后,手动把_prev清空,断开计算图引用。对于循环引用,用weakref。对于激活值缓存,用完后立即del。
5.3 数值稳定性:softmax与log-sum-exp技巧
直接算softmax会遇到溢出问题:exp(1000)是无穷大。解决方案是减去最大值:softmax(x) = softmax(x - max(x))。这个技巧叫log-sum-exp,是数值计算的基本功。
交叉熵损失也要注意。不要先算softmax再算log,而是直接用logits算,避免中间结果的精度损失。PyTorch的cross_entropy内部就是这么做的。
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 损失变NaN | 学习率过大 | 打印每步损失 | 降低学习率,加梯度裁剪 |
| 准确率不涨 | 初始化不当 | 检查初始权重分布 | 改用He/Xavier初始化 |
| 训练速度慢 | 数据加载瓶颈 | profiling数据加载 | 多进程预取,pin memory |
| 显存OOM | 批量太大 | 逐步减小batch | 梯度累积,混合精度 |
| 梯度为0 | 激活函数饱和 | 打印梯度直方图 | 换ReLU,加BN |
5.4 调试技巧:从单元测试到梯度检查
梯度检查是验证自动微分正确性的黄金标准。原理很简单:用数值微分近似解析梯度,两者对比。
def grad_check(f, x, eps=1e-4): # 数值梯度 num_grad = np.zeros_like(x.data) it = np.nditer(x.data, flags=['multi_index']) while not it.finished: idx = it.multi_index old = x.data[idx] x.data[idx] = old + eps f_pos = f(x).data x.data[idx] = old - eps f_neg = f(x).data x.data[idx] = old num_grad[idx] = (f_pos - f_neg) / (2 * eps) it.iternext() # 解析梯度 x.grad = None f(x).backward() ana_grad = x.grad # 对比 diff = np.abs(num_grad - ana_grad).max() print(f"Max gradient diff: {diff:.6e}") return diff < 1e-4这个检查要定期做,尤其是新实现一个算子之后。我踩过的坑是:ReLU在0点的导数,我定义成0,但数值微分在0附近会给出0.5,导致检查失败。后来统一约定0点导数为0,并在检查时避开0点。
6. 从原型到生产:还需要补哪些课
6.1 算子融合与图优化
原型跑通后,性能优化的大头在算子融合。比如matmul + bias + relu可以融合成一个kernel,减少内存读写。实现方式是在计算图上做模式匹配,找到可融合的子图,然后生成融合后的kernel。
这个工作难度不小,需要对硬件架构有深入理解。我的建议是先从简单的融合开始,比如element-wise操作的融合,再逐步扩展到复杂模式。
6.2 混合精度训练
用FP16代替FP32做前向和反向,可以显著减少显存占用和加速计算。但要注意数值稳定性:FP16的动态范围小,容易溢出。解决方案是用loss scaling,把损失放大,反向传播后再缩回来。
scale = 1024.0 loss_scaled = loss * scale loss_scaled.backward() for p in model.parameters(): p.grad /= scale6.3 分布式训练入门
单卡跑不动时,就要上多卡。数据并行是最简单的方案:每个卡跑一个batch,梯度all-reduce后更新。实现时要注意通信开销,用NCCL做GPU间通信,用ring all-reduce减少带宽压力。
这块内容展开能写一本书,这里只提个醒:分布式训练的bug特别难调,建议先用单机多卡验证,再上多机。
7. 我个人的一些实操体会
这个项目我断断续续做了三个月,最大的收获不是写出了多快的算子,而是对AI系统的敬畏。以前调包时觉得模型训练理所当然,自己实现后才知道,每一个看似简单的操作背后,都有无数工程细节在支撑。
如果让我给后来者建议,我会说:不要贪快,不要跳过数学推导,不要怕写单元测试。我见过太多人想一周搞定,结果卡在反向传播的链式法则上。慢就是快,把基础打牢,后面学什么都快。
另外,社区里有很多优秀的开源实现可以参考,比如micrograd、tinygrad、llm.c。不要闭门造车,但也不要直接抄,要理解每一行代码的意图。我自己的做法是:先看别人的实现,理解思路,然后关掉参考,自己从零写一遍,遇到卡壳再回去看。这样反复几次,知识就内化了。
最后分享一个小技巧:给每个算子写一个可视化测试。比如卷积算子,输入一个简单的图案,把输出画出来,肉眼检查是否正确。这比看数字直观得多,也更容易发现边界条件的bug。