AI工程化实战：从零搭建可复现的机器学习系统

这些年面试过不少自称“会AI”的候选人，开口就是模型指标能到几个点，可一聊到训练好的模型怎么部署、线上效果掉点怎么排查、数据回流怎么设计，很多人就开始含糊了。说白了，行业里缺的不是能调参的人，而是能把AI系统从零搭起来、稳定跑下去的人。这也是“ai-engineering-from-scratch”这个主题想聊透的事：不纠结某一个模型的内部结构，而是把“从业务问题到线上服务”的整条链路走通。

这篇内容适合两类人：一类是算法岗想补工程能力的，另一类是后端或全栈想往AI方向转的。我会按自己实际搭过的一套文本分类系统为例子，把环境管理、数据版本、实验追踪、模型服务、监控迭代这些环节逐个拆开讲，每个步骤都带上我踩过的坑和现在的习惯做法。看完你会发现，AI工程化没那么玄，它是一堆琐碎但可复现的规范拼起来的。

1. AI工程到底在解决什么问题

1.1 先分清“做模型”和“做系统”的区别

单独跑通一个模型，哪怕精度很高，也只能算完成了20%的工作。剩下的部分里，有数据怎么持续更新、特征怎么避免线上线下不一致、模型怎么平滑上线、出问题怎么快速回滚，这些才是工程的核心。

我用一个例子说明：你训练了一个文本分类模型，离线AUC到了0.95，但上线后用户反馈不准。为什么？很可能是线上请求的数据格式和训练时不一样，比如训练时做了分词和去停用词，线上接口却忘了做同样的预处理。类似这种问题，靠调模型是永远解决不了的，只能靠规范的工程流程来卡住。

“from scratch”的含义就在这里：不是给你一个训练好的模型让你调API，而是从空目录开始，一步步把数据管线、训练脚本、服务接口、监控告警全部搭出来。整个过程走一遍之后，你就有能力应对不同业务场景了。

1.2 工程化要解决的四类核心问题

第一是可复现性。同事跑你的代码，能不能得到一样的结果？一个月后你自己再跑，还能不能复现？如果答案是不确定，那项目就还没到工程化状态。可复现性的关键在于锁定环境依赖、固定随机种子、记录数据版本。

第二是自动化。从原始数据到特征、从训练到评估、从打包到部署，每一步都尽量脚本化。手工操作越少，出错概率就越低。我见过有人用Jupyter手动跑训练，每次都要手动切路径，这种流程一旦数据量上来就撑不住。

第三是可观测性。模型上线后不是万事大吉，要能知道它的输入分布有没有变、预测结果有没有异常、延迟和吞吐是否正常。没有监控的模型就是盲飞。

第四是迭代效率。业务变化了，新数据来了，模型多久能更新一版？如果每次更新都要手动跑一遍全流程，那就谈不上快速响应。工程化的目标是把“更新模型”这个动作从几天压缩到几小时甚至几分钟。

这四个问题，可以说就是“ai-engineering”这门手艺的骨架。后面所有的工具选择、流程设计，都是围绕它们展开的。

2. 从零搭建一套可复现的AI项目骨架

2.1 环境与依赖管理：先锁住版本再说别的

很多初学者喜欢“最新版”，但做工程恰恰要反着来——能用稳定版就别追新。我的习惯是，机器上统一装Python 3.10，不要用系统自带的Python，也不要多个Python版本混用，否则C扩展包编译会浪费大量时间。

推荐用conda建独立环境，再用pip-tools锁定全量依赖。具体做法分三步：

# 创建独立环境 conda create -n ai-eng python=3.10 -y conda activate ai-eng # 用requirements.in记录顶层依赖（只写直接引用的包） # 然后生成并冻结全量依赖 pip-compile requirements.in -o requirements.txt pip install -r requirements.txt

requirements.in里只写直接依赖，比如pandas、scikit-learn、fastapi、mlflow这些，pip-compile会自动解析它们的传递依赖并锁死版本。这个文件就是项目的“配方”，任何人都能通过它重建出完全一致的环境。

