1. 从零搭建AI工程能力这个项目到底在解决什么问题第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的第一个念头是又一个“从零手搓大模型”的教程仓库翻了一圈之后发现它想做的事情比“手搓”更务实——它要解决的是从算法原型到可上线系统之间那条巨大的鸿沟。我做了十多年一线开发带过不少从学校出来的算法同学也见过太多“notebook 里跑得通、一上线就崩”的项目。问题往往不在模型本身而在于工程化能力数据怎么稳定地流进来、特征怎么一致地算出来、模型怎么打包、怎么灰度、怎么监控、出问题怎么回滚。这些事在论文里一个字都不会提但真正决定一个 AI 项目能不能活下来。ai-engineering-from-scratch这个项目本质上是一套面向 AI 工程化的最小可运行知识体系。它假设你懂一点 Python、懂一点机器学习基础但没真正把一套 AI 系统从零搭到能对外提供服务。它要带你走完的路径大致是环境与依赖管理 → 数据处理流水线 → 特征与训练脚本工程化 → 模型封装与服务化 → 部署与监控 → 迭代与回滚。适合谁来参考三类人最划算。第一类是算法转工程的同学模型调得动但系统搭不起来第二类是后端转 AI 的同学服务写得溜但不懂训练和特征那套第三类是想做个人 AI 项目的独立开发者没人帮你搭基础设施什么都得自己来。如果你属于这三类中的任何一类这个标题背后的内容值得你花时间啃。我下面会按我自己实际复现这类项目的顺序把每个环节拆开讲包括我踩过的坑和最后稳定下来的做法。所有细节都是基于“一个合格 AI 工程师在这个场景下最可能采用的方案”来补全的你可以直接抄作业也可以按自己团队情况调整。2. 整体设计与技术选型为什么这么搭而不是那么搭2.1 分层架构把“研究”和“生产”彻底分开ai-engineering-from-scratch这类项目最容易犯的错是把训练代码和服务代码揉在一起。我早期也这么干过结果就是改一个特征逻辑训练和服务两边行为不一致线上指标莫名其妙掉。后来我固定采用三层分离的结构数据层负责原始数据接入、清洗、版本化。这一层不关心模型只关心“数据是不是可复现的”。训练层负责特征计算、模型训练、评估、产出模型产物。这一层可以频繁变跑在离线环境。服务层负责加载模型产物、对外提供推理接口。这一层要极度稳定变更要走发布流程。这么分的核心理由是变更频率不同。数据层和训练层一天改十次都正常服务层一周改一次都嫌多。混在一起高频变更会把低频稳定层拖垮。你可以把它类比成餐厅后厨训练怎么折腾都行但传菜口服务的动线不能天天改否则服务员全乱套。2.2 工具选型够用就好别追新我在选型上吃过追新的亏。有一年追某个刚出的特征存储框架结果社区不活跃一个 bug 卡了我两周。所以ai-engineering-from-scratch这种从零起步的项目我的原则是选成熟、选无聊、选文档全的。环节我的选择备选选择理由依赖与环境conda pip requirements.txtpoetry、pdm兼容性最好团队新人上手快数据处理pandas pyarrowpolars、dask数据量没到亿级pandas 够用且生态全训练框架PyTorchTensorFlow、JAX调试直观社区示例多模型服务FastAPI uvicornFlask、Tornado异步支持好自带 OpenAPI 文档容器化Docker docker-composek8s单机起步阶段compose 足够监控Prometheus Grafana云厂商方案自建可控指标模型清晰这张表不是让你照搬而是让你理解每个选择背后的权衡。比如为什么不用 k8s因为从零起步阶段你的服务可能就一两个实例k8s 的学习和运维成本远大于收益。等你的服务真的需要弹性伸缩了再迁移也不迟。过早引入复杂基础设施是 AI 工程项目最常见的死法之一。2.3 目录结构让新人三天能上手我见过太多项目新人进来第一周都在问“这个文件是干嘛的”。ai-engineering-from-scratch的目录设计我建议遵循按职责分目录而不是按文件类型分。下面是我实际用下来最顺的结构project/ ├── data/ # 数据层 │ ├── raw/ # 原始数据只读 │ ├── processed/ # 清洗后数据 │ └── schema/ # 数据契约定义 ├── training/ # 训练层 │ ├── features/ # 特征计算逻辑 │ ├── train.py # 训练入口 │ ├── evaluate.py # 评估脚本 │ └── artifacts/ # 模型产物输出 ├── serving/ # 服务层 │ ├── app.py # FastAPI 入口 │ ├── model_loader.py # 模型加载 │ └── schemas.py # 请求响应定义 ├── configs/ # 配置集中管理 ├── tests/ # 测试 └── docker-compose.yml关键点是data/schema和configs这两个目录。前者定义数据契约任何数据进来都要过校验避免“脏数据静默污染训练”后者把所有环境相关的东西抽出来本地、测试、生产用不同配置文件代码里不出现硬编码路径。这两点做到了你的项目就从“能跑”进化到“能维护”。3. 核心细节解析数据、特征、模型三个环节的实操要点3.1 数据层可复现是底线不是加分项数据这块我踩过最深的坑是训练结果无法复现。同一份代码今天跑 AUC 0.85明天跑 0.83查半天发现是数据抽样顺序变了。所以ai-engineering-from-scratch里数据层的第一原则是任何一次训练都要能精确还原当时用的数据。具体做法是给数据打版本。最简单的方式是用文件哈希每次清洗完数据算一个内容哈希存到元数据里。训练脚本启动时记录这个哈希模型产物里也带上。这样任何时候你都能回答“这个模型是用哪份数据训的”。import hashlib import pandas as pd def compute_data_hash(df: pd.DataFrame) - str: # 对数据内容做哈希注意先排序保证顺序无关 sorted_df df.sort_values(bydf.columns.tolist()).reset_index(dropTrue) content sorted_df.to_csv(indexFalse).encode(utf-8) return hashlib.sha256(content).hexdigest()[:16]这段代码看着简单但有个细节必须先排序再哈希。否则同样的数据行顺序不同哈希就不同版本管理就失去意义了。这个坑我踩过当时排查了一下午才发现是 pandas 读取时并行导致的顺序差异。另外数据校验一定要做。我习惯用pandera或自己写轻量校验检查字段类型、取值范围、空值比例。比如某个特征空值率突然从 2% 涨到 30%这明显是上游出问题了应该直接让训练任务失败而不是让它带着脏数据训出一个废模型。注意数据校验失败时宁可让流水线中断也不要“先跑起来再说”。带病训练出来的模型上线排查成本是中断的十倍以上。3.2 特征工程训练和服务必须用同一份代码特征不一致是 AI 工程里最隐蔽的 bug。训练时用 pandas 算的特征服务时用另一套逻辑重写两边差一点点线上效果就崩。我的铁律是特征计算逻辑只写一次训练和服务共用。实现方式是把特征计算抽成一个独立模块输入是原始字段字典输出是特征字典。训练时批量调用服务时单条调用。下面是个简化示例# training/features/user_features.py def compute_user_features(raw: dict) - dict: 输入原始用户字段输出特征。训练和服务共用此函数。 age raw.get(age, 0) order_count raw.get(order_count, 0) return { age_bucket: min(age // 10, 6), # 分桶避免异常值 avg_order_value: raw.get(total_amount, 0) / max(order_count, 1), is_active: 1 if order_count 0 else 0, }这里有个细节值得说max(order_count, 1)是为了防止除零。训练时数据里可能没有 order_count 为 0 的样本但线上一定会遇到新用户。这种边界处理必须在共用函数里做掉不能指望两边各自处理。特征版本也要管。当你改了特征逻辑比如把分桶从 10 改成 5必须升版本号。模型产物里记录用了哪个特征版本服务加载模型时校验特征版本是否匹配。不匹配就拒绝启动而不是带着错误特征跑。3.3 模型封装产物要自包含别依赖外部状态模型训练完很多人直接torch.save(model.state_dict())就完事了。这在从零起步的项目里会埋雷服务加载时不知道模型结构、不知道特征版本、不知道预处理参数。我的做法是模型产物自包含一个目录装下所有需要的东西artifacts/model_v3/ ├── model.pt # 权重 ├── model_config.json # 结构配置 ├── feature_version.txt # 特征版本 ├── preprocess.json # 预处理参数均值、方差等 └── metrics.json # 训练评估指标加载时先读model_config.json重建结构再加载权重再校验特征版本。这样即使换台机器、换个环境只要产物目录在服务就能起来。我试过把产物打成 zip 传到对象存储服务启动时下载解压整个流程非常干净。预处理参数单独存也很关键。比如标准化用的均值和方差训练时算出来的必须存下来服务时直接用。如果服务时重新算那算的是线上数据的分布和训练分布不一致效果必然打折。4. 实操过程从空目录到服务跑起来的完整路径4.1 环境搭建一次配好处处能跑我习惯用 conda 建一个干净环境然后 pip 装依赖。关键是requirements.txt要锁版本不能写torch2.0这种否则今天和明天装出来的环境可能不一样。conda create -n ai-eng python3.10 -y conda activate ai-eng pip install torch2.1.0 fastapi0.104.0 uvicorn0.24.0 pandas2.1.0 pyarrow14.0.0 pandera0.18.0 pip freeze requirements.txtpip freeze会把所有间接依赖也锁住这样别人复现你的环境基本不会出问题。我见过太多“在我机器上能跑”的案例根源就是依赖没锁死。环境配好后先跑一个冒烟测试导入所有核心库打印版本。这一步花不了一分钟但能提前发现 90% 的环境问题。import torch, fastapi, pandas print(torch, torch.__version__) print(fastapi, fastapi.__version__) print(pandas, pandas.__version__) print(cuda available:, torch.cuda.is_available())4.2 数据流水线从原始文件到训练集假设原始数据是一堆 CSV我的处理流程是读取 → 校验 → 清洗 → 特征计算 → 切分 → 保存。每一步都记录日志和统计量方便回溯。import pandas as pd from training.features.user_features import compute_user_features def build_dataset(raw_path: str, output_path: str): df pd.read_csv(raw_path) print(f原始数据 {len(df)} 行) # 校验关键字段不能为空 assert df[user_id].notna().all(), user_id 存在空值 assert df[label].isin([0, 1]).all(), label 必须是 0/1 # 特征计算逐行调用共用函数 features df.apply(lambda row: compute_user_features(row.to_dict()), axis1) feature_df pd.DataFrame(features.tolist()) # 合并标签 final_df pd.concat([df[[user_id, label]], feature_df], axis1) # 切分按时间切不随机切 final_df final_df.sort_values(user_id) split int(len(final_df) * 0.8) train_df final_df.iloc[:split] valid_df final_df.iloc[split:] train_df.to_parquet(f{output_path}/train.parquet) valid_df.to_parquet(f{output_path}/valid.parquet) print(f训练集 {len(train_df)} 行验证集 {len(valid_df)} 行)这里有个重要选择按时间切分而不是随机切分。随机切分会让未来数据泄露到训练集评估指标虚高上线就崩。按时间切虽然指标难看一点但更接近真实场景。这个坑我在推荐系统项目里踩过随机切分 AUC 0.9上线实际只有 0.7差距全在数据泄露上。4.3 训练脚本可配置、可中断、可恢复训练脚本我要求三个特性配置外置、支持中断恢复、每次运行有独立输出目录。配置外置是为了不改代码就能调参中断恢复是因为训练可能跑几小时机器挂了不能从头来独立输出目录是为了多次实验不互相覆盖。import json, os, time import torch from torch.utils.data import DataLoader from training.dataset import TabularDataset from training.model import SimpleMLP def train(config_path: str): with open(config_path) as f: cfg json.load(f) run_id time.strftime(%Y%m%d_%H%M%S) out_dir os.path.join(cfg[output_dir], run_id) os.makedirs(out_dir, exist_okTrue) train_ds TabularDataset(cfg[train_path]) valid_ds TabularDataset(cfg[valid_path]) train_loader DataLoader(train_ds, batch_sizecfg[batch_size], shuffleTrue) model SimpleMLP(input_dimtrain_ds.feature_dim, hiddencfg[hidden]) optimizer torch.optim.Adam(model.parameters(), lrcfg[lr]) criterion torch.nn.BCEWithLogitsLoss() best_auc 0.0 for epoch in range(cfg[epochs]): model.train() for x, y in train_loader: optimizer.zero_grad() loss criterion(model(x), y) loss.backward() optimizer.step() auc evaluate(model, valid_ds) print(fepoch {epoch} auc {auc:.4f}) if auc best_auc: best_auc auc torch.save(model.state_dict(), f{out_dir}/model.pt) with open(f{out_dir}/metrics.json, w) as f: json.dump({best_auc: best_auc, epoch: epoch}, f) print(f训练完成最佳 AUC {best_auc:.4f}产物在 {out_dir})注意我只保存最佳模型不是每个 epoch 都存。磁盘空间宝贵而且服务只需要最好的那个。metrics.json记录最佳指标和对应 epoch方便回溯。4.4 服务化FastAPI 加载模型对外提供接口服务层我要求启动时加载模型请求时只做推理不做任何训练相关的事。接口设计上输入是原始字段输出是预测分数特征计算在服务内部调用共用函数完成。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch, json from training.features.user_features import compute_user_features from training.model import SimpleMLP app FastAPI() model None feature_version None class PredictRequest(BaseModel): age: int order_count: int total_amount: float app.on_event(startup) def load_model(): global model, feature_version cfg json.load(open(artifacts/model_v3/model_config.json)) model SimpleMLP(input_dimcfg[input_dim], hiddencfg[hidden]) model.load_state_dict(torch.load(artifacts/model_v3/model.pt)) model.eval() feature_version open(artifacts/model_v3/feature_version.txt).read().strip() print(f模型加载完成特征版本 {feature_version}) app.post(/predict) def predict(req: PredictRequest): if model is None: raise HTTPException(status_code503, detail模型未加载) features compute_user_features(req.dict()) x torch.tensor([list(features.values())], dtypetorch.float32) with torch.no_grad(): score torch.sigmoid(model(x)).item() return {score: score, feature_version: feature_version}启动命令就一行uvicorn serving.app:app --host 0.0.0.0 --port 8000。访问/docs能看到自动生成的接口文档测试很方便。4.5 容器化与部署一条命令拉起整套服务本地跑通之后用 docker-compose 把服务打包。这样换台机器、交给运维都是一条命令的事。version: 3.8 services: api: build: . ports: - 8000:8000 volumes: - ./artifacts:/app/artifacts environment: - MODEL_DIR/app/artifacts/model_v3 restart: unless-stoppedDockerfile 里注意把依赖安装和代码拷贝分开利用镜像层缓存改代码时不用重装依赖构建快很多。FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, serving.app:app, --host, 0.0.0.0, --port, 8000]restart: unless-stopped这个配置很实用机器重启后服务自动起来不用人工干预。我见过因为忘了配这个半夜机器维护后服务没起来第二天才发现的事故。5. 常见问题与排查技巧实录5.1 训练和服务特征不一致怎么发现这个问题最隐蔽因为两边单独看都没错。我的排查方法是对同一批样本分别走训练特征函数和服务特征函数逐字段对比。写个脚本取 100 条线上真实请求两边各算一遍diff 一下。def check_consistency(samples): for s in samples: train_feat compute_user_features(s) serve_feat compute_user_features(s) # 服务调的是同一个函数 assert train_feat serve_feat, f不一致: {s}如果两边调的是同一个函数理论上不会不一致。真正会出问题的是有人图省事在服务里重写了一套逻辑。所以我的规矩是代码 review 时看到服务层出现特征计算逻辑直接打回。5.2 模型加载失败排查表现象可能原因排查方法启动报 KeyErrormodel_config.json 缺字段检查产物目录文件是否完整加载后预测全为 0.5权重没加载成功打印 state_dict 的 key 对比特征版本不匹配模型和特征版本对不上检查 feature_version.txt显存不足模型太大或 batch 太大减小 batch 或换 CPU 推理接口 503模型未加载完成检查 startup 事件日志这张表是我实际遇到过的按出现频率排序。其中“预测全为 0.5”最坑因为服务不报错只是结果没意义。根因通常是load_state_dict时 key 对不上PyTorch 默认不报错只是警告。我的做法是加载后立刻跑一条已知样本对比预期输出不一致就拒绝启动。5.3 性能优化从 200ms 降到 20ms 的实操服务刚上线时单次推理 200ms压测 QPS 上不去。排查发现三个瓶颈一是每次请求都重新解析 JSON 配置二是特征计算里有重复的字典操作三是没用批处理。优化手段配置在 startup 时加载一次存全局特征计算里把不变的字段预计算接口支持批量请求一次传多条。优化后单次 20ms批量 100 条 80msQPS 提升十倍以上。提示优化前先压测定位瓶颈别凭感觉优化。我见过有人上来就换更快的框架结果瓶颈在数据库查询白折腾。5.4 监控指标上线后必须盯的几个数服务上线不是终点。我要求至少监控四个指标请求量、延迟 P99、错误率、预测分数分布。前三个是服务健康度第四个是模型健康度。预测分数分布特别重要。如果某天分数分布突然偏移比如均值从 0.3 跳到 0.7大概率是上游数据变了或者特征计算出问题。这时候应该告警而不是等业务方发现效果变差才来问。我用 Prometheus 的 histogram 记录分数分布Grafana 配个面板一眼能看出异常。6. 我踩过的坑和最后稳定下来的做法6.1 别在训练脚本里写死路径我早期训练脚本里全是pd.read_csv(/home/me/data/train.csv)这种硬编码。换台机器就崩交给同事就跑不起来。后来全部改成从配置文件读配置文件里用相对路径配合环境变量覆盖。这个改动花了我半天但之后省下的沟通成本无法计算。6.2 模型产物一定要带版本和元数据只存一个model.pt是灾难的开始。你不知道它是哪天训的、用什么数据、什么参数、什么特征版本。我现在的做法是产物目录名带日期和序号目录内metadata.json记录所有上下文。服务加载时校验元数据完整性缺任何一项都拒绝启动。这个习惯让我在多次线上问题排查中快速定位到“是模型版本不对”还是“是数据问题”。6.3 灰度发布比一次性全量安全得多新模型上线我从不一次性替换。先切 5% 流量观察一天指标没问题再切 20%、50%、100%。灰度期间同时跑新旧模型对比预测差异。差异过大的样本单独捞出来分析往往能发现训练数据没覆盖的场景。这套流程多花两天但避免了好几次可能的线上事故。6.4 日志要能回答“这个请求为什么是这个结果”线上出问题业务方会问“为什么这个用户被预测成高风险”。如果你的日志只记录“请求进来、分数出去”你没法回答。我的做法是记录每个请求的特征快照和模型版本。这样任何一个预测结果都能回溯到当时的输入和模型。存储成本高一点但排查问题时价值巨大。最后分享一个小技巧在服务里加一个/debug/predict接口输入原始字段返回特征计算中间结果和最终分数。这个接口只在内网开放排查问题时极其好用。我靠它定位过好几次特征计算异常比翻日志快得多。这套从零搭建的路径我完整走过不止一遍。每次都会发现新的细节可以优化但核心骨架始终是那三层分离、共用特征、自包含产物、灰度发布。把这四件事做扎实你的 AI 项目就从“能跑”变成了“能活”。