1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调一下API跑通一个Demo然后发个朋友圈说“今天又搞定了一个AI项目”。我早期也这么干过结果到了真实业务场景里模型效果不稳定、推理延迟忽高忽低、数据管道三天两头断流排查问题的时候连从哪下手都不知道。后来我才意识到调包只能让你跑起来但从零构建才能让你真正掌控它。“ai-engineering-from-scratch”这个标题核心不在“AI”而在“from scratch”。它强调的不是让你去发明一个新的Transformer架构而是让你从最底层的工程视角把AI系统拆开来看数据怎么流进来、特征怎么处理、模型怎么加载、推理怎么调度、服务怎么暴露、监控怎么做。这套东西才是AI工程和“调个API”之间的本质区别。这篇文章适合谁看如果你已经会用Python写点脚本对机器学习有基本概念但一提到“工程化”就头疼那这篇内容就是为你准备的。我会按照一个真实项目的落地路径从环境搭建、数据管道、模型封装、服务部署到性能调优把每个环节的“为什么”和“怎么做”都讲清楚。所有代码和配置都是可以直接抄作业的但更重要的是你要理解每一步背后的工程逻辑。提示本文不会涉及任何特定云厂商的绑定操作所有方案都基于开源工具和通用工程实践确保你在任何环境下都能复现。2. 环境与依赖别让版本冲突毁掉你的第一周2.1 为什么我坚持用虚拟环境而不是全局安装我见过太多人拿到项目第一件事就是pip install -r requirements.txt然后跑着跑着发现某个包版本冲突整个环境崩掉最后只能重装系统。AI工程涉及的工具链特别长数值计算、深度学习框架、数据处理、服务框架、监控工具每个都有自己的依赖树。全局安装等于把所有鸡蛋放在一个篮子里一旦冲突全盘皆输。我的做法是每个项目一个独立的虚拟环境而且用conda而不是venv。原因很简单AI工程里很多底层库比如BLAS、CUDA相关的运行时用conda管理起来更省心尤其是当你需要在同一台机器上切换不同版本的深度学习框架时。下面是我常用的环境初始化流程# 创建独立环境指定Python版本 conda create -n ai-eng python3.10 -y conda activate ai-eng # 先装底层数值库再装上层框架 conda install numpy1.24 pandas2.0 scipy -y pip install torch2.1.0 --index-url https://download.pytorch.org/whl/cpu pip install fastapi uvicorn pydantic scikit-learn注意这里的顺序先conda装底层再pip装上层。因为conda的依赖解析器对二进制兼容性处理得更好而pip在纯Python包上更灵活。如果你反过来先用pip装了numpy再用conda装scipyconda可能会试图替换掉pip装的numpy导致环境混乱。2.2 依赖锁定的正确姿势requirements.txt只写包名和版本号是不够的。AI工程里一个包的间接依赖可能就有几十个而且不同平台Linux、macOS、Windows的解析结果还不一样。我推荐用pip-tools来生成锁定文件pip install pip-tools pip-compile requirements.in --output-file requirements.txt --generate-hashes pip-sync requirements.txt--generate-hashes会为每个包生成哈希校验值确保你安装的包没有被篡改。pip-sync则会严格按照锁定文件同步环境多装的包会被卸载少装的会补上。这套流程在团队协作里特别重要因为“在我机器上能跑”这句话90%的情况都是依赖版本不一致导致的。注意如果你在用GPU环境CUDA版本和深度学习框架版本的对应关系一定要查官方兼容性矩阵。我踩过最坑的一次是torch版本和cuda版本差了一个小版本号结果训练能跑推理直接段错误排查了整整两天。3. 数据管道AI工程里最容易被低估的脏活累活3.1 从原始数据到模型输入中间到底有多少步很多人以为数据管道就是“读CSV扔给模型”。真实项目里从原始数据到模型能用的张量中间至少隔着这些步骤数据采集、格式解析、缺失值处理、异常值检测、特征编码、归一化、数据集划分、批处理封装。每一步都有坑而且这些坑在Demo阶段完全不会暴露只有数据量上来、数据分布变化之后才会集中爆发。我习惯把数据管道拆成三个独立的层采集层、处理层、供给层。采集层负责从各种来源数据库、日志文件、消息队列把数据拉过来统一转成一种中间格式我一般用Parquet列式存储、压缩率高、读取快。处理层负责清洗、转换、特征工程输出的是模型可以直接消费的数值矩阵。供给层负责批处理、打乱、预取把数据高效地喂给训练或推理循环。这种分层的好处是每一层都可以独立测试和替换。比如采集层从MySQL换成Kafka处理层完全不用动处理层从pandas换成Spark供给层也不用改。工程化的核心思想就是解耦让变化的影响范围尽可能小。3.2 用PyTorch Dataset和DataLoader封装数据供给供给层我强烈建议用PyTorch的Dataset和DataLoader即使你用的不是PyTorch训练也可以借用这套抽象。下面是一个我常用的模板from torch.utils.data import Dataset, DataLoader import numpy as np class TabularDataset(Dataset): def __init__(self, features, labels, transformNone): self.features features.astype(np.float32) self.labels labels.astype(np.int64) self.transform transform def __len__(self): return len(self.labels) def __getitem__(self, idx): x self.features[idx] y self.labels[idx] if self.transform: x self.transform(x) return x, y # 使用示例 dataset TabularDataset(X_train, y_train) loader DataLoader( dataset, batch_size256, shuffleTrue, num_workers4, pin_memoryTrue, prefetch_factor2 )这里有几个参数值得展开说。num_workers4表示用4个子进程并行加载数据这个值不是越大越好一般设置为CPU核心数的50%到75%。设太大反而会因为进程切换开销导致吞吐下降。pin_memoryTrue会把数据锁在内存页里加速CPU到GPU的传输但只在GPU训练时有意义。prefetch_factor2表示每个worker预取2个batch这样GPU在计算当前batch时CPU已经在准备后面的数据了能有效避免GPU空转。实操心得如果你发现训练时GPU利用率忽高忽低大概率是数据供给跟不上。先用nvidia-smi看GPU利用率如果经常掉到0%就把num_workers调大或者检查__getitem__里有没有耗时的IO操作。我一般会把所有预处理都在__init__里做完__getitem__只做最轻量的索引和转换。3.3 数据版本管理别让“这次跑的结果和上次不一样”成为常态AI工程和传统软件工程最大的区别之一就是数据也是代码。你改了数据模型效果就变了但如果你不记录数据版本根本不知道是哪次改动导致的。我见过太多团队模型效果波动了大家互相甩锅最后发现是某个人偷偷更新了训练数据。我的做法是用DVCData Version Control来管理数据和模型文件。它和Git配合使用Git管代码DVC管数据。基本流程是dvc init dvc add data/raw/train.parquet git add data/raw/train.parquet.dvc data/raw/.gitignore git commit -m add training data v1dvc add会生成一个.dvc文件里面记录了数据的哈希值。数据本身存在本地或远程存储里Git仓库里只保留这个轻量级的指针文件。这样每次你切换Git分支对应的数据版本也会自动切换。没有数据版本管理的AI项目就像没有Git的代码项目迟早会乱成一锅粥。4. 模型封装从notebook到可复用的工程模块4.1 为什么notebook里的模型不能直接上生产notebook适合探索和实验但它的执行顺序是隐式的状态是全局的依赖是混乱的。你在notebook里定义了一个模型类改了三次最后一次跑通了但前两次的中间状态还留在内存里。这种代码拿到生产环境换一台机器、换一个执行顺序结果就可能完全不同。我的原则是任何要上生产的模型必须封装成独立的Python模块。这个模块对外只暴露两个接口train和predict。内部怎么实现外部不需要知道。下面是一个标准的模型封装模板# model.py import torch import torch.nn as nn from pathlib import Path class MLP(nn.Module): def __init__(self, input_dim, hidden_dim, output_dim): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden_dim, hidden_dim // 2), nn.ReLU(), nn.Linear(hidden_dim // 2, output_dim) ) def forward(self, x): return self.net(x) class ModelWrapper: def __init__(self, config): self.config config self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.model MLP( config[input_dim], config[hidden_dim], config[output_dim] ).to(self.device) def train(self, train_loader, val_loader, epochs10): optimizer torch.optim.Adam(self.model.parameters(), lrself.config[lr]) criterion nn.CrossEntropyLoss() best_val_loss float(inf) for epoch in range(epochs): self.model.train() for x, y in train_loader: x, y x.to(self.device), y.to(self.device) optimizer.zero_grad() loss criterion(self.model(x), y) loss.backward() optimizer.step() val_loss self._evaluate(val_loader, criterion) if val_loss best_val_loss: best_val_loss val_loss self._save_checkpoint() def predict(self, x): self.model.eval() with torch.no_grad(): x torch.tensor(x, dtypetorch.float32).to(self.device) logits self.model(x) return torch.softmax(logits, dim-1).cpu().numpy() def _evaluate(self, loader, criterion): self.model.eval() total_loss 0.0 with torch.no_grad(): for x, y in loader: x, y x.to(self.device), y.to(self.device) total_loss criterion(self.model(x), y).item() return total_loss / len(loader) def _save_checkpoint(self): path Path(self.config[checkpoint_dir]) path.mkdir(parentsTrue, exist_okTrue) torch.save(self.model.state_dict(), path / best.pt)这个封装的关键点在于配置和代码分离。所有超参数、路径、设备选择都通过config字典传入代码本身不硬编码任何环境相关的值。这样同一份代码可以在开发机、测试环境、生产环境无缝切换只需要换配置就行。4.2 模型序列化pickle、torch.save还是ONNX模型训练完要保存保存格式的选择直接影响后续的部署灵活性。我对比过几种常见方案格式优点缺点适用场景pickle简单直接Python原生版本兼容性差有安全风险临时实验torch.savePyTorch官方支持完整状态绑定PyTorch跨框架难PyTorch内部使用ONNX跨框架、跨平台推理优化好转换可能丢失自定义算子生产部署TorchScript脱离Python运行时性能好调试困难动态逻辑受限高性能推理我的建议是训练阶段用torch.save保存检查点部署阶段转成ONNX或TorchScript。ONNX的好处是你可以用ONNX Runtime来推理它比原生PyTorch推理快不少而且不依赖Python环境可以嵌入到C、Java等服务里。转换过程也不复杂import torch.onnx dummy_input torch.randn(1, config[input_dim]).to(device) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version14 )dynamic_axes这个参数很关键它允许模型接受可变长度的batch。如果不设置导出的ONNX模型只能处理固定batch size生产环境里会非常受限。踩坑记录ONNX转换时如果模型里有if分支或者循环转换可能会失败或者行为不一致。我遇到过一次模型里有个根据输入长度动态选择路径的逻辑转ONNX后直接报错。后来把逻辑改成用torch.where实现才转换成功。所以模型设计阶段就要考虑部署友好性别等到转换时才返工。5. 服务化与性能调优让模型真正跑起来5.1 用FastAPI把模型包装成HTTP服务模型封装好了下一步是把它暴露成服务。我选FastAPI的原因很直接异步支持好、自动生成API文档、类型校验强。下面是一个完整的服务模板# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import numpy as np import onnxruntime as ort app FastAPI(titleAI Model Service) class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): probabilities: list[float] predicted_class: int # 全局加载模型避免每次请求都重新加载 session ort.InferenceSession(model.onnx) app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): try: input_array np.array([request.features], dtypenp.float32) outputs session.run(None, {input: input_array}) probs outputs[0][0].tolist() pred_class int(np.argmax(probs)) return PredictResponse(probabilitiesprobs, predicted_classpred_class) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok}启动命令是uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4。--workers 4表示启动4个进程每个进程独立加载模型。这里有个坑如果你的模型很大比如几个GB4个worker会占4倍内存。这时候要么减少worker数量要么用共享内存的方式加载模型。ONNX Runtime支持内存映射多个进程可以共享同一份模型权重能省不少内存。5.2 批处理与动态批处理提升吞吐的关键单条推理的吞吐量是很低的因为每次请求都要走一遍完整的计算图GPU利用率极低。批处理是提升吞吐最直接的手段。但生产环境的请求是零散到来的你不能等凑够一个batch再处理那样延迟会爆炸。这时候就需要动态批处理服务端维护一个请求队列每隔几毫秒或者队列达到一定长度就把当前队列里的请求打包成一个batch一起推理。我用asyncio实现了一个简单的动态批处理逻辑import asyncio from collections import deque class DynamicBatcher: 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() self.lock asyncio.Lock() async def add_request(self, features): future asyncio.Future() async with self.lock: self.queue.append((features, future)) if len(self.queue) self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch list(self.queue) self.queue.clear() features np.array([item[0] for item in batch], dtypenp.float32) outputs session.run(None, {input: features}) for i, (_, future) in enumerate(batch): future.set_result(outputs[0][i].tolist()) async def run_loop(self): while True: await asyncio.sleep(self.max_wait) async with self.lock: if self.queue: await self._process_batch()这个逻辑的核心是请求进来先挂起等凑够batch或者超时后再统一处理。max_wait_ms设成10毫秒意味着最坏情况下请求会多等10毫秒但吞吐量能提升几倍甚至十几倍。这个权衡在生产环境里通常是值得的。5.3 监控与日志别等出事了才想起来看服务上线只是开始真正的挑战在于你怎么知道它现在是不是正常的。我见过太多项目服务跑起来了但没有任何监控直到用户投诉才发现模型输出全是NaN。AI服务的监控比普通服务多几个维度延迟P50、P95、P99分位数不能只看平均值吞吐每秒请求数、每秒处理的样本数错误率HTTP错误、推理异常、超时模型指标输入分布漂移、输出置信度分布、预测类别分布资源CPU、内存、GPU利用率、显存占用我用prometheus_client来暴露指标用Grafana来展示。关键代码是在推理前后打点from prometheus_client import Histogram, Counter INFERENCE_LATENCY Histogram( inference_latency_seconds, Model inference latency, buckets[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0] ) PREDICTION_COUNTER Counter( prediction_total, Total predictions, [predicted_class] ) app.post(/predict) async def predict(request: PredictRequest): with INFERENCE_LATENCY.time(): result await batcher.add_request(request.features) PREDICTION_COUNTER.labels(predicted_classstr(np.argmax(result))).inc() return PredictResponse(probabilitiesresult, predicted_classint(np.argmax(result)))输入分布漂移是AI服务特有的监控项。模型训练时的数据分布和线上推理时的数据分布可能不一致这种不一致会导致模型效果悄悄下降但不会报任何错误。我的做法是定期计算线上输入特征的均值和方差和训练集对比偏差超过阈值就告警。这个逻辑可以放在一个定时任务里不需要实时计算。实操心得监控指标不要贪多先覆盖延迟、错误率、吞吐这三个最基本的再逐步加模型相关的。我见过一个团队监控面板上几十个指标结果真出问题时没人知道该看哪个。监控的目的是快速定位问题不是展示技术实力。6. 从单机到可扩展当请求量涨上来之后6.1 水平扩展的前提无状态化单机服务总有上限请求量涨上来之后必须扩展。扩展的第一步是让服务无状态。什么叫无状态就是任何一个请求打到任何一个实例上结果都一样实例本身不保存任何请求相关的数据。模型权重是只读的可以每个实例加载一份请求队列是临时的处理完就销毁日志和监控数据发到外部系统。做到这些你就可以直接加机器前面挂个负载均衡请求量翻倍就加一倍实例。但这里有个问题动态批处理是有状态的。每个实例维护自己的请求队列负载均衡如果按请求轮询可能导致某些实例的队列很长某些很短。解决方案是用一致性哈希或者最少连接数策略让请求尽量均匀分布。更彻底的做法是把批处理逻辑抽出来做成一个独立的批处理服务所有推理实例从队列里拉取batch。但这会引入额外的网络跳数和复杂度除非请求量真的很大否则单机动态批处理已经够用了。6.2 模型热更新不重启服务换模型模型迭代是常态但每次换模型都重启服务会导致请求中断。热更新的目标是在不中断服务的前提下把新模型加载进来逐步切换流量。我的做法是维护两个模型槽位active和standby。新模型加载到standby预热完成后原子性地把active指针指向新模型旧模型等所有进行中的请求处理完再释放。class ModelManager: def __init__(self, model_path): self.active ort.InferenceSession(model_path) self.standby None self.lock threading.Lock() def load_new_model(self, new_path): new_session ort.InferenceSession(new_path) # 预热跑几次推理让ONNX Runtime完成图优化 dummy np.random.randn(1, 10).astype(np.float32) for _ in range(5): new_session.run(None, {input: dummy}) with self.lock: self.standby new_session def switch(self): with self.lock: if self.standby is not None: self.active, self.standby self.standby, None def predict(self, features): with self.lock: session self.active return session.run(None, {input: features})预热这一步很关键。ONNX Runtime在第一次推理时会做图优化、内存分配等操作如果不预热第一个请求的延迟可能是正常值的几十倍。我一般会跑5到10次预热推理确保延迟稳定后再切换流量。6.3 容量规划到底需要多少资源容量规划不是拍脑袋决定的需要基于实测数据。我的方法是先测单实例的极限吞吐再根据目标QPS计算实例数最后留30%的余量。测极限吞吐时用locust或者wrk压测逐步增加并发数观察延迟和错误率的变化。当P99延迟超过可接受阈值或者错误率开始上升时就是单实例的极限。举个例子假设你测出来单实例在P99延迟100毫秒以内能扛住200 QPS目标QPS是1000那至少需要5个实例。但考虑到流量波动和实例故障实际部署7个实例5个主力2个冗余。这个计算过程要定期重做因为模型更新、数据变化、依赖升级都可能改变单实例的性能特征。注意压测时一定要用真实的数据分布不要用随机生成的假数据。我踩过一次坑压测时用np.random.randn生成输入结果模型里有个分支对特定范围的输入会走更复杂的计算路径真实数据下延迟比压测结果高了3倍。假数据只能测出代码路径测不出真实性能。7. 一些让我少走弯路的工程习惯7.1 配置管理别把密码和路径写死在代码里我早期写代码数据库密码、模型路径、API密钥全是硬编码。结果代码传到Git上密码泄露换台机器跑路径全不对。后来我强制自己用环境变量加配置文件的方式管理配置。敏感信息走环境变量非敏感的默认值走配置文件配置文件用YAML格式支持多环境覆盖。# config/base.yaml model: input_dim: 128 hidden_dim: 256 output_dim: 10 checkpoint_dir: ./checkpoints server: host: 0.0.0.0 port: 8000 workers: 4import os import yaml def load_config(envdev): with open(config/base.yaml) as f: config yaml.safe_load(f) env_config_path fconfig/{env}.yaml if os.path.exists(env_config_path): with open(env_config_path) as f: env_config yaml.safe_load(f) config deep_merge(config, env_config) # 环境变量覆盖 if os.getenv(MODEL_CHECKPOINT_DIR): config[model][checkpoint_dir] os.getenv(MODEL_CHECKPOINT_DIR) return config这套机制的好处是开发环境用开发配置生产环境用生产配置代码完全一样。部署时只需要设置环境变量不需要改任何代码。7.2 测试AI项目也需要单元测试很多人觉得AI项目没法写测试因为模型输出是不确定的。但工程代码是可以测试的。数据预处理函数、特征编码逻辑、批处理队列、配置加载这些都应该有单元测试。模型本身可以用小规模数据做冒烟测试确保训练能跑通、推理不报错、输出形状正确。def test_preprocess(): raw {age: 25, income: 50000, category: A} result preprocess(raw) assert result.shape (1, 10) assert not np.isnan(result).any() def test_model_forward(): model MLP(input_dim10, hidden_dim32, output_dim3) x torch.randn(4, 10) out model(x) assert out.shape (4, 3)这些测试跑起来很快但能在早期发现大部分低级错误。我宁愿花半小时写测试也不愿意花两天排查一个本可以避免的bug。7.3 文档写给三个月后的自己AI工程项目里最容易被忽略的就是文档。但三个月后你回头看自己的代码如果没有文档你大概率会想“这写的什么鬼”。我的文档习惯是每个模块顶部写清楚这个模块的职责和对外接口每个关键函数写清楚输入输出和边界条件每个配置项写清楚含义和默认值。不需要长篇大论关键是让读的人能快速理解这个模块是干什么的、怎么用。def preprocess(raw: dict) - np.ndarray: 将原始输入转换为模型可用的特征向量。 Args: raw: 包含 age(int), income(float), category(str) 的字典 Returns: shape 为 (1, 10) 的 float32 数组已做归一化和独热编码 Raises: ValueError: 当 category 不在已知类别中时抛出 ...这种文档写起来不费劲但读起来非常省时间。工程效率的提升往往就藏在这些不起眼的习惯里。8. 最后分享几个我踩过的坑和对应的解法第一个坑是ONNX Runtime的线程数配置。默认情况下ONNX Runtime会使用所有可用的CPU核心但在容器环境里它可能读不到正确的核心数导致创建过多线程反而拖慢推理。解法是显式设置intra_op_num_threads和inter_op_num_threads一般设成物理核心数的一半到全部。第二个坑是FastAPI的异步阻塞。如果你在async def路由里调用了同步的阻塞函数比如session.run整个事件循环会被卡住其他请求全部排队。解法是用run_in_executor把阻塞调用放到线程池里执行或者直接用同步的def路由FastAPI会自动用线程池处理。第三个坑是日志级别设置不当导致性能下降。DEBUG级别的日志在推理路径上会打印大量信息IO开销很大。生产环境一定要把日志级别设成INFO或WARNING而且推理路径上不要打日志只在请求入口和出口打点。第四个坑是模型文件权限。容器里加载模型时如果模型文件的权限不对ONNX Runtime会报一个非常模糊的错误排查起来很费劲。解法是确保模型文件对运行用户可读而且路径不要有中文或特殊字符。这些坑看起来都是小事但每一个都曾让我花掉半天甚至一天的时间。AI工程的难点不在于算法有多复杂而在于这些琐碎的工程细节叠加起来形成了一个巨大的复杂度。从零构建的意义就是让你有机会把这些细节一个个吃透而不是被某个黑盒框架挡住视线。
从零手搓AI工程：告别调包，掌握核心构建与部署
1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调一下API跑通一个Demo然后发个朋友圈说“今天又搞定了一个AI项目”。我早期也这么干过结果到了真实业务场景里模型效果不稳定、推理延迟忽高忽低、数据管道三天两头断流排查问题的时候连从哪下手都不知道。后来我才意识到调包只能让你跑起来但从零构建才能让你真正掌控它。“ai-engineering-from-scratch”这个标题核心不在“AI”而在“from scratch”。它强调的不是让你去发明一个新的Transformer架构而是让你从最底层的工程视角把AI系统拆开来看数据怎么流进来、特征怎么处理、模型怎么加载、推理怎么调度、服务怎么暴露、监控怎么做。这套东西才是AI工程和“调个API”之间的本质区别。这篇文章适合谁看如果你已经会用Python写点脚本对机器学习有基本概念但一提到“工程化”就头疼那这篇内容就是为你准备的。我会按照一个真实项目的落地路径从环境搭建、数据管道、模型封装、服务部署到性能调优把每个环节的“为什么”和“怎么做”都讲清楚。所有代码和配置都是可以直接抄作业的但更重要的是你要理解每一步背后的工程逻辑。提示本文不会涉及任何特定云厂商的绑定操作所有方案都基于开源工具和通用工程实践确保你在任何环境下都能复现。2. 环境与依赖别让版本冲突毁掉你的第一周2.1 为什么我坚持用虚拟环境而不是全局安装我见过太多人拿到项目第一件事就是pip install -r requirements.txt然后跑着跑着发现某个包版本冲突整个环境崩掉最后只能重装系统。AI工程涉及的工具链特别长数值计算、深度学习框架、数据处理、服务框架、监控工具每个都有自己的依赖树。全局安装等于把所有鸡蛋放在一个篮子里一旦冲突全盘皆输。我的做法是每个项目一个独立的虚拟环境而且用conda而不是venv。原因很简单AI工程里很多底层库比如BLAS、CUDA相关的运行时用conda管理起来更省心尤其是当你需要在同一台机器上切换不同版本的深度学习框架时。下面是我常用的环境初始化流程# 创建独立环境指定Python版本 conda create -n ai-eng python3.10 -y conda activate ai-eng # 先装底层数值库再装上层框架 conda install numpy1.24 pandas2.0 scipy -y pip install torch2.1.0 --index-url https://download.pytorch.org/whl/cpu pip install fastapi uvicorn pydantic scikit-learn注意这里的顺序先conda装底层再pip装上层。因为conda的依赖解析器对二进制兼容性处理得更好而pip在纯Python包上更灵活。如果你反过来先用pip装了numpy再用conda装scipyconda可能会试图替换掉pip装的numpy导致环境混乱。2.2 依赖锁定的正确姿势requirements.txt只写包名和版本号是不够的。AI工程里一个包的间接依赖可能就有几十个而且不同平台Linux、macOS、Windows的解析结果还不一样。我推荐用pip-tools来生成锁定文件pip install pip-tools pip-compile requirements.in --output-file requirements.txt --generate-hashes pip-sync requirements.txt--generate-hashes会为每个包生成哈希校验值确保你安装的包没有被篡改。pip-sync则会严格按照锁定文件同步环境多装的包会被卸载少装的会补上。这套流程在团队协作里特别重要因为“在我机器上能跑”这句话90%的情况都是依赖版本不一致导致的。注意如果你在用GPU环境CUDA版本和深度学习框架版本的对应关系一定要查官方兼容性矩阵。我踩过最坑的一次是torch版本和cuda版本差了一个小版本号结果训练能跑推理直接段错误排查了整整两天。3. 数据管道AI工程里最容易被低估的脏活累活3.1 从原始数据到模型输入中间到底有多少步很多人以为数据管道就是“读CSV扔给模型”。真实项目里从原始数据到模型能用的张量中间至少隔着这些步骤数据采集、格式解析、缺失值处理、异常值检测、特征编码、归一化、数据集划分、批处理封装。每一步都有坑而且这些坑在Demo阶段完全不会暴露只有数据量上来、数据分布变化之后才会集中爆发。我习惯把数据管道拆成三个独立的层采集层、处理层、供给层。采集层负责从各种来源数据库、日志文件、消息队列把数据拉过来统一转成一种中间格式我一般用Parquet列式存储、压缩率高、读取快。处理层负责清洗、转换、特征工程输出的是模型可以直接消费的数值矩阵。供给层负责批处理、打乱、预取把数据高效地喂给训练或推理循环。这种分层的好处是每一层都可以独立测试和替换。比如采集层从MySQL换成Kafka处理层完全不用动处理层从pandas换成Spark供给层也不用改。工程化的核心思想就是解耦让变化的影响范围尽可能小。3.2 用PyTorch Dataset和DataLoader封装数据供给供给层我强烈建议用PyTorch的Dataset和DataLoader即使你用的不是PyTorch训练也可以借用这套抽象。下面是一个我常用的模板from torch.utils.data import Dataset, DataLoader import numpy as np class TabularDataset(Dataset): def __init__(self, features, labels, transformNone): self.features features.astype(np.float32) self.labels labels.astype(np.int64) self.transform transform def __len__(self): return len(self.labels) def __getitem__(self, idx): x self.features[idx] y self.labels[idx] if self.transform: x self.transform(x) return x, y # 使用示例 dataset TabularDataset(X_train, y_train) loader DataLoader( dataset, batch_size256, shuffleTrue, num_workers4, pin_memoryTrue, prefetch_factor2 )这里有几个参数值得展开说。num_workers4表示用4个子进程并行加载数据这个值不是越大越好一般设置为CPU核心数的50%到75%。设太大反而会因为进程切换开销导致吞吐下降。pin_memoryTrue会把数据锁在内存页里加速CPU到GPU的传输但只在GPU训练时有意义。prefetch_factor2表示每个worker预取2个batch这样GPU在计算当前batch时CPU已经在准备后面的数据了能有效避免GPU空转。实操心得如果你发现训练时GPU利用率忽高忽低大概率是数据供给跟不上。先用nvidia-smi看GPU利用率如果经常掉到0%就把num_workers调大或者检查__getitem__里有没有耗时的IO操作。我一般会把所有预处理都在__init__里做完__getitem__只做最轻量的索引和转换。3.3 数据版本管理别让“这次跑的结果和上次不一样”成为常态AI工程和传统软件工程最大的区别之一就是数据也是代码。你改了数据模型效果就变了但如果你不记录数据版本根本不知道是哪次改动导致的。我见过太多团队模型效果波动了大家互相甩锅最后发现是某个人偷偷更新了训练数据。我的做法是用DVCData Version Control来管理数据和模型文件。它和Git配合使用Git管代码DVC管数据。基本流程是dvc init dvc add data/raw/train.parquet git add data/raw/train.parquet.dvc data/raw/.gitignore git commit -m add training data v1dvc add会生成一个.dvc文件里面记录了数据的哈希值。数据本身存在本地或远程存储里Git仓库里只保留这个轻量级的指针文件。这样每次你切换Git分支对应的数据版本也会自动切换。没有数据版本管理的AI项目就像没有Git的代码项目迟早会乱成一锅粥。4. 模型封装从notebook到可复用的工程模块4.1 为什么notebook里的模型不能直接上生产notebook适合探索和实验但它的执行顺序是隐式的状态是全局的依赖是混乱的。你在notebook里定义了一个模型类改了三次最后一次跑通了但前两次的中间状态还留在内存里。这种代码拿到生产环境换一台机器、换一个执行顺序结果就可能完全不同。我的原则是任何要上生产的模型必须封装成独立的Python模块。这个模块对外只暴露两个接口train和predict。内部怎么实现外部不需要知道。下面是一个标准的模型封装模板# model.py import torch import torch.nn as nn from pathlib import Path class MLP(nn.Module): def __init__(self, input_dim, hidden_dim, output_dim): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden_dim, hidden_dim // 2), nn.ReLU(), nn.Linear(hidden_dim // 2, output_dim) ) def forward(self, x): return self.net(x) class ModelWrapper: def __init__(self, config): self.config config self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.model MLP( config[input_dim], config[hidden_dim], config[output_dim] ).to(self.device) def train(self, train_loader, val_loader, epochs10): optimizer torch.optim.Adam(self.model.parameters(), lrself.config[lr]) criterion nn.CrossEntropyLoss() best_val_loss float(inf) for epoch in range(epochs): self.model.train() for x, y in train_loader: x, y x.to(self.device), y.to(self.device) optimizer.zero_grad() loss criterion(self.model(x), y) loss.backward() optimizer.step() val_loss self._evaluate(val_loader, criterion) if val_loss best_val_loss: best_val_loss val_loss self._save_checkpoint() def predict(self, x): self.model.eval() with torch.no_grad(): x torch.tensor(x, dtypetorch.float32).to(self.device) logits self.model(x) return torch.softmax(logits, dim-1).cpu().numpy() def _evaluate(self, loader, criterion): self.model.eval() total_loss 0.0 with torch.no_grad(): for x, y in loader: x, y x.to(self.device), y.to(self.device) total_loss criterion(self.model(x), y).item() return total_loss / len(loader) def _save_checkpoint(self): path Path(self.config[checkpoint_dir]) path.mkdir(parentsTrue, exist_okTrue) torch.save(self.model.state_dict(), path / best.pt)这个封装的关键点在于配置和代码分离。所有超参数、路径、设备选择都通过config字典传入代码本身不硬编码任何环境相关的值。这样同一份代码可以在开发机、测试环境、生产环境无缝切换只需要换配置就行。4.2 模型序列化pickle、torch.save还是ONNX模型训练完要保存保存格式的选择直接影响后续的部署灵活性。我对比过几种常见方案格式优点缺点适用场景pickle简单直接Python原生版本兼容性差有安全风险临时实验torch.savePyTorch官方支持完整状态绑定PyTorch跨框架难PyTorch内部使用ONNX跨框架、跨平台推理优化好转换可能丢失自定义算子生产部署TorchScript脱离Python运行时性能好调试困难动态逻辑受限高性能推理我的建议是训练阶段用torch.save保存检查点部署阶段转成ONNX或TorchScript。ONNX的好处是你可以用ONNX Runtime来推理它比原生PyTorch推理快不少而且不依赖Python环境可以嵌入到C、Java等服务里。转换过程也不复杂import torch.onnx dummy_input torch.randn(1, config[input_dim]).to(device) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version14 )dynamic_axes这个参数很关键它允许模型接受可变长度的batch。如果不设置导出的ONNX模型只能处理固定batch size生产环境里会非常受限。踩坑记录ONNX转换时如果模型里有if分支或者循环转换可能会失败或者行为不一致。我遇到过一次模型里有个根据输入长度动态选择路径的逻辑转ONNX后直接报错。后来把逻辑改成用torch.where实现才转换成功。所以模型设计阶段就要考虑部署友好性别等到转换时才返工。5. 服务化与性能调优让模型真正跑起来5.1 用FastAPI把模型包装成HTTP服务模型封装好了下一步是把它暴露成服务。我选FastAPI的原因很直接异步支持好、自动生成API文档、类型校验强。下面是一个完整的服务模板# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import numpy as np import onnxruntime as ort app FastAPI(titleAI Model Service) class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): probabilities: list[float] predicted_class: int # 全局加载模型避免每次请求都重新加载 session ort.InferenceSession(model.onnx) app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): try: input_array np.array([request.features], dtypenp.float32) outputs session.run(None, {input: input_array}) probs outputs[0][0].tolist() pred_class int(np.argmax(probs)) return PredictResponse(probabilitiesprobs, predicted_classpred_class) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok}启动命令是uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4。--workers 4表示启动4个进程每个进程独立加载模型。这里有个坑如果你的模型很大比如几个GB4个worker会占4倍内存。这时候要么减少worker数量要么用共享内存的方式加载模型。ONNX Runtime支持内存映射多个进程可以共享同一份模型权重能省不少内存。5.2 批处理与动态批处理提升吞吐的关键单条推理的吞吐量是很低的因为每次请求都要走一遍完整的计算图GPU利用率极低。批处理是提升吞吐最直接的手段。但生产环境的请求是零散到来的你不能等凑够一个batch再处理那样延迟会爆炸。这时候就需要动态批处理服务端维护一个请求队列每隔几毫秒或者队列达到一定长度就把当前队列里的请求打包成一个batch一起推理。我用asyncio实现了一个简单的动态批处理逻辑import asyncio from collections import deque class DynamicBatcher: 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() self.lock asyncio.Lock() async def add_request(self, features): future asyncio.Future() async with self.lock: self.queue.append((features, future)) if len(self.queue) self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch list(self.queue) self.queue.clear() features np.array([item[0] for item in batch], dtypenp.float32) outputs session.run(None, {input: features}) for i, (_, future) in enumerate(batch): future.set_result(outputs[0][i].tolist()) async def run_loop(self): while True: await asyncio.sleep(self.max_wait) async with self.lock: if self.queue: await self._process_batch()这个逻辑的核心是请求进来先挂起等凑够batch或者超时后再统一处理。max_wait_ms设成10毫秒意味着最坏情况下请求会多等10毫秒但吞吐量能提升几倍甚至十几倍。这个权衡在生产环境里通常是值得的。5.3 监控与日志别等出事了才想起来看服务上线只是开始真正的挑战在于你怎么知道它现在是不是正常的。我见过太多项目服务跑起来了但没有任何监控直到用户投诉才发现模型输出全是NaN。AI服务的监控比普通服务多几个维度延迟P50、P95、P99分位数不能只看平均值吞吐每秒请求数、每秒处理的样本数错误率HTTP错误、推理异常、超时模型指标输入分布漂移、输出置信度分布、预测类别分布资源CPU、内存、GPU利用率、显存占用我用prometheus_client来暴露指标用Grafana来展示。关键代码是在推理前后打点from prometheus_client import Histogram, Counter INFERENCE_LATENCY Histogram( inference_latency_seconds, Model inference latency, buckets[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0] ) PREDICTION_COUNTER Counter( prediction_total, Total predictions, [predicted_class] ) app.post(/predict) async def predict(request: PredictRequest): with INFERENCE_LATENCY.time(): result await batcher.add_request(request.features) PREDICTION_COUNTER.labels(predicted_classstr(np.argmax(result))).inc() return PredictResponse(probabilitiesresult, predicted_classint(np.argmax(result)))输入分布漂移是AI服务特有的监控项。模型训练时的数据分布和线上推理时的数据分布可能不一致这种不一致会导致模型效果悄悄下降但不会报任何错误。我的做法是定期计算线上输入特征的均值和方差和训练集对比偏差超过阈值就告警。这个逻辑可以放在一个定时任务里不需要实时计算。实操心得监控指标不要贪多先覆盖延迟、错误率、吞吐这三个最基本的再逐步加模型相关的。我见过一个团队监控面板上几十个指标结果真出问题时没人知道该看哪个。监控的目的是快速定位问题不是展示技术实力。6. 从单机到可扩展当请求量涨上来之后6.1 水平扩展的前提无状态化单机服务总有上限请求量涨上来之后必须扩展。扩展的第一步是让服务无状态。什么叫无状态就是任何一个请求打到任何一个实例上结果都一样实例本身不保存任何请求相关的数据。模型权重是只读的可以每个实例加载一份请求队列是临时的处理完就销毁日志和监控数据发到外部系统。做到这些你就可以直接加机器前面挂个负载均衡请求量翻倍就加一倍实例。但这里有个问题动态批处理是有状态的。每个实例维护自己的请求队列负载均衡如果按请求轮询可能导致某些实例的队列很长某些很短。解决方案是用一致性哈希或者最少连接数策略让请求尽量均匀分布。更彻底的做法是把批处理逻辑抽出来做成一个独立的批处理服务所有推理实例从队列里拉取batch。但这会引入额外的网络跳数和复杂度除非请求量真的很大否则单机动态批处理已经够用了。6.2 模型热更新不重启服务换模型模型迭代是常态但每次换模型都重启服务会导致请求中断。热更新的目标是在不中断服务的前提下把新模型加载进来逐步切换流量。我的做法是维护两个模型槽位active和standby。新模型加载到standby预热完成后原子性地把active指针指向新模型旧模型等所有进行中的请求处理完再释放。class ModelManager: def __init__(self, model_path): self.active ort.InferenceSession(model_path) self.standby None self.lock threading.Lock() def load_new_model(self, new_path): new_session ort.InferenceSession(new_path) # 预热跑几次推理让ONNX Runtime完成图优化 dummy np.random.randn(1, 10).astype(np.float32) for _ in range(5): new_session.run(None, {input: dummy}) with self.lock: self.standby new_session def switch(self): with self.lock: if self.standby is not None: self.active, self.standby self.standby, None def predict(self, features): with self.lock: session self.active return session.run(None, {input: features})预热这一步很关键。ONNX Runtime在第一次推理时会做图优化、内存分配等操作如果不预热第一个请求的延迟可能是正常值的几十倍。我一般会跑5到10次预热推理确保延迟稳定后再切换流量。6.3 容量规划到底需要多少资源容量规划不是拍脑袋决定的需要基于实测数据。我的方法是先测单实例的极限吞吐再根据目标QPS计算实例数最后留30%的余量。测极限吞吐时用locust或者wrk压测逐步增加并发数观察延迟和错误率的变化。当P99延迟超过可接受阈值或者错误率开始上升时就是单实例的极限。举个例子假设你测出来单实例在P99延迟100毫秒以内能扛住200 QPS目标QPS是1000那至少需要5个实例。但考虑到流量波动和实例故障实际部署7个实例5个主力2个冗余。这个计算过程要定期重做因为模型更新、数据变化、依赖升级都可能改变单实例的性能特征。注意压测时一定要用真实的数据分布不要用随机生成的假数据。我踩过一次坑压测时用np.random.randn生成输入结果模型里有个分支对特定范围的输入会走更复杂的计算路径真实数据下延迟比压测结果高了3倍。假数据只能测出代码路径测不出真实性能。7. 一些让我少走弯路的工程习惯7.1 配置管理别把密码和路径写死在代码里我早期写代码数据库密码、模型路径、API密钥全是硬编码。结果代码传到Git上密码泄露换台机器跑路径全不对。后来我强制自己用环境变量加配置文件的方式管理配置。敏感信息走环境变量非敏感的默认值走配置文件配置文件用YAML格式支持多环境覆盖。# config/base.yaml model: input_dim: 128 hidden_dim: 256 output_dim: 10 checkpoint_dir: ./checkpoints server: host: 0.0.0.0 port: 8000 workers: 4import os import yaml def load_config(envdev): with open(config/base.yaml) as f: config yaml.safe_load(f) env_config_path fconfig/{env}.yaml if os.path.exists(env_config_path): with open(env_config_path) as f: env_config yaml.safe_load(f) config deep_merge(config, env_config) # 环境变量覆盖 if os.getenv(MODEL_CHECKPOINT_DIR): config[model][checkpoint_dir] os.getenv(MODEL_CHECKPOINT_DIR) return config这套机制的好处是开发环境用开发配置生产环境用生产配置代码完全一样。部署时只需要设置环境变量不需要改任何代码。7.2 测试AI项目也需要单元测试很多人觉得AI项目没法写测试因为模型输出是不确定的。但工程代码是可以测试的。数据预处理函数、特征编码逻辑、批处理队列、配置加载这些都应该有单元测试。模型本身可以用小规模数据做冒烟测试确保训练能跑通、推理不报错、输出形状正确。def test_preprocess(): raw {age: 25, income: 50000, category: A} result preprocess(raw) assert result.shape (1, 10) assert not np.isnan(result).any() def test_model_forward(): model MLP(input_dim10, hidden_dim32, output_dim3) x torch.randn(4, 10) out model(x) assert out.shape (4, 3)这些测试跑起来很快但能在早期发现大部分低级错误。我宁愿花半小时写测试也不愿意花两天排查一个本可以避免的bug。7.3 文档写给三个月后的自己AI工程项目里最容易被忽略的就是文档。但三个月后你回头看自己的代码如果没有文档你大概率会想“这写的什么鬼”。我的文档习惯是每个模块顶部写清楚这个模块的职责和对外接口每个关键函数写清楚输入输出和边界条件每个配置项写清楚含义和默认值。不需要长篇大论关键是让读的人能快速理解这个模块是干什么的、怎么用。def preprocess(raw: dict) - np.ndarray: 将原始输入转换为模型可用的特征向量。 Args: raw: 包含 age(int), income(float), category(str) 的字典 Returns: shape 为 (1, 10) 的 float32 数组已做归一化和独热编码 Raises: ValueError: 当 category 不在已知类别中时抛出 ...这种文档写起来不费劲但读起来非常省时间。工程效率的提升往往就藏在这些不起眼的习惯里。8. 最后分享几个我踩过的坑和对应的解法第一个坑是ONNX Runtime的线程数配置。默认情况下ONNX Runtime会使用所有可用的CPU核心但在容器环境里它可能读不到正确的核心数导致创建过多线程反而拖慢推理。解法是显式设置intra_op_num_threads和inter_op_num_threads一般设成物理核心数的一半到全部。第二个坑是FastAPI的异步阻塞。如果你在async def路由里调用了同步的阻塞函数比如session.run整个事件循环会被卡住其他请求全部排队。解法是用run_in_executor把阻塞调用放到线程池里执行或者直接用同步的def路由FastAPI会自动用线程池处理。第三个坑是日志级别设置不当导致性能下降。DEBUG级别的日志在推理路径上会打印大量信息IO开销很大。生产环境一定要把日志级别设成INFO或WARNING而且推理路径上不要打日志只在请求入口和出口打点。第四个坑是模型文件权限。容器里加载模型时如果模型文件的权限不对ONNX Runtime会报一个非常模糊的错误排查起来很费劲。解法是确保模型文件对运行用户可读而且路径不要有中文或特殊字符。这些坑看起来都是小事但每一个都曾让我花掉半天甚至一天的时间。AI工程的难点不在于算法有多复杂而在于这些琐碎的工程细节叠加起来形成了一个巨大的复杂度。从零构建的意义就是让你有机会把这些细节一个个吃透而不是被某个黑盒框架挡住视线。