还有一个容易忽略的点：训练环境和推理环境的依赖往往不一致。训练要pytorch、sklearn，推理可能只要numpy和fastapi。可以把依赖拆成requirements-train.txt和requirements-serve.txt，镜像不至于变得臃肿，攻击面也小一些。

2.2 项目目录结构：让新人5分钟内看懂

目录结构讲究的就是“约定大于配置”。我常用的结构是这样的：

. ├── config/ # 所有配置项，YAML格式 ├── data/ # 原始数据与中间数据（git忽略） ├── features/ # 特征工程脚本 ├── models/ # 模型定义与训练脚本 ├── services/ # 推理服务代码 ├── tests/ # 单元测试与集成测试 ├── scripts/ # 流水线脚本 ├── notebooks/ # 探索性分析，只做分析不做训练 ├── requirements.in ├── requirements.txt └── README.md

关键原则是：notebook只用来探索，正式训练必须进Python脚本。notebook里跑代码过程序没法自动化，而且状态容易乱。我自己的习惯是，在notebook里确认思路后，立刻把逻辑整理成模块化函数放进features/或models/，notebook只保留分析图和结论。

config/目录下的YAML配置也值得多说一句。训练参数、数据路径、模型版本这些都应该写在配置里，而不是散落各处。这样换参数只需要改配置，不需要改代码。

# config/train_config.yaml data: raw_path: data/raw/train.jsonl processed_path: data/processed/train.parquet validation_split: 0.2 random_seed: 42 model: name: lgbm_text_classifier max_features: 50000 params: n_estimators: 500 learning_rate: 0.05 max_depth: 6 train: target_metric: f1_macro experiment_name: text_cls_baseline

2.3 数据版本管理：没有它就没有可复现性

代码版本有Git管，数据版本如果没人管，AI项目就失去了可复现的基础。之前遇到过一种情况：模型指标神秘上涨，查了半天发现是有人手动往数据文件里加了几千条样本，数据变了指标当然变了——但这种变化没人记录，等于白干。

数据版本管理有两种常见做法。简单场景用DVC（Data Version Control），它对标的是Git思路：数据存在本地或对象存储，DVC记录的是元数据和哈希。用起来就是两条命令：

dvc init dvc add data/raw/train.jsonl git add data/raw/train.jsonl.dvc

之后每次数据更新，dvc add都会重新计算哈希，Git里记录的是不同版本的指针，想回到某个数据版本只需要dvc checkout。对于更大规模的数据湖场景，可以引入lakeFS或者Delta Lake这类专门方案，但中小团队用DVC足够。

还有一个容易被忽略的点：特征数据也要版本化。原始数据只保存一份，特征工程改了之后，生成的特征表可能是另一个分布。最稳的方式是特征脚本加上版本号，每个特征文件命名带上哈希前缀，比如features_v3_8f3d2a.parquet，从文件名就能知道是哪版代码产出的。

2.4 实验追踪：把每一次尝试都记录下来

模型训练过程中，最怕的就是“感觉这个改得不错，但忘了具体改了什么”。实验追踪工具解决的就是这个问题。我自己常用MLflow，因为它轻量、跟语言无关，而且自带模型注册和部署功能。

在训练脚本里加几行代码就能接入：

import mlflow mlflow.set_experiment("text_cls_baseline") with mlflow.start_run(run_name="lgbm_v3"): # 记录参数 mlflow.log_params(config["model"]["params"]) # 记录指标 mlflow.log_metric("f1_macro", f1_macro) mlflow.log_metric("auc", auc) # 记录模型文件 mlflow.sklearn.log_model(model, artifact_path="model")

之后在MLflow UI里就能看到每一次实验的参数、指标和产物对比，选模型不再靠感觉，而是看数据。这不是仪式感，是工程化的基本要求：没有记录的实验等于没做。

这里有个实操建议：实验命名要有规律，比如lgbm_v3_k5_fold、bert_finetune_lr2e5。命名混乱会让后续对比成本急剧上升。我自己现在每次开新的实验前都会在群里说一句“跑xxx实验，模型xxx，数据版本xxx”，哪怕只是给自己留个记录，一周后回看也能迅速定位。

3. 核心链路实操：从原始日志到线上文本分类服务

3.1 一个贯穿全链路的具体例子

