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

"ai-engineering-from-scratch"这个标题，第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地，但绝大多数都是教你import torch然后跑个预训练模型，或者调个API做个聊天机器人。真正从工程角度、从零开始把一套AI系统搭起来的内容，少得可怜。

我自己在这个行业摸爬滚打了十来年，带过团队，也踩过无数坑。最深的体会就是：能跑通一个demo和能上线一个AI系统，中间隔着一整个工程体系。这个项目标题的核心价值，恰恰在于它瞄准的是"from scratch"——不是从零写算法，而是从零构建一套完整的AI工程能力。

说白了，这个项目要解决的是这么几个问题：数据怎么管、模型怎么训、实验怎么追踪、服务怎么部署、线上怎么监控。这些东西，你在Jupyter Notebook里永远学不会。它适合谁？适合那些已经会写Python、懂一点机器学习基础，但一到真实项目就抓瞎的工程师；也适合那些想从数据分析、后端开发转型到AI工程方向的从业者。

我见过太多人，模型训出来准确率95%，结果一上线就崩。为什么？因为训练集和线上数据分布不一致，因为特征工程没有版本管理，因为推理服务没有做批处理优化。这些问题，调包解决不了，只有从工程层面系统性地搭建，才能根治。

所以这篇内容，我打算把"从零搭建AI工程体系"这件事掰开揉碎了讲。不聊虚的，全是实操层面的东西——目录怎么设计、工具怎么选、流程怎么串、坑怎么避。你可以直接抄作业，也可以根据自己的场景做裁剪。

2. 整体架构设计与技术选型思路

2.1 为什么我不推荐一上来就用大厂那套全家桶

很多人一提到AI工程化，第一反应就是上Kubeflow、上MLflow、上Feast，恨不得把整个MLOps生态都装一遍。我理解这种心态，觉得工具越全越专业。但实际干下来，小团队或者个人项目一上来就搞全家桶，基本等于自杀。

原因很简单：维护成本太高。Kubeflow那套东西，光是部署和调试就能耗掉你两周，而且版本兼容性问题能把你逼疯。我试过在一个三人团队里推Kubeflow，结果光是让pipeline跑通就花了三周，真正用来做模型迭代的时间反而被压缩了。

所以我的建议是：从最小可用架构开始，按需演进。什么叫最小可用？就是能完成"数据进→模型出→服务上线"这个闭环的最简配置。具体来说，一个典型的从零搭建的AI工程体系，应该包含这么几层：

这张表你直接拿去用，根据自己团队规模和业务需求选型。核心原则是：每一层都要能独立替换，层与层之间通过标准接口通信。这样你后面想升级某一层，不会牵一发动全身。

2.2 目录结构设计：别小看这件事

我见过太多项目，代码写得挺漂亮，但目录结构一塌糊涂。train.py、model.py、utils.py全堆在根目录，数据文件、配置文件、日志文件混在一起。这种项目，三个月后你自己都看不懂。

从零搭建AI工程体系，第一步就是把目录结构定好。我推荐的结构是这样的：

project/ ├── configs/ # 配置文件 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── data/ # 数据目录（gitignore） │ ├── raw/ │ ├── processed/ │ └── features/ ├── src/ # 源代码 │ ├── data/ # 数据处理模块 │ ├── features/ # 特征工程模块 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ └── serving/ # 推理服务 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 测试代码 ├── scripts/ # 运维脚本 ├── requirements.txt └── README.md

这个结构的关键在于关注点分离。src/下面按功能模块划分，每个模块只负责一件事。configs/用YAML管理不同环境的配置，避免硬编码。data/目录加入.gitignore，用DVC做版本管理。

注意：notebooks/目录只用于探索性分析，任何要上线的代码都必须从notebook迁移到src/下面。我踩过的坑就是，notebook里跑通的逻辑直接复制到生产代码，结果因为执行顺序问题导致结果不一致。

2.3 配置管理：别再把参数写死在代码里

