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

1. 从零搭建AI工程体系为什么我劝你别一上来就调包ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、做项目最痛的一个点太多人把AI工程等同于会调API或者会跑个demo但真正落到生产环境里从数据管道到模型服务、从特征管理到监控告警中间那一大段工程化的活儿几乎没人系统讲过。我自己是从传统后端转过来的最早做推荐系统那会儿踩过的坑现在想起来还肉疼。模型离线AUC 0.85上线之后线上效果直接腰斩特征在训练脚本里算一遍、在服务端又算一遍两边逻辑对不上排查了整整三天模型文件从实验室的pkl换成线上要用的格式光是序列化兼容问题就折腾了一周。这些事儿没有一件是算法问题全是工程问题。所以这篇内容我想聊的不是某个具体模型怎么训而是从零开始把一套AI工程体系搭起来这件事本身。它适合谁看如果你是刚入行的算法工程师只会写notebook不会写服务这篇能帮你补上工程这一课如果你是后端或者数据工程师想转AI方向但不知道从哪下手这篇能给你一条清晰的路径如果你是技术负责人正发愁团队里模型能跑但上不了线这篇里的很多坑你应该都似曾相识。核心关键词就一个ai-engineering-from-scratch从零构建AI工程能力。我会按整体设计思路 → 核心模块拆解 → 实操落地 → 问题排查这条线来讲每一块都尽量给到能直接抄作业的细节而不是泛泛而谈。2. 整体架构怎么设计先想清楚数据流再谈技术选型2.1 为什么从零不等于从轮子造起很多人对from scratch有个误解以为是要自己手写一个TensorFlow。这就跑偏了。AI工程里的从零指的是从零搭建一套完整的工程链路而不是从零实现底层框架。这个区别特别重要因为它直接决定了你的技术选型策略。我的原则是底层框架和成熟组件直接用业务链路的胶水层自己写。什么意思PyTorch、TensorFlow、Spark、Kafka这些人家几万人维护你没必要重造但数据怎么从Kafka流到特征库、特征怎么喂给模型、模型输出怎么回写业务库这一整条链路的编排逻辑必须你自己掌控因为这是你业务的命脉用现成的一站式平台往往会被绑死。我见过太多团队一上来就买了个AI中台结果发现业务需求稍微一变平台改不动最后又退回自己搭。所以从零开始反而是最灵活的路子。2.2 一条完整的AI工程链路长什么样抛开具体技术任何一套AI工程体系本质上都是这么一条数据流原始数据 → 数据清洗与校验 → 特征工程 → 特征存储 → 模型训练 → 模型评估 → 模型注册 → 在线服务 → 监控与回流这条链路里离线部分清洗、特征、训练、评估和在线部分服务、监控是两套完全不同的技术栈但它们必须共享同一份特征定义和模型契约否则就会出现我前面说的线上线下不一致。我一般会把整个体系拆成四个层次来设计层次职责典型组件关键产出数据层采集、清洗、校验Kafka、Spark、Great Expectations干净的结构化数据特征层特征计算、存储、复用Spark、Feast、Redis特征视图与特征服务模型层训练、评估、注册PyTorch、MLflow、Airflow可追溯的模型版本服务层在线推理、监控、回流FastAPI、Triton、Prometheus稳定的推理接口这四层里特征层是最容易被忽视、但最要命的一层。很多团队数据层和模型层都做得不错唯独特征层是训练脚本里现算结果就是线上线下两套逻辑早晚出事。2.3 技术选型背后的取舍逻辑选型这事儿我不喜欢给标准答案因为每个团队的规模、人力、业务节奏都不一样。但我可以给你几条我踩过坑之后总结的判断标准。第一优先选能被替换的组件。比如特征存储Feast和自研Redis方案我倾向先用Feast因为它有标准接口将来想换底层存储不用改业务代码。反过来如果你一上来就深度绑定某个商业平台迁移成本会高到你想哭。第二离线在线尽量用同一套计算引擎。如果你的离线特征用Spark算在线特征用Python算那两套逻辑必然会有细微差异浮点精度、空值处理、时间窗口边界。我的做法是核心特征用同一份SQL或同一份Python函数定义离线批量跑、在线单条跑从源头保证一致性。第三别过早引入重组件。团队就三五个人业务量也不大非要上Kubernetes Triton 特征平台全家桶运维成本能把你拖垮。我一般建议日请求量百万级以下FastAPI Redis 单机模型服务就够了等真扛不住了再升级。3. 核心模块拆解数据、特征、模型、服务四件套3.1 数据层校验比清洗更重要数据层最容易犯的错是把精力全花在清洗上却忽略了校验。清洗是修数据校验是发现数据什么时候坏了。生产环境里数据源突然改字段、上游任务延迟、编码格式变化这些都会让你的模型悄悄失效而你还蒙在鼓里。我的做法是在数据入口加一道数据契约校验。用Great Expectations或者自己写一套简单的规则引擎对每一批进来的数据做检查# 一个极简的数据校验示例实际项目里我会用GE或pandera import pandas as pd def validate_batch(df: pd.DataFrame) - list: errors [] # 1. 关键字段不能为空 for col in [user_id, item_id, event_time]: if df[col].isnull().any(): errors.append(f{col} 存在空值) # 2. 数值范围检查 if (df[price] 0).any(): errors.append(price 出现负值) # 3. 时间不能是未来 if (pd.to_datetime(df[event_time]) pd.Timestamp.now()).any(): errors.append(event_time 出现未来时间) return errors这段代码看着简单但它救过我很多次。有一次上游把价格单位从元改成了分数值直接放大100倍就是靠范围校验第一时间发现的。校验规则要跟着业务走每加一个字段就加一条规则别嫌麻烦。注意校验失败不要直接丢弃数据而是告警 落盘到隔离区。丢弃会让你丢失现场事后根本查不出问题出在哪。3.2 特征层一致性是命根子特征层的核心矛盾就一个离线算的特征和在线算的特征必须一模一样。这个一模一样包括同样的输入、同样的计算逻辑、同样的空值处理、同样的时间窗口。我推荐的做法是特征定义即代码。把每个特征写成一个纯函数离线批量调用、在线单条调用共用同一份实现# features/user_features.py def avg_order_amount_7d(orders: list) - float: 近7天平均订单金额离线和在线共用 if not orders: return 0.0 # 空值统一返回0别用None否则线上线下行为不一致 valid [o[amount] for o in orders if o[amount] is not None] if not valid: return 0.0 return sum(valid) / len(valid)离线跑的时候你把一个用户近7天的订单列表传进去在线跑的时候你从Redis里取出这个列表再传进去。逻辑只有一份就不会有偏差。特征存储这块小团队我建议直接用Redis 一张MySQL元数据表。Redis存特征值key是feature:user_id:feature_nameMySQL存特征的元信息谁定义的、什么类型、更新频率。等特征数量上千、团队多人协作了再考虑上Feast这类专业平台。3.3 模型层可追溯比高精度更重要模型层我见过最混乱的场景是线上跑着一个模型但没人知道它是哪天训的、用的哪份数据、参数是什么。出了问题想回滚发现旧模型文件找不到了。这就是典型的缺乏模型管理。我的铁律是每一个上线的模型都必须能回答三个问题——用哪份数据训的、用什么代码训的、评估指标是多少。这三个问题答不上来就不许上线。实现上MLflow是性价比最高的选择。它帮你记录每次实验的参数、指标、产物还能做模型注册。一个典型的训练脚本长这样import mlflow import mlflow.pytorch mlflow.set_experiment(recommendation_model) with mlflow.start_run(): # 记录超参数 mlflow.log_params({lr: 0.001, batch_size: 256, epochs: 10}) # 记录数据版本 mlflow.log_param(data_version, 2024-01-15) # 训练... # 记录指标 mlflow.log_metrics({auc: 0.85, logloss: 0.32}) # 保存模型 mlflow.pytorch.log_model(model, model)跑完之后MLflow的UI里能清楚看到每次实验的对比。模型注册环节我会给通过评估的模型打上staging标签人工验证后再转production绝不自动上线。3.4 服务层接口设计决定后期维护成本服务层是模型和业务之间的桥梁它的设计好坏直接决定你后期改需求时是改一行还是改一周。我的接口设计原则有三条第一请求和响应都用明确的schema。别用裸dict用Pydantic定义清楚from pydantic import BaseModel from typing import List class PredictRequest(BaseModel): user_id: str item_ids: List[str] context: dict {} class PredictResponse(BaseModel): scores: List[float] model_version: str第二响应里必须带模型版本号。这样出问题时你能立刻定位是哪个版本也方便做A/B测试。第三推理逻辑和业务逻辑分离。服务层只负责取特征 → 调模型 → 返回分数至于这个分数怎么用、要不要过滤、要不要排序交给业务层。这样模型迭代时业务代码不用动。4. 实操落地从零跑通一条最小可用链路4.1 环境准备与目录结构理论讲再多不如跑一遍。我给你一条最小可用链路一台4核8G的机器就能跑起来。先看目录结构这个结构我用了好几年清晰且好扩展ai-engineering/ ├── data/ # 数据相关 │ ├── ingest.py # 数据接入 │ └── validate.py # 数据校验 ├── features/ # 特征定义离线在线共用 │ └── user_features.py ├── training/ # 训练 │ └── train.py ├── serving/ # 服务 │ ├── app.py │ └── feature_client.py ├── configs/ # 配置 │ └── config.yaml └── requirements.txt依赖就几个核心的pandas、scikit-learn、fastapi、uvicorn、redis、mlflow、pyyaml。别一上来就装一堆够用就行。4.2 数据接入与校验的实操假设我们的场景是用户下单预测数据从CSV来真实场景换成Kafka消费即可。接入脚本的核心是幂等——同一批数据重复跑结果不能变。# data/ingest.py import pandas as pd from data.validate import validate_batch def ingest(file_path: str) - pd.DataFrame: df pd.read_csv(file_path) # 去重保证幂等 df df.drop_duplicates(subset[order_id]) # 校验 errors validate_batch(df) if errors: # 落盘隔离区别丢 df.to_parquet(fdata/quarantine/{pd.Timestamp.now().date()}.parquet) raise ValueError(f数据校验失败: {errors}) return df这里有个细节去重放在校验之前。因为重复数据本身可能触发校验规则比如同一订单出现两次先去重能减少误报。这个顺序我调过好几次才定下来。4.3 特征计算与存储的实操特征计算我坚持一份逻辑两处调用。先定义特征函数然后写两个入口一个批量算离线一个单条算在线。# features/user_features.py def compute_user_features(orders: list) - dict: 输入用户订单列表输出特征字典 amounts [o[amount] for o in orders if o.get(amount) is not None] return { order_count_7d: len(orders), avg_amount_7d: sum(amounts) / len(amounts) if amounts else 0.0, max_amount_7d: max(amounts) if amounts else 0.0, }离线批量算完写进Redisimport redis, json r redis.Redis(hostlocalhost, port6379) def save_features(user_id: str, features: dict): r.set(ffeature:{user_id}, json.dumps(features), ex86400) # 24小时过期在线取的时候直接从Redis读读不到就返回默认值千万别现场去数据库捞会拖垮服务def get_features(user_id: str) - dict: raw r.get(ffeature:{user_id}) if raw is None: return {order_count_7d: 0, avg_amount_7d: 0.0, max_amount_7d: 0.0} return json.loads(raw)4.4 训练与模型注册的实操训练脚本的关键是可复现。固定随机种子、记录数据版本、记录代码commit这三样缺一不可。# training/train.py import numpy as np, mlflow from sklearn.ensemble import GradientBoostingClassifier from sklearn.metrics import roc_auc_score np.random.seed(42) # 固定种子 def train(X_train, y_train, X_val, y_val): with mlflow.start_run(): mlflow.log_param(data_version, 2024-01-15) model GradientBoostingClassifier(n_estimators100, max_depth5) model.fit(X_train, y_train) auc roc_auc_score(y_val, model.predict_proba(X_val)[:, 1]) mlflow.log_metric(auc, auc) mlflow.sklearn.log_model(model, model) return model, auc跑完之后去MLflow UI里看结果。AUC达标比如0.75才进入注册环节注册时打上版本号人工确认后转生产。4.5 在线服务的实操服务层用FastAPI核心是启动时加载模型请求时只做推理别每次请求都重新加载模型。# serving/app.py from fastapi import FastAPI from pydantic import BaseModel from typing import List import mlflow.pyfunc app FastAPI() model None app.on_event(startup) def load_model(): global model # 从模型注册中心加载生产版本 model mlflow.pyfunc.load_model(models:/recommendation_model/production) class PredictRequest(BaseModel): user_id: str item_ids: List[str] app.post(/predict) def predict(req: PredictRequest): features get_features(req.user_id) # 构造模型输入... scores model.predict(...) return {scores: scores.tolist(), model_version: v1.2.0}启动命令就一句uvicorn serving.app:app --host 0.0.0.0 --port 8000。压测一下单机QPS跑到几百没问题小业务完全够用。5. 常见问题与排查技巧实录5.1 线上线下效果不一致怎么查这是最高频的问题没有之一。排查思路我总结成一张表排查方向具体检查点常见原因特征一致性同一用户离线在线特征值是否相同空值处理不同、时间窗口边界不同数据分布线上请求的特征分布 vs 训练集分布训练数据过时、线上新用户多模型加载线上加载的是不是最新版本版本号没更新、缓存未刷新预处理归一化/编码逻辑是否一致离线用了fit的参数在线重新fit了我的排查顺序是先比对特征再比对分布最后查模型版本。90%的问题出在特征上。具体做法是挑10个线上请求把它们的特征值dump出来和离线算的对比一眼就能看出差异。5.2 模型服务内存泄漏怎么办服务跑几天内存就涨满重启才好这是典型的内存泄漏。常见原因有三个模型对象被反复加载检查是不是每次请求都load_model了必须放startup里。缓存无上限本地缓存比如LRU没设maxsize请求多了就爆。特征对象没释放大对象用完及时del别指望GC。我一般会在服务里加一个/health接口返回当前内存占用配合Prometheus做监控涨到阈值就告警。5.3 特征更新延迟导致预测偏差特征不是实时更新的比如你每小时批量刷一次Redis那这一小时内的新行为就反映不到特征里。这个延迟对某些业务比如实时推荐是致命的。解决办法有两个一是缩短批量刷新周期比如改成5分钟一次二是对关键特征做实时更新用户下单后立刻更新该用户的特征。我一般对最近一次行为这类特征做实时更新对7天统计这类做批量更新两者结合。提示实时更新特征时注意并发写问题。多个请求同时更新同一用户特征要用Redis的原子操作或者加锁否则会丢更新。5.4 模型回滚的正确姿势模型上线后发现效果不好要回滚。这时候如果没做版本管理就只能干瞪眼。我的做法是每次上线都保留上一个生产版本回滚就是改一个指针。在MLflow里把production标签从新版本移回旧版本服务端监听标签变化或者定时拉取自动加载旧模型。整个过程不超过1分钟。千万别用重新训练一个旧模型来回滚那既慢又不可靠。5.5 几个我踩过的独家坑坑一时间窗口用自然日还是滚动24小时。离线训练时我用的是自然日今天0点到昨天0点在线却用了滚动24小时结果特征对不上。后来统一成滚动窗口问题消失。时间窗口的定义必须写进特征文档全团队统一。坑二浮点数精度。离线用float64算在线用float32存Redis取出来再算结果有微小差异。对大多数模型无所谓但对某些敏感模型比如排序分数卡阈值就会出问题。统一用float32或者干脆存字符串。坑三空值语义。离线把没有订单处理成0在线却处理成None模型输入直接报错。空值的处理方式必须和特征定义绑死写在一个函数里。6. 后续可以怎么扩展这套体系跑通最小链路之后这套体系还有很多可以往上加的东西。我按优先级给你排个序你可以根据自己的业务节奏来。第一优先级加监控。模型上线不是终点是起点。至少要监控三个指标请求量、延迟、预测分布。预测分布尤其重要如果某天分布突然偏移说明上游数据或业务变了模型可能已经失效。用Prometheus Grafana半天就能搭起来。第二优先级加A/B测试。新模型别直接全量先切5%流量对比核心业务指标。A/B测试的框架可以很简单请求进来时按user_id哈希分流不同流量走不同模型版本结果分别打点。第三优先级加自动化训练。用Airflow或者简单的cron定期比如每天拉新数据、重训模型、自动评估达标就推到staging。这一步能把你从手动炼丹里解放出来。第四优先级特征平台化。当特征数量超过几百个、多个团队共用时就该上Feast这类专业平台了。但记住平台是为人服务的别为了平台而平台小团队用Redis 元数据表完全够用很久。我个人在实际操作中的体会是AI工程最难的不是某个技术点而是一致性和可追溯这两件事。把这两件事做好你的系统就能稳定运行做不好再牛的模型也白搭。从零搭建的过程其实就是不断和不一致作斗争的过程。每次你觉得这里应该没问题吧往往就是问题所在。多写校验、多留日志、多做对比这三件事看起来笨但最管用。