理论讲再多，不如走一遍完整流程。下面用“客服工单自动分类”这个例子演示：输入是原始客服对话文本，输出是工单类别（比如“退款”“物流咨询”“产品故障”），目标是让模型代替人工完成初步分拣。

原始数据大概是这种JSON结构：

{"text": "我上周买的手机壳到现在还没发货，能帮我催一下吗？", "label": "物流咨询", "id": "t-001"}

数据量不大，五千条左右，但类别分布不均衡，“产品故障”只有不到5%。这类问题做起来比大规模数据更有代表性，因为样本不均衡、标注噪声这些问题在真实业务里反而最常见。

3.2 数据清洗与特征构建：别急着上模型

拿到文本数据第一件事不是建模，而是搞清楚数据长什么样。我会先在notebook里做一轮探索性分析，重点看：文本长度分布、类别占比、是否有空值、是否有重复样本、标签是否存在错标。

清洗环节容易走极端，有人喜欢堆几十个规则，有人干脆什么都不做。我的原则是只清理明确有问题的部分，不过度清洗。具体来说就是：

• 统一文本格式：去掉HTML标签、多余空白、不可见字符，繁体转简体。
• 处理噪声样本：比如长度小于5个字符且属于“其他”类的样本，直接丢弃。
• 合并极小类别：占比低于2%的类别要么合并到相近类别，要么标记为“其他”，否则模型很难学。

特征方面，如果走传统机器学习路线，TF-IDF仍然是一个很强也很稳的baseline。用sklearn的TfidfVectorizer时，有几个参数值得注意：ngram_range=(1, 2)可以兼顾单词和短语，min_df=2过滤掉只出现一次的词，sublinear_tf=True对高频词做压缩。这些设置能明显提升特征质量。

但我最后实际线上用的往往不是TF-IDF加LR，而是先用TF-IDF做baseline，再用LightGBM或者微调一个小规模的BERT模型对比。原因在于，业务初期快速上线一个baseline比追求极致精度重要得多。线上跑起来的弱模型，价值永远大于线下跑不起来的强模型。

3.3 训练环节的工程细节：切分要随机但也要小心泄漏

数据切分是一个特别容易“翻车”的环节。很多教程里直接train_test_split不分青红皂白，但遇到时间序列相关的数据（比如按天产生的工单）就要按时间切分，否则你评估的是“未来数据训练，过去数据验证”这种倒挂场景，指标虚高没有参考价值。

以我们的工单数据为例，如果业务上模型要处理的是明天的新工单，那么训练集应该是昨天的数据，验证集应该是今天的数据，测试集最好是“未来一周”的数据。具体切分逻辑：

from sklearn.model_selection import train_test_split # 假设数据带时间字段 timestamp train_df, temp_df = train_test_split( df, test_size=0.3, random_state=42, stratify=df["label"] ) val_df, test_df = train_test_split( temp_df, test_size=0.5, random_state=42, stratify=temp_df["label"] )

stratify参数能保证每个集合的类别分布和原始数据一致。这里有个细节：要先切分再做特征工程，否则特征工程会提前“看到”测试集的信息。比如你在全量数据上计算TF-IDF的词表，然后切分，维度和取值其实已经包含了测试集的信息，这属于泄漏。

训练环节我自己习惯跑5折交叉验证来选参数，但最终模型用全量数据重新训练一次。方法就是：用小数据集的交叉验证确定参数，然后用全部数据以这些参数训练、保存、上线。这样既防止过拟合，又不浪费数据。

3.4 模型评估：不能只看一个指标

文本分类二分类可以看AUC，但多分类场景我建议重点盯F1 macro和混淆矩阵。F1 macro对每个类别的表现都给了同等权重，不会因为大类表现好就掩盖小类的问题。

举个例子，之前那个工单分类模型，整体准确率有92%，看起来不错，但一看类别明细，“产品故障”这个类别的F1只有0.31。原因很简单：这个类在训练集里样本太少。这种问题靠换更大的模型是解决不了的，要么补充标注数据，要么做类别权重调整，要么换评估方式。

调类别权重可以直接在LightGBM里做，给少数类的样本更大的权重，相当于让模型更“在意”它们：