这是新手最容易犯的错误。学习率、batch size、数据路径，全写在代码里。换个数据集就要改代码，调个参数就要重新提交。正确的做法是用配置文件管理所有可变参数。

我习惯用Hydra这个库，它支持配置组合、命令行覆盖、多环境切换。举个例子：

# configs/base.yaml data: path: "data/processed/train.parquet" batch_size: 32 num_workers: 4 model: name: "resnet18" num_classes: 10 pretrained: true training: lr: 0.001 epochs: 50 optimizer: "adam" scheduler: "cosine"

然后在代码里这样用：

import hydra from omegaconf import DictConfig @hydra.main(config_path="configs", config_name="base") def train(cfg: DictConfig): print(f"Batch size: {cfg.data.batch_size}") print(f"Learning rate: {cfg.training.lr}") # 训练逻辑...

这样你想改参数，直接在命令行覆盖就行：python train.py training.lr=0.0001。不用改代码，不用重新提交，实验记录也清晰。

3. 核心模块拆解与实操要点

3.1 数据管道：AI工程的地基

数据管道是整个AI工程体系里最不起眼但最重要的部分。模型效果不好，80%的问题出在数据上。我见过太多团队，模型结构调了又调，最后发现是数据清洗逻辑有bug。

从零搭建数据管道，核心要解决三个问题：数据版本管理、数据验证、特征一致性。

数据版本管理我用DVC。它的原理很简单，就是把大文件存在本地或对象存储，git里只存一个指针文件。这样你每次git checkout到某个commit，对应的数据版本也能通过dvc checkout恢复。具体操作：

# 初始化DVC dvc init # 添加数据文件 dvc add data/raw/train.csv # 提交指针文件 git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m "Add training data v1" # 切换数据版本 git checkout <commit-hash> dvc checkout

数据验证我用Great Expectations。这个工具可以让你定义数据的期望属性，比如"年龄列不能有负数"、"类别列的取值范围是0到9"。每次数据更新后自动跑一遍验证，不通过就阻断流程。

