从零手搓AI工程：不调包，从张量到上线的完整实践

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

很多人一上来就想搞个大模型应用，第一反应是找个API接上，或者拉个开源框架跑个demo。结果呢？demo跑通了，一上真实数据就崩，一改需求就懵，一遇到性能瓶颈就束手无策。我见过太多这样的案例了，包括我自己早期也踩过这个坑。

ai-engineering-from-scratch这个标题，核心不在“AI”，而在“from scratch”。它代表的是一种学习路径和工程方法论：不依赖高层抽象，从最底层的矩阵运算、梯度计算、数据管道开始，亲手搭建一个能跑通、能调试、能扩展的AI工程骨架。这件事的价值不在于造轮子，而在于你造过一遍之后，再看那些框架的源码和文档，会有一种“原来如此”的通透感。

这篇文章适合谁？如果你是刚入行一两年、天天调包但心里没底的算法工程师；或者是想从后端、数据方向转AI工程，但被各种框架绕晕的开发者；又或者你单纯就是想搞明白，一个模型从数据到上线，中间到底经历了哪些环节、每个环节的坑在哪里。那这篇内容就是写给你的。我会把整个从零搭建的过程拆开，讲清楚每一步的设计意图、参数选择的依据、以及我实际踩过的坑。全文基于我自己的项目实践和教学经验，不是教科书式的罗列，而是从业者之间的经验交换。

2. 整体架构设计：先想清楚数据怎么流

2.1 为什么我选择“最小可运行闭环”作为起点

很多教程一上来就讲Transformer架构、讲注意力机制，这没错，但容易让人迷失在细节里。我的做法是：先定义一个最小可运行闭环。什么叫最小闭环？就是数据进来、经过一个可学习的模块、产生输出、计算损失、反向传播更新参数、再输出预测结果。这个闭环里，模型可以简单到只有一个线性层，但整个管道必须是完整的、可运行的。

为什么这么做？因为AI工程的核心难点从来不是模型结构本身，而是数据流的管理。你想想，实际工作中，80%的时间花在数据清洗、特征工程、管道搭建、性能调优上，真正改模型结构的时间可能不到20%。如果一开始就陷入模型细节，很容易忽略工程层面的问题，比如数据加载是否高效、内存是否溢出、梯度是否消失、检查点是否可靠。

我试过两种路径：一种是先搭一个简单模型跑通全流程，再逐步替换复杂模型；另一种是先设计一个复杂模型，再补全周边设施。实测下来，第一种路径的调试效率高得多。因为当你的管道是通的，你可以快速定位问题是出在数据、模型还是优化器上。而第二种路径，一旦出问题，你面对的是一个黑盒，排查成本极高。

2.2 模块划分与依赖关系

整个项目我划分为五个核心模块：数据模块、模型模块、训练模块、评估模块、推理模块。每个模块的职责必须清晰，接口必须稳定。

数据模块负责原始数据的读取、清洗、分词、向量化、批处理。这里的关键是定义一个统一的数据接口，比如Dataset类，对外暴露__len__和__getitem__，这样上层的DataLoader就可以用统一的方式处理不同来源的数据。我见过很多项目把数据预处理逻辑散落在训练脚本里，结果就是换个数据集就要重写一遍训练代码，维护成本极高。

模型模块负责定义网络结构、前向传播、参数初始化。这里我建议把模型定义和训练逻辑彻底分开。模型类只关心输入张量到输出张量的映射，不关心损失函数、优化器、设备管理。这样做的好处是，你可以单独测试模型的前向传播是否正确，而不需要启动整个训练流程。

训练模块负责循环控制、梯度计算、参数更新、日志记录、检查点保存。这是整个项目的调度中心。我习惯把训练循环写成一个独立的函数或类，接收模型、数据加载器、优化器、损失函数作为参数。这样你可以用同一套训练逻辑去训练不同的模型，也可以用不同的训练策略去训练同一个模型。

