企业官网建设流程全解析

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

1.1 一个让我彻底改变主意的真实场景

去年帮一个朋友排查线上推理服务的问题，现象很典型：模型在测试集上指标漂亮得不行，一上生产环境延迟直接飙到800ms以上，GPU利用率却只有30%出头。团队里几个小伙子围着代码看了两天，从模型结构查到数据预处理，愣是没找到瓶颈。最后我让他们把推理流程拆开逐段打点，才发现问题出在一个谁都没注意的地方——每次请求都在重新加载tokenizer的词表文件，一个不到2MB的JSON，硬生生被读了上万次。

这件事对我触动挺大。现在市面上讲AI工程的内容，绝大多数都在教你怎么调transformers的pipeline、怎么用LangChain搭个链、怎么在云平台上点几下部署一个端点。这些当然有用，但它们是“用工具”，不是“懂工程”。一旦出了问题，你连从哪儿下手都不知道，因为你根本不清楚工具内部发生了什么。

ai-engineering-from-scratch这个方向，说白了就是反过来走——先别急着调包，咱们从最底层把AI工程涉及的核心环节自己实现一遍。不是为了造轮子而造轮子，而是为了在真正需要调包的时候，你知道那个包里面大概长什么样，哪里可能出问题，出了问题该往哪个方向查。

1.2 这个项目到底适合谁，不适合谁

先说清楚定位，免得你花时间看了半天发现不是自己要的。

适合的人：有一定Python基础，用过至少一个深度学习框架（PyTorch或TensorFlow都行），但总觉得自己是在“黑盒操作”——模型能跑起来，但说不清里面每一步在干什么。想系统性地把AI工程从数据到部署这条链路打通，建立完整的工程直觉。

不太适合的人：完全零编程基础的朋友，建议先去补Python和基础数学；以及已经在做AI工程多年、对底层机制门儿清的老手，这里面的内容对你来说可能偏基础。

我自己的经验是，从零实现一遍的价值不在于你以后真的会手写这些代码，而在于你脑子里有了一张完整的地图。当别人跟你说“模型推理慢”的时候，你能立刻在脑子里定位到可能的几个环节：是预处理拖后腿了？是显存没管好？是batch策略有问题？还是后处理在反复做无用功？这种定位能力，是调包调不出来的。

1.3 整体思路：把AI工程拆成五块硬骨头

我打算按这个顺序来拆：数据处理流水线、模型构建与训练循环、推理优化与部署、监控与可观测性、以及工程化收尾。这五块基本覆盖了一个AI系统从原始数据到线上服务的完整生命周期。

为什么是这个顺序？因为它是数据流动的自然顺序。数据从哪儿来、怎么变成模型能吃的格式、模型怎么学、学完了怎么高效地跑起来、跑起来之后怎么知道它跑得好不好——这是一条线，顺着走不容易乱。

每一块我都会先讲“为什么需要它”，再讲“从零怎么实现一个最小可用版本”，最后讲“实际工程中会踩哪些坑”。代码我会尽量用PyTorch，因为它的动态图机制对理解底层逻辑更友好，而且现在工业界用它的比例也确实高。

注意：从零实现不等于生产级实现。我们写的代码是为了理解原理，性能上肯定比不过成熟库。但理解了原理之后，你再用成熟库的时候，就知道该关注哪些参数、该在哪些地方做优化。

2. 数据处理流水线：别让脏数据毁了你的一切

2.1 为什么数据加载是第一个性能杀手

很多人做AI项目，模型结构调了又调，超参数试了又试，最后发现瓶颈在数据加载上。这不是段子，是常态。我见过一个图像分类任务，模型本身推理只要15ms，但数据预处理加加载花了120ms，整个流水线被拖慢了8倍。

从零实现数据流水线，核心要解决三个问题：怎么高效读、怎么高效转、怎么高效喂。

