1. 从零搭建AI工程能力，为什么大多数人卡在第一步

聊到“从零开始做AI工程”这个话题，我脑子里第一反应不是某个框架、某个模型，而是一个很现实的问题：大部分人根本不知道自己该从哪一行代码写起。你可能已经看过不少教程，跟着跑通了几个Notebook，模型在MNIST上准确率刷到了99%，但一旦让你从零搭一个能用的AI系统，立刻就懵了。这不是你笨，而是因为“跑通一个模型”和“构建一套AI工程体系”之间，隔着一整条工程化的鸿沟。

ai-engineering-from-scratch这个标题，核心不在“AI”，而在“engineering”和“from scratch”。它要解决的不是“怎么训练一个模型”，而是“怎么像一名真正的AI工程师那样，从零开始把数据、模型、服务、监控这一整条链路搭起来”。适合谁看？适合那些已经会写Python、懂一点机器学习基础，但在实际项目中总觉得“差点意思”的人。也适合那些从后端、数据方向转过来，想搞清楚AI系统到底和普通软件系统有什么本质区别的工程师。

我见过太多人一上来就冲去学Transformer架构、研究注意力机制的数学推导，结果连一个最简单的推理服务都部署不明白。这就像你想学做菜，不去练刀工、火候，先花三个月研究小麦的分子结构。不是说底层原理不重要，而是工程能力的构建有它自己的顺序，搞反了顺序，投入产出比会低得让你怀疑人生。

这篇文章我会按照一个真实的从零构建路径来展开，把每个阶段该做什么、为什么这么做、容易踩什么坑，全部拆开讲清楚。你不需要有很深的数学背景，但需要有一点耐心，因为AI工程本质上是一门“手艺活”，光看是看不会的。

2. 先搞清楚AI工程和传统软件工程到底差在哪

2.1 确定性系统与概率性系统的根本分歧

传统软件工程的核心假设是：给定相同的输入，系统应该产生相同的输出。你写一个排序函数，输入[3,1,2]，永远得到[1,2,3]。测试写起来很直接——断言输出等于预期值就行了。但AI系统从根上就打破了这个假设。同一个输入，模型可能这次输出“正面情感”，下次输出“中性情感”，因为模型权重是浮点数，推理过程涉及大量矩阵运算，数值精度、硬件差异、甚至批次大小的变化都会影响最终结果。

这意味着你不能再用传统的单元测试思路来验证AI系统。你需要的是统计意义上的验证：在一个分布上，模型的准确率、召回率、F1分数是否达标。这不是说传统测试方法没用了，而是说你必须额外构建一套针对概率性系统的评估体系。很多从传统后端转过来的工程师，第一个跟头就栽在这里——他们花大量时间写精确断言，结果发现测试天天挂，最后干脆不写了，系统裸奔上线。

2.2 数据是代码的一部分，而且比代码更难管

在传统工程里，代码是核心资产，数据是附属品。但在AI工程里，数据的质量和分布直接决定了系统的上限。你模型架构再先进，训练数据里全是噪声，出来的东西就是垃圾。更麻烦的是，数据不像代码那样可以用Git做版本控制——一个数据集动辄几个GB到几个TB，你不可能每次改动都提交到仓库里。

所以AI工程从第一天起就要考虑数据版本管理的问题。常见做法是用DVC或者类似的工具来追踪数据集的变更，把数据文件的哈希值和元信息存到Git里，实际数据放在对象存储上。这个决策看起来很小，但如果你一开始不做，等到数据集迭代了十几版之后，你会发现根本搞不清楚哪个模型是用哪版数据训出来的，复现实验变成了一场考古。

2.3 部署之后才是真正麻烦的开始

传统软件上线之后，只要服务器不挂，行为基本是稳定的。AI系统上线之后，模型会随着时间推移而“退化”。原因很简单：真实世界的数据分布在变，而你的模型是在历史数据上训练的。比如一个电商推荐系统，训练时用的是去年的用户行为数据，今年用户的偏好变了，模型的推荐效果就会下降。这种现象叫“数据漂移”，是AI系统特有的问题。

