从零手搓AI工程：深入理解核心链路，告别调包困境

1. 从零手搓AI工程：为什么我不建议你直接调包

第一次看到ai-engineering-from-scratch这个项目名的时候，我正坐在工位上啃一个调了三天的模型部署脚本。那会儿我的日常就是pip install一堆框架，然后对着报错信息发呆，改改参数、换换版本，运气好跑通了就上线，运气不好就继续在依赖地狱里打转。直到有天leader问我一句：“你这个推理服务为什么延迟这么高？”我支支吾吾半天，只能说“可能是框架本身的问题”——其实我根本不知道框架内部发生了什么。

这就是我决定从零开始重写一遍AI工程核心链路的原因。ai-engineering-from-scratch不是一个具体的开源库，而是一种学习路径和工程实践思路：把AI应用从数据到推理的每一个环节，用最朴素的方式亲手实现一遍。它解决的核心问题是——当你只会调包的时候，你永远不知道性能瓶颈在哪、模型为什么崩、线上为什么和本地表现不一致。这个内容适合所有已经会用PyTorch或TensorFlow跑demo，但一遇到工程化问题就抓瞎的开发者；也适合那些想转AI工程方向、但被各种框架抽象层绕晕的后端同学。

我花了大概两个月时间，把一条完整的AI工程链路拆成了六个模块：数据处理、特征工程、模型训练、模型压缩、推理服务、监控与迭代。每个模块我都先用纯Python和NumPy实现一遍，再对比主流框架的做法。踩过的坑比我过去两年加起来都多，但收获也是实打实的——现在线上服务延迟降了60%，模型体积压到原来的四分之一，最关键的是，出了问题我能自己定位了。

下面我把整个从零构建的过程拆开讲，包括每一步的设计思路、核心代码、参数计算，以及那些只有亲手写过才会知道的坑。

2. 整体架构设计与技术选型思路

2.1 为什么选择“从零实现”而不是“直接调包”

很多人会问：现在PyTorch、TensorFlow、ONNX Runtime这么成熟，为什么还要自己写一遍？这不是重复造轮子吗？

我的回答是：造轮子不是为了替代轮子，而是为了理解轮子。你不需要在生产环境用自己写的矩阵乘法，但你需要知道矩阵乘法在CPU和GPU上的内存访问模式差异；你不需要自己实现一个完整的Transformer，但你需要知道注意力机制的计算复杂度为什么是O(n²)，以及怎么通过分块计算把它降到可接受的范围。

从工程角度看，直接调包有三个致命问题：

• 黑盒调试困难：模型输出异常时，你无法判断是数据问题、模型结构问题还是框架本身的bug。自己实现过一遍之后，你能快速定位到具体环节。
• 性能优化无方向：框架帮你做了太多自动优化，导致你不知道哪些操作是昂贵的。亲手写过之后，你会对内存拷贝、类型转换、算子融合有直观感受。
• 部署环境受限：某些边缘设备或特殊硬件上，主流框架可能跑不起来。这时候你需要有能力手写一个轻量级推理引擎。

我选择的技术栈很朴素：Python 3.10 + NumPy做基础计算，Cython做热点代码加速，Flask做服务框架，Prometheus做监控。没有用任何深度学习框架，所有模型都是手写的前向和反向传播。

2.2 模块划分与依赖关系

整个项目分为六个核心模块，依赖关系是单向的：

这个划分方式的好处是每个模块可以独立测试和替换。比如你不想用我写的量化方法，可以换成ONNX的量化工具，只要接口对齐就行。

2.3 核心设计原则

在动手之前，我定了三条原则：

第一，接口先行。每个模块的输入输出格式先定义清楚，用类型注解和dataclass约束。这样即使实现换了，上下游不用改。

第二，性能可测量。每个模块都要有基准测试，记录耗时、内存占用、CPU利用率。没有测量就没有优化。

第三，失败可复现。所有随机操作都要固定种子，所有配置都要版本化。这样出了问题能精确复现。

提示：如果你也想走一遍这个路径，建议先从推理服务模块开始。因为它是最终产出，能让你快速看到成果，有成就感之后再去啃训练和压缩。

3. 核心模块的从零实现细节

3.1 数据处理：别小看一个DataLoader