高效读的关键在于避免随机IO。机械硬盘的随机读取速度比顺序读取慢两个数量级，即使是SSD，大量小文件的随机读也会成为瓶颈。所以工程上的标准做法是：把所有训练数据预先打包成少数几个大文件（比如TFRecord、WebDataset的tar包、或者HDF5），读取的时候顺序读大块，然后在内存里做切分。

高效转的核心是并行化。CPU有多核，但Python的GIL让多线程在计算密集型任务上几乎没用。所以数据预处理要用多进程，torch.utils.data.DataLoader的num_workers参数就是干这个的。但这里有个坑：worker进程和主进程之间的数据传输是有成本的，如果每个样本很小，传输开销可能比计算开销还大。

高效喂的关键是预取。GPU计算的时候，CPU不能闲着，要提前把下一批数据准备好。这就是prefetch_factor参数的作用。理想情况下，数据加载和模型计算应该完全重叠，GPU永远不等数据。

2.2 手写一个带预取和并行加载的Dataset

先看最基础的版本，理解数据流：

import torch from torch.utils.data import Dataset, DataLoader import numpy as np class SimpleDataset(Dataset): def __init__(self, data_path, labels_path): # 实际工程中这里通常是内存映射，不是全量加载 self.data = np.load(data_path, mmap_mode='r') self.labels = np.load(labels_path) def __len__(self): return len(self.labels) def __getitem__(self, idx): # 这里做单样本的预处理 x = self.data[idx].astype(np.float32) / 255.0 y = self.labels[idx] return torch.from_numpy(x), torch.tensor(y)

这个版本能跑，但有几个问题。第一，__getitem__里做了类型转换和归一化，这些操作在每个epoch都会重复执行，浪费算力。第二，没有预取，GPU等数据。第三，如果数据量大，np.load全量加载会爆内存。

改进版本，把预处理结果缓存起来：

class CachedDataset(Dataset): def __init__(self, data_path, labels_path, cache_size=10000): self.data = np.load(data_path, mmap_mode='r') self.labels = np.load(labels_path) self.cache = {} self.cache_size = cache_size def __getitem__(self, idx): if idx in self.cache: return self.cache[idx] x = self.data[idx].astype(np.float32) / 255.0 y = self.labels[idx] result = (torch.from_numpy(x), torch.tensor(y)) # 简单的LRU策略 if len(self.cache) >= self.cache_size: self.cache.pop(next(iter(self.cache))) self.cache[idx] = result return result

但缓存有个问题：如果数据增强是随机的，缓存就失效了。所以实际工程中，缓存的是“确定性预处理”的结果，随机增强放在后面做。

真正高效的方案是用内存映射加预取：

from torch.utils.data import DataLoader # num_workers设为CPU核心数，prefetch_factor设为2-4 loader = DataLoader( dataset, batch_size=64, shuffle=True, num_workers=8, prefetch_factor=4, pin_memory=True, # 如果用的是GPU persistent_workers=True # 避免每个epoch重新创建worker )

pin_memory=True这个参数很多人忽略，但它对GPU训练速度影响很大。它把数据放在锁页内存里，GPU可以直接通过DMA访问，省去了从普通内存拷贝到显存的时间。实测在数据加载密集的场景下，能提升10%-20%的吞吐。

2.3 数据增强的工程化陷阱

数据增强是提升模型泛化能力的利器，但工程上坑很多。

第一个坑是增强操作放在了错误的位置。如果你在__getitem__里做增强，每个epoch都会重新计算，而且多进程下每个worker独立做增强，结果不可复现。更好的做法是把增强放在GPU上做，用torchvision.transforms.v2或者kornia，这样可以利用GPU的并行能力，而且增强参数可以统一控制。

第二个坑是增强强度没有随训练进程调整。训练初期模型欠拟合，强增强会拖慢收敛；训练后期模型容易过拟合，需要强增强来正则化。所以实际工程中常用的是渐进式增强策略，比如AutoAugment或者RandAugment的调度版本。

