1. 从零手搓AI工程：为什么我不建议你直接调包

第一次看到ai-engineering-from-scratch这个项目名的时候，我正坐在工位上啃一个调了三天都没收敛的推荐模型。当时第一反应是：又来了一个教人从零实现Transformer的教程仓库。但点进去翻了翻结构，发现它跟市面上那些“手写注意力机制”的玩具项目完全不是一回事——它真正想解决的是从算法原型到工程落地之间那条巨大的鸿沟。

说白了，这个项目标题里的“from scratch”不是指从零推导数学公式，而是指从零搭建一套能跑、能扩、能维护的AI工程体系。它覆盖的东西包括数据管道怎么设计、特征存储怎么选型、训练任务怎么调度、模型怎么打包上线、线上推理怎么压延迟。这些东西你在任何一本深度学习教材里都找不到，但每一个做过真实AI项目的人都知道，它们才是决定项目生死的关键。

我见过太多团队，算法同学在Jupyter Notebook里把指标刷到SOTA，结果一上生产环境就崩——特征对不齐、推理延迟爆炸、模型版本混乱、回滚都回不去。这个项目就是冲着这些痛点来的。它适合谁看？我认为有三类人最应该认真读：一是刚入行一两年、只会调库跑模型的算法工程师；二是从后端转过来做AI平台、对模型本身不太熟的工程同学；三是技术负责人，需要一套完整的工程checklist来审视自己团队的基建水平。

接下来我会把这个项目涉及的核心工程环节拆开，结合我自己踩过的坑，把每个模块的设计思路、实操要点和避坑经验讲透。文章会比较长，但每一段都是能直接抄作业的干货。

2. 项目整体架构与设计思路拆解

2.1 为什么“从零”意味着先做减法

很多人一上来就想搞一套大而全的MLOps平台，Kubeflow、MLflow、Feast、Airflow全给装上，结果光环境依赖就折腾了两周，业务代码一行没写。ai-engineering-from-scratch这个项目的核心哲学恰恰相反：先用最少的组件跑通端到端流程，再按需替换。

它的架构分层非常清晰，从下到上大致是：数据层（原始存储+特征计算）、训练层（任务编排+实验追踪）、服务层（模型打包+推理接口）、监控层（指标采集+告警）。每一层都提供了“最小可用实现”和“生产级替换方案”两条路径。比如特征存储，最小实现就是一个Parquet文件加一个YAML schema定义，生产级才换成Feast或Tecton。

这种设计的好处是，你可以在本地用Docker Compose把整条链路跑起来，理解每个环节的输入输出契约，然后再决定哪些组件值得引入重型工具。我自己的经验是，80%的AI项目根本用不上Kubeflow，一个Celery加Redis的简单任务队列就足够了。过早引入复杂编排系统，只会让调试成本指数级上升。

2.2 核心模块的职责边界划分

项目把整个AI工程链路切成了六个核心模块，每个模块之间有明确的接口定义。这个划分方式我觉得比很多大厂的内部规范还合理，值得单独拿出来说。

• 数据契约模块：定义原始数据、清洗后数据、特征向量三层的schema和版本。关键点是所有schema用代码定义（比如Pydantic模型），而不是散落在文档里。
• 特征工程模块：负责特征计算逻辑的版本管理和离线/在线一致性校验。这是最容易出bug的地方，后面会详细讲。
• 训练编排模块：管理训练任务的依赖关系、资源申请、失败重试。核心是幂等性设计——同一个任务重复执行结果必须一致。
• 模型注册模块：模型产物的版本管理、元数据记录、阶段标记（staging/production/archived）。
• 推理服务模块：模型加载、请求预处理、批量推理、响应后处理。重点是延迟和吞吐的平衡。
• 可观测模块：数据漂移检测、预测分布监控、性能指标采集。

这六个模块的边界不是拍脑袋定的，而是按照数据流向和变更频率两个维度切分的。数据契约变更频率最低，可观测模块变更频率最高。把变更频率不同的东西放在一起，是很多团队代码腐化的根源。

2.3 技术选型背后的取舍逻辑

项目在技术选型上有一套很务实的评判标准，我总结成三个问题：这个工具能不能用单文件配置？出问题时能不能在30分钟内定位到根因？社区是否还在活跃维护？

