从零搭建AI工程：数据、模型到监控的完整闭环实践

1. 从零开始做 AI 工程到底在做什么“ai-engineering-from-scratch”这个名字我第一次看到的时候第一反应是这不只是“训练一个模型”那么简单。它背后其实是一整套工程体系的搭建——从数据怎么进、模型怎么训、到服务怎么上线、上线之后怎么持续监控哪个环节断了整个项目都跑不起来。我见过太多人把 AI 项目等同于“写个 Notebook 跑通 Accuracy”但真正放到生产环境里你会发现 Notebook 里的 90 分和线上服务的 90 分完全是两码事。输入数据格式变了怎么办推理速度达不到 SLA 怎么办模型效果衰减了怎么发现这些问题没有一个能在 Notebook 里回答。所以“from scratch”这个限定词特别关键它意味着你不依赖现成的全栈平台而是从头把每个部件组装起来理解每一层的原因而不是只做一个调包侠。这篇文章就是围绕“从零开始搭建 AI 工程”这条主线把我实际踩过的坑、验证过的方案、以及每一步选择背后的逻辑都拆开讲清楚。无论你是刚入门想做第一个端到端项目还是已经在做模型但想补上工程化这块短板这篇文章都能给你一条可以照着走的路。我会从整体架构设计、数据处理、模型训练与评估、服务部署再到监控与持续迭代一条线把完整闭环讲透。2. 先想清楚边界你的 AI 系统到底要解决什么问题2.1 别急着选模型先把问题定义清楚很多人一上来就问“用 ChatGPT 还是用开源模型”我每次都会反问一句你要解决的任务输入是什么、输出是什么、错误代价有多高这三个问题没想明白选什么模型都是拍脑袋。所谓“问题定义”在 AI 工程里至少包含以下内容任务类型是分类、回归、排序还是生成式任务任务类型直接决定模型架构和损失函数。输入输出边界输入是纯文本、图片、结构化表格还是混合模态输出是标签、数值、向量还是文本序列评价指标离线指标用什么准确率、F1、BLEU、NDCG线上指标用什么延迟、吞吐、用户反馈两者之间的关系是什么。失败容忍度预测错了会怎样是推荐系统里少推一个商品还是风控系统里放了一笔欺诈交易这决定了需要人工兜底还是全自动决策。这里有一个很典型的例子。我在做一个文档分类系统时最初大家都觉得准确率最重要结果跑下来发现准确率卡在 92% 上不去了。后来一分析真正让业务方头疼的是“分错类导致工单派错部门”这个错误的代价远高于“分得不够细”。于是我们把目标从“最大化准确率”改成“保证高置信度样本的准确率低置信度样本直接转人工”加入了一个不确定性阈值。系统上线后业务方真正满意的不是准确率涨了而是错误样本变少了。所以从 scratch 开始的第一步不是写代码而是写清楚一个“问题说明书”。这份说明书里要包含上面讲的四个维度并且要跟业务方逐条对齐。这步做扎实了后面工程的每个环节都有据可依。2.2 设计整体架构把 AI 系统当软件系统来搭搞清楚要解决什么问题之后就需要设计系统的整体架构了。很多从零开始的项目失败不是因为模型不好而是因为架构压根没有把数据流、训练流、服务流分开结果所有环节耦合在一起改一个参数都要动全身。我习惯把 AI 工程架构拆成五个核心模块数据层负责数据采集、清洗、标注、版本管理、特征加工。训练层负责实验管理、模型训练、超参调优、模型评估与注册。服务层负责模型部署、推理 API、前后处理、流量调度。监控层负责模型指标监控、数据漂移检测、日志收集、告警通知。迭代层负责反馈数据回流、模型重训触发、AB 实验、灰度发布。这五个模块之间是解耦的每个模块都有明确的输入输出接口。举个例子数据层产出的是一份带版本号的数据集训练层消费这份数据产出的是一个带版本号且注册过的模型服务层只认模型注册表里的版本监控层监控线上行为并把反馈数据写回数据层。这样做的好处是任何一个模块替换或升级都不需要改动其他模块。从 scratch 搭建这套架构不需要一步到位上 Kubernetes、上 Feature Store那样反而会被基础设施拖死。我建议从小而美的单体起步先用 Python 包管理 配置文件把这五个模块串起来跑通一条端到端流水线然后再逐步把模块拆成独立服务。这个思路和软件工程里的“先单体、后微服务”是一样的逻辑——先做能用的系统再考虑规模化和分布式。3. 数据工程的细节决定了模型效果的天花板3.1 数据采集与清洗大部分脏活都在这里AI 工程里有一句老话Garbage in, garbage out。但“垃圾数据”到底是什么很多人其实没概念。以文本分类任务为例数据采集阶段常见的问题包括重复数据、标签噪声、类别不平衡、文本编码混乱、测试集与训练集重叠等。每一个问题都会在模型效果上体现出来。我自己做数据清洗时会按照固定流程走每一条都有对应的检查脚本去重对全文计算 MinHash 或 SimHash 相似度相似度超过阈值的样本只保留一条。单纯用文本完全匹配去重不够因为很多重复样本只是多了几个空格或换行。标签清洗这一步容易被忽略。很多时候标注团队对标签定义理解不一致需要随机抽一批样本做标签一致性校验用 Cohens Kappa 计算标注一致性低于 0.7 都需要组织标注返工。编码规整统一转成 UTF-8去掉不可见字符、零宽字符全角半角统一。这种不起眼的问题会在模型推理阶段引入莫名其妙的错误。类别平衡分析统计每个类别的样本数量如果长尾特别严重需要在训练时做重采样或者用损失函数加权而不是直接删样本。清洗流程跑完后我还会额外做一次“数据泄漏检查”。最常踩的坑是测试集里出现了训练集的同源数据。比如爬虫抓取的数据里同一篇文章稍微改几个字就同时出现在训练集和测试集里导致模型在离线评估时虚高一上线上就崩。检查方法可以算训练集与测试集之间的最小哈希相似度把相似度高于 0.8 的样本全部从测试集剔除。3.2 数据集版本管理让每个实验可复现模型训练的可复现性很大程度上取决于数据集的可复现性。你训练时用的数据是 V1.0结果过了两周发现数据有 bug 改了 V1.1结果实验对比时根本说不清效果变化是模型改进还是数据变化带来的。所以数据集必须做版本管理。这里不需要上重型系统最轻量的方案是用 DVCData Version Control把数据集目录挂到 Git 仓库每次数据变更生成一个新的版本号并在实验记录里关联数据版本。具体操作大致是# 初始化 DVC dvc init # 把数据目录纳入版本管理 dvc add data/raw/text_classification # 提交版本并推送远端存储 git add data/raw/text_classification.dvc git commit -m dataset: update label rules for category A dvc push每次训练时在实验配置里显式记录数据集版本。比如dataset: version: v1.1 path: data/processed/text_classification note: 修复类别A的标签噪声剔除32条重复样本这看起来有点繁琐但一旦你要回溯实验、排查线上问题数据版本就是救命稻草。没有版本管理遇到问题只能“凭感觉猜”有了版本管理每一步都有据可查。3.3 特征工程在深度学习时代仍然不能丢很多人觉得深度学习能自动学特征所以特征工程可以省略。这个认知在纯文本、纯图像的标准任务里勉强成立但在涉及业务数据、时序数据、多源异构数据的场景里完全不做特征工程会非常吃亏。我常用的做法是“三明治”结构底层保留原始信号中间做业务特征顶层用深度模型自动学习。也就是说特征工程不是替代模型而是给模型提供先验知识。比如做用户点击率预估用户的活跃时长、历史点击率这类业务特征直接拼进模型 embedding 层比让模型从头学要稳定得多。特征加工在工程上需要注意“训练/推理一致性”。很多坑都出在这里训练时用了整个时间窗口的特征推理时发现还有未来数据导致特征分布不一致。解决方法是给所有特征打上时间戳并且严格要求特征计算只依赖截止到预测时刻的已观察数据。这个约束在代码上做成统一的特征算子接口训练和推理共用同一份逻辑避免两头维护双份代码。4. 模型训练与评估别让离线指标骗了你4.1 模型选型任务类型决定架构规模决定成本模型选型是训练阶段最核心的决策。这里我给一个简洁的选型表格适用于大多数从零开始的场景任务类型首选方案备选方案适用规模短文本分类/匹配Transformer 编码器如 BERT 系列轻量模型如 distillation 版本万级以下样本用预训练 fine-tune长文档理解Longformer / 分段编码 聚合抽取式摘要减少输入长度十万级以上文档文本生成开源生成模型如 Qwen、Llama 等自回归模型 fine-tune有领域数据时 fine-tune表格数据GBDT如 LightGBM / XGBoostMLP 特征工程万级到百万级样本时序预测LightGBM 滞后特征Transformer 时序模型小样本场景慎用深度模型图像分类预训练 CNN / ViT轻量化 CNN千级到百万级图像选型一个重要的判断标准不是“哪个模型准”而是“哪个模型在你的数据量、算力、推理延迟约束下最划算”。我在实际项目里经常看到有人用几十亿参数的模型做情感分析准确率比千亿参数模型差不了多少但推理成本和延迟高了一个量级运维还特别麻烦。很多时候大模型不是最优解小而精的模型才是。4.2 训练流程实验记录、超参管理、检查点保存从 scratch 起步时训练脚本一定要从一开始就带上以下基础能力否则后面补全是灾难实验记录至少用 MLflow 或 WandB 记录每次实验的配置、指标、模型产物。我不要求团队一定用哪一家但这个环节不能省。没有实验记录调参就是玄学。超参管理所有超参集中在 config 文件里禁止散落在代码里。推荐用 Hydra 管理 yaml 配置支持命令行覆盖。# config/train.yaml model: name: bert-base-chinese max_length: 128 train: batch_size: 32 learning_rate: 2e-5 epochs: 5 optimizer: type: adamw weight_decay: 0.01 data: version: v1.1检查点保存保存不只是“保存最优模型”还要保存 optimizer 状态、当前 epoch、数据版本、随机种子。这样一旦训练中断可以精确恢复复现实验也不用赌运气。随机种子固定PyTorch 里需要同时设置 Python、NumPy、PyTorch、CUDA 的种子才能最大程度保证可复现。这些基础能力看起来是“浪费时间”但一旦实验数量上来效果差别立竿见影。没有这套机制你根本不知道当前这个指标是哪个 config 跑出来的。4.3 离线评估与上线效果之间的鸿沟我做得最久也最头疼的一个项目是离线 AUC 做到了 0.93上线后实际业务指标没有任何提升。排查了很久最后发现原因有三层一是离线评估用的特征分布和线上实时特征分布不一致二是线下的样本是历史数据线上的用户行为已经发生了变化三是业务指标比如点击率和离线指标AUC并不是完全单调关系。所以我在评估阶段会同时做几件事除了标准指标如 AUC、F1 之外还会做“数据切片评估”。也就是把测试集按业务维度拆成若干子集分别计算指标观察模型在哪些人群、哪些时间段、哪些内容类型上表现差。比如一个推荐模型整体 AUC 是 0.9但只看新用户子集时 AUC 只有 0.7那新用户体验肯定是不行的。这份“切片评估报告”比单一总指标有用得多。另外还有一个人人都会犯但要刻意避开的错误在测试集上反复调参调模型最后把测试集“学到过拟合”了。更稳健的做法是把数据拆成训练集、验证集、测试集同时在训练过程中只碰验证集测试集只在最终评估时碰一次。要是条件允许还可以再建一个时间上更靠后的“封板测试集”模拟真实上线后的分布漂移。5. 模型服务化从离线到线上差的不只是 API5.1 推理服务的标准形态与关键参数模型训好之后就要把它变成一个“能对外服务”的系统了。最轻量可靠的方式是把模型封装成一个 HTTP 服务用 FastAPI 或 Flask 实现。相比动不动上 Kubernetes 的方案单个服务模块在初期更容易调试和维护。一个标准的推理服务要做的事情远不止“加载模型、预测、返回结果”至少要覆盖输入校验不能等模型 inference 时才发现格式不对要在请求入口做 schema 校验Pydantic 是个好选择。前后处理tokenizer 编码、预测结果解析、映射回业务字段这部分逻辑必须跟训练时代码一致。超时控制每个请求必须有超时上限避免模型偶发卡死把服务拖垮。并发控制用信号量限制最大并发数防止突发流量把 GPU 显存打爆。优雅启停模型加载和 warmup 要在服务真正接流量之前完成。一个参考的 FastAPI 推理服务伪代码结构from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification app FastAPI() class Item(BaseModel): text: str class PredictRequest(BaseModel): items: list[Item] top_k: int 1 # 全局加载一次模型避免每次请求重新加载 tokenizer AutoTokenizer.from_pretrained(model_cache/bert_finetuned) model AutoModelForSequenceClassification.from_pretrained(model_cache/bert_finetuned) model.eval() app.on_event(startup) def warmup(): # 用一条真实样本跑一次推理确保 CUDA kernel 已加载 with torch.no_grad(): dummy tokenizer(warmup, return_tensorspt) model(**dummy) app.post(/predict) def predict(req: PredictRequest): if len(req.items) 100: raise HTTPException(status_code400, detailbatch too large) try: texts [item.text for item in req.items] encoded tokenizer(texts, paddingTrue, truncationTrue, max_length128, return_tensorspt) except Exception as e: raise HTTPException(status_code400, detailftokenize error: {e}) with torch.no_grad(): outputs model(**encoded) probs torch.nn.functional.softmax(outputs.logits, dim-1) scores, labels torch.topk(probs, kreq.top_k, dim-1) results [] for i in range(len(texts)): results.append({ text: texts[i], predictions: [ {label: int(labels[i][j]), score: float(scores[i][j])} for j in range(req.top_k) ] }) return {results: results}这里有一个值得强调的设计批量推理。线上服务很少单条请求预测因为批量推理可以显著提高 GPU 利用率。实际使用时可以把请求排队攒到一定数量或时间窗口再统一推理我见过用积压队列把吞吐量提升三倍以上的例子。5.2 模型版本管理灰度发布与回滚上面讲训练时要注册模型那服务侧就要消费模型注册表。一个最简单的模型管理流程如下训练产出的模型被推送到一个模型仓库可以是本地文件目录或对象存储并为每次推送打上版本号。配置文件里记录线上服务当前使用哪个版本。model_repository/ ├── text_classifier/ │ ├── v1.0/ │ │ ├── model.bin │ │ ├── config.json │ │ └── metadata.yaml │ └── v2.0/ │ ├── model.bin │ ├── config.json │ └── metadata.yaml └── production_version.txt灰度发布的做法是让线上服务支持同时加载一个版本通过流量比例分配请求。比如先让 5% 流量打到 v2.0观察监控指标没有异常再逐步提升到 20%、50%、100%。一旦指标异常迅速把流量切回 v1.0比重新部署快得多也更安全。回滚机制一定要提前设计好。很多人觉得回滚就是改个版本号重启但实际操作中如果新版模型改变了输出格式或业务字段映射旧版本服务不兼容就麻烦了。因此模型产出的 metadata 里必须包含输入输出 schema 版本服务端做字段兼容性检查这样灰度切换时才不会出现线上告警一片的错误。6. 线上监控与持续迭代一个不能少6.1 监控哪些指标怎么设阈值模型上线不是结束而上线后的观察才是真正的开始。AI 工程的监控和传统后端监控有重叠但也有 AI 特有的部分。我的监控清单通常是三层系统层请求量、延迟P50/P99、错误率、GPU 利用率、显存占用。这层异常说明服务本身有问题和模型效果无关。模型层预测置信度分布、类别分布、平均输出分数等。如果模型线上输出分布和训练时的验证集分布差距过大往往意味着输入发生了漂移。数据层输入数据的统计特征比如文本长度、关键词频次、图像亮度、表格字段缺失率等。数据层监控是发现“静默失效”的关键。关于阈值怎么定我建议不要凭感觉拍脑袋而是从历史数据里找基线。具体做法是上线前收集一到两周的灰度流量数据计算各项指标的均值和标准差设定阈值为均值 ± 3σ。超过这个范围就触发告警。这个方法的逻辑基础是3σ 之外的概率在正态分布假设下不到 0.3%虽然不是严格统计学检验但作为工程告警阈值很实用。6.2 漂移检测如何判断模型该重训了模型效果衰减是必然的问题只是你什么时候发现、怎么发现。如果只看业务指标通常要等用户反馈已经很严重了才能发现这时候损失已经造成了。所以提前量很重要要通过数据漂移检测来提前预警。漂移检测的常用方法按数据类型不同有区别数值特征用 PSIPopulation Stability Index。PSI 小于 0.1 表示无漂移0.1 到 0.25 表示轻度漂移大于 0.25 表示显著漂移需要重训。离散特征用卡方检验或者直接比较分布占比看训练期占比和线上最近一周占比的差异。文本特征可以计算句子 embedding 的中心距离或者对 token 频率分布做 KL 散度。预测分布模型输出的类别概率分布变化也是一个很好的漂移信号。代码实现上一个简单的 PSI 计算器可以这样写import numpy as np def calculate_psi(expected, actual, bins10): # 将两个分布切到相同的分箱 expected np.asarray(expected) actual np.asarray(actual) # 基于期望分布确定分位点 percentiles np.percentile(expected, np.linspace(0, 100, bins 1)) percentiles[-1] percentiles[-1] 1e-6 expected_counts, _ np.histogram(expected, binspercentiles) actual_counts, _ np.histogram(actual, binspercentiles) # 转成占比 expected_ratio expected_counts / expected_counts.sum() actual_ratio actual_counts / actual_counts.sum() # 处理 0 值 expected_ratio np.clip(expected_ratio, 1e-6, 1) actual_ratio np.clip(actual_ratio, 1e-6, 1) psi np.sum((actual_ratio - expected_ratio) * np.log(actual_ratio / expected_ratio)) return psi拿到漂移检测结果后要主动设定“重训触发条件”。触发条件不是“漂移超过阈值就立刻重训”因为重训本身也有成本。通常我会结合三个信号漂移预警连续触发 N 天、业务指标连续阈值下探、以及人工抽样确认数据质量出现问题。三个信号里至少两个成立才会进入重训流程。重训流程也要自动化。数据层会接收新的标注或反馈数据自动生成新版本数据集训练层基于新数据跑一轮训练任务产出的模型注册到模型仓库然后走灰度发布。整个闭环跑顺后从数据回收到新模型上线快的话一天以内就能完成这才算是一个真正“会自我更新”的 AI 系统。6.3 AB 实验与多策略对比最后再讲一个工程细节如何证明你的新模型确实比老模型好。很多人用“我本地验证集指标更高”来证明业务方通常不信因为最终要看用户效果。所以 AB 实验是必要环节。在服务层分配流量时用户维度或请求维度随机进入对照组老模型和实验组新模型同时记录用户行为反馈。观察周期根据业务类型不同一般一到两周。核心观察指标需要提前定好比如推荐场景看点击率、转化率、人均点击搜索场景看点击率、无结果率、长尾满意度。这里有三个容易踩的坑辛普森悖论整体实验组指标提升了但拆到各个流量来源后发现实验组在多个子群都变差了。所以 AB 实验结束后一定要做分层分析。样本偏差随机分流时没有做用户 id 哈希均匀头像分成两组但活跃度不同看起来是模型差异实际是用户差异。需要做分流均匀性校验。早停误判上线三天看到指标有提升就宣布成功其实可能是新鲜感效应。至少观察一个完整的用户行为周期再下结论。AB 实验不只是“证明哪边赢”更是持续迭代的驾驶舱。每次模型迭代都过一遍 AB 实验整个系统才会往稳定的方向演化而不是凭感觉“改改更健康”。7. 从零到一我踩过的三个大坑整个“ai-engineering-from-scratch”的搭建过程我走了不少弯路最有代表性的三个坑值得单独拿出来说因为它们几乎在每个项目里都会复现。第一个坑是“数据集版本管理缺失导致实验无法复现”。最开始我图省事数据文件直接存在共享盘上文件名带了个日期。结果一周后要复现当时的实验发现数据已经被人改过了之前的实验跑不出同样的指标。后来引入 DVC 和实验记录才彻底解决。这件事给我最大的教训是任何实验要想被别人甚至将来的自己复现数据版本、代码 commit、模型参数、环境依赖一个都不能少。第二个坑是“离线评估指标和线上效果严重脱节”。一次做文本分类项目离线 F1 到了 0.92但线上工单的自动分类准确率始终上不去最后发现是线上文本和训练数据来自不同渠道写作风格差别巨大。后续加了数据切片评估和线上分布监控才在早期发现类似问题。这个事让我养成了“先看分布、再看指标”的习惯。第三个坑是“回滚预案没做就上了灰度”。某次推新模型上线后半小时效果类监控指标异常想回滚到旧版结果发现旧版模型文件被误删服务直接不可用。从那以后我的模型仓库里永远保留最近三个版本的模型文件且生产服务支持随时一键切回。这些坑没有一个是高深的技术难题全是工程管理问题。AI 工程最难的不是算法而是把算法、数据、系统、流程组成一个可靠的整体。这也是“from scratch”真正教给你的东西。如果你正在规划自己的 AI 工程不妨从最小化的闭环开始一个小数据集、一个小模型、一台能跑 GPU 的机器、一个 Flask 服务、一份简单的监控脚本。先把这条线跑通再逐步做规模化和自动化。每一步都理解清楚为什么这么做比一步到位搭一套华丽但管不过来的系统有价值得多。