1. 从零搭建AI工程能力为什么“会用模型”和“会做工程”是两回事很多人第一次接触AI项目时都会经历一个相似的阶段在笔记本里跑通一个模型准确率看着还不错于是觉得“AI也就这么回事”。可一旦要把这个东西放到真实业务里问题就全冒出来了——模型加载慢、显存不够、接口响应超时、并发一上来就崩、日志里全是看不懂的报错。这时候你才会意识到训练一个模型和交付一个AI系统中间隔着一整条工程链路。ai-engineering-from-scratch这个标题核心讲的不是某个具体算法而是从零开始构建AI工程能力这件事本身。它面向的是那些已经懂一点Python、跑过几个demo但还没真正把AI系统落地过的人。关键词里的“from scratch”很关键它意味着不依赖现成的高级封装而是从最基础的环节开始把数据、模型、服务、部署、监控这条链路一层层搭起来。我见过太多人卡在“demo能跑上线就废”的阶段。原因往往不是模型不行而是工程没做。比如数据管道没有版本管理模型更新后效果回退却查不到原因比如推理服务没有做批处理GPU利用率长期低于20%比如没有健康检查服务挂了半小时才被发现。这些问题算法本身解决不了必须靠工程手段。所以这篇内容适合三类人第一类是想从算法岗转向AI工程岗的开发者需要补全工程视角第二类是做后端或运维突然被要求支持AI业务需要快速理解AI系统的特殊性第三类是自己做产品想把AI能力集成进去但不想被云厂商绑死。不管哪一类核心诉求都是一样的——把AI从“能跑”变成“能稳定跑、能规模化跑”。接下来我会按照一个真实项目的推进顺序从环境准备、数据管道、模型训练、推理服务、部署监控这几个环节把每个阶段最容易踩的坑和最关键的设计决策讲清楚。不会堆砌术语而是用我实际做项目时踩过的坑和总结的经验帮你少走弯路。2. 环境与依赖管理别让“在我机器上能跑”成为口头禅2.1 为什么AI项目的环境问题比普通后端更棘手普通后端项目的依赖通常比较稳定一个requirements.txt能管很久。但AI项目不一样它的依赖链条又长又脆弱Python版本、CUDA版本、深度学习框架版本、算子库版本、显卡驱动版本这五者之间必须严格匹配。我遇到过最离谱的一次是同事升级了CUDA驱动结果整个训练集群的PyTorch全部报错因为编译时的CUDA版本和运行时不兼容。更麻烦的是AI项目往往需要同时支持训练和推理两种场景。训练环境需要完整的框架和数据处理库推理环境只需要运行时体积越小越好。如果混在一起推理镜像动辄几个GB部署和扩缩容都会很痛苦。所以从零搭建AI工程能力第一件事就是把环境管理当成一个独立模块来设计而不是随手pip install了事。2.2 用容器隔离训练与推理环境的具体做法我的建议是至少准备两套镜像训练镜像和推理镜像。训练镜像可以大而全包含Jupyter、TensorBoard、完整的数据处理库推理镜像则要精简只保留模型加载和推理必需的依赖。具体操作上可以用多阶段构建。下面是一个推理镜像的Dockerfile示例FROM nvidia/cuda:12.1.1-runtime-ubuntu22.04 AS base RUN apt-get update apt-get install -y \ python3.10 \ python3-pip \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements-infer.txt . RUN pip3 install --no-cache-dir -r requirements-infer.txt COPY model/ ./model/ COPY serve.py . EXPOSE 8000 CMD [python3, serve.py]这里有几个细节值得注意。第一基础镜像用runtime而不是devel体积能小一半以上。第二requirements-infer.txt里只放推理必需的包比如torch、transformers、fastapi、uvicorn不要放pandas、matplotlib这些训练才用的东西。第三模型文件在构建时拷贝进去而不是运行时挂载这样镜像可以独立分发。训练镜像则可以用devel版本并且把常用工具都装上。但要注意训练镜像不要频繁重建因为每次重建都要重新下载几个GB的依赖。我的做法是把基础依赖层固定下来只在上层改代码。2.3 依赖锁定的实操细节与常见坑requirements.txt最大的问题是它只记录直接依赖不记录间接依赖的版本。今天装是好的明天某个间接依赖升级了可能就崩了。所以一定要用pip-compile或者poetry lock把完整依赖树锁死。我习惯用pip-tools流程是这样的# 在 requirements.in 里写直接依赖 echo torch2.1.0 requirements.in echo transformers4.35.0 requirements.in # 生成锁定的 requirements.txt pip-compile requirements.in --output-file requirements.txt生成的requirements.txt会包含所有间接依赖的精确版本这样在任何机器上安装结果都一致。还有一个坑是CUDA版本和框架版本的对应关系。PyTorch官网有一个兼容性表格但很多人不看直接pip install torch结果装到了CPU版本训练时才发现用不了GPU。正确的做法是指定index-urlpip install torch2.1.0 --index-url https://download.pytorch.org/whl/cu121提示每次升级框架版本前先在测试环境验证不要直接在生产环境升级。我见过因为升级transformers导致tokenizer行为变化线上效果直接掉点的案例。3. 数据管道AI工程里最容易被低估的环节3.1 数据版本管理为什么比代码版本管理更重要代码可以回滚数据回滚却很难。模型效果变差了你首先得知道是代码变了还是数据变了。如果数据没有版本管理这个问题根本查不了。我刚开始做AI项目时数据就是放在一个共享目录里谁都可以改。结果有一次模型效果突然下降排查了两天才发现是有人往训练集里加了一批脏数据。从那以后我强制要求所有训练数据必须经过版本管理。具体做法可以用DVCData Version Control它和Git配合使用把大文件存在对象存储里Git里只存元数据。基本流程dvc init dvc add data/train.csv git add data/train.csv.dvc data/.gitignore git commit -m add training data v1这样每次数据变更都会有一个对应的Git commit模型训练时记录commit hash就能追溯到用的是哪版数据。3.2 数据清洗与预处理的工程化落地数据清洗在notebook里做很容易工程化却很难。核心问题是清洗逻辑必须可复用、可测试、可监控。我的做法是把清洗逻辑封装成独立的Python模块每个函数只做一件事并且有对应的单元测试。比如def remove_duplicates(df, subset): 去除重复样本保留第一条 before len(df) df df.drop_duplicates(subsetsubset, keepfirst) after len(df) logger.info(fremoved {before - after} duplicates) return df def filter_by_length(df, text_col, min_len10, max_len512): 按文本长度过滤 df df[df[text_col].str.len().between(min_len, max_len)] return df然后在管道里按顺序调用这些函数。这样做的好处是每个步骤的输入输出都可以单独验证出问题时能快速定位是哪个环节出了问题。还有一个容易被忽略的点是数据分布监控。训练数据和线上推理数据的分布可能不一致这种偏移会导致模型效果下降。我通常会在数据管道里加一个统计模块定期计算关键特征的分布和训练时对比。如果偏移超过阈值就告警。3.3 训练/验证/测试集划分的常见错误很多人划分数据集就是简单train_test_split这在时序数据或分组数据上会出大问题。比如用户行为数据同一个用户的行为可能同时出现在训练集和测试集里导致测试效果虚高。正确的做法是按时间划分或者按用户分组划分。下面是一个按时间划分的例子def time_based_split(df, time_col, train_ratio0.7, val_ratio0.15): df df.sort_values(time_col) n len(df) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) train df.iloc[:train_end] val df.iloc[train_end:val_end] test df.iloc[val_end:] return train, val, test注意划分后一定要检查三个集合的标签分布是否一致。我见过因为随机种子设置不当导致验证集里某个类别的样本数为零的情况。4. 模型训练与实验管理让每次实验都可追溯4.1 实验跟踪不是可选项而是必选项没有实验跟踪的AI项目就像没有版本控制的代码项目。你改了超参数跑了一轮效果好了但忘了改了什么或者效果差了想回退到之前的配置却找不到记录。我用的工具是MLflow它轻量、开源、和Python生态集成好。核心用法很简单import mlflow mlflow.set_experiment(my-ai-project) with mlflow.start_run(): mlflow.log_params({lr: 1e-4, batch_size: 32, epochs: 10}) for epoch in range(10): train_loss train_one_epoch() val_loss validate() mlflow.log_metrics({train_loss: train_loss, val_loss: val_loss}, stepepoch) mlflow.pytorch.log_model(model, model)这样每次实验的参数、指标、模型文件都自动记录还能在UI里对比不同实验。关键是这些记录是结构化的可以程序化查询方便做自动化分析。4.2 超参数搜索的工程化实现超参数搜索如果靠手动调效率极低。但直接上贝叶斯优化又可能过度设计。我的经验是分阶段先粗粒度网格搜索确定大致范围再细粒度随机搜索。用Optuna做超参数搜索的示例import optuna def objective(trial): lr trial.suggest_float(lr, 1e-5, 1e-3, logTrue) batch_size trial.suggest_categorical(batch_size, [16, 32, 64]) dropout trial.suggest_float(dropout, 0.1, 0.5) model build_model(dropoutdropout) val_loss train_and_validate(model, lr, batch_size) return val_loss study optuna.create_study(directionminimize) study.optimize(objective, n_trials50)这里的关键是每次trial都要记录到MLflow否则搜索完了不知道哪组参数最好。另外搜索空间不要设太大否则50次trial可能都覆盖不到好区域。4.3 模型检查点与恢复训练的正确姿势训练大模型时中途中断是常态。如果没有检查点机制一次中断可能损失几天的训练成果。但检查点也不是随便存要考虑存储成本和恢复速度。我的做法是保存两类检查点定期检查点和最佳检查点。定期检查点每N个epoch保存一次用于恢复训练最佳检查点根据验证集指标保存用于最终部署。def save_checkpoint(model, optimizer, epoch, path): torch.save({ epoch: epoch, model_state: model.state_dict(), optimizer_state: optimizer.state_dict(), }, path) # 定期保存 if epoch % 5 0: save_checkpoint(model, optimizer, epoch, fckpt/epoch_{epoch}.pt) # 最佳保存 if val_loss best_val_loss: best_val_loss val_loss save_checkpoint(model, optimizer, epoch, ckpt/best.pt)提示恢复训练时不仅要恢复模型参数还要恢复优化器状态和epoch计数否则学习率调度会乱掉。这个坑我踩过恢复后loss直接飙升。5. 推理服务从单机脚本到高并发API的跨越5.1 推理服务的核心性能指标与优化方向推理服务和训练最大的区别是训练看吞吐推理看延迟。线上服务通常有明确的延迟要求比如P99小于200ms。如果达不到用户体验就会受影响。影响推理延迟的因素主要有几个模型大小、批处理策略、硬件利用率、请求排队。优化顺序应该是先做批处理再做模型量化最后考虑硬件升级。批处理是性价比最高的优化手段。GPU的算力很强但单条请求往往用不满。把多个请求攒成一批一起推理吞吐能提升几倍甚至十几倍。但批处理会增加延迟因为要等请求攒够。所以需要根据延迟要求设置合理的批处理窗口。5.2 用FastAPI搭建推理服务的完整示例下面是一个支持动态批处理的推理服务示例from fastapi import FastAPI from pydantic import BaseModel import torch import asyncio from queue import Queue from threading import Thread app FastAPI() class Request(BaseModel): text: str class BatchProcessor: def __init__(self, model, max_batch_size32, max_wait0.01): self.model model self.max_batch_size max_batch_size self.max_wait max_wait self.queue Queue() self.thread Thread(targetself._process, daemonTrue) self.thread.start() def _process(self): while True: batch [] # 攒批 while len(batch) self.max_batch_size: try: item self.queue.get(timeoutself.max_wait) batch.append(item) except: break if batch: texts [item[0] for item in batch] futures [item[1] for item in batch] results self.model.predict(texts) for future, result in zip(futures, results): future.set_result(result) async def predict(self, text): loop asyncio.get_event_loop() future loop.create_future() self.queue.put((text, future)) return await future processor BatchProcessor(model) app.post(/predict) async def predict(req: Request): result await processor.predict(req.text) return {result: result}这个实现里max_wait控制攒批窗口max_batch_size控制最大批大小。两个参数需要根据实际延迟和吞吐要求调优。5.3 模型量化与加速的实操经验如果批处理还不够可以考虑模型量化。量化是把FP32的权重转成INT8模型体积缩小4倍推理速度提升2-3倍精度损失通常在1%以内。PyTorch的动态量化很简单import torch.quantization model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )但量化不是万能的。有些模型对量化敏感精度掉得厉害。我的做法是量化后一定要在验证集上跑一遍确认精度损失在可接受范围内再上线。还有一个加速手段是ONNX Runtime。把PyTorch模型导出成ONNX用ONNX Runtime推理通常比原生PyTorch快20%-50%。导出命令torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch}, output: {0: batch}} )注意ONNX导出时如果用了动态shape要确保推理时也支持动态batch。否则批处理会报错。6. 部署与监控让AI系统真正“活”在线上6.1 容器化部署的关键配置推理服务容器化时有几个配置直接影响稳定性。第一是资源限制必须设置CPU和内存的request和limit否则一个服务可能把节点资源吃光。第二是健康检查包括存活探针和就绪探针。resources: requests: memory: 2Gi cpu: 1 limits: memory: 4Gi cpu: 2 livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: 8000 initialDelaySeconds: 10 periodSeconds: 5initialDelaySeconds要设得足够大因为模型加载可能需要几十秒。如果设太小服务还没加载完就被判定为不健康会反复重启。6.2 监控指标的设计与告警阈值AI服务的监控和普通后端不一样除了CPU、内存、QPS这些常规指标还要监控模型特有的指标推理延迟分布、批处理大小分布、输入数据分布、预测结果分布。推理延迟要关注P50、P95、P99而不是平均值。平均值会被大量快请求拉低掩盖长尾问题。我通常设置P99超过500ms就告警。输入数据分布监控也很重要。如果线上输入和训练分布差异过大模型效果会下降。可以定期计算输入文本长度的分布、词频分布和训练集对比。def monitor_input_distribution(texts, reference_stats): lengths [len(t.split()) for t in texts] current_mean np.mean(lengths) if abs(current_mean - reference_stats[mean_length]) reference_stats[std_length] * 3: alert(input distribution shifted)6.3 灰度发布与回滚机制模型更新不能一次性全量必须灰度。我的做法是按流量比例逐步放量1% - 10% - 50% - 100%。每个阶段观察至少半小时确认核心指标没有异常再进入下一阶段。回滚机制要提前准备好。模型文件、配置、镜像都要有版本记录回滚时直接切到上一个版本。关键是回滚要快最好能在5分钟内完成。所以镜像要提前构建好不要等回滚时再构建。提示灰度发布时新旧版本的流量要能区分否则监控指标混在一起看不出问题。可以在请求头里加版本标识或者在网关层做流量染色。7. 一些踩坑之后的个人体会做AI工程这几年最大的体会是算法决定上限工程决定下限。一个效果90分的模型如果工程没做好线上可能只能发挥60分而一个效果80分的模型工程做扎实了线上能稳定发挥78分。对于大多数业务场景稳定性比极致效果更重要。另一个体会是不要过早优化。我见过团队在模型还没调好的时候就花大量时间做推理加速结果模型换了加速工作全白费。正确的顺序是先跑通链路再优化效果最后优化性能。还有一点文档和自动化测试在AI项目里特别重要。因为AI项目的不确定性本来就高如果工程环节再不可靠排查问题会非常痛苦。每个模块的输入输出、每个配置项的含义、每个告警的处理方式都应该有文档记录。最后分享一个小技巧在推理服务里加一个/debug接口返回当前模型版本、配置、最近一次请求的输入输出。线上出问题时这个接口能帮你快速定位是模型问题还是数据问题。当然这个接口要做好权限控制不要暴露敏感信息。