按照这个标准，它推荐的技术栈是这样的：数据存储用Parquet加对象存储（而不是HDF5，因为Parquet的列式读取和谓词下推对特征回填太重要了）；任务编排用Prefect而不是Airflow（Prefect的Python原生DSL调试体验好太多）；实验追踪用MLflow（虽然UI丑但API稳定）；模型服务用FastAPI加ONNX Runtime（比TorchServe轻量，冷启动快）。

这里我要特别说一下ONNX Runtime的选择。很多人觉得把PyTorch模型转ONNX是多此一举，但实测下来，在CPU推理场景下ONNX Runtime的延迟通常比原生PyTorch低30%到50%，而且内存占用更可控。转换过程确实会碰到算子不支持的问题，但项目里提供了一套算子兼容性检查脚本，能在转换前就发现潜在问题，省去了大量试错时间。

3. 核心细节解析与实操要点

3.1 数据契约：用代码锁死schema

数据契约这个模块是整个项目的基石，也是最容易被忽视的环节。我见过太多项目，训练时特征列叫user_age，上线时变成userAge，然后线上指标直接掉一半，排查了两天才发现是命名不一致。

项目的做法是用Pydantic定义每一层数据的schema，并且在数据管道的每个出入口做强制校验。具体来说，原始数据层定义RawDataSchema，清洗后定义CleanDataSchema，特征层定义FeatureSchema。每个schema不仅定义字段名和类型，还定义取值范围和缺失率上限。

from pydantic import BaseModel, Field, validator class FeatureSchema(BaseModel): user_id: int = Field(..., gt=0) age: float = Field(..., ge=0, le=120) click_rate_7d: float = Field(..., ge=0, le=1) @validator('click_rate_7d') def check_missing_rate(cls, v): # 实际校验在批量层面做，这里只是示例 return v

注意：schema校验一定要在批量层面做缺失率检查，单条记录的validator只能检查取值范围。缺失率超过阈值（比如5%）应该直接阻断管道，而不是静默填充。

实操中有一个关键技巧：schema版本号要跟数据文件一起存储。比如特征文件命名为features_v3_20240115.parquet，同时在文件元数据里写入schema的哈希值。这样当schema变更时，旧文件不会被新代码误读。我自己的团队就是因为没做这件事，有一次更新了特征计算逻辑后，训练任务读到了新旧混合的特征文件，模型效果直接崩了。

3.2 特征一致性：离线在线对齐的三种方案

特征工程模块最核心的挑战是离线在线一致性。离线用Spark算特征，在线用Redis查特征，两边逻辑稍有差异就会导致训练服务偏差（training-serving skew）。项目里给出了三种对齐方案，按复杂度递增。

第一种是逻辑复用：把特征计算逻辑写成纯Python函数，离线和在线都调用同一个函数。优点是绝对一致，缺点是性能差，在线场景下每次请求都算一遍不现实。适合特征计算简单、QPS不高的场景。

第二种是离线预计算加在线缓存：离线用Spark批量算好特征写入Redis，在线直接查。优点是快，缺点是特征新鲜度受限于预计算频率。适合用户画像这类变化不频繁的特征。

第三种是流式计算加双写：用Flink或Kafka Streams实时算特征，同时写入离线存储和在线存储。优点是新鲜度高，缺点是架构复杂，运维成本大。适合风控、推荐这类对实时性要求高的场景。

项目推荐的做法是按特征分组，混合使用。比如用户基础属性用第二种，实时行为序列用第三种。关键是要有一套一致性校验机制：定期抽样一批请求，同时走离线和在线两条路径算特征，对比差异。差异超过阈值就告警。这个校验任务我建议每天跑一次，能提前发现很多隐蔽的bug。

3.3 训练任务的幂等性设计

训练编排模块里，幂等性是最容易被忽略但后果最严重的问题。什么叫幂等？就是同一个训练任务，不管执行多少次，产出的模型必须完全一致。听起来简单，做起来坑很多。

第一个坑是随机种子。PyTorch、NumPy、Python内置random的种子都要固定，而且DataLoader的worker进程也要设置种子，否则数据打乱顺序会变。项目里提供了一个set_all_seeds工具函数，把所有能想到的随机源都固定住。

第二个坑是数据读取顺序。如果训练数据是从数据库读的，没有ORDER BY，每次返回顺序可能不同。解决方案是在数据管道最后一步做一次显式排序，或者把数据先落盘成Parquet再读。

