1. 从零搭建AI工程体系，为什么我劝你别急着调库

很多人一上来就想跑通一个模型，装完环境直接pip install一堆框架，然后复制粘贴一段训练代码，看到loss降了就觉得自己入门了。我早期也这么干过，结果就是：模型跑起来了，但完全不知道数据从哪来、特征怎么处理、推理延迟为什么高、上线后怎么监控。说白了，那叫“调包”，不叫“工程”。

ai-engineering-from-scratch这个标题，核心不在“AI”，而在“from scratch”——从零开始。它要解决的不是“怎么调用一个现成的API”，而是“当你要从零构建一套能跑通、能维护、能扩展的AI工程体系时，每一步该怎么做、为什么这么做”。适合谁看？刚转行做AI工程的同学、从算法研究转向落地部署的工程师、以及想自己搭一套完整pipeline但不知道从哪下手的独立开发者。

我自己的经验是，AI工程和传统后端工程最大的区别在于：不确定性。传统后端接口，输入输出是确定的；AI系统里，数据分布会漂移、模型会退化、推理结果有随机性。所以“from scratch”搭建时，必须把可观测性、可复现性、可回滚性放在第一位，而不是先追求模型精度。这篇文章我会按我实际搭建过的一套最小可行AI工程体系来拆，从数据层到服务层，每一步都给出选型理由、参数计算和踩坑记录。

2. 整体架构设计：先画数据流，再写代码

2.1 为什么我坚持“数据先行”而不是“模型先行”

我见过太多项目，一开始就纠结用BERT还是LLM，结果数据管道一塌糊涂。正确的顺序是：先定义数据契约，再确定模型接口，最后写服务逻辑。数据契约包括：输入数据的schema（字段名、类型、取值范围）、输出数据的schema、以及数据在各个环节的流转格式。

举个例子，假设你要做一个文本分类服务。数据契约应该长这样：

这个表看起来简单，但实际搭建时，80%的bug都出在环节之间的格式不匹配。比如预处理输出的input_ids是int64，但模型要求int32；或者服务层期望confidence是浮点数，但模型输出的是logits没做softmax。先定契约，再写代码，能省掉大量调试时间。

2.2 分层架构：从数据层到服务层的五层模型

我习惯把AI工程体系分成五层，每层职责单一，层间通过明确定义的接口通信：

1. 数据层：负责数据采集、清洗、版本管理。核心工具是DVC（Data Version Control）或LakeFS，用来追踪数据集的变更。为什么不用Git？因为Git不适合大文件，而DVC用指针文件+远程存储的方式，既能版本化又不撑爆仓库。
2. 特征层：负责特征提取、转换、存储。离线用Spark或Pandas，在线用Redis或Feast。关键是要保证离线/在线特征一致性，否则训练时AUC 0.9，上线后掉到0.6。
3. 模型层：负责训练、调参、评估。工具选PyTorch Lightning或HuggingFace Trainer，因为它们把训练循环标准化了，减少重复代码。
4. 服务层：负责推理、批处理、API暴露。轻量级用FastAPI+ONNX Runtime，高并发用Triton Inference Server。
5. 监控层：负责性能监控、数据漂移检测、模型退化告警。工具用Prometheus+Grafana+Evidently。

这五层不是必须全上，但数据层和监控层绝对不能省。我见过太多项目上线后没有监控，模型效果掉了两周才发现，损失已经造成了。

2.3 技术选型的三个核心原则

选型时我遵循三个原则：可替换、可观测、可回滚。

• 可替换：每个组件都要有明确的接口，比如模型推理接口定义为predict(input: dict) -> dict，这样从PyTorch换到ONNX时，上层代码不用改。
• 可观测：每个环节都要打日志、埋指标。比如数据预处理阶段记录null_rate、outlier_rate；推理阶段记录latency_p99、throughput。
• 可回滚：模型版本、数据版本、配置版本都要能一键回滚。我用MLflow管理模型版本，用DVC管理数据版本，用Hydra管理配置版本。

注意：不要为了“技术先进”而选型。我试过用Kafka做实时特征管道，结果运维成本太高，小团队根本扛不住。后来换成Redis Stream，简单够用。

3. 核心细节解析：数据管道与特征工程

3.1 数据清洗：那些文档不会告诉你的脏数据陷阱

原始数据永远比你想的脏。我整理了一份常见脏数据类型和应对策略：

我踩过最坑的一次：训练集里text字段有大量HTML标签，我没清洗直接喂给模型，结果模型学会了根据<div>标签预测类别，上线后真实数据没有HTML标签，效果直接崩了。清洗规则必须和线上预处理逻辑一致，最好把清洗代码封装成函数，离线和在线共用。