第三个坑是验证集也做了增强。验证集的目的是评估模型在真实数据上的表现，做增强会引入噪声，导致指标虚高。验证集只做确定性的预处理，比如resize和归一化。

实操心得：我习惯在训练脚本里加一个debug模式，把增强后的样本可视化出来看几眼。很多增强参数配错了，光看代码看不出来，一看图就明白了。比如归一化的均值方差写反了，或者resize的插值方式选错了，这些都会悄悄影响模型性能。

3. 模型构建与训练循环：把每一行代码都搞清楚

3.1 从零写一个训练循环，而不是用Trainer

transformers.Trainer和PyTorch Lightning确实方便，但它们把太多东西藏起来了。从零写训练循环，你会被迫面对每一个细节：梯度什么时候清零、损失怎么累积、学习率怎么调度、梯度裁剪在哪里做。

先看最朴素的版本：

model = MyModel() optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4) criterion = torch.nn.CrossEntropyLoss() for epoch in range(num_epochs): model.train() for batch_idx, (data, target) in enumerate(train_loader): data, target = data.cuda(), target.cuda() optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step()

这个循环能跑，但有几个问题。第一，没有验证环节，你不知道模型有没有过拟合。第二，没有学习率调度，固定学习率很难收敛到最优。第三，没有梯度裁剪，遇到梯度爆炸直接NaN。第四，没有混合精度训练，显存利用率低。

加上这些之后：

scaler = torch.cuda.amp.GradScaler() scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=num_epochs) for epoch in range(num_epochs): model.train() train_loss = 0.0 for data, target in train_loader: data, target = data.cuda(), target.cuda() optimizer.zero_grad() with torch.cuda.amp.autocast(): output = model(data) loss = criterion(output, target) scaler.scale(loss).backward() scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) scaler.step(optimizer) scaler.update() train_loss += loss.item() scheduler.step() # 验证环节 model.eval() val_loss = 0.0 correct = 0 with torch.no_grad(): for data, target in val_loader: data, target = data.cuda(), target.cuda() output = model(data) val_loss += criterion(output, target).item() pred = output.argmax(dim=1) correct += pred.eq(target).sum().item() print(f"Epoch {epoch}: train_loss={train_loss/len(train_loader):.4f}, " f"val_loss={val_loss/len(val_loader):.4f}, " f"val_acc={correct/len(val_dataset):.4f}")

这里有几个关键点值得展开。

混合精度训练：autocast自动把部分运算转成float16，减少显存占用、加速计算。但float16的数值范围小，梯度容易下溢，所以需要GradScaler把损失放大，反向传播后再缩回来。这个机制不理解的话，遇到NaN根本不知道从哪儿查。

梯度裁剪：clip_grad_norm_把所有参数的梯度拼成一个向量，算它的L2范数，如果超过max_norm就整体缩放。这是防止梯度爆炸的标准手段，尤其在RNN和Transformer上几乎是必须的。

学习率调度：CosineAnnealing让学习率按余弦曲线从初始值降到0，比固定学习率收敛得更稳。但T_max要设对，一般等于总训练步数。设小了学习率提前降到0，模型还没收敛；设大了学习率降得太慢，后期震荡。

3.2 梯度累积：小显存跑大batch的实用技巧

显存不够是常态。想用大batch提升训练稳定性，但显卡装不下，怎么办？梯度累积。

原理很简单：正常是每个batch算一次梯度、更新一次参数。梯度累积是算N个batch的梯度、累加起来、再更新一次。效果上等价于把batch size放大了N倍，但显存占用不变。

accumulation_steps = 4 for i, (data, target) in enumerate(train_loader): data, target = data.cuda(), target.cuda() with torch.cuda.amp.autocast(): output = model(data) loss = criterion(output, target) / accumulation_steps scaler.scale(loss).backward() if (i + 1) % accumulation_steps == 0: scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) scaler.step(optimizer) scaler.update() optimizer.zero_grad()

注意损失要除以accumulation_steps，否则梯度会放大N倍。另外optimizer.zero_grad()要放在更新之后，不能每个batch都清。

