1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体框架而是一堆踩过的坑。过去几年里我参与过推荐系统、图像分类、文本抽取这几类项目几乎每一次都会遇到同一个尴尬模型在 Notebook 里跑得漂漂亮亮一旦要交给业务方用就立刻变成一团乱麻。数据管道是临时拼的特征处理散落在十几个脚本里模型版本靠文件名区分上线之后指标掉了还得靠人肉回滚。这套东西说白了就是“能跑就行”离真正的 AI 工程差着十万八千里。ai-engineering-from-scratch这个项目标题核心价值就在于它把视角从“训一个模型”拉高到了“搭一条能长期运转的工程链路”。它要解决的不是某个算法精度问题而是数据怎么进、特征怎么管、模型怎么训、版本怎么控、服务怎么出、监控怎么做这一整条链路的可复现问题。适合谁来参考我认为有三类人最该看一是刚转行做 AI 的工程师脑子里只有模型没有工程二是被“实验室模型”折磨过的后端或数据开发想搞清楚 AI 项目到底特殊在哪三是小团队里那个既要写模型又要管部署的“全干工程师”你需要一套能自己掌控、不依赖大厂平台的最小工程骨架。我打算按自己实际搭过的一套流程来讲从目录结构、数据契约、特征管理、训练编排、模型注册到服务与监控一层层拆开。里面会穿插我踩过的坑比如为什么我坚持把数据校验放在训练之前为什么模型注册表比想象中重要以及为什么“能复现”这三个字值多少钱。整套东西不依赖某个特定云厂商用 Python 生态里最常见的工具就能落地你可以直接抄作业也可以按自己团队情况裁剪。2. 整体架构设计与技术选型思路2.1 从“脚本思维”切换到“流水线思维”大多数人做 AI 项目的起点是一个train.py里面读数据、做特征、训模型、存权重一条龙。这种写法在探索阶段没问题但它有个致命缺陷每一步的输入输出都是隐式的。你改了特征处理逻辑没人知道旧模型是用哪版特征训的你换了个数据源也没人知道新数据和旧指标能不能比。ai-engineering-from-scratch要做的第一件事就是把这些隐式依赖全部显式化。我的做法是把整条链路拆成五个边界清晰的阶段数据接入、特征工程、训练编排、模型注册、在线服务。每个阶段只通过“契约”和下一阶段交互契约就是一份带版本号的 schema 或者配置。这样做的好处是任何一环出问题你都能定位到具体阶段而不是在一锅粥里捞针。举个实际例子有次线上 AUC 突然掉了 3 个点我顺着契约往回查发现是上游数据源某个字段的空值率从 2% 涨到了 40%而特征处理脚本对空值的填充策略没变导致大量样本被填成了默认值。如果没有显式的数据契约这种问题能查一整天。选型上我刻意保持克制。编排层用Prefect或者Dagster这类轻量工具而不是一上来就上 Kubeflow。原因很简单小团队维护不起重型平台而且很多平台把“魔法”藏得太深出问题你根本不知道它内部干了什么。特征存储我用Feast的本地模式起步够用且概念清晰。模型注册用MLflow它的 Tracking 和 Model Registry 两个模块几乎是小团队标配。服务层用FastAPI包一层简单直接。这套组合的学习曲线平缓每一层你都能看懂它在干什么这对“from scratch”这个定位至关重要。2.2 目录结构就是你的工程说明书我见过太多项目目录结构是“历史遗留”的产物新人进来根本不知道从哪看起。ai-engineering-from-scratch的目录设计我改过三版最终定下来是这样ai-engineering-from-scratch/ ├── configs/ # 所有配置按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── data/ # 数据契约与校验规则 │ ├── schemas/ │ └── expectations/ ├── features/ # 特征定义与计算逻辑 │ ├── definitions.py │ └── transformations.py ├── training/ # 训练编排 │ ├── pipeline.py │ └── tasks/ ├── registry/ # 模型注册与版本管理 │ └── model_card.py ├── serving/ # 在线服务 │ ├── app.py │ └── schemas.py ├── monitoring/ # 监控与告警 │ ├── metrics.py │ └── drift.py └── tests/ # 各层单测与集成测试这个结构的关键在于每个目录对应流水线的一个阶段且只暴露必要的接口。比如features/definitions.py里定义特征名、类型、来源transformations.py里写具体计算逻辑训练阶段只 import definitions不直接碰原始数据。这样特征逻辑变更时影响范围是可控的。我特别想强调configs/这一层很多人把配置硬编码在代码里结果换个环境就要改代码这是工程化的大忌。配置分层之后dev 和 prod 的差异只体现在 yaml 文件里代码完全一致复现性才有保障。2.3 数据契约被低估的工程基石数据契约这个词听起来很“重”但落地可以很轻。我的做法是在data/schemas/下为每个数据源写一份 schema 定义用Pydantic或者Pandera都行。schema 里声明字段名、类型、是否可空、取值范围。然后在数据接入阶段强制校验不通过就直接中断流水线而不是让脏数据流到下游。为什么我坚持“校验前置”因为 AI 项目里最贵的成本不是算力是调试时间。脏数据导致的模型异常往往要等到训练完、评估完才暴露那时候你已经烧了几小时 GPU。前置校验能在几秒内拦住问题。我踩过最惨的一次坑是某个 ID 字段混进了字符串类型的空值训练时被 pandas 静默转成了 NaN模型照常训完指标看着还行上线后才发现这批样本全部预测错误。从那以后我把数据校验做成了流水线的硬性关卡宁可误拦不可放过。3. 核心模块拆解与实操要点3.1 特征工程定义与计算必须分离特征工程最容易写成一团乱麻原因是大家习惯把“这个特征是什么”和“这个特征怎么算”混在一起写。ai-engineering-from-scratch里我强制把这两件事分开。definitions.py只做声明类似这样from feast import Entity, FeatureView, Field from feast.types import Float32, Int64 user Entity(nameuser_id, join_keys[user_id]) user_features FeatureView( nameuser_stats, entities[user], schema[ Field(nameorder_count_7d, dtypeInt64), Field(nameavg_amount_30d, dtypeFloat32), Field(namedays_since_last_order, dtypeInt64), ], source..., ttl..., )transformations.py里才写具体的 SQL 或者 pandas 计算逻辑。这样分离之后训练阶段只需要知道“我要用order_count_7d这个特征”不需要关心它是怎么算出来的。特征逻辑要改只动 transformationsdefinitions 里的接口保持不变下游训练代码零改动。这个设计我用了两年最大的收益是特征复用。同一个特征定义离线训练和在线服务都能引用避免了“训练用一套、线上用另一套”的经典事故。注意特征的时间窗口一定要显式声明。我见过太多人用“最近 7 天”这种模糊表述结果离线算的时候用的是 T-7 到 T线上算的时候用的是 T-7 到 T1差一天就能让指标对不上。窗口的起止必须精确到时间戳。3.2 训练编排让每一步都可追溯训练编排的核心不是“把任务串起来”而是“让每次运行都可追溯”。我用 Prefect 定义 pipeline每个 task 的输入输出都落盘或者记录到 MLflow。关键设计是每次训练运行都有一个唯一的 run_id所有中间产物都以 run_id 为前缀存储。这样你任何时候都能回答“这个模型是用哪份数据、哪版特征、哪套参数训出来的”。from prefect import flow, task import mlflow task def load_data(config): df read_source(config[data_source]) validate(df, config[schema]) return df task def compute_features(df, feature_defs): return transform(df, feature_defs) task def train_model(features, params): with mlflow.start_run() as run: mlflow.log_params(params) model fit(features, params) mlflow.log_metric(val_auc, evaluate(model)) mlflow.sklearn.log_model(model, model) return run.info.run_id flow def training_pipeline(config): df load_data(config) feats compute_features(df, config[features]) run_id train_model(feats, config[params]) return run_id这里有个细节值得说我把validate放在load_data里面而不是单独一个 task。原因是校验失败应该让整个 flow 立刻失败而不是产生一个“校验失败但继续跑”的中间状态。Prefect 的 task 失败会中断 flow这正是我要的行为。另外MLflow 的log_model会把模型和依赖环境一起打包这对复现至关重要。我曾经遇到过“本地能跑、服务器跑不了”的问题根源就是环境依赖没记录后来强制每次 log_model 都带上 conda 环境问题就消失了。3.3 模型注册别再用文件名管版本了模型注册这一步很多小团队直接跳过用model_v1.pkl、model_v2_final.pkl这种命名来管。我强烈建议不要这么干原因有三个第一文件名无法表达“这个模型是用什么数据训的”第二文件名无法表达“这个模型现在是不是线上版本”第三文件名无法做权限和审计。MLflow Model Registry 能解决这三个问题而且接入成本很低。我的做法是每次训练完把模型注册到 Registry并打上 stage 标签Staging、Production、Archived。同时写一份 model card记录训练数据范围、评估指标、已知偏差、负责人。这份 model card 在出问题时能救命。有次线上某个类别的召回率异常我翻 model card 发现训练数据里这个类别样本占比只有 0.3%属于长尾问题立刻定位了原因而不是盲目调参。注册项是否必填说明run_id是关联训练运行可追溯metrics是至少包含主指标和业务指标data_range是训练数据的时间范围feature_version是特征定义的版本号owner是负责人出问题能找到人known_issues否已知偏差或限制3.4 在线服务把模型包成稳定的接口服务层我用 FastAPI核心原则是接口稳定、内部可换。对外暴露的 API 只接受原始输入内部完成特征计算和模型推理。这样模型换版本时调用方无感知。接口的输入输出都用 Pydantic 定义 schema和训练阶段的数据契约保持一致避免“训练用一套字段、线上用另一套”的问题。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class PredictRequest(BaseModel): user_id: int context: dict class PredictResponse(BaseModel): score: float model_version: str app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): features compute_online_features(req.user_id, req.context) score model.predict(features) return PredictResponse(scorescore, model_versionMODEL_VERSION)这里有个实操心得响应里一定要带上 model_version。这样出问题时你能快速判断是哪个版本的模型在作妖而不是靠猜。另外在线特征计算要和离线特征计算共用同一套逻辑我通常把 transformations 抽成一个独立包离线和在线都 import 它从根上杜绝特征不一致。4. 完整实操流程与关键环节实现4.1 环境准备与依赖锁定第一步是把环境搭起来。我的习惯是用pyproject.toml管理依赖配合uv或者poetry做锁定。为什么强调锁定因为 AI 项目的依赖树很深numpy、pandas、torch 之间版本兼容性很敏感不锁定的话今天能跑的代码明天可能就崩。我踩过的坑是某次升级 pandas 大版本groupby的默认行为变了导致特征计算结果偏移模型指标掉了 1 个点查了两天才发现是依赖升级惹的祸。uv init ai-engineering-from-scratch uv add prefect mlflow feast fastapi pydantic pandera scikit-learn uv lock依赖锁定之后任何人 clone 下来执行uv sync都能得到完全一致的环境。这一步看似简单但它是“可复现”的第一道保障。我建议把 lock 文件纳入版本控制并且 CI 里强制检查 lock 文件是否和 pyproject 一致防止有人偷偷改了依赖没更新 lock。4.2 数据接入与校验的落地细节数据接入阶段我通常写一个read_source函数支持从数据库、对象存储、本地文件读取。关键是读取之后立刻做校验。校验规则我用 Pandera 定义因为它和 pandas 集成好报错信息也清晰。import pandera as pa from pandera import Column, DataFrameSchema, Check schema DataFrameSchema({ user_id: Column(int, Check.greater_than(0)), order_count_7d: Column(int, Check.greater_than_or_equal_to(0)), avg_amount_30d: Column(float, Check.in_range(0, 1e6)), event_time: Column(datetime64[ns]), }) validated schema.validate(raw_df, lazyTrue)lazyTrue这个参数很关键它会让校验收集所有错误再一次性抛出而不是遇到第一个错误就停。这样你能一次看到所有数据问题而不是修一个跑一次。校验失败时我会把失败样本落盘到data/rejected/目录方便排查。这个习惯帮我省了大量时间因为脏数据往往有模式看一眼失败样本就知道是上游哪个环节出了问题。4.3 特征计算与存储的实操特征计算我分成离线和在线两条路径但共用同一套 transformation 函数。离线用 Spark 或者 pandas 批量算结果写入 Feast 的离线存储在线用同样的函数输入是单条请求的上下文。Feast 的get_online_features会自动从在线存储取特征保证低延迟。from feast import FeatureStore store FeatureStore(repo_path./feature_repo) online_features store.get_online_features( features[user_stats:order_count_7d, user_stats:avg_amount_30d], entity_rows[{user_id: 12345}], ).to_dict()这里有个坑要提醒在线特征的时效性。Feast 的在线存储是物化的你需要定期把离线算好的特征同步到在线存储。同步频率取决于业务对时效的要求。我做过一个项目特征同步是每小时一次结果用户刚下的单在推荐里没体现被业务方投诉。后来改成事件驱动订单一产生就触发特征更新问题才解决。所以同步策略一定要和业务方对齐不能想当然。4.4 训练与评估的完整闭环训练阶段我把数据分成 train/val/test 三份train 用于拟合val 用于调参和早停test 只在最后评估一次。这个划分看起来是常识但我见过太多人反复用 test 调参导致指标虚高。划分时要注意时间维度如果是时序数据必须按时间切分不能随机切否则会引入未来信息泄露。def split_by_time(df, split_date): train df[df[event_time] split_date] val df[(df[event_time] split_date) (df[event_time] split_date pd.Timedelta(days7))] test df[df[event_time] split_date pd.Timedelta(days7)] return train, val, test评估指标我至少看三个主指标比如 AUC、业务指标比如转化率提升、稳定性指标比如不同时间段指标的方差。只看主指标容易过拟合加上业务指标和稳定性指标才能判断模型是不是真的可用。有次一个模型 AUC 涨了 0.5 个点但业务指标没动排查发现是样本权重设置有问题涨的那部分 AUC 来自不重要的样本。所以指标一定要多维度看。4.5 模型注册与上线的衔接训练完评估通过后把模型注册到 MLflow Registry并推进到 Staging。Staging 阶段我会跑一轮影子测试把新模型的预测和线上模型对比看差异是否在可接受范围。影子测试通过后再推进到 Production。这个流程能拦住大部分“指标好看但线上翻车”的情况。import mlflow client mlflow.tracking.MlflowClient() model_version client.create_model_version( nameuser_scorer, sourcefruns:/{run_id}/model, run_idrun_id, ) client.transition_model_version_stage( nameuser_scorer, versionmodel_version.version, stageStaging, )影子测试的实现方式是在服务层同时调用新旧两个模型但只返回旧模型的结果新模型的结果记录到日志里做对比。这样对用户无影响又能收集真实数据。我建议影子测试至少跑一个完整的业务周期比如一周覆盖工作日和周末的不同流量模式。5. 常见问题与排查技巧实录5.1 训练与线上指标不一致怎么查这是最经典的问题排查思路是从数据源头往下逐层对比。第一步确认线上请求的原始输入和离线训练数据的分布是否一致用 PSI 或者 KS 检验看特征漂移。第二步确认在线特征计算和离线特征计算的逻辑是否完全一致把同一条样本分别走两条路径对比中间结果。第三步确认模型版本是否一致看服务返回的 model_version 和 Registry 里的 Production 版本是否匹配。我遇到过一次排查到最后发现是在线特征计算里有个fillna(0)而离线计算里是fillna(mean)导致空值样本的特征值不同。这种问题靠肉眼 review 代码很难发现必须用同一条样本做端到端对比。所以我现在养成了一个习惯每次上线新特征都写一个对比测试确保离线和在线对同一条样本的输出完全一致。排查步骤检查内容常用工具1输入数据分布PSI、KS 检验2特征计算逻辑端到端对比测试3模型版本MLflow Registry4服务配置配置 diff5.2 数据漂移的监控与应对数据漂移是模型上线后最大的敌人。我的做法是在监控层定期计算训练数据和线上数据的特征分布差异超过阈值就告警。漂移分两种协变量漂移特征分布变了和概念漂移特征和标签的关系变了。前者容易监控后者需要标签回流才能发现。from scipy.stats import ks_2samp def detect_drift(train_feature, online_feature, threshold0.05): stat, p_value ks_2samp(train_feature, online_feature) return p_value threshold告警之后不要急着重训先分析漂移原因。如果是上游数据源变更导致的可能只需要调整特征处理如果是真实的业务变化才需要重训。我见过有人一告警就重训结果模型频繁切换反而影响了业务稳定性。漂移应对要有节奏不能一惊一乍。5.3 模型回滚的正确姿势模型出问题时回滚要快。我的做法是在 Registry 里保留最近 N 个 Production 版本回滚时只需要把旧版本重新推进到 Production服务层通过配置热加载切换。关键是回滚要能在分钟级完成而不是重新部署一遍。为此服务层启动时会把所有 Production 候选版本的模型都加载到内存切换时只改一个指针。注意回滚之后一定要保留现场把出问题的模型版本、当时的输入数据、日志都存档方便事后复盘。我见过有人回滚完就把日志清了结果同样的问题又犯了一次。5.4 小团队最容易忽略的三件事第一件是测试。AI 项目的测试不能只测模型精度还要测数据校验、特征计算、服务接口。我建议每个模块都有单测流水线有集成测试。第二件是文档。不是那种自动生成的 API 文档而是“为什么这么设计”的决策记录。第三件是成本监控。GPU 烧钱很快训练任务要有超时和资源上限防止一个死循环把预算烧光。这三件事在项目初期看不出价值但项目跑过半年之后它们决定了你是从容维护还是疲于救火。我现在接手任何 AI 项目第一件事就是看这三样东西有没有没有的话先补上再谈优化。6. 我在这套流程里踩过的坑和总结的经验搭这套ai-engineering-from-scratch流程前后迭代了大概一年半中间踩的坑能写一本书。挑几个最有代表性的说。第一个坑是过早引入重型平台。我一开始想上 Kubeflow结果光是搭环境就花了两周而且很多概念没吃透出了问题根本查不动。后来退回到 Prefect 这种轻量工具反而跑得更顺。这让我明白工具要匹配团队当前的成熟度超前一步是先进超前三步是灾难。第二个坑是特征逻辑复用不彻底。我一度以为把特征计算抽成函数就够了结果发现离线和在线的调用方式不同还是会出现不一致。后来强制要求离线和在线必须调用同一个函数且用同一份测试用例验证问题才根治。这件事的教训是工程约束要落到代码结构上光靠自觉是靠不住的。第三个坑是监控指标选得太多。我一开始恨不得把所有能算的指标都监控上结果告警天天响团队都麻木了。后来精简到三个核心指标主指标、数据漂移、服务延迟。告警少了但每个告警都是真问题响应效率反而高了。监控的目的是发现问题不是展示数据这个度要把握好。如果让我给刚起步的团队一个建议我会说先把数据契约和模型注册这两件事做扎实。这两件事投入不大但收益是长期的。数据契约保证你的输入可控模型注册保证你的输出可追溯。有了这两样后面加特征、换模型、调服务都是在这个稳固地基上做增量不会推倒重来。至于编排工具、特征存储这些可以随着团队规模慢慢升级不必一步到位。