第三个坑是分布式训练的梯度聚合顺序。多卡训练时，梯度all-reduce的顺序如果不确定，浮点误差累积会导致结果有微小差异。这个在大多数场景下可以忽略，但如果你的模型对数值精度极其敏感（比如某些强化学习场景），就需要用确定性算法。

实操心得：我建议在CI流程里加一个“幂等性测试”，同一个任务跑两次，对比模型权重的哈希值。如果不一致，说明有未固定的随机源。这个测试能帮你省下无数个排查“为什么两次训练结果不一样”的夜晚。

3.4 模型注册与版本管理

模型注册模块的核心是元数据管理。一个模型产物不只是一个权重文件，还包括：训练数据版本、特征schema版本、超参数、评估指标、代码commit hash、环境依赖版本。这些信息缺一个，后面出问题都很难追溯。

项目用MLflow的Model Registry来管理这些元数据，但我觉得它的核心价值在于定义了一套模型阶段流转规则。模型从staging到production需要满足几个条件：离线评估指标超过基线、在线A/B测试至少跑满一周、推理延迟P99低于阈值、没有数据漂移告警。这些条件用代码写成检查脚本，自动执行，避免人为拍脑袋上线。

版本命名上，项目推荐用{模型名}-{日期}-{短哈希}的格式，比如ctr-model-20240115-a3f2c1。日期让人一眼看出新旧，短哈希保证唯一性。不要用v1、v2这种命名，因为分支合并时很容易冲突。

4. 实操过程与核心环节实现

4.1 本地环境搭建：30分钟跑通全链路

项目的Quickstart设计得很好，用Docker Compose把依赖都打包好了。我实测下来，从clone到跑通第一个训练任务，大概30分钟。这里把关键步骤和可能踩的坑记录一下。

第一步是克隆仓库并安装Python依赖。项目用uv做包管理，比pip快很多。如果你还在用pip，建议花10分钟迁移到uv，长期收益很大。

git clone https://github.com/example/ai-engineering-from-scratch.git cd ai-engineering-from-scratch uv sync

第二步是启动基础设施。项目提供了一个docker-compose.dev.yml，包含PostgreSQL（存元数据）、MinIO（存模型文件）、Redis（存在线特征）、MLflow（实验追踪）。

docker compose -f docker-compose.dev.yml up -d

注意：MinIO的默认端口是9000和9001，如果你本地有服务占用了这两个端口，需要改compose文件里的端口映射。我第一次跑的时候就是9000被占，排查了半小时才发现。

第三步是生成模拟数据并跑通训练管道。项目提供了一个make demo命令，会自动生成一批模拟用户行为数据，计算特征，训练一个简单的LR模型，并注册到MLflow。

make demo

跑完之后，打开http://localhost:5000就能看到MLflow的UI，里面有实验记录和模型版本。这一步的目的是让你先看到完整链路长什么样，再深入每个模块的细节。

4.2 特征管道开发：从SQL到特征向量

实际项目中，特征计算逻辑通常是用SQL写的，因为数据分析师也能参与。但SQL的可测试性和可维护性很差，所以项目推荐用SQL加Python封装的方式：核心逻辑用SQL表达，但通过Python函数调用，并且每个SQL片段都有单元测试。

具体做法是，在features/目录下每个特征组一个Python文件，里面定义compute_xxx_features函数，函数内部调用SQL模板。SQL模板用Jinja2渲染，参数包括时间窗口、过滤条件等。

from jinja2 import Template CLICK_RATE_SQL = Template(""" SELECT user_id, COUNT(CASE WHEN event_type = 'click' THEN 1 END) * 1.0 / COUNT(*) AS click_rate_{{ window }}d FROM events WHERE event_time >= '{{ start_date }}' AND event_time < '{{ end_date }}' GROUP BY user_id """) def compute_click_rate(spark, start_date, end_date, window=7): sql = CLICK_RATE_SQL.render(window=window, start_date=start_date, end_date=end_date) return spark.sql(sql)

这种写法的好处是，SQL逻辑可以被测试。项目里每个特征函数都配了一个测试，用一小批构造数据验证输出是否符合预期。我强烈建议你也这么做，因为特征计算的bug往往很隐蔽，等到模型效果下降才发现就晚了。

4.3 训练任务编排：Prefect Flow实战

