从零搭建AI工程体系：数据、特征、训练、服务全链路实战

1. 从零搭建AI工程体系，为什么我劝你别急着调包

"ai-engineering-from-scratch"这个标题，第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地，但绝大多数都是教你import torch然后跑个预训练模型，或者调个API接口就完事。真正从零开始、把AI工程当作一门系统工程来拆解的内容，少得可怜。

我自己在这个坑里摸爬滚打了几年，带过几个从零起步的团队，也见过太多人卡在"模型跑通了但上线就崩"的阶段。所以看到这个标题，我特别有感触——它戳中的不是"怎么训一个模型"，而是"怎么把AI这件事当作一个工程问题来对待"。

这篇文章适合谁看？如果你是刚转行做AI的开发者，或者带团队从传统软件转向AI产品的技术负责人，又或者是那种"模型指标很好看但线上效果一塌糊涂"的从业者，那接下来的内容应该能帮你省下不少试错成本。我会从整体设计思路、核心环节拆解、实操落地、问题排查几个维度，把"从零搭建AI工程"这件事讲透。

需要提前说明的是，AI工程不等于算法研究。算法研究关心的是"这个模型能不能在benchmark上刷到SOTA"，而AI工程关心的是"这个系统能不能在真实场景下稳定、可维护、可迭代地跑下去"。两者的思维方式完全不同，这也是为什么很多算法很强的团队，做出来的产品却一塌糊涂。

2. 整体架构设计：先想清楚数据怎么流，再想模型怎么选

2.1 为什么"从零"反而是一种优势

很多人觉得从零开始是劣势，没有积累、没有现成的轮子。但我的经验恰恰相反——从零搭建AI工程体系，最大的好处是你被迫想清楚每一个环节的必要性。

如果你接手的是一个已经跑了两年的系统，里面可能堆满了历史遗留的补丁：数据管道里塞了一堆硬编码的清洗规则，模型服务里嵌了业务逻辑，监控只覆盖了CPU和内存。你想改任何一处，都牵一发而动全身。

从零开始，你可以按照"数据→特征→训练→评估→部署→监控→反馈"这条主线，把每个环节的边界划清楚。我通常会把整个体系分成四层：

• 数据层：负责原始数据的采集、清洗、版本管理
• 特征层：负责特征的计算、存储、服务化
• 模型层：负责训练、评估、版本管理
• 服务层：负责推理、监控、反馈闭环

这四层之间通过明确的接口通信，任何一层的改动不应该影响其他层的内部实现。这个原则听起来简单，但实际操作中，90%的团队都会在"特征层"和"模型层"之间糊掉——训练时用的特征计算逻辑和线上服务时用的逻辑不一致，这是最经典的坑。

2.2 技术选型的核心逻辑：别追新，追可控

选型这件事，我的原则是：优先选择你能完全掌控的技术栈，而不是社区最火的那个。

举个例子，特征存储这块，市面上有Feast、Tecton等各种方案。但如果你团队只有三五个人，我建议直接用Redis加一套简单的元数据管理就够了。为什么？因为Feast的抽象层很厚，出问题的时候你排查成本极高。而Redis你至少知道它什么时候会挂、挂了怎么恢复。

模型服务这块也是同理。TorchServe、Triton、BentoML各有各的好，但如果你对推理性能没有极致要求，直接用FastAPI包一层PyTorch模型，反而更灵活。我见过太多团队为了用Triton，花了两周搭环境，结果发现自己的模型根本用不上动态batching。

这里给一个我常用的选型对照表，供参考：

这张表不是标准答案，而是一个思考框架。核心逻辑是：你的团队规模决定了你能承受的抽象层厚度。人越少，抽象层越薄越好。

2.3 数据流设计：把"训练-服务一致性"放在第一位

AI工程里最容易被低估的问题，就是训练和服务的数据处理不一致。训练的时候你用Pandas做特征，线上你用Java做特征，两边逻辑稍微有点差异，模型效果就会大打折扣。

我的做法是：特征计算逻辑只写一次，训练和服务共用同一套代码。具体来说，我会把特征计算封装成独立的Python模块，训练时直接调用，服务时通过一个轻量级的服务框架（比如FastAPI）暴露出去。这样虽然牺牲了一点性能，但换来的是逻辑一致性。

如果性能确实扛不住，那就退而求其次：用同一套配置文件和同一套测试用例来保证两边逻辑一致。每次改动特征逻辑，必须同时更新训练和服务的测试用例，CI里跑通才能合并。

3. 核心环节拆解：数据、特征、训练、服务，每个环节都有坑

3.1 数据层：别急着清洗，先做好版本管理

数据层最容易犯的错误，就是一上来就写清洗脚本。今天发现缺失值多了，加一条填充规则；明天发现某个字段格式不对，加一条正则替换。三个月后，你的清洗脚本变成了一团乱麻，没人敢改。

