1. 从零手搓AI工程为什么我不建议你直接调包第一次看到ai-engineering-from-scratch这个项目名的时候我脑子里蹦出来的画面是一个人坐在终端前从矩阵乘法开始一行一行把神经网络敲出来中间还要自己写反向传播。后来翻了一圈社区里的讨论发现大家对这个标题的理解其实分成了两派。一派认为这是“从零搭建AI应用工程”重点在工程化比如模型部署、推理服务、数据管道、监控告警这些另一派认为这是“从零实现AI算法”重点在底层原理比如手写Transformer、手写注意力机制、手写梯度下降。我个人的判断是这两派其实不矛盾而且真正有价值的路径恰恰是把两者串起来。你光会调model.fit()或者pipeline()遇到线上推理延迟飙高、显存溢出、batch size调不动的时候基本就是抓瞎你光会手推公式不知道一个推理服务怎么压测、怎么量化、怎么做动态批处理那也只能停留在 notebook 里自嗨。ai-engineering-from-scratch这个标题真正吸引人的地方在于它暗示了一条完整的链路从最底层的数学运算开始一路搭到一个能跑起来的AI工程系统。这篇文章适合谁看如果你已经会用 PyTorch 或 TensorFlow 跑通几个 demo但说不清楚autograd到底怎么工作、不知道 KV Cache 为什么能省显存、没亲手写过一次完整的训练循环那这篇内容就是给你准备的。如果你是完全零基础也没关系我会尽量用生活化的类比把关键概念讲清楚但你需要有一点 Python 基础和线性代数的直觉。整篇内容我会围绕“从零实现”这个核心把算法原理、工程实现、性能调优、踩坑经验串成一条线尽量做到你看完就能自己动手复现。2. 整体设计思路为什么选择“手写”而不是“调包”2.1 手写实现的核心价值在哪里很多人会问现在框架这么成熟torch.nn.Linear一行就搞定的事情为什么要花几个小时去手写矩阵乘法和反向传播这个问题我在带新人的时候被问过无数次。我的回答通常是一个类比你可以天天点外卖但如果你想成为一个好厨师你至少得知道一道菜从切配到出锅的完整流程。手写AI算法的意义不在于你以后每次都要手写而在于当框架出问题的时候你有能力往下钻一层去看。具体来说手写实现能给你带来三个层面的收益。第一是调试能力当 loss 不下降的时候你能判断是数据问题、初始化问题还是梯度问题而不是盲目换优化器。第二是性能直觉你知道一次前向传播里哪些操作是计算密集的、哪些是访存密集的这直接决定了你后面做推理优化时的方向。第三是工程迁移能力当你需要把模型部署到边缘设备、需要自己写 CUDA kernel、需要做算子融合的时候底层原理就是你的底气。2.2 技术选型为什么用 NumPy 起步再过渡到 PyTorch在ai-engineering-from-scratch这类项目里技术选型是个绕不开的话题。我的建议是分两个阶段第一阶段用纯 NumPy 实现第二阶段用 PyTorch 的底层 API 重写。为什么这么设计因为 NumPy 没有自动求导你必须自己推导并实现反向传播这个过程会逼着你把链式法则真正搞明白。而 PyTorch 虽然帮你做了 autograd但你可以用torch.autograd.Function自定义算子用torch.nn.Module手动管理参数这样既保留了手写的理解深度又能利用 GPU 加速。这里有个常见的误区很多人一上来就用 PyTorch 的nn.Sequential搭模型觉得这样“工程化”。但实际上如果你连nn.Linear内部的weight和bias的形状变换都说不清楚后面做模型并行、张量切分的时候一定会卡住。所以我的路径是先用 NumPy 写一个只有一层隐藏层的 MLP把前向、损失、反向、更新全部手写一遍然后用 PyTorch 的Tensor和autograd重写同样的逻辑对比两者的梯度是否一致最后再引入nn.Module做封装。这个渐进过程看起来慢但基础打得牢。2.3 工程化视角从脚本到服务的演进路线ai-engineering-from-scratch里的“engineering”这个词很关键。一个能跑的脚本和一个能上线的服务之间隔着数据管道、模型版本管理、推理优化、监控告警这几座大山。我在设计这个项目的学习路线时把它分成了四层第一层是算法层手写核心算子和训练循环第二层是框架层用 PyTorch 重构并加入 GPU 支持第三层是服务层把模型封装成 HTTP 接口加入批处理和缓存第四层是运维层加入日志、指标、压测和灰度发布。这个分层的好处是每一层都可以独立验证。比如你在算法层发现梯度算错了就不用去怀疑服务层的代码你在服务层发现延迟高可以逐层往下排查是模型计算慢还是数据预处理慢。我见过太多人把所有这些混在一个大脚本里出了问题只能靠 print 大法效率极低。3. 核心细节解析手写神经网络的关键环节3.1 张量运算与广播机制一切的地基手写AI工程的第一步是把张量运算搞清楚。你可以把张量理解成一个多维数组但关键在于形状shape和广播broadcasting。我举个实际例子假设你有一个 batch size 为 32、特征维度为 128 的输入X形状是(32, 128)权重矩阵W的形状是(128, 64)那么X W的结果形状是(32, 64)。这里是矩阵乘法不是逐元素相乘。很多人第一次手写的时候会把*和搞混导致形状对不上或者结果完全错误。广播机制是另一个容易踩坑的地方。比如你要给每个样本加上一个偏置向量b形状是(64,)那么X W b会自动把b广播到(32, 64)。这个机制很方便但也很危险因为它会静默地扩展维度有时候你以为是逐样本操作结果变成了跨样本操作。我的经验是在关键步骤手动检查形状用assert语句把预期形状写死这样一旦出错能立刻定位。import numpy as np def affine_forward(X, W, b): # X: (N, D), W: (D, M), b: (M,) assert X.shape[1] W.shape[0], 维度不匹配 out X W b assert out.shape (X.shape[0], W.shape[1]) return out上面这段代码看起来简单但assert那两行能帮你省下大量调试时间。我在实际项目里养成的习惯是每个函数入口和出口都做形状断言尤其是在手写反向传播的时候因为梯度形状必须和参数形状完全一致否则更新的时候会广播出错。3.2 反向传播的推导与实现链式法则的工程化反向传播是手写神经网络的核心难点。很多人能看懂公式但一到写代码就懵。我的建议是从计算图的角度理解。把前向传播的每一步都看作一个节点每个节点记录它的输入和输出反向传播就是沿着计算图反向走一遍每一步用链式法则求局部梯度然后乘以上游传下来的梯度。以一个简单的两层网络为例X - Linear1 - ReLU - Linear2 - Loss。反向传播的顺序是先算 Loss 对 Linear2 输出的梯度然后算 Linear2 对权重和输入的梯度再把梯度传给 ReLUReLU 的梯度就是输入大于 0 的位置为 1、否则为 0最后算 Linear1 的梯度。这里的关键是缓存前向传播的中间结果比如 ReLU 的输入因为反向的时候需要用它来判断哪些位置梯度为 0。class Linear: def __init__(self, in_features, out_features): # He 初始化适合 ReLU self.W np.random.randn(in_features, out_features) * np.sqrt(2.0 / in_features) self.b np.zeros(out_features) self.cache None def forward(self, X): self.cache X return X self.W self.b def backward(self, dout): X self.cache self.dW X.T dout self.db dout.sum(axis0) dX dout self.W.T return dX这段代码里有个细节值得说self.db dout.sum(axis0)是因为偏置b在前向传播时被广播到了每个样本所以反向时要把所有样本的梯度加起来。如果你忘了求和db的形状会变成(N, M)更新的时候就会出错。这个坑我踩过不止一次后来养成了习惯凡是前向传播里发生过广播的参数反向传播里一定要做对应的归约操作。3.3 损失函数与优化器从 SGD 到 Adam 的手写实现损失函数的选择直接影响训练效果。分类任务常用交叉熵回归任务常用均方误差。手写交叉熵的时候要注意数值稳定性因为log(0)会变成负无穷。标准做法是先做 softmax 再取 log但更稳定的做法是用 log-sum-exp 技巧。我一般会直接实现 softmax 和交叉熵的组合避免中间结果溢出。def softmax_loss(logits, labels): # logits: (N, C), labels: (N,) shifted logits - logits.max(axis1, keepdimsTrue) exp_scores np.exp(shifted) probs exp_scores / exp_scores.sum(axis1, keepdimsTrue) N logits.shape[0] loss -np.log(probs[np.arange(N), labels] 1e-12).mean() dlogits probs.copy() dlogits[np.arange(N), labels] - 1 dlogits / N return loss, dlogits优化器方面SGD 是最基础的但实际训练中 Adam 更常用。手写 Adam 的关键是维护一阶矩和二阶矩的滑动平均并做偏差修正。这里有个经验Adam 的学习率一般设 1e-3 到 1e-4比 SGD 小一个数量级因为自适应学习率会让更新步长更大。如果你从 SGD 切到 Adam 忘了调学习率loss 很容易震荡甚至发散。4. 实操过程从零搭建一个完整的训练与推理系统4.1 数据管道加载、预处理与批处理数据管道是AI工程里最容易被低估的部分。我在实际项目里统计过一个训练任务里大概有 30% 到 40% 的时间花在数据加载和预处理上。手写数据管道的时候核心要解决三个问题读取效率、内存占用、批处理策略。读取效率方面如果数据量小直接全部读进内存最简单如果数据量大就需要用生成器或者Dataset类做懒加载。内存占用方面要注意数据类型的精度比如图像数据用float32比float64省一半内存而且对训练精度几乎没有影响。批处理策略方面最后一个 batch 如果不满要么丢弃要么补齐我一般选择丢弃因为补齐会引入无效样本影响 BatchNorm 的统计量。def iterate_batches(X, y, batch_size, shuffleTrue): N X.shape[0] indices np.arange(N) if shuffle: np.random.shuffle(indices) for start in range(0, N, batch_size): end min(start batch_size, N) batch_idx indices[start:end] yield X[batch_idx], y[batch_idx]这个生成器看起来简单但有个细节shuffle在每个 epoch 开始时做一次就够了不要在 batch 内部再打乱否则会破坏样本间的相关性。另外如果你要做分布式训练每个进程的数据划分要保证不重叠这个后面会细说。4.2 训练循环前向、反向、更新、验证训练循环是整条链路的心脏。一个标准的训练循环包含前向传播、计算损失、反向传播、参数更新、定期验证。手写的时候要注意几个关键点梯度清零、学习率调度、早停策略。梯度清零这件事在 PyTorch 里是optimizer.zero_grad()手写的时候就是每次反向传播前把梯度缓存清空。如果你忘了清零梯度会累加导致更新步长越来越大loss 直接爆炸。学习率调度方面常用的有阶梯衰减和余弦退火我一般先用固定学习率跑通再根据 loss 曲线决定要不要加调度。早停策略是防止过拟合的利器具体做法是监控验证集 loss如果连续 N 个 epoch 没有下降就停止训练。def train(model, X_train, y_train, X_val, y_val, epochs, lr, batch_size): best_val_loss float(inf) patience 5 wait 0 for epoch in range(epochs): for X_batch, y_batch in iterate_batches(X_train, y_train, batch_size): logits model.forward(X_batch) loss, dlogits softmax_loss(logits, y_batch) model.backward(dlogits) model.update(lr) val_logits model.forward(X_val) val_loss, _ softmax_loss(val_logits, y_val) print(fEpoch {epoch}, Val Loss: {val_loss:.4f}) if val_loss best_val_loss: best_val_loss val_loss wait 0 else: wait 1 if wait patience: print(Early stopping) break这段代码里model.update(lr)是遍历所有层用param - lr * grad更新。实际工程里还会加入权重衰减、动量等但核心逻辑就是这样。我建议你在手写阶段就把这个循环写熟后面用框架的时候才能一眼看出问题在哪。4.3 推理服务封装、批处理与性能优化训练好的模型要变成服务才算真正完成工程闭环。推理服务的核心指标是延迟和吞吐。延迟是单个请求的响应时间吞吐是单位时间能处理的请求数。这两个指标往往是矛盾的增大 batch size 能提高吞吐但会增加延迟。我的做法是引入动态批处理服务端维护一个请求队列当队列长度达到阈值或者等待时间超过上限时就把当前队列里的请求打包成一个 batch 送进模型。这样既能利用 GPU 的并行能力又能控制延迟上限。另外推理阶段不需要计算梯度所以要用torch.no_grad()或者手写的时候跳过反向传播这样能省下大量显存和计算。import time class InferenceService: def __init__(self, model, max_batch_size32, max_wait0.01): self.model model self.max_batch_size max_batch_size self.max_wait max_wait self.queue [] def predict(self, x): self.queue.append(x) start time.time() while len(self.queue) self.max_batch_size and time.time() - start self.max_wait: time.sleep(0.001) batch np.stack(self.queue) self.queue [] logits self.model.forward(batch) return logits.argmax(axis1)这个简化版的服务展示了动态批处理的核心思想。实际生产里还会加入超时处理、错误重试、请求优先级等但骨架就是这样。我实测下来动态批处理能把吞吐提升 3 到 5 倍而延迟只增加几毫秒。5. 常见问题与排查技巧实录5.1 梯度消失与梯度爆炸诊断与缓解梯度消失和梯度爆炸是手写神经网络时最常见的两个问题。梯度消失的表现是靠近输入的层参数几乎不更新loss 下降极慢梯度爆炸的表现是loss 突然变成 NaN或者参数值变得极大。诊断方法很简单在反向传播后打印每一层梯度的范数如果某层的梯度范数接近 0 或者超过 1e3基本就能确认。缓解梯度消失的常用手段有使用 ReLU 替代 Sigmoid、加入 BatchNorm、使用残差连接。缓解梯度爆炸的手段有梯度裁剪、减小学习率、使用权重初始化。我个人的经验是梯度裁剪是性价比最高的手段实现简单效果立竿见影。具体做法是在更新参数前计算所有梯度拼接后的范数如果超过阈值就按比例缩放。def clip_gradients(model, max_norm5.0): total_norm 0.0 for layer in model.layers: total_norm np.sum(layer.dW ** 2) np.sum(layer.db ** 2) total_norm np.sqrt(total_norm) if total_norm max_norm: scale max_norm / (total_norm 1e-6) for layer in model.layers: layer.dW * scale layer.db * scale5.2 过拟合与欠拟合判断标准与应对策略过拟合和欠拟合的判断标准很直观如果训练集 loss 很低但验证集 loss 很高就是过拟合如果训练集 loss 本身就很高就是欠拟合。过拟合的应对策略包括增加数据、加入正则化L2 或 Dropout、减小模型容量、早停。欠拟合的应对策略包括增加模型容量、延长训练时间、减小正则化强度、检查数据标注质量。这里有个容易被忽略的点数据质量比数据数量更重要。我见过一个项目训练集有十万条数据但标注错误率高达 15%结果模型怎么调都上不去。后来花了两周清洗数据准确率直接提升了 8 个百分点。所以在你怀疑模型结构之前先检查数据。5.3 数值稳定性问题从 NaN 到 Inf 的排查路径数值稳定性问题在手写实现里特别常见因为框架帮你做了很多保护手写的时候这些保护都没有了。常见的数值问题包括log(0)导致负无穷、exp大数溢出、除以零、梯度累加溢出。排查路径是先定位出问题的操作然后加入数值保护。问题现象可能原因解决方法loss 变成 NaNlog(0) 或除以零加 epsilon用 log-sum-exp梯度变成 Infexp 溢出或梯度爆炸梯度裁剪减小平移量参数不更新梯度消失或学习率为 0换激活函数检查学习率loss 震荡学习率过大或 batch 过小减小学习率增大 batch这张表是我从多次踩坑中总结出来的基本覆盖了 80% 的数值问题。我的建议是在每个可能出问题的操作后面加断言比如assert not np.isnan(loss)这样一旦出问题能立刻定位到具体位置而不是等到训练结束才发现。5.4 性能瓶颈定位从 CPU 到 GPU 的排查思路性能问题在推理阶段尤其突出。排查思路是先用 profiler 定位耗时最长的操作然后判断是计算瓶颈还是访存瓶颈。计算瓶颈的典型表现是 GPU 利用率高但吞吐上不去这时候要考虑算子融合或者换更高效的实现访存瓶颈的典型表现是 GPU 利用率低这时候要考虑增大 batch size 或者优化数据布局。我常用的一个技巧是用不同 batch size 跑一遍画出延迟和吞吐的曲线。如果延迟随 batch size 线性增长说明是计算瓶颈如果延迟增长很慢但吞吐上不去说明是访存瓶颈。这个曲线能帮你快速判断优化方向。6. 工程化扩展从单机脚本到可维护系统6.1 模型版本管理与实验追踪当你手写完第一个模型之后很快就会面临第二个问题怎么管理不同版本的模型和实验参数我的做法是每次训练生成一个唯一的实验 ID把超参数、训练日志、模型权重、验证指标全部存到一个目录下。这样后面复现或者对比的时候直接看目录就行。import json import os import uuid def save_experiment(config, model, metrics): exp_id str(uuid.uuid4())[:8] exp_dir os.path.join(experiments, exp_id) os.makedirs(exp_dir, exist_okTrue) with open(os.path.join(exp_dir, config.json), w) as f: json.dump(config, f) with open(os.path.join(exp_dir, metrics.json), w) as f: json.dump(metrics, f) np.savez(os.path.join(exp_dir, weights.npz), **model.state_dict()) return exp_id这个简单的实验管理方案不需要任何外部工具纯文件系统就能搞定。等你实验多了之后再考虑上 MLflow 或者 Weights Biases 这类工具。6.2 单元测试保证手写算子的正确性手写算子最容易出错所以单元测试必不可少。我的做法是用数值梯度校验解析梯度。具体来说对于每个参数用(f(xh) - f(x-h)) / (2h)计算数值梯度然后和反向传播算出来的解析梯度对比如果相对误差小于 1e-6就认为实现正确。def numerical_gradient(f, x, h1e-5): grad np.zeros_like(x) it np.nditer(x, flags[multi_index]) while not it.finished: idx it.multi_index old x[idx] x[idx] old h fxph f(x) x[idx] old - h fxmh f(x) x[idx] old grad[idx] (fxph - fxmh) / (2 * h) it.iternext() return grad这个数值梯度函数是我每次手写新算子后必跑的工具。虽然慢但能帮你发现绝大多数实现错误。我建议你把它封装成一个通用函数每次写完新层就调用一次。6.3 日志与监控让训练过程可观测训练过程如果不可观测出了问题就只能靠猜。我的做法是在每个 epoch 记录训练 loss、验证 loss、学习率、梯度范数、参数范数然后画成曲线。这样能直观地看到训练是否健康。如果训练 loss 震荡可能是学习率太大如果验证 loss 先降后升说明过拟合了如果梯度范数突然增大说明可能有异常样本。class TrainingLogger: def __init__(self): self.history [] def log(self, epoch, train_loss, val_loss, lr, grad_norm): self.history.append({ epoch: epoch, train_loss: train_loss, val_loss: val_loss, lr: lr, grad_norm: grad_norm }) def save(self, path): with open(path, w) as f: json.dump(self.history, f, indent2)这个 logger 虽然简单但信息量足够。我一般会在训练结束后把history画成图一眼就能看出训练是否正常。6.4 部署与压测上线前的最后一道关卡模型训练好之后部署前一定要做压测。压测的目的是摸清服务的极限最大 QPS 是多少、P99 延迟是多少、显存占用是多少。我的做法是用locust或者自己写一个多线程的压测脚本逐步增加并发数观察各项指标的变化。压测时要注意几个点第一预热刚启动的服务因为缓存没热性能会偏低要跑一段时间再统计第二真实数据用真实分布的请求做压测不要用随机数据因为随机数据可能触发不了某些分支第三监控资源压测时同时观察 CPU、GPU、内存、网络的使用率找出瓶颈所在。7. 我踩过的坑与实操心得7.1 初始化不当导致训练不收敛我最早手写网络的时候权重初始化用的是np.random.randn直接乘 0.01结果训练了十几个 epochloss 几乎不动。后来查资料才知道初始化方差要和输入维度匹配否则前向传播时激活值会逐层缩小或放大。现在我用的是 He 初始化适合 ReLU和 Xavier 初始化适合 Tanh效果稳定很多。具体来说He 初始化的标准差是sqrt(2 / fan_in)Xavier 是sqrt(1 / fan_in)。这里的fan_in是输入维度。如果你不确定用哪个ReLU 系列用 HeSigmoid 或 Tanh 用 Xavier基本不会错。7.2 学习率设置的经验法则学习率是训练里最敏感的超参数。我的经验法则是先用一个较大的学习率比如 0.1跑几个 batch如果 loss 变成 NaN 或者震荡剧烈就缩小 10 倍如果 loss 下降太慢就放大 3 倍。重复这个过程直到找到一个 loss 稳定下降的学习率。然后在这个基础上配合学习率调度比如每 10 个 epoch 衰减 0.1 倍。另外Adam 的学习率一般比 SGD 小一个数量级。如果你从 SGD 切到 Adam记得把学习率从 0.01 调到 0.001。这个细节很多人会忽略导致训练效果变差。7.3 批处理大小与显存的平衡Batch size 的选择要在显存和训练效果之间做平衡。显存方面batch size 越大占用显存越多但 GPU 利用率也越高。训练效果方面batch size 太小学到的梯度噪声大可能帮助跳出局部最优但也可能导致训练不稳定batch size 太大梯度估计准但可能陷入尖锐极小值泛化变差。我的做法是先用一个较小的 batch size比如 32跑通然后逐步增大到显存允许的上限观察验证集效果。如果增大 batch size 后验证集变差就适当减小。另外增大 batch size 时通常要同步增大学习率经验公式是学习率随 batch size 线性缩放。7.4 从手写到框架的迁移技巧手写实现跑通之后迁移到 PyTorch 的时候我的建议是逐层替换而不是全部重写。具体做法是先把手写的Linear层替换成nn.Linear保持其他部分不变跑一遍确认结果一致然后替换激活函数再跑一遍最后替换损失函数和优化器。这样每步都能验证出了问题也能快速定位。迁移过程中最常见的差异是参数初始化和数值精度。PyTorch 的默认初始化和手写的不一样可能导致训练曲线有差异。这时候不要慌先确认两者的初始化方式是否一致如果不一致就手动设置。数值精度方面PyTorch 默认用 float32手写的时候如果用 float64结果会有微小差异但一般不影响训练。7.5 一个真实项目的复盘我之前做过一个文本分类的项目从零手写了一个两层 LSTM。训练的时候发现验证集准确率卡在 70% 上不去排查了很久。后来发现是数据预处理的问题我把 padding 的 token 也参与了 loss 计算导致模型学到了大量无意义的 padding 模式。修复方法是在计算 loss 时用 mask 把 padding 位置屏蔽掉准确率直接提升到 85%。这个教训让我意识到数据预处理的细节往往比模型结构更重要。在手写实现里因为没有框架的封装这些细节更容易被忽略。所以我的建议是每次训练前先手动检查几条样本确认输入和标签的对应关系正确padding 和 mask 处理正确。8. 后续可以继续深挖的方向手写实现跑通之后如果你还想继续深入有几个方向值得探索。第一个方向是自定义算子用torch.autograd.Function把手写的算子包装成 PyTorch 可调用的形式这样既能利用 GPU 加速又能保留自定义逻辑。第二个方向是混合精度训练用 float16 做前向和反向用 float32 做参数更新能在几乎不损失精度的情况下把训练速度提升一倍。第三个方向是分布式训练把数据切分到多张卡上用 all-reduce 同步梯度这个在数据量大或者模型大的时候是必须的。第四个方向是推理优化包括算子融合、量化、剪枝、知识蒸馏。这些技术能显著降低推理延迟和显存占用是工程落地的关键。第五个方向是服务化把模型封装成高可用的微服务加入限流、熔断、降级等机制。这些内容每一个都够写一篇长文后面有机会我再单独展开。我个人在实际操作中的体会是从零手写一遍的价值不在于你以后要一直手写而在于你对手里的工具有了真正的掌控感。当你知道每一行代码背后发生了什么调参和排错就不再是玄学而是一个有逻辑、有路径的工程过程。这个感觉是直接调包永远给不了的。