数据处理看起来最简单，但实际上是坑最多的环节。我一开始觉得不就是读文件、转数组吗？结果第一个版本就遇到了内存爆炸的问题。

问题场景：我有一个50GB的图片数据集，直接全部加载到内存肯定不行。常规做法是用PyTorch的DataLoader，它帮你做了懒加载和多进程预取。但我要从零实现，就得自己处理。

我的方案是实现一个三级缓冲的DataLoader：

class ScratchDataLoader: def __init__(self, data_paths, batch_size=32, num_workers=4, prefetch_factor=2): self.data_paths = data_paths self.batch_size = batch_size self.num_workers = num_workers self.prefetch_factor = prefetch_factor self._queue = queue.Queue(maxsize=prefetch_factor * num_workers) def _worker(self, worker_id): # 每个worker负责一部分数据 for i in range(worker_id, len(self.data_paths), self.num_workers): batch = self._load_batch(self.data_paths[i:i+self.batch_size]) self._queue.put(batch) def __iter__(self): # 启动worker进程 workers = [multiprocessing.Process(target=self._worker, args=(i,)) for i in range(self.num_workers)] for w in workers: w.start() # 从队列取数据 while True: try: yield self._queue.get(timeout=1) except queue.Empty: break

这里的关键参数是prefetch_factor，它决定了预取多少批数据。设太小会导致GPU等数据，设太大会占内存。我的经验值是：prefetch_factor = 2 到 4 之间，具体取决于你的数据加载速度和模型计算速度的比值。

计算方式很简单：假设加载一批数据需要t_load秒，模型计算一批需要t_compute秒。如果t_load > t_compute，说明数据加载是瓶颈，需要增加worker数量；如果t_load < t_compute，说明计算是瓶颈，prefetch_factor设为2就够了。

我实测下来，对于图片数据，num_workers=4、prefetch_factor=2是一个比较稳的配置。再多的话，进程间通信的开销会抵消掉并行加载的收益。

注意事项：多进程加载时，每个worker都会复制一份数据索引，如果索引本身很大（比如几百万条），内存占用会成倍增加。解决办法是用共享内存或者内存映射文件。

3.2 特征工程：手写标准化和PCA

特征工程这一步，我实现了两个最常用的操作：标准化和PCA降维。别看这两个操作简单，手写一遍能让你理解很多细节。

标准化的坑：常规公式是(x - mean) / std。但这里有个问题：如果std为0怎么办？如果数据里有NaN怎么办？

我的实现：

def standardize(X, mean=None, std=None, eps=1e-8): if mean is None: mean = np.nanmean(X, axis=0) if std is None: std = np.nanstd(X, axis=0) # 防止除零 std = np.where(std < eps, eps, std) # 处理NaN X_normalized = (X - mean) / std X_normalized = np.nan_to_num(X_normalized, nan=0.0) return X_normalized, mean, std

这里eps=1e-8是一个经验值，太小了会导致数值不稳定，太大了会损失精度。我试过1e-6和1e-10，最后选了1e-8。

PCA的实现：PCA的核心是特征值分解。但直接对协方差矩阵做特征值分解，当特征维度很高时（比如几万维），计算量会非常大。我的做法是先做SVD：

def pca(X, n_components): # 中心化 X_centered = X - np.mean(X, axis=0) # SVD分解 U, S, Vt = np.linalg.svd(X_centered, full_matrices=False) # 取前n_components个主成分 components = Vt[:n_components] # 计算解释方差比 explained_variance_ratio = (S ** 2) / np.sum(S ** 2) return components, explained_variance_ratio[:n_components]

SVD的好处是不需要显式计算协方差矩阵，数值稳定性更好。但SVD的计算复杂度是O(min(mn², m²n))，当样本数和特征数都很大时，还是需要用到随机SVD或者增量PCA。

我实测下来，对于10万样本、1000维特征的数据，完整SVD大概需要30秒，而随机SVD只需要3秒，精度损失在1%以内。所以如果你的数据量很大，建议用随机SVD。

3.3 模型训练：手写反向传播和优化器

这是整个项目最核心的部分。我实现了一个两层的全连接网络，包含ReLU激活和Softmax输出。虽然结构简单，但涵盖了反向传播的所有关键点。

前向传播：

