1. 从零搭建AI工程能力为什么“会用模型”和“会做工程”是两回事很多人第一次接触AI项目时都会经历一个相似的阶段在笔记本里跑通一个模型准确率看着还不错于是觉得“AI也就这么回事”。可一旦要把这个模型放到真实业务里问题就全冒出来了——推理延迟高得离谱、显存动不动就爆、服务一重启模型就得重新加载、并发一上来整个进程直接卡死。这时候你才会意识到训练一个模型和交付一个AI系统中间隔着一整条工程链路。“ai-engineering-from-scratch”这个标题说的正是这条链路。它不是教你调某个库的API也不是带你复现某篇论文而是从零开始把AI工程里那些真正决定成败的环节一个个搭起来。适合谁看如果你已经会写Python、懂一点深度学习基础但一到部署、优化、服务化就发怵那这篇内容就是冲着你来的。如果你是完全的新手也没关系我会尽量把每个概念用生活化的方式讲清楚让你知道每一步“为什么要这么做”。我自己的经历是早期做AI项目时踩过最大的坑不是模型效果不好而是工程细节没处理好导致整个系统不可用。比如有一次线上服务突然大面积超时排查了半天才发现是某个预处理函数里用了Python的全局锁并发一高就串行化了。这种问题任何一本机器学习教材都不会告诉你但它在真实工程里却是致命的。所以这篇博文我会围绕“从零搭建AI工程能力”这个核心把环境、数据、推理、服务、监控这几个关键环节拆开讲每个环节都给出可落地的做法和我自己踩过的坑。2. 环境与依赖别让“在我机器上能跑”成为口头禅2.1 为什么环境隔离是AI工程的第一课AI项目的依赖复杂度远超普通后端项目。一个典型的AI工程环境里可能同时存在CUDA、cuDNN、PyTorch、TensorFlow、ONNX Runtime、各种图像/音频处理库而这些库之间的版本兼容性极其脆弱。我见过太多团队因为某个人升级了一个小版本导致整个训练流水线崩掉的情况。所以从零开始的第一件事就是把环境隔离做到极致。我的建议是训练环境用Conda管理推理环境用Docker固化。为什么这么分因为训练阶段需要频繁试错Conda的灵活性更好而推理阶段追求稳定和可复现Docker的镜像机制能保证“构建一次到处运行”。具体操作上训练环境我会这样组织conda create -n ai-train python3.10 conda activate ai-train pip install torch2.1.0 torchvision0.16.0 --index-url https://download.pytorch.org/whl/cu118这里有个细节PyTorch的版本和CUDA版本必须严格对应。很多人直接pip install torch装到的是CPU版本或者和驱动不匹配的版本然后花几个小时排查为什么GPU用不了。我的经验是先去PyTorch官网查好对应关系再执行安装命令。2.2 推理镜像的构建要点推理镜像和训练环境最大的区别是镜像要尽可能小启动要尽可能快。我见过一个推理镜像打了8个G每次扩容拉取镜像就要好几分钟这在弹性伸缩场景下是灾难性的。构建推理镜像时我会用多阶段构建FROM nvidia/cuda:11.8.0-runtime-ubuntu22.04 AS base RUN apt-get update apt-get install -y python3.10 python3-pip rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . /app WORKDIR /app CMD [python3, server.py]注意这里用的是runtime而不是devel镜像因为推理不需要编译工具链用runtime能省下好几个G。另外--no-cache-dir也很关键不然pip缓存会让镜像体积翻倍。提示如果你的模型需要自定义CUDA算子那还是得用devel镜像来编译但编译完可以把产物拷到runtime镜像里保持最终镜像精简。2.3 依赖锁定的实操方法“在我机器上能跑”这句话的根源就是依赖没有锁定。我的做法是训练环境用pip freeze导出完整依赖推理环境用pip-tools生成锁定文件。pip freeze requirements-lock.txt但这里有个坑pip freeze会把所有间接依赖都列出来有时候会包含一些平台相关的包。更稳妥的方式是用pip-compilepip install pip-tools pip-compile requirements.in --output-file requirements.txtrequirements.in里只写你直接依赖的包pip-compile会自动解析出所有间接依赖并锁定版本。这样既清晰又可靠。3. 数据处理流水线模型效果的上限往往在这里被决定3.1 为什么数据预处理要独立成服务很多AI项目初期数据预处理是直接写在推理代码里的。比如一个图像分类服务收到请求后先resize、再归一化、再转tensor然后喂给模型。这种做法在单机、低并发时没问题但一旦要扩展就会遇到两个问题一是预处理和推理耦合在一起没法单独优化二是不同语言的客户端可能重复实现预处理逻辑导致线上线下不一致。我的做法是把预处理独立成一个轻量服务推理服务只接收已经处理好的张量。这样做的好处是预处理可以用CPU集群横向扩展推理服务专注用GPU而且预处理逻辑只有一份不会出现“训练时用Python预处理线上用Java预处理”导致的效果偏差。3.2 数据版本管理别让“数据变了”成为玄学AI工程里有一个容易被忽视的环节数据版本管理。模型有版本代码有版本但数据往往没有。结果就是某天模型效果突然下降排查半天发现是上游数据源改了字段格式。我的经验是给每个训练数据集打上版本号并且记录它的来源、处理脚本、统计信息。可以用DVC这样的工具也可以简单点用文件命名加元数据JSON{ dataset_version: v20240115, source: s3://bucket/raw/2024-01-15/, preprocess_script: preprocess_v3.py, num_samples: 120000, label_distribution: {cat: 0.4, dog: 0.6} }这样当效果波动时你可以快速对比不同版本的数据集定位是不是数据本身变了。3.3 在线特征与离线特征的一致性如果你的AI系统用到特征工程那在线特征和离线特征的一致性是必须死守的底线。我见过一个推荐系统离线AUC 0.85上线后CTR惨不忍睹最后发现是离线用Spark算特征在线用Python算两边对空值的处理逻辑不一样。解决这个问题的标准做法是用同一套代码做特征计算。可以把特征逻辑封装成一个库离线和在线都调用它。如果性能不允许至少要用同一套测试用例来验证两边输出一致。4. 推理优化从“能跑”到“跑得快”的关键跨越4.1 模型格式转换的取舍训练出来的模型通常是PyTorch的.pt或TensorFlow的.pb但直接拿来做推理往往不是最优的。常见的推理格式有ONNX、TensorRT、OpenVINO等它们各自适合不同场景。格式适用场景优点缺点PyTorch原生快速验证无需转换推理速度一般ONNX跨平台部署通用性好部分算子支持不全TensorRTNVIDIA GPU速度极快绑定NVIDIAOpenVINOIntel CPUCPU优化好绑定Intel我的建议是先用ONNX做通用部署如果性能不够再针对特定硬件做深度优化。ONNX的转换相对简单import torch model MyModel() model.load_state_dict(torch.load(model.pt)) dummy_input torch.randn(1, 3, 224, 224) torch.onnx.export(model, dummy_input, model.onnx, opset_version13)这里opset_version的选择很关键太低可能不支持某些算子太高可能目标推理引擎不支持。我一般用13或14兼容性比较好。4.2 批处理与动态形状的平衡推理服务的一个核心矛盾是批处理能提高吞吐但会增加延迟。如果固定batch size小请求也要等凑够一批才处理延迟不可控如果batch size1GPU利用率又上不去。我的做法是动态批处理设置一个最大batch size和一个最大等待时间比如最多等10ms凑够8个请求就立即处理否则10ms后有多少处理多少。这样在吞吐和延迟之间取得平衡。实现上可以用Triton Inference Server它内置了动态批处理功能配置一下就行dynamic_batching { max_batch_size: 8 preferred_batch_size: [4, 8] max_queue_delay_microseconds: 10000 }4.3 量化与剪枝什么时候该用什么时候别碰量化把FP32转成INT8和剪枝去掉不重要的权重是常见的模型压缩手段但不是所有场景都适合。我的经验是如果模型是图像分类、检测这类对精度不那么敏感的任務量化通常能带来2-4倍加速精度损失在1%以内。如果是生成式模型或者对数值精度要求高的任务量化可能导致输出质量明显下降要谨慎。剪枝更适合参数量巨大的模型小模型剪枝收益有限。量化可以用PyTorch的torch.quantization也可以用ONNX Runtime的量化工具。我一般先用动态量化试试水import torch.quantization quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )动态量化只量化权重激活值还是浮点实现简单适合快速验证。5. 服务化与并发让模型真正扛住流量5.1 为什么Flask不适合做AI推理服务很多教程用Flask写推理服务因为简单。但Flask默认是同步阻塞的一个请求在推理时其他请求只能等着。虽然可以用gunicorn开多worker但每个worker都会加载一份模型显存直接翻倍。我的建议是用异步框架或者专门的推理服务器。如果坚持用PythonFastAPI是更好的选择它基于async配合uvicorn能处理更高并发。但要注意PyTorch推理本身是同步的需要用线程池来跑from fastapi import FastAPI from concurrent.futures import ThreadPoolExecutor import torch app FastAPI() executor ThreadPoolExecutor(max_workers4) model load_model() app.post(/predict) async def predict(data: dict): loop asyncio.get_event_loop() result await loop.run_in_executor(executor, inference, data) return result这样多个请求可以并发进入推理在线程池里排队执行不会阻塞事件循环。5.2 模型加载与热更新服务重启时重新加载模型会导致一段时间的不可用。对于大模型加载可能要好几分钟。我的做法是模型常驻内存通过信号触发热更新import signal model load_model(model_v1.pt) def reload_model(signum, frame): global model new_model load_model(model_v2.pt) model new_model signal.signal(signal.SIGUSR1, reload_model)这样新模型加载好之后才替换引用旧模型等没有请求引用了再释放实现无缝切换。5.3 并发下的显存管理GPU显存是稀缺资源并发一高就容易OOM。我的经验是限制并发数根据显存大小和单次推理占用算出最大并发用信号量控制。及时释放中间变量PyTorch的中间张量如果不释放会一直占着显存。用torch.no_grad()包裹推理代码并且手动del不用的变量。监控显存使用用torch.cuda.memory_allocated()定期打印及时发现泄漏。import torch with torch.no_grad(): output model(input_tensor) del input_tensor torch.cuda.empty_cache()注意torch.cuda.empty_cache()会释放缓存但频繁调用会影响性能建议在请求间隙调用。6. 监控与迭代上线只是开始不是结束6.1 推理服务的核心监控指标AI服务和普通后端服务的监控指标有重叠也有差异。除了QPS、延迟、错误率这些通用指标AI服务还要关注GPU利用率太低说明资源浪费太高说明可能成为瓶颈。显存占用持续增长可能是泄漏。输入数据分布如果输入突然变化比如图片尺寸变了可能导致预处理出错。模型输出分布如果输出置信度突然普遍降低可能是数据漂移。我一般用Prometheus采集指标Grafana做面板。GPU指标可以用nvidia-smi导出或者用DCGM Exporter。6.2 数据漂移的检测与应对数据漂移是AI系统效果下降的主要原因之一。检测方法有很多简单点的可以监控输入特征的均值和方差复杂点的可以用KS检验或者PSI。我的做法是每周跑一次离线评估对比当前线上数据和训练数据的分布。如果发现明显漂移就触发重新训练。这个流程可以自动化from scipy.stats import ks_2samp def detect_drift(train_data, live_data, threshold0.05): stat, p_value ks_2samp(train_data, live_data) if p_value threshold: return True # 有漂移 return False6.3 A/B测试与灰度发布新模型上线不能一下子全量要先灰度。我的做法是用流量比例控制先放1%流量到新模型观察核心指标没有异常再逐步放大。实现上可以在网关层做分流根据用户ID哈希取模。要注意的是同一个用户要尽量固定到同一个模型不然体验会不一致。def route_model(user_id, new_model_ratio0.01): if hash(user_id) % 100 new_model_ratio * 100: return new_model return old_model灰度期间要密切监控新模型的延迟、错误率和业务指标一旦异常立即回滚。7. 我踩过的几个典型坑与应对思路7.1 预处理用了多进程导致内存爆炸有一次做图像服务预处理用了multiprocessing来加速结果每个worker都复制了一份模型显存直接爆了。后来改成预处理和推理分离预处理用CPU多进程推理用GPU单进程问题解决。7.2 日志打印了张量导致性能骤降调试时习惯性把输入输出张量打印到日志结果日志量巨大IO成为瓶颈。后来改成只打印形状和统计量比如tensor.shape、tensor.mean()性能恢复正常。7.3 模型文件放在代码仓库导致镜像巨大早期把模型文件直接放在代码目录里每次构建镜像都要拷贝几百兆的模型构建慢、拉取也慢。后来改成模型文件放在对象存储启动时下载镜像体积从2G降到300M。7.4 没有设置超时导致请求堆积推理服务没有设置超时某个请求卡住后后续请求全部排队最终整个服务不可用。后来加了请求级超时超时直接返回错误保护服务整体可用性。import signal class TimeoutError(Exception): pass def handler(signum, frame): raise TimeoutError() signal.signal(signal.SIGALRM, handler) signal.alarm(5) # 5秒超时 try: result inference(data) finally: signal.alarm(0)这些坑看起来都不复杂但每一个都真实地影响过线上服务。AI工程和普通后端工程最大的区别就在于你要同时处理软件工程和机器学习两个领域的复杂性任何一个环节疏忽都可能导致系统不可用。8. 从零搭建的推荐路径与工具选型如果你现在要从零开始搭建AI工程能力我会建议按这个顺序推进先把单机推理跑通用FastAPI或Triton起一个服务能接收请求返回结果。加上Docker把环境固化确保换台机器也能跑。加上监控至少要有延迟、QPS、GPU利用率三个指标。做压力测试用locust或wrk模拟并发找到瓶颈。优化推理根据瓶颈决定是转ONNX、上TensorRT还是加批处理。做灰度发布新模型先小流量验证再全量。建立数据版本和漂移检测让系统能持续迭代。工具选型上我的偏好是推理服务用Triton功能全或FastAPI灵活监控用PrometheusGrafana容器用DockerK8s模型管理用MLflow或自己写简单的版本控制。不需要一上来就上全套按需逐步引入。这个领域变化很快但底层的东西——环境隔离、数据一致性、推理优化、服务稳定性——是不变的。把这些基础打牢上层用什么新框架、新模型都能快速接上。