1. 从零手搓AI工程为什么我不建议你直接调包很多人一上来就想搞个大模型应用第一反应是找API、装框架、跑通一个Demo然后觉得自己“入门AI工程”了。我刚开始也这么干过结果踩了一堆坑接口一改就崩、成本失控、延迟高得离谱、出了问题完全不知道从哪查。后来我才意识到AI工程的核心能力不是“会调包”而是理解从数据到模型再到服务的整条链路。这就是我决定从零开始手搓一遍的原因。“ai-engineering-from-scratch”这个方向说白了就是不依赖高级封装用最基础的工具把AI系统的关键环节自己实现一遍。它解决的不是“能不能跑通”的问题而是“跑通之后你能不能掌控它”的问题。适合谁看如果你已经会写Python、懂一点线性代数和概率但每次遇到模型效果不好、推理太慢、显存爆了就只能上网搜答案那这套东西就是给你准备的。我下面会按实际动手的顺序把数据管线、模型训练、推理优化、服务部署这几个环节拆开讲每个环节都给出可复现的代码思路和参数选择的理由。2. 整体设计思路为什么我要把“调包”拆成“手搓”2.1 先想清楚手搓到底练的是什么很多人对手搓有误解觉得是要自己写一个PyTorch出来。不是的。手搓的目的是让你对每一层抽象都有“掀开盖子”的能力。比如你知道model.fit()背后发生了什么吗梯度是怎么累积的学习率在哪个时刻衰减数据是怎么分批的这些细节在调包时全是黑盒但一旦你自己用NumPy实现一遍反向传播再用PyTorch对照验证你对训练过程的理解会完全不一样。我给自己定的原则是能用基础库就不用高级封装能自己写的模块就不调现成函数。但也不是什么都从零写像矩阵乘法这种底层算子直接用NumPy或者PyTorch的底层接口就行没必要自己写CUDA核。关键是理解每一层的输入输出和计算逻辑。2.2 技术选型为什么是Python NumPy PyTorch底层API选Python没什么好说的生态最全。NumPy用来做数据预处理和手写算法验证因为它足够底层你能看到每一个数组的形状变化。PyTorch我只用torch.tensor、torch.autograd和torch.nn.functional这些底层接口不用nn.Module的高级封装更不用Trainer。这样做的代价是代码量会多两三倍但好处是每一个超参数、每一次前向传播、每一次梯度更新都在你眼皮底下。有人会问为什么不直接用JAX或者TensorFlow。我的考虑是PyTorch的动态图机制对调试最友好而且它的底层API和NumPy的思维模式最接近从NumPy过渡到PyTorch几乎没有认知负担。JAX虽然快但函数式编程的风格对新手不太友好调试也麻烦。2.3 整体架构从数据到服务的四层拆分我把整个系统拆成四层每一层都可以独立测试和替换数据层负责原始数据的读取、清洗、分词、分批。这一层的关键是可复现同样的随机种子必须产生同样的批次顺序。模型层定义网络结构、损失函数、优化器。这一层的关键是可解释每一层的参数量、计算量、梯度流动都要能打印出来。训练层控制训练循环、学习率调度、梯度裁剪、模型保存。这一层的关键是可观测loss曲线、梯度范数、学习率变化都要实时记录。服务层把训练好的模型封装成API处理并发请求、批处理、超时。这一层的关键是可伸缩单机能跑多机也能扩。这四层之间的接口我全部用最朴素的Python字典和NumPy数组来传递不用任何框架特有的数据结构。这样做的目的是让每一层都可以单独拿出来测试比如我可以不启动训练直接用假数据测试服务层的吞吐量。3. 核心细节解析数据管线与模型训练的实操要点3.1 数据管线为什么你的模型效果不好八成是数据没弄对我见过太多人把精力全花在调模型结构上结果数据管线里藏着一堆bug。数据管线的第一原则是任何一步都要能单独验证。比如分词之后你要能随机抽几条出来看分批之后你要能打印出每个批次的形状和标签分布。具体怎么做我一般会写一个DataPipeline类里面每个方法只做一件事class DataPipeline: def __init__(self, raw_texts, labels, tokenizer, batch_size, seed42): self.raw_texts raw_texts self.labels labels self.tokenizer tokenizer self.batch_size batch_size self.rng np.random.default_rng(seed) def clean(self): # 去重、去空、去异常字符 cleaned [] for text in self.raw_texts: text text.strip() if len(text) 2: continue cleaned.append(text) return cleaned def tokenize(self, texts): # 这里用最简单的空格分词实际项目可以换成BPE return [self.tokenizer.encode(t) for t in texts] def batch(self, token_ids, labels): # 先打乱再按batch_size切分 indices self.rng.permutation(len(token_ids)) for i in range(0, len(indices), self.batch_size): batch_idx indices[i:iself.batch_size] yield [token_ids[j] for j in batch_idx], [labels[j] for j in batch_idx]注意几个细节随机种子要固定不然每次跑的结果都不一样没法对比实验。清洗规则要可配置不同数据集的最短长度要求不一样。分批之前一定要打乱不然模型会学到顺序信息这在很多任务里是致命的。还有一个坑padding的位置。如果你用固定长度的批次短句子后面补0那计算loss的时候一定要mask掉这些0不然模型会学着去预测padding。我一般会在batch方法里同时返回一个mask数组训练时用loss (loss * mask).sum() / mask.sum()来算真实loss。3.2 模型层手写一个Transformer的注意力机制既然是从零手搓那注意力机制肯定要自己写一遍。很多人觉得Transformer很复杂其实拆开看就是几个矩阵乘法和softmax。我用NumPy写一个最基础的单头注意力def attention(Q, K, V, maskNone): # Q, K, V的形状都是 (batch_size, seq_len, d_model) d_k Q.shape[-1] scores np.matmul(Q, K.transpose(0, 2, 1)) / np.sqrt(d_k) if mask is not None: scores np.where(mask 0, -1e9, scores) weights softmax(scores, axis-1) return np.matmul(weights, V), weights def softmax(x, axis-1): x_max np.max(x, axisaxis, keepdimsTrue) exp_x np.exp(x - x_max) return exp_x / np.sum(exp_x, axisaxis, keepdimsTrue)这里有几个关键点除以sqrt(d_k)是为了防止点积过大导致softmax梯度消失这个缩放因子不是随便选的是让方差保持在1左右。mask要在softmax之前加而且要用一个很大的负数而不是0因为softmax(0)是有值的会污染注意力分布。softmax要减去最大值这是数值稳定性的常规操作不然exp容易溢出。写完之后我会用PyTorch的torch.nn.functional.scaled_dot_product_attention对照验证确保输出一致。这一步很重要手搓的代码必须和成熟实现对齐不然你根本不知道是自己写错了还是模型本身效果不好。3.3 训练层学习率调度和梯度裁剪的实操参数训练循环看起来简单但里面的坑最多。我一般会记录四个东西训练loss、验证loss、梯度范数、学习率。这四个指标能覆盖90%的训练问题。学习率调度我用的是带warmup的余弦退火参数是这样选的warmup步数总步数的5%到10%。比如总共训练10000步warmup设500到1000步。warmup的作用是让模型在初期不要更新太猛避免梯度爆炸。最大学习率1e-4到3e-4之间。我一般从3e-4开始试如果loss震荡就降到1e-4。最小学习率最大学习率的十分之一。余弦退火到最后会降到这个值让模型在末期微调。梯度裁剪我设的是全局范数裁剪阈值1.0。具体做法是把所有参数的梯度拼成一个向量算它的L2范数如果超过1.0就按比例缩放。这个操作能防止个别批次的异常梯度把模型带偏。def clip_gradients(parameters, max_norm1.0): total_norm 0.0 for p in parameters: total_norm np.sum(p.grad ** 2) total_norm np.sqrt(total_norm) clip_coef max_norm / (total_norm 1e-6) if clip_coef 1.0: for p in parameters: p.grad * clip_coef return total_norm注意1e-6这个epsilon不能省不然total_norm为0的时候会除零。返回的total_norm要记录下来如果它一直很大说明学习率可能太高了。4. 实操过程从零搭建一个文本分类服务的完整记录4.1 环境准备与依赖安装我用的环境是Python 3.10依赖只有四个numpy、torch、flask、requests。不用transformers、不用datasets、不用accelerate。安装命令很简单pip install numpy torch flask requests有人会问不用transformers怎么加载预训练模型我的做法是自己写一个最小的模型加载器从HuggingFace的bin文件里读权重然后映射到我手写的网络结构上。这个过程很麻烦但能让你彻底搞清楚预训练模型的参数命名规则和结构。如果只是想快速验证也可以先用随机初始化的模型跑通流程再替换成预训练权重。4.2 数据准备用一个小数据集跑通全流程我用的是一个公开的中文情感分类数据集大概1万条数据正负样本各半。数据格式是每行一个JSON包含text和label两个字段。读取和清洗的代码如下import json def load_data(path): texts, labels [], [] with open(path, r, encodingutf-8) as f: for line in f: item json.loads(line) texts.append(item[text]) labels.append(item[label]) return texts, labels texts, labels load_data(sentiment.jsonl) print(f总样本数: {len(texts)}) print(f正样本比例: {sum(labels) / len(labels):.2f})打印正样本比例这一步很重要如果比例严重失衡准确率这个指标就没意义了得换F1或者AUC。我一般会先看一眼这个比例再决定用哪些评估指标。4.3 模型定义一个极简的文本分类网络我的模型结构很简单词嵌入 平均池化 全连接。没有用Transformer因为在这个数据量下简单模型反而更稳。结构如下import torch import torch.nn as nn class TextClassifier(nn.Module): def __init__(self, vocab_size, embed_dim, num_classes): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) self.fc nn.Linear(embed_dim, num_classes) def forward(self, input_ids, mask): # input_ids: (batch, seq_len) embeds self.embedding(input_ids) # (batch, seq_len, embed_dim) # 用mask做加权平均池化 mask mask.unsqueeze(-1).float() pooled (embeds * mask).sum(dim1) / mask.sum(dim1).clamp(min1e-6) logits self.fc(pooled) return logits注意padding_idx0这个参数它让padding位置的嵌入向量不参与梯度更新。池化的时候用mask加权平均而不是直接mean这样padding不会影响结果。clamp(min1e-6)是防止mask全0的时候除零。4.4 训练循环每一步都打印关键指标训练循环我写得比较啰嗦但每一步都记录了关键信息def train(model, dataloader, optimizer, scheduler, num_epochs): for epoch in range(num_epochs): model.train() total_loss 0.0 for step, (input_ids, mask, labels) in enumerate(dataloader): input_ids torch.tensor(input_ids, dtypetorch.long) mask torch.tensor(mask, dtypetorch.float) labels torch.tensor(labels, dtypetorch.long) logits model(input_ids, mask) loss nn.functional.cross_entropy(logits, labels) optimizer.zero_grad() loss.backward() # 梯度裁剪 grad_norm clip_gradients(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() total_loss loss.item() if step % 50 0: lr scheduler.get_last_lr()[0] print(fEpoch {epoch} Step {step} Loss {loss.item():.4f} fGradNorm {grad_norm:.4f} LR {lr:.6f}) avg_loss total_loss / len(dataloader) print(fEpoch {epoch} Avg Loss {avg_loss:.4f})这里有几个实操心得每50步打印一次太频繁会刷屏太稀疏会漏掉异常。梯度范数和学习率一起打印这样能看出它们之间的关联。验证集评估放在每个epoch结束不要放在训练循环里面不然会拖慢训练速度。4.5 服务层用Flask封装一个带批处理的推理接口训练完之后模型要能对外提供服务。我用Flask写了一个最简单的接口支持单条和批量推理from flask import Flask, request, jsonify import torch app Flask(__name__) model load_model(model.pt) model.eval() app.route(/predict, methods[POST]) def predict(): data request.get_json() texts data[texts] # 支持列表 if isinstance(texts, str): texts [texts] input_ids, mask tokenize_and_pad(texts) with torch.no_grad(): logits model(input_ids, mask) probs torch.softmax(logits, dim-1) preds torch.argmax(probs, dim-1) results [] for text, pred, prob in zip(texts, preds.tolist(), probs.tolist()): results.append({ text: text, label: pred, confidence: max(prob) }) return jsonify({results: results}) if __name__ __main__: app.run(host0.0.0.0, port5000)注意torch.no_grad()一定要加不然推理会构建计算图显存会爆。批处理接口比单条接口吞吐量高很多因为GPU的并行能力只有在大batch下才能发挥出来。我实测下来batch_size32的时候QPS是单条的8倍左右。5. 常见问题与排查技巧实录5.1 训练loss不下降从数据到梯度的排查顺序loss不降是最常见的问题我一般按这个顺序排查先看数据随机抽几条样本打印它们的token ids和label确认没有错位。我遇到过label和text反了的情况查了半天才发现是数据加载的时候字段名写错了。再看梯度打印每一层的梯度范数如果某一层梯度全是0说明那一层没参与计算。常见原因是mask写错了或者某一层的输入被detach了。然后看学习率如果学习率太大loss会震荡太小则下降很慢。我一般会跑一个学习率扫描从1e-5到1e-2每个跑100步看哪个loss降得最快。最后看模型结构如果以上都没问题那可能是模型容量不够或者结构有bug。我会先用一个极小的数据集比如100条过拟合一下如果连100条都过拟合不了那肯定是代码有问题。5.2 显存不够用几个立竿见影的优化手段显存不够的时候按这个优先级来优化减小batch_size最直接但会影响训练稳定性。我一般会配合梯度累积比如batch_size8累积4次等效batch_size32。用混合精度torch.cuda.amp能省一半显存速度还快。但要注意有些操作在fp16下会溢出需要用GradScaler。检查有没有不必要的张量保留比如在训练循环里把loss存到一个列表里如果loss是tensor那整个计算图都会被保留。正确做法是存loss.item()。用梯度检查点这个比较高级适合大模型。原理是不保存中间激活值反向传播时重新计算。代价是训练速度慢20%左右。5.3 推理延迟高从模型到服务的全链路优化推理延迟高先定位瓶颈在哪排查点可能原因优化手段模型前向层数太多、注意力计算量大剪枝、量化、换更小的模型数据预处理分词慢、padding太多缓存分词结果、动态padding服务框架单条推理、没有批处理加批处理、用异步框架硬件CPU推理、显存带宽不够换GPU、用TensorRT我实测下来动态padding对延迟的改善最明显。因为大部分句子的长度都远小于最大长度固定padding会浪费大量计算。动态padding就是每个batch按当前最长句子来padding能省30%到50%的计算量。5.4 常见问题速查表问题现象可能原因快速验证方法解决方案loss变成NaN学习率太大、梯度爆炸打印梯度范数降低学习率、加梯度裁剪验证loss上升过拟合对比训练和验证loss曲线加dropout、早停、数据增强预测结果全是同一类数据失衡、模型没学到打印预测分布重采样、换损失函数服务响应超时批处理太大、模型太慢打印每个请求的处理时间减小batch、加超时限制模型加载失败参数名不匹配、形状不对打印state_dict的key手动映射参数名6. 我踩过的坑和最后再分享几个小技巧第一个坑是随机种子没固定全。Python的random、NumPy的np.random、PyTorch的torch.manual_seed都要设而且DataLoader的worker_init_fn也要设不然多进程加载数据的时候顺序还是会变。我现在的做法是在训练脚本开头写一个set_seed(42)函数把所有能设的种子都设一遍。第二个坑是学习率调度器的step位置。PyTorch的CosineAnnealingLR是按epoch调的但OneCycleLR是按step调的。如果搞混了学习率曲线会完全不对。我现在的习惯是每个step都打印学习率这样一眼就能看出调度器有没有正常工作。第三个坑是模型保存和加载的不一致。训练的时候用了nn.DataParallel保存的state_dict的key会多一个module.前缀加载的时候如果不用DataParallel就会报错。解决办法是保存的时候用model.module.state_dict()或者加载的时候用OrderedDict把前缀去掉。最后分享一个小技巧在训练循环里加一个异常捕获把出错的batch的数据和中间结果保存下来。这样即使训练崩了你也能复现问题。我一般会在try块里跑训练except块里把当前batch的input_ids、mask、labels和loss都存成npy文件然后重新抛出异常。这个习惯帮我省了很多调试时间。还有一个技巧是用小数据集做快速迭代。每次改完代码先用100条数据跑10个step确认没有形状错误和NaN再上全量数据。这样能把调试周期从几小时缩短到几分钟。我现在的流程是改代码 - 小数据跑通 - 全量训练 - 验证集评估 - 服务部署每一步都有明确的检查点不会等到最后才发现问题。