项目用Prefect做任务编排，我一开始是抗拒的，因为觉得Airflow更成熟。但用下来发现Prefect的调试体验确实好很多——你可以在本地直接跑Flow，不需要启动调度器，出错时的堆栈信息也很清晰。

一个典型的训练Flow包含四个Task：数据加载、特征计算、模型训练、模型评估。每个Task都是纯函数，输入输出都是可序列化的对象。这样做的好处是Task可以单独测试，也可以缓存结果避免重复计算。

from prefect import flow, task @task(retries=3, retry_delay_seconds=60) def load_data(date: str): # 加载数据逻辑 return data_path @task def compute_features(data_path: str): # 特征计算逻辑 return feature_path @task def train_model(feature_path: str, params: dict): # 训练逻辑 return model_path @flow(name="training-pipeline") def training_flow(date: str, params: dict): data_path = load_data(date) feature_path = compute_features(data_path) model_path = train_model(feature_path, params) return model_path

实操心得：Prefect的retries参数很实用，但要注意重试的Task必须幂等。如果你的Task有副作用（比如写数据库），重试前要先清理上次的残留。我一般会在Task开头加一个cleanup步骤，确保状态干净。

4.4 推理服务部署：FastAPI加ONNX Runtime

推理服务模块用FastAPI做HTTP接口，ONNX Runtime做模型推理。这个组合的优点是轻量、启动快、依赖少。相比TorchServe，它不需要额外的模型仓库配置，一个Python文件就能起服务。

关键实现点是请求批处理。在线推理如果每条请求都单独跑一次模型，GPU利用率会很低。项目里实现了一个简单的批处理队列：请求先进入队列，攒够一批或者等待超时后一起推理。

import asyncio from collections import deque class BatchProcessor: def __init__(self, model, max_batch_size=32, max_wait_ms=10): self.model = model self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue = deque() self.lock = asyncio.Lock() async def predict(self, input_data): async with self.lock: self.queue.append(input_data) if len(self.queue) >= self.max_batch_size: return await self._process_batch() await asyncio.sleep(self.max_wait_ms / 1000) async with self.lock: if input_data in self.queue: return await self._process_batch() async def _process_batch(self): batch = list(self.queue) self.queue.clear() # 批量推理逻辑 return self.model(batch)

这个实现比较粗糙，生产环境建议用aiobatcher这类成熟库。但理解这个模式很重要，因为批处理是在线和离线推理最大的区别之一。

4.5 监控告警：数据漂移检测

可观测模块里，数据漂移检测是最有价值的部分。模型上线后效果下降，十有八九是输入数据分布变了。项目用PSI（Population Stability Index）来检测特征漂移，用KL散度检测预测分布漂移。

PSI的计算方式是：把特征值分桶，对比训练时和当前的分桶占比，算加权和。PSI小于0.1说明分布稳定，0.1到0.25说明有轻微漂移，大于0.25说明严重漂移需要告警。

import numpy as np def calculate_psi(expected, actual, buckets=10): def scale_range(data, buckets): min_val, max_val = data.min(), data.max() breakpoints = np.linspace(min_val, max_val, buckets + 1) return np.histogram(data, breakpoints)[0] / len(data) expected_percents = scale_range(expected, buckets) actual_percents = scale_range(actual, buckets) # 避免除零 expected_percents = np.where(expected_percents == 0, 0.0001, expected_percents) actual_percents = np.where(actual_percents == 0, 0.0001, actual_percents) psi = np.sum((actual_percents - expected_percents) * np.log(actual_percents / expected_percents)) return psi

注意：PSI对分桶数量敏感，分桶太少会漏检，太多会误报。我一般用10到20个桶，具体看特征取值分布。对于长尾分布的特征，建议用分位数分桶而不是等宽分桶。

5. 常见问题与排查技巧实录

5.1 训练服务偏差的排查路径

训练服务偏差是AI工程里最头疼的问题之一，表现是离线评估指标很好，线上一跑就崩。排查思路要系统化，不能东一榔头西一棒子。

第一步，确认特征一致性。抽样100条线上请求，把请求参数记录下来，用离线管道重新算一遍特征，对比差异。如果特征值不一致，问题就在特征工程模块。常见原因包括：时间窗口边界处理不同、缺失值填充策略不同、类别特征编码映射不同。

第二步，确认模型输入格式。有时候特征值对了，但拼接顺序错了，或者归一化参数用错了。检查模型训练时的输入schema和线上推理时的输入schema是否完全一致。