import great_expectations as ge df = ge.read_csv("data/raw/train.csv") df.expect_column_values_to_be_between("age", min_value=0, max_value=120) df.expect_column_values_to_be_in_set("label", [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) results = df.validate()

特征一致性是最容易被忽视的。训练时用的特征计算逻辑，和线上推理时必须完全一致。我的做法是把特征计算逻辑封装成独立的模块，训练和推理都调用同一份代码。如果特征计算依赖历史数据（比如用户过去7天的点击次数），那线上还需要一个特征存储来保证实时性。

实操心得：数据管道的每个环节都要有日志和校验。我习惯在数据加载后打印数据形状、缺失值比例、类别分布。这些信息在排查问题时非常有用。

3.2 实验追踪：别再用Excel记结果了

我刚开始做AI的时候，实验记录全靠Excel。跑了几十组实验后，完全记不清哪个参数对应哪个结果。后来用MLflow，才发现实验追踪原来可以这么优雅。

MLflow的核心概念就三个：Experiment、Run、Artifact。一个Experiment对应一个项目，一个Run对应一次训练，Artifact是这次训练产出的文件（模型权重、配置文件、日志）。

import mlflow mlflow.set_experiment("my-classification-task") with mlflow.start_run(): # 记录参数 mlflow.log_param("lr", 0.001) mlflow.log_param("batch_size", 32) # 训练模型... # 记录指标 mlflow.log_metric("accuracy", 0.95) mlflow.log_metric("f1_score", 0.93) # 记录模型 mlflow.pytorch.log_model(model, "model") # 记录其他文件 mlflow.log_artifact("configs/base.yaml")

跑完实验后，打开MLflow UI，所有实验的参数、指标、产出文件一目了然。你还可以对比不同Run的结果，找出最优参数组合。

MLflow的部署也很简单，本地开发直接用mlflow ui就行。团队协作的话，搭一个MLflow Tracking Server，后端存储用PostgreSQL，Artifact存储用MinIO或S3。

注意：MLflow的Artifact存储路径要规划好。我见过有人把模型权重直接存在本地磁盘，结果磁盘满了导致训练中断。建议一开始就用对象存储。

3.3 模型训练：从脚本到Pipeline

很多人训练模型就是写一个train.py，里面从头到尾串下来。这种脚本跑单次实验没问题，但要做超参搜索、要做多模型对比，就力不从心了。

我的做法是把训练流程拆成Pipeline，每个步骤独立可测试。用PyTorch Lightning或者HuggingFace Trainer这类框架，它们已经把训练循环、分布式训练、混合精度这些脏活累活封装好了。

import pytorch_lightning as pl class MyModel(pl.LightningModule): def __init__(self, lr=0.001): super().__init__() self.save_hyperparameters() self.model = create_model() def training_step(self, batch, batch_idx): x, y = batch logits = self.model(x) loss = F.cross_entropy(logits, y) self.log("train_loss", loss) return loss def validation_step(self, batch, batch_idx): x, y = batch logits = self.model(x) acc = (logits.argmax(dim=1) == y).float().mean() self.log("val_acc", acc) def configure_optimizers(self): return torch.optim.Adam(self.parameters(), lr=self.hparams.lr)

PyTorch Lightning的好处是，它强制你把模型定义、训练逻辑、优化器配置分开。这样代码更清晰，也更容易复用。而且它内置了分布式训练支持，你只需要改一个参数就能从单卡切换到多卡。

超参搜索我用Optuna。它支持多种采样算法（TPE、CMA-ES、随机搜索），还能做剪枝——把明显没希望的实验提前终止，节省计算资源。

import optuna def objective(trial): lr = trial.suggest_float("lr", 1e-5, 1e-2, log=True) batch_size = trial.suggest_categorical("batch_size", [16, 32, 64]) model = MyModel(lr=lr) trainer = pl.Trainer(max_epochs=10) trainer.fit(model, train_loader) return trainer.callback_metrics["val_acc"].item() study = optuna.create_study(direction="maximize") study.optimize(objective, n_trials=50)

3.4 模型服务：别让推理成为瓶颈

模型训好了，怎么暴露给业务方用？最简单的做法是写个Flask应用，加载模型，暴露一个/predict接口。但这种做法在生产环境问题很多：没有批处理、没有并发控制、没有版本管理。

我的推荐是用FastAPI + ONNX Runtime。FastAPI负责API层，ONNX Runtime负责推理加速。ONNX Runtime的好处是它支持多种硬件后端（CPU、GPU、TensorRT），而且推理速度比原生PyTorch快不少。

from fastapi import FastAPI import onnxruntime as ort import numpy as np app = FastAPI() session = ort.InferenceSession("model.onnx") @app.post("/predict") async def predict(data: dict): input_array = np.array(data["features"], dtype=np.float32) outputs = session.run(None, {"input": input_array}) return {"prediction": outputs[0].tolist()}

如果要支持高并发，可以用Triton Inference Server。它支持动态批处理、模型集成、多模型版本管理。不过Triton的配置比较复杂，建议在业务量上来之后再考虑。

实操心得：推理服务的输入输出一定要做严格的类型校验和范围校验。我遇到过线上传入空数组导致服务崩溃的情况，后来加了Pydantic做输入验证，问题就解决了。

4. 实操全流程：从零到一搭建一个图像分类系统

4.1 环境准备与依赖安装

光说不练假把式。这一节我带你走一遍完整流程，搭建一个图像分类系统。假设我们要做一个猫狗分类器，数据集有10000张图片。

首先创建项目目录，初始化git和DVC：

mkdir cat-dog-classifier && cd cat-dog-classifier git init dvc init

然后创建虚拟环境，安装依赖：

python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install torch torchvision pytorch-lightning pip install mlflow dvc great-expectations pip install fastapi uvicorn onnx onnxruntime pip install hydra-core optuna

依赖装好后，创建目录结构：

mkdir -p configs data/raw data/processed src/{data,models,training,serving} experiments notebooks tests scripts

4.2 数据准备与验证

把猫狗图片分别放到data/raw/cats/和data/raw/dogs/目录下。然后用DVC做版本管理：

dvc add data/raw git add data/raw.dvc data/.gitignore git commit -m "Add raw image data"

接下来写数据验证脚本。检查图片数量、尺寸分布、格式是否统一：

from PIL import Image import os from collections import Counter def validate_images(data_dir): stats = {"count": 0, "sizes": Counter(), "formats": Counter()} for root, _, files in os.walk(data_dir): for f in files: if f.lower().endswith(('.png', '.jpg', '.jpeg')): path = os.path.join(root, f) with Image.open(path) as img: stats["count"] += 1 stats["sizes"][img.size] += 1 stats["formats"][img.format] += 1 return stats stats = validate_images("data/raw") print(f"Total images: {stats['count']}") print(f"Size distribution: {stats['sizes'].most_common(5)}") print(f"Format distribution: {stats['formats']}")

如果发现尺寸差异太大，就需要统一resize。我一般统一到224x224，这是大多数预训练模型的输入尺寸。

4.3 模型训练与实验追踪

数据准备好后，开始写训练代码。用PyTorch Lightning定义模型：

import pytorch_lightning as pl import torch import torch.nn.functional as F from torchvision import models class CatDogClassifier(pl.LightningModule): def __init__(self, lr=0.001, pretrained=True): super().__init__() self.save_hyperparameters() self.model = models.resnet18(pretrained=pretrained) self.model.fc = torch.nn.Linear(512, 2) def forward(self, x): return self.model(x) def training_step(self, batch, batch_idx): x, y = batch logits = self(x) loss = F.cross_entropy(logits, y) acc = (logits.argmax(dim=1) == y).float().mean() self.log("train_loss", loss, prog_bar=True) self.log("train_acc", acc, prog_bar=True) return loss def validation_step(self, batch, batch_idx): x, y = batch logits = self(x) loss = F.cross_entropy(logits, y) acc = (logits.argmax(dim=1) == y).float().mean() self.log("val_loss", loss, prog_bar=True) self.log("val_acc", acc, prog_bar=True) def configure_optimizers(self): optimizer = torch.optim.Adam(self.parameters(), lr=self.hparams.lr) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=10) return [optimizer], [scheduler]

