1. 从零搭建AI工程能力为什么“会调包”远远不够很多人第一次接触AI工程是从一行pip install开始的。装完框架跑通一个官方Demo看着终端里跳出几行训练日志就觉得自己已经“入门”了。可真到了要自己搭一条能跑通、能复现、能交付的流水线时问题就全冒出来了数据怎么组织模型怎么存训练崩了怎么查推理延迟高得离谱又该从哪下手ai-engineering-from-scratch这个标题说的正是这件事——不是从零学AI理论而是从零把AI工程这套“手艺”搭起来。它面向的是那些已经懂一点Python、知道神经网络大概长什么样但一到工程落地就发怵的人。你可能是个转行的开发者也可能是个做数据分析想往AI方向靠的工程师甚至是个学生手里只有一台普通笔记本想搞清楚一个AI项目从数据到上线的完整链路到底长什么样。我自己的经历是最早做AI项目时代码全堆在一个train.py里数据路径写死超参数靠手改模型存成model_final_final_v2.pth。结果换台机器就跑不起来过两周自己都忘了当时改了什么。后来才慢慢明白AI工程的核心不是“模型多牛”而是整条链路的可控性和可复现性。这篇文章就围绕这个目标把从零搭建AI工程能力的关键环节拆开讲包括环境管理、数据管道、训练循环、实验追踪、推理服务以及那些只有踩过才知道的坑。2. 环境与依赖别让“在我机器上能跑”成为口头禅2.1 为什么虚拟环境不是可选项而是必选项刚入门的人最容易忽略的就是环境隔离。系统Python里装了几十个包版本互相打架今天装了个新库把旧项目搞崩了这种事太常见了。AI工程对依赖版本尤其敏感——PyTorch 1.x 和 2.x 的API差异、CUDA版本和驱动版本的匹配、numpy版本和框架的兼容性任何一个不对报错信息能让你查半天。我的做法是每个项目一个独立环境用conda或venv都行但一定要把环境配置写成文件。requirements.txt是最低要求更好的是environment.yml或者pyproject.toml。关键是要把版本号钉死不要写torch1.0这种模糊约束而是写torch2.1.0。因为AI框架的小版本更新经常引入行为变化今天能跑的代码明天可能就报warning甚至error。# 创建独立环境 conda create -n ai-eng python3.10 conda activate ai-eng # 安装核心依赖版本钉死 pip install torch2.1.0 torchvision0.16.0 pip install numpy1.24.3 pandas2.0.3 pip install scikit-learn1.3.0 matplotlib3.7.2提示如果你用GPU安装PyTorch时要去官网查对应的CUDA版本命令不要直接pip install torch否则可能装成CPU版本训练时才发现用不了GPU。2.2 项目目录结构一开始就规划好后面少受罪我见过太多人把数据、代码、模型、日志全扔在一个文件夹里最后自己都分不清哪个文件是干嘛的。一个清晰的目录结构是AI工程能力的第一课。下面是我用了几年、觉得最顺手的结构project/ ├── configs/ # 配置文件YAML或JSON ├── data/ │ ├── raw/ # 原始数据只读不改 │ ├── processed/ # 清洗后的数据 │ └── splits/ # 训练/验证/测试划分 ├── src/ │ ├── data/ # 数据加载和预处理代码 │ ├── models/ # 模型定义 │ ├── train.py # 训练入口 │ ├── evaluate.py # 评估入口 │ └── predict.py # 推理入口 ├── experiments/ # 每次实验的输出 │ └── exp_001/ │ ├── config.yaml │ ├── checkpoints/ │ └── logs/ ├── notebooks/ # 探索性分析不参与生产 ├── requirements.txt └── README.md这个结构的好处是数据和代码分离实验和代码分离。data/raw永远不动所有处理后的数据进data/processed。每次训练在experiments/下新建一个编号目录把当次用的配置、模型权重、日志全存进去。这样过一个月回头看你还能精确复现当时的实验。2.3 配置文件把超参数从代码里赶出去超参数写死在代码里是另一个经典坑。学习率、batch size、模型层数、数据路径这些都应该放在配置文件里。我用YAML因为可读性好也方便程序读取。# configs/base.yaml data: train_path: data/processed/train.csv val_path: data/processed/val.csv batch_size: 32 num_workers: 4 model: name: resnet18 num_classes: 10 pretrained: true train: epochs: 50 lr: 0.001 weight_decay: 0.0001 device: cuda seed: 42 output: dir: experiments/exp_001代码里只负责读配置不负责定义配置。这样改超参数不用动代码也方便做超参数搜索。seed这个字段特别重要固定随机种子能让你的实验可复现不然每次跑出来的结果都不一样根本没法对比。3. 数据管道AI工程里最脏最累但最不能省的活3.1 数据加载的三种境界数据加载这件事新手和老手的差距特别明显。第一层是“能读进来就行”直接pd.read_csv然后全部转成numpy数组塞进内存。数据小的时候没问题数据一大就内存爆炸。第二层是“会用Dataset和DataLoader”这是PyTorch的标准做法能分批加载、支持多进程。第三层是“知道什么时候不该用DataLoader”比如数据量极大、需要流式处理或者需要复杂的在线增强时可能要自己写迭代器。对于大多数从零开始的项目我建议直接用PyTorch的DatasetDataLoader组合。它的好处是接口统一支持num_workers多进程加载还能自定义collate_fn处理变长序列。from torch.utils.data import Dataset, DataLoader import pandas as pd from PIL import Image class MyDataset(Dataset): def __init__(self, csv_path, transformNone): self.df pd.read_csv(csv_path) self.transform transform def __len__(self): return len(self.df) def __getitem__(self, idx): row self.df.iloc[idx] image Image.open(row[image_path]).convert(RGB) label int(row[label]) if self.transform: image self.transform(image) return image, label train_dataset MyDataset(data/processed/train.csv, transformtrain_transform) train_loader DataLoader( train_dataset, batch_size32, shuffleTrue, num_workers4, pin_memoryTrue )num_workers设多少一般设成CPU核心数但不要超过8太多反而因为进程切换开销变慢。pin_memoryTrue在GPU训练时能加速数据传输这个细节很多人不知道。3.2 数据划分别在训练集上评估自己我见过最离谱的错误是有人把全部数据拿来训练然后用训练集上的准确率当最终指标结果上线后效果差得离谱。数据划分是AI工程的底线训练集、验证集、测试集必须严格分开。训练集用来更新参数验证集用来调超参数和早停测试集只在最后评估一次。划分比例看数据量。数据多的话98/1/1都行数据少的话70/15/15或者80/10/10。关键是划分要随机且可复现用固定种子。from sklearn.model_selection import train_test_split df pd.read_csv(data/raw/all.csv) train_df, temp_df train_test_split(df, test_size0.3, random_state42, stratifydf[label]) val_df, test_df train_test_split(temp_df, test_size0.5, random_state42, stratifytemp_df[label]) train_df.to_csv(data/splits/train.csv, indexFalse) val_df.to_csv(data/splits/val.csv, indexFalse) test_df.to_csv(data/splits/test.csv, indexFalse)stratify参数在分类任务里很重要它保证每个类别的比例在划分后保持一致。不然某个类别可能全被分到训练集验证集里一个样本都没有。3.3 数据增强小数据集的救命稻草数据不够的时候数据增强是最划算的手段。图像任务里随机裁剪、翻转、颜色抖动都是标配。但要注意增强只用在训练集验证集和测试集只能用确定性的预处理比如resize和归一化。因为评估时需要的是稳定、可复现的结果随机增强会让每次评估结果都不一样。from torchvision import transforms train_transform transforms.Compose([ transforms.RandomResizedCrop(224), transforms.RandomHorizontalFlip(), transforms.ColorJitter(brightness0.2, contrast0.2), transforms.ToTensor(), transforms.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]) ]) val_transform transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]) ])归一化用的均值和标准差是ImageNet的统计值如果你做的是完全不同的领域比如医学影像最好自己算一下数据集的均值和标准差效果会更好。4. 训练循环把“黑盒”拆成可控的零件4.1 一个标准训练循环该有哪些部分很多人写训练循环就是for epoch in range(epochs): for batch in loader: ...能跑但不够健壮。一个工程化的训练循环至少要有这些部分训练模式切换、梯度清零、前向传播、损失计算、反向传播、参数更新、验证评估、指标记录、模型保存、早停判断。import torch import torch.nn as nn from tqdm import tqdm def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss 0 correct 0 total 0 for images, labels in tqdm(loader, descTraining): images, labels images.to(device), labels.to(device) optimizer.zero_grad() outputs model(images) loss criterion(outputs, labels) loss.backward() optimizer.step() total_loss loss.item() * images.size(0) _, predicted outputs.max(1) correct predicted.eq(labels).sum().item() total labels.size(0) return total_loss / total, correct / total torch.no_grad() def evaluate(model, loader, criterion, device): model.eval() total_loss 0 correct 0 total 0 for images, labels in loader: images, labels images.to(device), labels.to(device) outputs model(images) loss criterion(outputs, labels) total_loss loss.item() * images.size(0) _, predicted outputs.max(1) correct predicted.eq(labels).sum().item() total labels.size(0) return total_loss / total, correct / totalmodel.train()和model.eval()这两个调用千万别省。它们会影响Dropout和BatchNorm的行为训练时用train模式评估时用eval模式搞反了结果会差很多。torch.no_grad()装饰器在评估时能省显存、加速计算因为不需要构建计算图。4.2 学习率调度让训练后期更稳固定学习率不是不行但通常效果不如动态调整。最常用的是余弦退火和阶梯下降。余弦退火让学习率从初始值平滑降到接近零训练后期模型能在局部最小值附近更精细地收敛。from torch.optim.lr_scheduler import CosineAnnealingLR optimizer torch.optim.AdamW(model.parameters(), lr1e-3, weight_decay1e-4) scheduler CosineAnnealingLR(optimizer, T_maxepochs, eta_min1e-6) for epoch in range(epochs): train_loss, train_acc train_one_epoch(...) val_loss, val_acc evaluate(...) scheduler.step() print(fEpoch {epoch}: train_loss{train_loss:.4f}, val_acc{val_acc:.4f}, lr{scheduler.get_last_lr()[0]:.6f})注意scheduler.step()的位置PyTorch 1.1之后应该在每个epoch结束后调用而不是每个batch后。放错位置会导致学习率变化不符合预期。4.3 模型保存与早停别等跑完才后悔训练过程中要定期保存模型尤其是验证集指标创新高的时候。不然训练到一半崩了前面全白跑。早停则是防止过拟合的简单有效手段验证集损失连续几个epoch不下降就停止训练。best_val_acc 0 patience 5 counter 0 for epoch in range(epochs): train_loss, train_acc train_one_epoch(...) val_loss, val_acc evaluate(...) scheduler.step() if val_acc best_val_acc: best_val_acc val_acc torch.save({ epoch: epoch, model_state_dict: model.state_dict(), optimizer_state_dict: optimizer.state_dict(), val_acc: val_acc, }, f{output_dir}/checkpoints/best.pth) counter 0 else: counter 1 if counter patience: print(fEarly stopping at epoch {epoch}) break保存的时候把optimizer的状态也存进去这样如果要从checkpoint继续训练优化器的动量等信息不会丢。只存model.state_dict()是不够的。5. 实验追踪与复现让每次实验都有据可查5.1 为什么需要实验追踪做AI研究或开发你一定会跑很多组实验换个学习率、改个模型结构、加个数据增强。如果没有记录过几天你根本记不清哪组配置对应哪个结果。实验追踪要记录的东西包括配置参数、代码版本、训练指标曲线、模型权重、环境信息。最轻量的做法是每次实验建一个目录把配置和日志写进去。进阶一点可以用TensorBoard或Weights Biases这类工具。我建议从轻量做法开始因为依赖少、可控性强。import json import os from datetime import datetime def create_experiment_dir(base_dir, config): timestamp datetime.now().strftime(%Y%m%d_%H%M%S) exp_dir os.path.join(base_dir, fexp_{timestamp}) os.makedirs(os.path.join(exp_dir, checkpoints), exist_okTrue) os.makedirs(os.path.join(exp_dir, logs), exist_okTrue) with open(os.path.join(exp_dir, config.json), w) as f: json.dump(config, f, indent2) return exp_dir每次实验的配置存成JSON训练过程中的指标写进CSV或TensorBoard。这样回头看的时候打开目录就知道当时用了什么参数、结果如何。5.2 随机种子复现的命门AI实验的复现性是个大问题。即使代码一样随机种子不同结果也可能差几个百分点。要复现必须固定所有随机源Python的random、numpy的random、PyTorch的random、CUDA的随机。import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark Falsecudnn.deterministic True会让cuDNN使用确定性算法代价是可能慢一点。benchmark False也是为了保证确定性。如果你追求极致速度可以设benchmark True但结果就不可复现了。这是个取舍看你的场景。注意即使固定了所有种子不同GPU型号、不同驱动版本、不同框架版本之间结果仍可能有微小差异。完全跨平台复现是很难的但同一台机器上复现是能做到的。5.3 日志记录别只用printprint调试可以但工程化的项目需要结构化日志。Python的logging模块能把日志同时输出到控制台和文件还能设置级别。import logging def setup_logger(log_file): logger logging.getLogger(train) logger.setLevel(logging.INFO) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) file_handler logging.FileHandler(log_file) file_handler.setFormatter(formatter) logger.addHandler(file_handler) console_handler logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) return logger logger setup_logger(f{exp_dir}/logs/train.log) logger.info(fStart training with lr{config[train][lr]})日志文件在排查问题时特别有用。训练崩了翻日志能看到崩之前最后几步发生了什么比盯着终端回滚强得多。6. 推理与部署模型训练完只是开始6.1 从checkpoint到可调用服务训练完的模型不能直接扔给业务用。你需要一个推理脚本加载模型权重接收输入返回预测结果。这个脚本要处理的事情包括模型加载、预处理、前向传播、后处理、批处理。import torch from PIL import Image class Predictor: def __init__(self, checkpoint_path, devicecuda): self.device torch.device(device) self.model build_model(num_classes10) checkpoint torch.load(checkpoint_path, map_locationself.device) self.model.load_state_dict(checkpoint[model_state_dict]) self.model.to(self.device) self.model.eval() self.transform val_transform torch.no_grad() def predict(self, image_path): image Image.open(image_path).convert(RGB) tensor self.transform(image).unsqueeze(0).to(self.device) output self.model(tensor) prob torch.softmax(output, dim1) pred prob.argmax(dim1).item() confidence prob.max().item() return pred, confidenceunsqueeze(0)是为了加一个batch维度因为模型期望输入是[batch, channel, height, width]。torch.no_grad()在推理时一定要加不然会构建计算图浪费显存。6.2 批处理与延迟优化单张推理在生产环境往往不够用需要支持批处理。批处理能显著提高吞吐量因为GPU擅长并行计算。但批处理会引入延迟因为要等一批数据凑齐。这里有个权衡批越大吞吐越高但单条延迟越大。torch.no_grad() def predict_batch(self, image_paths, batch_size32): results [] for i in range(0, len(image_paths), batch_size): batch_paths image_paths[i:ibatch_size] tensors torch.stack([ self.transform(Image.open(p).convert(RGB)) for p in batch_paths ]).to(self.device) outputs self.model(tensors) probs torch.softmax(outputs, dim1) preds probs.argmax(dim1).cpu().tolist() confs probs.max(dim1).values.cpu().tolist() results.extend(zip(preds, confs)) return results如果延迟要求高可以用动态批处理设置一个最大等待时间超时或凑够一批就推理。这个逻辑稍微复杂但能兼顾吞吐和延迟。6.3 模型导出脱离训练代码运行生产环境往往不希望装整个训练框架。PyTorch支持导出成TorchScript或ONNX这样推理时只需要轻量运行时。# 导出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)TorchScript的好处是序列化了模型结构加载时不需要原始Python类。ONNX则更通用能跨框架部署。导出时要注意模型里不能有依赖Python控制流的逻辑否则trace会失败需要用torch.jit.script。7. 那些只有踩过才知道的坑7.1 显存泄漏训练越跑越慢的元凶训练循环里如果累积了计算图显存会一直涨最后OOM。常见原因是在验证时忘了加torch.no_grad()或者在循环里把loss存进了列表但没detach。# 错误做法losses.append(loss) 会保留计算图 # 正确做法 losses.append(loss.item()) # 只存数值 # 或者 losses.append(loss.detach().cpu()) # detach后存tensorloss.item()会把tensor转成Python float不保留计算图。loss.detach()返回一个不参与梯度计算的tensor。两种都行看后续怎么用。7.2 数据加载成为瓶颈GPU在等CPU如果nvidia-smi显示GPU利用率忽高忽低经常掉到0那很可能是数据加载拖了后腿。解决办法增大num_workers、把数据预处理提前做好、用更快的存储。# 检查数据加载时间 import time start time.time() for images, labels in train_loader: pass print(fData loading time: {time.time() - start:.2f}s)如果加载时间比训练时间还长那就得优化数据管道了。把图片预先resize好存成更小的格式或者用LMDB、HDF5这类二进制格式能大幅加速。7.3 过拟合训练集99%验证集60%过拟合是AI工程里最常见的病。症状是训练指标一路涨验证指标涨到一定程度就掉头向下。药方有几个加数据增强、加Dropout、加权重衰减、减小模型、早停。# 在模型里加Dropout class MyModel(nn.Module): def __init__(self, num_classes): super().__init__() self.backbone ... self.dropout nn.Dropout(0.5) self.fc nn.Linear(512, num_classes) def forward(self, x): x self.backbone(x) x self.dropout(x) return self.fc(x)Dropout率一般设0.3到0.5。太高会欠拟合太低没效果。权重衰减用AdamW的话weight_decay1e-4或1e-2都常见看模型大小。7.4 类别不平衡少数类被模型无视分类任务里如果某个类别样本特别少模型会倾向于预测多数类因为这样总体准确率高。解决办法重采样、加权损失、focal loss。# 加权损失 class_counts [1000, 100, 50] weights 1.0 / torch.tensor(class_counts, dtypetorch.float) weights weights / weights.sum() criterion nn.CrossEntropyLoss(weightweights.to(device))权重的计算方式是样本数的倒数再归一化。这样少数类的损失权重更大模型会更关注它们。8. 从能跑到好用AI工程能力的进阶方向把上面这些环节都跑通你已经比大多数“只会调包”的人强了。但AI工程这条路还长后面可以往几个方向深入。第一个方向是自动化。把训练、评估、导出串成一条流水线用Makefile或DVC管理。每次改完代码一条命令跑完整个流程减少手动操作出错。第二个方向是超参数搜索。手动调参效率低可以用Optuna或Ray Tune做自动搜索。定义好搜索空间让程序自己找最优组合。import optuna def objective(trial): lr trial.suggest_float(lr, 1e-5, 1e-2, logTrue) batch_size trial.suggest_categorical(batch_size, [16, 32, 64]) # 训练并返回验证集准确率 return val_acc study optuna.create_study(directionmaximize) study.optimize(objective, n_trials50)第三个方向是模型压缩。生产环境对延迟和体积有要求时可以量化、剪枝、蒸馏。PyTorch支持动态量化和静态量化能把模型体积压到四分之一推理速度提升两三倍。# 动态量化 quantized_model torch.quantization.quantize_dynamic( model, {nn.Linear}, dtypetorch.qint8 )第四个方向是监控与迭代。模型上线不是终点要监控推理延迟、输入分布漂移、预测置信度分布。发现异常时能快速回滚或重新训练。我自己走下来最大的体会是AI工程能力不是靠看教程看出来的是靠一个个项目磨出来的。每踩一个坑就补上一块知识。从写死路径到配置文件从手动保存到自动checkpoint从print调试到结构化日志每一步都是被现实逼出来的。你不需要一开始就做到完美但要有意识地把每个环节做得比上次好一点。时间长了这套工程习惯就会变成你的肌肉记忆再面对新项目时就能快速搭起一条靠谱的流水线。