从零手搓AI工程：不调包构建完整AI系统的实践指南

1. 从零手搓AI工程为什么我不建议你直接调包很多人一上来就想跑通一个能对话的模型或者直接拉一个开源仓库改改参数就上线。我见过太多这样的项目最后卡在环境依赖、显存溢出、推理延迟这些看似“低级”的问题上。ai-engineering-from-scratch这个标题本身就带着一种态度不依赖现成的高层封装从最底层的张量操作、数据管道、模型结构开始把AI工程链路完整地走一遍。这件事的价值不在于造一个比肩大厂的模型而在于让你真正理解一个AI系统从数据到部署中间到底发生了什么。我最初接触这个方向是因为一个很实际的需求手头有一批垂直领域的文本数据想做一个轻量级的分类和生成工具但调用云端API成本太高本地部署大模型又跑不动。于是被迫从零开始用最基础的组件搭一套能用的东西。这个过程踩了无数坑也让我对“AI工程”这四个字有了完全不同的认识。它不是一个模型文件加一个接口那么简单而是数据清洗、分词、嵌入、模型结构设计、训练循环、评估、量化、服务化这一整条链路。这篇文章适合谁如果你已经会写Python了解基本的机器学习概念但从来没有完整地从零构建过一个AI系统那这篇内容就是为你准备的。我会把每个环节的“为什么”讲清楚把参数选择的依据摆出来把踩过的坑和绕过去的经验直接告诉你。整个过程不需要昂贵的GPU集群一台带消费级显卡的机器甚至CPU都能跟着走一遍。提示本文涉及的所有代码和配置都基于常见开源工具链不涉及任何特定平台或受限服务。所有操作均在本地环境完成数据也建议使用公开数据集或自有合规数据。2. 环境搭建与依赖管理别让版本冲突毁掉你的第一个模型2.1 为什么虚拟环境不是可选项而是必选项我刚开始做的时候觉得虚拟环境麻烦直接全局装包结果不同项目对同一个库的版本要求冲突导致一个项目能跑另一个就报错。AI工程涉及的依赖特别多numpy、torch、transformers、datasets、tokenizers这些库之间版本兼容性非常敏感。比如torch2.0 和transformers4.30 以上配合没问题但如果你装了一个很老的tokenizers可能在加载预训练分词器时直接段错误。我的做法是每个项目一个conda环境或者venv并且用pip freeze导出精确版本。对于ai-engineering-from-scratch这种从零开始的项目我建议只装最核心的几个包不要一上来就pip install transformers全家桶。先装numpy和torch把张量操作跑通再逐步加其他组件。python -m venv ai_env source ai_env/bin/activate # Windows用 ai_env\Scripts\activate pip install numpy torch --index-url https://download.pytorch.org/whl/cpu如果你有NVIDIA显卡把cpu换成对应的CUDA版本比如cu121。这里有个经验不要盲目追求最新CUDA版本先查一下你的显卡驱动支持到哪个版本。我见过有人装了CUDA 12.1的torch结果驱动只支持11.8跑起来直接报错。2.2 数据管道的依赖比模型本身更容易出问题很多人把注意力全放在模型结构上结果数据加载环节一堆坑。比如用datasets库加载本地文件时默认会做缓存如果你的数据路径变了或者文件内容改了缓存不刷新你拿到的还是旧数据。我建议在开发阶段直接用pandas或原生Python读取自己写一个简单的Dataset类继承torch.utils.data.Dataset实现__len__和__getitem__。这样你完全掌控数据流出了问题也好排查。from torch.utils.data import Dataset class TextDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_len128): self.texts texts self.labels labels self.tokenizer tokenizer self.max_len max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): encoding self.tokenizer( self.texts[idx], truncationTrue, paddingmax_length, max_lengthself.max_len, return_tensorspt ) return { input_ids: encoding[input_ids].squeeze(), attention_mask: encoding[attention_mask].squeeze(), label: torch.tensor(self.labels[idx], dtypetorch.long) }这个类看起来简单但有几个细节值得注意。paddingmax_length在训练时效率不高因为每个batch都填充到全局最大长度。更好的做法是用DataCollator做动态填充每个batch填充到该batch的最大长度。另外return_tensorspt返回的是PyTorch张量如果你后面用DataLoader的collate_fn可以在这里返回列表让collator统一处理。注意分词器的max_length不要设得太大。我一开始设了512结果显存直接爆了。对于分类任务128或256通常足够具体取决于你的文本长度分布。可以先统计一下数据中90%分位数的长度以此作为参考。3. 分词器与嵌入层从字符到向量的第一道关3.1 为什么我不推荐从头训练一个分词器ai-engineering-from-scratch的精神是从零构建但分词器这一块我建议直接用现成的。原因很简单训练一个高质量的分词器需要海量文本和大量计算资源而且分词质量对最终效果影响巨大。你可以用tokenizers库加载一个预训练好的分词器比如BERT的bert-base-chinese或者多语言的xlm-roberta-base。这些分词器已经在通用语料上训练过对中文、英文都有不错的切分效果。如果你坚持要从零训练那至少要知道背后的逻辑。分词器本质上是在做子词切分常见算法有BPE、WordPiece和Unigram。BPE从字符开始不断合并频率最高的相邻对WordPiece类似但用似然概率选择合并Unigram则从大词表开始逐步裁剪。训练时你需要设置vocab_size这个值直接影响模型参数量和嵌入层大小。对于垂直领域30000左右的词表通常够用太小会导致很多词被切成单字太大则嵌入矩阵稀疏。from tokenizers import Tokenizer from tokenizers.models import BPE from tokenizers.trainers import BpeTrainer from tokenizers.pre_tokenizers import Whitespace tokenizer Tokenizer(BPE(unk_token[UNK])) tokenizer.pre_tokenizer Whitespace() trainer BpeTrainer(special_tokens[[UNK], [CLS], [SEP], [PAD], [MASK]], vocab_size30000) tokenizer.train(files[corpus.txt], trainertrainer) tokenizer.save(my_tokenizer.json)训练完之后一定要检查分词效果。我试过用一份技术文档训练分词器结果因为文档里英文缩写太多很多词被切得乱七八糟。后来加了预处理步骤把英文和数字用空格隔开效果才好起来。3.2 嵌入层初始化随机初始化还是加载预训练权重嵌入层就是把离散的token ID映射成连续向量。如果你从零训练嵌入矩阵就是随机初始化的然后通过反向传播学习。但随机初始化在数据量少的时候收敛很慢而且容易过拟合。我的做法是加载预训练的词向量比如用gensim加载Word2Vec或GloVe然后对齐到你的词表。如果词表里有预训练词向量没有的词就用随机值填充但把这些词的嵌入设为可训练其他词先冻结训练几轮后再解冻。import numpy as np import torch.nn as nn def create_embedding_layer(vocab_size, embed_dim, pretrained_vectorsNone): embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) if pretrained_vectors is not None: embedding.weight.data.copy_(torch.from_numpy(pretrained_vectors)) embedding.weight.requires_grad False # 先冻结 return embedding这里有个坑padding_idx0一定要设否则填充位置的嵌入也会参与梯度更新影响模型效果。另外冻结嵌入层后训练几个epoch再解冻通常比一直冻结或一直可训练效果更好。我实测下来在文本分类任务上这种“先冻结后解冻”的策略比随机初始化收敛快30%左右。4. 模型结构设计从简单到复杂的演进路线4.1 先跑通一个最简单的基线模型很多人一上来就想搞Transformer结果调参调到怀疑人生。我的建议是先搭一个最简单的模型比如嵌入层加平均池化加全连接层。这个模型虽然简单但能让你快速验证数据管道、训练循环和评估逻辑是否正确。如果这个基线模型都跑不通那问题一定不在模型结构上。import torch import torch.nn as nn class BaselineClassifier(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, attention_mask): embedded self.embedding(input_ids) # (batch, seq_len, embed_dim) mask attention_mask.unsqueeze(-1).float() pooled (embedded * mask).sum(dim1) / mask.sum(dim1).clamp(min1e-9) logits self.fc(pooled) return logits这个模型的核心是平均池化把序列维度压缩掉。注意这里用了attention_mask来排除填充位置的影响否则填充的0向量会拉低平均值。这个细节很多教程不讲但实际用的时候如果忘了模型效果会差很多。4.2 引入注意力机制什么时候该加什么时候不该加当你发现基线模型效果不够时下一步就是加注意力。但注意力不是万能的它增加了参数量和计算量在小数据集上反而容易过拟合。我的经验是如果训练集样本少于1万条先别急着上注意力试试把嵌入维度加大或者加几层全连接。如果数据量在几万到几十万可以加一个单头注意力或者轻量级的Transformer层。class AttentionClassifier(nn.Module): def __init__(self, vocab_size, embed_dim, num_classes, num_heads4): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) self.attention nn.MultiheadAttention(embed_dim, num_heads, batch_firstTrue) self.fc nn.Linear(embed_dim, num_classes) def forward(self, input_ids, attention_mask): embedded self.embedding(input_ids) # 把padding位置的attention mask设为负无穷 attn_mask (attention_mask 0) attn_output, _ self.attention(embedded, embedded, embedded, key_padding_maskattn_mask) pooled attn_output.mean(dim1) logits self.fc(pooled) return logits这里的关键是key_padding_mask它告诉注意力机制哪些位置是填充的不要关注。如果不设这个模型会把填充位置也当作有效信息效果会大打折扣。另外batch_firstTrue这个参数在PyTorch 1.9以后才支持老版本需要手动转置维度。提示注意力层的参数量是4 * embed_dim * embed_dim如果embed_dim256单层注意力就有26万参数。在小数据集上这足以让模型记住所有训练样本。所以加注意力之前先确认你的数据量够不够。5. 训练循环与优化策略让模型真正学起来5.1 损失函数选择交叉熵之外还有什么分类任务默认用交叉熵损失但如果你遇到类别不平衡交叉熵就不够用了。比如正样本占90%负样本占10%模型只要全预测正样本就能达到90%准确率但这样的模型没有实际价值。这时候可以用带权重的交叉熵给少数类更高的权重。class_weights torch.tensor([1.0, 5.0]) # 负样本权重是正样本的5倍 criterion nn.CrossEntropyLoss(weightclass_weights)权重的设置可以根据类别频率的倒数来定。如果正负比是9:1那负样本权重可以设为9正样本设为1。但也不要设得太极端否则模型会过度偏向少数类导致多数类准确率下降。我一般会试几个权重值看验证集上的F1分数哪个最高。另外如果任务是多标签分类交叉熵就不适用了要用二元交叉熵BCEWithLogitsLoss。这个损失函数会对每个类别独立计算二分类损失适合一个样本属于多个类别的场景。5.2 学习率调度为什么固定学习率往往不是最优我一开始训练模型用固定学习率结果要么收敛太慢要么在最优解附近震荡。后来改用学习率预热加余弦退火效果稳定很多。预热就是在训练初期把学习率从很小的值线性增加到设定值避免一开始就大步长更新导致模型发散。余弦退火则是在训练后期逐渐降低学习率让模型精细调整。from torch.optim.lr_scheduler import CosineAnnealingWarmRestarts optimizer torch.optim.AdamW(model.parameters(), lr2e-5, weight_decay0.01) scheduler CosineAnnealingWarmRestarts(optimizer, T_010, T_mult2)T_010表示每10个epoch重启一次学习率T_mult2表示每次重启后周期翻倍。这个策略在训练Transformer类模型时特别有效。AdamW的weight_decay建议设在0.01到0.1之间太小起不到正则化作用太大又会导致欠拟合。还有一个细节梯度裁剪。当梯度范数超过阈值时按比例缩小梯度防止梯度爆炸。这个在训练RNN或深层Transformer时几乎是必须的。torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)5.3 早停与模型保存别让过拟合偷走你的时间训练过程中要监控验证集损失如果连续几个epoch验证损失不再下降就停止训练。我一般设patience3也就是连续3个epoch验证损失没改善就停。同时保存验证损失最低的那个模型权重而不是最后一个epoch的权重。best_val_loss float(inf) patience_counter 0 patience 3 for epoch in range(num_epochs): train_loss train_one_epoch(model, train_loader, optimizer, criterion) val_loss evaluate(model, val_loader, criterion) if val_loss best_val_loss: best_val_loss val_loss torch.save(model.state_dict(), best_model.pt) patience_counter 0 else: patience_counter 1 if patience_counter patience: print(fEarly stopping at epoch {epoch}) break这里有个经验验证集和训练集的损失曲线要一起看。如果训练损失持续下降但验证损失开始上升那就是过拟合了早停能帮你省下大量计算时间。如果两者都下不去那是欠拟合需要加大模型容量或降低正则化强度。6. 模型评估与错误分析准确率之外你该看什么6.1 混淆矩阵比准确率更能说明问题准确率在类别不平衡时具有欺骗性。我习惯在验证集上画混淆矩阵看看模型到底在哪些类别上犯错。比如一个三分类任务如果模型把A类全预测成B类准确率可能还有60%但实际完全不可用。混淆矩阵能让你一眼看出问题所在。from sklearn.metrics import confusion_matrix, classification_report import seaborn as sns import matplotlib.pyplot as plt def plot_confusion_matrix(y_true, y_pred, labels): cm confusion_matrix(y_true, y_pred) sns.heatmap(cm, annotTrue, fmtd, xticklabelslabels, yticklabelslabels) plt.xlabel(Predicted) plt.ylabel(True) plt.show() print(classification_report(y_true, y_pred, target_nameslabels))classification_report会输出每个类别的精确率、召回率和F1分数。精确率高但召回率低说明模型很保守只预测最有把握的样本召回率高但精确率低说明模型很激进把很多不确定的样本也预测成了正类。根据业务需求你可以调整分类阈值来平衡这两个指标。6.2 错误样本分析从个例中找到系统性问题我习惯把预测错误的样本单独拿出来看按错误类型分类。比如文本分类任务中错误可能集中在短文本、含特殊符号的文本、或者类别边界模糊的文本上。找到这些模式后可以针对性地补充训练数据或调整预处理逻辑。有一次我发现模型总是把带引号的句子分错原因是分词器把引号单独切成了一个token而这个token在训练数据中出现频率很低嵌入向量没学好。后来我在预处理阶段把中文引号统一替换成英文引号问题就解决了。这种问题不看错误样本是永远发现不了的。注意错误分析不要只看一两个样本就下结论至少要看几十个按类别分组统计。另外测试集上的错误分析要在模型完全训练好之后做不要用验证集反复调参后再看测试集那样测试集就失去了客观性。7. 模型压缩与推理优化让模型真正能跑起来7.1 量化用精度换速度的实用手段训练好的模型如果参数量大推理时显存占用高、延迟大。量化是把浮点参数转换成低精度整数比如从FP32转成INT8模型大小直接缩小4倍推理速度也能提升2到3倍。PyTorch提供了动态量化和静态量化两种方式。动态量化最简单一行代码就能搞定。import torch.quantization quantized_model torch.quantization.quantize_dynamic( model, {nn.Linear, nn.Embedding}, dtypetorch.qint8 ) torch.save(quantized_model.state_dict(), quantized_model.pt)动态量化对Linear和Embedding层效果最好因为这两类层的计算主要是矩阵乘法量化后精度损失可控。但注意力层的MultiheadAttention量化后效果可能下降明显因为注意力分数对精度比较敏感。我的做法是只量化嵌入层和全连接层注意力层保持FP32。7.2 ONNX导出与推理跨平台部署的通用方案如果你需要把模型部署到非Python环境或者想用TensorRT加速ONNX是个不错的中间格式。PyTorch支持直接导出ONNX。dummy_input torch.randint(0, vocab_size, (1, 128)) dummy_mask torch.ones(1, 128) torch.onnx.export( model, (dummy_input, dummy_mask), model.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{ input_ids: {0: batch, 1: sequence}, attention_mask: {0: batch, 1: sequence}, logits: {0: batch} }, opset_version14 )dynamic_axes很重要它允许模型接受可变长度的输入。如果不设这个导出的ONNX模型只能处理固定长度的序列实际用起来很不方便。opset_version建议用14或更高对Transformer类模型的支持更好。导出后可以用onnxruntime加载推理速度通常比原生PyTorch快20%到50%具体取决于模型结构和硬件。import onnxruntime as ort import numpy as np session ort.InferenceSession(model.onnx) inputs { input_ids: np.array([[1, 2, 3, 4]], dtypenp.int64), attention_mask: np.array([[1, 1, 1, 1]], dtypenp.int64) } logits session.run([logits], inputs)[0]8. 服务化与接口设计从脚本到可用服务8.1 用FastAPI包装模型推理模型训练好之后最终要对外提供服务。最简单的方式是用FastAPI写一个HTTP接口。FastAPI自带异步支持和自动文档开发效率很高。from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model load_model() tokenizer load_tokenizer() class Request(BaseModel): text: str class Response(BaseModel): label: str confidence: float app.post(/predict, response_modelResponse) async def predict(req: Request): inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length128) with torch.no_grad(): logits model(**inputs) probs torch.softmax(logits, dim-1) confidence, pred torch.max(probs, dim-1) return Response(labelid2label[pred.item()], confidenceconfidence.item())这里有几个优化点。第一模型加载要在服务启动时完成不要每次请求都加载。第二推理时用torch.no_grad()关闭梯度计算减少内存占用。第三如果并发量高可以用batch推理把多个请求攒成一个batch一起算吞吐量能提升好几倍。8.2 批处理与超时控制生产环境必须考虑的事单条推理延迟可能只有几十毫秒但QPS一高GPU利用率上不去。批处理是解决这个问题的关键。你可以用一个队列缓存请求每隔几毫秒或者队列长度达到阈值时取出所有请求组成一个batch推理。import asyncio from collections import deque request_queue deque() batch_size 8 batch_timeout 0.01 # 10ms async def batch_worker(): while True: if len(request_queue) batch_size: batch [request_queue.popleft() for _ in range(batch_size)] await process_batch(batch) elif request_queue: await asyncio.sleep(batch_timeout) batch list(request_queue) request_queue.clear() await process_batch(batch) else: await asyncio.sleep(0.001)超时控制也很重要。如果某个请求在队列里等了太久还没被处理应该直接返回超时错误而不是让客户端一直等。我一般设总超时时间为200ms到500ms具体取决于业务对延迟的容忍度。提示生产环境一定要加监控记录每个请求的延迟、batch大小、GPU利用率。这些指标能帮你判断是否需要扩容或者调整batch参数。我见过一个服务因为batch设得太大导致单次推理时间过长反而拖慢了整体吞吐。9. 我踩过的那些坑和绕过去的经验第一个坑是数据泄露。我在划分训练集和验证集时用了随机划分结果同一个用户的多条记录被分到了不同集合导致验证集准确率虚高。后来改成按用户ID划分确保同一个用户的数据只出现在一个集合里验证结果才真实可信。这个坑在推荐系统和文本分类中特别常见一定要检查你的数据划分逻辑。第二个坑是分词器的特殊token处理。BERT分词器会自动在句首加[CLS]句尾加[SEP]但如果你自己训练的分词器没有这些特殊token模型就学不到句子边界信息。我一开始没注意模型效果一直上不去后来加上特殊token并确保分词器正确添加效果立刻提升了一截。第三个坑是学习率设得太大。我一开始用1e-3的学习率训练Transformer结果损失直接变成NaN。后来降到2e-5才稳定下来。Transformer类模型对学习率非常敏感通常需要比CNN小一到两个数量级。如果你不确定从1e-5开始试用学习率预热慢慢加上去。第四个坑是忘了设随机种子。深度学习训练有很多随机性数据打乱、参数初始化、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 Falsecudnn.deterministicTrue会让CUDA卷积操作确定化但可能会降低一点速度。如果对复现性要求极高就设成True如果更看重速度可以设成False但结果会有微小差异。第五个坑是显存碎片化。长时间训练或者频繁创建销毁张量会导致显存碎片明明总显存够用却报OOM。解决办法是定期调用torch.cuda.empty_cache()或者用PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True环境变量让PyTorch更灵活地管理显存。这个坑在训练大模型时特别常见有时候重启一下进程就能解决。10. 后续可以继续深挖的方向这套从零构建的流程跑通之后你可以往几个方向继续深入。一个是尝试不同的模型结构比如把注意力层换成卷积层或者循环层对比不同结构在你数据集上的表现。另一个是引入更复杂的训练技巧比如对抗训练、对比学习、知识蒸馏。还有一个方向是优化推理性能试试TensorRT或者OpenVINO看看能压到多低的延迟。我个人最感兴趣的是数据侧的工作。模型结构再花哨数据质量不行都是白搭。我最近在尝试用主动学习的方式挑选最有价值的样本进行标注在标注预算有限的情况下最大化模型效果。这个方向在实际项目中非常实用尤其是当你面对海量未标注数据时。另外如果你想把模型部署到移动端或者边缘设备可以研究一下模型剪枝和神经架构搜索。这些技术能帮你找到更轻量但效果不差的模型结构。不过这些方向都需要一定的实践积累建议先把基础流程走稳再逐步深入。