正确的做法是：先把原始数据原封不动地存下来，做好版本管理，然后再在上面做清洗。

原始数据存储这块，我建议用对象存储（比如S3兼容的MinIO），按日期和来源分目录。每次数据更新，打一个版本标签。这样你随时可以回到任何一个历史版本，复现当时的训练结果。

清洗逻辑则单独放在一个模块里，每个清洗步骤都有明确的输入和输出。我习惯用DVC来管理数据版本，配合一个简单的dvc.yaml定义清洗流水线：

stages: clean: cmd: python clean.py --input data/raw --output data/clean deps: - data/raw - clean.py outs: - data/clean

这样每次跑dvc repro，DVC会自动检查依赖是否变化，决定要不要重新执行清洗。数据版本和代码版本绑定在一起，复现实验的时候不会出现"数据对不上"的问题。

注意：原始数据永远不要覆盖。我见过团队为了省存储空间，清洗后直接覆盖原始数据，结果后来发现清洗逻辑有bug，想回滚都回不去。

3.2 特征层：离线在线一致性是生命线

特征层的核心挑战，就是离线特征和在线特征的一致性。离线训练时你用Spark算特征，在线服务时你用Redis查特征，两边如果对不上，模型效果直接崩盘。

我的解决方案是：特征定义用同一份配置，离线在线各自实现，但用同一套测试用例验证。

具体来说，我会定义一个特征配置文件，比如features.yaml：

features: user_avg_order_value: type: float source: orders computation: mean(amount) over last 30 days user_order_count_7d: type: int source: orders computation: count(*) over last 7 days

离线侧用Spark读取这个配置，生成特征计算任务；在线侧用Python读取同一个配置，生成Redis查询逻辑。然后写一套测试用例，用同一批模拟数据分别跑离线和在线，对比结果是否一致。

这套机制听起来有点重，但它是保证特征一致性的唯一可靠办法。我试过靠人工review来保证一致性，结果每次发版都提心吊胆，后来还是老老实实上了自动化测试。

3.3 训练层：实验管理不是可选项，是必选项

训练层最容易失控的地方，就是实验管理。今天调个学习率，明天换个网络结构，如果没有系统化的实验管理，两周后你根本记不清哪个模型对应哪组参数。

MLflow是我用得最顺手的实验管理工具。每次训练自动记录参数、指标、模型文件，还可以把模型注册到Model Registry里，标记哪个版本可以上线。

一个典型的训练脚本大概长这样：

import mlflow import mlflow.pytorch with mlflow.start_run(): mlflow.log_params({"lr": 0.001, "batch_size": 64, "epochs": 10}) for epoch in range(10): train_loss = train_one_epoch(model, train_loader) val_loss = validate(model, val_loader) mlflow.log_metrics({"train_loss": train_loss, "val_loss": val_loss}, step=epoch) mlflow.pytorch.log_model(model, "model")

这样每次训练完，你都能在MLflow UI里看到完整的实验记录。更重要的是，模型注册后，服务层可以直接从Registry拉取指定版本的模型，不需要手动拷贝文件。

实操心得：实验命名一定要有规范。我习惯用{项目名}-{日期}-{改动点}的格式，比如recommend-20240115-add-user-features。这样半年后回头看，还能快速定位到某次实验的上下文。

3.4 服务层：推理性能不是唯一指标

服务层这块，很多人一上来就盯着QPS和延迟。当然这很重要，但我想说的是：可观测性比性能更重要。

一个推理服务，如果QPS很高但出了问题时你完全不知道哪里出了错，那这个服务就是不可维护的。我通常会在服务层加三类监控：

• 基础监控：CPU、内存、GPU利用率、请求量、延迟分布
• 业务监控：输入特征分布、输出预测分布、异常输入比例
• 模型监控：预测置信度分布、特征缺失率、模型版本

业务监控和模型监控是很多人会忽略的。举个例子，如果某天输入特征的分布突然偏移了（比如用户年龄字段突然全是0），模型预测结果肯定会出问题。如果你只监控了CPU和延迟，根本发现不了这个问题。

我习惯用Prometheus加自定义指标来实现这套监控。比如在FastAPI里加一个中间件，记录每次请求的输入特征统计信息：

from prometheus_client import Histogram feature_age_hist = Histogram('feature_age', 'Distribution of user age') @app.middleware("http") async def monitor_features(request, call_next): body = await request.json() if 'age' in body: feature_age_hist.observe(body['age']) response = await call_next(request) return response

这样在Grafana里就能看到特征分布的变化趋势，一旦发现异常可以及时告警。

4. 实操落地：从零搭建一个可运行的AI工程原型

4.1 环境准备与项目结构

