从零搭建AI工程能力这件事,我前前后后折腾过好几轮。最早的时候我也走过弯路——买了一堆讲Transformer推导的书,把注意力机制的公式背得滚瓜烂熟,结果真到要上线一个文本分类服务的时候,连推理延迟怎么压、显存怎么省、模型版本怎么管都一头雾水。后来我才想明白一件事:AI工程不是机器学习理论的延伸,它是一门独立的工程学科,核心矛盾在于“模型的不确定性与工程系统的确定性要求之间的冲突”。这篇内容就是把我这些年从零构建AI工程能力体系的过程拆开讲,从底层原理到落地实操,适合那些已经会调API、但想真正搞懂“一个AI系统是怎么被工程化地搭起来”的开发者。
1. 先搞清楚AI工程到底在工程什么
1.1 它和传统软件工程的根本分歧在哪
传统软件工程里,输入是确定的,逻辑是确定的,输出也是确定的。你写一个订单金额计算函数,给定输入,输出永远一致。但AI系统不是这样——同一个输入,模型可能给出不同的输出(尤其是生成式模型),而且你很难用单元测试去断言“这个输出是对的”。
这就引出了AI工程最核心的几个工程问题:
- 概率性输出的可测试性:你没法写
assert output == expected,只能写“输出在某个可接受范围内”或者“输出不包含某些违规内容”。 - 数据依赖的隐蔽性:传统软件的bug在代码里,AI系统的bug可能在训练数据里、在特征管道里、在推理时的预处理逻辑里。
- 性能瓶颈的转移:传统后端瓶颈通常在IO或数据库,AI系统的瓶颈可能在GPU显存、在批处理策略、在tokenizer的效率上。
我踩过最典型的一个坑:早期做一个情感分析服务,离线评估准确率92%,上线后用户投诉不断。排查了两天才发现,训练时的文本预处理用了jieba分词,而线上服务为了图快直接按空格切词,中文根本没空格,等于把整句话当成一个token喂进去了。这个bug不在模型里,不在代码逻辑里,在“训练和推理的预处理不一致”这个工程缝隙里。
所以AI工程的第一课,不是学模型,是学“怎么让一个概率系统在确定性的工程约束下稳定运行”。
1.2 从零构建需要覆盖的五个能力层
我把AI工程能力拆成五层,从下往上依次是:
| 层级 | 能力域 | 典型工作内容 | 常见工具 |
|---|---|---|---|
| L1 | 环境与依赖管理 | GPU驱动、CUDA版本、Python环境隔离 | conda、docker、poetry |
| L2 | 数据处理管道 | 数据清洗、特征工程、数据集版本管理 | pandas、DVC、Great Expectations |
| L3 | 模型训练与微调 | 训练循环、超参管理、分布式训练 | PyTorch、Lightning、wandb |
| L4 | 推理服务化 | 模型导出、服务框架、批处理、量化 | ONNX、Triton、vLLM |
| L5 | 运维与监控 | 日志、指标、漂移检测、A/B测试 | Prometheus、Grafana、Evidently |
很多人一上来就扎进L3,觉得训模型才是“AI”,结果L1的环境问题就能卡一周,L4的服务化更是完全没概念。我的建议是从L1和L4两头往中间做——先把环境搞稳、把推理服务跑通,再回头补训练和数据的细节。因为推理服务是最终交付物,先看到交付物的样子,你才知道中间每一步是为了什么。
1.3 一个最小可用的AI工程骨架长什么样
在展开每一层之前,先给你一个我常用的最小骨架,让你有个全局感:
project/ ├── configs/ # 所有配置,不硬编码 │ ├── train.yaml │ └── serve.yaml ├── data/ # 数据版本管理 │ ├── raw/ │ └── processed/ ├── src/ │ ├── data/ # 数据处理 │ ├── model/ # 模型定义 │ ├── train/ # 训练逻辑 │ └── serve/ # 推理服务 ├── tests/ # 测试,包括数据测试 ├── docker/ # 环境固化 └── Makefile # 一键复现这个骨架的关键设计原则是:配置与代码分离、数据与代码分离、训练与推理共享同一套预处理代码。最后一条尤其重要,前面那个分词不一致的坑,根源就是训练和推理各写了一套预处理。
2. 环境与依赖:为什么你的CUDA总是装不对
2.1 CUDA版本地狱的本质原因
新手最容易被CUDA折磨。你装了个PyTorch,提示CUDA版本不匹配;换了个版本,又和驱动不兼容。这个问题的本质是三层版本必须严格对齐:
- GPU驱动版本:决定了你最高能支持的CUDA版本(向下兼容)
- CUDA Toolkit版本:PyTorch等框架编译时依赖的版本
- 框架版本:PyTorch/TensorFlow各自绑定了特定的CUDA版本
它们的关系是:驱动版本 ≥ 框架要求的CUDA版本。比如你的驱动支持到CUDA 12.4,那你可以跑CUDA 11.8编译的PyTorch,但不能跑CUDA 12.6编译的。
我常用的排查命令:
# 查看驱动支持的CUDA版本 nvidia-smi # 查看当前CUDA Toolkit版本 nvcc --version # 查看PyTorch实际使用的CUDA版本 python -c "import torch; print(torch.version.cuda)"注意:
nvidia-smi显示的CUDA版本是驱动支持的最高版本,不是你实际安装的版本。很多人看到这个数字就以为装对了,其实不是。
2.2 用容器把环境问题一次性解决
我现在的做法是本地只装驱动,所有Python环境和CUDA都跑在容器里。这样换机器、换项目都不会互相污染。一个典型的Dockerfile片段:
FROM nvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04 RUN apt-get update && apt-get install -y python3.10 python3-pip RUN pip install torch==2.2.0 --index-url https://download.pytorch.org/whl/cu121 WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt这里有个经验:基础镜像选runtime而不是devel,除非你需要编译CUDA算子。runtime镜像小很多,部署时拉取快。另外cudnn版本也要和CUDA对齐,12.1.0-cudnn8这种tag就是官方帮你配好的组合。
2.3 依赖锁定:别让“昨天还能跑”变成玄学
Python依赖的坑在于,pip install默认装最新版,今天能跑的代码明天可能就崩了。我的做法是:
- 用
poetry或pip-tools生成锁定文件(poetry.lock/requirements.txt带精确版本号) - 区分
requirements.in(宽松约束)和requirements.txt(精确锁定) - CI里用锁定文件安装,保证本地和线上一致
# 用pip-tools的流程 pip-compile requirements.in -o requirements.txt # 生成锁定 pip-sync requirements.txt # 按锁定安装这套流程看起来麻烦,但能省掉无数次“在我机器上是好的”的扯皮。
3. 数据处理管道:AI系统里最脏最累的活
3.1 为什么数据管道值得单独工程化
我见过太多项目把数据处理写成一个个散落的脚本:clean.py、split.py、augment.py,跑的时候靠记忆按顺序执行。这种做法的致命问题是不可复现——三个月后你想重新训练,根本记不清当时的数据是怎么处理的。
数据管道工程化的核心目标是:给定原始数据和一份配置,能一键复现出完全相同的训练集。这需要三个东西:
- 数据版本管理:原始数据、处理后数据都要有版本标识
- 处理逻辑代码化:每一步转换都是可测试的函数
- 数据校验:处理前后自动检查数据质量
3.2 用DVC做数据版本管理
Git管代码,DVC管数据。基本用法:
dvc init dvc add data/raw/dataset.csv # 生成dataset.csv.dvc指针文件 git add data/raw/dataset.csv.dvc data/raw/.gitignore git commit -m "add raw dataset v1"DVC会把大文件存到远程存储(S3、本地NAS都行),Git里只留一个指针文件。这样你切换Git分支时,dvc checkout就能拿到对应版本的数据。
我踩过的坑:DVC的缓存目录默认在项目内,容易把磁盘撑爆。建议一开始就配置外部缓存:
dvc cache dir /mnt/large-disk/dvc-cache3.3 数据校验:在训练前拦住脏数据
数据问题如果等到训练完才发现,浪费的是几小时甚至几天的GPU时间。我习惯在管道里加一层校验,用pydantic或Great Expectations都行。一个轻量做法是用pydantic定义数据schema:
from pydantic import BaseModel, validator class TrainingSample(BaseModel): text: str label: int @validator('text') def text_not_empty(cls, v): if not v.strip(): raise ValueError('text不能为空') return v @validator('label') def label_in_range(cls, v): if v not in (0, 1): raise ValueError('label必须是0或1') return v然后在数据加载时逐条校验,把不合规的样本记录下来而不是直接丢弃——丢弃会让你丢失数据分布的信息,记录下来才能分析是采集问题还是标注问题。
3.4 训练/推理预处理必须共享代码
回到开头那个分词坑。解决方案是把预处理逻辑抽成一个独立模块,训练和推理都import它:
# src/data/preprocess.py def preprocess_text(text: str) -> str: """训练和推理共用的预处理""" text = text.strip().lower() tokens = jieba.lcut(text) return ' '.join(tokens)训练脚本和推理服务都调用这个函数。任何预处理逻辑的修改,都必须同时影响训练和推理,否则就会出现训练-推理偏差(training-serving skew)。这是AI工程里最隐蔽也最致命的一类bug。
4. 训练工程化:让实验可复现、可比较
4.1 配置管理:告别硬编码超参
新手写训练脚本,超参直接写在代码里:
lr = 1e-4 batch_size = 32 epochs = 10改一次跑一次,改多了自己都记不清哪个配置对应哪个结果。正确做法是用配置文件:
# configs/train.yaml model: name: bert-base-chinese num_labels: 2 training: lr: 1e-4 batch_size: 32 epochs: 10 warmup_ratio: 0.1 data: train_path: data/processed/train.csv max_length: 128用hydra或omegaconf加载,命令行还能覆盖:
python train.py training.lr=5e-5 training.batch_size=64这样每次实验的完整配置都能被记录下来,配合wandb或mlflow,实验对比一目了然。
4.2 实验追踪:别靠脑子记结果
我早期用Excel记实验结果,记到第20次就乱了。后来换成wandb,每次训练自动记录loss曲线、指标、配置、甚至GPU利用率。关键代码就几行:
import wandb wandb.init(project="sentiment", config=cfg) for epoch in range(epochs): train_loss = train_one_epoch() val_f1 = evaluate() wandb.log({"train_loss": train_loss, "val_f1": val_f1})实验追踪的价值不在于记录,而在于对比。当你想知道“为什么这次比上次差”时,能直接diff两次实验的配置差异,而不是靠回忆。
4.3 检查点策略:什么时候存、存什么
模型检查点不是越多越好,占磁盘还容易搞混。我的策略是:
- 按验证指标存最优:只保留验证集上表现最好的N个
- 定期存:每N个epoch存一次,防止训练崩溃丢失进度
- 存完整状态:不只是模型权重,还要存优化器状态、学习率调度器状态、当前epoch,这样才能断点续训
torch.save({ 'epoch': epoch, 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'scheduler_state_dict': scheduler.state_dict(), 'best_metric': best_metric, }, f'checkpoints/epoch_{epoch}.pt')提示:断点续训时,优化器状态和调度器状态必须一起恢复,否则学习率曲线会错乱,训练效果可能明显变差。
4.4 分布式训练:什么时候需要、怎么上手
单卡能跑就别上分布式,分布式调试成本高。判断标准很简单:单卡跑一个epoch的时间 × 总epoch数 > 你能接受的等待时间,就该考虑分布式了。
PyTorch的DDP(DistributedDataParallel)是目前最成熟的选择。核心改动:
import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP dist.init_process_group("nccl") model = model.to(local_rank) model = DDP(model, device_ids=[local_rank])配合torchrun启动:
torchrun --nproc_per_node=4 train.py踩过的坑:DDP下每个进程的随机种子要不同,否则数据增强会重复。还有,验证和保存检查点只在rank 0上做,否则会写冲突。
5. 推理服务化:从模型文件到可用API
5.1 模型导出:为什么不能直接pickle
训练完的PyTorch模型直接torch.save然后线上torch.load,看起来能用,但有几个问题:
- 依赖训练代码:加载时需要模型类定义,线上服务得把训练代码也带上
- 性能差:PyTorch的eager模式推理有额外开销
- 跨框架难:想换推理引擎(如TensorRT)就卡住了
更好的做法是导出成中间格式。ONNX是通用选择:
torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input_ids", "attention_mask"], output_names=["logits"], dynamic_axes={"input_ids": {0: "batch", 1: "seq"}}, opset_version=14 )dynamic_axes很关键,不设置的话batch size和序列长度会被固定死,线上只能处理特定形状的输入。
5.2 推理引擎选型:ONNX Runtime vs Triton vs vLLM
不同场景选不同引擎,我整理了一个对比:
| 引擎 | 适用场景 | 优势 | 局限 |
|---|---|---|---|
| ONNX Runtime | 中小模型、CPU/GPU通用 | 轻量、易集成 | 大模型支持一般 |
| Triton Inference Server | 多模型、多框架混合 | 动态批处理、模型编排 | 部署复杂 |
| vLLM | 大语言模型 | PagedAttention、高吞吐 | 只支持LLM类 |
我的经验:判别式模型(分类、NER)用ONNX Runtime就够了,生成式大模型直接上vLLM。Triton适合模型多、需要统一管理的场景,但学习曲线陡。
5.3 动态批处理:吞吐量的关键
线上请求是零散到达的,如果每个请求单独推理,GPU利用率极低。动态批处理把短时间内到达的请求攒成一批一起推理:
# 简化的动态批处理逻辑 class BatchScheduler: def __init__(self, max_batch_size=32, max_wait_ms=10): self.queue = [] self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms async def add_request(self, request): self.queue.append(request) if len(self.queue) >= self.max_batch_size: return await self.flush() await asyncio.sleep(self.max_wait_ms / 1000) return await self.flush()max_wait_ms是延迟和吞吐的权衡点:设大了吞吐高但延迟高,设小了反之。我一般从10ms起步,根据实际P99延迟调整。
5.4 量化:用精度换显存和速度
模型量化把FP32权重压成INT8甚至INT4,显存占用能降一半以上,推理速度也能提升。常见做法:
from onnxruntime.quantization import quantize_dynamic quantize_dynamic( "model.onnx", "model_int8.onnx", weight_type=QuantType.QInt8 )注意:量化会带来精度损失,必须在上线前用验证集评估。我遇到过量化后F1掉3个点的情况,这种就不能接受。一般分类任务掉1个点以内可以接受,生成任务要更谨慎。
6. 监控与运维:上线只是开始
6.1 模型监控和传统监控的区别
传统服务监控看QPS、延迟、错误率就够了。AI服务还要额外看:
- 输入分布漂移:线上输入的特征分布和训练时是否一致
- 输出分布漂移:模型输出的分布是否发生偏移
- 预测置信度:置信度整体下降往往预示模型失效
我见过一个案例:一个推荐模型上线三个月效果逐渐变差,排查发现是用户行为模式变了,但模型没重新训练。如果当时监控了输入特征分布,就能提前发现。
6.2 用Evidently做漂移检测
Evidently是个轻量的漂移检测库,几行代码就能生成报告:
from evidently.report import Report from evidently.metric_preset import DataDriftPreset report = Report(metrics=[DataDriftPreset()]) report.run(reference_data=train_df, current_data=prod_df) report.save_html("drift_report.html")它会告诉你哪些特征发生了漂移、漂移程度如何。我一般设个阈值,漂移超过阈值就触发告警,人工判断是否需要重新训练。
6.3 日志设计:出问题时能查到什么
AI服务的日志要比普通服务更详细。我必记的字段:
- 请求ID(用于追踪单次请求全链路)
- 原始输入(脱敏后)
- 预处理后的输入
- 模型输出和置信度
- 推理耗时(区分预处理、推理、后处理)
- 模型版本号
logger.info({ "request_id": req_id, "model_version": "v1.2.3", "input_len": len(text), "prediction": pred, "confidence": conf, "preprocess_ms": pre_ms, "inference_ms": inf_ms, })模型版本号尤其重要,出问题时第一件事就是确认线上跑的是哪个版本,以及这个版本是什么时候上线的。
6.4 灰度发布与回滚
模型更新不能全量直接上。我的做法是:
- 新模型先接5%流量,观察指标
- 指标正常逐步放量到50%、100%
- 任何指标异常立即回滚到旧版本
这要求服务支持多模型版本共存,通过请求头或配置决定走哪个版本。用Triton的话,它原生支持多版本模型;自己写服务的话,可以维护一个版本路由表。
7. 一些让我少走弯路的实操心得
7.1 先跑通端到端,再优化每一环
新手最容易犯的错是“完美主义”——非要把数据处理做到极致才开始训练,非要把模型调到SOTA才上线。我的建议是先用最粗糙的方式跑通端到端:随便清洗下数据、用现成模型、写个最简单的Flask服务。跑通之后你才知道瓶颈在哪,再针对性优化。我见过太多项目卡在“数据还没准备好”阶段几个月,最后不了了之。
7.2 测试要覆盖数据而不只是代码
传统测试测代码逻辑,AI工程还要测数据。我习惯加这几类测试:
- 数据schema测试:字段类型、取值范围
- 数据分布测试:关键特征的均值、方差在合理范围
- 预处理一致性测试:训练和推理的预处理函数输出一致
- 模型输出测试:给定固定输入,输出在预期范围(不是精确相等)
def test_preprocess_consistency(): text = "这个产品很好用" train_out = preprocess_text(text) serve_out = preprocess_text(text) assert train_out == serve_out7.3 把“重新训练”做成一条命令
模型需要定期重新训练,如果每次都要手动跑一堆脚本,迟早会出错。我的目标是一条命令完成从数据到部署的全流程:
retrain: dvc pull python src/data/process.py python src/train/train.py python src/serve/export.py docker build -t model-service:latest .用Makefile或Airflow都行,关键是流程固化、可重复执行。
7.4 显存不够时的排查顺序
显存OOM是高频问题,我的排查顺序是:
- batch size是不是太大:先减半试试
- 有没有梯度累积的误用:梯度累积不省显存,只省batch维度
- 中间激活有没有及时释放:用
del加torch.cuda.empty_cache() - 是不是用了混合精度:AMP能省不少显存
- 模型本身是不是太大:考虑量化或换小模型
from torch.cuda.amp import autocast, GradScaler scaler = GradScaler() with autocast(): output = model(input) loss = criterion(output, target) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()混合精度通常能省30%-50%显存,而且速度更快,几乎是无脑该开的选项。
7.5 别忽视CPU侧的瓶颈
大家盯着GPU,但很多时候瓶颈在CPU:tokenizer太慢、数据加载是单线程、后处理逻辑复杂。我遇到过一个服务GPU利用率只有20%,排查发现是tokenizer成了瓶颈。解决方案是用fast tokenizer(Rust实现)或者把tokenization放到GPU上做。
from transformers import AutoTokenizer # 用fast版本,快很多 tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese", use_fast=True)7.6 版本管理要覆盖模型、数据、代码三者
一个AI系统的“版本”是三维的:代码版本(Git commit)、数据版本(DVC hash)、模型版本(训练产出的checkpoint)。三者必须能对应起来。我的做法是在模型元数据里记录:
{ "model_version": "v1.2.3", "git_commit": "a1b2c3d", "data_version": "dataset-v2-hash456", "training_config": "configs/train_v3.yaml", "trained_at": "2024-01-15T10:30:00Z" }这样任何一次线上问题,都能追溯到确切的代码、数据、配置组合。
8. 从零到一的路线图建议
如果你现在要从零构建AI工程能力,我建议按这个顺序推进,每一步都有明确的交付物:
第一阶段(1-2周):把环境跑通,能用Docker起一个PyTorch环境,跑通一个预训练模型的推理。交付物是一个能接收文本、返回分类结果的HTTP服务。
第二阶段(2-3周):加上数据处理管道,用DVC管理数据版本,训练和推理共享预处理代码。交付物是一条能从原始数据到训练出模型的完整命令。
第三阶段(2-3周):引入实验追踪和配置管理,能对比不同超参的实验结果。交付物是一份能复现最佳实验的配置文件。
第四阶段(2-3周):做推理优化,导出ONNX、加动态批处理、尝试量化。交付物是一份推理性能对比报告(延迟、吞吐、显存)。
第五阶段(持续):加上监控和漂移检测,建立灰度发布流程。交付物是一套能自动告警的监控面板。
这个路线图的关键是每个阶段都有可运行的交付物,而不是学完所有理论再动手。AI工程是实践性极强的领域,很多坑只有真正跑起来才会遇到。
我个人在实际操作中的体会是,AI工程最难的不是某个具体技术点,而是把散落的技术点串成一条稳定可靠的流水线。模型会换、数据会变、需求会调整,但那条流水线的骨架——配置管理、版本控制、测试、监控——是相对稳定的。把这条骨架搭好,后面换什么模型都是往里填的事。