1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。市面上讲AI的文章十篇里有八篇在教你pip install之后怎么调API剩下两篇在讲怎么改config里的超参。真正从零开始、把AI工程当成一门手艺来拆解的少得可怜。我自己在这个坑里摸爬滚打了几年从最早用numpy手写全连接层到后来带团队做推理服务踩过的坑能写满一个笔记本。所以看到这个标题我特别有共鸣——它说的不是“从零学AI理论”而是“从零搭AI工程”这两件事差别巨大。先把这个标题拆开看。AI工程这四个字核心不在“AI”在“工程”。模型是算法工程师的事但把模型变成一个能扛住流量、能持续迭代、能被人维护的系统是工程的事。From scratch也不是让你从晶体管开始造计算机而是说你要理解每一层抽象下面到底发生了什么而不是把model.fit()当成黑盒。这个项目适合谁适合那些已经会调包、但一遇到线上问题就抓瞎的人适合那些想搞清楚“为什么我的模型本地跑得好好的一上线就崩”的人也适合刚入行、想建立完整知识框架的新人。它解决的核心问题是把AI从“实验室玩具”变成“生产系统”之间那条巨大的鸿沟。我见过太多团队模型指标刷得漂漂亮亮一到部署就原形毕露。显存爆了、延迟飙了、并发上不去、版本对不上、数据漂移了没人发现。这些问题的根因往往不是模型不行而是工程没做扎实。所以这篇博文我想按照一个真实的搭建路径把AI工程从零到一的关键环节掰开揉碎讲一遍。不是理论综述是我自己趟出来的路线图。2. 整体设计思路AI工程到底该分几层2.1 为什么不能“一个脚本走天下”刚入门的时候我最喜欢写那种“一个train.py搞定一切”的脚本。读数据、预处理、建模型、训练、评估、保存全塞在一个文件里。本地跑没问题一旦要改点东西就痛苦了——换个数据集要改五处换个模型要动十行想加个日志得从头翻到尾。这就是典型的“没有工程分层”的后果。AI工程的本质是把一个复杂的、不确定的系统拆成若干个职责清晰的模块让每个模块可以独立开发、独立测试、独立替换。我后来总结出一个比较通用的分层思路从下往上大致是数据层、特征层、模型层、服务层、监控层。每一层只跟相邻的层打交道层与层之间通过明确的接口通信。这样做的好处是当你想把sklearn换成pytorch时只需要动模型层当你想把flask换成fastapi时只需要动服务层。提示分层不是目的解耦才是。如果你的项目只有一个人维护、生命周期不超过一个月过度分层反而是负担。分层的前提是你预期这个系统会长期演进。2.2 从零搭建的四个阶段我把整个搭建过程分成四个阶段每个阶段有明确的产出物和验收标准。这个划分方式是我带过几个项目之后沉淀下来的比按技术栈划分更贴近实际推进节奏。阶段核心目标关键产出验收标准阶段一跑通闭环数据到预测的完整链路可复现的训练脚本换台机器能跑出同样结果阶段二工程化改造模块解耦、配置管理分层代码结构改配置不改代码阶段三服务化部署对外提供稳定接口API服务容器镜像压测达标、能回滚阶段四可观测与迭代监控、告警、持续训练监控面板自动化流水线问题可定位、模型可更新很多人卡在阶段一和阶段二之间。阶段一能跑通就觉得“差不多了”直接跳到部署结果阶段三各种问题爆发。我的建议是阶段二该花的功夫一定要花它是后面所有工作的地基。2.3 技术选型的取舍逻辑选型这件事没有银弹只有权衡。我列几个我实际做过的决策以及背后的理由。框架选型PyTorch还是TensorFlow我的判断标准是看团队背景和部署环境。如果团队偏研究、需要快速实验PyTorch的动态图更顺手如果部署环境对推理性能要求极高、且需要成熟的移动端支持TensorFlow的生态更完整。但说实话到了2024年这个时间点两者差距已经很小了选哪个都能干活关键是团队熟悉哪个。服务框架Flask、FastAPI还是Triton小规模、低并发用Flask足够需要异步、自动文档、类型校验用FastAPI如果模型推理本身是瓶颈、需要批处理和张量并行直接上Triton Inference Server。我一般先用FastAPI把业务逻辑跑通等推理成为瓶颈再引入Triton。配置管理硬编码、argparse还是Hydra实验阶段argparse够用但一旦配置项超过二十个Hydra的层级配置和命令行覆盖能力就体现出价值了。我踩过的坑是早期用argparse后来配置爆炸改一个参数要翻半天代码。3. 核心细节解析每个环节的坑与解法3.1 数据层别让脏数据毁掉整个系统数据层是AI工程里最容易被低估的部分。我见过太多项目模型结构设计得精妙绝伦结果因为训练数据和推理数据的预处理不一致线上效果直接腰斩。这个问题的专业叫法是训练-服务偏差Training-Serving Skew。解决这个问题的核心原则是训练和推理必须共用同一套预处理逻辑。具体做法是把预处理代码抽成一个独立的模块训练时调用它推理时也调用它。千万不要训练时用pandas处理推理时用numpy重写一遍哪怕逻辑看起来一样浮点精度、缺失值处理、类别编码顺序都可能不同。# preprocessing.py —— 训练和推理共用的预处理模块 import numpy as np class Preprocessor: def __init__(self, mean, std, categories): self.mean mean self.std std self.categories categories # 类别到索引的映射 def transform(self, raw): # 数值特征标准化 num (raw[numeric] - self.mean) / (self.std 1e-8) # 类别特征编码未知类别统一映射到 0 cat np.array([self.categories.get(c, 0) for c in raw[category]]) return np.concatenate([num, cat])这个类在训练时被fit出mean、std、categories然后序列化保存推理时直接加载保证逻辑完全一致。我实测下来光是这一条就能消掉一大半“线上线下不一致”的诡异问题。注意标准化用的mean和std必须来自训练集绝对不能用全量数据计算否则就是数据泄露。这个坑我在早期项目里踩过线下指标虚高线上一塌糊涂。3.2 特征层特征存储的必要性判断特征工程是AI工程里最“脏”的活。早期我都是把特征计算写在训练脚本里后来发现同一个特征在训练、评估、推理三个地方各算一遍三份代码迟早会不一致。这时候就需要特征存储Feature Store。但我要泼一盆冷水不是所有项目都需要特征存储。特征存储解决的是“多团队共享特征”“训练推理特征一致性”“特征版本管理”这三个问题。如果你就一个模型、一个团队、特征不超过五十个自己写个feature.py模块就够了上特征存储是杀鸡用牛刀。判断标准很简单当你发现同一个特征被两个以上的模型使用或者你需要回溯“上周三那个模型用的是哪个版本的特征”时就该考虑特征存储了。轻量级的方案可以用Feast重量级的可以用Tecton但我的经验是大部分中小团队自己用Redis加一张元数据表就能搞定。3.3 模型层版本管理与可复现性模型版本管理是另一个血泪教训。我曾经遇到过线上模型效果突然下降排查了两天才发现是有人重新训练了模型、覆盖了原来的文件但没人记录这次训练用了什么数据、什么参数。从那以后我强制要求每次训练必须产出三样东西模型权重、训练配置、数据版本号。# 训练产出的目录结构 runs/ 20240115_143022/ model.pt config.yaml data_version.txt # 记录数据集的哈希或版本标签 metrics.json train.log用时间戳做目录名配合git commit hash基本能保证任何一次训练都可追溯、可复现。如果团队规模再大一点可以引入MLflow或Weights Biases做实验追踪但核心思想是一样的没有记录的训练等于没训练。3.4 服务层延迟与吞吐的平衡术服务层是AI工程最考验功力的地方。模型推理的延迟和吞吐是一对矛盾体批处理能提高吞吐但会增加单条请求的延迟单条推理延迟低但吞吐上不去。怎么平衡我的做法是动态批处理Dynamic Batching。服务端维护一个请求队列攒够一定数量或者等待超过一定时间就触发一次推理。这个“一定数量”和“一定时间”就是需要调的参数。等待时间设得太短批处理效果不明显设得太长用户等得着急。我一般从max_batch_size32、max_wait_ms10开始调根据实际压测结果微调。# 简化的动态批处理逻辑 import asyncio from collections import deque class BatchScheduler: def __init__(self, max_batch_size32, max_wait_ms10): self.max_batch_size max_batch_size self.max_wait max_wait_ms / 1000 self.queue deque() async def add_request(self, data): future asyncio.Future() self.queue.append((data, future)) if len(self.queue) self.max_batch_size: await self._flush() return await future async def _flush(self): batch list(self.queue) self.queue.clear() # 调用模型推理把结果 set 到各自的 future ...这段代码是简化版真实场景还要考虑超时、异常、优先级等。但核心思路就是用一点点延迟换吞吐用队列做缓冲。4. 实操过程从空目录到可服务系统4.1 环境准备与依赖锁定从零开始的第一步不是写代码是搭环境。我见过太多“在我机器上能跑”的悲剧根因都是依赖版本不一致。所以第一件事是锁定依赖。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装依赖并锁定版本 pip install torch2.1.0 fastapi0.104.0 uvicorn0.24.0 pip freeze requirements.txtrequirements.txt里必须写死版本号不能用torch2.0这种模糊写法。如果项目复杂建议用poetry或pipenv管理它们能处理依赖冲突。我个人的习惯是用poetry因为它的lock文件能保证所有人装出来的环境完全一致。提示如果涉及GPU还要记录CUDA版本和驱动版本。torch.cuda.version和nvidia-smi的输出都要存档否则换台机器可能就跑不起来了。4.2 数据管道搭建数据管道我一般用PyTorch的Dataset和DataLoader来搭因为它们的抽象足够好支持多进程加载、打乱、批处理。关键是要把数据读取和预处理分开Dataset只负责“取一条原始数据”预处理交给前面说的Preprocessor。from torch.utils.data import Dataset, DataLoader class MyDataset(Dataset): def __init__(self, file_paths, preprocessor): self.files file_paths self.preprocessor preprocessor def __len__(self): return len(self.files) def __getitem__(self, idx): raw load_raw(self.files[idx]) # 只负责读取 features self.preprocessor.transform(raw) # 预处理 label raw[label] return features, label # 使用 dataset MyDataset(train_files, preprocessor) loader DataLoader(dataset, batch_size64, shuffleTrue, num_workers4)num_workers这个参数很关键。设成0表示在主进程加载会阻塞训练设成4或8能并行加载但设太大反而会因为进程切换开销降低效率。我的经验是设成CPU核心数的一半左右比较合适。4.3 训练循环与检查点训练循环看起来简单但细节很多。我列几个必须处理的点梯度裁剪防止梯度爆炸、学习率调度让训练后期更稳定、检查点保存防止训练中断白跑、早停防止过拟合。import torch from torch.optim.lr_scheduler import CosineAnnealingLR def train(model, loader, epochs, lr, save_dir): optimizer torch.optim.Adam(model.parameters(), lrlr) scheduler CosineAnnealingLR(optimizer, T_maxepochs) best_loss float(inf) for epoch in range(epochs): model.train() total_loss 0 for features, labels in loader: optimizer.zero_grad() outputs model(features) loss torch.nn.functional.cross_entropy(outputs, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() total_loss loss.item() scheduler.step() avg_loss total_loss / len(loader) # 保存最佳检查点 if avg_loss best_loss: best_loss avg_loss torch.save(model.state_dict(), f{save_dir}/best.pt) # 每轮都保存防止中断 torch.save(model.state_dict(), f{save_dir}/last.pt) print(fEpoch {epoch}, Loss {avg_loss:.4f})clip_grad_norm_的max_norm设成1.0是个比较通用的起点如果训练不稳定可以调小如果收敛太慢可以调大。CosineAnnealingLR适合大多数场景比固定学习率效果好。4.4 服务封装与容器化模型训练好之后要封装成服务。我用FastAPI比较多因为它自带文档、支持异步、类型校验方便。from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model load_model(runs/best.pt) preprocessor load_preprocessor(runs/preprocessor.pkl) class Request(BaseModel): numeric: list[float] category: list[str] app.post(/predict) async def predict(req: Request): features preprocessor.transform(req.dict()) with torch.no_grad(): logits model(torch.tensor(features).unsqueeze(0)) prob torch.softmax(logits, dim-1) return {probabilities: prob.tolist()}容器化用DockerDockerfile要基于nvidia/cuda镜像如果用了GPU并且把模型文件复制进去。注意镜像层缓存把requirements.txt的安装放在复制代码之前这样改代码不会触发依赖重装。FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04 WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]4.5 压测与性能调优服务上线前必须压测。我用locust或wrk模拟不同并发下的延迟和吞吐。关键指标是P99延迟和QPS。P99延迟比平均延迟重要得多因为用户感知到的是最慢的那1%。压测时我会重点观察三个地方GPU利用率、CPU利用率、内存占用。如果GPU利用率低说明瓶颈在数据预处理或网络传输如果CPU利用率高说明预处理太重考虑用GPU做预处理或者优化代码如果内存持续增长可能有内存泄漏检查是否有全局变量在累积。5. 常见问题与排查技巧实录5.1 线上线下效果不一致这是最高频的问题。排查思路按顺序来先查预处理是否一致再查特征计算是否一致最后查模型版本是否一致。我做过统计八成的问题出在预处理。具体做法是拿一条线上请求的原始数据分别走训练预处理和推理预处理对比输出。如果不一样问题就找到了。5.2 推理延迟突然飙升延迟飙升通常有三个原因请求量突增导致排队、某个请求触发了慢路径、资源被其他进程抢占。排查时先看监控面板的QPS曲线如果QPS没变但延迟涨了大概率是慢路径如果QPS涨了是容量问题。慢路径常见于输入数据异常比如超长文本、超大图片需要在预处理阶段加长度限制。5.3 模型更新后效果回退新模型上线效果变差先别急着回滚。检查新模型的训练数据分布是否和线上一致检查评估指标是否选对了检查A/B测试的流量分配是否均匀。我遇到过一次新模型在离线评估上更好但线上更差最后发现是离线评估用了旧的数据划分存在数据泄露。5.4 常见问题速查表现象可能原因排查方法解决方向线上线下不一致预处理逻辑不同对比同一条数据的处理结果抽取共用预处理模块延迟飙升请求排队/慢路径看QPS和P99曲线扩容/加输入限制效果回退数据分布漂移对比训练和线上数据统计重新训练/加监控内存泄漏全局变量累积长时间压测看内存曲线检查缓存和日志训练不收敛学习率过大/数据未归一化看loss曲线调小学习率/加标准化5.5 几个我踩过的坑第一个坑是日志打太多。早期我在推理服务里把每条请求的完整输入输出都打到日志里结果磁盘两天就满了而且日志IO拖慢了服务。后来改成只打关键字段和采样日志问题解决。第二个坑是模型文件太大导致容器启动慢。一个几百MB的模型每次启动都要从镜像里加载冷启动要几十秒。后来我把模型放到对象存储启动时下载配合本地缓存冷启动降到几秒。第三个坑是没有做优雅关闭。服务收到终止信号时直接退出正在处理的请求全部失败。后来加了信号处理等当前批次处理完再退出用户体验好很多。6. 监控与持续迭代让系统自己会说话6.1 必须监控的四个指标AI系统的监控和普通后端系统不一样除了CPU、内存、延迟这些常规指标还要监控模型层面的指标。我一般盯四个输入数据分布、预测结果分布、特征缺失率、模型置信度。输入数据分布用PSIPopulation Stability Index衡量超过0.2就说明分布漂移了。预测结果分布看类别比例如果突然某个类别占比暴涨可能是数据问题。特征缺失率能提前发现上游数据管道故障。模型置信度下降往往预示着模型需要重新训练了。6.2 持续训练的触发条件持续训练不是定时任务而是事件驱动。我设置的触发条件有三个PSI超过阈值、模型置信度连续下降、人工反馈的bad case累积到一定数量。满足任一条件就触发重新训练训练完先影子部署不接真实流量对比新旧模型效果达标再切流量。6.3 回滚机制回滚机制是保命的。我的做法是每次部署都保留上一个版本的镜像和模型文件切流量用配置中心控制出问题一键切回。回滚时间要控制在分钟级超过五分钟的回滚等于没有回滚。7. 一些个人体会这套从零搭建的路径我完整走过不止一遍。最大的感受是AI工程的难点从来不在算法而在一致性和可观测性。一致性保证训练和推理行为相同可观测性保证出问题能快速定位。把这两件事做好系统就稳了一大半。另外别追求一步到位。我见过太多团队一上来就想搭一套“完美”的MLOps平台结果半年过去了还在搭平台模型一个没上线。正确的做法是先跑通最小闭环再逐步加工程化能力。每加一层都要问自己这一层解决了什么具体问题如果答不上来就不该加。最后分享一个小技巧给每个模型训练任务起一个有意义的名字包含日期、数据版本、关键参数比如20240115_v2data_lr1e-3。这个习惯看起来不起眼但当你三个月后需要回溯某个模型时会感谢当时的自己。