这两年 ai-engineering 这个概念热度涨得很快但大部分人讨论的还是“怎么训练一个更准的模型”很少有人认真聊“怎么把一个能跑的模型变成一条稳定、可维护、出问题能快速定位的生产链路”。我年初给自己立了个项目仓库名就叫 ai-engineering-from-scratch想从零把 AI 工程化这条链路完整走一遍。前两天翻提交记录发现这个项目已经迭代了八个多月踩了不少坑也沉淀下来一套自己觉得还比较顺手的实践方法。这篇就当是阶段总结吧给同样想从“能跑通 Notebook”跨到“能上线扛流量”的朋友一个参考。我见过太多人学 AI 学到一半就卡住不是卡在数学而是卡在“下一步干什么”上。模型调参调到头也就那样数据管道一塌糊涂、实验记录乱七八糟、上线后模型坏了都不知道怎么回事。这些都是 ai-engineering 要解决的问题也是从零开始最容易被忽视的部分。这篇文章我会把我实际搭项目的路径、选型逻辑、代码结构、还有踩过的坑全部拆开讲适合刚开始做工程化转型的人也适合想系统梳理 AI 工程知识体系的朋友。1. 先想清楚AI 工程到底在解决什么问题1.1 它和机器学习之间的本质差别很多人把 ai-engineering 理解成“机器学习的工程化版本”这么说对了一半但很容易把人带偏。传统做机器学习核心焦点是模型本身特征怎么提、参数怎么调、精度怎么涨。做 AI 工程就不一样了模型只是整条链路里的一个环节它不是终点而是起点。我用一个生活化的类比讲清楚这个差别。做机器学习像在家做一顿饭你只需要关心菜做得好不好吃、火候到不到位。做 AI 工程更像开一家餐厅除了菜好吃你还得管食材采购、库存管理、厨房动线、出餐速度、客户投诉处理。菜做得好只是基础真正决定餐厅能不能活下去的是那套支撑“稳定出餐”的体系。落到实际工作里一个典型的 AI 工程链路至少包括数据采集与清洗、数据版本管理、特征工程、实验追踪、模型训练、模型评估、模型注册、服务部署、在线监控、告警通知、模型定期重训。这些环节每一个单独拆出来都不算难但把它们串成一条能稳定运转的流水线才是 ai-engineering 真正的门槛。1.2 从零开始的合理学习顺序我一开始犯过一个典型错误就是上来就啃深度学习理论。结果两个月过去神经网络是懂了一点但连一个能拿去面试的完整项目都做不出来。后来我把思路完全反过来先跑通一个最小闭环再往回补理论。所谓最小闭环就是“数据进、预测出”的完整链路。我最终的路径是先补 Python 工程化基础重点学虚拟环境、面向对象、类型注解、单元测试然后用一个文本分类项目把数据清洗、特征提取、模型训练、模型保存、FastAPI 部署、性能压测全部走一遍最后再去补 MLflow、Docker、CI/CD 这些工具链。理论部分只在遇到问题时按需补比如做推荐系统时才发现协同过滤的向量相似度计算不够快再回头查 Faiss 的原理。这个顺序的好处是每一步都有明确的产出物学习的正反馈很强。学完一个阶段你能拿出一个实际能跑的东西而不是一堆笔记。对于 90% 的人来说工程能力上的短板比算法理论上的短板更影响找工作、做项目。2. 核心技能栈解析我把每个环节拆开讲清楚2.1 数据层比模型更值得花时间的地方数据是整条 AI 工程链路上最容易被轻视、也最容易翻车的环节。我刚开始时天真地以为数据工作就是把 CSV 读进来、drop 掉空值、然后开始训练。后来线上出了问题才意识到数据层的工程化程度直接决定模型上线后的稳定性。我实际搭建的数据层包含四件事数据版本管理、数据校验、数据血缘、数据切片。数据版本管理我用的是 DVC。你可能会问数据不都已经存在数据库里了吗为什么还要额外管理因为训练数据是会被更新的今天数据里多了一批新样本明天某个字段的口径变了如果不对数据做版本控制你根本说不清楚“这个模型当时是用哪批数据训出来的”。DVC 的做法很轻量它不存储数据本身只存储数据的元信息和哈希值文件存在本地或云端对象存储里。每条实验记录对应一个确定的数据版本出了问题能精准回滚。数据校验这块我强烈推荐 Great Expectations。它的思路很简单先定义一个“数据应该长什么样”的规则集比如“年龄字段必须在 0 到 120 之间”“订单金额不能为负数”“用户 ID 不能重复”然后在每次数据更新后自动跑一遍校验。这个习惯养成以后可以避免大量线上事故。我碰到过的一个典型情况是上游业务方改了字段含义把单位从“天”改成了“小时”模型没有及时适应导致预测结果全面偏差。如果数据校验里预设了取值范围规则这个问题在数据进管道的第一秒就会被拦截。2.2 实验追踪与模型注册把“试错”变成“资产”做 AI 工程实验管理能力决定了你的迭代速度。我发现很多团队做实验还停留在“跑完一个 Notebook记住准确率然后继续改参数”的阶段。这种方式的隐患在于你永远不会知道当前跑出来的结果对应的是哪份代码、哪份数据、哪组参数。等过两个星期回来看当时的实验早就被覆盖了想复现都无从下手。实验追踪工具我首选 MLflow因为它是开源的项目里最接近“全家桶”方案的。MLflow 自带四个模块Tracking记录参数、指标、代码版本、Projects打包代码、Models模型存储与注册、Registry模型生命周期管理。对于个人项目和中小团队不需要自己拼凑太多工具。用 MLflow 的时候要养成几个好习惯。第一每个实验启动前先把当前代码的 Git commit hash 记下来。第二除了记录模型的准确率、F1 这些指标还要记录训练集的样本量、特征的哈希值、数据集的版本号。第三模型注册到 Registry 时要打上 stage 标签比如 Staging、Production。这样当你在 Production 阶段发现模型效果下降时可以直接在 Registry 里一键把上一个版本切回去整个操作有迹可循。我自己的实验命名规则是项目名_模型类型_数据版本_时间比如spam_classifier_xgb_v3_20241115。刚开始觉得繁琐但真正需要回滚模型的时候你会庆幸当初多花了这一分钟。2.3 模型服务化从离线脚本到稳定接口模型服务和部署这部分是“能不能上生产”的分水岭。训练好的模型文件只是静态产物要让业务真正用起来必须把它包装成一个接口、塞进一个能扛流量的进程里。这里有几个不同的方案我用一张表做个对比方案适用场景延迟优点缺点离线批预测推荐物料的预生成、报表类预测分钟到小时级实现简单、吞吐高结果滞后不适合实时决策在线 API 预测风控、搜索排序、实时推荐毫秒到百毫秒级实时性强、易于接入需要关注并发和延迟流式预测日志实时分析、监控告警秒级连续处理、管道天然衔接基础设施复杂维护成本高我做最小闭环时选择的是在线 API 预测用 FastAPI ONNX Runtime 这套组合。选 FastAPI 纯粹是因为它写起来快、支持异步、自动生成 OpenAPI 文档对 Python 生态的项目几乎零成本。模型推理则用 ONNX Runtime因为 ONNX 把模型固化成了一个跨语言、跨平台的中间表示比直接部署原始模型框架更灵活而且在 CPU 上做了大量算子优化推理速度往往比原框架快不少。转 ONNX 有个常见的坑如果模型里有动态维度比如变长的文本序列导出时必须指定动态轴否则导出后的模型只能处理固定长度的输入。我第一次导出文本分类模型时就栽在这里明明训练时能处理任意长度的句子导出后一测长文本就报错。解决办法是在torch.onnx.export或sklearn-onnx里显式声明dynamic_axes把序列长度维度标记为动态。2.4 评估与监控生产环境里准确率只是一个起点模型上线后评估指标和监控指标的体系要完全换一套。离线评估时大家都盯着准确率、F1、AUC但上线以后你真正关心的应该是线上输入分布有没有变化、单次推理耗时是否稳定、预测结果的置信度是否下降。监控我按三类来设计系统监控、模型监控、业务监控。系统监控就是常规的 CPU、内存、QPS、延迟用 Prometheus Grafana 就能搞定。模型监控则需要关注输入特征的分布漂移我用 PSIPopulation Stability Index这个指标来量化。原理很简单对比当前输入数据的特征分布和训练集的特征分布如果差异超过某个阈值就告警。PSI 的计算代码不复杂我后面会贴一个可以直接用的版本。业务监控是很多人漏掉的一环。模型预测完以后下游业务有没有采纳这个结果用户对这个推荐的点击率怎么样这部分数据往往回流得慢需要定期做离线分析。我一开始完全没做业务监控后来发现模型指标一切正常但业务方反馈效果变差了最后追踪到是业务策略变了导致模型输出的分布跟着变了。所以说模型监控和业务监控要配合着看才不会被单方面指标的“正常”误导。3. 实操记录我如何搭出一个最小可用的 AI 工程闭环3.1 项目结构每一层各司其职我搭项目时没有沿用“一堆 Notebook 文件夹”的结构而是按照数据管道、训练模块、服务模块、监控模块四个层次组织。这个结构不是一开始就设计好的是被踩坑教育出来的后来发现跟很多成熟项目的标准布局基本一致。ai-engineering-from-scratch/ ├── data/ │ ├── raw/ # 原始数据只增不改 │ ├── processed/ # 清洗后的建模数据 │ └── versions/ # DVC 版本记录 ├── src/ │ ├── data/ # 数据采集、清洗、校验脚本 │ ├── features/ # 特征工程代码 │ ├── models/ # 模型定义、训练、评估脚本 │ └── servers/ # FastAPI 服务代码 ├── configs/ # 配置文件按环境拆分 ├── scripts/ # 流水线编排脚本 ├── tests/ # 单元测试、集成测试 ├── docker/ # Dockerfile、docker-compose └── mlruns/ # MLflow 实验记录这套结构最重要的设计原则是原始数据目录只允许追加不允许修改。因为数据处理过程中很可能有人改动原始文件一旦改了后续所有从这个文件衍生出来的模型结果都失去了可比性。为了强制这一点我在原始数据摄入脚本里加了文件哈希校验文件哈希变了直接报错。这招简单粗暴但确实能防止“原始数据被悄悄污染”的情况。3.2 数据准备与训练把 pipeline 焊死我做的第一个项目是垃圾短信分类样本量不大但非常适合走完整条链路。数据处理部分我用 scikit-learn 的 Pipeline 来组织把清洗、分词、向量化串成一条固定的数据流。from sklearn.pipeline import Pipeline from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.naive_bayes import MultinomialNB import joblib def build_pipeline(): return Pipeline([ (tfidf, TfidfVectorizer(max_features5000, ngram_range(1, 2))), (clf, MultinomialNB(alpha0.1)), ]) if __name__ __main__: import pandas as pd from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report df pd.read_csv(data/processed/sms_spam.csv) X_train, X_test, y_train, y_test train_test_split( df[text], df[label], test_size0.2, random_state42, stratifydf[label] ) pipeline build_pipeline() pipeline.fit(X_train, y_train) report classification_report(y_test, pipeline.predict(X_test)) print(report) joblib.dump(pipeline, models/sms_spam_pipeline_v1.pkl)这段代码里有几个细节值得展开说。第一用 Pipeline 的好处是特征工程和模型被绑定成了同一个对象后续做预测时不需要再单独调用向量化器不容易出现“训练时和推理时特征处理不一致”的问题。第二stratifydf[label]保证了训练集和测试集的类别比例一致防止类别不平衡时划分出的测试集失真。第三保存的是完整的 pipeline 而不是只保存模型参数这样加载后可以直接喂原始文本避免漏掉预处理步骤。训练这块我建议你在小项目上也走完整的流程定义可复现的随机种子、记录全部超参数、把实验指标写入 MLflow。哪怕只是一个朴素贝叶斯模型这套习惯越早养成越好。3.3 模型导出与服务化让模型真正跑起来训练完成后下一步是把模型导出并部署成 API。我用的是sklearn-onnx把 pipeline 转成 ONNX 格式然后用 FastAPI 封装成接口。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort import numpy as np app FastAPI(titleSMS Spam Classifier) session ort.InferenceSession( models/sms_spam_pipeline_v1.onnx, providers[CPUExecutionProvider] ) class SMSRequest(BaseModel): text: str class SMSResponse(BaseModel): label: int confidence: float app.post(/predict) async def predict(req: SMSRequest): if not req.text.strip(): raise HTTPException(status_code400, detailtext must not be empty) inputs {input: np.array([req.text], dtypenp.str_)} outputs session.run(None, inputs)[0] label int(np.argmax(outputs[0])) confidence float(np.max(outputs[0])) return SMSResponse(labellabel, confidenceconfidence) app.get(/health) async def health(): return {status: ok}这个服务看起来简单但有几个工程要点我是踩过坑才明白的。第一接口入参一定要用 Pydantic 做校验空文本这种边缘情况如果不校验就拿去跑模型轻则返回错误结果重则引发内存异常。第二ONNX Runtime 的 session 应该在服务启动时初始化一次而不是每个请求现场加载一次模型。模型加载耗时长放请求路径里会把 P99 延迟拉到不可用的程度。第三健康检查接口必不可少。Kubernetes 或者 Docker Compose 做容器探活的时候如果找不到健康检查端点服务一启动就会被重启。部署我用 Docker 打包镜像里只装了fastapi、uvicorn、onnxruntime、numpy比直接装完整训练环境的镜像小了将近 3GB。这里有一个经验训练环境和服务环境的依赖应该分开锁。训练环境可以宽松一点方便调试服务环境必须用 requirements.lock 锁死每一个依赖的精确版本。服务端的任何依赖升级都可能导致行为变化所以环境越稳定越好。3.4 监控告警用一段可复用的 PSI 检测数据漂移服务跑起来以后最怕的不是机器宕机而是“数据悄悄变了”。输入分布一旦漂移模型准确率会肉眼可见地掉但你一开始根本发现不了。我写了一段轻量级的 PSI 计算函数用来检测输入特征的分布变化。import numpy as np def calculate_psi(expected: np.ndarray, actual: np.ndarray, bins: int 10): expected_counts, edges np.histogram(expected, binsbins) actual_counts, _ np.histogram(actual, binsedges) expected_pct (expected_counts 1e-6) / expected_counts.sum() actual_pct (actual_counts 1e-6) / actual_counts.sum() psi np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct)) return psi def check_and_alert(): # 加载训练集特征分布作为 expected # 加载最近一周线上输入特征作为 actual psi calculate_psi(expected, actual) if psi 0.1: print(fALERT: data drift detected, PSI{psi:.4f}) return psiPSI 的判读标准一般是小于 0.1 表示分布无显著变化0.1 到 0.25 表示轻度漂移需要关注大于 0.25 表示显著漂移模型必须重新评估或重训。这套阈值来自风控行业的经验直接拿过来用基本可靠。这个监控要跑得勤但不用实时在线计算。我是每天凌晨用定时任务把前一天线上请求的特征字段抽出来跟训练集分布做对比结果写入日志和看板。第一次成功触发告警的时候排查出来的原因是业务方上线了一个新的短信模板文本风格突变垃圾短信和正常短信的关键词分布全部变了。如果没有这个监控这个问题至少要到业务方反馈“最近拦截效果变差了”才会被发现。4. 常见问题与排查技巧实录4.1 环境依赖地狱怎么破Python 项目的环境问题简直是个永恒的话题。我遇到过最夸张的一次是同一个项目在开发机上跑得好好的部署到生产服务器上直接 import 报错查了半天发现是 protobuf 版本冲突。后来我彻底切换到 poetry 管理依赖并且把所有环境统一构建成 Docker 镜像。镜像构建时的关键操作是锁定基础镜像版本比如python:3.11-slim而不是python:3.11否则镜像每次构建都可能拉到一个新版本行为不可控。如果项目已经有历史包袱、短时间内不能切到 docker折中的方案是至少要做到用pyenv锁定 Python 版本用requirements-lock.txt锁死依赖版本禁止任何人手动pip install xxx装到全局环境。依赖管理这事前期多花半小时能省后期好几天。4.2 线上模型效果下降时的排查顺序模型效果下降是 AI 工程里最让人头大的问题。我处理过几次之后总结了一套排查顺序现在基本可以做到快速定位。先看数据分布算一下近期输入特征的 PSI如果显著漂移问题大概率出在数据层。再看数据质量有没有某天开始某个字段出现大量空值比如上游日志格式调整导致字段对不上。然后看模型服务本身是不是有人改了模型文件、重启了服务、升级了依赖。最后才看重训必要性如果数据和代码都没问题那就是模型本身跟不上概念漂移了需要收集新数据做重训。按这个顺序排查绝大部分问题都能在 30 分钟内定位到根因。我的经验是八成以上的“模型变差了”问题根源根本不在模型而在数据或服务链路。4.3 实验对比时结果复现不出来做多组实验对比时最尴尬的就是“昨天同样的代码同样的参数今天跑出来的结果不一样”。这不是玄学基本逃不出三个原因随机种子没固定、数据被改动过、依赖版本变了。固定随机种子的正确做法不是只调random.seed(42)因为 PyTorch 有自己的随机源、NumPy 有自己的随机源Python 的标准库 random 也有。需要把每个框架的随机源都固定而且要在 DataLoader 里设置shuffleFalse或者固定 shuffle 的 seed。数据方面要依赖前文提到的 DVC 版本控制来保证。依赖方面在一个项目的生命周期里锁文件只允许定向升级不允许随手pip install --upgrade。4.4 服务端推理延迟忽高忽低FastAPI 的接口看起来响应很快但扛并发时经常出现 P99 延迟飙升的情况。我压测后发现问题出在文本向量化阶段用了纯 Python 做分词在并发请求多时 GIL 让分词环节几乎串行执行。优化方案有两个一是把分词步骤放到独立进程中用消息队列或者 multiprocessing pool 来处理二是把文本预处理全部批量化显式传入 batch size 让 ONNX Runtime 做批推理。第二个方案在 CPU 服务上效果明显单条请求的延迟可能没降但整体吞吐可以提高一倍以上。另外要关注的是“冷启动”问题。模型第一次加载到内存时非常慢如果服务实例频繁重启P99 延迟会长期不稳定。解决办法是加一个预热函数服务启动后先发一条空的健康检查请求触发模型加载完成后再暴露流量入口。5. 从个人项目到生产落地一些进阶建议5.1 用 CI/CD 把机器学习流程自动化个人项目做顺手以后我建议尽早引入简单的自动化流程。GitHub Actions 加上 CML 这个工具能实现最基本的 ML 流水线代码 push 后自动跑单元测试、数据测试、模型训练并把实验报告作为 comment 更新到 PR 里。这一步看似多了一层复杂度但它的价值是强制要求每次代码变更都必须通过测试和数据校验否则模型结果不被信任。这比你自己手动跑一遍再截图汇报要可靠得多也省时间得多。自动化测试的核心不只是测代码逻辑更要测数据和模型。数据测试可以用 Great Expectations 跑一套完整性规则模型测试可以设定“新模型在验证集上的指标不得低于当前生产模型指标的 2%”否则自动拦截上线。有了这层保障以后迭代模型时你不需要每次都提心吊胆。5.2 往 LLMOps 和 RAG 方向扩展时要注意什么现在聊 AI 工程绕不开大语言模型这条线。我个人的体会是如果把前面那些基础工程能力打扎实了切换到 LLM 方向其实是顺水推舟的事。LLMOps 里最核心的 RAG检索增强生成流程本质上就是“检索系统 生成系统 评估闭环”你前面积累的数据管道、实验追踪、模型评估经验全部可以直接复用。最开始学 RAG 时不要直接上 LangChain先自己用向量数据库和模型 API 手搓一个最简版本。手搓的过程能帮你理解 embedding 是怎么存储和检索的、chunk 切分对回答质量的影响、prompt 模板在系统里的位置。一旦这些底层逻辑清楚了再去用框架你会发现自己是在“用框架解决问题”而不是“被框架限制”。最后说一点实在的感受从ai-engineering-from-scratch这个项目开始到现在我最深的体会是做 AI 工程最大的门槛从来不是某个具体的技术而是“完整视角”的建立。你只有把数据、训练、部署、监控、迭代这五个环节真正跑通过一轮才会明白为什么很多公司里模型效果最好的那个人往往不是数学最好的人而是工程链路最熟的人。如果只能给一条建议我会说别急着追新框架、新模型先拿一个最普通的任务把从数据到监控的全链路跑通。等你亲手解决过一次“线上模型精度掉到不可接受但不知道哪里出了问题”的经历你就真正入门 ai-engineering 了。