然后写训练脚本，集成MLflow：

import hydra from omegaconf import DictConfig import mlflow import pytorch_lightning as pl from pytorch_lightning.loggers import MLFlowLogger @hydra.main(config_path="../configs", config_name="base") def train(cfg: DictConfig): mlflow.set_experiment("cat-dog-classification") with mlflow.start_run(): mlflow.log_params({ "lr": cfg.training.lr, "batch_size": cfg.data.batch_size, "epochs": cfg.training.epochs }) model = CatDogClassifier(lr=cfg.training.lr) logger = MLFlowLogger(experiment_name="cat-dog-classification") trainer = pl.Trainer( max_epochs=cfg.training.epochs, logger=logger, gpus=1 if torch.cuda.is_available() else 0 ) trainer.fit(model, train_loader, val_loader) # 保存模型 trainer.save_checkpoint("experiments/model.ckpt") mlflow.log_artifact("experiments/model.ckpt")

跑几组不同参数的实验，然后在MLflow UI里对比结果。我实测下来，ResNet18在猫狗分类任务上，学习率0.001、batch size 32、训练20个epoch，验证集准确率能到96%左右。

4.4 模型导出与推理服务

训练完成后，把PyTorch模型导出为ONNX格式：

import torch from src.models.classifier import CatDogClassifier model = CatDogClassifier.load_from_checkpoint("experiments/model.ckpt") model.eval() dummy_input = torch.randn(1, 3, 224, 224) torch.onnx.export( model, dummy_input, "experiments/model.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}} )

然后写FastAPI服务：

from fastapi import FastAPI, UploadFile from PIL import Image import io import numpy as np import onnxruntime as ort from torchvision import transforms app = FastAPI() session = ort.InferenceSession("experiments/model.onnx") preprocess = transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) @app.post("/predict") async def predict(file: UploadFile): image = Image.open(io.BytesIO(await file.read())).convert("RGB") input_tensor = preprocess(image).unsqueeze(0).numpy() outputs = session.run(None, {"input": input_tensor}) probs = np.exp(outputs[0]) / np.exp(outputs[0]).sum() label = "dog" if probs[0][1] > 0.5 else "cat" confidence = float(max(probs[0])) return {"label": label, "confidence": confidence}