所以AI工程必须包含监控和反馈闭环。你需要持续追踪模型的输入分布、输出分布、以及业务指标（点击率、转化率等），一旦发现异常就要触发重新训练。这套机制在传统软件里根本不存在，你得从零设计。这也是为什么我说AI工程师不能只懂模型，还得懂数据管道、懂服务架构、懂监控告警。

3. 从零构建的第一个阶段：把数据管道跑通

3.1 为什么数据管道比模型更值得先投入

我见过太多项目，模型代码写得漂漂亮亮，但数据管道一团糟。训练脚本里硬编码了数据路径，预处理逻辑散落在十几个文件里，每次换数据集都要改半天代码。这种项目基本上活不过三个月。正确的做法是：先把数据管道做成一个独立的、可测试的、可复用的模块，然后再往上叠模型。

数据管道要解决的核心问题就三个：数据从哪里来、怎么变成模型能吃的格式、怎么保证每次训练用的数据是一致的。听起来简单，但每个环节都有坑。比如数据来源可能是数据库、CSV文件、API接口，格式五花八门；预处理可能涉及缺失值填充、类别编码、归一化，每一步的参数都需要保存下来，推理时要用同样的参数处理新数据；一致性则要求你记录每次训练的输入数据版本和预处理配置。

3.2 一个最小可用的数据管道长什么样

假设你现在要做一个文本分类任务，从零开始搭数据管道。我会建议你按这个顺序来：

第一步，定义一个数据加载层。不要直接在训练脚本里写pd.read_csv()，而是封装一个DataLoader类，负责从不同来源读取原始数据，返回一个统一的DataFrame格式。这样做的好处是，以后换数据源只需要改这一个类，训练代码完全不用动。

第二步，定义预处理层。把所有预处理逻辑写成一个Preprocessor类，包含fit()和transform()两个方法。fit()在训练数据上计算统计量（比如均值、方差、词表），transform()用这些统计量处理数据。关键是要把fit()得到的参数保存到磁盘上，推理时加载同样的参数。这一步是保证训练和推理一致性的核心。

第三步，定义数据集划分逻辑。训练集、验证集、测试集的划分要固定随机种子，并且把划分结果保存下来。不要每次跑训练都重新划分，否则你的实验结果根本没法比较。

import pandas as pd import numpy as np import pickle from sklearn.model_selection import train_test_split class DataLoader: def __init__(self, source_path): self.source_path = source_path def load(self): df = pd.read_csv(self.source_path) df = df.dropna(subset=['text', 'label']) return df class Preprocessor: def __init__(self): self.vocab = {} self.max_len = 128 def fit(self, texts): from collections import Counter word_counts = Counter() for text in texts: word_counts.update(text.lower().split()) self.vocab = {word: idx+1 for idx, (word, _) in enumerate(word_counts.most_common(10000))} def transform(self, texts): sequences = [] for text in texts: seq = [self.vocab.get(w, 0) for w in text.lower().split()] seq = seq[:self.max_len] seq = seq + [0] * (self.max_len - len(seq)) sequences.append(seq) return np.array(sequences) def save(self, path): with open(path, 'wb') as f: pickle.dump({'vocab': self.vocab, 'max_len': self.max_len}, f) def load(self, path): with open(path, 'rb') as f: params = pickle.load(f) self.vocab = params['vocab'] self.max_len = params['max_len']

这段代码看起来很朴素，但它解决了一个关键问题：预处理参数和模型权重是分开保存的。推理时你先加载预处理器参数，再加载模型权重，两者必须配套使用。我见过有人把预处理逻辑直接写死在训练脚本里，结果部署的时候忘了同步，导致线上效果和离线评估差了十几个百分点。

3.3 数据版本管理的最小实践