踩过的坑：有一次我忘了除accumulation_steps，结果梯度大了4倍，学习率没调，直接训练发散。排查了半天才发现是这个低级错误。所以这种地方一定要写注释，提醒自己。

3.3 模型保存与恢复：别只存state_dict

很多人保存模型只存model.state_dict()，恢复的时候还要重新定义模型结构。这在实验阶段没问题，但到了部署阶段就是灾难——你得确保部署环境的代码和训练环境完全一致，否则加载就报错。

更稳妥的做法是保存完整的模型对象，或者至少保存模型结构信息：

# 保存 checkpoint = { 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'scheduler_state_dict': scheduler.state_dict(), 'epoch': epoch, 'config': model_config, # 模型结构配置 'normalization_stats': {'mean': mean, 'std': std} # 预处理参数 } torch.save(checkpoint, 'checkpoint.pt') # 恢复 checkpoint = torch.load('checkpoint.pt') model = MyModel(**checkpoint['config']) model.load_state_dict(checkpoint['model_state_dict'])

把预处理参数也存进去很重要。我见过太多次训练时用了某种归一化，部署时忘了，导致线上效果和离线评估对不上。这种问题排查起来极其痛苦，因为模型本身没问题，是数据分布变了。

4. 推理优化与部署：让模型真正跑得快

4.1 推理和训练到底有什么不同

训练和推理看起来都是前向传播，但工程上的关注点完全不同。

训练关注的是吞吐量——每秒能处理多少样本，因为训练要跑很多个epoch。推理关注的是延迟——单个请求从进到出要多久，因为用户等不了。

训练可以容忍较大的显存占用，因为batch size大反而有助于收敛。推理要尽量省显存，因为要同时服务多个请求。

训练时dropout、batch norm这些层是激活的，推理时要关掉。这个切换用model.eval()完成，但很多人忘了调，导致推理结果和验证结果对不上。

从零实现推理优化，核心是三个方向：减少计算量、减少内存访问、增加并行度。

4.2 手写一个简单的推理服务

先看最朴素的版本，用Flask起一个HTTP服务：

from flask import Flask, request, jsonify import torch import numpy as np app = Flask(__name__) model = MyModel() model.load_state_dict(torch.load('model.pt')) model.eval() model.cuda() @app.route('/predict', methods=['POST']) def predict(): data = request.json['data'] x = torch.tensor(data, dtype=torch.float32).cuda() with torch.no_grad(): output = model(x.unsqueeze(0)) pred = output.argmax(dim=1).item() return jsonify({'prediction': pred})

这个版本能跑，但性能很差。每个请求都要做一次完整的模型前向，没有batch，GPU利用率极低。而且Flask默认是单线程的，并发请求会排队。

改进方向有几个。第一，用动态batch——把短时间内到达的多个请求攒成一个batch一起推理。第二，用异步框架比如FastAPI加uvicorn，支持更高的并发。第三，把模型导出成ONNX或TorchScript，减少Python解释器的开销。

动态batch的实现思路：

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 add_request(self, data): future = asyncio.Future() async with self.lock: self.queue.append((data, future)) if len(self.queue) >= self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch = list(self.queue) self.queue.clear() data = torch.stack([item[0] for item in batch]).cuda() with torch.no_grad(): outputs = self.model(data) for i, (_, future) in enumerate(batch): future.set_result(outputs[i].cpu().numpy())

这个逻辑是：请求进来先放队列，攒够max_batch_size或者等够max_wait_ms就一起推理。这样GPU的利用率能提升好几倍。

4.3 模型量化：用精度换速度的实操

量化是把float32的权重和激活值转成int8，模型大小直接缩小4倍，推理速度提升2-4倍，精度损失通常在1%以内。对于大多数应用场景，这个 trade-off 非常划算。

PyTorch支持动态量化和静态量化。动态量化最简单，一行代码：

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

