从零搭建AI工程全链路：文本分类模型训练与部署实战

看到 ai-engineering-from-scratch 这个标题我第一反应是又一个 from-scratch 仓库。网上这类东西太多了大多是把教程 demo 抄一遍真正能让人从零把 AI 工程全链路跑通、跑懂的内容反而很少。所以这篇文章不打算讲“AI 很酷、未来已来”这种正确的废话我想用一套真实可复现的文本分类项目把从原始数据到线上服务的每个环节按实际动手的顺序一步步拆给你看。这套东西适合谁准备转 AI 开发但只会调库的工程师、刚入门想补全“模型之外”那部分知识的同学以及被各种 AutoML 和云平台一键训练宠坏、想找回底层手感的人。你不需要 GPU8G 内存的普通笔记本就能跑完全程。1. 整体设计与思路拆解1.1 “从零”到底零到哪一步先解决一个很实际的问题from scratch 的“零”在 AI 工程里并没有统一标准。同样是“从零做”有人是打开 Jupyter Notebook 调库有人是把 Keras 换成 PyTorch还有人连反向传播都要自己写。我整理了一张分级表方便你对照定位层级做法典型工具适用场景L0直接调现成 API云平台 NLP 接口快速验证想法L1全流程 AutoML云平台自动训练不想碰数据之外的细节L2高级封装框架Keras、HF Trainer标准任务快速迭代L3基础库手写PyTorch transformers可定制、懂原理L4连框架都自己写numpy 手写反向传播纯学习目的这篇文章定位在 L3 偏 L2.5pyTorch 和 HuggingFace 的 tokenizer 这类基础设施直接用但训练循环、评估逻辑、模型导出、服务化部署全部自己写。这么定位有三个理由。第一是可控性。用高级框架训练你只能看到框架想让你看到的日志数据怎么 shuffle、梯度怎么裁剪、学习率怎么变化都要钻进源码里才知道。自己写训练循环每一步都在你手里出了问题能直接定位。第二是调试可见性。线上预测出错时你需要快速定位是数据清洗的问题、是模型的问题、还是服务代码的问题。如果整个链路都是自己搭的排查路径会非常短。第三是可扩展性。手写的代码没有框架魔法后面换成 BERT、做多任务学习、加对抗训练你都能清楚地知道改哪里。1.2 为什么选“工单分类”当主线任务选主线任务的标准很简单数据容易拿到、指标清晰、能覆盖全链路所有环节。“工单分类”完美符合。我这次做的是在线客服系统的用户工单自动分类一共 6 个类别账户问题、支付问题、物流问题、退换货、技术故障、投诉建议。这个任务的复现成本极低你甚至可以用自己公司客服后台的历史工单或者从公开数据集里找类似的中文文本。分类边界也足够清晰——它不像情感分析那样主观不像命名实体识别那样需要复杂标注拿来做从零到一的全流程演示非常合适。更重要的是这个任务是有迁移价值的。工单分类做通之后情感分析、意图识别、内容审核本质上都是“把文本映射到固定类别集合”把分类头换一下数据组织方式变一下整条管线可以直接平移。1.3 系统分层架构整个系统我划分为六个层次每一层解决一类问题数据层原始工单 CSV 的清洗、脱敏、标签映射特征层中文分词、词表构建、序列 padding 与 batch 组织模型层TextCNN 网络结构设计训练层训练循环、梯度裁剪、学习率调度、早停评估层准确率、宏平均 F1、混淆矩阵与错误分析服务层模型导出、FastAPI 接口、并发推理这个顺序就是后面几章的展开顺序。很多初学者把注意力全放在模型层但实际项目里数据层和服务层占的工作量往往超过一半。这也是我写这篇文章最想传达的一件事AI 工程不等于训练模型。2. 工程前置准备环境、依赖与随机性控制2.1 环境与依赖清单我是在 Ubuntu 22.04 上跑的Python 3.10。依赖清单非常克制全是基础组件torch2.0.0 transformers4.30.0 pandas2.0.0 scikit-learn1.2.0 fastapi0.100.0 uvicorn0.23.0 onnxruntime1.15.0 tqdm4.65.0 jieba0.42.1这里有个容易被误解的点transformers我用它只是为了加载 BERT 的中文分词器并不是用它来封装训练。你完全可以用jieba自建词表做分词但预训练分词器的好处是对中文的支持更成熟而且后面你要切换到 BERT 时tokenizer 不用换。基础设施复用核心逻辑手写这是 L3 的定位。装环境建议用 venv 或者 conda不要图省事直接pip install到全局。这步不做好后面装 onnxruntime 时版本冲突哭都来不及。我的做法是python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt2.2 随机性控制复现的第一要素这是我踩过最深的坑之一。第一次跑实验时结果还挺好第二天换台机器复现F1 直接从 0.87 掉到 0.83找了好久才发现是随机种子的问题。神经网络里有三个随机源模型初始化、DataLoader 的 shuffle、dropout 的随机失活。如果不固定种子每次训练结果都会有波动你根本没法判断一个改动到底是“真改善了模型”还是“运气好”。我写了一个通用的 seed 函数训练前调用一次import torch import numpy as np import random def seed_everything(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调用之后相同代码、相同数据、相同 seed 的结果就是可复现的。这篇文章后面所有实验数据都是在seed42下产出的。3. 数据管线把凌乱文本变成可训练样本3.1 原始数据长什么样先看真实场景里工单长什么样。我拿到的原始 CSV 有四个字段工单 ID、提交时间、工单内容、人工标注的类别。后面训练只用文本和类别但时间字段留着因为划分数据集时要用它排查时间泄漏问题。原始文本非常脏随便抽两条【求助】你们这个退款流程也太反人类了吧订单号 20230815001申请退款点了三天都没反应 您好我收到了pf230801234打开发现屏幕有一条线可以换货吗发票需要一起寄回吗这类文本至少有四个问题HTML 残留、全半角混用、私密信息订单号、手机号、口语化语气词。清洗不是玄学每一条规则的背后都是真实数据的分布特征。3.2 文本清洗三步我的清洗管线是三步走每一步都是正则表达式能解决的小事但组合起来效果非常明显import re def clean_text(text: str) - str: # 第一步去 HTML 实体和标签 text re.sub(r[^], , text) text re.sub(r(amp|lt|gt|nbsp);, , text) # 第二步统一全角转半角 text text.replace(, ,).replace(。, .).replace(, ?) text text.replace(, !).replace(, :).replace(, ().replace(, )) # 第三步脱敏订单号/手机号换掉 text re.sub(r订单号?s?[:]?\s*[A-Za-z0-9], 订单号, text) text re.sub(r1[3-9]\d{9}, 手机号, text) return text.strip()注意第三步脱敏不能直接把号码删掉而是替换成“订单号”“手机号”这种占位词。因为“订单号”这个词本身对支付类工单有很强的区分信号删掉反而损失信息。这个细节是我在错误分析时发现的。清洗后先统计每条文本的长度分布我这次 12000 条样本95% 的文本在 50 到 250 字之间。这个分布决定了后面的模型参数固定最大长度 200超出截断不足 padding。3.3 标签映射与类别分布工单类别在 CSV 里是中文我需要映射成整数索引。这里有个工程细节类别个数要写死还是动态算我的建议是动态算从数据里取类别集合排序后生成映射。这样以后新增类别代码不用改。labels sorted(data[label].unique()) label2id {label: i for i, label in enumerate(labels)} id2label {i: label for label, i in label2id.items()}然后统计分布结果很有意思类别样本数占比账户问题280023.3%支付问题190015.8%物流问题210017.5%退换货250020.8%技术故障130010.8%投诉建议140011.7%整体差距不算极端但“技术故障”最低会拉低 macro-F1。这个分布我会在第五章回到——它直接影响评估指标的选择。3.4 数据集划分分层采样与时间泄漏划分数据集是初学者最容易翻车的地方。直接train_test_split(data, test_size0.2)是不行的因为原始数据是按时间排列的正样本和负样本在时间线上不均衡直接随机切会把时间顺序打乱导致模型隐式学习时间分布。我做了两层处理。第一层是分层采样保证训练集和验证集的类别比例一致from sklearn.model_selection import train_test_split train_df, val_df train_test_split( data, test_size0.2, stratifydata[label_id], random_state42 )第二层是时间泄漏检查。哪怕分了层也不能保证时间上没有重叠。正确的做法是画出训练集和验证集的时间分布图确认两者在时间轴上是交错的。如果验证集全部来自 10 月以后而训练集都是 10 月以前那模型记住的可能是时间规律而不是文本规律。这次的数据时间跨度六个月我检查后确认没有这个问题。3.5 词表构建与 DataLoader有了清洗后的文本接下来是分词和词表构建。中文不能按空格分我用jieba.cut然后统计词频取前 30000 个词作为词表from collections import Counter import jieba vocab_counter Counter() for text in train_df[clean_text]: vocab_counter.update(jieba.cut(text)) vocab [PAD, UNK] [w for w, _ in vocab_counter.most_common(30000)] word2idx {w: i for i, w in enumerate(vocab)}词表构建只用训练集不用验证集这是基本规则。否则验证集里出现生僻词模型就能“作弊”。然后写 Dataset 和 DataLoader。我习惯把 tokenization 和 padding 放在自定义的 collate_fn 里而不是写死在 Dataset 的__getitem__这样 batch 处理更高效import torch from torch.utils.data import Dataset class TextDataset(Dataset): def __init__(self, texts, labels, word2idx, max_len200): self.texts texts self.labels labels self.word2idx word2idx self.max_len max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): tokens jieba.cut(self.texts[idx]) token_ids [self.word2idx.get(w, self.word2idx[UNK]) for w in tokens][:self.max_len] return token_ids, self.labels[idx] def collate_fn(batch): token_ids, labels zip(*batch) lengths [len(x) for x in token_ids] max_len max(lengths) padded torch.zeros(len(batch), max_len, dtypetorch.long) for i, seq in enumerate(token_ids): padded[i, :len(seq)] torch.tensor(seq, dtypetorch.long) return padded, torch.tensor(labels, dtypetorch.long)DataLoader 的参数有几个细节。shuffleTrue只对训练集开验证集必须shuffleFalse。num_workers在 CPU 机器上设 2 就够设太高反而慢。pin_memoryTrue在 GPU 场景有用CPU 场景无所谓但保留也无妨。4. 模型设计与训练循环手写 TextCNN 的 PyTorch 实现4.1 TextCNN 原理与模型代码TextCNN 的核心思想用一句话说用不同尺寸的卷积核在文本序列上滑动捕捉 n-gram 级别的局部特征再通过最大池化提取每个特征最强的信号。生活化类比一个人用不同大小的放大镜扫文本小放大镜看到词和词的组合大放大镜看到短语级别的模式。每一个卷积核就是一把放大镜。模型代码就这么点东西import torch.nn as nn class TextCNN(nn.Module): def __init__(self, vocab_size, embed_size128, num_filters128, kernel_sizes[2, 3, 4], num_classes6): super().__init__() self.embedding nn.Embedding(vocab_size, embed_size, padding_idx0) self.convs nn.ModuleList([ nn.Conv1d(embed_size, num_filters, k, paddingk // 2) for k in kernel_sizes ]) self.dropout nn.Dropout(0.5) self.fc nn.Linear(num_filters * len(kernel_sizes), num_classes) def forward(self, x): # x: (batch, seq_len) emb self.embedding(x) # (batch, seq_len, embed_size) emb emb.transpose(1, 2) # (batch, embed_size, seq_len) pooled [] for conv in self.convs: c torch.relu(conv(emb)) # (batch, num_filters, seq_len) p torch.max_pool1d(c, c.size(2)).squeeze(2) pooled.append(p) out torch.cat(pooled, dim1) out self.dropout(out) return self.fc(out)padding_idx0很重要它告诉 Embedding 层ID 为 0 的 PAD token 在反向传播时梯度不更新保持全零向量避免 padding 影响特征提取。为什么选 TextCNN 而不是 LSTM 或者 Transformer第一训练快CPU 上 10 轮 10 分钟跑完第二文本分类任务里 n-gram 特征足以覆盖大部分模式“退款”“物流”“发票”这些关键词本身就是很强的信号第三模型结构简单任何一个 bug 都藏不住。LSTM 在长程依赖上有优势但工单文本普遍短这个优势不明显。Transformer 效果好但那是在大数据量 大模型的前提下从零手写复杂度会翻好几倍。4.2 训练循环与超参选择手写训练循环是这个项目最核心的部分。你可以清楚地看到每一步前向、计算 loss、反向、裁剪梯度、更新参数、调度学习率。import torch from tqdm import tqdm def train_one_epoch(model, dataloader, optimizer, criterion, clip1.0): model.train() total_loss, total_correct, total 0, 0, 0 for batch_texts, batch_labels in tqdm(dataloader): optimizer.zero_grad() logits model(batch_texts) loss criterion(logits, batch_labels) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), clip) optimizer.step() total_loss loss.item() * len(batch_labels) preds logits.argmax(dim1) total_correct (preds batch_labels).sum().item() total len(batch_labels) return total_loss / total, total_correct / total超参选择不是拍脑袋每一个都能讲出理由embed_size128这是性价比很高的区间太小特征表达不足太大容易过拟合且训练变慢。kernel_sizes[2, 3, 4]对应中文里的二元、三元、四元词组。工单里“退款失败”“物流太慢”“屏幕裂了”刚好都是这个长度。num_filters128每个卷积核尺寸 128 个滤波器三个尺寸拼接后得到 384 维特征分类头容量足够。lr1e-3Adam 默认学习率就是 1e-3配合梯度裁剪不会炸。batch_size64内存有限时 64 是一个稳定的数值。太小梯度噪声大太大每轮迭代次数少、收敛不稳。max_epochs10配合早停实际训练在第 6 轮左右就会停。训练主体就是一个循环早停逻辑放在验证之后best_f1, bad_epochs 0, 0 for epoch in range(max_epochs): train_loss, train_acc train_one_epoch(...) val_loss, val_f1 evaluate(model, val_dataloader, criterion) if val_f1 best_f1: best_f1 val_f1 torch.save(model.state_dict(), best_model.pt) bad_epochs 0 else: bad_epochs 1 if bad_epochs 3: print(fEarly stop at epoch {epoch}) break print(fepoch {epoch}: train_loss{train_loss:.4f}, ftrain_acc{train_acc:.4f}, val_loss{val_loss:.4f}, val_f1{val_f1:.4f})4.3 过拟合信号与实战调优我这次训练的 loss 曲线非常典型epoch 1: train_loss0.9821, val_loss0.8753 epoch 2: train_loss0.4523, val_loss0.3871 epoch 3: train_loss0.2537, val_loss0.2152 epoch 4: train_loss0.1785, val_loss0.1904 epoch 5: train_loss0.1243, val_loss0.2345 -- 过拟合信号出现第三轮之前训练损失和验证损失同步下降这是正常拟合。从第四轮开始训练损失还在降但验证损失回升说明模型开始把训练集里的噪声当成规律记住。如果继续硬训第六轮以后验证 F1 反而比第四轮低。我的处理组合拳是dropout0.5加早停。Dropout 在训练时随机屏蔽一半神经元迫使模型不依赖单一特征路径早停保证训练在最优点附近停住。此外在文本层面可以做同义词替换增强比如把“快递”替换成“物流”把“退款”替换成“退钱”但这次任务早停已经够用没有加。5. 评估体系与迭代闭环不只盯准确率5.1 指标选择准确率的陷阱训练结束后我同时输出准确率和宏平均 F1。为什么不用准确率看这个场景如果全部预测成“账户问题”准确率是 23.3%看起来像“有效模型”但实际一点用没有。准确率只适合类别完全均衡的时候而真实工单永远是不均衡的。我用的评估代码from sklearn.metrics import classification_report, f1_score def evaluate(model, dataloader): model.eval() all_preds, all_labels [], [] with torch.no_grad(): for batch_texts, batch_labels in dataloader: logits model(batch_texts) preds logits.argmax(dim1) all_preds.extend(preds.tolist()) all_labels.extend(batch_labels.tolist()) return f1_score(all_labels, all_preds, averagemacro)宏平均 F1 对每个类别一视同仁不会因为多数类表现好就掩盖少数类的问题。最终结果类别精确率召回率F1账户问题0.910.900.91支付问题0.840.820.83物流问题0.880.870.88退换货0.920.910.92技术故障0.760.740.75投诉建议0.800.830.82宏平均0.850.850.85技术故障的 F1 明显偏低这和它的样本量最少有关。但如果只看准确率 0.89这个问题会被完全掩盖。这就是指标选择的价值。5.2 错误分析与人工复盘评估不是看两个数字结束错误分析才是提升闭环的入口。我从混淆矩阵里抽出 20 条预测错误的样本一条条读发现三大类问题第一类是标注噪声。工单本身被客服标错类别原标签是“技术故障”但文本明明在骂物流。这种错不在模型在数据质量。处理方式是挑出来修正或剔除。第二类是类别边界模糊。“收到商品有破损”到底算物流问题还是退换货模型和人工标注者的判断都不稳定。这类冲突是常态解决办法是定义规则破损发生在运输阶段算物流签收后的质量问题算退换货。规则写进标注手册重新清洗部分数据后F1 整体升了 1.2 个点。第三类是特征缺失。有些技术故障工单描述得极其简短“蓝屏了”“闪退”“打不开”上下文太少模型确实很难学。这类工单在真实场景里要靠追问但数据里没有。我的处理是暂时接受现状等后续接入多轮对话。5.3 实验记录与闭环迭代AI 工程迭代是常态每次改数据、改参数都要有记录。我一开始用备注写在文件名里后来发现根本没法追溯。后面我用一个 CSV 当实验台账experiment_log pd.DataFrame(columns[time, lr, batch_size, seed, data_version, model, macro_f1]) # 每跑完一次实验就加一行 experiment_log.to_csv(experiment_log.csv, indexFalse)有条件的可以用 MLflow 这类实验追踪工具但小项目一个 CSV 足够。关键是把数据版本、代码版本、超参和结果对应起来。我这次真正有价值的涨点来自“修正标注噪声”和“清洗规则优化”而不是改模型结构。这个经验我想强调一万遍。6. 服务化部署把模型包成一个 HTTP 接口6.1 模型导出从 PyTorch 到 ONNX模型训练完直接保存state_dict可以用但生产环境里我建议导出成 ONNX。原因有三个ONNX 推理不依赖 PyTorch 运行时部署环境更轻ONNX 支持量化CPU 推理能提速ONNX 是中间格式以后要切到其他推理引擎不用重新训练。导出代码import torch.onnx model.eval() dummy_input torch.zeros(1, 200, dtypetorch.long) torch.onnx.export( model, dummy_input, textcnn.onnx, input_names[input_ids], output_names[logits], dynamic_axes{input_ids: {0: batch}, logits: {0: batch}}, opset_version17 )dynamic_axes是必选项不设置的话 ONNX 会把 batch 维度固定成 1生产环境一次只能推理一条文本。设置之后可以自由传入不同 batch 大小。6.2 FastAPI 接口与模型生命周期管理服务端代码用 FastAPI逻辑很直接接收文本、走清洗和分词、转成 token IDs、模型推理、返回类别和置信度。from fastapi import FastAPI from pydantic import BaseModel import onnxruntime as ort import numpy as np app FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str confidence: float app.on_event(startup) def load_model(): global session, word2idx session ort.InferenceSession(textcnn.onnx, providers[CPUExecutionProvider]) word2idx build_word2idx(./data/train.csv) app.post(/predict) def predict(req: PredictRequest): cleaned clean_text(req.text) token_ids [word2idx.get(w, word2idx[UNK]) for w in jieba.cut(cleaned)][:200] if len(token_ids) 1: return PredictResponse(label未知, confidence0.0) token_ids np.pad(token_ids, (0, 200 - len(token_ids)), constant_values0) logits session.run(None, {input_ids: np.array([token_ids], dtypenp.int64)})[0] probs softmax(logits[0]) idx int(np.argmax(probs)) return PredictResponse(labelid2label[idx], confidencefloat(probs[idx]))几个关键点。模型加载要放在app.on_event(startup)里用全局变量保存千万不要在请求函数内部重复加载否则每个请求都重建一次 session延迟会暴涨。clean_text和word2idx这些函数和训练时保持一致数据和推理之间最怕的就是“训练时清洗一套、推理时另一套”。启动服务uvicorn main:app --host 0.0.0.0 --port 8000 --workers 46.3 压测与性能优化我用 Python 写了个简单压测脚本1000 条混合请求单 worker 条件下平均延迟 40msQPS 约 25。4 个 worker 时 QPS 能到 90 左右。对于客服工单分类这种异步场景完全够用。如果还想优化方向有两个第一是 ONNX 动态量化把权重从 FP32 降到 INT8延迟能降 30% 左右但 F1 可能掉 0.5 到 1 个点需要评估第二是把 pair 接口改成 batch 接口一次传多一条文本减少请求往返的开销。这些优化都要以压测数据为准不要为了优化而优化。7. 常见问题与排查技巧实录7.1 高频问题速查表这条表是我两次完整跑通这个项目后整理出来的高频问题。新手踩坑率几乎是百分之百建议直接存下来现象可能原因排查方法训练 loss 变成 NaN学习率过大或梯度爆炸调小学习率加梯度裁剪验证 F1 接近 0.5标签泄露或数据划分错误检查是否把验证集混进训练预测结果全是某一个类类别不平衡但没做任何处理用分层划分考虑类别权重加载模型后结果与训练时差异大忘了model.eval()Dropout 未关闭推理前必须调 eval 模式中文乱码CSV 编码问题统一用 utf-8-sig 读写推理延迟高未导出 ONNX或模型重复加载导出 ONNX检查 session 复用训练太慢文本里夹了大量无意义长文本设 max_len 截断清理超长文本7.2 几个容易翻车的细节第一固定 seed 之后torch.backends.cudnn.deterministic要单独写。只用manual_seed并不能保证 CuDNN 的确定性尤其是在 CNN 模型上不设置的话同一份代码跑两次结果还是有微小的差异。第二保存模型和加载模型的严格模式。如果模型定义里有 Embedding加载时load_state_dict一定要用strictTrue它默认就是 True但要注意当模型结构改了比如 kernel_sizes 从[2,3,4]改成[2,3]旧的权重文件里多出来的 key 会让加载直接报错。这不是坏事反而能提醒你结构不一致。第三推理时输入数据要过一遍和训练时一模一样的清洗函数。我见过太多人推理时直接对原始文本做 tokenize结果因为全角混淆导致线上效果远低于训练时的评估指标。把清洗函数做成共享模块训练和推理从同一个函数 import这个是必须的工程习惯。第四文本长度截断要记得看分布。不要拍脑袋定一个最大长度而是先统计训练集长度分布。工单文本 95% 在 200 字以内那就定 200如果你做的是长文档任务200 可能会截掉一半信息评估数据会非常难看。第五CPU 推理时不要忽略 batch 带来的提升。单个请求一个 batch吞吐量和多请求 batch 化差好几倍。FastAPI 的接口如果并发量预估比较高提前设计成可以接受list[str]输入服务端自动做 batch再拆开返回。我后来就是这么改的QPS 提升非常明显。最后再说几句这套流程走完之后我对“AI 工程”这四个字的理解变了很多。训练模型只是其中一个环节数据清洗、指标设计、服务化部署每个环节都有大量坑。真正从零做一遍收获最大的不是模型精度多高而是你对整个系统有了完整的手感——每个环节出了什么问题你都能在脑子里快速定位。如果你时间有限我只建议你可以把全文的七个章节都照着跑一遍。如果只能挑两件事来做我会选手写训练循环和 ONNX 导出。前者让你跟黑盒框架彻底解绑后者让你亲手把一个训练好的模型变成真正可用的服务。做完这两件事再去用 Trainer 这类工具心态会完全不一样——你知道它替你做了什么也知道它没替你做什么。这套代码再往后扩展的方向也清晰把输入从单条文本换成对话历史模型从 TextCNN 换成 BERT评估体系加一个线上 AB 测试服务层加日志链路追踪。动任何一个地方你都知道它在整个系统里的位置这就是 from scratch 带来的底气。