启动服务：uvicorn src.serving.app:app --host 0.0.0.0 --port 8000。然后可以用curl测试：

curl -X POST -F "file=@test.jpg" http://localhost:8000/predict

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

5.1 训练loss不下降，先查这五个地方

这是新手最常遇到的问题。模型训了半天，loss纹丝不动。别急着调模型结构，先按这个顺序排查：

第一，检查数据标签是否对应正确。我遇到过有人把标签文件读反了，猫的图片标成狗，狗的标成猫。模型当然学不会。排查方法很简单：随机抽几张图，人工核对标签。

第二，检查学习率是否合适。学习率太大，loss会震荡甚至发散；学习率太小，loss下降极慢。建议先用一个较小的学习率（比如1e-4）跑几个epoch，观察loss变化。如果loss几乎不变，可以尝试增大10倍。

第三，检查数据预处理是否一致。训练时用了归一化，验证时没用，或者训练时resize到224，验证时resize到256。这种不一致会导致模型在验证集上表现很差。

第四，检查模型是否真的在更新参数。有时候优化器配置错误，或者参数被requires_grad=False冻结了，模型根本不会更新。可以在训练循环里打印几个参数的梯度范数。

第五，检查batch size是否太小。Batch size太小（比如1或2），梯度估计噪声大，训练不稳定。建议至少设为16或32。

5.2 线上推理结果和离线不一致，怎么排查

这个问题我遇到过好几次，每次原因都不一样。整理成一个排查表：

我踩过最坑的一次是，训练时用了torchvision的归一化参数，推理时手写归一化，结果均值方差写反了。排查了一整天，最后发现是mean=[0.485, 0.456, 0.406]写成了mean=[0.406, 0.456, 0.485]。这种低级错误，真的防不胜防。

避坑技巧：把预处理逻辑封装成一个类，训练和推理都调用同一个类。这样就不会出现不一致的问题。

5.3 实验记录混乱，找不到最优模型

这是实验管理没做好。我的建议是：

• 每次实验必须记录完整的配置，包括数据版本、代码commit、超参数
• 实验命名要有规律，比如resnet18_lr0.001_bs32_20240101
• 重要的实验打tag，比如best、baseline
• 定期清理无效实验，避免MLflow里堆积太多Run

MLflow支持给Run打tag和加note，善用这些功能。我习惯在找到最优模型后，给它打上production标签，然后导出模型文件到指定目录。

5.4 推理服务响应慢，怎么优化

推理服务慢，通常有三个原因：模型太大、没有批处理、CPU推理。

模型太大：考虑模型量化（INT8）、剪枝、蒸馏。ONNX Runtime支持动态量化，能把模型大小压缩到原来的1/4，推理速度提升2-3倍。

没有批处理：单个请求单个推理，GPU利用率极低。可以用Triton的动态批处理，或者自己在服务层做请求聚合。

CPU推理：如果GPU资源紧张，可以用ONNX Runtime的CPU优化，开启多线程和SIMD指令集。实测下来，ResNet18在CPU上单张推理大概20ms，批处理32张大概200ms，吞吐量提升明显。

# ONNX Runtime CPU优化配置 options = ort.SessionOptions() options.intra_op_num_threads = 4 options.inter_op_num_threads = 4 options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL session = ort.InferenceSession("model.onnx", options)