评估模块负责在验证集和测试集上计算指标，比如准确率、F1、AUC等。评估逻辑必须和训练逻辑解耦，因为评估时不需要梯度计算，而且可能需要更复杂的指标计算。我通常会把评估函数写成纯函数，输入模型和数据集，输出指标字典。

推理模块负责加载训练好的模型，对单条或批量数据进行预测。这个模块看起来简单，但实际部署时问题最多，比如输入格式不一致、设备不匹配、批处理大小影响结果等。我建议在训练阶段就写好推理接口，并用单元测试保证训练和推理的结果一致性。

2.3 目录结构设计

一个清晰的目录结构能省下大量找文件的时间。我的习惯是这样的：

project/ ├── configs/ # 配置文件 │ ├── default.yaml │ └── experiment_01.yaml ├── data/ # 数据存放 │ ├── raw/ │ ├── processed/ │ └── interim/ ├── src/ # 源代码 │ ├── data/ # 数据模块 │ │ ├── dataset.py │ │ └── preprocess.py │ ├── models/ # 模型模块 │ │ ├── linear.py │ │ └── mlp.py │ ├── training/ # 训练模块 │ │ ├── trainer.py │ │ └── optimizer.py │ ├── evaluation/ # 评估模块 │ │ └── metrics.py │ └── inference/ # 推理模块 │ └── predictor.py ├── scripts/ # 运行脚本 │ ├── train.py │ └── evaluate.py ├── tests/ # 单元测试 ├── requirements.txt └── README.md

这个结构的好处是，每个模块的职责一目了然，新人接手时能快速定位代码。配置文件独立存放，方便做实验对比。数据和代码分离，避免把大文件提交到版本控制。测试目录保证核心逻辑的稳定性。

注意：不要把所有代码都塞进一个main.py里。我见过一个项目，训练、评估、推理全在一个文件里，三千多行，改一个参数要翻半天。模块化不是为了好看，是为了可维护。

3. 核心细节解析：从张量操作到梯度检查

3.1 数据管道的构建与优化

数据管道是AI工程的地基。地基不牢，后面全是坑。我见过太多项目在数据加载上浪费大量时间，GPU利用率长期低于30%，原因就是数据加载成了瓶颈。

构建数据管道的第一步是确定数据格式。对于文本数据，我通常先统一转换成JSON Lines格式，每行一个样本，包含输入和标签。这样做的好处是流式读取方便，不需要一次性加载到内存。对于图像数据，我会预处理成固定大小的张量，保存为二进制文件，读取时用内存映射。

第二步是定义Dataset类。这里有个关键决策：数据预处理放在__getitem__里还是提前做好？我的经验是，如果预处理操作是确定性的且计算量不大，可以放在__getitem__里，比如归一化、简单的分词。如果预处理涉及随机增强或者复杂计算，最好提前做好并缓存，否则每个epoch都要重复计算，浪费大量时间。

第三步是DataLoader的配置。batch_size的选择需要权衡内存和梯度稳定性。太小了梯度噪声大，太大了内存吃不消。我的经验是从32或64开始试，观察GPU内存占用和损失曲线的平滑程度。num_workers的设置取决于CPU核心数和数据加载的复杂度。如果数据加载是IO密集型的，可以设大一点，比如4到8；如果是CPU密集型的，设成CPU核心数即可。pin_memory在GPU训练时建议开启，可以加速数据传输。

from torch.utils.data import Dataset, DataLoader class TextDataset(Dataset): def __init__(self, file_path, tokenizer, max_len=128): self.samples = [] with open(file_path, 'r', encoding='utf-8') as f: for line in f: self.samples.append(json.loads(line)) self.tokenizer = tokenizer self.max_len = max_len def __len__(self): return len(self.samples) def __getitem__(self, idx): sample = self.samples[idx] encoding = self.tokenizer( sample['text'], max_length=self.max_len, padding='max_length', truncation=True, return_tensors='pt' ) return { 'input_ids': encoding['input_ids'].squeeze(0), 'attention_mask': encoding['attention_mask'].squeeze(0), 'label': torch.tensor(sample['label'], dtype=torch.long) } train_loader = DataLoader( train_dataset, batch_size=64, shuffle=True, num_workers=4, pin_memory=True, drop_last=True )