你不需要一上来就搞一套复杂的数据版本管理系统。一个最简单的做法是：每次数据集发生变更时，计算整个数据文件的MD5哈希值，把这个哈希值和变更说明记录到一个data_versions.csv文件里。训练时在日志中记录使用了哪个哈希值的数据。这样至少能保证你事后能追溯。

更进一步的做法是用DVC。DVC的工作原理是在Git里存一个很小的.dvc文件，里面记录了数据文件的哈希值和存储路径，实际数据存在本地目录或对象存储上。当你执行dvc add data.csv时，DVC会计算哈希、把数据移到缓存目录、生成.dvc文件。这样你的Git仓库保持轻量，同时数据版本和代码版本是关联的。

注意：数据版本管理不是可选项。如果你打算长期维护一个AI系统，从第一天就要做。否则三个月后你一定会遇到“这个模型到底是用哪版数据训的”这种问题。

4. 模型训练环节：从脚本到可复现的实验

4.1 为什么你的实验总是无法复现

“上次跑出来F1是0.85，这次怎么只有0.82？”——这是AI工程师最常遇到的灵魂拷问。原因通常有这几个：随机种子没固定、数据划分变了、依赖库版本升级了、硬件环境不同导致浮点运算结果有差异。要解决这个问题，你需要把训练过程当成一个实验来管理，而不是当成一个脚本随便跑跑。

实验管理的核心是记录。每次训练要记录：代码版本（Git commit hash）、数据版本（数据哈希）、超参数配置、环境信息（Python版本、关键库版本、CUDA版本）、随机种子、以及最终的评估指标。这些信息看起来很多，但如果你用配置文件来管理超参数，用工具来自动记录环境信息，实际工作量并不大。

4.2 配置文件驱动的训练脚本

我强烈建议你把所有超参数写到一个YAML或JSON配置文件里，训练脚本只负责读取配置、执行训练、保存结果。这样做的好处是：实验配置可以版本控制，不同实验之间的差异一目了然，复现时只需要用同一个配置文件跑一遍。

# config/train_config.yaml data: source_path: "data/raw/dataset.csv" test_size: 0.2 val_size: 0.1 random_seed: 42 preprocessing: max_vocab_size: 10000 max_sequence_length: 128 model: embedding_dim: 128 hidden_dim: 256 num_layers: 2 dropout: 0.3 training: batch_size: 64 learning_rate: 0.001 epochs: 20 early_stopping_patience: 3 checkpoint_dir: "checkpoints/"

训练脚本读取这个配置，用配置里的随机种子初始化所有随机数生成器，训练完成后把最佳模型的权重、预处理器的参数、以及完整的配置一起保存到一个以时间戳命名的目录里。这样每个实验都是自包含的，不会互相干扰。

4.3 评估指标的选择比模型架构更重要

新手最容易犯的错误是只看准确率。在类别不平衡的数据集上，准确率会骗人。比如一个二分类任务，正样本占5%，负样本占95%，你的模型只要全部预测为负，准确率就有95%，但实际上一事无成。这时候你需要看精确率、召回率、F1分数，甚至AUC-ROC曲线。

选择什么指标取决于你的业务场景。如果是垃圾邮件过滤，你更关心精确率（不能把正常邮件误判为垃圾）；如果是疾病筛查，你更关心召回率（不能漏掉真正的病人）。没有万能的指标，只有适合场景的指标。在训练脚本里，我通常会同时计算多个指标，保存到日志里，方便后续分析。

还有一个容易被忽略的点：评估要在固定的测试集上做。不要每次训练都重新划分测试集，否则你的指标波动里有一部分是数据划分带来的噪声，你根本分不清模型是真的变好了还是运气好。

5. 把模型变成服务：推理部署的关键决策

5.1 批处理、实时推理、流式推理怎么选

模型训练好之后，下一步是让它能被调用。这里有三种常见的部署模式，选择哪种取决于你的业务需求：