说了这么多理论，接下来我带你从零搭一个可运行的AI工程原型。这个原型包含数据清洗、特征计算、模型训练、模型服务、监控五个环节，代码量不大，但结构完整。

先看项目结构：

ai-engineering-from-scratch/ ├── data/ │ ├── raw/ │ └── clean/ ├── features/ │ ├── features.yaml │ ├── offline.py │ └── online.py ├── training/ │ ├── train.py │ └── evaluate.py ├── serving/ │ ├── app.py │ └── monitor.py ├── tests/ │ └── test_feature_consistency.py ├── dvc.yaml └── requirements.txt

这个结构的关键在于：features/目录下的offline.py和online.py共用同一份features.yaml配置，tests/目录下的测试用例保证两边逻辑一致。

环境准备很简单，Python 3.9以上，装几个核心依赖：

pip install pandas scikit-learn fastapi uvicorn prometheus-client dvc mlflow

4.2 数据清洗与特征计算

假设我们的场景是用户购买行为预测，原始数据是一份订单记录CSV。清洗逻辑很简单：去掉金额为负的记录，填充缺失的用户年龄。

# clean.py import pandas as pd def clean_data(input_path, output_path): df = pd.read_csv(input_path) df = df[df['amount'] > 0] df['age'] = df['age'].fillna(df['age'].median()) df.to_csv(output_path, index=False) if __name__ == "__main__": clean_data("data/raw/orders.csv", "data/clean/orders.csv")

特征计算这块，离线用Pandas，在线用Redis查询，但都读取同一份features.yaml：

# features/offline.py import pandas as pd import yaml def compute_features(df, config_path="features/features.yaml"): with open(config_path) as f: config = yaml.safe_load(f) features = pd.DataFrame() features['user_avg_order_value'] = df.groupby('user_id')['amount'].mean() features['user_order_count_7d'] = df.groupby('user_id')['amount'].count() return features

在线侧的逻辑类似，只是数据源从DataFrame变成了Redis：

# features/online.py import redis import yaml r = redis.Redis(host='localhost', port=6379) def get_features(user_id, config_path="features/features.yaml"): with open(config_path) as f: config = yaml.safe_load(f) features = {} features['user_avg_order_value'] = float(r.get(f"user:{user_id}:avg_order_value") or 0) features['user_order_count_7d'] = int(r.get(f"user:{user_id}:order_count_7d") or 0) return features

4.3 模型训练与评估

训练脚本用MLflow记录实验，模型用scikit-learn的随机森林：

# training/train.py import mlflow import mlflow.sklearn import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score def train(): df = pd.read_csv("data/clean/orders.csv") features = pd.read_csv("data/clean/features.csv") X = features y = df['is_high_value'] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2) with mlflow.start_run(): model = RandomForestClassifier(n_estimators=100, max_depth=5) model.fit(X_train, y_train) preds = model.predict(X_test) acc = accuracy_score(y_test, preds) mlflow.log_param("n_estimators", 100) mlflow.log_param("max_depth", 5) mlflow.log_metric("accuracy", acc) mlflow.sklearn.log_model(model, "model") print(f"Accuracy: {acc}") if __name__ == "__main__": train()

评估这块，除了准确率，我还会看特征重要性和混淆矩阵。特征重要性可以帮你判断模型是不是学到了合理的模式，混淆矩阵可以帮你判断模型在哪个类别上表现差。

4.4 模型服务与监控

服务层用FastAPI包一层，加载MLflow注册的模型：

# serving/app.py import mlflow.pyfunc from fastapi import FastAPI from features.online import get_features from prometheus_client import Histogram, make_asgi_app app = FastAPI() model = mlflow.pyfunc.load_model("models:/order_value_model/Production") prediction_hist = Histogram('prediction_score', 'Distribution of prediction scores') @app.post("/predict") def predict(user_id: str): features = get_features(user_id) input_df = pd.DataFrame([features]) score = model.predict(input_df)[0] prediction_hist.observe(score) return {"user_id": user_id, "score": float(score)} app.mount("/metrics", make_asgi_app())

监控这块，除了Prometheus的基础指标，我还加了一个预测分数分布的直方图。如果某天预测分数突然集中到某个值，说明输入特征可能出了问题，需要排查。

4.5 特征一致性测试

最后是特征一致性测试，这是保证离线在线一致的关键：

# tests/test_feature_consistency.py import pandas as pd from features.offline import compute_features from features.online import get_features def test_feature_consistency(): df = pd.DataFrame({ 'user_id': ['u1', 'u1', 'u2'], 'amount': [10.0, 20.0, 30.0] }) offline_features = compute_features(df) # 模拟在线特征（实际测试中需要先写入Redis） online_features = { 'u1': get_features('u1'), 'u2': get_features('u2') } for user_id in offline_features.index: for feature_name in offline_features.columns: offline_val = offline_features.loc[user_id, feature_name] online_val = online_features[user_id][feature_name] assert abs(offline_val - online_val) < 1e-6, \ f"Mismatch for {user_id}.{feature_name}: {offline_val} vs {online_val}"