提示：drop_last=True在训练时建议开启，避免最后一个不完整的batch影响BatchNorm的统计量。但在评估时应该设为False，确保所有样本都被评估到。

3.2 模型定义中的参数初始化与正则化

模型定义看起来简单，但参数初始化和正则化策略直接影响训练能否收敛。我见过太多人直接使用默认初始化，结果训练不动或者梯度爆炸。

对于线性层，PyTorch默认使用Kaiming均匀初始化，这在ReLU激活下表现不错。但如果你用的是Sigmoid或Tanh，建议改用Xavier初始化。为什么？因为Sigmoid在输入较大或较小时梯度接近零，Xavier初始化能保证前向传播的方差一致，避免信号在层间衰减。

import torch.nn as nn class MLP(nn.Module): def __init__(self, input_dim, hidden_dim, output_dim, dropout=0.3): super().__init__() self.fc1 = nn.Linear(input_dim, hidden_dim) self.fc2 = nn.Linear(hidden_dim, hidden_dim) self.fc3 = nn.Linear(hidden_dim, output_dim) self.dropout = nn.Dropout(dropout) self.relu = nn.ReLU() self._init_weights() def _init_weights(self): for m in self.modules(): if isinstance(m, nn.Linear): nn.init.xavier_uniform_(m.weight) if m.bias is not None: nn.init.zeros_(m.bias) def forward(self, x): x = self.relu(self.fc1(x)) x = self.dropout(x) x = self.relu(self.fc2(x)) x = self.dropout(x) x = self.fc3(x) return x

Dropout的概率选择也有讲究。一般来说，全连接层用0.3到0.5，卷积层用0.1到0.3。如果模型参数量大、训练数据少，Dropout可以设大一点；如果模型本身就不大，Dropout太大反而欠拟合。我通常从0.3开始试，观察训练损失和验证损失的差距来调整。

权重衰减是另一种正则化手段。Adam优化器的weight_decay参数通常设为1e-4到1e-2。注意，权重衰减和Dropout不要同时设得太大，否则模型会欠拟合。我的经验是，如果用了较强的Dropout，权重衰减可以设小一点，比如1e-5。

3.3 损失函数的选择与标签平滑

损失函数的选择取决于任务类型。分类任务用交叉熵，回归任务用均方误差，多标签分类用二元交叉熵。这些是基础，但有几个细节容易被忽略。

第一个细节是类别不平衡。如果某个类别的样本数远少于其他类别，交叉熵损失会被多数类主导，导致模型偏向多数类。解决方法有两种：一是给损失函数加权重，少数类的权重设大一点；二是用Focal Loss，降低易分类样本的权重，让模型关注难分类样本。

class FocalLoss(nn.Module): def __init__(self, alpha=1, gamma=2, reduction='mean'): super().__init__() self.alpha = alpha self.gamma = gamma self.reduction = reduction def forward(self, inputs, targets): ce_loss = nn.functional.cross_entropy(inputs, targets, reduction='none') pt = torch.exp(-ce_loss) focal_loss = self.alpha * (1 - pt) ** self.gamma * ce_loss if self.reduction == 'mean': return focal_loss.mean() elif self.reduction == 'sum': return focal_loss.sum() return focal_loss

第二个细节是标签平滑。在分类任务中，硬标签（0或1）容易导致模型过度自信，泛化能力下降。标签平滑把硬标签变成软标签，比如把1变成0.9，把0变成0.1。这样模型不会追求极端的概率输出，泛化能力更好。PyTorch的CrossEntropyLoss自带label_smoothing参数，直接设成0.1即可。