3.2 特征工程：从原始文本到模型输入的完整链路

以文本分类为例，完整链路是：原始文本 → 分词 → ID映射 → 截断/填充 → 注意力掩码 → 张量。

分词器选型：中文用BertTokenizer或SentencePiece，英文用ByteLevelBPETokenizer。关键参数是max_length，怎么定？统计训练集文本长度的分布，取95分位数。比如95%的文本长度≤128，那就设max_length=128，这样只截断5%的样本，信息损失最小。

import numpy as np from transformers import BertTokenizer tokenizer = BertTokenizer.from_pretrained('bert-base-chinese') lengths = [len(tokenizer.encode(t)) for t in texts] max_len = int(np.percentile(lengths, 95)) print(f"建议max_length={max_len}")

截断策略：优先保留头部和尾部，中间截断。因为文本分类任务中，开头和结尾往往包含关键信息。填充策略：用pad_token_id填充到max_length，同时生成attention_mask标记真实token位置。

实操心得：分词后的input_ids要检查是否超出词表范围。我遇到过tokenizer.vocab_size=21128，但数据里出现了input_ids=21129，原因是分词器版本和数据版本不匹配。每次加载分词器后，先跑一遍assert max(input_ids) < tokenizer.vocab_size。

3.3 数据版本管理：为什么你的实验无法复现

实验无法复现的根源通常是：数据变了、代码变了、环境变了。数据版本管理用DVC：

dvc init dvc add data/raw/train.csv dvc remote add -d myremote s3://mybucket/dvcstore dvc push

这样每次数据变更都会生成新的.dvc文件，记录哈希值。代码版本用Git，环境版本用conda env export > environment.yml。三者结合，才能保证git checkout到某个commit时，能完整复现实验。

我习惯在每次实验前打tag：git tag -a exp-001 -m "bert-base, max_len=128, lr=2e-5"，然后在MLflow里记录对应的tag。这样半年后回头看，还能知道当时跑了什么。

4. 实操过程：从训练到部署的完整实现

4.1 训练脚本的标准化模板

我不建议用Trainer一把梭，而是自己写训练循环，因为这样能精确控制每个环节。下面是我常用的模板：

import torch from torch.utils.data import DataLoader from transformers import AdamW, get_linear_schedule_with_warmup def train_epoch(model, dataloader, optimizer, scheduler, device): model.train() total_loss = 0 for batch in dataloader: input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['labels'].to(device) optimizer.zero_grad() outputs = model(input_ids, attention_mask=attention_mask, labels=labels) loss = outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() scheduler.step() total_loss += loss.item() return total_loss / len(dataloader)

关键点：梯度裁剪（clip_grad_norm_）防止梯度爆炸，学习率预热（get_linear_schedule_with_warmup）让训练更稳定。warmup_steps一般设为总步数的10%。

4.2 超参数选择：学习率、批大小、epoch的黄金组合

学习率：BERT类模型常用2e-5到5e-5。我试过1e-4，loss直接飞了。批大小：受显存限制，一般16或32。如果显存不够，用梯度累积：

accumulation_steps = 4 loss = loss / accumulation_steps loss.backward() if (step + 1) % accumulation_steps == 0: optimizer.step() optimizer.zero_grad()

这样等效批大小=batch_size * accumulation_steps。epoch：一般3-5轮，用早停（Early Stopping）防止过拟合。监控验证集loss，连续2轮不下降就停。

4.3 模型导出与推理优化

训练完的PyTorch模型直接部署，推理延迟可能很高。我通常导出为ONNX：

torch.onnx.export( model, (dummy_input_ids, dummy_attention_mask), "model.onnx", input_names=['input_ids', 'attention_mask'], output_names=['logits'], dynamic_axes={ 'input_ids': {0: 'batch', 1: 'seq'}, 'attention_mask': {0: 'batch', 1: 'seq'}, 'logits': {0: 'batch'} }, opset_version=13 )

然后用ONNX Runtime推理：

import onnxruntime as ort session = ort.InferenceSession("model.onnx") logits = session.run(None, { 'input_ids': input_ids.numpy(), 'attention_mask': attention_mask.numpy() })[0]

实测下来，ONNX Runtime比原生PyTorch推理快1.5-2倍，而且内存占用更低。如果追求极致性能，可以用TensorRT，但转换成本较高，小项目没必要。

4.4 服务层实现：FastAPI + ONNX Runtime

服务层用FastAPI，因为它的异步支持和自动文档生成很省事：

