1. 这个项目到底在解决什么问题第一次看到 ai-engineering-from-scratch 这个标题我脑子里蹦出来的第一个念头是又是一个教人调包的教程但仔细琢磨了一下 from scratch 这四个字我意识到它想做的事情可能完全不一样。市面上讲 AI 工程的内容绝大多数都是从pip install transformers开始然后from_pretrained一把梭跑通了就皆大欢喜。可一旦模型效果不对、推理速度上不去、显存爆了、部署到生产环境各种超时很多人就懵了——因为底层的那些东西从来没自己动手搭过。这个项目要解决的核心痛点就在这里让 AI 工程师真正理解从零构建一个 AI 系统需要经历哪些环节而不是停留在调库的层面。它适合那些已经会用 PyTorch 或 TensorFlow 跑通 demo但想进一步搞清楚为什么这样设计底层到底发生了什么如果不用现成框架我该怎么写的人。说白了就是给那些不满足于当调包侠、想往深水区走的工程师看的。我自己的经历是早年做推荐系统的时候一直用现成的特征工程库和模型训练框架直到有一次线上出了个诡异的 bug排查了三天才发现是对底层张量运算的理解有偏差。从那以后我就特别重视从零实现这件事——哪怕你最终还是要用现成的库但你自己写过一遍看问题的视角完全不一样。这个项目标题里的 AI Engineering 也值得说道说道。它强调的不是 AI Research 或者 Machine Learning而是 Engineering。这意味着重点不在发明新算法而在于如何把算法变成可靠、可维护、可扩展的系统。数据管道怎么搭、模型怎么训练和评估、推理服务怎么部署、监控怎么做——这些才是 AI 工程的核心。而 from scratch 则意味着这些环节你都要亲手实现一遍哪怕是最简陋的版本。2. 从零构建 AI 系统的整体设计思路2.1 为什么非要从零不可很多人会问现在工具链这么成熟为什么还要从零写这不是重复造轮子吗我的回答是造轮子不是为了用而是为了懂。你从零实现一个简单的梯度下降和你直接调optimizer.step()对反向传播的理解深度是完全不同的。前者你会被迫思考链式法则怎么展开、计算图怎么构建、梯度怎么累积后者你只需要知道这一步会更新参数。从工程角度讲从零构建还有一个巨大的好处当系统出问题时你有能力定位到最底层。我见过太多团队模型训练 loss 不下降只能靠猜——换个学习率、换个初始化、换个 batch size试来试去。但如果你自己实现过训练循环你会知道去看梯度范数、看激活值分布、看每一层的输出范围这些都是有据可查的。这个项目的设计思路我理解应该是分层递进的。第一层是数学和算法基础张量运算、自动微分、常见层的实现。第二层是训练基础设施数据加载、模型保存、日志记录、超参管理。第三层是推理和部署模型导出、服务封装、性能优化。每一层都要求你手写核心逻辑而不是调库。2.2 技术选型的取舍逻辑从零构建不代表什么都要用纯 Python 写。合理的做法是核心算法用 NumPy 或纯 Python 实现工程基础设施可以用成熟库。比如自动微分引擎你可以用 NumPy 手写一个简易版但没必要自己实现一个完整的深度学习框架。数据加载可以用 PyTorch 的 DataLoader但你要理解它内部的 sampler、collate_fn 是怎么工作的。为什么这么选因为从零构建的目的是建立直觉而不是替代生产工具。你手写的版本性能肯定不如优化过的库但它能让你看清每个环节的输入输出、每个参数的影响。等你理解了这些再用回成熟库的时候你就知道该关注哪些配置、该在哪些地方做优化。具体到技术栈我建议的路线是Python NumPy 作为基础PyTorch 作为对照参考不是直接用而是用来验证你手写实现的正确性FastAPI 或 Flask 做推理服务Docker 做环境封装。这套组合的好处是门槛低、资料多、社区活跃遇到问题容易找到答案。2.3 项目结构的规划一个合理的从零构建项目目录结构应该长这样ai-engineering-from-scratch/ ├── core/ # 核心算法实现 │ ├── tensor.py # 张量基础操作 │ ├── autograd.py # 自动微分引擎 │ ├── layers.py # 网络层实现 │ └── optim.py # 优化器实现 ├── data/ # 数据处理 │ ├── dataset.py # 数据集封装 │ └── transforms.py # 数据增强 ├── train/ # 训练流程 │ ├── loop.py # 训练循环 │ └── metrics.py # 评估指标 ├── serve/ # 推理服务 │ ├── app.py # API 服务 │ └── export.py # 模型导出 └── tests/ # 测试用例 └── test_core.py # 核心功能验证这个结构的关键在于关注点分离核心算法、数据处理、训练流程、推理服务各自独立方便单独测试和替换。我踩过的坑是早期把所有代码堆在一个文件里后来想换个优化器或者加个新层牵一发动全身改起来特别痛苦。3. 核心模块的从零实现细节3.1 张量基础一切从数组开始AI 系统的基石是张量运算。从零实现的第一步就是搞清楚张量到底是什么。简单说张量就是多维数组加上一组运算规则。你可以用 NumPy 的ndarray作为底层存储然后在上面封装自己的操作。关键要实现的运算包括加法、乘法、矩阵乘法、转置、reshape、广播。其中广播机制是最容易出错的地方。比如一个形状为(3, 1)的张量和一个形状为(1, 4)的张量相加结果形状是(3, 4)。这个规则看起来简单但在反向传播的时候梯度的形状需要和原始输入对齐这里很容易写错。我的建议是先实现一个最简单的Tensor类只支持标量运算然后逐步扩展到多维。每加一个功能就写一个测试用例验证。比如import numpy as np class Tensor: def __init__(self, data, requires_gradFalse): self.data np.array(data, dtypenp.float32) self.requires_grad requires_grad self.grad None self._backward lambda: None self._prev set() def __add__(self, other): other other if isinstance(other, Tensor) else Tensor(other) out Tensor(self.data other.data, requires_gradself.requires_grad or other.requires_grad) out._prev {self, other} def _backward(): if self.requires_grad: self.grad (self.grad or 0) out.grad if other.requires_grad: other.grad (other.grad or 0) out.grad out._backward _backward return out这段代码虽然简陋但它揭示了自动微分的核心思想每个操作记录自己的反向传播逻辑计算图通过_prev连接起来。等你把这个机制吃透了再看 PyTorch 的 autograd就会发现原理完全一样只是工程实现更复杂。注意手写张量运算时一定要处理好数据类型。NumPy 默认的 float64 在深度学习里通常用不上统一用 float32 可以省一半内存而且大多数 GPU 对 float32 的支持最好。3.2 自动微分计算图的构建与反向传播自动微分是深度学习框架的灵魂。从零实现一个简易版你需要理解两个核心概念前向传播构建计算图反向传播沿计算图求梯度。前向传播时每个操作都要记录输入是什么、输出是什么、反向传播时梯度怎么算。反向传播时从损失函数开始沿着计算图反向遍历用链式法则逐层计算梯度。这里的关键难点是拓扑排序。因为计算图是一个有向无环图DAG反向传播必须保证一个节点的所有下游节点都处理完了才能处理它自己。实现方式通常是深度优先搜索加后序遍历def backward(self): topo [] visited set() def build_topo(v): if v not in visited: visited.add(v) for child in v._prev: build_topo(child) topo.append(v) build_topo(self) self.grad np.ones_like(self.data) for node in reversed(topo): node._backward()这段代码我建议你亲手敲一遍然后拿一个简单的函数比如y (a b) * c手动推导梯度再和代码运行结果对比。手动推导和代码验证结合是理解自动微分最快的方式。实操心得写反向传播的时候最容易犯的错误是梯度累积。比如一个张量被用了两次它的梯度应该是两次贡献之和而不是覆盖。我在实现的时候统一用self.grad (self.grad or 0) new_grad这种写法避免遗漏。3.3 网络层实现从全连接到卷积有了张量和自动微分就可以实现网络层了。最基础的是全连接层Linear核心就是矩阵乘法加偏置class Linear: def __init__(self, in_features, out_features): self.weight Tensor(np.random.randn(in_features, out_features) * 0.01, requires_gradTrue) self.bias Tensor(np.zeros(out_features), requires_gradTrue) def __call__(self, x): return x self.weight self.bias卷积层稍微复杂一点但原理是一样的滑动窗口加乘积累加。从零实现卷积你可以先用最朴素的循环写法虽然慢但逻辑清晰。等理解透了再用im2col或者 FFT 加速。激活函数也是必写的。ReLU 最简单前向是max(0, x)反向是梯度在 x0 时原样传回否则为 0。Sigmoid 和 Tanh 涉及指数运算反向传播时要注意数值稳定性——当输入很大或很小时梯度会趋近于 0导致梯度消失。提示实现激活函数时建议同时实现前向和反向并写单元测试验证。比如 ReLU 在 x-1 时梯度应该是 0在 x1 时梯度应该是 1。这些测试用例能帮你快速定位 bug。3.4 优化器梯度下降的变体优化器的任务是根据梯度更新参数。最基础的是随机梯度下降SGDclass SGD: def __init__(self, params, lr0.01): self.params params self.lr lr def step(self): for p in self.params: if p.grad is not None: p.data - self.lr * p.grad def zero_grad(self): for p in self.params: p.grad None在此基础上可以实现 Momentum、RMSProp、Adam 等变体。Adam 的公式看起来复杂但拆开看就是一阶矩估计梯度的指数移动平均、二阶矩估计梯度平方的指数移动平均、偏差修正、参数更新。每一步都有明确的数学含义建议对照论文自己推导一遍。参数选择方面学习率是最关键的。我通常从 1e-3 开始试如果 loss 震荡就调小如果下降太慢就调大。Adam 的默认参数beta10.9, beta20.999, eps1e-8在大多数场景下都能用但如果你发现训练不稳定可以试试把 eps 调大到 1e-6。4. 训练流程与工程化实践4.1 数据管道从原始数据到批次数据管道的核心任务是把原始数据转换成模型可以消费的批次。这个环节看起来简单但实际工程中问题最多。常见的问题包括数据格式不统一、内存不够、加载速度慢、批次内数据长度不一致。从零实现一个数据管道你需要处理数据集封装怎么读取单条数据、采样策略怎么组成一个批次、预处理归一化、编码、增强、多进程加载避免 IO 成为瓶颈。我的经验是先把单进程版本写清楚确保逻辑正确再考虑加速。多进程加载的坑很多比如 Windows 下的 spawn 模式、共享内存的序列化开销、worker 异常处理等。如果数据量不大单进程加预取就够了。class Dataset: def __init__(self, data, labels): self.data data self.labels labels def __len__(self): return len(self.data) def __getitem__(self, idx): return self.data[idx], self.labels[idx] class DataLoader: def __init__(self, dataset, batch_size32, shuffleTrue): self.dataset dataset self.batch_size batch_size self.shuffle shuffle def __iter__(self): indices np.arange(len(self.dataset)) if self.shuffle: np.random.shuffle(indices) for i in range(0, len(indices), self.batch_size): batch_idx indices[i:iself.batch_size] batch_data [self.dataset[j] for j in batch_idx] yield self.collate(batch_data) def collate(self, batch): data, labels zip(*batch) return np.stack(data), np.stack(labels)这个实现虽然简单但涵盖了数据管道的核心逻辑。你可以在此基础上加缓存、加预取、加多进程逐步优化。4.2 训练循环每个 epoch 在做什么训练循环是 AI 工程的核心流程。一个标准的训练循环包括前向传播、计算损失、反向传播、更新参数、记录指标。看起来简单但每个环节都有讲究。前向传播时要注意模型的 train/eval 模式切换。Dropout 和 BatchNorm 在训练和推理时的行为不同忘了切换会导致结果异常。损失函数的选择取决于任务分类用交叉熵回归用 MSE排序用 pairwise loss。反向传播之前一定要清空梯度。PyTorch 里是optimizer.zero_grad()手写实现里就是把所有参数的 grad 置为 None 或 0。忘了这一步梯度会累积训练结果完全错误。def train_epoch(model, dataloader, optimizer, criterion): model.train() total_loss 0 for batch_data, batch_labels in dataloader: optimizer.zero_grad() outputs model(batch_data) loss criterion(outputs, batch_labels) loss.backward() optimizer.step() total_loss loss.data return total_loss / len(dataloader)验证集上的评估要切换到 eval 模式并且不计算梯度torch.no_grad()或手动设置requires_gradFalse这样可以省显存、加速。4.3 模型保存与加载模型保存看起来简单但实际工程中有很多细节。保存什么通常包括模型参数、优化器状态、当前 epoch、最佳指标。只保存参数的话恢复训练时优化器的动量信息就丢了可能导致训练不稳定。保存格式方面PyTorch 用state_dict加 pickleTensorFlow 用 SavedModel。从零实现的话我建议用 NumPy 的np.savez保存参数用 JSON 保存元信息。这样跨平台、跨版本兼容性好不依赖特定框架。def save_checkpoint(model, optimizer, epoch, path): checkpoint { model: {name: param.data for name, param in model.named_parameters()}, optimizer: optimizer.state_dict(), epoch: epoch } np.savez(path, **checkpoint)加载的时候要注意版本兼容。如果模型结构变了参数名对不上加载就会失败。我的做法是在 checkpoint 里保存模型结构的配置信息加载时先重建模型再加载参数。注意保存模型时如果用了 DataParallel 或 DistributedDataParallel参数名会多一个module.前缀。加载到单卡模型时需要去掉这个前缀否则会报 key 不匹配。4.4 日志与监控训练过程中的日志记录是排查问题的关键。至少要记录每个 step 的 loss、学习率、梯度范数每个 epoch 的验证指标、耗时。这些数据可以帮助你判断训练是否正常、是否过拟合、学习率是否合适。梯度范数特别有用。如果梯度范数突然变大可能是遇到了异常样本或者学习率太高如果梯度范数一直很小可能是梯度消失需要检查网络结构或初始化。我习惯用 TensorBoard 或者 Weights Biases 来可视化这些指标。但从零构建的话先用简单的文本日志加 matplotlib 画图就够了。关键是要有记录的意识而不是等出了问题再回头加日志。5. 推理部署与性能优化5.1 模型导出从训练到推理训练好的模型要部署到生产环境第一步是导出。导出的核心要求是脱离训练代码独立运行。PyTorch 用 TorchScript 或 ONNXTensorFlow 用 SavedModel。从零构建的话你可以把参数保存成 NumPy 数组推理时用纯 NumPy 实现前向传播。这样做的好处是依赖少、启动快缺点是性能可能不如优化过的推理引擎。但对于学习目的来说完全够用。而且当你自己实现过一遍推理流程后再用 ONNX Runtime 或 TensorRT就知道该关注哪些优化点了。导出时要注意确保推理时的预处理和训练时一致。我见过太多案例训练时用了某种归一化推理时忘了导致效果大幅下降。建议把预处理逻辑也一起导出或者至少写成配置文件。5.2 服务封装API 设计与并发处理推理服务通常以 HTTP API 的形式提供。用 FastAPI 写一个简单的服务from fastapi import FastAPI import numpy as np app FastAPI() model load_model() app.post(/predict) async def predict(request: dict): data np.array(request[data]) output model(data) return {prediction: output.tolist()}这个服务能跑但离生产级还差得远。需要考虑并发处理多个请求同时进来怎么办、批处理把小请求合并成大批次提高吞吐、超时控制避免慢请求拖垮服务、健康检查K8s 需要 liveness 和 readiness 探针。并发方面Python 的 GIL 限制了多线程的并行能力。如果推理是 CPU 密集型的用多进程如果是 IO 密集型的比如等 GPU用异步。FastAPI 支持 async def但要注意不要在 async 函数里做同步的 CPU 计算会阻塞事件循环。5.3 性能优化延迟与吞吐的平衡推理性能的两个核心指标是延迟单个请求的响应时间和吞吐单位时间处理的请求数。这两者往往需要权衡增大批次可以提高吞吐但会增加延迟。优化的方向包括模型量化float32 转 int8减少内存和计算量、算子融合把多个小操作合并成一个大操作、缓存对重复输入直接返回缓存结果、硬件加速用 GPU、TPU 或专用推理芯片。从零构建的话我建议先做 profiling找到瓶颈在哪里。是预处理慢还是模型前向慢还是后处理慢用time.time()或者cProfile打点数据说话。很多时候瓶颈不在模型本身而在数据拷贝或格式转换上。提示NumPy 的np.ascontiguousarray可以把不连续的内存布局转成连续有时能显著提升后续运算速度。这个技巧在处理转置或切片后的数组时特别有用。6. 常见问题与排查技巧实录6.1 训练不收敛的排查思路训练不收敛是最常见的问题。我的排查顺序是先看数据再看模型最后看超参。数据方面检查标签是否正确、是否有异常值、预处理是否合理。我遇到过一次标签编码时把类别 0 和类别 1 搞反了模型怎么训都学不会。还有一次数据归一化时用了全局均值但训练集和验证集的分布差异很大导致验证集效果很差。模型方面检查初始化、检查网络结构、检查是否有梯度消失或爆炸。初始化太小会导致信号逐层衰减太大又会导致梯度爆炸。可以用torch.nn.init.kaiming_normal_或xavier_normal_做初始化。超参方面学习率是最关键的。太大导致震荡太小导致收敛慢。建议用学习率扫描从 1e-5 到 1e-1 各试一下看 loss 曲线的形态。6.2 显存不足的应对策略显存不足是另一个高频问题。解决方法包括减小批次大小、梯度累积小批次多次前向后再更新、混合精度训练float16 加 float32 混合、梯度检查点用计算换显存、模型并行把模型拆到多张卡上。梯度累积的实现很简单每算完一个批次的梯度不立即更新而是累积几次后再更新。这样等效于增大了批次大小但显存占用不变。accumulation_steps 4 for i, (data, labels) in enumerate(dataloader): loss model(data, labels) / accumulation_steps loss.backward() if (i 1) % accumulation_steps 0: optimizer.step() optimizer.zero_grad()混合精度训练用torch.cuda.amp可以自动管理但要注意有些操作在 float16 下会溢出需要用autocast上下文管理器控制。6.3 推理速度慢的优化手段推理速度慢首先要定位瓶颈。用torch.profiler或者简单的计时看时间花在哪里。常见原因和解决方案问题可能原因解决方案预处理慢Python 循环、重复计算向量化、缓存模型前向慢算子未融合、批次太小算子融合、增大批次后处理慢解码逻辑复杂简化逻辑、异步处理数据传输慢CPU-GPU 频繁拷贝减少拷贝、pin memory服务框架慢同步阻塞、序列化开销异步、用更快的序列化我踩过的一个坑是推理服务里每次请求都重新加载模型导致延迟极高。后来改成全局加载一次延迟直接降了一个数量级。这个错误看起来很蠢但在快速原型阶段很容易犯。6.4 常见问题速查表现象可能原因排查方法Loss 不下降学习率太小、梯度消失检查梯度范数、调大学习率Loss 震荡学习率太大、批次太小调小学习率、增大批次验证集效果差过拟合、数据分布不一致加正则化、检查数据划分训练速度慢数据加载瓶颈、模型太大profiling、多进程加载显存溢出批次太大、模型太大减小批次、梯度累积推理结果异常预处理不一致、模式未切换对比训练和推理流程这张表是我自己排查问题时总结的覆盖了大部分常见场景。实际遇到问题时建议先按表排查再深入分析。7. 从零构建的进阶方向7.1 分布式训练入门单机单卡跑通之后下一步是分布式训练。核心思想是数据并行每张卡持有完整的模型副本处理不同的数据批次梯度通过 all-reduce 同步。PyTorch 的 DistributedDataParallel 封装了这些细节但理解底层原理很重要。关键概念包括进程组哪些进程参与通信、通信后端NCCL 用于 GPUGloo 用于 CPU、梯度同步all-reduce 的时机和方式。从零实现的话可以用torch.distributed的底层 API手动控制梯度同步。分布式训练的坑很多端口冲突、进程启动失败、梯度不同步、性能不升反降。建议先在单机多卡上跑通再扩展到多机。7.2 模型压缩与加速模型压缩的目标是在保持效果的前提下减小模型体积、提升推理速度。主要手段包括剪枝去掉不重要的权重、量化降低数值精度、知识蒸馏用大模型教小模型。剪枝最简单的是权重剪枝把绝对值小于阈值的权重置零。但这样得到的稀疏矩阵普通硬件不一定能加速需要专门的稀疏计算库。结构化剪枝去掉整个通道或层对硬件更友好但实现更复杂。量化方面PyTorch 提供了动态量化、静态量化、量化感知训练三种模式。动态量化最简单一行代码就能用但加速效果有限。静态量化需要校准数据效果更好。量化感知训练在训练时就模拟量化误差效果最好但最复杂。7.3 持续学习与模型更新生产环境中的模型不是一成不变的。数据分布会漂移业务需求会变化模型需要持续更新。这就涉及到增量学习和在线学习。增量学习的核心挑战是灾难性遗忘学新数据时忘了旧知识。解决方法包括经验回放保留旧数据的一部分、弹性权重巩固限制重要权重的变化、参数隔离不同任务用不同参数。在线学习则是数据流式到达模型实时更新。这对系统的要求更高需要处理概念漂移、需要保证更新稳定、需要监控效果变化。从零构建的话可以先实现一个简单的滑动窗口训练定期用最近的数据重新训练模型。8. 我个人的一些实操体会这个项目做下来最大的收获不是学会了某个具体技术而是建立了一套从问题到方案的完整思维链路。以前遇到问题第一反应是搜XX报错怎么解决现在会先想这个问题的本质是什么涉及哪些环节每个环节的输入输出是什么然后有针对性地去排查。另一个体会是从零构建和用现成工具并不矛盾。你从零实现过一遍再用现成工具时就知道该关注哪些参数、该在哪些地方做定制。比如用 PyTorch 的 DataLoader你知道num_workers设多少合适、pin_memory什么时候开、collate_fn怎么写。这些细节没自己实现过的人往往是一脸懵的。最后分享一个小技巧每实现一个模块就写一个最小可复现的测试。比如实现了 Linear 层就写一个测试输入随机数据对比手写实现和 PyTorch 的输出是否一致。这个习惯能帮你快速定位 bug也能加深对每个模块的理解。我自己的测试文件比实现代码还长但正是这些测试让我在重构时有了底气。这个方向后续还可以往很多地方扩展比如实现一个简易的 Transformer、做一个端到端的图像分类系统、或者把训练好的模型部署到移动端。关键是把从零构建的思路保持下去遇到新东西先想如果我自己写会怎么写然后再去看别人是怎么写的。这个习惯坚持下来技术深度会有质的提升。