import lightgbm as lgb # 根据类别样本量反比计算权重 from sklearn.utils.class_weight import compute_class_weight weights = compute_class_weight("balanced", classes=labels, y=train_y) sample_weight = [weights[int(i)] for i in train_y] lgb_train = lgb.Dataset(train_X, train_y, weight=sample_weight)

不要指望类别权重能创造奇迹，它的作用更多是让模型的预测分布不那么偏。真实业务场景里，“标注更多少数类样本”比任何算法技巧都有效，这点务必理解清楚。

3.5 模型服务化：FastAPI实现一个可用的推理接口

模型训练好只是开始，要让它对外服务，我用的方案是FastAPI加MLflow模型注册。FastAPI的优点在于原生异步支持、自动生成OpenAPI文档、类型校验方便，非常适合做模型服务的HTTP接口。

实际部署时，服务代码很简短，核心部分就是把模型加载进来做预测：

import pickle from fastapi import FastAPI from pydantic import BaseModel import numpy as np app = FastAPI() # 加载模型文件和预处理配置 with open("models/lgbm_text_cls.pkl", "rb") as f: model = pickle.load(f) with open("models/vectorizer.pkl", "rb") as f: vectorizer = pickle.load(f) class PredictRequest(BaseModel): text: str @app.post("/predict") def predict(req: PredictRequest): vec = vectorizer.transform([req.text]) proba = model.predict_proba(vec)[0] # 返回概率最高的类别和置信度 top_idx = int(np.argmax(proba)) return { "label": str(model.classes_[top_idx]), "confidence": float(proba[top_idx]), "probabilities": dict(zip(model.classes_, proba.tolist())), }

启动服务用uvicorn：

uvicorn services.app:app --host 0.0.0.0 --port 8080 --workers 2

这里的workers数量不是越大越好，它受限于GIL和模型对象大小。每个worker都会加载一份模型到内存，比如模型500MB，4个worker就是2GB。先算内存再定worker数，这是我多次踩过内存崩溃后的教训。

部署到生产环境我习惯用Docker，多阶段构建可以大幅减小镜像体积：

FROM python:3.10-slim as base WORKDIR /app COPY requirements-serve.txt . RUN pip install --no-cache-dir -r requirements-serve.txt FROM base as runtime COPY services/ ./services/ COPY models/ ./models/ EXPOSE 8080 CMD ["uvicorn", "services.app:app", "--host", "0.0.0.0", "--port", "8080"]

镜像里只装推理需要的包，不装训练相关的东西。这样镜像小、启动快、安全隐患也少。

4. 从上线到持续迭代：AI系统的运维与进化

4.1 模型监控：不盯着就一定会出问题

模型上线之后的第一周是最关键的。我在上线初期会盯三个监控项：

• 请求量与平均延迟：确认服务没有被打挂、响应速度正常。
• 预测分布：模型输出的类别分布如果出现明显偏移，比如原来“退款”类占30%，上线两天突然变成60%，很有可能是线上输入和训练分布不一致了。
• 置信度分布：如果大量请求的置信度低到0.3以下，说明模型遇到了“没见过”的输入。

监控用什么工具不是重点，Prometheus加Grafana是标配，但小团队可以直接用简单的日志统计加定时脚本。重要的是“有监控”而不是“用什么监控”。我自己初期没有搭完整监控体系，只是写了个脚本每小时统计一次请求量和预测分布，发现异常再人工介入，低成本就够用。

4.2 反馈数据回流与重训策略

模型不是一次性产品。业务变了、用户习惯变了，模型就会慢慢失效，这叫作模型漂移。对抗漂移的办法是持续收集反馈数据并定期重训。

反馈数据来源一般有两类：一是用户明确反馈“这个分类错了”，二是系统对低置信度预测做人工复核。这些数据会进入一个feedback_db，累积到一定量级后触发重训。

重训的触发条件可以这么设计：

• 每周定时重训一次，用最新一周的数据。
• 如果监控发现预测分布偏差超过阈值，立即触发重训。
• 新数据超过1000条且标注质量校验通过，触发一次增量训练。

