尧图网页设计 
1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体项目而是一种很熟悉的状态手上有现成的框架、现成的API、现成的预训练权重但一旦线上出问题或者老板问一句“这个指标为什么掉了”整个人就卡住了。这个标题戳中的正是当下很多做AI应用的人共同的软肋——会用工具但不懂工具背后的工程链路。ai-engineering-from-scratch从字面拆解核心是三个词AI、Engineering、From Scratch。它不是教你调某个库的某个函数而是把AI系统当成一个完整工程来对待从最底层的数据处理、模型训练、推理服务、部署监控一层一层自己搭起来。能做什么能让你在没有任何现成脚手架的情况下独立跑通一条从原始数据到线上服务的完整链路。解决什么问题解决“只会调包、不会排障、不懂取舍”的工程能力断层。适合谁适合已经会用Python、跑过几个demo、但想真正把AI系统落地到生产环境的开发者也适合想转AI工程方向的 backend 或 data 同学。我自己带过几个从算法转工程的同学最常见的误区就是以为模型精度高就万事大吉。结果一上线QPS上不去、显存爆掉、数据管道堵死、日志里全是超时。这些问题调包是学不会的只有自己从零搭一遍才会真正理解每个环节的边界在哪里。这篇就围绕这个标题把我自己从零搭AI工程链路时踩过的坑、做过的取舍、以及那些文档里不会写的细节完整拆一遍。2. 整体设计思路为什么“从零”不等于“重复造轮子”2.1 先想清楚“从零”的边界在哪里很多人一听“from scratch”第一反应是什么都自己写连矩阵乘法都手撸。这其实是误解。ai-engineering-from-scratch里的“从零”指的是工程链路的从零而不是数学库的从零。你完全可以用 NumPy 做数值计算、用 PyTorch 做自动求导但数据怎么切分、特征怎么落盘、模型怎么序列化、服务怎么暴露、监控怎么埋点这些必须自己设计一遍。为什么这么定边界因为AI工程的核心难点从来不是“实现一个卷积”而是“让一个卷积在真实流量下稳定跑三个月”。框架帮你解决了算子问题但解决不了工程问题。我见过太多团队模型代码写得漂漂亮亮结果数据版本对不上、训练和推理的特征处理不一致、线上模型和离线评估用的不是同一份权重最后排查花的时间比开发还多。所以整体设计的第一条原则凡是涉及“状态”和“边界”的地方必须自己掌控。数据版本、模型版本、配置版本、服务版本这四个东西必须能对得上。框架可以帮你算但不能帮你管。2.2 分层设计把AI系统拆成五层我习惯把一条完整的AI工程链路拆成五层从下往上依次是数据层原始数据采集、清洗、切分、版本管理特征层特征计算、特征存储、训练/推理一致性保障模型层训练脚本、超参管理、模型评估、模型注册服务层推理接口、批处理、并发控制、资源隔离观测层日志、指标、追踪、告警这五层不是随便分的而是对应了AI系统最容易出问题的五个位置。数据层出问题表现为“模型效果莫名其妙下降”特征层出问题表现为“离线AUC很高线上效果很差”模型层出问题表现为“复现不了上次的结果”服务层出问题表现为“一上量就崩”观测层出问题表现为“出了问题不知道去哪查”。ai-engineering-from-scratch的价值就是让你把这五层都亲手搭一遍知道每层的输入输出是什么、边界在哪里、失败模式是什么。下面我逐层拆。2.3 技术选型的取舍逻辑从零搭不代表要用最原始的工具。我的选型原则是越靠近数据越用轻量工具越靠近服务越用成熟组件。数据层我用 Pandas Parquet DVC 的思路。Pandas 处理中小规模数据足够Parquet 列式存储省空间且读取快DVC 做数据版本管理避免“数据被覆盖了但没人知道”。为什么不直接上 Spark因为大多数团队的初期数据量根本不到需要分布式的程度过早引入 Spark 只会增加运维负担。特征层我用 Feast 或者自己写一套简单的特征注册表。核心是保证训练和推理用同一份特征计算逻辑。这一点极其关键后面会展开。模型层用 PyTorch MLflow。PyTorch 灵活MLflow 做实验追踪和模型注册。为什么不用更重的平台因为从零搭的目的是理解链路不是被平台绑架。服务层用 FastAPI Uvicorn ONNX Runtime。FastAPI 写接口快Uvicorn 做 ASGI 服务ONNX Runtime 做推理加速。这套组合轻量、可控、容易排查。观测层用 Prometheus Grafana 结构化日志。指标采集用 Prometheus 客户端库日志用 JSON 格式方便后续检索。这套选型的核心逻辑是每一层都要能单独替换、单独测试、单独观测。如果你搭出来的系统换个模型就得改服务代码那说明分层没做好。3. 核心细节解析数据、特征、模型三层的实操要点3.1 数据层版本管理比清洗更重要数据层最容易犯的错是把精力全花在清洗上忽略了版本管理。清洗逻辑可以慢慢迭代但版本管理一旦缺失后面所有实验都不可复现。我的做法是原始数据只读永远不覆盖。每次清洗产出新版本用data/v1、data/v2这样的目录区分同时记录一份manifest.json包含{ version: v2, created_at: 2025-01-15T10:30:00Z, source: data/raw/2025-01-14, rows: 1250000, columns: [user_id, item_id, label, ts], cleaning_script: scripts/clean_v2.py, hash: a3f8c2... }这个hash是对整个数据文件算的用来校验数据有没有被意外修改。别小看这一步我遇到过好几次“明明没动数据但结果变了”的情况最后发现是有人手动改了一个CSV。切分数据时时间序列场景一定要按时间切不能随机切。随机切会导致未来信息泄露离线指标虚高。我一般用 70% 训练、15% 验证、15% 测试且验证集和测试集在时间上晚于训练集。注意数据切分后训练集、验证集、测试集的分布要单独统计一遍。如果某个类别在测试集里占比异常说明切分有问题别急着训练。3.2 特征层训练/推理一致性是命门特征层最核心的问题只有一个训练时算的特征和推理时算的特征必须是同一套逻辑。我见过最典型的翻车场景训练时用 Pandas 做归一化推理时用 NumPy 手写了一遍结果均值方差算错了线上效果直接掉一半。这种问题在离线评估里完全看不出来因为离线用的还是训练那套逻辑。解决办法是把特征计算逻辑抽成一个独立模块训练和推理都调它。比如# features/user_features.py class UserFeatureCalculator: def __init__(self, stats): self.mean stats[mean] self.std stats[std] def compute(self, df): df[age_norm] (df[age] - self.mean) / self.std return df训练时从统计文件加载stats推理时也从同一份统计文件加载。统计文件本身也要版本化跟模型版本绑定。特征存储方面离线特征落 Parquet在线特征落 Redis 或本地缓存。关键是要有一份特征注册表记录每个特征的名称、类型、计算逻辑、依赖数据、负责人。这份注册表就是特征层的“合同”谁改了特征都要更新它。3.3 模型层实验追踪不是可选项模型层最容易忽略的是实验追踪。很多人跑实验靠记笔记结果一周后回头看不知道哪个参数对应哪个结果。ai-engineering-from-scratch要求你把实验追踪当成基础设施来搭。我用 MLflow 记录每次实验的超参数学习率、batch size、层数等指标loss、AUC、F1等产物模型权重、配置文件、特征统计代码版本git commit hash这样任何一次结果都能复现。复现的时候checkout 对应 commit加载对应数据版本和特征统计跑一遍就能得到一样的结果。模型评估不能只看一个指标。分类任务我至少看 AUC、F1、RecallK排序任务看 NDCG、MAP。而且要分桶看比如按用户活跃度分桶避免模型只在头部用户上好用。模型注册环节每个上线的模型都要有明确的版本号、评估报告、回滚方案。我一般用model_name/v1、model_name/v2这样的命名配合 MLflow 的 Model Registry 管理阶段Staging、Production、Archived。4. 实操过程从原始数据到线上服务的完整链路4.1 环境准备与依赖锁定第一步永远是环境。我习惯用pyproject.toml管理依赖配合uv或poetry锁定版本。为什么强调锁定因为AI工程链路长任何一个依赖版本漂移都可能导致结果不一致。# 初始化项目 uv init ai-engineering-from-scratch cd ai-engineering-from-scratch # 添加核心依赖 uv add numpy pandas pyarrow scikit-learn uv add torch --index-url https://download.pytorch.org/whl/cpu uv add fastapi uvicorn onnx onnxruntime uv add mlflow prometheus-client依赖锁定后生成uv.lock提交到 git。任何人 clone 下来跑uv sync就能得到完全一致的环境。提示PyTorch 的 CPU 版和 GPU 版依赖不同团队内要统一。如果本地开发用 CPU、线上用 GPU至少保证算子行为一致避免出现“本地能跑线上报错”。4.2 数据管道搭建从原始文件到训练集假设原始数据是一批 CSV放在data/raw/下。第一步是写一个清洗脚本产出data/processed/v1/。# scripts/build_dataset.py import pandas as pd import hashlib import json from pathlib import Path RAW_DIR Path(data/raw) OUT_DIR Path(data/processed/v1) OUT_DIR.mkdir(parentsTrue, exist_okTrue) def load_raw(): frames [] for f in RAW_DIR.glob(*.csv): frames.append(pd.read_csv(f)) return pd.concat(frames, ignore_indexTrue) def clean(df): df df.dropna(subset[user_id, item_id, label]) df df.drop_duplicates(subset[user_id, item_id, ts]) df[ts] pd.to_datetime(df[ts]) return df def split_by_time(df): df df.sort_values(ts) n len(df) train df.iloc[: int(n * 0.7)] valid df.iloc[int(n * 0.7) : int(n * 0.85)] test df.iloc[int(n * 0.85) :] return train, valid, test if __name__ __main__: df clean(load_raw()) train, valid, test split_by_time(df) train.to_parquet(OUT_DIR / train.parquet) valid.to_parquet(OUT_DIR / valid.parquet) test.to_parquet(OUT_DIR / test.parquet) manifest { version: v1, rows: len(df), train_rows: len(train), valid_rows: len(valid), test_rows: len(test), } (OUT_DIR / manifest.json).write_text(json.dumps(manifest, indent2))这个脚本跑完数据层就有了第一个可复现的版本。注意manifest.json里记录了行数后续任何环节发现行数对不上都能快速定位。4.3 特征计算与统计落盘特征计算脚本要独立于训练脚本。我一般放在features/目录下训练和推理都 import 它。# features/build_features.py import pandas as pd import json from pathlib import Path class FeatureBuilder: def __init__(self, statsNone): self.stats stats or {} def fit(self, df): self.stats[age_mean] float(df[age].mean()) self.stats[age_std] float(df[age].std()) return self def transform(self, df): df df.copy() df[age_norm] (df[age] - self.stats[age_mean]) / self.stats[age_std] df[hour] df[ts].dt.hour return df def save(self, path): Path(path).write_text(json.dumps(self.stats, indent2)) classmethod def load(cls, path): stats json.loads(Path(path).read_text()) return cls(stats)训练时builder FeatureBuilder().fit(train_df) builder.save(artifacts/feature_stats_v1.json) train_feat builder.transform(train_df) valid_feat builder.transform(valid_df)推理时builder FeatureBuilder.load(artifacts/feature_stats_v1.json) input_feat builder.transform(input_df)这样训练和推理用的是同一份统计、同一套逻辑一致性有保障。4.4 模型训练与评估训练脚本要记录所有超参和指标。我用 MLflow 做追踪import mlflow import torch mlflow.set_experiment(ai-engineering-from-scratch) with mlflow.start_run(): mlflow.log_params({lr: 1e-3, batch_size: 256, epochs: 10}) # ... 训练循环 ... mlflow.log_metric(valid_auc, 0.87) mlflow.pytorch.log_model(model, model)评估不能只看整体指标要分桶。比如按用户活跃度分桶def evaluate_by_bucket(df, preds): df df.copy() df[pred] preds df[bucket] pd.qcut(df[user_activity], q4, labels[low, mid_low, mid_high, high]) return df.groupby(bucket).apply(lambda g: roc_auc_score(g[label], g[pred]))如果发现某个桶的AUC明显低说明模型在这个人群上表现差需要针对性优化。4.5 模型导出与推理服务训练完的 PyTorch 模型导出成 ONNX推理时用 ONNX Runtime性能更好且不依赖 PyTorch。import torch.onnx dummy_input torch.randn(1, feature_dim) torch.onnx.export( model, dummy_input, artifacts/model_v1.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch}, output: {0: batch}}, )服务用 FastAPIfrom fastapi import FastAPI import onnxruntime as ort import numpy as np app FastAPI() session ort.InferenceSession(artifacts/model_v1.onnx) builder FeatureBuilder.load(artifacts/feature_stats_v1.json) app.post(/predict) def predict(payload: dict): df pd.DataFrame([payload]) feat builder.transform(df) x feat[FEATURE_COLUMNS].values.astype(np.float32) out session.run(None, {input: x})[0] return {score: float(out[0][0])}启动uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4--workers 4表示起4个进程充分利用多核。具体起几个看CPU核数和单次推理耗时。我一般按核数 / 2起步压测后再调。4.6 观测层埋点服务起来后必须埋指标。Prometheus 客户端库很简单from prometheus_client import Counter, Histogram, make_asgi_app REQUEST_COUNT Counter(predict_requests_total, Total predict requests) REQUEST_LATENCY Histogram(predict_latency_seconds, Predict latency) ERROR_COUNT Counter(predict_errors_total, Total predict errors) app.post(/predict) def predict(payload: dict): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): try: # ... 推理逻辑 ... return {score: score} except Exception: ERROR_COUNT.inc() raise再把/metrics挂到 FastAPIapp.mount(/metrics, make_asgi_app())Grafana 里配几个面板QPS、P99延迟、错误率、模型版本分布。这四个指标基本能覆盖大部分线上问题。5. 常见问题与排查技巧实录5.1 离线指标好、线上效果差这是最经典的问题。排查顺序特征一致性训练和推理的特征计算逻辑是否完全一致统计文件是否同一版本数据分布线上流量分布和训练数据分布是否差异过大看特征均值、方差、缺失率。标签延迟线上标签是不是延迟回传的如果是评估时要用延迟后的标签。模型版本线上加载的模型是不是最新版本看服务启动日志里的模型hash。我遇到过一次排查两天最后发现是推理服务加载的特征统计文件是旧的归一化参数不对。所以现在我的服务启动日志里一定会打印特征统计文件的hash和模型文件的hash。5.2 服务一上量就崩通常是资源问题。排查清单现象可能原因排查方法QPS上不去worker数不够看CPU利用率增加worker延迟飙升单次推理太慢profile推理代码看瓶颈在特征还是模型内存爆掉批处理太大或内存泄漏限制batch size检查是否有全局缓存超时增多下游依赖慢看日志里的耗时分布我的经验是推理服务一定要设超时和限流。FastAPI 可以用asyncio.wait_for包一层超时直接返回降级结果。限流用slowapi或自己在网关层做。5.3 实验结果复现不了九成是版本问题。检查代码版本git commit数据版本manifest hash特征统计版本依赖版本lock文件随机种子随机种子要显式设置import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed)注意即使设了种子GPU上的某些算子仍可能有非确定性。如果要求严格复现要开torch.use_deterministic_algorithms(True)但会牺牲一些性能。5.4 数据管道跑得越来越慢数据量涨了之后Pandas 单机处理会变慢。优化顺序用 Parquet 替代 CSV读取快很多只读需要的列用columns参数用category类型存低基数列分块处理用chunksize还不行再考虑 Polars 或 DuckDB我实测下来同样一份数据Parquet 列裁剪比全量CSV读取快5到10倍。大多数场景根本不需要上分布式。注意数据管道优化前先profile别凭感觉优化。用cProfile或py-spy看时间花在哪。5.5 模型更新后如何安全回滚模型回滚必须能在分钟级完成。我的做法是模型文件按版本命名model_v1.onnx、model_v2.onnx服务启动时从配置读模型路径配置支持热更新或者用双buffer切换保留最近3个版本旧的归档回滚时只改配置重启服务加载旧版本模型。整个过程不超过5分钟。如果做不到说明部署流程有问题。6. 我踩过的坑和几条实在建议搭这套链路的过程中有几个坑我印象特别深。第一个是特征统计文件没版本化训练用v1、推理用v2线上效果掉了30%排查了一整天才定位到。从那以后我把特征统计、模型、配置三者的版本号绑在一起任何一个变了都要重新走评估流程。第二个坑是ONNX导出时没设dynamic_axes结果只能处理batch size为1的请求批量推理直接报错。这个错误在导出时不会提示只有实际跑批量请求才暴露。所以导出后一定要用不同batch size测一遍。第三个坑是日志打太多把磁盘写满了。推理服务每个请求打一条详细日志QPS一高日志量爆炸。后来改成采样打日志正常请求只打指标异常请求才打详细日志。几条实在建议第一任何环节都要能单独测试数据管道、特征计算、模型推理各自有单元测试别等联调才发现问题。第二版本号要贯穿始终数据、特征、模型、配置四个版本号能对上排查问题就快一半。第三观测层要提前搭别等出问题才想起来加日志那时候已经晚了。第四从零搭的目的是理解链路不是拒绝工具该用框架的地方用框架但要知道框架帮你做了什么、没做什么。这套东西搭完你对AI工程的理解会上一个台阶。以后再遇到线上问题脑子里会有一张完整的链路图知道从哪查、怎么查。这比会调多少个API值钱得多。 从零手搓AI工程链路：数据、特征、模型、服务与观测五层实操指南
1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体项目而是一种很熟悉的状态手上有现成的框架、现成的API、现成的预训练权重但一旦线上出问题或者老板问一句“这个指标为什么掉了”整个人就卡住了。这个标题戳中的正是当下很多做AI应用的人共同的软肋——会用工具但不懂工具背后的工程链路。ai-engineering-from-scratch从字面拆解核心是三个词AI、Engineering、From Scratch。它不是教你调某个库的某个函数而是把AI系统当成一个完整工程来对待从最底层的数据处理、模型训练、推理服务、部署监控一层一层自己搭起来。能做什么能让你在没有任何现成脚手架的情况下独立跑通一条从原始数据到线上服务的完整链路。解决什么问题解决“只会调包、不会排障、不懂取舍”的工程能力断层。适合谁适合已经会用Python、跑过几个demo、但想真正把AI系统落地到生产环境的开发者也适合想转AI工程方向的 backend 或 data 同学。我自己带过几个从算法转工程的同学最常见的误区就是以为模型精度高就万事大吉。结果一上线QPS上不去、显存爆掉、数据管道堵死、日志里全是超时。这些问题调包是学不会的只有自己从零搭一遍才会真正理解每个环节的边界在哪里。这篇就围绕这个标题把我自己从零搭AI工程链路时踩过的坑、做过的取舍、以及那些文档里不会写的细节完整拆一遍。2. 整体设计思路为什么“从零”不等于“重复造轮子”2.1 先想清楚“从零”的边界在哪里很多人一听“from scratch”第一反应是什么都自己写连矩阵乘法都手撸。这其实是误解。ai-engineering-from-scratch里的“从零”指的是工程链路的从零而不是数学库的从零。你完全可以用 NumPy 做数值计算、用 PyTorch 做自动求导但数据怎么切分、特征怎么落盘、模型怎么序列化、服务怎么暴露、监控怎么埋点这些必须自己设计一遍。为什么这么定边界因为AI工程的核心难点从来不是“实现一个卷积”而是“让一个卷积在真实流量下稳定跑三个月”。框架帮你解决了算子问题但解决不了工程问题。我见过太多团队模型代码写得漂漂亮亮结果数据版本对不上、训练和推理的特征处理不一致、线上模型和离线评估用的不是同一份权重最后排查花的时间比开发还多。所以整体设计的第一条原则凡是涉及“状态”和“边界”的地方必须自己掌控。数据版本、模型版本、配置版本、服务版本这四个东西必须能对得上。框架可以帮你算但不能帮你管。2.2 分层设计把AI系统拆成五层我习惯把一条完整的AI工程链路拆成五层从下往上依次是数据层原始数据采集、清洗、切分、版本管理特征层特征计算、特征存储、训练/推理一致性保障模型层训练脚本、超参管理、模型评估、模型注册服务层推理接口、批处理、并发控制、资源隔离观测层日志、指标、追踪、告警这五层不是随便分的而是对应了AI系统最容易出问题的五个位置。数据层出问题表现为“模型效果莫名其妙下降”特征层出问题表现为“离线AUC很高线上效果很差”模型层出问题表现为“复现不了上次的结果”服务层出问题表现为“一上量就崩”观测层出问题表现为“出了问题不知道去哪查”。ai-engineering-from-scratch的价值就是让你把这五层都亲手搭一遍知道每层的输入输出是什么、边界在哪里、失败模式是什么。下面我逐层拆。2.3 技术选型的取舍逻辑从零搭不代表要用最原始的工具。我的选型原则是越靠近数据越用轻量工具越靠近服务越用成熟组件。数据层我用 Pandas Parquet DVC 的思路。Pandas 处理中小规模数据足够Parquet 列式存储省空间且读取快DVC 做数据版本管理避免“数据被覆盖了但没人知道”。为什么不直接上 Spark因为大多数团队的初期数据量根本不到需要分布式的程度过早引入 Spark 只会增加运维负担。特征层我用 Feast 或者自己写一套简单的特征注册表。核心是保证训练和推理用同一份特征计算逻辑。这一点极其关键后面会展开。模型层用 PyTorch MLflow。PyTorch 灵活MLflow 做实验追踪和模型注册。为什么不用更重的平台因为从零搭的目的是理解链路不是被平台绑架。服务层用 FastAPI Uvicorn ONNX Runtime。FastAPI 写接口快Uvicorn 做 ASGI 服务ONNX Runtime 做推理加速。这套组合轻量、可控、容易排查。观测层用 Prometheus Grafana 结构化日志。指标采集用 Prometheus 客户端库日志用 JSON 格式方便后续检索。这套选型的核心逻辑是每一层都要能单独替换、单独测试、单独观测。如果你搭出来的系统换个模型就得改服务代码那说明分层没做好。3. 核心细节解析数据、特征、模型三层的实操要点3.1 数据层版本管理比清洗更重要数据层最容易犯的错是把精力全花在清洗上忽略了版本管理。清洗逻辑可以慢慢迭代但版本管理一旦缺失后面所有实验都不可复现。我的做法是原始数据只读永远不覆盖。每次清洗产出新版本用data/v1、data/v2这样的目录区分同时记录一份manifest.json包含{ version: v2, created_at: 2025-01-15T10:30:00Z, source: data/raw/2025-01-14, rows: 1250000, columns: [user_id, item_id, label, ts], cleaning_script: scripts/clean_v2.py, hash: a3f8c2... }这个hash是对整个数据文件算的用来校验数据有没有被意外修改。别小看这一步我遇到过好几次“明明没动数据但结果变了”的情况最后发现是有人手动改了一个CSV。切分数据时时间序列场景一定要按时间切不能随机切。随机切会导致未来信息泄露离线指标虚高。我一般用 70% 训练、15% 验证、15% 测试且验证集和测试集在时间上晚于训练集。注意数据切分后训练集、验证集、测试集的分布要单独统计一遍。如果某个类别在测试集里占比异常说明切分有问题别急着训练。3.2 特征层训练/推理一致性是命门特征层最核心的问题只有一个训练时算的特征和推理时算的特征必须是同一套逻辑。我见过最典型的翻车场景训练时用 Pandas 做归一化推理时用 NumPy 手写了一遍结果均值方差算错了线上效果直接掉一半。这种问题在离线评估里完全看不出来因为离线用的还是训练那套逻辑。解决办法是把特征计算逻辑抽成一个独立模块训练和推理都调它。比如# features/user_features.py class UserFeatureCalculator: def __init__(self, stats): self.mean stats[mean] self.std stats[std] def compute(self, df): df[age_norm] (df[age] - self.mean) / self.std return df训练时从统计文件加载stats推理时也从同一份统计文件加载。统计文件本身也要版本化跟模型版本绑定。特征存储方面离线特征落 Parquet在线特征落 Redis 或本地缓存。关键是要有一份特征注册表记录每个特征的名称、类型、计算逻辑、依赖数据、负责人。这份注册表就是特征层的“合同”谁改了特征都要更新它。3.3 模型层实验追踪不是可选项模型层最容易忽略的是实验追踪。很多人跑实验靠记笔记结果一周后回头看不知道哪个参数对应哪个结果。ai-engineering-from-scratch要求你把实验追踪当成基础设施来搭。我用 MLflow 记录每次实验的超参数学习率、batch size、层数等指标loss、AUC、F1等产物模型权重、配置文件、特征统计代码版本git commit hash这样任何一次结果都能复现。复现的时候checkout 对应 commit加载对应数据版本和特征统计跑一遍就能得到一样的结果。模型评估不能只看一个指标。分类任务我至少看 AUC、F1、RecallK排序任务看 NDCG、MAP。而且要分桶看比如按用户活跃度分桶避免模型只在头部用户上好用。模型注册环节每个上线的模型都要有明确的版本号、评估报告、回滚方案。我一般用model_name/v1、model_name/v2这样的命名配合 MLflow 的 Model Registry 管理阶段Staging、Production、Archived。4. 实操过程从原始数据到线上服务的完整链路4.1 环境准备与依赖锁定第一步永远是环境。我习惯用pyproject.toml管理依赖配合uv或poetry锁定版本。为什么强调锁定因为AI工程链路长任何一个依赖版本漂移都可能导致结果不一致。# 初始化项目 uv init ai-engineering-from-scratch cd ai-engineering-from-scratch # 添加核心依赖 uv add numpy pandas pyarrow scikit-learn uv add torch --index-url https://download.pytorch.org/whl/cpu uv add fastapi uvicorn onnx onnxruntime uv add mlflow prometheus-client依赖锁定后生成uv.lock提交到 git。任何人 clone 下来跑uv sync就能得到完全一致的环境。提示PyTorch 的 CPU 版和 GPU 版依赖不同团队内要统一。如果本地开发用 CPU、线上用 GPU至少保证算子行为一致避免出现“本地能跑线上报错”。4.2 数据管道搭建从原始文件到训练集假设原始数据是一批 CSV放在data/raw/下。第一步是写一个清洗脚本产出data/processed/v1/。# scripts/build_dataset.py import pandas as pd import hashlib import json from pathlib import Path RAW_DIR Path(data/raw) OUT_DIR Path(data/processed/v1) OUT_DIR.mkdir(parentsTrue, exist_okTrue) def load_raw(): frames [] for f in RAW_DIR.glob(*.csv): frames.append(pd.read_csv(f)) return pd.concat(frames, ignore_indexTrue) def clean(df): df df.dropna(subset[user_id, item_id, label]) df df.drop_duplicates(subset[user_id, item_id, ts]) df[ts] pd.to_datetime(df[ts]) return df def split_by_time(df): df df.sort_values(ts) n len(df) train df.iloc[: int(n * 0.7)] valid df.iloc[int(n * 0.7) : int(n * 0.85)] test df.iloc[int(n * 0.85) :] return train, valid, test if __name__ __main__: df clean(load_raw()) train, valid, test split_by_time(df) train.to_parquet(OUT_DIR / train.parquet) valid.to_parquet(OUT_DIR / valid.parquet) test.to_parquet(OUT_DIR / test.parquet) manifest { version: v1, rows: len(df), train_rows: len(train), valid_rows: len(valid), test_rows: len(test), } (OUT_DIR / manifest.json).write_text(json.dumps(manifest, indent2))这个脚本跑完数据层就有了第一个可复现的版本。注意manifest.json里记录了行数后续任何环节发现行数对不上都能快速定位。4.3 特征计算与统计落盘特征计算脚本要独立于训练脚本。我一般放在features/目录下训练和推理都 import 它。# features/build_features.py import pandas as pd import json from pathlib import Path class FeatureBuilder: def __init__(self, statsNone): self.stats stats or {} def fit(self, df): self.stats[age_mean] float(df[age].mean()) self.stats[age_std] float(df[age].std()) return self def transform(self, df): df df.copy() df[age_norm] (df[age] - self.stats[age_mean]) / self.stats[age_std] df[hour] df[ts].dt.hour return df def save(self, path): Path(path).write_text(json.dumps(self.stats, indent2)) classmethod def load(cls, path): stats json.loads(Path(path).read_text()) return cls(stats)训练时builder FeatureBuilder().fit(train_df) builder.save(artifacts/feature_stats_v1.json) train_feat builder.transform(train_df) valid_feat builder.transform(valid_df)推理时builder FeatureBuilder.load(artifacts/feature_stats_v1.json) input_feat builder.transform(input_df)这样训练和推理用的是同一份统计、同一套逻辑一致性有保障。4.4 模型训练与评估训练脚本要记录所有超参和指标。我用 MLflow 做追踪import mlflow import torch mlflow.set_experiment(ai-engineering-from-scratch) with mlflow.start_run(): mlflow.log_params({lr: 1e-3, batch_size: 256, epochs: 10}) # ... 训练循环 ... mlflow.log_metric(valid_auc, 0.87) mlflow.pytorch.log_model(model, model)评估不能只看整体指标要分桶。比如按用户活跃度分桶def evaluate_by_bucket(df, preds): df df.copy() df[pred] preds df[bucket] pd.qcut(df[user_activity], q4, labels[low, mid_low, mid_high, high]) return df.groupby(bucket).apply(lambda g: roc_auc_score(g[label], g[pred]))如果发现某个桶的AUC明显低说明模型在这个人群上表现差需要针对性优化。4.5 模型导出与推理服务训练完的 PyTorch 模型导出成 ONNX推理时用 ONNX Runtime性能更好且不依赖 PyTorch。import torch.onnx dummy_input torch.randn(1, feature_dim) torch.onnx.export( model, dummy_input, artifacts/model_v1.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch}, output: {0: batch}}, )服务用 FastAPIfrom fastapi import FastAPI import onnxruntime as ort import numpy as np app FastAPI() session ort.InferenceSession(artifacts/model_v1.onnx) builder FeatureBuilder.load(artifacts/feature_stats_v1.json) app.post(/predict) def predict(payload: dict): df pd.DataFrame([payload]) feat builder.transform(df) x feat[FEATURE_COLUMNS].values.astype(np.float32) out session.run(None, {input: x})[0] return {score: float(out[0][0])}启动uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4--workers 4表示起4个进程充分利用多核。具体起几个看CPU核数和单次推理耗时。我一般按核数 / 2起步压测后再调。4.6 观测层埋点服务起来后必须埋指标。Prometheus 客户端库很简单from prometheus_client import Counter, Histogram, make_asgi_app REQUEST_COUNT Counter(predict_requests_total, Total predict requests) REQUEST_LATENCY Histogram(predict_latency_seconds, Predict latency) ERROR_COUNT Counter(predict_errors_total, Total predict errors) app.post(/predict) def predict(payload: dict): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): try: # ... 推理逻辑 ... return {score: score} except Exception: ERROR_COUNT.inc() raise再把/metrics挂到 FastAPIapp.mount(/metrics, make_asgi_app())Grafana 里配几个面板QPS、P99延迟、错误率、模型版本分布。这四个指标基本能覆盖大部分线上问题。5. 常见问题与排查技巧实录5.1 离线指标好、线上效果差这是最经典的问题。排查顺序特征一致性训练和推理的特征计算逻辑是否完全一致统计文件是否同一版本数据分布线上流量分布和训练数据分布是否差异过大看特征均值、方差、缺失率。标签延迟线上标签是不是延迟回传的如果是评估时要用延迟后的标签。模型版本线上加载的模型是不是最新版本看服务启动日志里的模型hash。我遇到过一次排查两天最后发现是推理服务加载的特征统计文件是旧的归一化参数不对。所以现在我的服务启动日志里一定会打印特征统计文件的hash和模型文件的hash。5.2 服务一上量就崩通常是资源问题。排查清单现象可能原因排查方法QPS上不去worker数不够看CPU利用率增加worker延迟飙升单次推理太慢profile推理代码看瓶颈在特征还是模型内存爆掉批处理太大或内存泄漏限制batch size检查是否有全局缓存超时增多下游依赖慢看日志里的耗时分布我的经验是推理服务一定要设超时和限流。FastAPI 可以用asyncio.wait_for包一层超时直接返回降级结果。限流用slowapi或自己在网关层做。5.3 实验结果复现不了九成是版本问题。检查代码版本git commit数据版本manifest hash特征统计版本依赖版本lock文件随机种子随机种子要显式设置import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed)注意即使设了种子GPU上的某些算子仍可能有非确定性。如果要求严格复现要开torch.use_deterministic_algorithms(True)但会牺牲一些性能。5.4 数据管道跑得越来越慢数据量涨了之后Pandas 单机处理会变慢。优化顺序用 Parquet 替代 CSV读取快很多只读需要的列用columns参数用category类型存低基数列分块处理用chunksize还不行再考虑 Polars 或 DuckDB我实测下来同样一份数据Parquet 列裁剪比全量CSV读取快5到10倍。大多数场景根本不需要上分布式。注意数据管道优化前先profile别凭感觉优化。用cProfile或py-spy看时间花在哪。5.5 模型更新后如何安全回滚模型回滚必须能在分钟级完成。我的做法是模型文件按版本命名model_v1.onnx、model_v2.onnx服务启动时从配置读模型路径配置支持热更新或者用双buffer切换保留最近3个版本旧的归档回滚时只改配置重启服务加载旧版本模型。整个过程不超过5分钟。如果做不到说明部署流程有问题。6. 我踩过的坑和几条实在建议搭这套链路的过程中有几个坑我印象特别深。第一个是特征统计文件没版本化训练用v1、推理用v2线上效果掉了30%排查了一整天才定位到。从那以后我把特征统计、模型、配置三者的版本号绑在一起任何一个变了都要重新走评估流程。第二个坑是ONNX导出时没设dynamic_axes结果只能处理batch size为1的请求批量推理直接报错。这个错误在导出时不会提示只有实际跑批量请求才暴露。所以导出后一定要用不同batch size测一遍。第三个坑是日志打太多把磁盘写满了。推理服务每个请求打一条详细日志QPS一高日志量爆炸。后来改成采样打日志正常请求只打指标异常请求才打详细日志。几条实在建议第一任何环节都要能单独测试数据管道、特征计算、模型推理各自有单元测试别等联调才发现问题。第二版本号要贯穿始终数据、特征、模型、配置四个版本号能对上排查问题就快一半。第三观测层要提前搭别等出问题才想起来加日志那时候已经晚了。第四从零搭的目的是理解链路不是拒绝工具该用框架的地方用框架但要知道框架帮你做了什么、没做什么。这套东西搭完你对AI工程的理解会上一个台阶。以后再遇到线上问题脑子里会有一张完整的链路图知道从哪查、怎么查。这比会调多少个API值钱得多。