批处理最简单，写个脚本定时跑就行。实时推理需要你搭一个HTTP服务，通常用FastAPI或Flask。流式推理则需要消息队列（如Kafka）和流处理框架，复杂度最高。我的建议是：从批处理开始，确认模型效果稳定后再做实时服务。不要一上来就搞微服务架构，那是给自己找麻烦。

5.2 用FastAPI搭一个最小推理服务

FastAPI是目前Python生态里做推理服务最顺手的选择。它自带异步支持、自动生成API文档、性能也够用。一个最小的推理服务大概长这样：

from fastapi import FastAPI from pydantic import BaseModel import numpy as np import pickle import torch app = FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str confidence: float # 启动时加载模型和预处理器 preprocessor = Preprocessor() preprocessor.load("artifacts/preprocessor.pkl") model = torch.load("artifacts/model.pt", map_location="cpu") model.eval() LABELS = ["负面", "正面"] @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): seq = preprocessor.transform([request.text]) tensor = torch.LongTensor(seq) with torch.no_grad(): logits = model(tensor) probs = torch.softmax(logits, dim=-1) pred_idx = torch.argmax(probs, dim=-1).item() confidence = probs[0][pred_idx].item() return PredictResponse( label=LABELS[pred_idx], confidence=round(confidence, 4) )

这个服务看起来很简单，但有几个关键点：模型在启动时加载一次，不要每次请求都重新加载；推理时用torch.no_grad()关闭梯度计算，节省内存；输入输出用Pydantic模型定义，自动做类型校验和文档生成。

5.3 推理服务的性能优化思路

如果你的服务QPS要求不高（比如每秒几十个请求），上面这个版本完全够用。但如果要求更高，就需要做一些优化。最常见的优化手段是批处理：把多个请求攒成一个批次一起推理，充分利用GPU的并行能力。实现方式可以是服务端定时攒批，也可以是客户端主动发批量请求。

另一个优化方向是模型量化。把FP32的权重转成INT8，模型体积缩小4倍，推理速度提升2-3倍，精度损失通常在1%以内。PyTorch提供了动态量化和静态量化两种方式，动态量化最简单，一行代码就能搞定：

quantized_model = torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtype=torch.qint8 )

但要注意，量化后的模型在CPU上加速明显，在GPU上反而不一定。所以量化之前先确认你的部署环境是什么。

提示：推理服务的第一个版本不要追求极致性能。先让它跑起来，能正确返回结果，然后再根据实际压测数据做优化。过早优化是万恶之源。

6. 上线之后：监控、反馈与持续迭代

6.1 模型监控到底该盯哪些指标

模型上线不是终点，而是起点。你需要持续监控三类指标：系统指标、数据指标、业务指标。系统指标包括请求延迟、QPS、错误率、CPU/内存使用率，这些和普通服务监控一样。数据指标包括输入文本的长度分布、词汇分布、以及模型输出的置信度分布。业务指标则是最终的转化率、点击率等。

数据指标是AI系统特有的。如果输入分布发生了明显变化（比如突然出现大量超长文本），模型的输出可能变得不可靠。一个实用的做法是计算输入数据的统计量（均值、方差、分位数），和训练数据的统计量做对比，偏差超过阈值就触发告警。

6.2 建立反馈闭环的最小方案

监控发现问题之后，你需要一个反馈闭环来修复。最简单的闭环是：记录线上推理的输入和输出，定期人工标注一部分样本，用新标注的数据重新训练模型。这个流程不需要很复杂，但必须要有。

具体操作上，你可以在推理服务里加一个日志模块，把每次请求的输入文本、模型输出、置信度、时间戳写到一个日志文件或数据库里。然后每周或每月抽样一批低置信度的样本，交给人工标注，把标注结果加入训练集。重新训练后，在固定的测试集上评估，确认效果提升后再上线。

