1. 从零手搓AI工程为什么我不建议你直接调包很多人一上来就想跑通一个能对话的模型结果卡在环境配置、显存不足、依赖冲突上三天热情耗尽直接放弃。我见过太多这样的案例包括我自己早期也是这么过来的。ai-engineering-from-scratch这个方向的核心价值不是让你复现一个GPT-4而是让你亲手把数据、模型、训练循环、推理服务这条链路完整走一遍。走完之后你再去看那些封装好的框架会有一种“原来底层就这么回事”的踏实感。这篇文章适合谁如果你已经会写Python懂一点线性代数和概率论但从来没自己从头搭过一个能跑起来的AI工程那这篇就是写给你的。我会把整个流程拆成可操作的步骤每一步都告诉你为什么这么做、不这么做会出什么问题。全程不依赖任何需要特殊网络环境才能访问的资源所有工具和库都是公开可获取的。先明确一个概念所谓“从零”不是让你用汇编去写矩阵乘法而是指你不依赖高度封装的训练框架比如直接调model.fit()那种而是自己定义模型结构、自己写训练循环、自己处理数据管道。PyTorch 或 JAX 这类基础张量库可以用因为它们本身就是“从零”的一部分。真正的“从零”是指你对每一行代码在干什么都有掌控感。我个人的经验是第一次走通这个流程大概需要两到三周每天投入两三个小时。如果你之前完全没有接触过深度学习可能需要一个月。但走通之后你再学任何新模型、新框架速度会快三到五倍。因为你已经知道那些抽象层下面藏着什么。2. 环境搭建把地基打牢再动工2.1 硬件选择的现实考量先说硬件。如果你有一张显存8GB以上的NVIDIA显卡那是最好的。如果没有CPU也能跑只是训练小模型的时候速度会慢很多。我实测下来一个参数量在100万左右的简单模型用CPU训练一个epoch大概需要几分钟用GPU可能只要几秒。但重点是流程走通不是追求速度。如果你用的是Apple Silicon的MacPyTorch对MPS后端有支持也能用。但要注意某些操作在MPS上可能没有实现会回退到CPU。我建议第一次走流程的时候先用CPU跑通确保逻辑没问题再切到GPU加速。内存方面16GB是底线32GB会更从容。因为数据处理的时候如果你把整个数据集加载到内存里很容易爆。后面我会讲怎么用流式加载来规避这个问题。2.2 Python环境与依赖管理我强烈建议用conda或者venv创建一个独立的环境。不要在你的系统Python里直接装包否则版本冲突会让你痛不欲生。我踩过的坑是之前用系统Python装了一个版本的PyTorch后来又装了一个需要不同CUDA版本的库结果两个都用不了最后只能重装系统。conda create -n ai-scratch python3.10 conda activate ai-scratchPython版本选3.10或3.11都可以太新的版本有些库还没适配。然后安装核心依赖pip install torch numpy matplotlib tqdm如果你有NVIDIA显卡去PyTorch官网查一下对应的CUDA版本安装命令。不要直接pip install torch那样装的是CPU版本。我见过有人装了CPU版本训练了半天发现没用上显卡白白浪费几个小时。2.3 目录结构的设计在写代码之前先把目录结构定好。这不是强迫症而是为了后面不混乱。我的习惯是这样的ai-from-scratch/ ├── data/ # 存放原始数据和预处理后的数据 ├── src/ # 源代码 │ ├── model.py # 模型定义 │ ├── dataset.py # 数据加载 │ ├── train.py # 训练循环 │ └── inference.py # 推理服务 ├── checkpoints/ # 保存模型权重 ├── logs/ # 训练日志 └── configs/ # 配置文件这个结构看起来简单但能帮你省很多事。比如你想换一个模型试试只需要改model.py其他文件不用动。想换数据集只改dataset.py。这种模块化的思路是从零做工程的关键。注意不要把所有代码写在一个文件里。我早期图省事一个main.py写了八百行后来想改一个参数找了半天才找到在哪。模块化不是为了好看是为了你自己后面能维护。3. 数据管道模型吃进去的每一口都要干净3.1 数据加载的三种方式与选择逻辑数据加载看起来简单其实是最容易出问题的地方。我把它分成三种方式第一种是全部加载到内存。适合小数据集比如几万条文本或者几千张图片。优点是速度快缺点是内存占用大。如果你有32GB内存加载一个几GB的数据集没问题。第二种是流式加载。每次只读一个batch的数据用完就释放。适合大数据集比如几十GB的图片。缺点是IO速度可能成为瓶颈需要配合多进程预读取。第三种是内存映射。把数据文件映射到虚拟内存操作系统帮你管理换入换出。适合中等规模、访问模式比较随机的数据。我一般先用第一种方式快速验证流程跑通之后再根据数据规模切换到第二种或第三种。不要一上来就搞复杂的先把简单流程跑通。3.2 自定义Dataset的写法与常见陷阱PyTorch提供了Dataset和DataLoader两个类。你需要继承Dataset实现__len__和__getitem__两个方法。看起来简单但有几个坑第一个坑是__getitem__里不要做太重的计算。比如你每次取数据都做一次复杂的增强那训练速度会被拖慢。正确的做法是在__init__里做好预处理或者用collate_fn在batch层面做。第二个坑是随机种子。如果你在__getitem__里用了随机数做数据增强但没有固定种子那每次取到的数据都不一样调试的时候很难复现问题。我的做法是在__init__里创建一个random.Random(seed)实例每个worker用自己的种子。第三个坑是多进程加载时的内存泄漏。如果你用了num_workers 0每个worker会复制一份数据集对象。如果数据集很大内存会成倍增长。解决办法是用mmap或者把数据路径存起来在__getitem__里再读取。class TextDataset(Dataset): def __init__(self, file_path, tokenizer, max_len128): self.data [] with open(file_path, r, encodingutf-8) as f: for line in f: self.data.append(line.strip()) self.tokenizer tokenizer self.max_len max_len def __len__(self): return len(self.data) def __getitem__(self, idx): text self.data[idx] tokens self.tokenizer(text, max_lengthself.max_len, paddingmax_length, truncationTrue) return { input_ids: torch.tensor(tokens[input_ids]), attention_mask: torch.tensor(tokens[attention_mask]) }这个写法适合小数据集。如果数据量大把self.data换成文件路径列表在__getitem__里按行读取。3.3 批处理与填充的策略批处理的时候同一个batch里的序列长度要一致所以需要填充。填充的策略有两种一种是填充到固定长度比如128一种是填充到当前batch的最大长度。固定长度的好处是形状统一方便调试。坏处是如果大部分序列都很短会浪费很多计算。动态填充的好处是效率高坏处是每个batch的形状不一样某些操作会报错。我的建议是训练的时候用动态填充配合collate_fn自己实现。推理的时候用固定长度方便部署。def collate_fn(batch): max_len max([len(item[input_ids]) for item in batch]) input_ids torch.stack([ F.pad(item[input_ids], (0, max_len - len(item[input_ids]))) for item in batch ]) attention_mask torch.stack([ F.pad(item[attention_mask], (0, max_len - len(item[attention_mask]))) for item in batch ]) return {input_ids: input_ids, attention_mask: attention_mask}提示填充的token id要和padding token的id一致。如果你用的是BERT类的tokenizerpadding token通常是0。但如果你自己训练tokenizer要确认一下。4. 模型定义从矩阵乘法到Transformer4.1 最小可训练模型的结构设计第一次从零做不要一上来就写Transformer。先用一个最简单的模型把流程跑通。比如一个文本分类模型嵌入层 平均池化 全连接层。这个模型只有几千个参数训练几分钟就能看到loss下降。class SimpleClassifier(nn.Module): def __init__(self, vocab_size, embed_dim, num_classes): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim) self.fc nn.Linear(embed_dim, num_classes) def forward(self, input_ids, attention_mask): embeds self.embedding(input_ids) # (batch, seq_len, embed_dim) mask attention_mask.unsqueeze(-1).float() pooled (embeds * mask).sum(dim1) / mask.sum(dim1) logits self.fc(pooled) return logits这个模型虽然简单但包含了所有核心要素嵌入、掩码、池化、分类头。跑通之后你再把中间部分换成Transformer层就水到渠成了。4.2 注意力机制的代码实现与维度验证注意力机制是Transformer的核心。我第一次写的时候维度搞错了矩阵乘法报错调了两个小时才找到问题。所以这里我把维度变化写清楚。输入是(batch, seq_len, d_model)。经过三个线性层得到Q、K、V形状都是(batch, seq_len, d_model)。然后Q乘以K的转置得到(batch, seq_len, seq_len)的注意力分数。除以sqrt(d_k)做缩放再经过softmax最后乘以V得到(batch, seq_len, d_model)的输出。class SelfAttention(nn.Module): def __init__(self, d_model, num_heads): super().__init__() self.num_heads num_heads self.d_k d_model // num_heads self.q_linear nn.Linear(d_model, d_model) self.k_linear nn.Linear(d_model, d_model) self.v_linear nn.Linear(d_model, d_model) self.out_linear nn.Linear(d_model, d_model) def forward(self, x, maskNone): batch, seq_len, _ x.shape q self.q_linear(x).view(batch, seq_len, self.num_heads, self.d_k).transpose(1, 2) k self.k_linear(x).view(batch, seq_len, self.num_heads, self.d_k).transpose(1, 2) v self.v_linear(x).view(batch, seq_len, self.num_heads, self.d_k).transpose(1, 2) scores torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) attn F.softmax(scores, dim-1) out torch.matmul(attn, v) out out.transpose(1, 2).contiguous().view(batch, seq_len, -1) return self.out_linear(out)维度验证的方法在forward里加print看每一步的形状。或者用torchsummary这个库。我习惯在第一次写的时候加print确认没问题再删掉。4.3 残差连接与层归一化的位置选择残差连接和层归一化的位置有两种常见方案Post-LN和Pre-LN。原始Transformer用的是Post-LN就是先做注意力或前馈再加残差最后做层归一化。Pre-LN是先做层归一化再做注意力或前馈最后加残差。Post-LN的问题是训练深层模型时梯度容易爆炸需要小心的学习率预热。Pre-LN更稳定现在大部分模型都用Pre-LN。我建议从零做的时候直接用Pre-LN省去调学习率的麻烦。class TransformerBlock(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout0.1): super().__init__() self.attn SelfAttention(d_model, num_heads) self.norm1 nn.LayerNorm(d_model) self.norm2 nn.LayerNorm(d_model) self.ff nn.Sequential( nn.Linear(d_model, d_ff), nn.ReLU(), nn.Dropout(dropout), nn.Linear(d_ff, d_model) ) self.dropout nn.Dropout(dropout) def forward(self, x, maskNone): x x self.dropout(self.attn(self.norm1(x), mask)) x x self.dropout(self.ff(self.norm2(x))) return x注意Pre-LN的输出在最后还需要一个额外的层归一化。这个细节容易漏掉导致输出分布不稳定。5. 训练循环让loss降下去的艺术5.1 损失函数与优化器的搭配逻辑损失函数的选择取决于任务。分类任务用交叉熵回归任务用均方误差序列生成任务用交叉熵但需要shift。我见过有人做分类用了MSEloss也能降但收敛很慢而且精度上不去。优化器方面Adam和AdamW是最常用的。AdamW是Adam的改进版把权重衰减从梯度更新里分离出来效果更好。我建议直接用AdamW学习率设1e-4到3e-4之间。如果模型很小可以用1e-3。optimizer torch.optim.AdamW(model.parameters(), lr3e-4, weight_decay0.01) criterion nn.CrossEntropyLoss()学习率调度也很重要。我一般用余弦退火或者线性预热加衰减。预热的意思是前几百步学习率从0慢慢升到设定值避免一开始梯度太大把模型带偏。from torch.optim.lr_scheduler import CosineAnnealingLR, LinearLR, SequentialLR warmup LinearLR(optimizer, start_factor0.1, total_iters500) cosine CosineAnnealingLR(optimizer, T_maxtotal_steps - 500) scheduler SequentialLR(optimizer, schedulers[warmup, cosine], milestones[500])5.2 梯度裁剪与混合精度训练的实操细节梯度裁剪是防止梯度爆炸的常用手段。特别是用RNN或者深层Transformer的时候梯度可能会变得非常大。裁剪的方法很简单torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)这个操作要在loss.backward()之后、optimizer.step()之前做。max_norm设1.0是常见值也可以设0.5或5.0根据实际情况调。混合精度训练是用float16代替float32做前向和反向可以节省显存、加快速度。PyTorch提供了torch.cuda.amp来实现scaler torch.cuda.amp.GradScaler() for batch in dataloader: optimizer.zero_grad() with torch.cuda.amp.autocast(): outputs model(batch[input_ids], batch[attention_mask]) loss criterion(outputs, batch[labels]) scaler.scale(loss).backward() scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) scaler.step(optimizer) scaler.update()注意scaler.unscale_要在裁剪之前调用否则裁剪的是缩放后的梯度不对。这个坑我踩过loss一直不降查了半天才发现是顺序错了。5.3 训练过程中的监控与日志记录训练的时候一定要记录loss和准确率。我习惯用tqdm显示进度条同时把每个epoch的指标写到日志文件里。from tqdm import tqdm for epoch in range(num_epochs): model.train() total_loss 0 pbar tqdm(dataloader, descfEpoch {epoch}) for batch in pbar: # ... training step ... total_loss loss.item() pbar.set_postfix({loss: loss.item()}) avg_loss total_loss / len(dataloader) with open(logs/train.log, a) as f: f.write(fEpoch {epoch}, loss: {avg_loss}\n)除了loss还要看梯度范数。如果梯度范数突然变得很大说明可能有问题。可以在裁剪之前记录一下grad_norm torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) if grad_norm 10: print(fWarning: large grad norm {grad_norm})提示如果loss变成nan先检查学习率是不是太大再检查数据里有没有异常值比如全零的输入最后检查有没有除零操作。6. 推理与部署让模型真正跑起来6.1 模型导出与加载的注意事项训练完之后要把模型权重保存下来。最简单的方式是用torch.savetorch.save(model.state_dict(), checkpoints/model.pt)加载的时候先创建同样结构的模型再加载权重model SimpleClassifier(vocab_size, embed_dim, num_classes) model.load_state_dict(torch.load(checkpoints/model.pt)) model.eval()注意model.eval()一定要调用否则dropout和batch norm的行为会和训练时不一样导致推理结果不稳定。如果要在没有PyTorch的环境里部署可以用ONNX导出dummy_input torch.randint(0, vocab_size, (1, 128)) torch.onnx.export(model, dummy_input, model.onnx, input_names[input_ids], output_names[logits])ONNX的好处是跨平台坏处是某些自定义操作可能不支持。导出之后要用ONNX Runtime验证一下输出是否一致。6.2 批处理推理与延迟优化推理的时候如果一条一条来延迟会很高。批处理可以显著提高吞吐量。但批处理也有代价需要等凑够一个batch才能处理增加了等待时间。我的做法是设置一个最大batch size和一个最大等待时间。比如最多等50毫秒或者凑够32条就立即处理。这样在延迟和吞吐之间取得平衡。import time class BatchInference: def __init__(self, model, max_batch32, max_wait0.05): self.model model self.max_batch max_batch self.max_wait max_wait self.buffer [] self.last_time time.time() def add(self, input_ids): self.buffer.append(input_ids) if len(self.buffer) self.max_batch or \ time.time() - self.last_time self.max_wait: return self.flush() return None def flush(self): if not self.buffer: return None batch torch.stack(self.buffer) with torch.no_grad(): outputs self.model(batch) self.buffer [] self.last_time time.time() return outputs这个模式在线上服务里很常见。如果你只是本地测试直接批处理就行不用这么复杂。6.3 常见推理错误的排查思路推理时最常见的错误是形状不匹配。训练的时候batch size是32推理的时候输入是1某些操作比如batch norm会报错。解决办法是在模型里用BatchNorm的时候设置track_running_statsTrue或者在推理时手动扩展维度。另一个常见错误是设备不匹配。模型在GPU上输入在CPU上会报错。解决办法是统一设备device torch.device(cuda if torch.cuda.is_available() else cpu) model model.to(device) input_ids input_ids.to(device)还有一个坑是tokenizer的padding side。有些tokenizer默认在左边填充有些在右边。如果训练和推理的填充方向不一致结果会差很多。我建议统一用右边填充然后在模型里用attention mask来处理。7. 从跑通到跑好我踩过的五个坑7.1 数据泄漏为什么你的验证集准确率虚高数据泄漏是指训练集和验证集有重叠。最常见的情况是你先做了数据增强然后才划分训练集和验证集。这样同一个样本的增强版本可能同时出现在训练集和验证集里导致验证集准确率虚高。正确的做法是先划分再分别做增强。如果数据量很小可以用交叉验证但也要确保每一折的验证集是独立的。我踩过这个坑做一个文本分类任务验证集准确率到了95%但上线之后效果很差。查了半天才发现同一个文本的不同截断版本分别进了训练集和验证集。7.2 学习率设置太大震荡太小不动学习率是训练中最难调的参数。太大loss震荡不收敛太小loss下降极慢。我的经验是先用一个中等值比如1e-3跑几百步看loss曲线。如果震荡除以10如果几乎不动乘以10。还有一个技巧是学习率范围测试从很小的值开始每步指数增长记录loss。loss下降最快的点就是合适的学习率。这个方法叫LR Range Test Leslie Smith提出的很实用。7.3 过拟合与欠拟合的判断与应对过拟合的表现是训练loss持续下降验证loss先降后升。应对方法有增加数据、加正则化dropout、weight decay、早停、减小模型。欠拟合的表现是训练loss和验证loss都很高降不下去。应对方法有增大模型、增加训练轮数、提高学习率、检查数据是否有问题。我一般先看训练loss能不能降到很低。如果训练loss都降不下去那肯定是欠拟合先加大模型或者调学习率。如果训练loss很低但验证loss高那就是过拟合加正则化或者加数据。7.4 随机种子为什么你的结果无法复现深度学习里有很多随机操作参数初始化、数据打乱、dropout。如果不固定种子每次跑的结果都不一样。这对调试很不友好。import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False注意cudnn.deterministic True会降低速度但能保证结果可复现。调试的时候打开最终训练的时候可以关掉。7.5 显存不足从batch size到梯度累积显存不足是最常见的问题。解决办法有几个减小batch size、用梯度累积、用混合精度、用梯度检查点。梯度累积是模拟大batch的效果每算一个mini-batch的梯度不立即更新参数而是累加梯度等累积到一定步数再更新。accumulation_steps 4 for i, batch in enumerate(dataloader): loss model(batch) / accumulation_steps loss.backward() if (i 1) % accumulation_steps 0: optimizer.step() optimizer.zero_grad()这样等效于batch size扩大了4倍但显存占用不变。代价是训练速度会慢一些因为多了几次前向反向。注意用梯度累积的时候学习率也要相应调整。如果等效batch size扩大了4倍学习率也可以适当增大但不要直接乘4一般乘2左右比较稳妥。8. 下一步可以往哪里走走通上面这个流程之后你已经有了一个可以工作的AI工程原型。接下来可以往几个方向深入第一个方向是换更大的模型。把SimpleClassifier换成Transformer把层数从2层加到12层看看效果能提升多少。这个过程会让你理解模型容量和数据量之间的关系。第二个方向是换更复杂的任务。从分类换成生成比如训练一个小的语言模型。生成任务需要处理自回归解码、beam search、温度采样等比分类复杂得多。第三个方向是优化推理性能。试试量化、剪枝、蒸馏看看能在保持精度的前提下把模型压缩多少。这些技术在部署到资源受限的设备时非常有用。第四个方向是搭建完整的训练流水线。把数据预处理、训练、评估、导出串起来用一个配置文件控制所有参数。这样你换数据集或者换模型的时候只需要改配置不用改代码。我个人在走完第一遍之后最大的收获不是学会了某个具体技术而是建立了一种“我知道每一步在干什么”的掌控感。这种感觉在你面对新模型、新框架的时候特别重要因为你知道哪些部分是核心哪些部分只是封装。最后分享一个小技巧每次跑实验之前先用一个极小的数据集比如100条跑一遍完整流程确保没有形状错误、设备错误、路径错误。这个过程只要几分钟但能帮你省下几个小时。我现在的习惯是任何新代码都先用小数据跑通再上全量数据。这个习惯让我少熬了很多夜。