注意：标签平滑在蒸馏任务中要慎用，因为教师模型的软标签本身已经包含了类别间的关系信息，再加标签平滑可能会破坏这种信息。

3.4 优化器与学习率调度

优化器的选择直接影响收敛速度和最终性能。Adam系列是默认选择，但SGD在某些任务上能获得更好的泛化性能。我的建议是：如果训练数据大、模型复杂，先用Adam快速收敛，再考虑用SGD微调。如果训练数据小、模型简单，直接上SGD加动量。

学习率是最重要的超参数。太大不收敛，太小收敛慢。我通常用学习率扫描来确定初始学习率：从1e-6开始，每个epoch乘以10，观察损失曲线，找到损失下降最快的那个学习率，然后取它的一半作为初始值。

学习率调度策略也很关键。常用的有StepLR、CosineAnnealing、ReduceLROnPlateau。我的经验是，CosineAnnealing配合Warmup效果最稳定。Warmup在前几个epoch把学习率从0线性增加到初始值，避免训练初期梯度不稳定。CosineAnnealing让学习率按余弦曲线下降，后期学习率很小，有助于模型收敛到更优的局部极小值。

from torch.optim.lr_scheduler import CosineAnnealingLR, LambdaLR optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4) warmup_epochs = 5 total_epochs = 50 def lr_lambda(epoch): if epoch < warmup_epochs: return (epoch + 1) / warmup_epochs progress = (epoch - warmup_epochs) / (total_epochs - warmup_epochs) return 0.5 * (1 + math.cos(math.pi * progress)) scheduler = LambdaLR(optimizer, lr_lambda)

3.5 梯度检查与数值稳定性

梯度检查是调试模型的重要手段。我习惯在训练前跑一次梯度检查，确认反向传播的实现是否正确。方法很简单：用数值梯度近似解析梯度，比较两者的差异。如果差异大于1e-5，说明反向传播有问题。

def gradient_check(model, loss_fn, inputs, targets, epsilon=1e-6): model.zero_grad() outputs = model(inputs) loss = loss_fn(outputs, targets) loss.backward() param = next(model.parameters()) analytic_grad = param.grad.clone() numeric_grad = torch.zeros_like(param) for i in range(min(param.numel(), 10)): param_flat = param.view(-1) orig = param_flat[i].item() param_flat[i] = orig + epsilon loss_plus = loss_fn(model(inputs), targets).item() param_flat[i] = orig - epsilon loss_minus = loss_fn(model(inputs), targets).item() numeric_grad.view(-1)[i] = (loss_plus - loss_minus) / (2 * epsilon) param_flat[i] = orig diff = (analytic_grad - numeric_grad).abs().max() print(f"Max gradient difference: {diff.item()}") return diff.item()

数值稳定性问题在深度学习中很常见。比如Softmax在输入很大时会溢出，LogSumExp技巧可以解决。交叉熵损失内部已经用了LogSumExp，所以直接调用F.cross_entropy是安全的。但如果你自己实现Softmax，一定要注意减去最大值。

另一个常见问题是梯度爆炸。解决方法有两种：梯度裁剪和梯度归一化。梯度裁剪把梯度的范数限制在一个阈值内，比如1.0。梯度归一化把梯度除以它的范数，保持方向不变但范数为1。我通常用梯度裁剪，因为它更简单且效果稳定。

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

4. 实操过程：从数据到上线的完整流程

4.1 环境准备与依赖管理

环境准备看起来简单，但版本冲突是新手最大的坑。我的建议是用虚拟环境加requirements.txt锁定版本。不要用pip install直接装最新版，因为最新版可能和你的代码不兼容。