这个闭环的关键是低置信度样本的利用。模型不确定的样本往往是最有信息量的，人工标注这些样本的性价比最高。不要随机抽样，那样会浪费大量标注预算在模型已经很有把握的样本上。

6.3 模型更新的灰度发布策略

新模型训练好之后，不要直接全量替换。先做灰度发布：把一小部分流量（比如5%）导到新模型上，对比新旧模型在相同流量下的业务指标。如果新模型指标更好，逐步扩大流量比例；如果变差了，立刻回滚。

灰度发布需要一个路由层，根据请求的某些特征（比如用户ID的哈希值）决定走哪个模型。这个路由逻辑可以放在推理服务里，也可以放在网关层。关键是保证同一个用户在灰度期间始终走同一个模型，否则用户体验会不一致。

7. 那些没人告诉你但一定会踩的坑

7.1 训练环境和推理环境不一致

这是最经典的坑。训练时用的是PyTorch 1.12 + CUDA 11.6，推理服务器上装的是PyTorch 1.13 + CUDA 11.7，结果模型加载报错，或者推理结果有细微差异。解决办法是用Docker把训练和推理环境统一起来，用同一个镜像。如果做不到，至少要把依赖版本写死在requirements.txt里，并且在CI流程里验证推理服务能正常加载模型。

7.2 预处理逻辑在训练和推理时不一致

训练时用sklearn的StandardScaler做了归一化，推理时忘了做，或者用了不同的参数。这种错误不会报错，但模型效果会莫名其妙地差。解决办法是把预处理逻辑封装成一个独立的类，训练和推理共用同一份代码，参数从磁盘加载。

7.3 忽略了对模型输入的长度限制

文本分类模型通常有最大序列长度限制。训练时你截断到了128个token，推理时用户输入了一篇1000字的文章，你没有截断就直接喂给模型，结果要么报错，要么模型只看了前128个token，后面的信息全丢了。解决办法是在推理服务里加一个截断逻辑，和训练时保持一致。

7.4 没有做输入校验

用户输入空字符串、纯空格、超长文本、特殊字符，你的服务直接崩溃。推理服务必须做输入校验：检查文本是否为空、长度是否超限、是否包含非法字符。校验不通过就返回明确的错误信息，不要让异常穿透到模型层。

7.5 模型文件太大导致部署困难

有些模型动辄几个GB，部署到生产环境时传输和加载都很慢。解决办法是在保存模型时只保存必要的权重，不要保存优化器状态；如果模型确实很大，考虑量化或剪枝；部署时用对象存储分发模型文件，服务启动时异步加载。

8. 从零到一的路线图：我建议你这样安排时间

如果你现在要从零开始构建AI工程能力，我建议按这个顺序推进：

第一阶段（1-2周）：把数据管道跑通。选一个简单的数据集，写一个可复用的数据加载和预处理模块，实现数据版本管理。这个阶段的目标是让你对“数据在AI系统里怎么流动”有一个具体的感知。

第二阶段（2-3周）：把训练过程规范化。用配置文件管理超参数，固定随机种子，记录实验元信息，保存预处理参数和模型权重。这个阶段的目标是让你的实验可复现。

第三阶段（1-2周）：搭一个最小推理服务。用FastAPI把模型包装成HTTP接口，加上输入校验和日志记录。这个阶段的目标是让你理解“模型怎么变成服务”。

第四阶段（持续）：建立监控和反馈闭环。记录线上数据，定期评估模型表现，用新数据迭代模型。这个阶段没有终点，因为AI系统的维护是一个持续的过程。

每个阶段都不要追求完美，先跑通再优化。我见过太多人卡在“选哪个框架”这种问题上纠结好几天，其实FastAPI和Flask都能用，PyTorch和TensorFlow都能训，选一个顺手的先干起来，后面不合适再换。AI工程是一门实践性极强的技能，看十篇文章不如自己动手搭一遍。你在第一个阶段踩的坑，会比这篇文章里写的所有注意事项都更让你印象深刻。