这个测试跑通，才能保证上线后模型效果和离线评估一致。

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

5.1 模型上线后效果暴跌，怎么排查

这是最经典的问题。模型离线AUC 0.85，上线后业务指标反而下降。排查思路按以下顺序来：

1. 检查特征一致性：跑一遍特征一致性测试，看离线在线特征是否对齐。这是最常见的原因，占我遇到问题的60%以上。
2. 检查数据分布偏移：对比训练数据和线上推理数据的特征分布。如果线上某个特征分布明显偏移，说明训练数据不能代表线上场景。
3. 检查标签定义：有时候离线标签和业务指标的定义不一致。比如离线用"7天内是否复购"作为标签，但业务关心的是"30天GMV"，两者优化目标不同。
4. 检查服务延迟：如果推理延迟过高，可能导致请求超时，实际生效的请求比例很低。

我整理了一个排查速查表：

5.2 特征计算太慢，训练等不起怎么办

特征计算慢是另一个高频问题。我的优化思路分三步：

第一步，增量计算。不要每次全量重算，只计算新增数据对应的特征。比如用户过去30天订单数，只需要在昨天的基础上加上今天的新订单，减去30天前的旧订单。

第二步，并行化。用Spark或者Dask把特征计算并行化，按用户ID分片。这一步通常能带来10倍以上的加速。

第三步，预计算。把常用的特征提前算好存到Redis或者特征存储里，训练时直接读取。代价是存储成本增加，但训练速度会快很多。

实操心得：特征计算优化不要一步到位。先跑通全量计算，确认逻辑正确，再做增量化和并行化。我见过团队一上来就搞增量计算，结果逻辑写错了，排查了两周才发现。

5.3 模型版本管理混乱，怎么治理

模型版本管理混乱的典型表现是：线上跑的是哪个模型没人知道，回滚的时候找不到上一个版本。

治理方案很简单：所有模型必须通过Model Registry管理，禁止手动拷贝模型文件。

MLflow的Model Registry可以给模型打标签，比如Staging、Production、Archived。服务层加载模型时，只认Production标签，不认具体版本号。这样回滚的时候，只需要把上一个版本重新标记为Production，服务层自动加载新版本。

from mlflow.tracking import MlflowClient client = MlflowClient() client.transition_model_version_stage( name="order_value_model", version=3, stage="Production" )

这套机制配合CI/CD，可以实现模型的自动化发布和回滚。

5.4 监控告警太多，怎么降噪

监控告警太多，最后大家都不看了，这是监控体系失效的开始。降噪的核心思路是：只对可行动的问题告警。

什么叫可行动？就是收到告警的人知道该做什么。比如"CPU利用率超过90%"是可行动的，因为可以扩容；但"某个特征均值偏移了0.1%"就不可行动，因为不知道要不要处理。

我的做法是分两级告警：

• P0告警：服务不可用、错误率超过阈值、延迟超过阈值。这类告警直接打电话。
• P1告警：特征分布偏移、预测分数异常、模型版本不一致。这类告警发到群里，工作时间处理。

P1告警的阈值不要设得太敏感，否则每天几十条告警，大家很快就麻木了。我通常会用过去7天的数据计算一个基线，只有偏移超过3个标准差才告警。

6. 一些踩坑之后的个人体会

从零搭建AI工程体系这件事，我最大的体会是：工程能力比算法能力更稀缺。

我见过太多算法很强的团队，模型指标刷得很高，但工程体系一塌糊涂，上线就崩。也见过算法一般但工程扎实的团队，模型效果虽然不是顶尖，但系统稳定可靠，业务价值反而更大。

如果你正在从零搭建AI工程体系，我的建议是：先把数据层和特征层做扎实，这两层是地基。模型层和服务层可以先用最简单的方案跑通，后面再逐步优化。不要一上来就追求大而全的架构，那样很容易陷入"架构很漂亮但跑不起来"的困境。

另外，测试和监控一定要从第一天就加上。我试过先跑通再补测试，结果补测试的时候发现一堆隐藏问题，改起来比从头写还麻烦。特征一致性测试、模型版本管理、基础监控，这三样是底线，不能省。

最后分享一个我常用的检查清单，每次上线新模型前过一遍：

• 特征一致性测试是否通过
• 模型是否注册到Model Registry并标记为Production
• 服务层是否加载了正确的模型版本
• 基础监控和业务监控是否覆盖
• 回滚方案是否准备好

这个清单不长，但能挡住80%的上线事故。