但动态量化只对Linear层有效，而且是在运行时动态计算量化参数，加速有限。静态量化需要校准数据，提前算好量化参数，加速更明显：

model.qconfig = torch.quantization.get_default_qconfig('fbgemm') model_prepared = torch.quantization.prepare(model) # 用校准数据跑一遍，收集激活值分布 with torch.no_grad(): for data, _ in calib_loader: model_prepared(data) model_quantized = torch.quantization.convert(model_prepared)

校准数据的选择很关键。要用真实分布的数据，不能用随机噪声。一般从训练集里抽几百到几千个样本就够了。

注意事项：量化对某些操作不友好，比如LayerNorm、Softmax这些对数值范围敏感的层。如果量化后精度掉得厉害，可以只量化部分层，保留这些敏感层为float32。PyTorch支持混合量化，通过qconfig精细控制。

4.4 推理引擎选型：ONNX Runtime vs TorchScript

模型训练完，部署的时候不一定非要用PyTorch。ONNX Runtime和TorchScript是两个主流选择。

TorchScript是PyTorch自带的，把模型编译成中间表示，脱离Python解释器运行。优点是兼容性好，PyTorch支持的操作它基本都支持。缺点是优化程度一般。

ONNX Runtime是微软主导的跨框架推理引擎，支持图优化、算子融合、多后端加速。在CPU上通常比TorchScript快，在GPU上差距不大。

导出ONNX的代码：

dummy_input = torch.randn(1, 3, 224, 224).cuda() torch.onnx.export( model, dummy_input, "model.onnx", input_names=['input'], output_names=['output'], dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}}, opset_version=13 )

dynamic_axes很重要，它让导出的模型支持变长batch。不设的话，模型只能处理固定batch size，部署时很受限。

选型建议：如果团队已经在用PyTorch生态，TorchScript上手最快；如果追求极致性能，或者需要跨框架部署，ONNX Runtime更合适。实测在CPU推理场景下，ONNX Runtime通常比TorchScript快20%-50%。

5. 监控与可观测性：上线只是开始

5.1 为什么模型上线后性能会下降

模型在离线评估时指标很好，上线后却慢慢变差，这是AI工程中最常见也最头疼的问题。原因通常有三个：数据漂移、概念漂移、以及工程层面的问题。

数据漂移是输入数据的分布变了。比如训练时用户主要是年轻人，上线后中老年用户增多，输入特征分布就变了。概念漂移是输入和输出的关系变了。比如推荐系统里，用户的兴趣随时间变化，同样的行为对应的偏好不同了。

工程层面的问题更隐蔽。比如特征计算逻辑在训练和推理时不一致，训练时用了未来信息（数据泄露），推理时没有；或者某个特征在训练时是数值型，推理时变成了字符串，类型转换悄悄失败，变成了默认值。

从零实现监控，核心是记录三类指标：输入指标、输出指标、以及性能指标。

5.2 手写一个轻量级监控模块

不需要一上来就上Prometheus加Grafana，先用最朴素的方式把数据记下来：

import json import time from collections import defaultdict import numpy as np class ModelMonitor: def __init__(self, window_size=1000): self.window_size = window_size self.input_stats = defaultdict(list) self.output_stats = defaultdict(list) self.latencies = [] self.errors = 0 self.total = 0 def record(self, inputs, outputs, latency_ms, error=None): self.total += 1 if error: self.errors += 1 return # 记录输入统计 for name, value in inputs.items(): self.input_stats[name].append(float(value)) if len(self.input_stats[name]) > self.window_size: self.input_stats[name].pop(0) # 记录输出统计 if isinstance(outputs, np.ndarray): self.output_stats['mean'].append(float(outputs.mean())) self.output_stats['std'].append(float(outputs.std())) # 记录延迟 self.latencies.append(latency_ms) if len(self.latencies) > self.window_size: self.latencies.pop(0) def get_report(self): report = { 'total_requests': self.total, 'error_rate': self.errors / max(self.total, 1), 'latency_p50': np.percentile(self.latencies, 50) if self.latencies else 0, 'latency_p95': np.percentile(self.latencies, 95) if self.latencies else 0, 'latency_p99': np.percentile(self.latencies, 99) if self.latencies else 0, } for name, values in self.input_stats.items(): if values: report[f'input_{name}_mean'] = np.mean(values) report[f'input_{name}_std'] = np.std(values) return report

