
1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调一下API跑通一个Demo然后发个朋友圈说“今天又搞定了一个AI项目”。我刚开始也是这么干的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出和请求超时我才意识到——那些被封装得严严实实的接口正在悄悄剥夺我定位问题和优化性能的能力。ai-engineering-from-scratch这个标题说白了就是“从零开始做AI工程”。它不是教你如何调用某个现成的模型接口而是带你走一遍一个AI系统从数据到部署的完整链路。你可能会问现在工具链这么成熟为什么还要自己造轮子我的回答很直接造轮子不是为了替代轮子而是为了在轮子爆胎的时候你知道该换哪个零件。这篇文章适合三类人第一类是有一定Python基础想从传统后端或数据分析转AI工程方向的开发者第二类是在做AI应用但总感觉“心里没底”遇到性能问题只能重启服务的工程师第三类是对AI系统内部机制好奇想亲手拆开看看里面到底有什么的技术爱好者。我会围绕数据管道、模型训练、推理服务、性能调优这几个核心环节把我在实际项目中踩过的坑、总结的经验、以及那些文档里不会写的细节全部摊开来讲。需要提前说明的是本文不会涉及任何特定云厂商的绑定操作也不会推荐需要特殊网络环境才能使用的工具。所有内容都基于开源生态和本地可复现的环境你可以在一台普通的开发机上跟着操作。2. 数据管道AI工程里最容易被低估的脏活累活2.1 为什么数据加载会成为训练瓶颈我刚入行的时候总觉得模型结构才是决定训练速度的关键。后来在一个图像分类项目里我把ResNet换成了更轻量的MobileNet结果训练速度只提升了不到15%。用性能分析工具一查发现GPU利用率长期在30%以下大部分时间都在等数据加载。这就是典型的数据管道瓶颈。AI工程和传统软件工程最大的区别之一就是数据量级完全不在一个维度。传统业务一次查询可能就几十条记录而AI训练一个batch可能就是几千张图片或者几万条文本。如果你的数据加载逻辑还是“读文件-解析-预处理-送GPU”这种串行模式那GPU大部分时间都在摸鱼。解决这个问题的核心思路是预取和并行。具体来说你需要构建一个多进程的数据加载器让CPU在GPU计算当前batch的时候就已经把下一个batch准备好了。在PyTorch里DataLoader的num_workers参数就是干这个的。但这里有个坑num_workers不是越大越好。我实测下来在4核8线程的机器上num_workers设为4到6比较合适设成16反而会因为进程切换开销导致性能下降。from torch.utils.data import DataLoader # 一个典型的数据加载器配置 train_loader DataLoader( datasettrain_dataset, batch_size64, shuffleTrue, num_workers4, # 根据CPU核心数调整 pin_memoryTrue, # 锁页内存加速CPU到GPU的传输 prefetch_factor2, # 每个worker预取2个batch persistent_workersTrue # 保持worker进程存活避免重复创建 )pin_memoryTrue这个参数特别值得说一下。它会把数据加载到锁页内存里这样从CPU内存拷贝到GPU显存的时候可以用DMA直接传输不需要CPU介入。实测下来这个操作能减少大约20%到30%的数据传输时间。但注意如果你的数据集很小或者内存本身就很紧张开这个参数可能会适得其反。2.2 数据版本管理别再用文件名区分了我见过太多项目的数据集管理方式是train_v1.csv、train_v2_fixed.csv、train_v2_fixed_real.csv。这种命名方式在单人开发的时候还能凑合一旦多人协作或者需要回溯实验就是灾难。你根本不知道哪个版本对应哪次实验也不知道某个版本里到底改了什么。我的做法是引入数据版本控制的概念。最简单的方案是用DVCData Version Control它可以把大数据集用Git的方式管理起来。你不需要把数据本身提交到Git仓库只需要提交一个.dvc文件里面记录了数据的哈希值和存储路径。每次数据变更都会生成一个新的版本号实验记录里只需要写版本号就行。# 初始化DVC dvc init # 添加数据集 dvc add data/train.csv # 提交变更 git add data/train.csv.dvc data/.gitignore git commit -m update training data to v2如果觉得DVC太重也可以自己用哈希值做版本管理。每次数据预处理完成后计算整个数据集的MD5把哈希值写进实验配置文件。这样虽然原始但足够可靠。关键是养成习惯任何一次模型训练都必须能追溯到确切的数据版本。2.3 数据预处理的那些“隐形陷阱”数据预处理看起来简单但里面藏着不少坑。我挑几个最典型的说说。第一个是归一化参数的计算。很多人做图像归一化的时候直接抄了ImageNet的均值和方差。但如果你的数据集和ImageNet分布差异很大比如医学影像或者工业质检图像用ImageNet的参数反而会拖累模型效果。正确的做法是在你自己的训练集上统计均值和方差。注意只能用训练集统计不能用验证集或测试集否则会造成数据泄露。第二个是文本分词的截断策略。处理长文本的时候通常会设置一个最大长度比如512个token。但截断方式有讲究是从头截断、从尾截断还是头尾各保留一部分这取决于你的任务。如果是情感分类关键信息可能在开头或结尾如果是问答任务答案可能藏在中间。我一般会先分析一下文本长度的分布如果超过最大长度的样本比例很高就需要考虑换更长的模型或者做分段处理。第三个是类别不平衡的处理时机。很多人一上来就做重采样或者加权但忽略了数据增强本身就可以缓解不平衡问题。我的经验是先做数据增强再看效果如果还不够再考虑加权损失函数最后才用重采样。因为重采样会改变数据的真实分布可能引入新的偏差。3. 模型训练从能跑到跑得好的距离3.1 训练循环里那些必须自己写的逻辑现在有很多训练框架比如PyTorch Lightning、HuggingFace Trainer它们把训练循环封装得很好。但我建议你在初期至少手写一次完整的训练循环因为只有自己写过才能理解那些框架到底帮你做了什么。一个完整的训练循环至少包含这几个部分前向传播、损失计算、反向传播、参数更新、梯度清零。听起来简单但每个环节都有细节。比如梯度清零如果你忘了写optimizer.zero_grad()梯度就会累加导致模型更新方向完全错误。这个bug不会报错只会让你的loss曲线看起来很奇怪排查起来非常费劲。# 一个最小但完整的训练循环 model.train() for epoch in range(num_epochs): for batch_idx, (data, target) in enumerate(train_loader): data, target data.to(device), target.to(device) optimizer.zero_grad() # 梯度清零 output model(data) # 前向传播 loss criterion(output, target) # 损失计算 loss.backward() # 反向传播 # 梯度裁剪防止梯度爆炸 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() # 参数更新 # 学习率调度 scheduler.step()梯度裁剪这个操作在训练Transformer类模型时几乎是必须的。我遇到过好几次loss突然变成NaN的情况最后查出来都是梯度爆炸导致的。加上clip_grad_norm_之后训练稳定性明显提升。max_norm一般设1.0或者5.0具体看模型和任务。3.2 学习率调度不是越复杂越好学习率是训练中最重要的超参数之一。我见过有人用余弦退火加上热重启再加上自适应调整搞得非常复杂结果反而不如一个简单的阶梯下降。我的建议是先从简单的策略开始确认模型能正常收敛后再尝试更复杂的调度。最常用的三种学习率调度策略策略适用场景优点缺点阶梯下降传统CNN训练简单可控需要手动设置下降节点余弦退火Transformer类模型平滑下降后期学习率小训练周期需要预设线性预热衰减大模型微调防止初期震荡预热步数需要调我个人的习惯是如果是从头训练一个小模型用阶梯下降每30个epoch降为原来的0.1如果是微调预训练模型用线性预热加余弦衰减预热步数设为总步数的10%。这个配置在大多数场景下都能work。还有一个细节warmup阶段的学习率不能从0开始。如果从0开始前几个step的梯度更新几乎为0相当于浪费了。一般从base_lr * 0.01或者base_lr * 0.1开始线性增加到base_lr。3.3 模型保存与恢复别只保存权重很多人保存模型的时候只保存state_dict()也就是模型的权重参数。这样做的问题是你恢复模型的时候必须重新定义模型结构而且如果模型结构改了旧权重可能加载失败。更麻烦的是你丢失了优化器的状态恢复训练的时候优化器要从头开始这会导致loss曲线出现明显的跳变。我的做法是保存一个完整的checkpoint包含以下内容checkpoint { epoch: epoch, model_state_dict: model.state_dict(), optimizer_state_dict: optimizer.state_dict(), scheduler_state_dict: scheduler.state_dict(), best_metric: best_metric, config: config, # 训练配置 } torch.save(checkpoint, fcheckpoint_epoch_{epoch}.pt)这样恢复的时候模型、优化器、调度器的状态都能完整还原训练可以无缝继续。另外我建议同时保存一个“最佳模型”和一个“最新模型”。最佳模型用于推理部署最新模型用于恢复训练。如果磁盘空间允许还可以保留最近几个epoch的checkpoint防止训练崩溃后损失太大。3.4 实验追踪别靠记忆和Excel我早期做实验的时候用Excel记录每次训练的配置和结果。后来实验多了Excel里几百行数据根本分不清哪次是哪次。更糟糕的是有时候改了代码忘了记录结果复现不出来之前的好结果。后来我用了实验追踪工具情况才好转。这类工具的核心功能很简单记录每次实验的超参数、指标曲线、代码版本、数据版本。你可以用TensorBoard、Weights Biases、MLflow或者自己用SQLite搭一个简单的记录系统。关键不是用哪个工具而是养成每次实验都记录的习惯。我自己的最小记录方案是这样的每次训练启动时自动生成一个实验ID把配置文件和Git commit hash写进一个JSON文件训练过程中的loss和指标追加写入CSV。这样即使不用任何第三方工具也能保证实验可追溯。4. 推理服务从实验室到生产环境的惊险一跃4.1 为什么你的模型在线上变慢了实验室里推理一张图片可能只要10毫秒部署到线上后却变成了100毫秒。这种性能落差非常常见原因通常有这几个第一批处理策略不同。实验室里你可能一次推理一个batch线上请求是逐个到达的。如果每个请求都单独推理GPU利用率极低。解决方案是动态批处理把短时间内到达的多个请求攒成一个batch一起推理。这个攒批的时间窗口需要权衡太长会增加延迟太短则批大小不够。第二预处理和后处理的开销。实验室里你可能忽略了图片解码、缩放、归一化的时间但这些操作在线上是实打实要消耗CPU的。我建议把预处理逻辑用C或者Rust重写或者至少用多线程并行处理。第三模型没有做推理优化。PyTorch的默认推理模式并没有做图优化。你可以用torch.jit.trace或者torch.jit.script把模型转成TorchScript或者用ONNX Runtime、TensorRT做进一步优化。实测下来这些优化能带来2到5倍的性能提升。# 将PyTorch模型转为TorchScript model.eval() example_input torch.randn(1, 3, 224, 224).to(device) traced_model torch.jit.trace(model, example_input) traced_model.save(model_traced.pt) # 推理时直接加载 loaded_model torch.jit.load(model_traced.pt)4.2 服务框架选型FastAPI还是Triton推理服务的框架选择取决于你的场景复杂度。如果只是简单的模型推理FastAPI加上Uvicorn就足够了。它的优势是轻量、灵活、Python生态好。但如果你的场景涉及多模型编排、动态批处理、模型版本管理那NVIDIA Triton或者TorchServe会更合适。我用FastAPI搭过一个图像分类服务核心代码不到50行from fastapi import FastAPI, File, UploadFile import torch from PIL import Image import io app FastAPI() model torch.jit.load(model_traced.pt) model.eval() app.post(/predict) async def predict(file: UploadFile File(...)): image_bytes await file.read() image Image.open(io.BytesIO(image_bytes)).convert(RGB) # 预处理 tensor preprocess(image).unsqueeze(0) with torch.no_grad(): output model(tensor) # 后处理 result postprocess(output) return {class: result}这个服务跑起来很简单但有几个问题没有批处理、没有并发控制、没有健康检查。如果要做生产级部署至少还需要加上请求队列、超时控制、优雅关闭这些逻辑。4.3 显存管理推理服务最头疼的问题推理服务和训练最大的区别是训练可以慢慢跑推理必须快速响应。而显存是推理服务最稀缺的资源。我遇到过好几次因为显存泄漏导致服务崩溃的情况。显存泄漏的常见原因有几个第一在推理循环里不断创建新的tensor而没有释放第二使用了torch.no_grad()但某些操作仍然记录了计算图第三多个请求共享模型实例时某些中间变量没有被正确清理。我的经验是推理代码里所有涉及tensor的操作都要用with torch.no_grad():包起来。另外定期用torch.cuda.empty_cache()清理缓存但不要频繁调用因为这会强制同步影响性能。更好的做法是监控显存使用情况设置一个阈值超过阈值时触发清理。还有一个技巧是限制并发请求数。如果显存只够同时处理4个请求那就用信号量把并发数控制在4。超出的请求排队等待而不是直接拒绝。这样虽然会增加一些延迟但能保证服务不崩溃。5. 性能调优让AI系统跑得更快更稳5.1 混合精度训练省显存还能加速混合精度训练是我最推荐的优化手段之一。它的核心思想是在保持模型权重为FP32的同时用FP16来做前向和反向计算。这样既能减少显存占用又能利用GPU的FP16计算单元加速。在PyTorch里用torch.cuda.amp可以很方便地实现from torch.cuda.amp import autocast, GradScaler scaler GradScaler() for data, target in train_loader: optimizer.zero_grad() with autocast(): output model(data) loss criterion(output, target) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()GradScaler的作用是防止FP16的梯度下溢。因为FP16能表示的最小正数比FP32大小梯度会变成0。GradScaler会先把loss放大计算梯度后再缩回来保证梯度不丢失。实测下来混合精度训练能减少大约40%的显存占用训练速度提升20%到30%。但注意有些操作在FP16下不稳定比如softmax、layer norm这些层通常会自动保持FP32。如果遇到loss变成NaN可以尝试用torch.cuda.amp.autocast(enabledFalse)把某些层排除。5.2 梯度累积小显存跑大batch如果你的显存不够大但又想用大batch训练梯度累积是一个好办法。它的原理是连续做多次前向和反向但不更新参数等累积到一定步数后再统一更新。这样等效于大batch训练但显存占用不变。accumulation_steps 4 for i, (data, target) in enumerate(train_loader): with autocast(): output model(data) loss criterion(output, target) / accumulation_steps scaler.scale(loss).backward() if (i 1) % accumulation_steps 0: scaler.step(optimizer) scaler.update() optimizer.zero_grad()注意loss要除以accumulation_steps这样累积后的梯度才是正确的平均值。另外BatchNorm层在梯度累积下会有问题因为每个micro-batch的统计量不同。解决方案是用SyncBatchNorm或者GroupNorm替代。5.3 推理延迟优化从100ms到20ms的实战记录我之前优化过一个文本分类服务初始延迟是100ms左右。经过一系列优化最终降到了20ms。这里把优化过程完整记录一下。第一步定位瓶颈。用cProfile和torch.profiler分析发现60%的时间花在了分词和预处理上只有30%在模型推理。这很反直觉但确实如此。第二步优化分词。原来的分词是用Python循环逐个处理改成批量分词后时间从40ms降到了8ms。HuggingFace的tokenizer支持批量输入直接传一个列表进去就行。第三步模型推理优化。把PyTorch模型转成ONNX再用ONNX Runtime推理。这一步把模型推理时间从30ms降到了12ms。ONNX Runtime对算子做了融合和优化而且支持多线程并行。第四步后处理优化。原来的后处理用了pandas开销很大。改成纯numpy操作后时间从10ms降到了2ms。最终总延迟从100ms降到了22ms左右。这个案例说明推理服务的瓶颈往往不在模型本身而在周边的数据处理逻辑。6. 踩坑实录那些让我熬夜排查的典型问题6.1 数据加载器死锁一个多进程的经典陷阱有一次训练跑着跑着就卡住了GPU利用率变成0但进程还在。用py-spy抓了一下堆栈发现主进程卡在DataLoader的__next__上worker进程卡在某个锁上。这是典型的多进程死锁。原因是我的数据集类里用了一个全局的threading.Lock而DataLoader的多个worker进程会各自复制一份这个锁。当某个worker持有锁的时候其他worker在等待但主进程又在等待所有worker返回数据形成了循环等待。解决方案很简单不要在数据集类里使用全局锁。如果确实需要共享资源用multiprocessing.Manager或者把资源做成只读的。另外DataLoader的persistent_workersTrue在某些情况下也会加剧死锁问题如果遇到卡死可以先把这个参数关掉试试。6.2 显存碎片化为什么empty_cache不管用显存碎片化是另一个让人头疼的问题。表现是明明nvidia-smi显示还有不少空闲显存但一申请就OOM。这是因为空闲显存不是连续的无法满足大块内存的申请。torch.cuda.empty_cache()只能释放缓存分配器里未使用的显存但不能解决碎片化问题。真正有效的做法是在训练开始前就分配好足够大的显存池。PyTorch提供了PYTORCH_CUDA_ALLOC_CONF环境变量可以设置max_split_size_mb来减少碎片。export PYTORCH_CUDA_ALLOC_CONFmax_split_size_mb:128这个设置的意思是大于128MB的内存块不会被拆分。这样可以减少小碎片但可能会浪费一些显存。需要根据你的模型大小来调整。另外如果碎片化非常严重重启服务是最彻底的解决方案。所以生产环境最好有自动重启机制。6.3 模型精度下降一个被忽略的BN层问题有一次我把训练好的模型部署到线上发现精度比验证集低了5个百分点。排查了很久最后发现是BatchNorm层在推理时没有切换到eval模式。BatchNorm在训练时用当前batch的均值和方差在推理时用训练阶段累积的滑动平均。如果推理时忘了调model.eval()BatchNorm就会用推理batch的统计量导致结果不稳定。更隐蔽的是如果推理batch大小为1方差为0输出会完全错误。这个问题的教训是推理代码里必须显式调用model.eval()。另外如果你用了Dropout层model.eval()也会把它关掉。所以这个调用是必须的不能省。还有一个相关的问题如果训练时用了SyncBatchNorm保存的模型在单卡推理时可能会有问题。解决方案是在保存模型前把SyncBatchNorm转回普通的BatchNorm。7. 写给想入坑AI工程的朋友7.1 工具链学习路径先深后广AI工程涉及的工具非常多PyTorch、TensorFlow、ONNX、Triton、Docker、Kubernetes、MLflow、DVC……如果每个都学很容易陷入“什么都会一点什么都不精”的困境。我的建议是先在一个方向上做深再横向扩展。比如你选择PyTorch作为主力框架那就把PyTorch的训练、推理、部署链路全部摸透。等你对PyTorch的生态非常熟悉了再去看TensorFlow或者ONNX会发现很多概念是相通的学习成本大大降低。具体的学习路径可以是先用PyTorch手写一个完整的训练循环理解前向、反向、优化器这些核心概念然后学习如何保存和加载模型理解checkpoint的组成接着学习如何把模型部署成服务理解推理和训练的区别最后学习性能优化包括混合精度、梯度累积、推理加速这些技巧。7.2 调试AI系统的正确姿势调试AI系统和调试传统软件有很大不同。传统软件的bug通常是确定性的输入A一定得到错误B。但AI系统的bug往往是不确定的同样的输入可能得到不同的输出这让排查变得困难。我的调试原则是先固定随机种子再逐步缩小范围。固定随机种子可以消除随机性带来的干扰让问题可复现。然后从数据、模型、训练、推理四个环节逐一排查。数据环节看数据是否正确加载和预处理模型环节看结构是否正确、参数是否正常更新训练环节看loss曲线是否合理推理环节看输入输出是否匹配。另外可视化是调试AI系统的利器。把图片、特征图、注意力权重可视化出来很多问题一眼就能看出来。我遇到过好几次数据标注错误都是通过可视化发现的。7.3 关于“从零开始”的一点个人体会最后说点个人感受。ai-engineering-from-scratch这个标题听起来很硬核好像要从零实现所有东西。但实际上“从零开始”的核心不是拒绝使用工具而是理解工具背后的原理。你可以用PyTorch但你要知道DataLoader的num_workers到底在干什么你可以用ONNX Runtime但你要知道它为什么比原生PyTorch快你可以用Triton但你要知道动态批处理是怎么实现的。这种理解才是你在遇到问题时能快速定位和解决的关键。我见过太多人工具用得很溜但一旦出了问题就束手无策只能重启或者换工具。而真正有经验的工程师能够从日志、指标、堆栈中读出问题的根源然后精准地修复它。这种能力不是靠调包能获得的必须通过亲手实践和不断踩坑来积累。所以如果你真的想入坑AI工程我的建议是找一个实际的小项目从数据准备到模型部署完整地走一遍。不要怕踩坑每一个坑都是你成长的阶梯。等你踩过足够多的坑回头看那些曾经让你熬夜的问题你会发现它们其实都很简单。