1. 从零搭建AI工程体系为什么我劝你别急着调包ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都在教你import torch然后跑一个预训练模型真正从零把AI工程这条链路搭起来的内容少得可怜。我自己带过几个刚入行的同学发现一个特别普遍的现象他们能背出Transformer的结构图能说清楚注意力机制的公式但你让他从数据采集开始一路走到模型上线服务中间每一步该做什么、为什么这么做基本是懵的。这就是from scratch的价值所在。它不是让你从零手写一个PyTorch而是让你从零理解并搭建一套完整的AI工程体系——数据怎么进来、特征怎么处理、模型怎么训练、实验怎么管理、服务怎么部署、线上怎么监控。这套东西才是真正区分会调包和能做AI工程的分水岭。我写这篇东西是想把我在实际项目里踩过的坑、总结出来的流程完整地摊开讲一遍。适合谁看如果你已经会写Python懂一点机器学习基础但从来没独立负责过一个从数据到上线的完整AI项目那这篇就是给你准备的。如果你已经做过几个项目但总觉得流程很乱、每次都在重复造轮子那这篇也能帮你把工程化的思路理清楚。核心关键词就一个ai-engineering-from-scratch。我会围绕它把整个AI工程从零搭建的骨架、血肉、神经都讲透。2. 整体架构设计先想清楚数据怎么流再动手写代码2.1 为什么我坚持数据流优先的设计原则很多人做AI项目上来就打开Jupyter Notebook开始pd.read_csv然后一顿model.fit。这种做法的后果是等到项目要上线了发现数据处理逻辑散落在十几个notebook里训练脚本和推理脚本用的预处理代码不一致线上效果和离线评估对不上。我见过最离谱的一个案例离线AUC 0.85上线之后掉到0.62排查了三天才发现是特征工程里一个归一化参数在训练时用了全量数据统计推理时用的是单条数据统计。所以我的第一条经验就是先画数据流图再写代码。数据从哪来、经过哪些处理、存到哪里、训练时怎么读、推理时怎么读这条链路必须在动手之前就想清楚。具体来说我会把整个AI工程分成五个层次数据层负责原始数据的采集、存储、版本管理特征层负责数据清洗、特征提取、特征存储训练层负责模型定义、训练循环、实验管理服务层负责模型打包、API封装、推理优化监控层负责线上指标采集、数据漂移检测、模型重训触发这五层不是随便分的每一层都有明确的输入输出契约。数据层的输出是带版本号的原始数据集特征层的输出是特征向量加特征元数据训练层的输出是模型文件加训练配置服务层的输出是HTTP/gRPC接口监控层的输出是告警和重训信号。层与层之间通过标准化的接口通信任何一层的实现替换不影响其他层。2.2 技术选型的取舍逻辑别为了用新工具而用新工具说到工具选型我特别想吐槽一种现象有些人一提到AI工程张口就是Kubeflow、MLflow、Feast、Airflow全套上齐。工具本身没问题但如果你是一个三五人的小团队或者你只是想先把一个MVP跑通上这么多组件就是给自己找罪受。我的选型原则很简单先用最少的组件跑通闭环再根据瓶颈逐步替换。具体到每个层层次起步方案进阶方案替换触发条件数据层本地文件Git LFS对象存储DVC数据量超过10GB或多人协作特征层Pandas脚本Feast/Feast-like特征复用超过3个项目训练层单机脚本TensorBoardMLflow分布式训练单次训练超过4小时服务层FlaskPickleFastAPIONNX RuntimeQPS超过50或延迟要求100ms监控层日志定时脚本PrometheusGrafana线上模型超过2个这个表是我自己项目里实际用过的演进路径。起步阶段用FlaskPickle听起来很土但它能让你在一天之内把整个闭环跑通。等你发现Pickle加载慢、Flask并发差的时候再换ONNX Runtime和FastAPI这时候你已经有足够的理由和数据支撑这个决策了。注意不要一开始就追求生产级。我见过太多项目死在过度设计上架构图画得漂漂亮亮三个月过去了连一个能跑的demo都没有。2.3 目录结构设计让项目自己会说话一个从零搭建的AI工程项目目录结构应该让人一眼就能看出数据流。我常用的模板是这样的ai-project/ ├── configs/ # 配置文件按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── data/ │ ├── raw/ # 原始数据只读 │ ├── interim/ # 中间处理结果 │ └── processed/ # 最终训练数据 ├── src/ │ ├── data/ # 数据加载与处理 │ ├── features/ # 特征工程 │ ├── models/ # 模型定义 │ ├── training/ # 训练循环 │ ├── serving/ # 推理服务 │ └── monitoring/ # 监控脚本 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析不参与生产 ├── tests/ # 单元测试与集成测试 └── scripts/ # 一键运行脚本这个结构的关键在于data/raw永远只读任何处理都写到interim或processednotebooks里的代码不允许被src引用configs里的参数不允许硬编码在代码里。这三条规矩看起来简单但能避免90%的这个结果是怎么来的问题。3. 核心细节解析数据、特征、训练三层的实操要点3.1 数据层版本管理不是可选项是必选项数据版本管理这件事我是吃了大亏才重视起来的。早期做一个推荐项目训练数据每周更新一次有一次线上效果突然下降想回滚到上一版模型结果发现上一版模型对应的训练数据已经被覆盖了根本没法复现。从那以后我强制要求所有训练数据必须带版本号。具体做法很简单每次数据更新生成一个data_version格式是日期_哈希前8位比如20240115_a3f2b1c9。这个版本号会写入训练配置训练产出的模型文件里也记录这个版本号。这样任何时候都能通过模型文件反查到训练数据。数据存储方面小数据量1GB直接用文件系统加Git LFS就够了。数据量大了之后我推荐用对象存储配合DVC做版本控制。DVC的好处是它不把数据本身放进Git只存元数据和哈希值拉取的时候按需下载。# DVC初始化与数据版本管理 dvc init dvc add data/raw/train.csv git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m add training data v20240115实操心得DVC的远程存储配置一定要在项目初期就做好不要等到数据几十个GB了才想起来配。我一般用对象存储的桶作为DVC remote团队每个人都能拉取。3.2 特征层训练和推理必须共用同一套代码特征工程最大的坑就是训练时用一套代码推理时用另一套代码。这两套代码只要有一点不一致线上效果就会打折扣。我的解决方案是特征计算逻辑只写一次训练和推理都调用同一个函数。具体来说我会把每个特征的计算封装成一个纯函数输入是原始数据字典输出是特征值。这个函数不依赖任何全局状态不依赖数据库连接所有需要的参数都通过参数传入。# src/features/feature_defs.py def compute_user_avg_amount(user_history, window_days30): 计算用户近30天平均交易金额 cutoff datetime.now() - timedelta(dayswindow_days) recent [t for t in user_history if t[timestamp] cutoff] if not recent: return 0.0 return sum(t[amount] for t in recent) / len(recent)训练时我批量调用这个函数推理时我实时调用这个函数。唯一的区别是数据来源不同但计算逻辑完全一致。为了进一步保证一致性我会写一个测试用同一批原始数据分别走训练路径和推理路径断言输出的特征向量完全相等。特征存储方面如果特征计算耗时较长比如需要查多个表做聚合我会引入特征存储把计算结果缓存起来。但起步阶段直接用Redis缓存就够了没必要上Feast。3.3 训练层实验管理从第一天就要做训练层最容易失控的地方是实验管理。你改了学习率、换了优化器、调了batch size跑出来一堆模型文件过了一周根本记不清哪个模型对应哪组参数。我的做法是每次训练必须生成一个实验记录包含完整的配置和指标。最轻量的方案是用TensorBoard加一个JSON配置文件。每次训练启动时把配置写进experiments/{experiment_id}/config.json训练过程中的指标写进TensorBoard最终模型存到experiments/{experiment_id}/model.pkl。experiment_id用时间戳加随机后缀生成。# src/training/train.py import json, uuid, datetime def train(config): exp_id f{datetime.datetime.now():%Y%m%d_%H%M%S}_{uuid.uuid4().hex[:6]} exp_dir Path(fexperiments/{exp_id}) exp_dir.mkdir(parentsTrue) with open(exp_dir / config.json, w) as f: json.dump(config, f, indent2) # ... 训练逻辑 ... # 保存模型和指标 torch.save(model.state_dict(), exp_dir / model.pt) with open(exp_dir / metrics.json, w) as f: json.dump(final_metrics, f, indent2) return exp_id这套方案土是土了点但它保证了任何一次训练都可追溯。等你实验数量超过50个再考虑上MLflow也不迟。注意随机种子一定要固定并记录。我遇到过两次结果不可复现的情况都是因为忘了固定种子。现在我的配置里强制包含seed字段训练脚本第一行就是set_seed(config[seed])。4. 实操过程从零跑通一个完整AI项目的七个步骤4.1 第一步环境准备与依赖锁定环境准备听起来简单但它是团队协作中最容易出问题的地方。我强烈建议用conda创建独立环境并且用pip-compile锁定依赖版本。# 创建环境 conda create -n ai-project python3.10 -y conda activate ai-project # 安装基础依赖 pip install numpy pandas scikit-learn torch flask # 锁定依赖 pip freeze requirements.txtrequirements.txt里必须包含所有依赖的精确版本号比如numpy1.24.3不能写numpy1.24。我吃过这个亏本地开发时numpy是1.24服务器上自动装了1.26结果某个API行为变了排查了半天。4.2 第二步数据采集与初步清洗数据采集的方式取决于数据源。如果是数据库我一般用SQLAlchemy写一个抽取脚本支持增量抽取。如果是文件直接读进来做初步清洗。初步清洗要做的事情去重、处理缺失值、类型转换、异常值标记。这一步不要做太复杂的特征工程只做最基本的清洗把数据变成干净但原始的状态。# src/data/clean.py def basic_clean(df): # 去重 df df.drop_duplicates(subset[id]) # 缺失值处理 df[age] df[age].fillna(df[age].median()) df[category] df[category].fillna(unknown) # 类型转换 df[timestamp] pd.to_datetime(df[timestamp]) # 异常值标记不删除只标记 df[amount_outlier] (df[amount] df[amount].quantile(0.999)).astype(int) return df异常值我倾向于标记而不是删除因为删除可能会丢失重要信息而且删除逻辑本身也可能有偏。4.3 第三步特征工程与特征验证特征工程是AI项目里最耗时的部分通常占整个项目60%以上的时间。我的做法是先做特征探索确定哪些特征有用再固化成特征函数。特征验证这一步很多人会跳过但它极其重要。我会对每个特征做三件事检查分布是否合理、检查训练集和验证集的分布是否一致、检查特征与标签的相关性。# 特征分布检查 def validate_feature(train_df, val_df, feature_name): train_mean train_df[feature_name].mean() val_mean val_df[feature_name].mean() diff_ratio abs(train_mean - val_mean) / (abs(train_mean) 1e-8) if diff_ratio 0.1: print(f警告特征{feature_name}训练集与验证集分布差异过大{diff_ratio:.2%})如果某个特征在训练集和验证集上分布差异超过10%我会重点排查是不是数据划分有问题或者这个特征本身不稳定。4.4 第四步模型训练与超参数调优模型训练的第一步是建立一个baseline。baseline不需要多复杂逻辑回归或者简单的决策树就够了。baseline的作用是给你一个参照系后面所有复杂模型都要跟它比。超参数调优我推荐用Optuna它比GridSearch和RandomSearch都高效。但要注意调优的搜索空间不要设太大否则容易过拟合验证集。# 用Optuna调优 import optuna def objective(trial): lr trial.suggest_float(lr, 1e-4, 1e-2, logTrue) batch_size trial.suggest_categorical(batch_size, [32, 64, 128]) hidden_dim trial.suggest_int(hidden_dim, 64, 256, step64) model build_model(hidden_dim) metric train_and_eval(model, lr, batch_size) return metric study optuna.create_study(directionmaximize) study.optimize(objective, n_trials50)50次试验通常就够了超过100次边际收益很低而且容易过拟合验证集。4.5 第五步模型评估与选择模型评估不能只看一个指标。分类问题我至少看AUC、F1、Precision、Recall四个指标回归问题看MAE、RMSE、R²。而且一定要看混淆矩阵或者残差分布这些能暴露单一指标掩盖的问题。模型选择的原则是在满足业务约束的前提下选最简单的模型。如果逻辑回归和深度模型效果差不多选逻辑回归因为推理快、可解释、维护成本低。4.6 第六步模型服务化与API封装服务化最核心的要求是推理逻辑与训练逻辑一致。我会把特征计算函数抽出来服务端直接调用同一个函数。# src/serving/app.py from flask import Flask, request, jsonify from src.features.feature_defs import compute_features import pickle app Flask(__name__) model pickle.load(open(experiments/best/model.pkl, rb)) app.route(/predict, methods[POST]) def predict(): raw_data request.json features compute_features(raw_data) prediction model.predict([features])[0] return jsonify({prediction: float(prediction)})服务化之后一定要做压力测试用locust或者wrk测一下QPS和P99延迟。如果延迟不达标再考虑用ONNX Runtime或者TensorRT优化。4.7 第七步线上监控与重训触发线上监控要监控三类指标服务指标QPS、延迟、错误率、模型指标预测分布、特征分布、业务指标点击率、转化率。前两类是技术监控第三类是业务监控。数据漂移检测我一般用PSIPopulation Stability Index计算线上特征分布与训练特征分布的差异。PSI超过0.2就触发告警超过0.3就触发重训。def calculate_psi(expected, actual, buckets10): breakpoints np.percentile(expected, np.linspace(0, 100, buckets1)) expected_perc np.histogram(expected, breakpoints)[0] / len(expected) actual_perc np.histogram(actual, breakpoints)[0] / len(actual) expected_perc np.clip(expected_perc, 1e-6, None) actual_perc np.clip(actual_perc, 1e-6, None) psi np.sum((expected_perc - actual_perc) * np.log(expected_perc / actual_perc)) return psi5. 常见问题与排查技巧实录5.1 训练和推理结果不一致的排查思路这是AI工程里最经典的问题。排查顺序应该是先查特征再查模型最后查数值精度。特征层面用同一批原始数据分别走训练和推理路径逐特征对比输出值。如果发现某个特征不一致检查该特征的计算函数是否有依赖全局状态、是否用了不同的时间窗口、是否在推理时缺失了某些输入。模型层面检查模型是否处于eval模式。PyTorch里Dropout和BatchNorm在train和eval模式下行为不同如果推理时忘了model.eval()结果会差很多。数值精度层面检查训练和推理是否用了相同的浮点精度。混合精度训练FP16的模型直接用于FP32推理可能会有微小差异通常不影响但如果差异很大就要考虑精度转换问题。5.2 线上效果下降的快速定位方法线上效果下降先看是全局下降还是局部下降。全局下降通常是数据漂移或服务bug局部下降通常是某些用户群体或某些场景出了问题。我的排查清单排查项检查方法常见原因特征分布计算PSI上游数据源变更预测分布对比线上线下预测均值模型加载错误服务延迟查看P99延迟资源不足或代码bug错误日志搜索异常堆栈空值处理缺失业务指标分渠道/分用户群对比流量结构变化5.3 模型重训的触发条件与流程模型重训不能太频繁也不能太久不训。我的经验是定期重训比如每周一次加上漂移触发重训两者结合。重训流程要自动化拉取最新数据、跑特征工程、训练模型、评估、如果效果超过当前线上模型则替换否则保留原模型。这个流程用Airflow或者简单的cron脚本都能实现。实操心得重训后的模型不要直接全量替换先做小流量A/B测试。我一般先放5%流量跑一天确认核心指标不降再逐步放大。5.4 小团队资源有限时的优先级排序资源有限的时候优先级应该是数据质量 特征工程 模型复杂度 服务优化 监控完善。数据质量不行再复杂的模型也白搭。我见过一个团队花了两周调模型结构最后发现是训练数据里有一批标签标错了修正标签后简单模型就超过了复杂模型。6. 从零搭建AI工程体系的个人体会这套从零搭建的流程我在三个项目里完整跑过。最大的体会是AI工程的难点不在算法在工程。算法层面的东西论文里都写清楚了开源的实现也很多。但工程层面的东西每个项目都不一样数据怎么流、特征怎么管、服务怎么部署这些没有标准答案只能根据实际情况做取舍。另一个体会是不要追求一步到位。我第一个项目试图一开始就上全套MLOps结果光搭环境就花了一个月真正做模型的时间反而被压缩了。后来我学乖了先用最土的办法跑通闭环再根据实际瓶颈逐步优化。这样每一步优化都有明确的收益而不是为了架构好看而优化。最后分享一个小技巧每次项目复盘的时候把如果重来一次我会怎么做写下来。我攒了十几条这样的复盘笔记后来做新项目的时候直接翻出来对照能避开很多重复的坑。比如数据版本管理要在第一天做、特征函数必须训练推理共用、实验记录必须包含随机种子这些都是从复盘里来的。这个内容后续还可以这样扩展把监控层做深加入自动化重训的完整流水线或者把服务层做深加入模型量化、蒸馏、剪枝的实操。但那是另一个话题了先把从零到一的闭环跑通比什么都重要。