重训流程走的就是第2章搭好的那套骨架：拉新数据 → 更新数据版本 → 跑特征工程 → 训练 → 评估 → 注册新模型 → 灰度发布。全流程脚本化之后，一次重训可以控制在十分钟内。

4.3 灰度发布与回滚：别让模型“硬切”

新模型上线不要直接替换旧模型，最好走灰度。最简单的灰度方式是：按请求流量比例路由，比如先放10%的流量到新模型，观察几小时，指标没问题再逐步放量。

实现上不需要什么高级框架，在FastAPI里加一段路由逻辑就可以：

import random # config里配置灰度比例 NEW_MODEL_RATIO = 0.1 def route_model(): return "new" if random.random() < NEW_MODEL_RATIO else "old"

这段逻辑当然不能上线后改代码，灰度比例应该放到配置中心或环境变量里。工程上的习惯是：任何策略调整都不能靠改代码完成，要能通过配置动态生效。

回滚也一样，模型服务的镜像上多个版本，注册中心切换版本就行。只要训练与服务分离、模型文件不写死在镜像里，回滚就是一次配置操作。

5. 常见问题与排查技巧实录

5.1 关键问题速查表

5.2 数据泄漏：最隐蔽也最致命的问题

数据泄漏通常不会报错，只会让你在离线评估时过于乐观。除了前面提到的先切分再特征工程，还有几个容易忽略的泄漏源：

• 正则化信息泄漏：用全量测试集计算文本长度均值去做归一化，测试集信息提前进入了训练过程。
• 去重不当泄漏：训练集和测试集里出现几乎相同的文本（比如客服标准回复模板），模型相当于“记住”了答案。
• 时间泄漏：BERT类模型在预训练时可能“见过”你的业务文本。这个没法完全避免，但评估时要注意长尾新词的表现。

我自己的做法是上线前做一次“模拟上线评估”：只用截止到某时间点的数据训练，然后预测那之后时间点的数据，看指标下滑幅度。这个数字更接近真实线上表现。

5.3 环境依赖冲突处理

AI项目里最常见的报错之一就是“Package conflict”。比如numpy版本太新会导致某个旧版scikit-learn报错，pandas和pyarrow版本对不上会导致parquet读写失败。

处理这个问题的关键不是“装最新”，而是“锁版本”。pip-compile生成的requirements.txt锁住了全部传递依赖的版本。如果后续要升级某个包，不要手动改版本号再跑pip install，正确的做法是改requirements.in，重新执行pip-compile生成新锁文件，再用pip install -r requirements.txt安装。

依赖升级还有一个经验法则：训练环境和部署环境分开之后，升级要分两次验证。先升级训练环境，跑一遍完整实验确认指标没问题，再升级部署环境，跑一遍服务测试和压测，两个环境都验证通过才算升级完成。

5.4 推理性能优化：每次上线前必做性能测试

文本分类这种计算量小的任务，性能瓶颈通常不在模型本身，而在输入处理和数据传输上。最简单的优化方案是把预处理结果缓存起来。

比如TF-IDF向量化如果每次都重新加载和计算，会很慢。正确做法是：把训练好的vectorizer和模型一起序列化保存，推理时直接transform，不重新训练也不重新加载词表。

我之前做过一次压测，修改前后单线程QPS从120提升到280。原因就是排查后发现旧代码每次请求都重新执行了正则清洗和停用词过滤，这些可以用缓存解决的计算被重复执行了。

6. 写在最后：如果让我重来一次

从零开始做AI工程，我最后悔的一件事是初期花了大量时间研究各种炫酷的模型结构，却在数据规范和流程设计上偷了懒。模型效果不满意可以换模型，但工程骨架不扎实，后续每一步都在还债。

如果让我给刚开始的人一句话，我会说：先把项目目录脚手架搭好、把数据和实验追踪做起来、把部署和监控链路跑通，然后再去调模型指标。顺序反了，回头补工程课的成本比一开始就按工程规范做要高得多。

另外说一个大家容易忽视的小习惯：写README。每完成一个阶段，把当前的数据情况、模型效果、已知问题、下一步计划记下来。这条看起来不“技术”，但半年后回看项目时，你会发现这些记录比任何代码注释都管用。AI工程是长跑，不是冲刺，把路标立清楚的人才能走得更远。