def forward(X, W1, b1, W2, b2): Z1 = X @ W1 + b1 A1 = np.maximum(0, Z1) # ReLU Z2 = A1 @ W2 + b2 # Softmax exp_Z2 = np.exp(Z2 - np.max(Z2, axis=1, keepdims=True)) A2 = exp_Z2 / np.sum(exp_Z2, axis=1, keepdims=True) return Z1, A1, Z2, A2

这里np.max(Z2, axis=1, keepdims=True)是为了数值稳定性，防止exp溢出。这个技巧在教科书里经常被忽略，但实际工程中必须加上。

反向传播：

def backward(X, Y, Z1, A1, Z2, A2, W1, W2): m = X.shape[0] # 输出层梯度 dZ2 = A2 - Y dW2 = A1.T @ dZ2 / m db2 = np.sum(dZ2, axis=0) / m # 隐藏层梯度 dA1 = dZ2 @ W2.T dZ1 = dA1 * (Z1 > 0) # ReLU导数 dW1 = X.T @ dZ1 / m db1 = np.sum(dZ1, axis=0) / m return dW1, db1, dW2, db2

ReLU的导数就是(Z1 > 0)，看起来简单，但这里有个坑：如果Z1恰好等于0，导数取0还是1？实践中取0还是1影响不大，但理论上应该取0.5。我试过两种方式，最终准确率差异在0.1%以内，所以选了更简单的(Z1 > 0)。

优化器：我实现了SGD with Momentum和Adam。Adam的公式看起来复杂，但核心就是两个指数移动平均：

class Adam: def __init__(self, lr=0.001, beta1=0.9, beta2=0.999, eps=1e-8): self.lr = lr self.beta1 = beta1 self.beta2 = beta2 self.eps = eps self.m = {} self.v = {} self.t = 0 def update(self, params, grads): self.t += 1 for key in params: if key not in self.m: self.m[key] = np.zeros_like(params[key]) self.v[key] = np.zeros_like(params[key]) # 一阶矩估计 self.m[key] = self.beta1 * self.m[key] + (1 - self.beta1) * grads[key] # 二阶矩估计 self.v[key] = self.beta2 * self.v[key] + (1 - self.beta2) * (grads[key] ** 2) # 偏差修正 m_hat = self.m[key] / (1 - self.beta1 ** self.t) v_hat = self.v[key] / (1 - self.beta2 ** self.t) # 更新参数 params[key] -= self.lr * m_hat / (np.sqrt(v_hat) + self.eps)

Adam的默认学习率是0.001，这个值在大多数情况下都能work。但如果你发现loss震荡得厉害，可以降到0.0001；如果收敛太慢，可以升到0.003。我实测下来，对于小批量（batch_size=32）训练，0.001是最稳的。

注意事项：手写反向传播最容易出错的地方是矩阵维度。建议每写一行都打印一下shape，确认无误后再继续。我一开始就因为转置搞错，调了整整一个下午。

3.4 模型压缩：量化、剪枝与蒸馏的取舍

模型训练完之后，下一步就是压缩。我实现了三种方法：量化、剪枝和蒸馏。

量化：把float32的权重转成int8。核心是计算scale和zero_point：

def quantize(W, num_bits=8): # 计算量化范围 W_min, W_max = np.min(W), np.max(W) # 计算scale和zero_point scale = (W_max - W_min) / (2 ** num_bits - 1) zero_point = np.round(-W_min / scale).astype(np.int32) # 量化 W_int = np.round(W / scale + zero_point).astype(np.int32) W_int = np.clip(W_int, 0, 2 ** num_bits - 1) return W_int, scale, zero_point

量化的关键是校准。我用了两种校准方法：最小最大值校准和移动平均校准。前者简单但容易受异常值影响，后者更稳定但需要更多数据。我实测下来，对于权重分布比较均匀的模型，最小最大值校准就够了；如果权重有长尾分布，建议用移动平均。

剪枝：把不重要的权重置零。我实现了基于权重大小的剪枝：

def prune(W, sparsity=0.5): # 计算阈值 threshold = np.percentile(np.abs(W), sparsity * 100) # 剪枝 W_pruned = W.copy() W_pruned[np.abs(W) < threshold] = 0 return W_pruned

