资讯动态

从零手搓AI工程:拆解黑盒,掌握生产级AI系统核心

发布时间:2026/10/3 16:04:44 来源:尧图企业网站定制
1. 从零手搓AI工程为什么我不建议你直接调包很多人一提到AI工程脑子里第一反应就是pip install transformers然后写个pipeline就觉得自己入门了。我刚开始也这么干过结果到了真实项目里模型推理慢得像蜗牛显存炸得比气球还快部署上线后并发一上来直接崩盘。那时候我才意识到会调库和会做AI工程中间隔着一整条鸿沟。ai-engineering-from-scratch这个标题核心不是教你从零训练一个GPT而是让你从工程视角重新理解AI系统的每一层。它解决的是“知其然不知其所以然”的问题——你知道model.generate()能出结果但不知道KV Cache怎么省显存不知道算子融合为什么能提速不知道量化到底损失了什么。适合谁看适合那些已经能跑通demo但一上生产就抓瞎的开发者也适合想转行AI工程但被各种框架抽象层绕晕的初学者。我写这篇东西的出发点很简单把我自己踩过的坑、拆过的轮子、算过的账原原本本摊开。不堆公式不抄文档就讲人话。你跟着走一遍至少能明白一个AI系统从数据进来到结果出去中间到底发生了什么哪些地方能动手脚哪些地方千万别乱碰。2. 整体设计思路把黑盒拆成白盒2.1 为什么选择“从零实现”而不是“直接调包”直接调包最大的问题是抽象泄漏。框架帮你封装了太多东西一旦出问题你连报错都看不懂。比如PyTorch的DataLoader你用num_workers8觉得挺快但你知道每个worker怎么复制数据、怎么共享内存、什么时候会死锁吗不知道。从零实现一遍哪怕只是写个最简陋的版本你也能把这条链路摸清楚。另一个原因是性能调优的需要。生产环境里推理延迟每降低10ms可能就省下一台服务器。而这些优化点往往藏在框架底层内存布局、计算图优化、算子选择。你不拆开看根本不知道从哪里下手。我试过直接拿HuggingFace的pipeline做服务QPS死活上不去后来自己重写了预处理和后处理把Python循环换成向量化操作性能直接翻了三倍。这就是从零理解的价值。还有一点很现实面试造火箭。现在AI岗位面试动不动就问你“手写Attention”“推导反向传播”“实现一个简单的Transformer”。你光会调包第一轮技术面就挂了。从零实现一遍这些题都是送分题。2.2 核心架构分层从数据到部署的完整链路一个完整的AI工程系统我习惯把它分成五层。这个分层不是教科书上的是我自己项目里总结出来的怎么顺手怎么来。第一层数据管道。负责数据的读取、清洗、增强、批处理。这层最容易被忽视但实际项目中80%的时间都花在这里。我见过太多人模型调得飞起结果数据里全是脏样本训出来的东西根本不能用。第二层模型定义。包括网络结构、损失函数、初始化策略。这层是核心但也是最成熟的部分大部分时候你不需要从零写但必须能看懂。第三层训练循环。优化器、学习率调度、梯度裁剪、混合精度、分布式训练。这层是工程含量最高的地方也是从零实现最能学到东西的地方。第四层推理优化。量化、剪枝、算子融合、KV Cache、批处理策略。这层直接决定你的服务能不能上线。第五层服务部署。API设计、并发处理、监控告警、版本管理。这层是AI工程和传统后端工程的交叉点。每一层我都会在后面的章节里展开讲但你先记住这个结构数据进结果出中间每一层都有坑。2.3 技术选型为什么用PyTorch而不是TensorFlow这个问题我被问过无数次。我的答案很简单调试体验。PyTorch是动态图你可以像写普通Python一样print中间结果pdb断点调试。TensorFlow 1.x的静态图时代调试简直是噩梦得用tf.Print这种反人类的东西。虽然TF 2.x也支持动态图了但生态和社区惯性已经倒向PyTorch。另一个原因是研究友好。你看最新的论文90%以上都提供PyTorch实现。你要复现一个模型用PyTorch能直接抄代码用TensorFlow得自己翻译。时间就是生命。当然TensorFlow在部署端有优势TF Serving、TFLite这些工具很成熟。但现在PyTorch有TorchServe、ONNX Runtime差距已经很小了。所以我的建议是训练用PyTorch部署看情况转ONNX。这个组合我用了三年没出过大问题。3. 核心细节解析那些文档里不会写的实操要点3.1 数据管道别让IO成为你的瓶颈数据管道的第一原则能并行就并行能预取就预取。我见过太多项目GPU利用率只有30%一查发现是数据加载卡住了。CPU在那慢悠悠地读图片、做增强GPU干等着。具体怎么做PyTorch的DataLoader有num_workers和prefetch_factor两个参数。num_workers设成CPU核心数prefetch_factor设成2到4。但注意num_workers不是越大越好每个worker都会复制一份数据集内存会爆。我一般设成min(8, cpu_count)。还有一个坑Windows上num_workers0会出问题。因为Windows用spawn而不是fork来创建子进程数据集对象必须可pickle。我当初在Windows上调试了一下午最后发现是自定义Dataset里有个lambda函数没法序列化。解决办法很简单把lambda改成普通函数或者用num_workers0先跑通。数据增强这块能在线做就别离线做。离线增强会占用大量磁盘空间而且每个epoch看到的都是同样的增强结果降低了数据多样性。在线增强虽然消耗CPU但配合多worker完全扛得住。我习惯用Albumentations库比torchvision的transforms快不少而且支持更多增强策略。注意数据增强一定要区分训练集和验证集。训练集可以随机裁剪、翻转、颜色抖动验证集只能做中心裁剪和归一化。我见过有人验证集也做随机增强结果指标忽高忽低排查了半天才发现是数据的问题。3.2 模型定义初始化不是随便写写模型初始化有多重要我做过一个实验同样的网络结构用默认初始化和用Kaiming初始化训练收敛速度差了三倍。默认初始化PyTorch的reset_parameters对深层网络来说方差太小信号传到后面层就消失了。Kaiming初始化的核心思想是保持每一层输出的方差一致。具体公式是std sqrt(2 / fan_in)其中fan_in是输入神经元数量。PyTorch里直接用nn.init.kaiming_normal_就行。但注意这个初始化是针对ReLU激活函数的。如果你用SELU或者GELU得用不同的初始化策略。还有一个细节偏置项通常初始化为0。但有些特殊情况比如你用了BatchNorm偏置其实可以省略因为BN会把它抵消掉。我一般会把BN层前面的卷积偏置设成False省一点参数和计算量。损失函数这块CrossEntropyLoss自带Softmax你千万别在前面再加一个Softmax。我当初就犯过这个错模型输出经过两次Softmax梯度直接爆炸。还有如果标签有类别不平衡记得加weight参数或者用Focal Loss。Focal Loss的公式很简单FL -alpha * (1-p)^gamma * log(p)gamma一般取2alpha根据类别频率设置。3.3 训练循环混合精度是把双刃剑混合精度训练AMP能省显存、提速但用不好会出大问题。核心原理是前向和反向用FP16参数更新用FP32。这样既享受了FP16的计算速度又避免了FP32的精度损失。PyTorch里用torch.cuda.amp很简单scaler torch.cuda.amp.GradScaler() for data, target in dataloader: optimizer.zero_grad() with torch.cuda.amp.autocast(): output model(data) loss criterion(output, target) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()但注意有些操作不支持FP16比如softmax在FP16下容易溢出。autocast会自动把这些操作转成FP32但你得知道哪些操作被转了。我一般会在关键位置加print(tensor.dtype)来确认。还有一个坑梯度裁剪和AMP一起用的时候要先unscale再裁剪。因为scaler.scale(loss)把梯度放大了你直接裁剪裁的是放大后的梯度比例不对。正确做法是scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(optimizer) scaler.update()学习率调度这块OneCycleLR是我最推荐的。它先 warmup 再退火训练初期稳定后期收敛快。参数设置也简单max_lr设成你手动调出来的最佳学习率的3到5倍total_steps就是总迭代次数。我实测下来比StepLR和CosineAnnealing都要稳。4. 实操过程从零搭一个图像分类系统4.1 环境准备与依赖安装先列一下我用的环境你照着装就行python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118 pip install albumentations opencv-python-headless pip install tensorboard tqdm pip install onnx onnxruntime版本方面PyTorch 2.0以上都行CUDA版本根据你的显卡驱动选。我用的CUDA 11.8对应PyTorch的cu118版本。如果你没有GPU把cu118换成cpu但训练速度会慢很多建议至少用Google Colab的免费GPU。提示安装PyTorch时一定要去官网复制对应的安装命令不要直接pip install torch。默认源装的是CPU版本你跑半天发现GPU没用上白等。4.2 数据集准备与预处理我用CIFAR-10做例子因为小、快、经典。但实际项目中你面对的是自己的数据所以我会把数据加载部分写得通用一点。import torch from torch.utils.data import Dataset, DataLoader import albumentations as A from albumentations.pytorch import ToTensorV2 import cv2 import numpy as np class CustomDataset(Dataset): def __init__(self, image_paths, labels, transformNone): self.image_paths image_paths self.labels labels self.transform transform def __len__(self): return len(self.image_paths) def __getitem__(self, idx): image cv2.imread(self.image_paths[idx]) image cv2.cvtColor(image, cv2.COLOR_BGR2RGB) label self.labels[idx] if self.transform: augmented self.transform(imageimage) image augmented[image] return image, label train_transform A.Compose([ A.RandomResizedCrop(224, 224, scale(0.8, 1.0)), A.HorizontalFlip(p0.5), A.ColorJitter(brightness0.2, contrast0.2, saturation0.2, hue0.1, p0.5), A.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]), ToTensorV2(), ]) val_transform A.Compose([ A.Resize(256, 256), A.CenterCrop(224, 224), A.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]), ToTensorV2(), ])这里有几个细节。RandomResizedCrop的scale参数我设成(0.8, 1.0)意思是随机裁剪原图80%到100%的区域。这个范围不能太大否则裁出来的图太小语义信息丢失。ColorJitter的hue参数我设成0.1因为色相变化太大会让物体颜色失真影响分类。归一化的均值和标准差是ImageNet的统计值。如果你用自己的数据集最好重新算一下。计算方法很简单遍历所有训练图片算每个通道的均值和标准差。我一般会写个小脚本跑一遍几分钟的事。4.3 模型搭建手写一个ResNet残差块虽然torchvision.models.resnet18可以直接用但我建议你手写一遍残差块理解为什么它能训练那么深。import torch.nn as nn import torch.nn.functional as F class BasicBlock(nn.Module): expansion 1 def __init__(self, in_planes, planes, stride1): super(BasicBlock, self).__init__() self.conv1 nn.Conv2d(in_planes, planes, kernel_size3, stridestride, padding1, biasFalse) self.bn1 nn.BatchNorm2d(planes) self.conv2 nn.Conv2d(planes, planes, kernel_size3, stride1, padding1, biasFalse) self.bn2 nn.BatchNorm2d(planes) self.shortcut nn.Sequential() if stride ! 1 or in_planes ! self.expansion * planes: self.shortcut nn.Sequential( nn.Conv2d(in_planes, self.expansion * planes, kernel_size1, stridestride, biasFalse), nn.BatchNorm2d(self.expansion * planes) ) def forward(self, x): out F.relu(self.bn1(self.conv1(x))) out self.bn2(self.conv2(out)) out self.shortcut(x) out F.relu(out) return out残差连接的核心是out self.shortcut(x)。这个加法的梯度是1所以反向传播时梯度可以直接传过去不会消失。shortcut在维度不匹配时用1x1卷积调整匹配时就是恒等映射。我当初手写的时候犯过一个错self.shortcut定义成None然后在forward里判断if self.shortcut is not None。这样写逻辑没错但PyTorch的state_dict里不会保存None加载模型时会报错。正确做法是像上面那样用nn.Sequential()空容器这样state_dict里是空的加载时不会出问题。4.4 训练循环与验证训练循环我写了个模板你直接抄def train_one_epoch(model, dataloader, criterion, optimizer, scaler, device): model.train() total_loss 0 correct 0 total 0 for images, labels in dataloader: images, labels images.to(device), labels.to(device) optimizer.zero_grad() with torch.cuda.amp.autocast(): outputs model(images) loss criterion(outputs, labels) scaler.scale(loss).backward() scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(optimizer) scaler.update() total_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() return total_loss / len(dataloader), 100. * correct / total验证循环类似但要去掉梯度计算和AMPtorch.no_grad() def validate(model, dataloader, criterion, device): model.eval() total_loss 0 correct 0 total 0 for images, labels in dataloader: images, labels images.to(device), labels.to(device) outputs model(images) loss criterion(outputs, labels) total_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() return total_loss / len(dataloader), 100. * correct / total注意model.eval()和torch.no_grad()缺一不可。eval()会把BN和Dropout切到推理模式no_grad()会关掉梯度计算省显存。我见过有人忘了eval()验证指标一直很低排查半天才发现是BN在作怪。4.5 模型导出与ONNX推理训练完了要部署ONNX是通用格式。导出很简单dummy_input torch.randn(1, 3, 224, 224).to(device) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version11 )dynamic_axes很重要不设的话导出的模型只能跑固定batch size。opset_version我一般用11兼容性好。导出后可以用onnxruntime验证一下import onnxruntime as ort import numpy as np sess ort.InferenceSession(model.onnx) input_name sess.get_inputs()[0].name dummy np.random.randn(1, 3, 224, 224).astype(np.float32) output sess.run(None, {input_name: dummy}) print(output[0].shape)ONNX推理的速度比PyTorch原生快不少尤其是CPU上。我实测ResNet18在CPU上ONNX比PyTorch快2倍左右。GPU上差距小一些但ONNX Runtime的显存占用更低。5. 常见问题与排查技巧实录5.1 损失不下降先查这五个地方损失不下降是新手最常遇到的问题。我按排查优先级列一下排查项可能原因解决方法学习率太大导致震荡太小导致不收敛用LR Finder找最佳值或从1e-3开始试数据标签标签和图像不对应随机抽几张图可视化确认标签正确损失函数用错损失函数分类用CrossEntropy回归用MSE初始化参数初始化方差不对用Kaiming或Xavier初始化梯度梯度消失或爆炸加BN、残差连接或梯度裁剪我遇到最多的是学习率太大。表现是损失一开始下降然后突然跳到NaN。解决办法很简单把学习率除以10再试。如果损失下降太慢就乘以10。反复几次就能找到合适的量级。还有一个隐蔽的问题数据没有归一化。如果输入图片像素值是0到255而模型期望的是0到1那第一层的输出会非常大导致梯度爆炸。我当初用自定义数据集时就忘了归一化训练了三个epoch损失都不动后来加了ToTensorV2()它会自动除以255才正常。5.2 显存不够用试试这些招显存不够是另一个高频问题。我按效果排序减小batch size。最直接但会影响BN的效果。如果batch size小于8建议把BN换成GroupNorm。混合精度训练。省显存30%到50%几乎不影响精度。梯度累积。用accumulation_steps4等效于batch size乘以4但显存不变。梯度检查点。用torch.utils.checkpoint省显存但慢20%左右。模型并行。把不同层放到不同GPU上适合超大模型。我一般先用混合精度不够再梯度累积。梯度检查点太慢不到万不得已不用。注意torch.cuda.empty_cache()能释放缓存显存但别频繁调用会影响性能。它只在显存碎片化严重时用一下。5.3 推理速度慢从这三个层面优化推理速度优化分三个层面模型层面、算子层面、服务层面。模型层面换更小的模型ResNet18换MobileNet或者剪枝、量化。量化我用torch.quantization把FP32转成INT8速度提升2到4倍精度损失1%以内。算子层面用ONNX Runtime或TensorRT。TensorRT在NVIDIA GPU上最快但配置麻烦。ONNX Runtime通用性好CPU和GPU都支持。服务层面批处理。把多个请求攒成一个batch一起推理吞吐量能提升好几倍。但会增加延迟需要权衡。我一般设个最大等待时间比如10ms超时就单独推理。5.4 模型上线后精度下降排查数据漂移训练时精度95%上线后只有80%这种情况太常见了。原因通常是数据漂移线上数据的分布和训练数据不一样。排查方法把线上数据采样一批和训练数据做对比。看均值、方差、直方图。如果差异大说明需要重新训练或者做领域适应。我遇到过一个案例训练数据是白天拍的线上数据有夜间拍的。模型在夜间图片上表现极差。解决办法是在训练数据里加入夜间图片或者用风格迁移把白天转成夜间。另一个原因是预处理不一致。训练时用OpenCV读图BGR线上用PIL读图RGB通道顺序反了精度直接崩。这种问题最隐蔽但排查起来也简单把同一张图用两种方式读一遍对比一下。6. 工具选型与效率提升6.1 实验管理别再用Excel记结果了我刚开始做实验的时候用Excel记超参数和指标结果文件多了根本找不到。后来换了TensorBoard再后来用Weights BiasesWB。WB能自动记录超参数、指标、梯度直方图还能对比不同实验。免费版够个人用。如果你不想用在线工具TensorBoard也行。SummaryWriter几行代码就能记录from torch.utils.tensorboard import SummaryWriter writer SummaryWriter(runs/experiment_1) writer.add_scalar(Loss/train, train_loss, epoch) writer.add_scalar(Accuracy/train, train_acc, epoch)关键是每次实验都记清楚配置。我习惯在代码开头写个config字典所有超参数从里面读然后存成JSON。这样复现的时候直接加载JSON就行。6.2 代码组织别把所有东西塞一个文件我见过太多项目一个train.py两千行改个参数得翻半天。好的组织方式是按功能分模块project/ ├── configs/ │ └── default.yaml ├── data/ │ ├── dataset.py │ └── transforms.py ├── models/ │ ├── resnet.py │ └── losses.py ├── engine/ │ ├── trainer.py │ └── validator.py ├── utils/ │ ├── logger.py │ └── checkpoint.py └── train.py这样改数据增强去transforms.py改模型去resnet.py互不干扰。配置文件用YAML比argparse灵活支持嵌套结构。6.3 版本管理模型和代码要一起管代码用Git管这不用多说。但模型权重和数据集怎么管小文件直接Git LFS大文件用DVC。DVC能跟踪数据版本和Git配合使用。我一般把dvc.yaml和.dvc文件提交到Git实际数据存在远程存储比如S3或本地NAS。模型权重我习惯按{experiment_name}_{epoch}_{metric}.pth命名比如resnet18_epoch50_acc0.95.pth。这样一眼就能看出是哪个实验、第几个epoch、什么指标。别用best.pth这种名字过两天你就忘了best是哪个best。7. 从零实现的边界哪些轮子值得造哪些别碰7.1 值得自己写的部分数据管道必须自己写。每个项目的数据格式都不一样通用库解决不了所有问题。而且数据管道是性能瓶颈的高发区自己写才能优化到位。训练循环建议自己写。虽然PyTorch Lightning、HuggingFace Trainer能帮你省事但它们的抽象层太厚出问题不好排查。自己写一遍以后用框架也能知道底层在干什么。推理服务建议自己写。FastAPI加ONNX Runtime几十行代码就能搭一个高性能服务。用Triton Inference Server当然更强大但配置复杂小项目没必要。7.2 不值得自己写的部分底层算子别自己写CUDA核函数。除非你是专门做高性能计算的否则用cuDNN、MKL这些库就行。自己写的算子大概率比库慢还容易出bug。分布式训练别自己实现AllReduce。用torch.distributed或者Horovod。分布式训练的坑太多了网络通信、同步策略、容错处理每一个都能让你调一周。自动微分别自己写反向传播。PyTorch的autograd已经足够好你手写的反向传播大概率有数值误差。除非你在做研究需要自定义梯度否则用现成的。7.3 一个实用的判断标准我判断一个轮子该不该造看三点是否影响性能、是否影响正确性、是否有现成的可靠方案。如果影响性能比如数据加载、推理优化自己写。如果影响正确性比如损失函数、评估指标自己写一遍确保理解。如果有现成可靠方案比如优化器、分布式通信直接用。这个标准帮我省了很多时间。我见过有人花两周自己写了个Adam优化器结果和PyTorch的差0.1%精度图啥呢8. 后续扩展方向从单机到生产8.1 分布式训练从DataParallel到DDP单卡训练不够用了第一步是上多卡。PyTorch有两种方式DataParallelDP和DistributedDataParallelDDP。别用DP它只在单机多卡有效而且主卡显存占用高效率低。用DDP支持多机多卡效率高。DDP的核心是DistributedSampler和init_process_groupimport torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP dist.init_process_group(backendnccl) model model.to(rank) model DDP(model, device_ids[rank]) sampler DistributedSampler(dataset) dataloader DataLoader(dataset, samplersampler, batch_sizebatch_size)注意每个epoch开始前要调sampler.set_epoch(epoch)否则每个epoch的数据顺序都一样影响训练效果。8.2 模型压缩量化、剪枝、蒸馏生产环境对模型大小和速度有要求压缩是必经之路。量化最简单torch.quantization.quantize_dynamic一行代码就能把Linear层转成INT8。剪枝复杂一些需要迭代训练和剪枝。蒸馏需要一个大模型当老师训练一个小模型。我一般先量化不够再剪枝。蒸馏用的少因为需要额外训练一个大模型成本高。8.3 持续学习模型更新不重训线上模型需要定期更新但全量重训成本太高。持续学习Continual Learning能在不遗忘旧知识的前提下学习新数据。方法有EWC、SI、Replay Buffer等。我试过Replay Buffer简单有效存一批旧数据和新数据混着训。但这个方向坑很多灾难性遗忘很难彻底解决。如果数据分布变化不大我建议直接全量重训省心。8.4 监控与告警别等用户投诉才发现问题上线不是终点是起点。你需要监控延迟、吞吐量、错误率、GPU利用率、显存占用。Prometheus加Grafana是标配。告警规则我一般设延迟P99超过100ms告警错误率超过1%告警GPU利用率持续低于30%告警说明有优化空间。还有一个容易被忽视的数据分布监控。线上数据的均值、方差、类别分布和训练数据对比。如果漂移超过阈值触发重新训练。这个我踩过坑模型上线三个月没管精度从95%掉到70%用户投诉了才发现。我个人在实际操作中的体会是AI工程最难的不是模型本身而是工程化。模型结构就那些论文里写得清清楚楚。但怎么让它在生产环境稳定跑起来怎么优化性能怎么排查问题这些才是真正拉开差距的地方。从零实现一遍不是为了造轮子是为了理解轮子。理解了之后你用任何框架都能得心应手。最后分享一个小技巧每次遇到问题先别急着搜解决方案自己先猜一下原因然后验证。猜对了你加深了理解猜错了你知道了自己的盲区。这个习惯我坚持了三年比看任何教程都管用。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