1. 从零手搓AI工程:为什么我不建议你直接调包
1.1 一个让我彻底改变主意的真实场景
去年帮一个朋友排查线上推理服务的问题,现象很典型:模型在测试集上指标漂亮,一上生产环境延迟就飙到800ms以上,GPU利用率却只有30%出头。团队里几个工程师围着代码看了两天,怀疑是模型太大、怀疑是并发不够、怀疑是框架版本不对,换了一轮依赖,问题依旧。最后我让他们把推理链路里每一段耗时打出来,才发现真正的瓶颈根本不在模型本身,而在数据预处理阶段——一个看起来人畜无害的tokenizer调用,因为每次请求都重新加载词表,硬生生吃掉了将近一半的时间。
这件事对我触动很大。现在市面上讲AI的教程,绝大多数都在教你“怎么调用一个现成的API”或者“怎么用三行代码跑通一个预训练模型”。这当然没错,工程效率至上。但问题在于,当你只会调包的时候,你对整个系统的认知是黑盒的。一旦出了问题,你连从哪儿下手都不知道,只能靠猜、靠换、靠玄学。
ai-engineering-from-scratch这个方向,说白了就是反其道而行之:不依赖高层封装,从最底层的矩阵运算开始,一步步把AI工程里那些核心组件亲手搭出来。它解决的不是“怎么快速出结果”的问题,而是“怎么真正理解结果是怎么来的”的问题。适合谁看?我认为有三类人最该走一遍这条路:一是刚入行、只会调库但说不清原理的算法工程师;二是做后端或数据方向、想转AI工程但被各种框架绕晕的开发者;三是带团队的技术负责人,你需要知道每个环节的坑在哪,才能做出靠谱的架构决策。
1.2 从零构建到底“零”到什么程度
先把预期对齐,免得你中途放弃。所谓“from scratch”,不是让你用汇编去写矩阵乘法,也不是让你从晶体管开始造GPU。合理的“零”起点应该是这样的:
- 不依赖自动微分框架:PyTorch、TensorFlow这些可以用,但只用来做对照验证,核心的前向传播、反向传播、梯度更新,你自己用NumPy实现一遍。
- 不依赖高层训练封装:
Trainer、fit()这类一键训练接口先放一边,数据加载、批次划分、损失计算、参数更新,全部手写。 - 不依赖现成的推理服务框架:模型怎么加载、请求怎么排队、批处理怎么做、显存怎么管理,自己设计一套最小可用的方案。
这个起点听起来吓人,但实际做下来你会发现,真正核心的东西并没有想象中那么多。一个能跑通的全连接网络,核心代码量可能也就两三百行。难的不是写代码,而是理解每一步为什么这么设计。
提示:不要一上来就挑战Transformer。先用一个两层全连接网络在MNIST上跑通完整流程,把前向、反向、更新这三个环节彻底吃透,再往上叠复杂度。我见过太多人直接冲Transformer,结果卡在矩阵维度对不上,挫败感极强。
1.3 这条路能给你带来什么实际收益
说点实在的。走完一遍从零构建的流程,你至少能在三个层面获得明显提升。
第一是调试能力。当loss不下降的时候,你不再只会调学习率。你会知道去检查梯度是不是消失了、初始化是不是有问题、数据归一化是不是漏了。这些判断力,调包是调不出来的。
第二是性能优化能力。你知道每一步计算在干什么,就能精准定位瓶颈。前面那个tokenizer的例子就是典型——如果你自己实现过词表加载和编码过程,你第一时间就会怀疑那里。
第三是架构设计能力。当你要设计一个推理服务的时候,你会清楚哪些环节可以并行、哪些必须串行、显存峰值出现在哪里、批处理大小怎么定。这些决策的质量,直接决定了系统的成本和稳定性。
2. 核心组件拆解:一个AI工程到底由哪些零件组成
2.1 数据管道:最容易被低估的环节
很多人把注意力全放在模型结构上,觉得数据管道就是“读文件、转格式”这种脏活累活。但实际工程中,数据管道往往是问题最多、优化空间最大的地方。
从零构建数据管道,你需要自己实现几个核心模块。数据加载器要支持按批次读取、打乱顺序、多进程预取。预处理流水线要处理归一化、编码、填充截断。缓存机制要避免重复计算,尤其是那些耗时较长的特征工程步骤。
这里有个关键设计决策:预处理放在训练前还是训练中?放在训练前,一次性处理好存成文件,训练时读取快,但占用磁盘空间,且数据增强的灵活性差。放在训练中,每次读取都实时处理,灵活但可能成为瓶颈。我的建议是折中——把不变的预处理(如分词、归一化)提前做好,把需要随机性的增强(如裁剪、旋转)放在训练时做。
# 一个极简的数据加载器核心逻辑示意 class SimpleDataLoader: def __init__(self, data, labels, batch_size, 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.indices), self.batch_size): batch_idx = self.indices[start:start + self.batch_size] yield self.data[batch_idx], self.labels[batch_idx]这段代码看起来简单,但你要想清楚几个问题:为什么用索引而不是直接切数据?因为直接切数据会复制内存,数据量大时开销惊人。为什么__iter__里才打乱?因为每次epoch都要重新打乱,放在初始化里只打乱一次就失去意义了。
注意:多进程预取是把双刃剑。
num_workers设大了,进程间通信开销可能超过计算本身;设小了,GPU等数据。经验值是从2开始试,逐步加到4或8,观察GPU利用率变化。如果加到8利用率还在涨,说明数据管道确实是瓶颈,需要考虑更激进的优化,比如把数据预加载到内存或显存。
2.2 模型定义:从矩阵乘法开始搭积木
从零定义模型,核心就三件事:初始化参数、定义前向计算、准备反向传播。我们以一个两层全连接网络为例,把每一步拆开看。
参数初始化不是随便给个随机数就完事。全零初始化会导致所有神经元对称,学不到东西。太小的随机数会让信号逐层衰减,太大的会让梯度爆炸。常用的He初始化和Xavier初始化,本质上是根据输入输出维度来缩放随机数范围,保证每层输出的方差大致稳定。
# He初始化:适用于ReLU激活函数 def he_init(fan_in, fan_out): std = np.sqrt(2.0 / fan_in) return np.random.randn(fan_in, fan_out) * std # Xavier初始化:适用于tanh、sigmoid等 def xavier_init(fan_in, fan_out): std = np.sqrt(2.0 / (fan_in + fan_out)) return np.random.randn(fan_in, fan_out) * std前向传播就是矩阵乘法加激活函数。但这里有个细节:偏置项怎么加?是每一层单独加,还是合并到权重矩阵里?单独加更清晰,合并更高效。从零构建阶段建议单独加,便于理解。
反向传播是整个流程里最烧脑的部分。核心是链式法则,但落到代码上,你需要手动推导每一层的梯度表达式。以两层网络为例,输出层的梯度直接来自损失函数,隐藏层的梯度需要从输出层往回传。这个过程用计算图来理解会清晰很多——每个操作是一个节点,梯度沿着图反向流动。
提示:手推梯度的时候,强烈建议用数值梯度做校验。具体做法是给某个参数加一个极小扰动,观察损失变化,算出数值梯度,和你推导的解析梯度对比。如果差距在1e-6以内,说明推导正确。这个校验步骤能帮你省下大量debug时间。
2.3 训练循环:看似简单,坑最多
训练循环的骨架就四步:前向计算损失、反向计算梯度、更新参数、清零梯度。但每一步都有讲究。
损失函数的选择要和任务匹配。分类用交叉熵,回归用均方误差,这是常识。但交叉熵配合softmax的时候,数值稳定性是个大问题。直接算log(softmax(x)),当x很大时softmax会溢出。正确的做法是用log-sum-exp技巧,把最大值提出来再算。
梯度更新不只是param -= lr * grad这么简单。学习率太大会震荡,太小会收敛慢。带动量的更新能加速收敛,自适应学习率方法(如Adam)能自动调整每个参数的学习率。从零实现一遍Adam,你会对“一阶矩估计”和“二阶矩估计”这两个概念有完全不同的理解。
# Adam优化器核心逻辑 class Adam: def __init__(self, params, lr=1e-3, beta1=0.9, beta2=0.999, eps=1e-8): self.params = params self.lr = lr self.beta1 = beta1 self.beta2 = beta2 self.eps = eps self.m = [np.zeros_like(p) for p in params] self.v = [np.zeros_like(p) for p in params] self.t = 0 def step(self, grads): self.t += 1 for i, (p, g) in enumerate(zip(self.params, grads)): self.m[i] = self.beta1 * self.m[i] + (1 - self.beta1) * g self.v[i] = self.beta2 * self.v[i] + (1 - self.beta2) * g ** 2 m_hat = self.m[i] / (1 - self.beta1 ** self.t) v_hat = self.v[i] / (1 - self.beta2 ** self.t) p -= self.lr * m_hat / (np.sqrt(v_hat) + self.eps)梯度清零是最容易被遗忘的一步。如果你不清零,梯度会累加,相当于变相增大了batch size,训练动态完全改变。这个坑我踩过不止一次,每次都是loss曲线诡异才想起来。
2.4 推理服务:从模型文件到可用接口
训练出模型只是第一步,把它变成能对外提供服务的接口,是另一个工程问题。
模型加载要考虑冷启动时间。每次请求都重新加载模型,延迟无法接受。正确的做法是服务启动时加载一次,常驻内存。但模型大了显存吃紧,就需要考虑量化、剪枝或者多模型共享底层参数。
请求处理要设计并发模型。同步阻塞式最简单,但吞吐量上不去。异步非阻塞能提高并发,但代码复杂度上升。我的建议是先用同步方式跑通,确认功能正确后再逐步引入异步。
批处理是提升吞吐量的关键手段。把多个请求攒成一个批次一起推理,能充分利用GPU的并行能力。但攒批会引入额外延迟,需要根据业务对延迟的容忍度来定策略。常见做法是设置一个最大等待时间,比如10ms,超时或者攒够一定数量就触发推理。
显存管理是推理服务的命门。你要清楚模型参数占多少、激活值占多少、输入输出占多少。峰值显存出现在哪个环节,决定了你能支持多大的批次。一个实用技巧是预留一部分显存作为缓冲,避免OOM导致服务崩溃。
3. 完整实操:手搓一个可用的图像分类服务
3.1 环境准备与依赖选择
这一节我们从头到尾走一遍完整流程。任务是在MNIST上训练一个分类器,并封装成HTTP接口。
环境方面,Python 3.9以上,NumPy是核心依赖,用来做所有矩阵运算。Flask用来提供HTTP服务,轻量够用。不需要PyTorch或TensorFlow,但如果你想做对照验证,可以装一个CPU版本的PyTorch,只用来验证梯度计算是否正确。
pip install numpy flask pillow数据准备很简单,MNIST可以从公开渠道获取,也可以自己生成一份小规模的模拟数据先跑通流程。我建议先用模拟数据,因为真实数据加载涉及文件解析,容易在早期引入不必要的复杂度。
# 生成模拟数据:1000个样本,每个样本784维,10个类别 np.random.seed(42) X_train = np.random.randn(1000, 784).astype(np.float32) y_train = np.random.randint(0, 10, size=1000) X_test = np.random.randn(200, 784).astype(np.float32) y_test = np.random.randint(0, 10, size=200)3.2 网络定义与前向传播实现
我们定义一个三层网络:输入784维,隐藏层128维,输出10维。激活函数用ReLU,输出层用softmax。
class ThreeLayerNet: def __init__(self, input_dim=784, hidden_dim=128, output_dim=10): self.W1 = he_init(input_dim, hidden_dim) self.b1 = np.zeros(hidden_dim) self.W2 = he_init(hidden_dim, output_dim) self.b2 = np.zeros(output_dim) def forward(self, x): # 第一层:线性变换 + ReLU self.z1 = x @ self.W1 + self.b1 self.a1 = np.maximum(0, self.z1) # 第二层:线性变换 + Softmax self.z2 = self.a1 @ self.W2 + self.b2 # 数值稳定的softmax z2_shifted = self.z2 - np.max(self.z2, axis=1, keepdims=True) exp_z2 = np.exp(z2_shifted) self.probs = exp_z2 / np.sum(exp_z2, axis=1, keepdims=True) return self.probs这里有个细节值得展开:softmax的数值稳定性处理。如果不减去最大值,当z2里有比较大的数时,exp会溢出成inf,整个计算就废了。减去最大值之后,最大的指数变成exp(0)=1,不会溢出,而且数学上等价,因为softmax对输入平移不变。
3.3 反向传播的手动推导与代码实现
反向传播的核心是链式法则。我们从损失函数开始往回推。损失用交叉熵,对于单个样本,损失是-log(probs[正确类别])。
输出层的梯度推导:交叉熵配合softmax有个漂亮的性质,梯度就是probs - one_hot(label)。这个结论建议自己推一遍,推完之后你会对为什么softmax和交叉熵是绝配有一个直观理解。
def backward(self, x, y_true): batch_size = x.shape[0] # 输出层梯度:softmax + 交叉熵的经典结果 dz2 = self.probs.copy() dz2[np.arange(batch_size), y_true] -= 1 dz2 /= batch_size # 第二层参数梯度 dW2 = self.a1.T @ dz2 db2 = np.sum(dz2, axis=0) # 传到隐藏层 da1 = dz2 @ self.W2.T # ReLU的梯度:大于0的地方为1,否则为0 dz1 = da1 * (self.z1 > 0) # 第一层参数梯度 dW1 = x.T @ dz1 db1 = np.sum(dz1, axis=0) return dW1, db1, dW2, db2ReLU的梯度实现有个容易忽略的点:self.z1 > 0产生的是布尔数组,乘上da1时会自动转成0和1。但要注意,如果z1恰好等于0,梯度取0还是1?实践中取0,因为ReLU在0点不可导,取次梯度即可。
3.4 训练过程与超参数选择
训练循环把前面的模块串起来。批次大小选64,学习率从0.001开始试,用Adam优化器。
model = ThreeLayerNet() optimizer = Adam([model.W1, model.b1, model.W2, model.b2], lr=1e-3) for epoch in range(20): loader = SimpleDataLoader(X_train, y_train, batch_size=64) epoch_loss = 0 for batch_x, batch_y in loader: probs = model.forward(batch_x) # 交叉熵损失 loss = -np.mean(np.log(probs[np.arange(len(batch_y)), batch_y] + 1e-8)) epoch_loss += loss grads = model.backward(batch_x, batch_y) optimizer.step(grads) # 每个epoch结束后评估 test_probs = model.forward(X_test) preds = np.argmax(test_probs, axis=1) acc = np.mean(preds == y_test) print(f"Epoch {epoch+1}, Loss: {epoch_loss:.4f}, Test Acc: {acc:.4f}")超参数选择上,学习率是最关键的。我的经验是先用一个较大的值(如0.01)跑几个epoch,如果loss震荡或者变成NaN,就除以10再试。Adam通常比SGD对学习率更宽容,但也不是万能的。批次大小影响梯度估计的方差,太小方差大,太大收敛慢。64到256之间是比较安全的范围。
提示:如果loss在前几个epoch就变成NaN,八成是学习率太大或者数据没归一化。先检查数据范围,确保输入在合理区间(比如均值0方差1),再把学习率降一个数量级。
3.5 封装成HTTP推理接口
训练好的模型要能对外服务。用Flask写一个最简单的接口,接收图片数据,返回预测类别。
from flask import Flask, request, jsonify import numpy as np app = Flask(__name__) model = ThreeLayerNet() # 假设已经训练好并加载了参数 # model.W1, model.b1, ... = load_params() @app.route('/predict', methods=['POST']) def predict(): data = request.get_json() # 输入是784维的列表 x = np.array(data['features'], dtype=np.float32).reshape(1, -1) probs = model.forward(x) pred = int(np.argmax(probs, axis=1)[0]) confidence = float(np.max(probs)) return jsonify({'prediction': pred, 'confidence': confidence}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)这个接口很简陋,但五脏俱全。实际部署时需要考虑的问题包括:输入校验(维度对不对、数值范围是否合理)、错误处理(模型推理抛异常怎么办)、日志记录(每个请求的耗时和结果)、并发控制(同时来100个请求怎么处理)。
4. 踩坑实录:从零构建路上最常见的六个问题
4.1 梯度消失与梯度爆炸的排查思路
这是从零实现反向传播时最常遇到的问题。现象是loss不下降,或者下降几个epoch后突然变成NaN。
排查的第一步是打印每一层梯度的范数。如果某一层的梯度范数接近0,说明梯度消失了;如果大得离谱(比如1e5以上),说明梯度爆炸了。
梯度消失的常见原因:激活函数选择不当(sigmoid在饱和区梯度接近0)、网络太深、初始化太小。解决办法:换ReLU、用He初始化、加残差连接。
梯度爆炸的常见原因:初始化太大、学习率太高、损失函数没有做数值稳定处理。解决办法:梯度裁剪(把梯度范数限制在一个阈值内)、降低学习率、检查softmax和log的实现。
# 梯度裁剪示例 def clip_gradients(grads, max_norm=5.0): total_norm = np.sqrt(sum(np.sum(g**2) for g in grads)) if total_norm > max_norm: scale = max_norm / (total_norm + 1e-8) grads = [g * scale for g in grads] return grads4.2 数值稳定性:那些让你loss变NaN的隐形杀手
除了softmax溢出,还有几个常见的数值陷阱。
对数运算:log(0)是负无穷,log(负数)是NaN。计算交叉熵时,概率值要加一个极小量(如1e-8)防止log(0)。
除法:分母为0会产生inf或NaN。归一化的时候,标准差可能为0,要加epsilon。
指数运算:exp大数溢出,前面softmax已经处理了。但如果你自己实现sigmoid,也要注意同样的问题。
累加:浮点数累加误差会累积。如果要做大量累加,用Kahan求和或者定期重置累加器。
注意:NaN一旦出现,会像病毒一样传播。一个NaN梯度更新后,所有参数都变NaN,整个模型就废了。所以要在关键位置加检查,比如每次更新前检查梯度是否包含NaN,一旦发现立即停止并报警。
4.3 训练不收敛的五个常见原因速查
| 现象 | 可能原因 | 排查方法 | 解决办法 |
|---|---|---|---|
| loss震荡不下降 | 学习率太大 | 打印每步loss | 降低学习率10倍 |
| loss下降但很慢 | 学习率太小或初始化不好 | 检查梯度范数 | 调大学习率或换初始化 |
| loss下降后反弹 | 过拟合 | 对比训练和验证loss | 加正则化或早停 |
| loss一直很高 | 标签错误或损失函数用错 | 检查数据标签 | 修正标签或换损失函数 |
| loss变NaN | 数值溢出或梯度爆炸 | 检查中间值范围 | 加数值稳定处理或梯度裁剪 |
这张表是我自己debug时总结的,覆盖了八成以上的训练问题。遇到问题先查表,能省不少时间。
4.4 推理服务的性能瓶颈定位
服务上线后如果延迟高,按这个顺序排查。
先看模型推理本身耗时多少。在代码里打时间戳,把前向传播的耗时单独统计出来。如果这部分就占了大部分时间,说明模型太大或者批次太小,GPU没吃满。
再看数据预处理耗时。图像解码、归一化、维度变换,这些操作看起来快,但累积起来可能很可观。尤其是如果每次请求都重新初始化一些对象,开销更大。
然后看请求排队耗时。如果并发高,请求在队列里等待的时间可能超过推理本身。这时候要考虑加机器或者优化批处理策略。
最后看网络传输耗时。请求体和响应体的大小,如果传输的是原始图像,可能比推理还慢。考虑压缩或者只传必要字段。
4.5 显存不够用的几种应对策略
显存不够是推理服务最常见的资源问题。应对策略按优先级排列。
降低批次大小是最直接的办法,但会降低吞吐量。使用半精度(FP16)能把显存占用减半,但要注意数值范围是否够用。模型量化把参数从FP32压到INT8,显存降到四分之一,但精度可能损失。梯度检查点用计算换显存,适合训练阶段。模型并行把不同层放到不同设备上,适合超大模型。
我的建议是先用半精度,成本最低效果最明显。如果还不够,再考虑量化。模型并行复杂度高,除非必要不轻易上。
4.6 从单机到服务的工程化 checklist
最后给一份上线前的检查清单,照着过一遍能避开大部分坑。
- 模型文件是否独立于代码,方便更新
- 服务启动时是否预加载模型,避免首次请求超时
- 输入是否做了维度和范围校验
- 是否有超时机制,防止单个请求卡死整个服务
- 是否记录了每个请求的耗时和结果,便于排查
- 是否有健康检查接口,方便负载均衡器探活
- 显存使用是否有监控和告警
- 是否做了压力测试,知道单机能扛多少QPS
- 模型更新时是否支持热加载,不中断服务
- 是否有降级策略,模型服务挂了怎么办
这份清单是我在多次上线中逐步积累的,每次上线前过一遍,能省下不少半夜被叫起来修服务的痛苦。
我个人在实际操作中的体会是,从零构建最大的价值不在于你写出了多高效的代码,而在于你建立了一套完整的心智模型。当你知道每一步在干什么、为什么这么干、哪里可能出问题,你再用回那些高层框架的时候,感觉是完全不一样的。你不再是一个调包侠,而是一个真正理解系统的人。这个转变,值得你花时间走一遍。