from fastapi import FastAPI from pydantic import BaseModel import onnxruntime as ort import numpy as np app = FastAPI() session = ort.InferenceSession("model.onnx") class Request(BaseModel): text: str class Response(BaseModel): label: int confidence: float @app.post("/predict", response_model=Response) async def predict(req: Request): inputs = tokenizer(req.text, return_tensors="np", max_length=128, truncation=True, padding="max_length") logits = session.run(None, { 'input_ids': inputs['input_ids'].astype(np.int64), 'attention_mask': inputs['attention_mask'].astype(np.int64) })[0] probs = softmax(logits, axis=-1)[0] label = int(np.argmax(probs)) confidence = float(probs[label]) return Response(label=label, confidence=confidence)

注意：tokenizer要在服务启动时加载一次，不要每次请求都加载。我见过有人在请求里from_pretrained，QPS直接掉到个位数。

4.5 监控层搭建：Prometheus + Grafana + Evidently

监控分三块：系统指标（CPU、内存、延迟）、业务指标（QPS、错误率）、模型指标（数据漂移、预测分布）。

系统指标用prometheus-fastapi-instrumentator自动埋点：

from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(app).expose(app)

数据漂移检测用Evidently，定期跑：

from evidently.report import Report from evidently.metric_preset import DataDriftPreset report = Report(metrics=[DataDriftPreset()]) report.run(reference_data=train_df, current_data=last_week_df) report.save_html("drift_report.html")

如果漂移分数超过阈值（比如0.5），就触发告警。我一般每周跑一次，如果连续两周漂移，就考虑重新训练。

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

5.1 训练不收敛：从loss曲线定位问题

loss曲线是诊断训练问题的第一手资料。常见模式和对策：

我遇到过一次loss突然变NaN，排查发现是某条样本的input_ids全是pad_token_id，导致attention全为0，softmax后出现inf。数据清洗时一定要过滤掉全padding的样本。

5.2 推理延迟高：从CPU到GPU的优化路径

推理延迟高的原因通常有：模型太大、批处理不当、CPU推理、序列太长。优化路径：

1. 量化：把FP32转成INT8，模型大小减4倍，推理速度提升2-3倍。用ONNX Runtime的量化工具：

from onnxruntime.quantization import quantize_dynamic, QuantType quantize_dynamic("model.onnx", "model_int8.onnx", weight_type=QuantType.QUInt8)

2. 批处理：把多个请求攒成一批推理。但要注意延迟和吞吐的权衡，批大小越大，吞吐越高，但单请求延迟也越高。我一般设max_batch_size=32，timeout=10ms。

3. 序列截断：如果max_length=512但实际95%的文本≤128，那就设max_length=128，推理速度提升约3倍。

4. GPU推理：如果QPS高，用GPU。但小模型在GPU上可能因为数据传输开销反而更慢，要实测。

5.3 数据漂移：上线后效果下降的隐形杀手

数据漂移分两种：协变量漂移（输入分布变了）和概念漂移（输入输出关系变了）。检测方法：

• 协变量漂移：用KS检验或PSI（Population Stability Index）比较训练集和线上数据的特征分布。
• 概念漂移：监控线上准确率（如果有标签）或预测置信度分布。

我踩过的坑：线上数据里突然出现大量新词（比如新品牌名），分词器把它们都切成[UNK]，导致模型效果下降。解决方案是定期更新分词器词表，或者用SentencePiece的subword机制，对未登录词更鲁棒。

实操心得：上线前一定要做影子模式（Shadow Mode），即线上流量同时打到新旧模型，对比输出差异。我一般跑一周影子模式，确认新模型没有异常后再切换。

5.4 常见问题速查表

6. 我个人的经验体会与后续扩展方向

这套从零搭建的AI工程体系，我前后迭代了三个版本。第一版只关注模型训练，结果上线后各种问题；第二版加了监控和回滚，稳定了很多；第三版把数据版本和特征一致性做扎实了，才算真正能维护。

如果后续要扩展，我会优先做两件事：自动化重训练和A/B测试框架。自动化重训练就是当数据漂移超过阈值时，自动触发训练管道，训练完自动评估，达标后自动部署到影子模式。A/B测试框架则是把流量按用户ID哈希分流，对比新旧模型的业务指标（比如点击率、转化率），而不是只看准确率。

最后分享一个小技巧：每次上线新模型前，先跑一遍历史数据回测。把过去一个月的线上请求日志拿出来，用新模型重新推理，对比新旧模型的输出差异。如果差异超过10%，就要仔细分析原因。这个步骤能拦住大部分低级错误，比如预处理逻辑不一致、模型版本搞错等。