1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经掌握了。我刚开始也是这么想的直到有一次线上推理服务在凌晨两点崩了日志里全是显存溢出的报错而我对着监控面板完全不知道从哪下手——因为底层的东西全是黑盒我连模型是怎么加载进显存的都说不清楚。那次事故之后我开始系统性地从零搭建AI工程链路把每一个环节都拆开来看这才真正理解了什么叫“工程”。ai-engineering-from-scratch这个标题核心不是教你调包而是教你理解AI系统从数据到推理的完整生命周期。它适合那些已经会用PyTorch或TensorFlow跑通模型但一遇到部署、优化、监控就抓瞎的开发者也适合想转行AI工程、但被各种框架抽象层绕晕的初学者。这篇文章我会把从零搭建AI工程的核心环节拆成几个模块每个模块都讲清楚“为什么这么做”和“不这么做会怎样”并且给出可以直接复现的代码和配置。先说一个反直觉的结论从零搭建AI工程最难的不是模型本身而是数据管道和推理服务的稳定性。模型架构在论文里写得清清楚楚但数据怎么清洗、怎么版本化、怎么在训练和推理之间保持一致这些才是真正吃经验的地方。我见过太多团队在模型上反复调参结果上线后发现训练时的特征处理和推理时的特征处理不一致导致线上效果直接掉十几个点。这种问题调包是发现不了的只有从零搭建过完整链路的人才会在早期就设计好特征一致性校验。所以这篇文章不会给你一个“万能框架”而是带你走一遍我实际踩过坑的完整流程。从环境隔离、数据版本控制、训练脚本的工程化改造到模型导出、推理服务封装、性能压测和监控告警每个环节我都会给出具体的工具选型和操作步骤并且解释为什么在这个场景下选A不选B。你不需要全部照搬但理解这些决策背后的逻辑比记住几个命令重要得多。2. 环境隔离与依赖管理别让“在我机器上能跑”成为常态2.1 为什么conda和pip混用迟早出事我刚开始做AI项目的时候习惯用conda建环境然后用pip装一些conda里没有的包。前几个月一切正常直到有一次在服务器上复现实验发现同样的requirements.txt装出来的环境本地能跑服务器上就报CUDA版本不匹配。排查了一整天最后发现是conda在安装PyTorch时自动带了一个cudatoolkit而pip又装了一个不同版本的nvidia-cudnn两者冲突了。这个坑的本质是conda和pip的依赖解析器是独立的它们互相不知道对方装了什么。conda管的是二进制包和系统级依赖pip管的是Python包当两者同时操作同一个包的不同版本时就会出现“依赖地狱”。我的建议是在一个项目里只选一种包管理工具。如果你需要管理CUDA、cuDNN这种系统级依赖就用conda从头管到尾如果只是纯Python包用pip加venv就够了。具体操作上我现在每个AI项目都会建一个独立的conda环境并且把环境导出为environment.yml而不是requirements.txt。因为environment.yml会记录conda渠道和pip安装的包复现性更好。导出命令是conda env export --no-builds environment.yml--no-builds参数很重要它去掉了具体的build号避免因为build号不同导致跨平台复现失败。然后在另一台机器上复现时conda env create -f environment.yml注意如果项目里用了pip安装的包environment.yml里会有一个pip:小节复现时conda会自动调用pip安装这些包。但前提是目标机器上conda环境已经激活否则pip会装到系统Python里。2.2 用Docker做最终的环境交付conda环境解决了开发阶段的依赖问题但到了部署阶段我强烈建议用Docker。原因很简单conda环境依赖宿主机器的CUDA驱动版本而Docker可以锁定整个运行时环境。我遇到过好几次开发机是CUDA 11.8服务器是CUDA 12.1conda环境导过去后PyTorch找不到对应的cudatoolkit直接报错。用Docker的话基础镜像里已经包含了匹配的CUDA运行时只要宿主机的驱动版本不低于镜像要求就能跑。我的Dockerfile通常长这样FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04 RUN apt-get update apt-get install -y python3.10 python3-pip COPY environment.yml /tmp/environment.yml RUN conda env create -f /tmp/environment.yml SHELL [conda, run, -n, myenv, /bin/bash, -c] COPY . /app WORKDIR /app CMD [python, serve.py]这里有个细节nvidia/cuda镜像的tag要选runtime而不是devel因为devel镜像包含编译工具体积大很多而推理服务不需要编译CUDA代码。另外conda env create在Docker build阶段执行这样镜像构建完成后环境就固化了启动容器时不需要再装依赖冷启动时间从几分钟降到几秒。2.3 依赖锁定的实操心得还有一个容易被忽略的点PyTorch的版本和CUDA版本必须严格对应。我见过有人用pip install torch不带版本号结果装了一个CPU版本的PyTorch训练时发现用不了GPU还以为是驱动问题。正确的做法是去PyTorch官网查版本对应表然后写死版本号。比如pip install torch2.1.0cu118 torchvision0.16.0cu118 -f https://download.pytorch.org/whl/torch_stable.html这个cu118后缀明确指定了CUDA 11.8版本不带这个后缀默认装CPU版。这个坑我踩过两次后来养成了习惯任何AI项目的依赖文件里PyTorch相关包必须带CUDA后缀和完整版本号。3. 数据管道的工程化从原始文件到可训练数据集3.1 数据版本控制为什么不能用GitAI工程和传统软件工程最大的区别之一就是数据。代码可以用Git管理但数据集动辄几个GB放Git里直接把仓库撑爆。我早期试过用Git LFS但LFS的免费额度很快用完而且每次拉取大文件都很慢。后来我转向了DVCData Version Control它把大文件存在远程存储比如S3或本地NASGit里只存一个指向文件的元数据指针。DVC的基本用法很简单dvc init dvc add data/raw/train.csv git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m add training data dvc push这样train.csv本身不会被提交到Git只有train.csv.dvc这个元数据文件被提交。别人克隆仓库后执行dvc pull就能从远程存储拉取数据。DVC的好处是你可以像切换代码分支一样切换数据版本git checkout到某个commit再dvc checkout数据就回到那个版本了。提示DVC的远程存储配置在.dvc/config里建议用环境变量存访问密钥不要硬编码在配置文件里。我一般用dvc remote add -d myremote s3://mybucket/dvcstore然后通过环境变量AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY传密钥。3.2 数据清洗的流水线设计数据清洗最忌讳的是“一次性脚本”——写一个clean.py跑完输出一个clean.csv然后训练脚本直接读这个clean.csv。这种做法的问题是清洗逻辑和训练逻辑耦合一旦清洗规则变了你无法追溯之前训练用的到底是哪个版本的数据。我的做法是把清洗拆成多个步骤每个步骤输出一个中间文件并且用DVC管理每个中间文件。比如一个典型的文本分类任务清洗流水线可能是原始数据去重dedup.py文本归一化比如全角转半角、去除特殊字符normalize.py标签清洗去除标注不一致的样本label_clean.py划分训练集、验证集、测试集split.py每个脚本都是独立的输入输出都是文件路径通过命令行参数传递。这样我可以单独重跑某一步而不需要从头开始。而且每个中间文件都被DVC跟踪任何时候都能复现出完全一样的数据集。3.3 训练和推理的特征一致性校验这是我最想强调的一点训练时用的特征处理逻辑必须和推理时完全一致。我见过一个推荐系统项目训练时用pandas做特征归一化推理时用numpy手写了一个归一化函数结果因为pandas的mean()默认跳过NaN而numpy的mean()不跳过导致推理时特征分布偏移线上CTR直接掉了8%。解决这个问题的办法是把特征处理逻辑封装成一个独立的模块训练和推理都调用同一个模块。比如# feature_processor.py class FeatureProcessor: def __init__(self, mean, std): self.mean mean self.std std def transform(self, x): return (x - self.mean) / (self.std 1e-8) def save(self, path): with open(path, wb) as f: pickle.dump({mean: self.mean, std: self.std}, f) classmethod def load(cls, path): with open(path, rb) as f: params pickle.load(f) return cls(params[mean], params[std])训练时用训练集算出mean和std保存到文件推理时加载这个文件用同样的transform方法。这样就能保证一致性。另外我还会在训练脚本里加一个校验步骤用保存的FeatureProcessor重新处理一遍训练集和训练时的特征对比确保没有偏差。4. 训练脚本的工程化改造从Notebook到可复现实验4.1 为什么Notebook不适合做正式训练Jupyter Notebook适合探索性分析但绝对不适合做正式的训练脚本。原因有三个第一Notebook的执行顺序是线性的但你可以跳着执行单元格导致状态混乱第二Notebook的版本控制很痛苦Git diff出来是一堆JSON根本看不出改了什么第三Notebook很难做参数化你没法用命令行传超参数。我的做法是探索阶段用Notebook一旦确定了模型结构和训练流程立刻重写成Python脚本。脚本的结构一般是# train.py import argparse from model import MyModel from data import load_data from trainer import Trainer def main(): parser argparse.ArgumentParser() parser.add_argument(--lr, typefloat, default1e-3) parser.add_argument(--batch_size, typeint, default32) parser.add_argument(--epochs, typeint, default10) parser.add_argument(--data_dir, typestr, requiredTrue) parser.add_argument(--output_dir, typestr, requiredTrue) args parser.parse_args() train_data, val_data load_data(args.data_dir) model MyModel() trainer Trainer(model, args) trainer.train(train_data, val_data) if __name__ __main__: main()这样我可以用python train.py --lr 1e-4 --batch_size 64来跑不同的实验每次实验的输出目录不同互不干扰。4.2 实验跟踪别再用Excel记结果了我早期做实验的时候用Excel表格记录每次的超参数和指标结果实验一多就乱了经常分不清哪个模型对应哪个结果。后来我用了MLflow它可以在训练脚本里自动记录超参数、指标和模型文件并且提供一个Web界面来对比不同实验。MLflow的集成很简单import mlflow mlflow.set_experiment(my_experiment) with mlflow.start_run(): mlflow.log_params(vars(args)) for epoch in range(args.epochs): train_loss train_one_epoch() val_loss, val_acc validate() mlflow.log_metrics({train_loss: train_loss, val_loss: val_loss, val_acc: val_acc}, stepepoch) mlflow.pytorch.log_model(model, model)跑完实验后打开mlflow ui就能看到所有实验的对比表格还能画图看指标随epoch的变化。这个工具帮我省了大量整理实验结果的时间而且团队协作时大家都能看到彼此的实验记录避免重复劳动。4.3 检查点保存与恢复训练训练大模型时中途断电或者显存溢出是常有的事。如果没有检查点机制几个小时的训练就白费了。我的做法是每个epoch结束后保存一次检查点包括模型参数、优化器状态、学习率调度器状态和当前epoch数。恢复训练时加载这些状态从断点继续。def save_checkpoint(model, optimizer, scheduler, epoch, path): torch.save({ epoch: epoch, model_state_dict: model.state_dict(), optimizer_state_dict: optimizer.state_dict(), scheduler_state_dict: scheduler.state_dict(), }, path) def load_checkpoint(model, optimizer, scheduler, path): checkpoint torch.load(path) model.load_state_dict(checkpoint[model_state_dict]) optimizer.load_state_dict(checkpoint[optimizer_state_dict]) scheduler.load_state_dict(checkpoint[scheduler_state_dict]) return checkpoint[epoch]注意保存检查点时如果用了混合精度训练AMP还需要保存scaler.state_dict()否则恢复训练时梯度缩放会不一致导致训练不稳定。这个坑我在一个图像分类项目里踩过恢复训练后loss突然飙升排查了半天才发现是scaler状态没恢复。5. 模型导出与推理服务封装从checkpoint到线上API5.1 TorchScript还是ONNX导出格式的选择训练完模型后下一步是导出成推理引擎能加载的格式。PyTorch原生支持两种导出方式TorchScript和ONNX。TorchScript是PyTorch自己的序列化格式导出后可以用C加载不依赖PythonONNX是跨框架的开放格式可以用ONNX Runtime或TensorRT加速。我的选择逻辑是如果推理服务用Python写直接用TorchScript就够了如果需要用C部署或者要用TensorRT做极致加速就导出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.pt)但torch.jit.trace有个坑它只记录实际执行的操作如果模型里有if分支trace只会记录当前输入走的那条分支。对于有动态控制流的模型要用torch.jit.script而不是trace。我一般先用trace试如果推理结果和Python端不一致再换script。ONNX导出稍微麻烦一点需要指定输入输出的名字和动态维度torch.onnx.export( model, example_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version13 )dynamic_axes参数很重要它允许推理时batch size可变。如果不设置导出的ONNX模型会固定batch size为1推理时传多个样本就会报错。5.2 用FastAPI封装推理接口推理服务的封装我推荐用FastAPI因为它异步性能好自动生成API文档而且和PyTorch的集成很简单。一个基本的推理服务长这样from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model torch.jit.load(model.pt) model.eval() class Request(BaseModel): text: str class Response(BaseModel): label: str score: float app.post(/predict, response_modelResponse) async def predict(req: Request): inputs tokenizer(req.text, return_tensorspt) with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, pred torch.max(probs, dim-1) return Response(labelstr(pred.item()), scorescore.item())启动命令是uvicorn serve:app --host 0.0.0.0 --port 8000 --workers 4。--workers参数指定了工作进程数一般设置为CPU核心数这样可以并行处理多个请求。注意如果模型加载在GPU上多个worker会各自加载一份模型到GPU显存占用会翻倍。这种情况下要么减少worker数要么用共享内存的方式让多个worker共用一份模型。我一般用--workers 1然后在服务内部用异步批处理来提升吞吐。5.3 批处理与动态填充的工程细节推理服务如果一次只处理一个请求GPU利用率会很低。我的做法是在服务内部做一个微批处理收集一小段时间内的请求凑成一个batch再送进模型。这样可以显著提升吞吐但会增加单次请求的延迟。需要根据业务场景权衡batch size和等待时间。另外文本模型推理时同一个batch里的样本长度不同需要padding到相同长度。如果padding到全局最大长度短样本会浪费大量计算。更好的做法是动态padding即padding到当前batch内的最大长度。HuggingFace的tokenizer支持这个功能inputs tokenizer(texts, paddingTrue, truncationTrue, return_tensorspt)paddingTrue就是动态padding它会padding到batch内最长样本的长度。这个细节在离线评估时影响不大但在线推理时能省不少显存和计算时间。6. 性能压测与监控上线前的最后一道防线6.1 用Locust做推理服务的压测服务写好了别急着上线先压测。我用Locust来模拟并发请求它可以定义用户行为然后逐步增加并发数观察响应时间和错误率的变化。一个简单的压测脚本from locust import HttpUser, task, between class ModelUser(HttpUser): wait_time between(0.1, 0.5) task def predict(self): self.client.post(/predict, json{text: 这是一个测试样本})启动Locust后在Web界面设置并发用户数和每秒新增用户数就能看到实时的QPS和延迟曲线。我一般会找到服务开始出现超时或错误率上升的并发点那个点就是当前配置下的最大承载能力。压测时要注意压测客户端本身不能成为瓶颈。如果Locust跑在同一台机器上它消耗的CPU和内存会影响服务性能。最好用单独的机器跑Locust或者至少用多台机器分布式压测。6.2 监控指标延迟、吞吐、显存、错误率上线后监控是必不可少的。我关注的指标有四类指标类型具体指标告警阈值说明延迟P50/P95/P99延迟P99 500msP99比平均延迟更能反映用户体验吞吐QPS低于基线20%突然下降可能意味着服务异常显存GPU显存使用率 90%接近上限时容易OOM错误率HTTP 5xx比例 1%任何5xx都需要立即排查这些指标可以用Prometheus采集Grafana展示。FastAPI可以通过prometheus-fastapi-instrumentator自动暴露指标from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(app).expose(app)GPU显存指标需要额外安装nvidia-dcgm-exporter它会把GPU指标暴露成Prometheus格式。6.3 我踩过的监控坑指标有了但没人看监控搭好了但如果没有告警指标就是摆设。我早期犯过一个错误Grafana面板做得很漂亮但没人盯着看结果服务挂了半小时才被发现。后来我加了告警规则用Alertmanager把告警发到团队群里。告警规则要设置合理的阈值和持续时间避免误报。比如“P99延迟超过500ms持续2分钟”才告警而不是一超过就报否则网络抖动会导致大量误报。另一个坑是告警疲劳。如果告警太频繁大家就会麻木真正的问题反而被忽略。我的做法是把告警分级P0是服务不可用立即打电话P1是性能下降发群消息P2是资源使用率偏高记录到日报里。这样团队对告警的响应才有优先级。7. 从零搭建AI工程的个人体会走完这一整套流程我最大的感受是AI工程的难点不在AI而在工程。模型架构可以看论文超参数可以调但数据管道、环境隔离、推理服务、监控告警这些“脏活累活”才是决定一个AI系统能不能稳定跑起来的关键。我见过太多团队在模型上投入80%的精力结果上线后因为数据不一致、显存泄漏、服务超时这些问题疲于奔命。如果你正在从零搭建AI工程我的建议是先把数据管道和推理服务的基础设施搭好再回头优化模型。一个准确率85%但稳定运行的服务比一个准确率90%但三天两头崩的服务有价值得多。另外不要追求一步到位我现在的这套流程也是经过多次迭代才稳定下来的。每次踩坑后把解决方案固化到脚本或配置里慢慢就形成了一套适合自己的工程体系。最后分享一个我最近在用的技巧用Makefile管理整个训练和部署流程。把数据清洗、训练、导出、压测这些命令写成Makefile的target这样一条make all就能跑完整个流水线而且每个步骤的依赖关系一目了然。对于团队协作来说这比写一堆README文档管用得多。