剪枝率sparsity的选择很关键。我试过0.3、0.5、0.7、0.9，最后发现0.5是一个比较好的平衡点：模型体积减半，准确率下降不到1%。如果剪到0.9，准确率会掉5%以上。

蒸馏：用大模型教小模型。核心是损失函数的设计：

def distillation_loss(y_true, y_pred, y_teacher, alpha=0.5, T=3.0): # 硬标签损失 hard_loss = cross_entropy(y_true, y_pred) # 软标签损失 soft_loss = cross_entropy(softmax(y_teacher / T), softmax(y_pred / T)) return alpha * hard_loss + (1 - alpha) * soft_loss * T * T

温度参数T控制软标签的平滑程度。T越大，软标签越平滑，学生模型能学到更多类别间的相对关系。我试过T=1、3、5、10，最后选了3。alpha控制硬标签和软标签的权重，0.5是一个比较稳的默认值。

三种方法的对比：

我最终的选择是：先剪枝再量化。剪枝把模型体积减半，量化再把剩下的权重压到int8，总体压缩率能达到8倍，准确率损失控制在2%以内。

3.5 推理服务：从Flask到自研批处理

推理服务是整个链路的出口。我一开始用Flask写了一个最简单的版本：

@app.route('/predict', methods=['POST']) def predict(): data = request.json X = preprocess(data) y = model.forward(X) return jsonify({'prediction': y.tolist()})

这个版本能跑，但性能很差。问题在于：每个请求都单独做一次前向传播，GPU利用率极低。解决办法是批处理：把多个请求攒成一个batch，一起送进模型。

我实现了一个简单的批处理调度器：

class BatchScheduler: def __init__(self, model, max_batch_size=32, max_wait_time=0.01): self.model = model self.max_batch_size = max_batch_size self.max_wait_time = max_wait_time self.queue = [] self.lock = threading.Lock() def add_request(self, X): with self.lock: self.queue.append(X) if len(self.queue) >= self.max_batch_size: return self._process_batch() # 等待一小段时间，看能不能攒更多请求 time.sleep(self.max_wait_time) with self.lock: if self.queue: return self._process_batch() return None def _process_batch(self): batch = np.stack(self.queue) self.queue = [] return self.model.forward(batch)

这里有两个关键参数：max_batch_size和max_wait_time。前者决定了单次计算的最大吞吐量，后者决定了延迟的上限。我实测下来，max_batch_size=32、max_wait_time=0.01是一个比较好的平衡：吞吐量提升了8倍，延迟只增加了10毫秒。

注意事项：批处理会引入额外的延迟，如果你的应用对延迟极其敏感（比如实时交互），建议把max_wait_time设小一点，比如0.001秒。但这样批处理的收益也会降低。

3.6 监控与迭代：指标采集和告警

服务上线之后，监控是必不可少的。我采集了四类指标：

• 请求指标：QPS、延迟分布、错误率
• 系统指标：CPU利用率、内存占用、GPU利用率
• 模型指标：预测分布、置信度分布、特征漂移
• 业务指标：点击率、转化率、用户反馈

采集方式很简单，用Prometheus的Python客户端：

from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT = Counter('request_count', 'Total request count') REQUEST_LATENCY = Histogram('request_latency_seconds', 'Request latency') MODEL_CONFIDENCE = Gauge('model_confidence', 'Average model confidence') @app.route('/predict', methods=['POST']) def predict(): REQUEST_COUNT.inc() start_time = time.time() # ... 处理请求 ... REQUEST_LATENCY.observe(time.time() - start_time) MODEL_CONFIDENCE.set(np.mean(confidence)) return jsonify(result)

告警规则我设了三条：

1. 延迟告警：P99延迟超过100毫秒，持续1分钟
2. 错误率告警：错误率超过1%，持续5分钟
3. 置信度告警：平均置信度低于0.6，持续10分钟

第三条特别有用。当模型遇到分布外数据时，置信度会明显下降，这时候就需要触发模型更新或者人工介入。

4. 实操过程中的常见问题与排查技巧

4.1 数值稳定性问题

手写实现最容易遇到的就是数值问题。我整理了一个速查表：

我遇到最诡异的一次是loss在训练初期正常下降，但到第10个epoch突然变成NaN。排查了半天，发现是学习率在后期相对梯度来说太大了。解决办法是加学习率衰减：

