1. 为什么我要从零手搓一套AI工程化流程第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是“又一个教程仓库”而是过去两年踩过的那些坑模型在 notebook 里跑得飞起一上生产就各种超时离线评测指标漂亮得不行线上 A/B 测试却毫无提升同事交接时甩过来一句“环境在我机器上是好的”然后你花两天时间对齐 CUDA、Python 和依赖版本。这些问题不是模型能力问题而是工程化能力问题。ai-engineering-from-scratch这个项目标题核心讲的就是一件事不依赖任何现成的 MLOps 平台、不套用大厂的开源全家桶从最基础的组件开始把一套 AI 系统从数据到推理再到监控的完整链路亲手搭起来。它解决的是“会用模型但不会做工程”这个断层适合那些已经能跑通model.fit()和model.predict()但一遇到并发、版本管理、回滚、成本控制就发怵的开发者。我写这篇东西就是想把我自己从零搭这套流程时的思路、选型、参数计算和踩坑记录完整摊开让你能直接抄作业也能理解每一步为什么这么做。这套流程的价值不在于“造轮子”而在于你亲手造过一遍轮子之后再用任何现成框架都能一眼看穿它在哪个抽象层偷了懒。下面我按整体设计、核心细节、实操实现、问题排查四个大块来讲中间会穿插大量我实际跑过的参数和配置。2. 整体架构设计与技术选型思路2.1 从“能跑”到“能扛”的四个断层很多人以为 AI 工程化就是加个 Docker、写个 FastAPI。我一开始也这么想直到线上出了三次事故才明白从 notebook 到生产之间横着四个断层。第一个断层是环境断层。你的训练环境有 GPU、有特定版本的 cuDNN推理环境可能只有 CPU或者 GPU 型号不同。第二个断层是数据断层。训练时数据是静态的 parquet 文件线上数据是流式的、有缺失值、有分布漂移。第三个断层是性能断层。单条推理 50ms 看着没问题但 QPS 上到 200 时Python GIL 和内存拷贝会把延迟推到 2 秒以上。第四个断层是运维断层。模型更新后效果变差你想回滚却发现旧版本权重没存、旧版本依赖没锁。ai-engineering-from-scratch的思路就是针对这四个断层一层一层用最朴素的工具去填。我不建议一上来就上 Kubernetes 和 Feature Store那会让你在还没理解问题之前就被复杂度淹没。我的选型原则是每个组件只解决一个断层组件之间用最笨但最可靠的方式连接。2.2 为什么我选择“手写编排”而不是现成平台市面上有大量 MLOps 平台从实验追踪到模型注册到在线服务一应俱全。我试过其中几个最后回到手写编排原因有三个。第一调试成本。平台出问题时你面对的是一个黑盒。日志告诉你“pipeline failed”但到底是数据读取超时还是模型加载 OOM你得翻三层抽象才能定位。手写编排虽然土但每一行都是你自己的报错栈直接指向问题。第二成本可控。很多平台按推理次数或节点数收费小规模场景下比自己搭贵一个数量级。我算过一笔账一个中等规模的推荐模型用平台托管每月成本大约是自己搭的 3 到 5 倍而且流量突增时账单不可预测。第三知识沉淀。手写一遍之后你对数据版本、模型版本、依赖版本三者的对应关系会有肌肉记忆。这种理解在换任何平台时都能迁移而平台操作技能换一家就作废。当然手写编排不是让你重复造所有轮子。我的做法是数据版本用 DVC实验追踪用 MLflow服务用 FastAPI ONNX Runtime监控用 Prometheus 自写指标。这些工具都是轻量、可替换、社区成熟的组合起来就是一套完整的从零方案。2.3 分层架构把每一层做成可替换的积木我把整套系统分成五层每层之间通过明确的接口通信这样任何一层出问题或需要升级都不会牵动全局。层级职责我的选型替换成本数据层数据版本、清洗、特征存储DVC Parquet Redis低训练层实验追踪、超参管理、模型产出MLflow PyTorch中制品层模型格式转换、版本注册ONNX 自建注册表低服务层推理 API、批处理、并发控制FastAPI ONNX Runtime中监控层延迟、吞吐、漂移、告警Prometheus Grafana低这个分层的关键在于制品层。很多人忽略这一层训练完直接把.pt文件丢给服务层结果服务层被 PyTorch 的依赖和加载时间拖死。我坚持在制品层把模型转成 ONNX原因后面会详细讲。3. 核心细节解析与实操要点3.1 数据版本管理为什么不能用 Git 管数据数据版本是 AI 工程化的地基。我见过太多团队用 Git LFS 管数据集结果仓库膨胀到几十 GBclone 一次要半小时。ai-engineering-from-scratch的做法是用 DVC 把数据指针和实际存储分离。具体操作是数据文件本身放在对象存储或共享磁盘Git 里只存一个.dvc文件里面记录了数据的哈希值。这样切换数据版本就像切换 Git 分支一样快。我实测下来一个 20GB 的数据集用 DVC 切换版本耗时不到 3 秒而重新下载要 10 分钟以上。注意DVC 的缓存目录默认在项目内会占用大量本地空间。建议在初始化时用dvc cache dir把缓存指向大容量磁盘否则你的系统盘很快会满。数据清洗环节我的经验是把清洗逻辑写成纯函数并版本化。很多人清洗数据是在 notebook 里一步步跑的下次换数据就重跑一遍结果两次清洗逻辑不一致。我的做法是把每个清洗步骤写成一个函数输入输出都是 DataFrame然后用一个pipeline.yaml描述步骤顺序。这样清洗逻辑本身也进了版本控制出问题可以精确回滚到某一步。3.2 模型格式转换ONNX 到底省了什么训练用 PyTorch服务用 ONNX Runtime这是我在多个项目里验证过的最稳组合。为什么不在服务端直接用 PyTorch三个原因。第一加载速度。PyTorch 加载一个 300MB 的模型大约需要 4 到 6 秒ONNX Runtime 加载同样模型只要 0.8 到 1.2 秒。在需要快速扩缩容的场景下这个差距直接决定冷启动体验。第二依赖体积。PyTorch 的 Docker 镜像动辄 2GB 以上ONNX Runtime 的 CPU 版本只有 200MB 左右。镜像小意味着拉取快、攻击面小、成本低。第三推理性能。在 CPU 上ONNX Runtime 通过图优化和算子融合通常比 PyTorch 原生推理快 1.5 到 2 倍。我实测过一个 BERT-base 模型PyTorch 单条推理 45msONNX Runtime 只要 23ms。转换过程本身不复杂但有几个坑必须注意。导出时要用torch.onnx.export并且把dynamic_axes设好否则 batch size 会被固定死。我一般这样写torch.onnx.export( model, dummy_input, model.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{ input_ids: {0: batch, 1: sequence}, attention_mask: {0: batch, 1: sequence}, logits: {0: batch} }, opset_version14 )opset_version我固定在 14因为 15 以上在某些旧版 ONNX Runtime 上兼容性不好而 13 以下缺少一些融合优化。导出后一定要用onnxruntime跑一遍验证对比 PyTorch 和 ONNX 的输出差异我一般要求最大绝对误差小于 1e-4。3.3 服务层并发控制别让 GIL 成为瓶颈FastAPI 本身是异步的但如果你在异步函数里直接调用同步的推理代码整个事件循环会被阻塞。我见过最典型的错误写法是在async def predict里直接调session.run()结果 QPS 上到 50 就卡死。正确的做法是用线程池把推理调用包起来。ONNX Runtime 的InferenceSession.run会释放 GIL所以多线程能真正并行。我的配置是from concurrent.futures import ThreadPoolExecutor import onnxruntime as ort session ort.InferenceSession( model.onnx, providers[CPUExecutionProvider], sess_optionsort.SessionOptions() ) session.intra_op_num_threads 4 session.inter_op_num_threads 2 executor ThreadPoolExecutor(max_workers8) async def predict(inputs): loop asyncio.get_event_loop() return await loop.run_in_executor(executor, _run_inference, inputs)这里intra_op_num_threads控制单个算子内部的并行度inter_op_num_threads控制算子之间的并行度。我的经验是CPU 核数在 8 到 16 之间时intra_op设为核数的一半inter_op设为 2整体吞吐最高。线程池大小设为核数的 1 到 1.5 倍超过这个数反而因为上下文切换导致延迟上升。提示如果你的服务同时处理多种模型给每个模型单独配一个线程池避免一个慢模型拖垮所有请求。3.4 监控指标只看延迟是不够的监控层我见过最常见的误区是只监控 API 延迟和错误率。这两个指标只能告诉你“服务活着”不能告诉你“服务是否在退化”。AI 系统特有的退化包括数据漂移、预测分布偏移、特征缺失率上升。我的监控指标分三组。第一组是系统指标P50/P95/P99 延迟、QPS、错误率、CPU/内存使用率。第二组是模型指标预测置信度分布、各类别预测占比、输入特征缺失率。第三组是业务指标点击率、转化率等下游信号。数据漂移检测我用的是 PSIPopulation Stability Index计算方式是def calculate_psi(expected, actual, buckets10): breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) expected_perc np.histogram(expected, breakpoints)[0] / len(expected) actual_perc np.histogram(actual, breakpoints)[0] / len(actual) expected_perc np.clip(expected_perc, 1e-6, None) actual_perc np.clip(actual_perc, 1e-6, None) return np.sum((actual_perc - expected_perc) * np.log(actual_perc / expected_perc))PSI 小于 0.1 表示分布稳定0.1 到 0.25 表示有轻微漂移需要关注大于 0.25 表示显著漂移需要告警。这个阈值不是绝对的我一般会结合业务影响调整。4. 完整实操流程与关键环节实现4.1 环境准备从裸机到可复现环境我假设你有一台干净的 Linux 机器有 GPU 更好没有也能跑通 CPU 流程。第一步是装 Python 和虚拟环境管理工具。我不用 conda因为 conda 的依赖解析在跨平台时经常出问题改用pyenvvenv。# 安装 pyenv curl https://pyenv.run | bash # 安装 Python 3.10 pyenv install 3.10.13 pyenv local 3.10.13 # 创建虚拟环境 python -m venv .venv source .venv/bin/activate为什么固定 Python 3.10因为 3.11 在某些 ONNX Runtime 版本上有兼容性问题3.9 又缺少一些类型语法。3.10 是我实测最稳的版本。依赖管理我用pip-tools把直接依赖写在requirements.in然后编译出锁定的requirements.txt。这样每次安装的版本完全一致。pip install pip-tools pip-compile requirements.in -o requirements.txt pip-sync requirements.txtrequirements.in里我一般只写这些torch2.1.0 onnx1.15.0 onnxruntime1.16.3 fastapi0.104.1 uvicorn0.24.0 mlflow2.8.0 dvc3.30.0 prometheus-client0.18.04.2 数据流水线搭建从原始数据到特征假设我们有一个用户行为预测任务原始数据是 CSV 格式的日志。第一步是用 DVC 初始化数据版本控制。dvc init dvc remote add -d storage /mnt/data/dvc-storage dvc add data/raw/logs.csv git add data/raw/logs.csv.dvc .gitignore git commit -m add raw data然后写清洗脚本src/data/clean.py把原始日志转成特征表。我的清洗逻辑分四步去重、缺失值处理、异常值截断、特征编码。每一步都写成独立函数方便单测。def remove_duplicates(df, key_cols): return df.drop_duplicates(subsetkey_cols, keeplast) def fill_missing(df, num_cols, cat_cols): for col in num_cols: df[col] df[col].fillna(df[col].median()) for col in cat_cols: df[col] df[col].fillna(__missing__) return df def clip_outliers(df, num_cols, lower_q0.01, upper_q0.99): for col in num_cols: lo df[col].quantile(lower_q) hi df[col].quantile(upper_q) df[col] df[col].clip(lo, hi) return df清洗完的特征表用 Parquet 格式存储因为 Parquet 列式存储读取快、压缩率高。我实测一个 500 万行的特征表CSV 格式 1.2GBParquet 只要 180MB读取速度快 4 倍。4.3 训练与实验追踪让每次实验都可追溯训练脚本我用 MLflow 做追踪。关键是把超参、指标、模型文件都记录下来并且给每次运行打上数据版本标签。import mlflow mlflow.set_tracking_uri(http://localhost:5000) mlflow.set_experiment(user-behavior-prediction) with mlflow.start_run(): mlflow.log_params({ learning_rate: 1e-3, batch_size: 64, epochs: 10, data_version: v1.2 }) for epoch in range(epochs): train_loss train_one_epoch(model, loader) val_auc evaluate(model, val_loader) mlflow.log_metrics({ train_loss: train_loss, val_auc: val_auc }, stepepoch) mlflow.pytorch.log_model(model, model)这里data_version标签很重要。当线上效果变差时你可以通过这个标签快速定位是哪个数据版本训练的模型然后对比数据分布找出问题。训练完成后我把最佳模型导出为 ONNX并记录转换前后的输出差异。这个差异值也存进 MLflow作为模型质量的一部分。4.4 模型注册与版本管理自建轻量注册表MLflow 自带模型注册表但我更倾向于自建一个轻量注册表因为 MLflow 的注册表在跨团队协作时权限管理比较麻烦。我的做法是用一个 JSON 文件加对象存储。{ models: { user-behavior: { production: { version: v1.2.0, path: s3://models/user-behavior/v1.2.0/model.onnx, data_version: v1.2, metrics: {val_auc: 0.842}, created_at: 2024-01-15T10:30:00Z }, staging: { version: v1.3.0-rc1, path: s3://models/user-behavior/v1.3.0-rc1/model.onnx, data_version: v1.3, metrics: {val_auc: 0.851}, created_at: 2024-01-20T14:00:00Z } } } }服务层启动时读取这个 JSON根据环境变量决定加载 production 还是 staging 版本。回滚只需要改 JSON 里的版本号然后重启服务。这个方案土但极其可靠而且任何人都能看懂。4.5 服务部署从单机到多副本服务用 FastAPI 写启动命令是uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4--workers的数量我一般设为 CPU 核数。但注意如果每个 worker 都加载一份模型内存会成倍增长。对于大模型我改用单 worker 线程池的方案前面用 Nginx 做负载均衡。Nginx 配置的关键是超时和重试upstream inference { server 127.0.0.1:8000; keepalive 32; } location /predict { proxy_pass http://inference; proxy_read_timeout 5s; proxy_connect_timeout 1s; proxy_next_upstream error timeout; }proxy_read_timeout设为 5 秒因为我的 P99 延迟在 800ms 左右5 秒足够覆盖异常情况。keepalive设为 32 减少连接建立开销。4.6 监控告警从指标到行动Prometheus 抓取指标Grafana 展示告警规则写在 Prometheus 的 rules 文件里。我的核心告警规则有三条。第一条P95 延迟超过 1 秒持续 5 分钟。第二条错误率超过 1% 持续 3 分钟。第三条PSI 超过 0.25。groups: - name: inference rules: - alert: HighLatency expr: histogram_quantile(0.95, rate(inference_latency_bucket[5m])) 1 for: 5m labels: severity: warning annotations: summary: P95 latency above 1s - alert: HighErrorRate expr: rate(inference_errors_total[3m]) / rate(inference_requests_total[3m]) 0.01 for: 3m labels: severity: critical - alert: DataDrift expr: model_psi 0.25 for: 10m labels: severity: warning告警不是终点关键是告警后的行动。我的 runbook 里写清楚了每种告警的第一步排查动作延迟高先看 CPU 和线程池队列错误率高先看模型加载和输入格式漂移告警先看上游数据源。5. 常见问题与排查技巧实录5.1 模型加载慢导致冷启动超时这个问题我遇到过三次每次都是因为模型文件太大或者加载逻辑有问题。排查思路是先用time命令测加载耗时如果超过 3 秒就要优化。优化手段有三个。第一把模型转成 ONNX 并启用图优化。第二用onnxruntime的sess_options.optimized_model_filepath把优化后的模型存下来下次直接加载优化版。第三如果模型真的很大考虑用内存映射文件加载。sess_options ort.SessionOptions() sess_options.optimized_model_filepath model_optimized.onnx session ort.InferenceSession(model.onnx, sess_options)第一次加载会慢但之后加载model_optimized.onnx会快很多。我实测一个 500MB 的模型优化后加载时间从 6 秒降到 1.5 秒。5.2 推理结果与训练时不一致这是最隐蔽的问题通常由三个原因导致。第一预处理不一致。训练时用的归一化参数和线上不同。第二输入维度顺序不同。第三ONNX 转换时某些算子行为有差异。排查方法是逐层对比。把同一个样本分别喂给 PyTorch 和 ONNX对比每一层的输出。我一般从第一层开始找到第一个差异超过 1e-4 的层然后检查那个层的算子。注意ONNX 的Softmax和 PyTorch 的Softmax在数值稳定性上可能有细微差异通常不影响结果但如果你的阈值卡得很紧就要留意。5.3 内存泄漏导致服务崩溃FastAPI 服务跑几天后内存涨到几个 GB最后 OOM。这个问题我排查了两天最后发现是日志里存了完整的请求体而请求体里有大数组。解决方法是日志只记录元数据不记录原始数据。如果确实需要记录输入用于调试用采样比如每 1000 个请求记录一个并且把大数组截断。import logging logger logging.getLogger(__name__) def log_request(request_id, inputs): logger.info({ request_id: request_id, input_shape: inputs.shape, input_dtype: str(inputs.dtype), input_sample: inputs.flatten()[:5].tolist() })5.4 常见问题速查表现象可能原因排查动作解决手段冷启动超时模型加载慢测加载耗时ONNX 优化、内存映射结果不一致预处理差异逐层对比输出统一预处理代码内存泄漏日志存大对象看内存增长曲线日志采样、截断延迟突增线程池满看队列长度调大线程池、限流漂移告警上游数据变化对比特征分布重训、加特征校验版本回滚失败旧模型未存检查注册表强制保留最近 5 个版本5.5 我踩过的三个坑第一个坑是用 Git 管模型文件。一个 200MB 的模型提交了 10 次仓库变成 2GBclone 一次要 5 分钟。后来改用对象存储加版本号仓库干净了回滚也快了。第二个坑是在异步函数里做同步 IO。我在async def里直接读 Redis结果事件循环被阻塞QPS 上不去。后来所有 IO 都改成异步客户端或者用线程池包起来。第三个坑是监控只加不删。指标越加越多Prometheus 存储爆炸查询变慢。后来我定了规矩每个指标必须有明确的告警规则或看板用途否则不加。6. 后续扩展与个人体会这套从零搭的流程跑通之后扩展方向其实很多。比如把批处理推理单独拆成一个服务用消息队列解耦比如加一个 A/B 测试框架按用户 ID 分流到不同模型版本比如把特征计算下沉到 Flink 做实时特征。但我的建议是不要一次全上每加一个组件先问自己当前系统真的扛不住了吗如果答案是否定的就先不加。我个人在实际操作中的体会是AI 工程化最难的不是技术而是克制。克制住一上来就上全家桶的冲动克制住为了“先进”而引入复杂度的冲动。ai-engineering-from-scratch这个思路的核心价值就是让你在亲手搭过一遍之后清楚地知道每个组件的边界在哪里什么时候该用它什么时候该换掉它。这套流程我用了两年中间换过模型、换过硬件、换过团队但分层结构和核心接口一直没变这大概就是它最稳的地方。
从零手搓AI工程化流程：数据版本、ONNX推理与监控实战
1. 为什么我要从零手搓一套AI工程化流程第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是“又一个教程仓库”而是过去两年踩过的那些坑模型在 notebook 里跑得飞起一上生产就各种超时离线评测指标漂亮得不行线上 A/B 测试却毫无提升同事交接时甩过来一句“环境在我机器上是好的”然后你花两天时间对齐 CUDA、Python 和依赖版本。这些问题不是模型能力问题而是工程化能力问题。ai-engineering-from-scratch这个项目标题核心讲的就是一件事不依赖任何现成的 MLOps 平台、不套用大厂的开源全家桶从最基础的组件开始把一套 AI 系统从数据到推理再到监控的完整链路亲手搭起来。它解决的是“会用模型但不会做工程”这个断层适合那些已经能跑通model.fit()和model.predict()但一遇到并发、版本管理、回滚、成本控制就发怵的开发者。我写这篇东西就是想把我自己从零搭这套流程时的思路、选型、参数计算和踩坑记录完整摊开让你能直接抄作业也能理解每一步为什么这么做。这套流程的价值不在于“造轮子”而在于你亲手造过一遍轮子之后再用任何现成框架都能一眼看穿它在哪个抽象层偷了懒。下面我按整体设计、核心细节、实操实现、问题排查四个大块来讲中间会穿插大量我实际跑过的参数和配置。2. 整体架构设计与技术选型思路2.1 从“能跑”到“能扛”的四个断层很多人以为 AI 工程化就是加个 Docker、写个 FastAPI。我一开始也这么想直到线上出了三次事故才明白从 notebook 到生产之间横着四个断层。第一个断层是环境断层。你的训练环境有 GPU、有特定版本的 cuDNN推理环境可能只有 CPU或者 GPU 型号不同。第二个断层是数据断层。训练时数据是静态的 parquet 文件线上数据是流式的、有缺失值、有分布漂移。第三个断层是性能断层。单条推理 50ms 看着没问题但 QPS 上到 200 时Python GIL 和内存拷贝会把延迟推到 2 秒以上。第四个断层是运维断层。模型更新后效果变差你想回滚却发现旧版本权重没存、旧版本依赖没锁。ai-engineering-from-scratch的思路就是针对这四个断层一层一层用最朴素的工具去填。我不建议一上来就上 Kubernetes 和 Feature Store那会让你在还没理解问题之前就被复杂度淹没。我的选型原则是每个组件只解决一个断层组件之间用最笨但最可靠的方式连接。2.2 为什么我选择“手写编排”而不是现成平台市面上有大量 MLOps 平台从实验追踪到模型注册到在线服务一应俱全。我试过其中几个最后回到手写编排原因有三个。第一调试成本。平台出问题时你面对的是一个黑盒。日志告诉你“pipeline failed”但到底是数据读取超时还是模型加载 OOM你得翻三层抽象才能定位。手写编排虽然土但每一行都是你自己的报错栈直接指向问题。第二成本可控。很多平台按推理次数或节点数收费小规模场景下比自己搭贵一个数量级。我算过一笔账一个中等规模的推荐模型用平台托管每月成本大约是自己搭的 3 到 5 倍而且流量突增时账单不可预测。第三知识沉淀。手写一遍之后你对数据版本、模型版本、依赖版本三者的对应关系会有肌肉记忆。这种理解在换任何平台时都能迁移而平台操作技能换一家就作废。当然手写编排不是让你重复造所有轮子。我的做法是数据版本用 DVC实验追踪用 MLflow服务用 FastAPI ONNX Runtime监控用 Prometheus 自写指标。这些工具都是轻量、可替换、社区成熟的组合起来就是一套完整的从零方案。2.3 分层架构把每一层做成可替换的积木我把整套系统分成五层每层之间通过明确的接口通信这样任何一层出问题或需要升级都不会牵动全局。层级职责我的选型替换成本数据层数据版本、清洗、特征存储DVC Parquet Redis低训练层实验追踪、超参管理、模型产出MLflow PyTorch中制品层模型格式转换、版本注册ONNX 自建注册表低服务层推理 API、批处理、并发控制FastAPI ONNX Runtime中监控层延迟、吞吐、漂移、告警Prometheus Grafana低这个分层的关键在于制品层。很多人忽略这一层训练完直接把.pt文件丢给服务层结果服务层被 PyTorch 的依赖和加载时间拖死。我坚持在制品层把模型转成 ONNX原因后面会详细讲。3. 核心细节解析与实操要点3.1 数据版本管理为什么不能用 Git 管数据数据版本是 AI 工程化的地基。我见过太多团队用 Git LFS 管数据集结果仓库膨胀到几十 GBclone 一次要半小时。ai-engineering-from-scratch的做法是用 DVC 把数据指针和实际存储分离。具体操作是数据文件本身放在对象存储或共享磁盘Git 里只存一个.dvc文件里面记录了数据的哈希值。这样切换数据版本就像切换 Git 分支一样快。我实测下来一个 20GB 的数据集用 DVC 切换版本耗时不到 3 秒而重新下载要 10 分钟以上。注意DVC 的缓存目录默认在项目内会占用大量本地空间。建议在初始化时用dvc cache dir把缓存指向大容量磁盘否则你的系统盘很快会满。数据清洗环节我的经验是把清洗逻辑写成纯函数并版本化。很多人清洗数据是在 notebook 里一步步跑的下次换数据就重跑一遍结果两次清洗逻辑不一致。我的做法是把每个清洗步骤写成一个函数输入输出都是 DataFrame然后用一个pipeline.yaml描述步骤顺序。这样清洗逻辑本身也进了版本控制出问题可以精确回滚到某一步。3.2 模型格式转换ONNX 到底省了什么训练用 PyTorch服务用 ONNX Runtime这是我在多个项目里验证过的最稳组合。为什么不在服务端直接用 PyTorch三个原因。第一加载速度。PyTorch 加载一个 300MB 的模型大约需要 4 到 6 秒ONNX Runtime 加载同样模型只要 0.8 到 1.2 秒。在需要快速扩缩容的场景下这个差距直接决定冷启动体验。第二依赖体积。PyTorch 的 Docker 镜像动辄 2GB 以上ONNX Runtime 的 CPU 版本只有 200MB 左右。镜像小意味着拉取快、攻击面小、成本低。第三推理性能。在 CPU 上ONNX Runtime 通过图优化和算子融合通常比 PyTorch 原生推理快 1.5 到 2 倍。我实测过一个 BERT-base 模型PyTorch 单条推理 45msONNX Runtime 只要 23ms。转换过程本身不复杂但有几个坑必须注意。导出时要用torch.onnx.export并且把dynamic_axes设好否则 batch size 会被固定死。我一般这样写torch.onnx.export( model, dummy_input, model.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{ input_ids: {0: batch, 1: sequence}, attention_mask: {0: batch, 1: sequence}, logits: {0: batch} }, opset_version14 )opset_version我固定在 14因为 15 以上在某些旧版 ONNX Runtime 上兼容性不好而 13 以下缺少一些融合优化。导出后一定要用onnxruntime跑一遍验证对比 PyTorch 和 ONNX 的输出差异我一般要求最大绝对误差小于 1e-4。3.3 服务层并发控制别让 GIL 成为瓶颈FastAPI 本身是异步的但如果你在异步函数里直接调用同步的推理代码整个事件循环会被阻塞。我见过最典型的错误写法是在async def predict里直接调session.run()结果 QPS 上到 50 就卡死。正确的做法是用线程池把推理调用包起来。ONNX Runtime 的InferenceSession.run会释放 GIL所以多线程能真正并行。我的配置是from concurrent.futures import ThreadPoolExecutor import onnxruntime as ort session ort.InferenceSession( model.onnx, providers[CPUExecutionProvider], sess_optionsort.SessionOptions() ) session.intra_op_num_threads 4 session.inter_op_num_threads 2 executor ThreadPoolExecutor(max_workers8) async def predict(inputs): loop asyncio.get_event_loop() return await loop.run_in_executor(executor, _run_inference, inputs)这里intra_op_num_threads控制单个算子内部的并行度inter_op_num_threads控制算子之间的并行度。我的经验是CPU 核数在 8 到 16 之间时intra_op设为核数的一半inter_op设为 2整体吞吐最高。线程池大小设为核数的 1 到 1.5 倍超过这个数反而因为上下文切换导致延迟上升。提示如果你的服务同时处理多种模型给每个模型单独配一个线程池避免一个慢模型拖垮所有请求。3.4 监控指标只看延迟是不够的监控层我见过最常见的误区是只监控 API 延迟和错误率。这两个指标只能告诉你“服务活着”不能告诉你“服务是否在退化”。AI 系统特有的退化包括数据漂移、预测分布偏移、特征缺失率上升。我的监控指标分三组。第一组是系统指标P50/P95/P99 延迟、QPS、错误率、CPU/内存使用率。第二组是模型指标预测置信度分布、各类别预测占比、输入特征缺失率。第三组是业务指标点击率、转化率等下游信号。数据漂移检测我用的是 PSIPopulation Stability Index计算方式是def calculate_psi(expected, actual, buckets10): breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) expected_perc np.histogram(expected, breakpoints)[0] / len(expected) actual_perc np.histogram(actual, breakpoints)[0] / len(actual) expected_perc np.clip(expected_perc, 1e-6, None) actual_perc np.clip(actual_perc, 1e-6, None) return np.sum((actual_perc - expected_perc) * np.log(actual_perc / expected_perc))PSI 小于 0.1 表示分布稳定0.1 到 0.25 表示有轻微漂移需要关注大于 0.25 表示显著漂移需要告警。这个阈值不是绝对的我一般会结合业务影响调整。4. 完整实操流程与关键环节实现4.1 环境准备从裸机到可复现环境我假设你有一台干净的 Linux 机器有 GPU 更好没有也能跑通 CPU 流程。第一步是装 Python 和虚拟环境管理工具。我不用 conda因为 conda 的依赖解析在跨平台时经常出问题改用pyenvvenv。# 安装 pyenv curl https://pyenv.run | bash # 安装 Python 3.10 pyenv install 3.10.13 pyenv local 3.10.13 # 创建虚拟环境 python -m venv .venv source .venv/bin/activate为什么固定 Python 3.10因为 3.11 在某些 ONNX Runtime 版本上有兼容性问题3.9 又缺少一些类型语法。3.10 是我实测最稳的版本。依赖管理我用pip-tools把直接依赖写在requirements.in然后编译出锁定的requirements.txt。这样每次安装的版本完全一致。pip install pip-tools pip-compile requirements.in -o requirements.txt pip-sync requirements.txtrequirements.in里我一般只写这些torch2.1.0 onnx1.15.0 onnxruntime1.16.3 fastapi0.104.1 uvicorn0.24.0 mlflow2.8.0 dvc3.30.0 prometheus-client0.18.04.2 数据流水线搭建从原始数据到特征假设我们有一个用户行为预测任务原始数据是 CSV 格式的日志。第一步是用 DVC 初始化数据版本控制。dvc init dvc remote add -d storage /mnt/data/dvc-storage dvc add data/raw/logs.csv git add data/raw/logs.csv.dvc .gitignore git commit -m add raw data然后写清洗脚本src/data/clean.py把原始日志转成特征表。我的清洗逻辑分四步去重、缺失值处理、异常值截断、特征编码。每一步都写成独立函数方便单测。def remove_duplicates(df, key_cols): return df.drop_duplicates(subsetkey_cols, keeplast) def fill_missing(df, num_cols, cat_cols): for col in num_cols: df[col] df[col].fillna(df[col].median()) for col in cat_cols: df[col] df[col].fillna(__missing__) return df def clip_outliers(df, num_cols, lower_q0.01, upper_q0.99): for col in num_cols: lo df[col].quantile(lower_q) hi df[col].quantile(upper_q) df[col] df[col].clip(lo, hi) return df清洗完的特征表用 Parquet 格式存储因为 Parquet 列式存储读取快、压缩率高。我实测一个 500 万行的特征表CSV 格式 1.2GBParquet 只要 180MB读取速度快 4 倍。4.3 训练与实验追踪让每次实验都可追溯训练脚本我用 MLflow 做追踪。关键是把超参、指标、模型文件都记录下来并且给每次运行打上数据版本标签。import mlflow mlflow.set_tracking_uri(http://localhost:5000) mlflow.set_experiment(user-behavior-prediction) with mlflow.start_run(): mlflow.log_params({ learning_rate: 1e-3, batch_size: 64, epochs: 10, data_version: v1.2 }) for epoch in range(epochs): train_loss train_one_epoch(model, loader) val_auc evaluate(model, val_loader) mlflow.log_metrics({ train_loss: train_loss, val_auc: val_auc }, stepepoch) mlflow.pytorch.log_model(model, model)这里data_version标签很重要。当线上效果变差时你可以通过这个标签快速定位是哪个数据版本训练的模型然后对比数据分布找出问题。训练完成后我把最佳模型导出为 ONNX并记录转换前后的输出差异。这个差异值也存进 MLflow作为模型质量的一部分。4.4 模型注册与版本管理自建轻量注册表MLflow 自带模型注册表但我更倾向于自建一个轻量注册表因为 MLflow 的注册表在跨团队协作时权限管理比较麻烦。我的做法是用一个 JSON 文件加对象存储。{ models: { user-behavior: { production: { version: v1.2.0, path: s3://models/user-behavior/v1.2.0/model.onnx, data_version: v1.2, metrics: {val_auc: 0.842}, created_at: 2024-01-15T10:30:00Z }, staging: { version: v1.3.0-rc1, path: s3://models/user-behavior/v1.3.0-rc1/model.onnx, data_version: v1.3, metrics: {val_auc: 0.851}, created_at: 2024-01-20T14:00:00Z } } } }服务层启动时读取这个 JSON根据环境变量决定加载 production 还是 staging 版本。回滚只需要改 JSON 里的版本号然后重启服务。这个方案土但极其可靠而且任何人都能看懂。4.5 服务部署从单机到多副本服务用 FastAPI 写启动命令是uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4--workers的数量我一般设为 CPU 核数。但注意如果每个 worker 都加载一份模型内存会成倍增长。对于大模型我改用单 worker 线程池的方案前面用 Nginx 做负载均衡。Nginx 配置的关键是超时和重试upstream inference { server 127.0.0.1:8000; keepalive 32; } location /predict { proxy_pass http://inference; proxy_read_timeout 5s; proxy_connect_timeout 1s; proxy_next_upstream error timeout; }proxy_read_timeout设为 5 秒因为我的 P99 延迟在 800ms 左右5 秒足够覆盖异常情况。keepalive设为 32 减少连接建立开销。4.6 监控告警从指标到行动Prometheus 抓取指标Grafana 展示告警规则写在 Prometheus 的 rules 文件里。我的核心告警规则有三条。第一条P95 延迟超过 1 秒持续 5 分钟。第二条错误率超过 1% 持续 3 分钟。第三条PSI 超过 0.25。groups: - name: inference rules: - alert: HighLatency expr: histogram_quantile(0.95, rate(inference_latency_bucket[5m])) 1 for: 5m labels: severity: warning annotations: summary: P95 latency above 1s - alert: HighErrorRate expr: rate(inference_errors_total[3m]) / rate(inference_requests_total[3m]) 0.01 for: 3m labels: severity: critical - alert: DataDrift expr: model_psi 0.25 for: 10m labels: severity: warning告警不是终点关键是告警后的行动。我的 runbook 里写清楚了每种告警的第一步排查动作延迟高先看 CPU 和线程池队列错误率高先看模型加载和输入格式漂移告警先看上游数据源。5. 常见问题与排查技巧实录5.1 模型加载慢导致冷启动超时这个问题我遇到过三次每次都是因为模型文件太大或者加载逻辑有问题。排查思路是先用time命令测加载耗时如果超过 3 秒就要优化。优化手段有三个。第一把模型转成 ONNX 并启用图优化。第二用onnxruntime的sess_options.optimized_model_filepath把优化后的模型存下来下次直接加载优化版。第三如果模型真的很大考虑用内存映射文件加载。sess_options ort.SessionOptions() sess_options.optimized_model_filepath model_optimized.onnx session ort.InferenceSession(model.onnx, sess_options)第一次加载会慢但之后加载model_optimized.onnx会快很多。我实测一个 500MB 的模型优化后加载时间从 6 秒降到 1.5 秒。5.2 推理结果与训练时不一致这是最隐蔽的问题通常由三个原因导致。第一预处理不一致。训练时用的归一化参数和线上不同。第二输入维度顺序不同。第三ONNX 转换时某些算子行为有差异。排查方法是逐层对比。把同一个样本分别喂给 PyTorch 和 ONNX对比每一层的输出。我一般从第一层开始找到第一个差异超过 1e-4 的层然后检查那个层的算子。注意ONNX 的Softmax和 PyTorch 的Softmax在数值稳定性上可能有细微差异通常不影响结果但如果你的阈值卡得很紧就要留意。5.3 内存泄漏导致服务崩溃FastAPI 服务跑几天后内存涨到几个 GB最后 OOM。这个问题我排查了两天最后发现是日志里存了完整的请求体而请求体里有大数组。解决方法是日志只记录元数据不记录原始数据。如果确实需要记录输入用于调试用采样比如每 1000 个请求记录一个并且把大数组截断。import logging logger logging.getLogger(__name__) def log_request(request_id, inputs): logger.info({ request_id: request_id, input_shape: inputs.shape, input_dtype: str(inputs.dtype), input_sample: inputs.flatten()[:5].tolist() })5.4 常见问题速查表现象可能原因排查动作解决手段冷启动超时模型加载慢测加载耗时ONNX 优化、内存映射结果不一致预处理差异逐层对比输出统一预处理代码内存泄漏日志存大对象看内存增长曲线日志采样、截断延迟突增线程池满看队列长度调大线程池、限流漂移告警上游数据变化对比特征分布重训、加特征校验版本回滚失败旧模型未存检查注册表强制保留最近 5 个版本5.5 我踩过的三个坑第一个坑是用 Git 管模型文件。一个 200MB 的模型提交了 10 次仓库变成 2GBclone 一次要 5 分钟。后来改用对象存储加版本号仓库干净了回滚也快了。第二个坑是在异步函数里做同步 IO。我在async def里直接读 Redis结果事件循环被阻塞QPS 上不去。后来所有 IO 都改成异步客户端或者用线程池包起来。第三个坑是监控只加不删。指标越加越多Prometheus 存储爆炸查询变慢。后来我定了规矩每个指标必须有明确的告警规则或看板用途否则不加。6. 后续扩展与个人体会这套从零搭的流程跑通之后扩展方向其实很多。比如把批处理推理单独拆成一个服务用消息队列解耦比如加一个 A/B 测试框架按用户 ID 分流到不同模型版本比如把特征计算下沉到 Flink 做实时特征。但我的建议是不要一次全上每加一个组件先问自己当前系统真的扛不住了吗如果答案是否定的就先不加。我个人在实际操作中的体会是AI 工程化最难的不是技术而是克制。克制住一上来就上全家桶的冲动克制住为了“先进”而引入复杂度的冲动。ai-engineering-from-scratch这个思路的核心价值就是让你在亲手搭过一遍之后清楚地知道每个组件的边界在哪里什么时候该用它什么时候该换掉它。这套流程我用了两年中间换过模型、换过硬件、换过团队但分层结构和核心接口一直没变这大概就是它最稳的地方。