python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install torch==2.0.1 torchvision==0.15.2 pip install numpy==1.24.3 pandas==2.0.3 scikit-learn==1.3.0 pip install pyyaml==6.0 tqdm==4.65.0 tensorboard==2.13.0 pip freeze > requirements.txt

提示：PyTorch的版本要和CUDA版本匹配。用nvidia-smi查看CUDA版本，然后去PyTorch官网找对应的安装命令。不要直接pip install torch，那样装的是CPU版本。

4.2 数据预处理与特征工程

数据预处理的目标是把原始数据转换成模型能吃的张量。对于文本数据，流程是：读取原始文本、清洗（去HTML标签、去特殊字符）、分词、构建词表、转换成ID序列、填充或截断到固定长度。

构建词表时，我通常保留频率最高的前30000个词，其余用<UNK>代替。为什么是30000？因为大部分任务中，前30000个词已经覆盖了95%以上的文本内容，再增加词表只会增加模型参数量，收益很小。

from collections import Counter def build_vocab(texts, max_size=30000, min_freq=2): counter = Counter() for text in texts: counter.update(text.split()) vocab = {'<PAD>': 0, '<UNK>': 1, '<CLS>': 2, '<SEP>': 3} for word, freq in counter.most_common(max_size): if freq >= min_freq: vocab[word] = len(vocab) return vocab

对于数值特征，标准化是必须的。我通常用Z-score标准化：减去均值，除以标准差。注意，均值和标准差只能从训练集计算，然后应用到验证集和测试集。如果从整个数据集计算，会造成数据泄露。

from sklearn.preprocessing import StandardScaler scaler = StandardScaler() train_features = scaler.fit_transform(train_features) val_features = scaler.transform(val_features) test_features = scaler.transform(test_features)

4.3 训练循环的完整实现

训练循环是项目的核心。我把它拆成几个关键步骤：前向传播、损失计算、反向传播、梯度裁剪、参数更新、学习率调整、日志记录、检查点保存。

