1. 从零手搓AI工程为什么我不建议你直接调包1.1 一个让我彻底改变主意的真实场景去年帮一个朋友排查线上推理服务的问题现象很典型模型在测试集上指标漂亮得不行一上生产环境延迟直接飙到800ms以上GPU利用率却只有30%出头。团队里几个小伙子围着代码看了两天从模型结构查到数据预处理愣是没找到瓶颈。最后我让他们把推理流程拆开逐段打点才发现问题出在一个谁都没注意的地方——每次请求都在重新加载tokenizer的词表文件一个不到2MB的JSON硬生生被读了上万次。这件事对我触动挺大。现在市面上讲AI工程的内容绝大多数都在教你怎么调transformers的pipeline、怎么用LangChain搭个链、怎么在云平台上点几下部署一个端点。这些当然有用但它们是“用工具”不是“懂工程”。一旦出了问题你连从哪儿下手都不知道因为你根本不清楚工具内部发生了什么。ai-engineering-from-scratch这个方向说白了就是反过来走——先别急着调包咱们从最底层把AI工程涉及的核心环节自己实现一遍。不是为了造轮子而造轮子而是为了在真正需要调包的时候你知道那个包里面大概长什么样哪里可能出问题出了问题该往哪个方向查。1.2 这个项目到底适合谁不适合谁先说清楚定位免得你花时间看了半天发现不是自己要的。适合的人有一定Python基础用过至少一个深度学习框架PyTorch或TensorFlow都行但总觉得自己是在“黑盒操作”——模型能跑起来但说不清里面每一步在干什么。想系统性地把AI工程从数据到部署这条链路打通建立完整的工程直觉。不太适合的人完全零编程基础的朋友建议先去补Python和基础数学以及已经在做AI工程多年、对底层机制门儿清的老手这里面的内容对你来说可能偏基础。我自己的经验是从零实现一遍的价值不在于你以后真的会手写这些代码而在于你脑子里有了一张完整的地图。当别人跟你说“模型推理慢”的时候你能立刻在脑子里定位到可能的几个环节是预处理拖后腿了是显存没管好是batch策略有问题还是后处理在反复做无用功这种定位能力是调包调不出来的。1.3 整体思路把AI工程拆成五块硬骨头我打算按这个顺序来拆数据处理流水线、模型构建与训练循环、推理优化与部署、监控与可观测性、以及工程化收尾。这五块基本覆盖了一个AI系统从原始数据到线上服务的完整生命周期。为什么是这个顺序因为它是数据流动的自然顺序。数据从哪儿来、怎么变成模型能吃的格式、模型怎么学、学完了怎么高效地跑起来、跑起来之后怎么知道它跑得好不好——这是一条线顺着走不容易乱。每一块我都会先讲“为什么需要它”再讲“从零怎么实现一个最小可用版本”最后讲“实际工程中会踩哪些坑”。代码我会尽量用PyTorch因为它的动态图机制对理解底层逻辑更友好而且现在工业界用它的比例也确实高。注意从零实现不等于生产级实现。我们写的代码是为了理解原理性能上肯定比不过成熟库。但理解了原理之后你再用成熟库的时候就知道该关注哪些参数、该在哪些地方做优化。2. 数据处理流水线别让脏数据毁了你的一切2.1 为什么数据加载是第一个性能杀手很多人做AI项目模型结构调了又调超参数试了又试最后发现瓶颈在数据加载上。这不是段子是常态。我见过一个图像分类任务模型本身推理只要15ms但数据预处理加加载花了120ms整个流水线被拖慢了8倍。从零实现数据流水线核心要解决三个问题怎么高效读、怎么高效转、怎么高效喂。高效读的关键在于避免随机IO。机械硬盘的随机读取速度比顺序读取慢两个数量级即使是SSD大量小文件的随机读也会成为瓶颈。所以工程上的标准做法是把所有训练数据预先打包成少数几个大文件比如TFRecord、WebDataset的tar包、或者HDF5读取的时候顺序读大块然后在内存里做切分。高效转的核心是并行化。CPU有多核但Python的GIL让多线程在计算密集型任务上几乎没用。所以数据预处理要用多进程torch.utils.data.DataLoader的num_workers参数就是干这个的。但这里有个坑worker进程和主进程之间的数据传输是有成本的如果每个样本很小传输开销可能比计算开销还大。高效喂的关键是预取。GPU计算的时候CPU不能闲着要提前把下一批数据准备好。这就是prefetch_factor参数的作用。理想情况下数据加载和模型计算应该完全重叠GPU永远不等数据。2.2 手写一个带预取和并行加载的Dataset先看最基础的版本理解数据流import torch from torch.utils.data import Dataset, DataLoader import numpy as np class SimpleDataset(Dataset): def __init__(self, data_path, labels_path): # 实际工程中这里通常是内存映射不是全量加载 self.data np.load(data_path, mmap_moder) self.labels np.load(labels_path) def __len__(self): return len(self.labels) def __getitem__(self, idx): # 这里做单样本的预处理 x self.data[idx].astype(np.float32) / 255.0 y self.labels[idx] return torch.from_numpy(x), torch.tensor(y)这个版本能跑但有几个问题。第一__getitem__里做了类型转换和归一化这些操作在每个epoch都会重复执行浪费算力。第二没有预取GPU等数据。第三如果数据量大np.load全量加载会爆内存。改进版本把预处理结果缓存起来class CachedDataset(Dataset): def __init__(self, data_path, labels_path, cache_size10000): self.data np.load(data_path, mmap_moder) self.labels np.load(labels_path) self.cache {} self.cache_size cache_size def __getitem__(self, idx): if idx in self.cache: return self.cache[idx] x self.data[idx].astype(np.float32) / 255.0 y self.labels[idx] result (torch.from_numpy(x), torch.tensor(y)) # 简单的LRU策略 if len(self.cache) self.cache_size: self.cache.pop(next(iter(self.cache))) self.cache[idx] result return result但缓存有个问题如果数据增强是随机的缓存就失效了。所以实际工程中缓存的是“确定性预处理”的结果随机增强放在后面做。真正高效的方案是用内存映射加预取from torch.utils.data import DataLoader # num_workers设为CPU核心数prefetch_factor设为2-4 loader DataLoader( dataset, batch_size64, shuffleTrue, num_workers8, prefetch_factor4, pin_memoryTrue, # 如果用的是GPU persistent_workersTrue # 避免每个epoch重新创建worker )pin_memoryTrue这个参数很多人忽略但它对GPU训练速度影响很大。它把数据放在锁页内存里GPU可以直接通过DMA访问省去了从普通内存拷贝到显存的时间。实测在数据加载密集的场景下能提升10%-20%的吞吐。2.3 数据增强的工程化陷阱数据增强是提升模型泛化能力的利器但工程上坑很多。第一个坑是增强操作放在了错误的位置。如果你在__getitem__里做增强每个epoch都会重新计算而且多进程下每个worker独立做增强结果不可复现。更好的做法是把增强放在GPU上做用torchvision.transforms.v2或者kornia这样可以利用GPU的并行能力而且增强参数可以统一控制。第二个坑是增强强度没有随训练进程调整。训练初期模型欠拟合强增强会拖慢收敛训练后期模型容易过拟合需要强增强来正则化。所以实际工程中常用的是渐进式增强策略比如AutoAugment或者RandAugment的调度版本。第三个坑是验证集也做了增强。验证集的目的是评估模型在真实数据上的表现做增强会引入噪声导致指标虚高。验证集只做确定性的预处理比如resize和归一化。实操心得我习惯在训练脚本里加一个debug模式把增强后的样本可视化出来看几眼。很多增强参数配错了光看代码看不出来一看图就明白了。比如归一化的均值方差写反了或者resize的插值方式选错了这些都会悄悄影响模型性能。3. 模型构建与训练循环把每一行代码都搞清楚3.1 从零写一个训练循环而不是用Trainertransformers.Trainer和PyTorch Lightning确实方便但它们把太多东西藏起来了。从零写训练循环你会被迫面对每一个细节梯度什么时候清零、损失怎么累积、学习率怎么调度、梯度裁剪在哪里做。先看最朴素的版本model MyModel() optimizer torch.optim.AdamW(model.parameters(), lr1e-4) criterion torch.nn.CrossEntropyLoss() for epoch in range(num_epochs): model.train() for batch_idx, (data, target) in enumerate(train_loader): data, target data.cuda(), target.cuda() optimizer.zero_grad() output model(data) loss criterion(output, target) loss.backward() optimizer.step()这个循环能跑但有几个问题。第一没有验证环节你不知道模型有没有过拟合。第二没有学习率调度固定学习率很难收敛到最优。第三没有梯度裁剪遇到梯度爆炸直接NaN。第四没有混合精度训练显存利用率低。加上这些之后scaler torch.cuda.amp.GradScaler() scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxnum_epochs) for epoch in range(num_epochs): model.train() train_loss 0.0 for data, target in train_loader: data, target data.cuda(), target.cuda() optimizer.zero_grad() with torch.cuda.amp.autocast(): output model(data) loss criterion(output, target) scaler.scale(loss).backward() scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(optimizer) scaler.update() train_loss loss.item() scheduler.step() # 验证环节 model.eval() val_loss 0.0 correct 0 with torch.no_grad(): for data, target in val_loader: data, target data.cuda(), target.cuda() output model(data) val_loss criterion(output, target).item() pred output.argmax(dim1) correct pred.eq(target).sum().item() print(fEpoch {epoch}: train_loss{train_loss/len(train_loader):.4f}, fval_loss{val_loss/len(val_loader):.4f}, fval_acc{correct/len(val_dataset):.4f})这里有几个关键点值得展开。混合精度训练autocast自动把部分运算转成float16减少显存占用、加速计算。但float16的数值范围小梯度容易下溢所以需要GradScaler把损失放大反向传播后再缩回来。这个机制不理解的话遇到NaN根本不知道从哪儿查。梯度裁剪clip_grad_norm_把所有参数的梯度拼成一个向量算它的L2范数如果超过max_norm就整体缩放。这是防止梯度爆炸的标准手段尤其在RNN和Transformer上几乎是必须的。学习率调度CosineAnnealing让学习率按余弦曲线从初始值降到0比固定学习率收敛得更稳。但T_max要设对一般等于总训练步数。设小了学习率提前降到0模型还没收敛设大了学习率降得太慢后期震荡。3.2 梯度累积小显存跑大batch的实用技巧显存不够是常态。想用大batch提升训练稳定性但显卡装不下怎么办梯度累积。原理很简单正常是每个batch算一次梯度、更新一次参数。梯度累积是算N个batch的梯度、累加起来、再更新一次。效果上等价于把batch size放大了N倍但显存占用不变。accumulation_steps 4 for i, (data, target) in enumerate(train_loader): data, target data.cuda(), target.cuda() with torch.cuda.amp.autocast(): output model(data) loss criterion(output, target) / accumulation_steps scaler.scale(loss).backward() if (i 1) % accumulation_steps 0: scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(optimizer) scaler.update() optimizer.zero_grad()注意损失要除以accumulation_steps否则梯度会放大N倍。另外optimizer.zero_grad()要放在更新之后不能每个batch都清。踩过的坑有一次我忘了除accumulation_steps结果梯度大了4倍学习率没调直接训练发散。排查了半天才发现是这个低级错误。所以这种地方一定要写注释提醒自己。3.3 模型保存与恢复别只存state_dict很多人保存模型只存model.state_dict()恢复的时候还要重新定义模型结构。这在实验阶段没问题但到了部署阶段就是灾难——你得确保部署环境的代码和训练环境完全一致否则加载就报错。更稳妥的做法是保存完整的模型对象或者至少保存模型结构信息# 保存 checkpoint { model_state_dict: model.state_dict(), optimizer_state_dict: optimizer.state_dict(), scheduler_state_dict: scheduler.state_dict(), epoch: epoch, config: model_config, # 模型结构配置 normalization_stats: {mean: mean, std: std} # 预处理参数 } torch.save(checkpoint, checkpoint.pt) # 恢复 checkpoint torch.load(checkpoint.pt) model MyModel(**checkpoint[config]) model.load_state_dict(checkpoint[model_state_dict])把预处理参数也存进去很重要。我见过太多次训练时用了某种归一化部署时忘了导致线上效果和离线评估对不上。这种问题排查起来极其痛苦因为模型本身没问题是数据分布变了。4. 推理优化与部署让模型真正跑得快4.1 推理和训练到底有什么不同训练和推理看起来都是前向传播但工程上的关注点完全不同。训练关注的是吞吐量——每秒能处理多少样本因为训练要跑很多个epoch。推理关注的是延迟——单个请求从进到出要多久因为用户等不了。训练可以容忍较大的显存占用因为batch size大反而有助于收敛。推理要尽量省显存因为要同时服务多个请求。训练时dropout、batch norm这些层是激活的推理时要关掉。这个切换用model.eval()完成但很多人忘了调导致推理结果和验证结果对不上。从零实现推理优化核心是三个方向减少计算量、减少内存访问、增加并行度。4.2 手写一个简单的推理服务先看最朴素的版本用Flask起一个HTTP服务from flask import Flask, request, jsonify import torch import numpy as np app Flask(__name__) model MyModel() model.load_state_dict(torch.load(model.pt)) model.eval() model.cuda() app.route(/predict, methods[POST]) def predict(): data request.json[data] x torch.tensor(data, dtypetorch.float32).cuda() with torch.no_grad(): output model(x.unsqueeze(0)) pred output.argmax(dim1).item() return jsonify({prediction: pred})这个版本能跑但性能很差。每个请求都要做一次完整的模型前向没有batchGPU利用率极低。而且Flask默认是单线程的并发请求会排队。改进方向有几个。第一用动态batch——把短时间内到达的多个请求攒成一个batch一起推理。第二用异步框架比如FastAPI加uvicorn支持更高的并发。第三把模型导出成ONNX或TorchScript减少Python解释器的开销。动态batch的实现思路import asyncio from collections import deque class BatchProcessor: def __init__(self, model, max_batch_size32, max_wait_ms10): self.model model self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue deque() self.lock asyncio.Lock() async def add_request(self, data): future asyncio.Future() async with self.lock: self.queue.append((data, future)) if len(self.queue) self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch list(self.queue) self.queue.clear() data torch.stack([item[0] for item in batch]).cuda() with torch.no_grad(): outputs self.model(data) for i, (_, future) in enumerate(batch): future.set_result(outputs[i].cpu().numpy())这个逻辑是请求进来先放队列攒够max_batch_size或者等够max_wait_ms就一起推理。这样GPU的利用率能提升好几倍。4.3 模型量化用精度换速度的实操量化是把float32的权重和激活值转成int8模型大小直接缩小4倍推理速度提升2-4倍精度损失通常在1%以内。对于大多数应用场景这个 trade-off 非常划算。PyTorch支持动态量化和静态量化。动态量化最简单一行代码quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )但动态量化只对Linear层有效而且是在运行时动态计算量化参数加速有限。静态量化需要校准数据提前算好量化参数加速更明显model.qconfig torch.quantization.get_default_qconfig(fbgemm) model_prepared torch.quantization.prepare(model) # 用校准数据跑一遍收集激活值分布 with torch.no_grad(): for data, _ in calib_loader: model_prepared(data) model_quantized torch.quantization.convert(model_prepared)校准数据的选择很关键。要用真实分布的数据不能用随机噪声。一般从训练集里抽几百到几千个样本就够了。注意事项量化对某些操作不友好比如LayerNorm、Softmax这些对数值范围敏感的层。如果量化后精度掉得厉害可以只量化部分层保留这些敏感层为float32。PyTorch支持混合量化通过qconfig精细控制。4.4 推理引擎选型ONNX Runtime vs TorchScript模型训练完部署的时候不一定非要用PyTorch。ONNX Runtime和TorchScript是两个主流选择。TorchScript是PyTorch自带的把模型编译成中间表示脱离Python解释器运行。优点是兼容性好PyTorch支持的操作它基本都支持。缺点是优化程度一般。ONNX Runtime是微软主导的跨框架推理引擎支持图优化、算子融合、多后端加速。在CPU上通常比TorchScript快在GPU上差距不大。导出ONNX的代码dummy_input torch.randn(1, 3, 224, 224).cuda() 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_version13 )dynamic_axes很重要它让导出的模型支持变长batch。不设的话模型只能处理固定batch size部署时很受限。选型建议如果团队已经在用PyTorch生态TorchScript上手最快如果追求极致性能或者需要跨框架部署ONNX Runtime更合适。实测在CPU推理场景下ONNX Runtime通常比TorchScript快20%-50%。5. 监控与可观测性上线只是开始5.1 为什么模型上线后性能会下降模型在离线评估时指标很好上线后却慢慢变差这是AI工程中最常见也最头疼的问题。原因通常有三个数据漂移、概念漂移、以及工程层面的问题。数据漂移是输入数据的分布变了。比如训练时用户主要是年轻人上线后中老年用户增多输入特征分布就变了。概念漂移是输入和输出的关系变了。比如推荐系统里用户的兴趣随时间变化同样的行为对应的偏好不同了。工程层面的问题更隐蔽。比如特征计算逻辑在训练和推理时不一致训练时用了未来信息数据泄露推理时没有或者某个特征在训练时是数值型推理时变成了字符串类型转换悄悄失败变成了默认值。从零实现监控核心是记录三类指标输入指标、输出指标、以及性能指标。5.2 手写一个轻量级监控模块不需要一上来就上Prometheus加Grafana先用最朴素的方式把数据记下来import json import time from collections import defaultdict import numpy as np class ModelMonitor: def __init__(self, window_size1000): self.window_size window_size self.input_stats defaultdict(list) self.output_stats defaultdict(list) self.latencies [] self.errors 0 self.total 0 def record(self, inputs, outputs, latency_ms, errorNone): self.total 1 if error: self.errors 1 return # 记录输入统计 for name, value in inputs.items(): self.input_stats[name].append(float(value)) if len(self.input_stats[name]) self.window_size: self.input_stats[name].pop(0) # 记录输出统计 if isinstance(outputs, np.ndarray): self.output_stats[mean].append(float(outputs.mean())) self.output_stats[std].append(float(outputs.std())) # 记录延迟 self.latencies.append(latency_ms) if len(self.latencies) self.window_size: self.latencies.pop(0) def get_report(self): report { total_requests: self.total, error_rate: self.errors / max(self.total, 1), latency_p50: np.percentile(self.latencies, 50) if self.latencies else 0, latency_p95: np.percentile(self.latencies, 95) if self.latencies else 0, latency_p99: np.percentile(self.latencies, 99) if self.latencies else 0, } for name, values in self.input_stats.items(): if values: report[finput_{name}_mean] np.mean(values) report[finput_{name}_std] np.std(values) return report这个模块记录输入特征的均值和标准差、输出分布的均值和标准差、以及延迟的分位数。当输入特征的均值偏离训练时的均值超过一定阈值就说明可能发生了数据漂移。延迟的P99比P50更重要。用户对平均延迟不敏感但对最慢的那1%的请求极其敏感。如果P99延迟是2秒意味着每100个用户就有1个要等2秒这个体验是灾难性的。5.3 数据漂移检测的实用方法检测数据漂移工程上常用三种方法。统计检验对每个特征比较当前窗口的分布和训练时的分布。连续特征用KS检验类别特征用卡方检验。p值小于0.05就认为有显著差异。PSIPopulation Stability Index把特征分桶比较各桶的占比变化。PSI小于0.1说明分布稳定0.1到0.25说明有轻微漂移大于0.25说明有显著漂移。这个方法在金融风控领域用得很多因为它对分桶方式不敏感结果比较稳健。模型置信度监控如果模型输出的置信度整体下降说明输入数据可能偏离了训练分布。这个方法不需要标签适合实时监控。def calculate_psi(expected, actual, buckets10): # 用训练数据的分位数分桶 breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) breakpoints[0] -np.inf breakpoints[-1] np.inf expected_counts np.histogram(expected, binsbreakpoints)[0] / len(expected) actual_counts np.histogram(actual, binsbreakpoints)[0] / len(actual) # 避免除零 expected_counts np.clip(expected_counts, 1e-6, None) actual_counts np.clip(actual_counts, 1e-6, None) psi np.sum((actual_counts - expected_counts) * np.log(actual_counts / expected_counts)) return psi实操心得PSI的计算依赖分桶分桶方式不同结果会有差异。我的经验是对于长尾分布的特征用等频分桶比等宽分桶更稳定。另外PSI要定期算不能只看一次。我习惯每天算一次画成趋势图这样能提前发现缓慢的漂移。5.4 日志与追踪出问题时怎么快速定位监控告诉你“有问题”日志和追踪告诉你“问题在哪儿”。结构化日志是基础。不要用print用logging模块输出JSON格式方便后续检索和分析import logging import json logger logging.getLogger(__name__) def log_prediction(request_id, inputs, outputs, latency_ms, model_version): logger.info(json.dumps({ request_id: request_id, model_version: model_version, latency_ms: latency_ms, input_shape: list(inputs.shape) if hasattr(inputs, shape) else None, output_summary: { mean: float(outputs.mean()), std: float(outputs.std()), max: float(outputs.max()), }, timestamp: time.time() }))request_id是关键它让一个请求在多个服务之间的调用链可以串起来。如果推理服务调用了特征服务特征服务又调用了数据库通过request_id能把整条链路串起来定位瓶颈在哪个环节。分布式追踪用OpenTelemetry它能在每个服务里自动埋点生成调用链的火焰图。火焰图上看哪个span最宽瓶颈就在哪儿。这个工具上手有点门槛但一旦用起来排查性能问题效率提升巨大。6. 工程化收尾那些没人告诉你但很重要的事6.1 配置管理别把参数写死在代码里实验阶段把学习率、batch size写死在代码里没问题但到了工程阶段这些参数必须外置。原因很简单不同环境开发、测试、生产需要不同的配置写死在代码里意味着每次改配置都要改代码、重新部署。用YAML或JSON管理配置# config/production.yaml model: name: resnet50 num_classes: 1000 pretrained: true training: batch_size: 128 learning_rate: 0.001 epochs: 100 optimizer: adamw weight_decay: 0.01 data: train_path: /data/train val_path: /data/val num_workers: 8 image_size: 224加载配置的代码import yaml def load_config(envdevelopment): with open(fconfig/{env}.yaml, r) as f: config yaml.safe_load(f) return config配置管理的一个常见坑是配置项没有默认值。如果某个配置项在某个环境里忘了写程序直接崩溃。所以加载配置后要做校验确保所有必填项都存在。6.2 版本管理模型、数据、代码一个都不能少AI工程和传统软件工程最大的区别是除了代码还有模型和数据需要版本管理。代码用Git这个不用多说。模型版本管理小团队可以用文件命名规范比如model_v1.2.3_20240101.pt大团队建议用MLflow或DVC。数据版本管理更复杂因为数据量大不能直接放Git。DVC的思路是数据文件本身存在对象存储里Git里只存一个指针文件记录数据的哈希值。我自己的习惯是每次训练完把模型文件、训练配置、评估指标、以及训练数据的哈希值打包成一个版本存到对象存储里。这样任何时候都能复现某个版本的模型。6.3 测试AI系统怎么测AI系统的测试比传统软件复杂因为输出不是确定性的。但有几类测试是必须做的。数据测试检查数据格式、范围、分布是否符合预期。比如图像数据要检查像素值是否在0-255之间标签是否在合法范围内。用great_expectations或pandera可以自动化这些检查。模型测试检查模型输出是否在合理范围内是否对特定输入有预期行为。比如一个分类模型对随机噪声的输入输出的置信度应该很低对明显属于某个类别的输入输出的对应类别概率应该很高。集成测试检查整个流水线从数据加载到推理输出是否正常工作。这个测试要覆盖边界情况比如空输入、超大输入、格式错误的输入。回归测试每次模型更新后在固定的测试集上评估确保指标没有下降。这个测试要自动化集成到CI/CD流程里。def test_model_output_range(): model load_model() dummy_input torch.randn(1, 3, 224, 224) output model(dummy_input) # 输出应该是概率分布 assert output.shape (1, 1000) assert torch.allclose(output.sum(dim1), torch.ones(1), atol1e-5) assert (output 0).all() and (output 1).all() def test_model_deterministic(): model load_model() model.eval() dummy_input torch.randn(1, 3, 224, 224) with torch.no_grad(): output1 model(dummy_input) output2 model(dummy_input) # eval模式下同样输入应该得到同样输出 assert torch.allclose(output1, output2)注意test_model_deterministic只在model.eval()模式下成立。训练模式下有dropout和batch norm的随机性输出不确定。这个测试能帮你发现忘了调eval()的低级错误。6.4 文档写给三个月后的自己AI项目最容易烂尾的地方是文档。实验阶段大家都不写文档等到要交接或者要复现的时候发现什么都记不清了。我的经验是至少写三份文档。README项目是干什么的、怎么安装、怎么跑起来。实验记录每次实验的配置、结果、结论用表格记录方便对比。决策日志为什么选了这个方案而不是那个当时的考虑是什么。这份文档在几个月后回头看价值最大。实验记录可以用Markdown表格实验ID模型学习率Batch Size数据增强Val Acc备注exp001ResNet501e-364无76.2%基线exp002ResNet501e-364RandomCrop78.5%增强有效exp003ResNet501e-4128RandomCrop79.1%最佳决策日志用时间线记录2024-01-15: 选择ResNet50而不是ViT因为训练数据只有10万张ViT容易过拟合。 2024-01-20: 学习率从1e-3降到1e-4因为loss震荡严重。 2024-01-25: 加入梯度累积因为显存不够batch size从128降到32累积4步。这些记录看起来琐碎但当你三个月后需要复现某个实验或者新人接手项目的时候它们能省下大量时间。6.5 持续集成与持续部署让更新变得安全AI系统的CI/CD和传统软件不同因为模型更新需要额外的验证步骤。典型的流程是代码提交触发CI跑单元测试和集成测试测试通过后触发模型训练或模型评估评估指标超过基线才允许部署到生产环境。部署策略上金丝雀发布Canary Release比全量发布安全得多。先把新模型部署到一小部分流量上观察一段时间指标正常再逐步扩大流量。如果指标异常自动回滚到旧模型。# 简单的金丝雀发布逻辑 class ModelRouter: def __init__(self, stable_model, canary_model, canary_ratio0.1): self.stable_model stable_model self.canary_model canary_model self.canary_ratio canary_ratio def predict(self, request_id, inputs): # 用request_id做哈希保证同一个用户始终路由到同一个模型 if hash(request_id) % 100 self.canary_ratio * 100: return self.canary_model(inputs), canary return self.stable_model(inputs), stable用request_id做哈希而不是随机数是为了保证同一个用户的请求始终由同一个模型处理。否则用户刷新一下页面模型变了结果也变了体验很差。回滚机制要自动化。监控系统检测到新模型的错误率或延迟超过阈值自动把流量切回旧模型。这个阈值要提前设好不能等出了问题再临时决定。7. 一些零散但值钱的经验7.1 关于工具选型我的真实看法工具选型没有绝对的对错只有适不适合。我的原则是先用最简单的方案把事做成遇到瓶颈再换。比如推理服务一开始用Flask就够了QPS上不去再换FastAPI再上不去再考虑Triton Inference Server。不要一上来就上最复杂的方案因为复杂方案的学习成本和维护成本都很高而你的业务可能根本用不到那些高级特性。再比如实验管理一开始用Excel记结果完全没问题实验多了再用MLflow。工具是为了解决问题不是为了炫技。7.2 性能优化的优先级性能优化要按ROI排序。我的经验是数据加载 模型推理 模型训练 代码微优化。数据加载优化通常收益最大因为很多项目的瓶颈都在IO上。把数据打包成大文件、用内存映射、开多进程加载这些改动能带来数倍的提升。模型推理优化次之量化、ONNX Runtime、动态batch这些能带来2-4倍的提升。模型训练优化再次之混合精度、梯度累积、分布式训练这些能带来1.5-3倍的提升。代码微优化收益最小比如把Python循环改成向量化操作通常只有10%-30%的提升。除非其他都优化完了否则不值得花时间。7.3 关于“从零实现”的边界从零实现是为了理解原理不是为了替代成熟库。你不需要手写一个比PyTorch还快的卷积算子但你需要知道卷积在GPU上是怎么并行计算的这样你才能理解为什么某些形状的输入特别慢。我的建议是核心环节自己实现一遍边缘环节用成熟库。比如数据加载、训练循环、推理服务这些核心环节自己写一遍收获很大。但像分布式通信、CUDA内核这些理解原理就行没必要自己造。7.4 最后分享一个小技巧如果你刚开始做AI工程不知道从哪儿下手我的建议是找一个已经跑通的开源项目把它拆了重写一遍。不要照着抄而是理解每一部分在干什么然后用自己的方式实现。遇到不懂的地方查文档、看源码、做实验。这个过程很慢但收获比看十篇教程都大。我自己就是这么过来的。第一个项目拆的是HuggingFace的transformers里的Trainer花了整整一周才把训练循环、梯度累积、混合精度这些逻辑理清楚。但理清楚之后后面再看其他框架的代码基本都是一眼就懂。AI工程这个领域变化很快但底层的东西变化很慢。数据怎么加载、梯度怎么算、模型怎么部署这些核心问题十年都没变过。把底层搞扎实了上层工具怎么变都不慌。