1. 从零搭建AI工程能力：这个项目到底在解决什么问题
第一次看到ai-engineering-from-scratch这个标题，我脑子里蹦出来的第一个念头是：又一个教人调包的教程？但仔细琢磨了一下“from scratch”这个限定词，再结合这两年带团队、面候选人的实际感受，我意识到它瞄准的其实是一个很具体的痛点——市面上大量所谓“AI入门”内容，本质上是在教你怎么用现成框架拼装，而不是让你理解一个AI系统从数据到上线到底经历了什么。
我自己在工业界做AI系统落地差不多十年，从早期用传统机器学习做风控，到后来做推荐、做多模态检索，踩过的坑基本都集中在“工程”这两个字上。模型结构本身反而是最不容易出问题的部分，真正让人熬夜的是数据管道断了、特征对不齐、推理延迟抖动、线上指标和离线对不上。所以当我看到这个标题时，第一反应是：如果它真的能做到“from scratch”，那它应该覆盖的是数据工程、训练工程、推理工程、评估工程这一整条链路，而不是只讲模型。
这个项目适合谁？我认为有三类人值得认真看。第一类是刚入行、只会调库的算法工程师，他们需要补上“模型之外”的那一半知识；第二类是从后端或数据方向转过来的工程师，他们有工程底子但缺AI系统的全局观；第三类是想自己动手做一个小型AI产品的人，比如做一个文本分类服务或者图像检索demo，他们需要一条能跑通的最小闭环路径。这篇文章我会按照我自己理解一个AI工程项目的方式，把这条路径拆开讲清楚，包括每一步为什么这么做、参数怎么定、坑在哪里。
2. 整体设计思路：为什么“从零”比“调包”更难也更有价值
2.1 从零构建的核心逻辑：把黑盒拆成可验证的环节
很多人对“from scratch”有误解，以为是要手写反向传播、手写卷积核。其实在工程语境下，from scratch 的核心含义是不依赖高度封装的端到端框架，而是把每个环节单独实现或至少单独验证。举个例子，你可以用 PyTorch 定义模型，但数据加载、特征处理、训练循环、评估指标、推理服务这些部分，你要能说清楚每一步的输入输出是什么、边界条件在哪里。
我之所以强调这一点，是因为在实际工作中，一个AI系统出问题，90%的情况不是模型结构错了，而是某个环节的数据分布变了、某个预处理步骤在训练和推理时不一致、某个指标的计算方式有歧义。如果你一直用高度封装的pipeline，这些问题会被掩盖，直到线上炸了才暴露。从零构建的价值就在于，每个环节都是显式的、可单独测试的。
2.2 技术选型的取舍：为什么我建议从轻量级工具链入手
在工具选型上，我的建议很明确：训练侧用 PyTorch，数据处理用 pandas + numpy 起步，服务侧用 FastAPI，实验管理先用文件系统加命名规范。为什么不推荐一上来就上 Spark、TFX、MLflow 这些重型工具？因为学习阶段最重要的是缩短反馈循环。你用 pandas 处理几万条数据，几秒钟就能看到结果；你用 Spark 处理同样的数据，光环境配置和任务提交就要几分钟，调试成本完全不一样。
这里有个我自己的经验：在数据量小于100万条、特征维度小于1000维的情况下，单机 pandas + numpy 完全够用。我做过一个文本分类项目，80万条样本、5000维稀疏特征，用 pandas 做特征工程加上 PyTorch 训练，单机16核CPU加一张消费级显卡，整个流程跑下来不到20分钟。这个效率对于学习和原型验证来说绰绰有余。等你真的遇到单机扛不住的数据量，再迁移到分布式框架，那时候你对数据流的理解已经足够深，迁移只是换API的事。
2.3 项目模块划分：一条最小但完整的AI工程链路
我把这条链路拆成五个模块，每个模块都有明确的输入输出和验收标准：
| 模块 | 输入 | 输出 | 验收标准 |
|---|---|---|---|
| 数据准备 | 原始文件/数据库 | 清洗后的结构化数据 | 无缺失值、类型正确、分布合理 |
| 特征工程 | 结构化数据 | 数值化特征矩阵 | 训练/推理一致、无泄漏 |
| 模型训练 | 特征矩阵+标签 | 模型权重+训练日志 | 损失收敛、验证集指标达标 |
| 评估验证 | 模型+测试集 | 评估报告 | 指标可复现、切片分析完整 |
| 推理服务 | 模型+请求 | 预测结果 | 延迟达标、错误可追踪 |
这张表看起来简单，但每个环节都有大量细节。比如“训练/推理一致”这一条，我见过太多项目栽在这里：训练时用了某个归一化参数，推理时忘了加载；训练时对缺失值填了均值，推理时直接报错。这些问题的根源都是没有把特征工程当成一个独立的、可序列化的组件来对待。
3. 核心细节解析：数据、特征与训练中的关键决策
3.1 数据清洗：先做“体检”再动手
拿到一份原始数据，我的习惯是先做三件事：看形状、看类型、看分布。看形状就是df.shape，知道有多少行多少列；看类型就是df.dtypes，确认数值列是不是被读成了字符串；看分布就是df.describe()加上对类别列的value_counts()。这三步做完，基本能发现80%的数据问题。
举个实际例子。我之前处理一份用户行为日志，timestamp列读进来是字符串，格式是2024-01-15 08:30:00。如果直接丢给模型肯定不行，需要转成数值特征。我的做法是拆成四个特征：小时、星期几、是否周末、距今天数。为什么这么拆？因为小时反映日内模式，星期几反映周内模式，是否周末是二值强特征，距今天数反映时效性。这四个特征加起来，比直接用一个时间戳数值效果好得多，因为树模型对原始时间戳的数值大小没有语义理解。
注意：时间特征拆分时一定要确认时区。我踩过一次坑，训练数据用的是UTC，推理时服务器本地时间是东八区，导致小时特征整体偏移8，线上效果直接掉了一截。后来统一在数据入口处强制转UTC，问题才解决。
3.2 特征工程：训练和推理必须共用同一套代码
这是我最想强调的一点。很多教程把特征工程写在训练脚本里，推理时重新写一遍，这是灾难的根源。正确的做法是把特征处理逻辑封装成一个类或一组函数，训练和推理都调用同一份代码。
具体怎么做？我通常定义一个FeatureProcessor类，包含fit和transform两个方法。fit在训练集上计算统计量（均值、方差、类别编码映射等），transform用这些统计量做转换。训练时先fit再transform，推理时只transform，并且把fit得到的统计量保存成文件，推理服务启动时加载。
class FeatureProcessor: def __init__(self): self.mean = None self.std = None self.category_map = {} def fit(self, df): self.mean = df['age'].mean() self.std = df['age'].std() for col in ['city', 'gender']: self.category_map[col] = {v: i for i, v in enumerate(df[col].unique())} def transform(self, df): df = df.copy() df['age'] = (df['age'] - self.mean) / self.std for col in ['city', 'gender']: df[col] = df[col].map(self.category_map[col]).fillna(-1) return df这段代码看起来简单，但它解决了一个大问题：推理时的数据转换和训练时完全一致。保存mean、std、category_map这些统计量，本质上就是把训练数据的“状态”固化下来，推理时复现这个状态。
3.3 训练循环：别急着上高级技巧，先把基础打牢
训练循环的核心就四步：前向传播、计算损失、反向传播、更新参数。但就是这四步，有很多细节决定成败。
批次大小的选择：我一般从32或64开始试。太小了梯度噪声大，太大了显存吃紧且泛化可能变差。有个经验公式可以参考：批次大小在32到256之间，学习率相应地在1e-4到1e-2之间调整。如果批次翻倍，学习率也可以适当翻倍，这是线性缩放规则的大致思路。
学习率调度：我习惯用余弦退火或者带热重启的余弦退火。为什么？因为固定学习率很难同时兼顾初期快速下降和后期精细收敛。余弦退火让学习率从初始值平滑降到接近零，后期模型能在局部最小值附近稳定下来。实测下来，同样的模型和數據，余弦退火比固定学习率在验证集上通常能高0.5到1个点。
早停策略：验证集损失连续N个epoch不下降就停。N一般取5到10。这个策略帮我省了大量训练时间，也避免了过拟合。但要注意，早停的“不下降”最好用平滑后的指标，比如移动平均，避免被单次波动误导。
3.4 评估验证：离线指标好不代表线上好
评估环节最容易犯的错误是只看一个总体指标。比如分类任务只看准确率，如果类别不平衡，准确率会严重误导。我通常会看四个东西：混淆矩阵、每个类别的精确率和召回率、AUC、以及按关键维度切片的指标。
切片分析特别重要。举个例子，一个推荐模型整体AUC是0.78，看起来不错。但按用户活跃度切片后发现，高活跃用户AUC是0.82，低活跃用户只有0.65。这说明模型对低活跃用户预测能力差，而低活跃用户恰恰是拉新的重点。如果不做切片，这个问题根本发现不了。
实操心得：切片维度不要太多，选3到5个业务上最关心的维度就够了。每个切片样本量不能太少，少于1000条的切片指标波动太大，参考价值有限。
4. 实操过程：从原始数据到可调用服务的完整实现
4.1 环境准备与依赖管理
我强烈建议用 conda 或 venv 创建独立环境，依赖写进requirements.txt。核心依赖就几个：numpy、pandas、scikit-learn、torch、fastapi、uvicorn、joblib。版本尽量固定，比如torch==2.1.0，避免不同机器上行为不一致。
python -m venv venv source venv/bin/activate pip install numpy pandas scikit-learn torch fastapi uvicorn joblib pip freeze > requirements.txt这一步看起来简单，但我见过太多人因为环境问题浪费半天。固定版本号能避免“在我机器上能跑”的经典问题。
4.2 数据加载与清洗的实操步骤
假设我们有一份CSV文件raw_data.csv，包含用户ID、年龄、城市、性别、历史行为次数、是否转化六个字段。目标是根据前五个字段预测“是否转化”。
第一步，加载并检查：
import pandas as pd df = pd.read_csv('raw_data.csv') print(df.shape) print(df.dtypes) print(df.isnull().sum()) print(df['是否转化'].value_counts())第二步，处理缺失值。年龄缺失用中位数填充，城市和性别缺失用“未知”填充。为什么年龄用中位数而不是均值？因为年龄分布通常有偏，中位数更稳健。
第三步，处理异常值。年龄小于0或大于120的置为缺失再填充，历史行为次数为负的置为0。
df.loc[(df['年龄'] < 0) | (df['年龄'] > 120), '年龄'] = None df['年龄'].fillna(df['年龄'].median(), inplace=True) df['城市'].fillna('未知', inplace=True) df['性别'].fillna('未知', inplace=True) df.loc[df['历史行为次数'] < 0, '历史行为次数'] = 04.3 特征处理与数据集划分
特征处理按前面说的FeatureProcessor来做。划分数据集时，我习惯用train_test_split，测试集占20%，随机种子固定为42，保证可复现。
from sklearn.model_selection import train_test_split train_df, test_df = train_test_split(df, test_size=0.2, random_state=42, stratify=df['是否转化']) processor = FeatureProcessor() processor.fit(train_df) train_X = processor.transform(train_df.drop(columns=['是否转化', '用户ID'])) test_X = processor.transform(test_df.drop(columns=['是否转化', '用户ID'])) train_y = train_df['是否转化'].values test_y = test_df['是否转化'].values注意stratify参数，它保证训练集和测试集的类别比例一致。如果类别不平衡，这个参数很重要。
4.4 模型定义与训练循环实现
模型用一个简单的多层感知机，输入维度等于特征数，隐藏层128和64，输出层2类。
import torch import torch.nn as nn class MLP(nn.Module): def __init__(self, input_dim): super().__init__() self.net = nn.Sequential( nn.Linear(input_dim, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.3), nn.Linear(64, 2) ) def forward(self, x): return self.net(x)训练循环：
model = MLP(train_X.shape[1]) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50) criterion = nn.CrossEntropyLoss() train_tensor = torch.tensor(train_X.values, dtype=torch.float32) train_label = torch.tensor(train_y, dtype=torch.long) dataset = torch.utils.data.TensorDataset(train_tensor, train_label) loader = torch.utils.data.DataLoader(dataset, batch_size=64, shuffle=True) for epoch in range(50): model.train() total_loss = 0 for batch_x, batch_y in loader: optimizer.zero_grad() output = model(batch_x) loss = criterion(output, batch_y) loss.backward() optimizer.step() total_loss += loss.item() scheduler.step() print(f'Epoch {epoch}, Loss: {total_loss / len(loader):.4f}')这里有几个细节：Dropout(0.3)防止过拟合，CosineAnnealingLR让学习率平滑下降，shuffle=True打乱批次顺序。这些看起来是小事，但组合起来对最终效果影响很大。
4.5 模型保存与推理服务封装
训练完成后保存模型权重和特征处理器：
torch.save(model.state_dict(), 'model.pt') import joblib joblib.dump(processor, 'processor.pkl')推理服务用 FastAPI：
from fastapi import FastAPI import joblib import torch import numpy as np app = FastAPI() model = MLP(input_dim=5) model.load_state_dict(torch.load('model.pt')) model.eval() processor = joblib.load('processor.pkl') @app.post('/predict') def predict(data: dict): df = pd.DataFrame([data]) X = processor.transform(df) tensor = torch.tensor(X.values, dtype=torch.float32) with torch.no_grad(): output = model(tensor) prob = torch.softmax(output, dim=1)[0][1].item() return {'probability': prob, 'label': int(prob > 0.5)}启动命令：uvicorn main:app --host 0.0.0.0 --port 8000。这样你就有了一个可调用的预测接口。
5. 常见问题与排查技巧实录
5.1 训练不收敛或损失震荡
这是最常见的问题。排查顺序我一般是：先看学习率，再看数据，最后看模型。学习率太大导致震荡，太小导致收敛慢。我通常先用1e-3试，如果损失震荡就降到1e-4，如果下降太慢就升到1e-2。数据方面，检查特征是否归一化、标签是否有问题。模型方面，检查层数是否过深、激活函数是否合适。
有个容易被忽略的点：输入特征的尺度差异过大。比如年龄是0到100，历史行为次数是0到10000，如果不做标准化，梯度会被大数值特征主导。我的做法是对所有数值特征做标准化，对类别特征做嵌入或独热编码。
5.2 过拟合：训练集好测试集差
过拟合的信号很明确：训练损失持续下降，验证损失先降后升。应对手段按优先级排序：增加数据量 > 加正则化 > 减小模型 > 早停。增加数据量最有效但成本最高；加正则化包括Dropout、权重衰减、数据增强；减小模型就是减少层数或隐藏单元；早停是最省事的兜底方案。
我自己的习惯是同时用Dropout和权重衰减，Dropout率0.2到0.5之间，权重衰减1e-4到1e-2之间。这两个参数需要根据验证集表现微调。
5.3 推理延迟过高
如果单次推理超过100毫秒，对于实时服务来说就偏高了。优化方向有几个：模型量化、批处理、ONNX导出、减少特征计算。模型量化把float32转成int8，速度能提升2到4倍，精度损失通常在1个点以内。批处理是把多个请求攒在一起推理，吞吐量能大幅提升，但单次延迟会增加。ONNX导出配合ONNX Runtime，推理速度通常比原生PyTorch快20%到50%。
避坑技巧：量化后的模型一定要在验证集上重新评估，我遇到过量化后AUC掉3个点的情况，原因是某些层的数值范围太窄，int8表示精度不够。这时候可以只量化部分层，或者用动态量化代替静态量化。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 损失为NaN | 学习率过大/数据有异常值 | 打印每步损失和输入 | 降低学习率、裁剪梯度、清洗数据 |
| 验证指标远低于训练 | 过拟合 | 对比训练/验证损失曲线 | 加Dropout、早停、增加数据 |
| 推理结果与训练不一致 | 特征处理不一致 | 对比训练和推理的中间输出 | 统一特征处理代码 |
| 服务启动报错 | 依赖版本冲突 | 检查requirements.txt | 固定版本、重建环境 |
| 预测概率全为0.5 | 模型未加载或输入全零 | 检查模型权重和输入 | 重新加载模型、检查特征 |
6. 从原型到可用：我踩过的坑和总结的经验
6.1 数据泄漏：最隐蔽也最致命
数据泄漏是指训练时用到了推理时拿不到的信息。最常见的场景是用未来数据预测过去。比如做用户流失预测，特征里包含了“最近一次登录距今天数”，但如果这个特征是在标签时间点之后计算的，就泄漏了。排查方法是：对每个特征问一句“这个特征在预测时刻真的能拿到吗？”如果答案是否定的，就必须去掉。
我踩过一次严重的泄漏坑：做商品推荐时，特征里包含了商品的“总销量”，但这个销量是包含预测时间点之后的。结果离线AUC高达0.95，上线后直接掉到0.6。后来把销量改成“截至预测时间点的历史销量”，离线AUC降到0.78，但线上一致了。这个教训让我明白，离线指标虚高不一定是好事，可能是泄漏的信号。
6.2 版本管理：模型、数据、代码要一起管
AI项目和传统软件项目最大的区别是，模型效果依赖于数据、代码、超参数三者的组合。只管理代码版本是不够的。我的做法是：每次训练生成一个实验ID，把代码commit hash、数据版本、超参数配置、模型权重、评估结果都关联到这个ID上。这样任何时候都能复现某个模型。
具体实现可以很简单：建一个experiments目录，每个实验一个子目录，里面放config.json、metrics.json、model.pt。不需要上MLflow也能做到基本可追溯。
6.3 监控：上线只是开始
模型上线后，数据分布会变，模型效果会衰减。必须监控三个东西：输入特征分布、预测结果分布、业务指标。输入特征分布偏移检测可以用PSI或KL散度，预测结果分布看均值方差变化，业务指标就是点击率、转化率这些。
我一般设置这样的告警规则：PSI大于0.2触发警告，大于0.5触发严重告警；预测均值连续三天偏移超过10%触发告警。这些阈值不是绝对的，要根据业务容忍度调整。
6.4 最后分享几个小技巧
第一个技巧：训练前先跑一个极小数据集。取100条数据，跑几个epoch，确认整个流程能跑通、损失能下降。这能提前发现大部分代码错误，比在全量数据上跑半天才发现问题高效得多。
第二个技巧：把随机种子固定住。torch.manual_seed(42)、np.random.seed(42)、random.seed(42)三件套。这样每次实验结果可复现，排查问题时不会因为随机性干扰判断。
第三个技巧：保存最佳模型而不是最后一个模型。训练过程中验证集指标最好的那个epoch的权重才是你要的。我通常在每个epoch结束后判断验证指标是否提升，提升就保存，不提升就跳过。
第四个技巧：推理服务加一个健康检查接口。/health返回模型加载状态和版本号，方便运维排查。这个接口不涉及预测逻辑，只是确认服务活着、模型加载成功。
这条从零构建AI工程能力的路径，我自己走过很多遍，每次都能发现新的细节。最开始觉得麻烦，但当你真正遇到线上问题、需要快速定位的时候，你会发现这些“麻烦”的步骤恰恰是救命的。数据要可追溯、特征要一致、模型要可复现、服务要可监控，这四句话是我这些年做AI工程最深的体会。