6. 工程化落地的几个关键决策

6.1 什么时候该上分布式训练

分布式训练不是银弹。我见过有人单卡能跑完的实验，非要上8卡分布式，结果调试花了两天，训练时间只节省了不到一半。

判断标准很简单：单卡训练时间超过你能接受的等待时间，就考虑分布式。比如单卡要训3天，那上4卡能压缩到1天以内，值得折腾。如果单卡只要2小时，那分布式带来的收益远不及调试成本。

分布式训练用PyTorch Lightning的DDP模式，基本不用改代码，只需要改Trainer的配置：

trainer = pl.Trainer( accelerator="gpu", devices=4, strategy="ddp", max_epochs=50 )

但要注意，DDP模式下数据加载器要用DistributedSampler，否则每个进程会加载全量数据，导致重复训练。

6.2 模型版本管理怎么做

模型版本管理是AI工程里容易被忽视的一环。我的做法是：

• 每次训练产出的模型，用model_name_version_date格式命名
• 模型文件存到对象存储，元数据（训练配置、指标、数据版本）存到数据库
• 线上服务加载模型时，从配置中心读取版本号，支持热更新

简单点的话，可以用MLflow的Model Registry。它提供了模型版本管理、阶段转换（Staging、Production、Archived）的功能。

import mlflow # 注册模型 mlflow.register_model( "runs:/<run_id>/model", "cat-dog-classifier" ) # 转换阶段 client = mlflow.tracking.MlflowClient() client.transition_model_version_stage( name="cat-dog-classifier", version=1, stage="Production" )

6.3 监控告警怎么搭

模型上线不是终点，而是起点。线上监控要关注三类指标：

服务指标：QPS、延迟、错误率。这些用Prometheus + Grafana就能搞定。

模型指标：预测分布、置信度分布。如果线上预测的类别分布突然偏移，说明数据分布可能变了。

数据指标：输入特征的统计量（均值、方差、缺失率）。如果线上输入和训练数据分布差异大，模型效果会下降。

数据漂移检测可以用Evidently。它支持多种漂移检测方法（PSI、KL散度、KS检验），还能生成可视化报告。

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

我一般设置每周跑一次漂移检测，如果漂移程度超过阈值，就触发告警，提醒团队考虑重新训练模型。

7. 我踩过的那些坑和最后的建议

做AI工程这些年，踩过的坑真的不少。有些是技术问题，有些是流程问题，还有些纯粹是沟通问题。

技术层面，最大的坑是过度设计。一开始就想着搭一套完美的系统，结果花了两周搭架子，真正跑通第一个模型又花了两周。后来我学乖了，先用最土的办法跑通闭环，然后再逐步优化。比如数据版本管理，一开始直接用文件夹加日期命名，后面数据量大了再上DVC。

流程层面，最大的坑是训练和推理代码不一致。这个问题我强调过很多次，但每次带新人都还会遇到。后来我强制要求：任何特征计算逻辑，必须封装成独立函数，训练和推理都调用同一份代码。代码review的时候，重点检查这一点。

沟通层面，最大的坑是和业务方对齐预期。业务方觉得AI是魔法，输入数据就能出结果。实际上AI系统有它的局限性，数据质量、场景覆盖、边界情况都需要提前沟通。我现在的习惯是，项目启动前先和业务方明确：模型能解决什么问题，不能解决什么问题，效果指标怎么定义。

最后分享一个小技巧：建立自己的实验日志。除了MLflow自动记录的内容，我还会手写一些笔记，记录每次实验的直觉判断、遇到的异常现象、下一步的想法。这些非结构化的信息，在后续排查问题时往往比结构化数据更有用。

这个项目后续还可以扩展的方向很多：加入A/B测试框架，支持多模型灰度发布；加入自动化重训练流程，当数据漂移超过阈值时自动触发；加入模型解释性工具，帮助业务方理解模型决策。但这些都是锦上添花，核心的工程体系搭好了，后面加什么都顺手。