def lr_schedule(epoch, initial_lr=0.001, decay_rate=0.9, decay_steps=5): return initial_lr * (decay_rate ** (epoch // decay_steps))

这个策略是每5个epoch把学习率乘以0.9。实测下来，比固定学习率稳定很多。

4.2 内存与性能瓶颈

从零实现的一个好处是你能精确控制内存。我总结了几个优化技巧：

第一，用原地操作。比如X += 1比X = X + 1省内存，因为后者会创建新数组。

第二，用视图而不是拷贝。NumPy的切片返回的是视图，不占额外内存。但要注意，修改视图会影响原数组。

第三，用float32而不是float64。精度损失很小，但内存和计算量都减半。

第四，用Cython加速热点代码。比如矩阵乘法，用Cython重写之后能快3-5倍。

我实测下来，一个原本需要2GB内存的模型，经过这些优化之后，内存占用降到了800MB。

4.3 线上服务与本地不一致

这是最让人头疼的问题。本地跑得好好的，一上线就出问题。常见原因有：

• 数据分布不一致：线上数据有更多异常值或缺失值
• 版本不一致：本地和线上的依赖版本不同
• 并发问题：多线程或多进程导致的状态竞争
• 资源限制：线上CPU或内存受限，导致超时

我的解决办法是：在本地模拟线上环境。用Docker限制CPU和内存，用压力测试工具模拟并发请求，用线上数据的一个子集做验证。这样能在上线前发现大部分问题。

提示：建议在服务里加一个/health接口，返回模型版本、依赖版本、当前负载等信息。出问题的时候，第一件事就是查这个接口。

4.4 模型更新与回滚

模型更新是另一个容易出问题的环节。我的做法是：

1. 灰度发布：新模型先接10%的流量，观察24小时
2. A/B测试：对比新旧模型的业务指标
3. 快速回滚：如果新模型出问题，能在1分钟内切回旧模型

实现方式很简单，用配置中心控制流量比例：

def get_model(): traffic_ratio = config.get('new_model_traffic_ratio', 0.0) if random.random() < traffic_ratio: return new_model return old_model

回滚就是把traffic_ratio设回0。这个操作要能在不重启服务的情况下完成，所以配置要支持热更新。

5. 从零实现之后的收获与扩展方向

走完这一整条链路之后，我最大的感受是：AI工程的核心不是模型，而是工程。模型结构可以调包，但数据处理、性能优化、服务部署、监控迭代这些环节，必须自己理解才能做好。

具体来说，我有三个方面的收获：

第一，性能优化有了方向。以前只知道模型慢，现在能精确到是哪个算子慢、为什么慢、怎么优化。比如我发现ReLU激活函数在NumPy里的实现比手写循环快50倍，因为NumPy底层用了SIMD指令。

第二，调试能力大幅提升。以前遇到问题只能猜，现在能通过打印中间结果、对比梯度、检查数值稳定性来定位。有一次线上服务返回全零，我5分钟就定位到是输入数据没有做标准化。

第三，架构设计更有底气。以前设计系统只能参考别人的方案，现在能根据自己的需求做取舍。比如我知道批处理能提升吞吐量但会增加延迟，所以能根据业务场景选择最合适的参数。

这个项目后续还可以往几个方向扩展：

• 支持更多模型结构：目前只实现了全连接网络，可以扩展到CNN、RNN、Transformer
• 支持更多硬件：目前只跑了CPU，可以扩展到GPU、NPU、FPGA
• 支持分布式训练：目前只支持单机，可以扩展到多机多卡
• 支持自动化调参：目前靠手动调参，可以引入贝叶斯优化或强化学习

最后分享一个小技巧：如果你也想走一遍这个路径，建议从推理服务开始，倒着往训练走。因为推理服务是最终产出，能让你快速看到成果，有成就感之后再去啃训练和压缩。而且推理服务遇到的问题（比如延迟、内存、并发）更直观，更容易理解。

我在实际操作中的体会是，从零实现一遍之后，再回头看那些框架的文档，会有一种“原来如此”的感觉。以前觉得神秘的自动微分、算子融合、内存池，现在都能理解背后的原理了。这种理解带来的自信，是调包永远给不了的。