这个模块记录输入特征的均值和标准差、输出分布的均值和标准差、以及延迟的分位数。当输入特征的均值偏离训练时的均值超过一定阈值，就说明可能发生了数据漂移。

延迟的P99比P50更重要。用户对平均延迟不敏感，但对最慢的那1%的请求极其敏感。如果P99延迟是2秒，意味着每100个用户就有1个要等2秒，这个体验是灾难性的。

5.3 数据漂移检测的实用方法

检测数据漂移，工程上常用三种方法。

统计检验：对每个特征，比较当前窗口的分布和训练时的分布。连续特征用KS检验，类别特征用卡方检验。p值小于0.05就认为有显著差异。

PSI（Population Stability Index）：把特征分桶，比较各桶的占比变化。PSI小于0.1说明分布稳定，0.1到0.25说明有轻微漂移，大于0.25说明有显著漂移。这个方法在金融风控领域用得很多，因为它对分桶方式不敏感，结果比较稳健。

模型置信度监控：如果模型输出的置信度整体下降，说明输入数据可能偏离了训练分布。这个方法不需要标签，适合实时监控。

def calculate_psi(expected, actual, buckets=10): # 用训练数据的分位数分桶 breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1)) breakpoints[0] = -np.inf breakpoints[-1] = np.inf expected_counts = np.histogram(expected, bins=breakpoints)[0] / len(expected) actual_counts = np.histogram(actual, bins=breakpoints)[0] / len(actual) # 避免除零 expected_counts = np.clip(expected_counts, 1e-6, None) actual_counts = np.clip(actual_counts, 1e-6, None) psi = np.sum((actual_counts - expected_counts) * np.log(actual_counts / expected_counts)) return psi

实操心得：PSI的计算依赖分桶，分桶方式不同结果会有差异。我的经验是，对于长尾分布的特征，用等频分桶比等宽分桶更稳定。另外，PSI要定期算，不能只看一次。我习惯每天算一次，画成趋势图，这样能提前发现缓慢的漂移。

5.4 日志与追踪：出问题时怎么快速定位

监控告诉你“有问题”，日志和追踪告诉你“问题在哪儿”。

结构化日志是基础。不要用print，用logging模块，输出JSON格式，方便后续检索和分析：

import logging import json logger = logging.getLogger(__name__) def log_prediction(request_id, inputs, outputs, latency_ms, model_version): logger.info(json.dumps({ 'request_id': request_id, 'model_version': model_version, 'latency_ms': latency_ms, 'input_shape': list(inputs.shape) if hasattr(inputs, 'shape') else None, 'output_summary': { 'mean': float(outputs.mean()), 'std': float(outputs.std()), 'max': float(outputs.max()), }, 'timestamp': time.time() }))

request_id是关键，它让一个请求在多个服务之间的调用链可以串起来。如果推理服务调用了特征服务，特征服务又调用了数据库，通过request_id能把整条链路串起来，定位瓶颈在哪个环节。

分布式追踪用OpenTelemetry，它能在每个服务里自动埋点，生成调用链的火焰图。火焰图上看哪个span最宽，瓶颈就在哪儿。这个工具上手有点门槛，但一旦用起来，排查性能问题效率提升巨大。

6. 工程化收尾：那些没人告诉你但很重要的事

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

实验阶段把学习率、batch size写死在代码里没问题，但到了工程阶段，这些参数必须外置。原因很简单：不同环境（开发、测试、生产）需要不同的配置，写死在代码里意味着每次改配置都要改代码、重新部署。

用YAML或JSON管理配置：

