从零手搓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_len=128): 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], truncation=True, padding='max_length', max_length=self.max_len, return_tensors='pt' ) return { 'input_ids': encoding['input_ids'].squeeze(), 'attention_mask': encoding['attention_mask'].squeeze(), 'label': torch.tensor(self.labels[idx], dtype=torch.long) }

这个类看起来简单，但有几个细节值得注意。padding='max_length'在训练时效率不高，因为每个batch都填充到全局最大长度。更好的做法是用DataCollator做动态填充，每个batch填充到该batch的最大长度。另外return_tensors='pt'返回的是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_size=30000) tokenizer.train(files=["corpus.txt"], trainer=trainer) 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_vectors=None): embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0) if pretrained_vectors is not None: embedding.weight.data.copy_(torch.from_numpy(pretrained_vectors)) embedding.weight.requires_grad = False # 先冻结 return embedding

这里有个坑：padding_idx=0一定要设，否则填充位置的嵌入也会参与梯度更新，影响模型效果。另外冻结嵌入层后，训练几个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_idx=0) 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(dim=1) / mask.sum(dim=1).clamp(min=1e-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_heads=4): super().__init__() self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0) self.attention = nn.MultiheadAttention(embed_dim, num_heads, batch_first=True) 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_mask=attn_mask) pooled = attn_output.mean(dim=1) logits = self.fc(pooled) return logits

这里的关键是key_padding_mask，它告诉注意力机制哪些位置是填充的，不要关注。如果不设这个，模型会把填充位置也当作有效信息，效果会大打折扣。另外batch_first=True这个参数在PyTorch 1.9以后才支持，老版本需要手动转置维度。

提示：注意力层的参数量是4 * embed_dim * embed_dim，如果embed_dim=256，单层注意力就有26万参数。在小数据集上，这足以让模型记住所有训练样本。所以加注意力之前，先确认你的数据量够不够。

5. 训练循环与优化策略：让模型真正学起来

5.1 损失函数选择：交叉熵之外还有什么

分类任务默认用交叉熵损失，但如果你遇到类别不平衡，交叉熵就不够用了。比如正样本占90%，负样本占10%，模型只要全预测正样本就能达到90%准确率，但这样的模型没有实际价值。这时候可以用带权重的交叉熵，给少数类更高的权重。

class_weights = torch.tensor([1.0, 5.0]) # 负样本权重是正样本的5倍 criterion = nn.CrossEntropyLoss(weight=class_weights)

权重的设置可以根据类别频率的倒数来定。如果正负比是9:1，那负样本权重可以设为9，正样本设为1。但也不要设得太极端，否则模型会过度偏向少数类，导致多数类准确率下降。我一般会试几个权重值，看验证集上的F1分数哪个最高。

另外，如果任务是多标签分类，交叉熵就不适用了，要用二元交叉熵（BCEWithLogitsLoss）。这个损失函数会对每个类别独立计算二分类损失，适合一个样本属于多个类别的场景。

5.2 学习率调度：为什么固定学习率往往不是最优

我一开始训练模型用固定学习率，结果要么收敛太慢，要么在最优解附近震荡。后来改用学习率预热加余弦退火，效果稳定很多。预热就是在训练初期把学习率从很小的值线性增加到设定值，避免一开始就大步长更新导致模型发散。余弦退火则是在训练后期逐渐降低学习率，让模型精细调整。

from torch.optim.lr_scheduler import CosineAnnealingWarmRestarts optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5, weight_decay=0.01) scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=10, T_mult=2)

T_0=10表示每10个epoch重启一次学习率，T_mult=2表示每次重启后周期翻倍。这个策略在训练Transformer类模型时特别有效。AdamW的weight_decay建议设在0.01到0.1之间，太小起不到正则化作用，太大又会导致欠拟合。

还有一个细节：梯度裁剪。当梯度范数超过阈值时，按比例缩小梯度，防止梯度爆炸。这个在训练RNN或深层Transformer时几乎是必须的。

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

5.3 早停与模型保存：别让过拟合偷走你的时间

训练过程中要监控验证集损失，如果连续几个epoch验证损失不再下降，就停止训练。我一般设patience=3，也就是连续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(f"Early 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, annot=True, fmt='d', xticklabels=labels, yticklabels=labels) plt.xlabel('Predicted') plt.ylabel('True') plt.show() print(classification_report(y_true, y_pred, target_names=labels))

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}, dtype=torch.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_version=14 )

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]], dtype=np.int64), 'attention_mask': np.array([[1, 1, 1, 1]], dtype=np.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_model=Response) async def predict(req: Request): inputs = tokenizer(req.text, return_tensors='pt', truncation=True, max_length=128) with torch.no_grad(): logits = model(**inputs) probs = torch.softmax(logits, dim=-1) confidence, pred = torch.max(probs, dim=-1) return Response(label=id2label[pred.item()], confidence=confidence.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(seed=42): 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会让CUDA卷积操作确定化，但可能会降低一点速度。如果对复现性要求极高，就设成True；如果更看重速度，可以设成False，但结果会有微小差异。

第五个坑是显存碎片化。长时间训练或者频繁创建销毁张量会导致显存碎片，明明总显存够用却报OOM。解决办法是定期调用torch.cuda.empty_cache()，或者用PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True环境变量让PyTorch更灵活地管理显存。这个坑在训练大模型时特别常见，有时候重启一下进程就能解决。

10. 后续可以继续深挖的方向

这套从零构建的流程跑通之后，你可以往几个方向继续深入。一个是尝试不同的模型结构，比如把注意力层换成卷积层或者循环层，对比不同结构在你数据集上的表现。另一个是引入更复杂的训练技巧，比如对抗训练、对比学习、知识蒸馏。还有一个方向是优化推理性能，试试TensorRT或者OpenVINO，看看能压到多低的延迟。

我个人最感兴趣的是数据侧的工作。模型结构再花哨，数据质量不行都是白搭。我最近在尝试用主动学习的方式挑选最有价值的样本进行标注，在标注预算有限的情况下最大化模型效果。这个方向在实际项目中非常实用，尤其是当你面对海量未标注数据时。

另外，如果你想把模型部署到移动端或者边缘设备，可以研究一下模型剪枝和神经架构搜索。这些技术能帮你找到更轻量但效果不差的模型结构。不过这些方向都需要一定的实践积累，建议先把基础流程走稳，再逐步深入。