def train_one_epoch(model, dataloader, optimizer, scheduler, loss_fn, device, epoch): model.train() total_loss = 0 correct = 0 total = 0 for batch_idx, batch in enumerate(dataloader): input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['label'].to(device) optimizer.zero_grad() outputs = model(input_ids, attention_mask) loss = loss_fn(outputs, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() preds = outputs.argmax(dim=-1) correct += (preds == labels).sum().item() total += labels.size(0) if batch_idx % 100 == 0: print(f"Epoch {epoch} | Batch {batch_idx} | Loss {loss.item():.4f}") scheduler.step() avg_loss = total_loss / len(dataloader) accuracy = correct / total return avg_loss, accuracy

验证循环和训练循环类似，但要加上torch.no_grad()上下文管理器，关闭梯度计算，节省内存和计算时间。

@torch.no_grad() def evaluate(model, dataloader, loss_fn, device): model.eval() total_loss = 0 correct = 0 total = 0 for batch in dataloader: input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['label'].to(device) outputs = model(input_ids, attention_mask) loss = loss_fn(outputs, labels) total_loss += loss.item() preds = outputs.argmax(dim=-1) correct += (preds == labels).sum().item() total += labels.size(0) avg_loss = total_loss / len(dataloader) accuracy = correct / total return avg_loss, accuracy

4.4 模型保存与加载

模型保存有两种方式：保存整个模型和保存状态字典。我强烈建议保存状态字典，因为保存整个模型会把模型类的定义也序列化进去，一旦代码结构变了，加载就会失败。状态字典只保存参数张量，加载时先实例化模型类，再加载参数。

# 保存 torch.save({ 'epoch': epoch, 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'scheduler_state_dict': scheduler.state_dict(), 'best_metric': best_metric, }, 'checkpoint.pth') # 加载 checkpoint = torch.load('checkpoint.pth', map_location=device) model = MLP(input_dim, hidden_dim, output_dim) model.load_state_dict(checkpoint['model_state_dict']) optimizer.load_state_dict(checkpoint['optimizer_state_dict']) scheduler.load_state_dict(checkpoint['scheduler_state_dict']) start_epoch = checkpoint['epoch'] + 1

注意：map_location参数在跨设备加载时很重要。如果你在GPU上训练，在CPU上推理，不加这个参数会报错。

4.5 推理接口的设计与测试

推理接口的设计目标是：输入原始数据，输出预测结果，中间的所有预处理和后处理都封装在接口内部。这样调用方不需要关心模型细节。

class Predictor: def __init__(self, model_path, vocab, max_len=128, device='cpu'): self.device = device self.max_len = max_len self.vocab = vocab self.model = MLP(input_dim=len(vocab), hidden_dim=256, output_dim=2) checkpoint = torch.load(model_path, map_location=device) self.model.load_state_dict(checkpoint['model_state_dict']) self.model.to(device) self.model.eval() def predict(self, text): tokens = text.split()[:self.max_len] input_ids = [self.vocab.get(t, self.vocab['<UNK>']) for t in tokens] input_ids = input_ids + [self.vocab['<PAD>']] * (self.max_len - len(input_ids)) input_tensor = torch.tensor([input_ids], dtype=torch.long).to(self.device) with torch.no_grad(): outputs = self.model(input_tensor) probs = torch.softmax(outputs, dim=-1) pred = outputs.argmax(dim=-1).item() return {'label': pred, 'probability': probs[0][pred].item()}

推理接口写完后，一定要做一致性测试：用训练时的评估函数和推理接口分别对同一批数据预测，比较结果是否一致。如果不一致，说明预处理或后处理有差异，需要排查。

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

5.1 损失不下降的排查思路

损失不下降是最常见的问题。我的排查顺序是：先检查数据，再检查模型，最后检查优化器。

数据方面，检查输入和标签是否对应正确。我遇到过一次，数据加载时shuffle了输入但没shuffle标签，导致模型学的是随机映射。检查方法是：取一个batch，打印输入和标签，人工确认对应关系。

模型方面，检查前向传播的输出是否合理。比如分类任务，输出应该是logits，经过Softmax后概率和应为1。如果输出全是NaN，说明有数值溢出，检查是否有除零或log零的操作。

优化器方面，检查学习率是否太大或太小。太大导致震荡，太小导致收敛慢。可以尝试把学习率乘以10或除以10，观察损失变化。

5.2 过拟合与欠拟合的识别与处理

过拟合的标志是训练损失持续下降，验证损失先下降后上升。处理方法是增加正则化（Dropout、权重衰减）、减少模型参数量、增加训练数据、早停。

欠拟合的标志是训练损失和验证损失都很高且下降缓慢。处理方法是增加模型复杂度、减少正则化、增加训练轮数、调整学习率。

我通常用一张表来快速判断：

5.3 GPU内存不足的优化策略

GPU内存不足是训练大模型时的常见问题。优化策略按优先级排序：

第一，减小batch size。这是最直接的方法，但可能影响梯度稳定性。可以用梯度累积来模拟大batch：每累积N个batch的梯度再更新一次参数。

第二，使用混合精度训练。PyTorch的torch.cuda.amp可以把部分计算转成float16，减少内存占用并加速计算。

from torch.cuda.amp import autocast, GradScaler scaler = GradScaler() for batch in dataloader: optimizer.zero_grad() with autocast(): outputs = model(inputs) loss = loss_fn(outputs, labels) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()

第三，检查是否有内存泄漏。比如在训练循环里不断往列表里追加张量，导致内存持续增长。解决方法是用.item()提取标量，或者用del及时释放不再使用的张量。

第四，使用梯度检查点。梯度检查点用计算时间换内存空间，在前向传播时不保存中间激活值，反向传播时重新计算。PyTorch的torch.utils.checkpoint可以实现。

5.4 训练速度慢的排查与加速

训练速度慢的原因可能有很多。先定位瓶颈：是数据加载慢、前向传播慢、还是反向传播慢？

用torch.cuda.synchronize()配合time.time()测量每个步骤的耗时。如果数据加载耗时占比高，增加num_workers或优化预处理逻辑。如果前向传播慢，检查模型是否有大量小算子，可以合并。如果反向传播慢，检查是否有不必要的梯度计算。

另一个常见原因是数据在CPU和GPU之间频繁传输。解决方法是一次性把数据移到GPU，或者用pin_memory加non_blocking=True异步传输。

input_ids = batch['input_ids'].to(device, non_blocking=True)

5.5 常见问题速查表

提示：遇到问题先别急着改代码，先打印中间结果。我习惯在关键步骤加print或assert，确认数据形状、数值范围、设备位置都符合预期。很多问题看一眼中间结果就清楚了。

6. 从零搭建的延伸与扩展

6.1 从单机到分布式的演进路径

单机训练跑通后，如果数据量或模型规模上来了，就需要考虑分布式。分布式训练有两种模式：数据并行和模型并行。数据并行是把数据切分到多个GPU，每个GPU有完整的模型副本，梯度汇总后更新。模型并行是把模型切分到多个GPU，每个GPU负责一部分计算。

我建议先从数据并行开始，因为实现简单，PyTorch的DistributedDataParallel几行代码就能搞定。模型并行复杂得多，除非模型大到单卡放不下，否则不建议。

import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP dist.init_process_group(backend='nccl') model = model.to(rank) model = DDP(model, device_ids=[rank])

6.2 模型部署的几种方案对比

模型训练好后，部署方式有几种选择：REST API、gRPC、批处理、边缘部署。REST API适合实时性要求不高的场景，用FastAPI或Flask几行代码就能搭起来。gRPC适合高性能、低延迟的场景，但需要定义protobuf接口。批处理适合离线任务，用脚本定时跑。边缘部署需要模型压缩和量化，把模型变小变快。

我通常先用REST API快速上线，验证业务效果后再考虑优化。不要一上来就追求高性能，先跑通再优化。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() predictor = Predictor('model.pth', vocab) class Request(BaseModel): text: str @app.post('/predict') def predict(request: Request): result = predictor.predict(request.text) return result

6.3 持续集成与自动化测试

AI项目的测试和传统软件不同，除了单元测试，还需要数据测试和模型测试。数据测试检查数据分布是否漂移，模型测试检查模型性能是否下降。

我习惯在CI流程里加几个关键检查：代码风格检查（flake8）、单元测试（pytest）、数据验证（检查缺失值、异常值）、模型性能基线（新模型在固定测试集上的指标不能低于基线）。

# .github/workflows/ci.yml name: CI on: [push] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.9' - name: Install dependencies run: pip install -r requirements.txt - name: Run tests run: pytest tests/ - name: Check model performance run: python scripts/check_baseline.py

6.4 项目复盘与经验总结

这个项目我从零开始搭了三遍，每一遍都有新的体会。第一遍是照着教程抄，跑通了但不知道为什么要这么写。第二遍是带着问题改，把每个模块的接口重新设计了一遍，理解了模块解耦的重要性。第三遍是教别人做，发现了很多自己之前忽略的细节，比如数据泄露、梯度检查、推理一致性。

我的核心体会是：从零搭建的价值不在于代码本身，而在于建立对AI工程全流程的直觉。当你亲手处理过数据加载的瓶颈、梯度爆炸的调试、推理结果的不一致，再看那些框架和工具，你就知道它们解决了什么问题、为什么这么设计、什么时候该用什么时候不该用。

最后分享一个小技巧：每次遇到问题，先别急着搜解决方案，先自己打印中间结果、画损失曲线、做消融实验。自己排查出来的问题，印象最深，也最能积累经验。搜到的答案往往是针对特定场景的，换个场景就不适用了。自己搞明白原理，才能举一反三。