# config/production.yaml model: name: resnet50 num_classes: 1000 pretrained: true training: batch_size: 128 learning_rate: 0.001 epochs: 100 optimizer: adamw weight_decay: 0.01 data: train_path: /data/train val_path: /data/val num_workers: 8 image_size: 224

加载配置的代码：

import yaml def load_config(env='development'): with open(f'config/{env}.yaml', 'r') as f: config = yaml.safe_load(f) return config

配置管理的一个常见坑是配置项没有默认值。如果某个配置项在某个环境里忘了写，程序直接崩溃。所以加载配置后要做校验，确保所有必填项都存在。

6.2 版本管理：模型、数据、代码一个都不能少

AI工程和传统软件工程最大的区别是：除了代码，还有模型和数据需要版本管理。

代码用Git，这个不用多说。模型版本管理，小团队可以用文件命名规范，比如model_v1.2.3_20240101.pt，大团队建议用MLflow或DVC。数据版本管理更复杂，因为数据量大，不能直接放Git。DVC的思路是：数据文件本身存在对象存储里，Git里只存一个指针文件，记录数据的哈希值。

我自己的习惯是，每次训练完，把模型文件、训练配置、评估指标、以及训练数据的哈希值打包成一个版本，存到对象存储里。这样任何时候都能复现某个版本的模型。

6.3 测试：AI系统怎么测

AI系统的测试比传统软件复杂，因为输出不是确定性的。但有几类测试是必须做的。

数据测试：检查数据格式、范围、分布是否符合预期。比如图像数据要检查像素值是否在0-255之间，标签是否在合法范围内。用great_expectations或pandera可以自动化这些检查。

模型测试：检查模型输出是否在合理范围内，是否对特定输入有预期行为。比如一个分类模型，对随机噪声的输入，输出的置信度应该很低；对明显属于某个类别的输入，输出的对应类别概率应该很高。

集成测试：检查整个流水线从数据加载到推理输出是否正常工作。这个测试要覆盖边界情况，比如空输入、超大输入、格式错误的输入。

回归测试：每次模型更新后，在固定的测试集上评估，确保指标没有下降。这个测试要自动化，集成到CI/CD流程里。

def test_model_output_range(): model = load_model() dummy_input = torch.randn(1, 3, 224, 224) output = model(dummy_input) # 输出应该是概率分布 assert output.shape == (1, 1000) assert torch.allclose(output.sum(dim=1), torch.ones(1), atol=1e-5) assert (output >= 0).all() and (output <= 1).all() def test_model_deterministic(): model = load_model() model.eval() dummy_input = torch.randn(1, 3, 224, 224) with torch.no_grad(): output1 = model(dummy_input) output2 = model(dummy_input) # eval模式下，同样输入应该得到同样输出 assert torch.allclose(output1, output2)

注意：test_model_deterministic只在model.eval()模式下成立。训练模式下有dropout和batch norm的随机性，输出不确定。这个测试能帮你发现忘了调eval()的低级错误。

6.4 文档：写给三个月后的自己

AI项目最容易烂尾的地方是文档。实验阶段大家都不写文档，等到要交接或者要复现的时候，发现什么都记不清了。

我的经验是，至少写三份文档。README：项目是干什么的、怎么安装、怎么跑起来。实验记录：每次实验的配置、结果、结论，用表格记录，方便对比。决策日志：为什么选了这个方案而不是那个，当时的考虑是什么。这份文档在几个月后回头看，价值最大。

实验记录可以用Markdown表格：

决策日志用时间线记录：

2024-01-15: 选择ResNet50而不是ViT，因为训练数据只有10万张，ViT容易过拟合。 2024-01-20: 学习率从1e-3降到1e-4，因为loss震荡严重。 2024-01-25: 加入梯度累积，因为显存不够，batch size从128降到32，累积4步。

这些记录看起来琐碎，但当你三个月后需要复现某个实验，或者新人接手项目的时候，它们能省下大量时间。

6.5 持续集成与持续部署：让更新变得安全