第三步，确认评估口径。离线评估用的指标和线上监控的指标定义可能不同。比如离线AUC是全量算的，线上AUC是按天算再平均的，两者本来就不应该相等。

我整理了一个排查速查表，按出现频率排序：

5.2 特征回填的坑与解决方案

特征回填是指用新的特征计算逻辑重新计算历史特征。这件事听起来简单，实际上坑很多。最大的坑是时间旅行：用未来的数据算历史的特征。

比如你要算“用户过去7天点击率”，如果回填时用的是当前时间的数据，那算出来的特征就包含了未来信息。训练时模型会学到这些未来信息，上线后没有未来信息，效果自然崩。

解决方案是严格按时间点计算。回填某一天的特征时，只能用那一天之前的数据。项目里提供了一个point_in_time_join工具函数，确保特征计算的时间边界正确。

def point_in_time_join(events_df, feature_df, event_time_col, feature_time_col): # 对每个事件，只关联事件时间之前的特征 return events_df.merge( feature_df, on='user_id', how='left', suffixes=('', '_feature') ).query(f'{feature_time_col} <= {event_time_col}')

实操心得：特征回填一定要在测试环境先跑一小批数据，验证时间边界正确后再全量跑。我见过一个团队直接在生产环境回填，结果把训练数据污染了，模型重新训练了三次才发现问题。

5.3 模型上线后的性能退化处理

模型上线后性能退化是常态，关键是要能快速定位和回滚。项目里推荐的做法是影子模式加金丝雀发布。

影子模式是指新模型上线后，不直接服务线上流量，而是复制一份流量跑新模型，对比新旧模型的预测差异。差异在可接受范围内，再切正式流量。金丝雀发布是指先切1%流量到新模型，观察一段时间没问题再逐步扩大。

回滚机制要自动化。当监控指标触发告警时，自动把流量切回旧模型。这个切换要在秒级完成，所以模型加载要支持热更新——新模型提前加载好，切换时只改路由配置。

我自己的经验是，模型回滚比模型上线更重要。上线可以慢一点，但回滚一定要快。因为线上出问题时，每一秒都在损失业务。建议把回滚演练纳入常规运维流程，每个月至少练一次。

5.4 资源不足时的降级策略

推理服务在高负载下容易资源不足，这时候需要有降级策略。项目里给出了三级降级方案：

第一级，降低批处理等待时间。从10ms降到5ms，牺牲一点吞吐换延迟。适合延迟敏感的场景。

第二级，切换到轻量模型。提前准备一个蒸馏后的小模型，资源不足时自动切换。小模型效果差一些，但能保证服务不挂。

第三级，返回默认预测。当所有模型都无法服务时，返回一个兜底的默认值（比如全局平均点击率）。这是最后的保底手段，避免服务完全不可用。

注意：降级策略要提前配置好阈值和切换逻辑，不能等出问题了再临时改代码。我建议在压测环境模拟资源不足场景，验证降级逻辑是否正确触发。

6. 工程化落地的个人体会

做AI工程这些年，我最大的体会是：算法决定上限，工程决定下限。一个模型效果再好，如果工程链路不稳定，上线后也是灾难。ai-engineering-from-scratch这个项目最大的价值，是把那些散落在各个团队内部文档里的工程经验，系统化地整理成了一套可复用的框架。

它不完美，有些模块的实现比较简陋，有些技术选型也有争议。但它的思路是对的：先用最小实现跑通链路，理解每个环节的契约，再按需引入重型工具。这个思路能帮你避免“过度工程化”的陷阱——很多团队死在引入太多工具上，而不是工具不够。

如果你正在从零搭建AI工程体系，我建议你按这个顺序推进：先做数据契约和特征一致性校验，这是地基；再做训练幂等性和模型注册，这是骨架；最后做监控告警和降级策略，这是保险。每一步都跑通了再进下一步，不要跳步。

最后分享一个我踩过的最大的坑：曾经有一个项目，特征计算逻辑用Spark SQL写，在线服务用Java写，两边逻辑分别维护。结果有一次SQL里改了一个过滤条件，Java代码忘了同步，导致线上线下特征不一致，模型效果掉了20%。从那以后，我坚持一个原则：特征计算逻辑只能有一份实现，要么都用Python，要么都用SQL，绝不两边各写一套。这个原则帮我省下了无数个排查bug的夜晚。