AI系统的CI/CD和传统软件不同，因为模型更新需要额外的验证步骤。

典型的流程是：代码提交触发CI，跑单元测试和集成测试；测试通过后，触发模型训练或模型评估；评估指标超过基线，才允许部署到生产环境。

部署策略上，金丝雀发布（Canary Release）比全量发布安全得多。先把新模型部署到一小部分流量上，观察一段时间，指标正常再逐步扩大流量。如果指标异常，自动回滚到旧模型。

# 简单的金丝雀发布逻辑 class ModelRouter: def __init__(self, stable_model, canary_model, canary_ratio=0.1): self.stable_model = stable_model self.canary_model = canary_model self.canary_ratio = canary_ratio def predict(self, request_id, inputs): # 用request_id做哈希，保证同一个用户始终路由到同一个模型 if hash(request_id) % 100 < self.canary_ratio * 100: return self.canary_model(inputs), 'canary' return self.stable_model(inputs), 'stable'

用request_id做哈希而不是随机数，是为了保证同一个用户的请求始终由同一个模型处理。否则用户刷新一下页面，模型变了，结果也变了，体验很差。

回滚机制要自动化。监控系统检测到新模型的错误率或延迟超过阈值，自动把流量切回旧模型。这个阈值要提前设好，不能等出了问题再临时决定。

7. 一些零散但值钱的经验

7.1 关于工具选型，我的真实看法

工具选型没有绝对的对错，只有适不适合。我的原则是：先用最简单的方案把事做成，遇到瓶颈再换。

比如推理服务，一开始用Flask就够了，QPS上不去再换FastAPI，再上不去再考虑Triton Inference Server。不要一上来就上最复杂的方案，因为复杂方案的学习成本和维护成本都很高，而你的业务可能根本用不到那些高级特性。

再比如实验管理，一开始用Excel记结果完全没问题，实验多了再用MLflow。工具是为了解决问题，不是为了炫技。

7.2 性能优化的优先级

性能优化要按ROI排序。我的经验是：数据加载 > 模型推理 > 模型训练 > 代码微优化。

数据加载优化通常收益最大，因为很多项目的瓶颈都在IO上。把数据打包成大文件、用内存映射、开多进程加载，这些改动能带来数倍的提升。

模型推理优化次之，量化、ONNX Runtime、动态batch，这些能带来2-4倍的提升。

模型训练优化再次之，混合精度、梯度累积、分布式训练，这些能带来1.5-3倍的提升。

代码微优化收益最小，比如把Python循环改成向量化操作，通常只有10%-30%的提升。除非其他都优化完了，否则不值得花时间。

7.3 关于“从零实现”的边界

从零实现是为了理解原理，不是为了替代成熟库。你不需要手写一个比PyTorch还快的卷积算子，但你需要知道卷积在GPU上是怎么并行计算的，这样你才能理解为什么某些形状的输入特别慢。

我的建议是：核心环节自己实现一遍，边缘环节用成熟库。比如数据加载、训练循环、推理服务这些核心环节，自己写一遍收获很大。但像分布式通信、CUDA内核这些，理解原理就行，没必要自己造。

7.4 最后分享一个小技巧

如果你刚开始做AI工程，不知道从哪儿下手，我的建议是：找一个已经跑通的开源项目，把它拆了重写一遍。

不要照着抄，而是理解每一部分在干什么，然后用自己的方式实现。遇到不懂的地方，查文档、看源码、做实验。这个过程很慢，但收获比看十篇教程都大。

我自己就是这么过来的。第一个项目拆的是HuggingFace的transformers里的Trainer，花了整整一周才把训练循环、梯度累积、混合精度这些逻辑理清楚。但理清楚之后，后面再看其他框架的代码，基本都是一眼就懂。

AI工程这个领域，变化很快，但底层的东西变化很慢。数据怎么加载、梯度怎么算、模型怎么部署，这些核心问题十年都没变过。把底层搞扎实了，上层工具怎么变都不慌。