ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI工程从零实战:客服工单自动分类系统全流程

AI工程从零实战:客服工单自动分类系统全流程 如果你打开招聘网站搜“AI工程师”你会发现大部分岗位描述长得像同一个模板熟悉PyTorch、会训模型、懂部署、有落地经验。但真上线做项目时很多人卡住的点反而不是那些花哨的模型结构而是环境怎么搭、数据怎么洗、训练挂了怎么查、模型怎么接进业务系统。这些东西在论文里看不到在开源仓库里往往也是零散的。这篇“ai-engineering-from-scratch”想写的就是一条把AI工程完整走一遍的路径从零开始不依赖任何现成的中台和封装好的低代码平台纯手动把数据、训练、部署、监控整条链路跑通。我从自己的实际项目出发用一个“客服工单自动分类”的小系统当主线把每一步的关键决策、踩过的坑、可复用的代码片段都写清楚。适合两类人看一是想独立做一个AI项目的开发者二是已经在用框架但没搞明白底层到底在发生什么的同学。这里面没有玄学都是能直接落地的东西。1. AI工程到底在“工程”什么1.1 从“会调API”到“能做工程”之间隔了什么先说一个常见误区。会调用现成API和真正做一个AI工程是两码事。用API是消费别人封装好的能力你只需要组织输入输出做AI工程则是要把一个模型从数据处理、训练调优、模型部署到服务监控完整地建立和维护起来。后者包含太多“看不见的活”。举个例子。我接到过不少类似“帮我做个文本分类”的需求听起来很简单但真正动手会撞上一连串问题类别分布严重不均衡、数据里有大量脏文本、模型预测偶尔出错但没人知道为什么、上线后推理延迟忽高忽低、模型更新后老接口出兼容问题。每一个单独拿出来都不致命串在一起就足以把项目拖垮。做AI工程和写普通后端代码的思维差异也很大。传统软件开发是确定性逻辑输入合法输出就可预期。模型不是这样的它的输出是概率分布有置信度、有错误率、有退化风险。你写的不是“一定会做什么”的逻辑代码而是“大概率会怎样”的统计系统。这要求你在工程上额外补很多保障数据校验、版本管理、回滚机制、监控告警缺一样生产环境就会给你颜色看。1.2 从零到一的技术全景图做AI工程技术栈是分层的从下往上分别是基础设施层、数据层、模型层、服务层、监控层。很多人一上来就盯着模型层疯狂换模型调参数结果发现项目还是推不动因为他忽略了下层的基建。以我做的客服工单分类为例全景是这样的层涉及内容我用的具体方案基础设施GPU、容器、环境管理Linux Docker NVIDIA驱动数据层采集、清洗、标注、管线Python脚本 pandas HuggingFace Dataset模型层预训练模型、微调、评估PyTorch Transformers LoRA服务层推理接口、并发、性能FastAPI PyTorch推理监控层延迟、错误率、数据漂移Prometheus 自定义日志这么拆解的意义在于你能清楚地看到每个阶段该干什么不会因为某一步卡住就整个项目停摆。比如环境搭不起来的时候你可以先用云端的Notebook验证模型效果数据来不及清洗的时候可以先跑通一个最小demo验证流程。每一层都有独立的交付物这就是工程化的思维。1.3 一条主线案例客服工单自动分类系统后面的内容我都围绕一个具体项目展开假设你是一家电商公司的工程师客服部门每天收到几百到几千条用户工单需要把它们自动分到对应类目比如售后、物流、产品咨询、退款、投诉。传统做法是人工看标题和内容费时费力而且新人培训成本高。这个项目足够典型它覆盖了AI工程里几乎所有核心环节它是文本分类任务数据可以用历史工单模型可以用中文预训练模型微调预测结果需要接进工单系统业务方不仅要知道“分到哪一类”还要知道“置信度有多少”方便处理低置信度工单。整个项目用一个周末就能跑出第一个可用版本非常适合用来讲“from scratch”的完整路径。2. 从零开始的第一步开发环境与工具链2.1 硬件与资源没有A100也能起步很多人第一步就卡在“我没钱买GPU”。这里先说清楚做AI工程和做大规模预训练完全是两个量级的需求。你不需要训练GPT你只需要在别人的基座模型上做微调显存需求远没有想象中大。以工单分类为例用中文预训练模型如BERT或更小的模型序列长度设到128或256batch size设成16或32一块GTX 306012GB甚至更小的显卡就能跑得非常舒服。LoRA微调会让显存压力进一步下降。如果你用的是云服务器租一张T4或V100按量付费整个训练周期可能就几块钱到几十块钱远比想象中便宜。云上的选择多一些但有一点必须提醒训练和推理的硬件选择策略是不同的。训练看的是总算力和显存推理看的是延迟和吞吐。建议训练用按量付费的高配机器跑完就释放推理反而适合长期持有稳定的GPU或直接用CPU推理因为很多模型量化后CPU跑的延迟也能接受。别一上来就买最高配那是预算浪费。2.2 Python虚拟环境与CUDA版本那点事软件环境的坑我在自己动手的这些年里踩过太多次几乎每次都能碰到“换个机器就全乱套”的情况。最典型的例子是PyTorch和CUDA版本不匹配torch.cuda.is_available()返回False然后你会浪费半天时间去找原因。我的习惯是这样的# 1. 先用 conda 创建独立环境避免污染系统 Python conda create -n ai-eng python3.10 -y conda activate ai-eng # 2. 安装对应 CUDA 版本的 PyTorch # 注意先看本机显卡驱动支持的 CUDA 版本再装 PyTorch pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121 # 3. 验证是否可用 python -c import torch; print(torch.cuda.is_available())验证torch.cuda.is_available()是False的话不要急着重装系统。先运行nvidia-smi看驱动支持的CUDA版本再用nvcc --version看本机工具包版本PyTorch的CUDA运行时和驱动的CUDA版本可以不一致只要PyTorch要求的CUDA版本不超过驱动支持的上限就行。这个坑是绝大多数环境类问题的源头。2.3 用Docker锁死环境杜绝“在我机器上好好的”环境问题在你自己机器上重装一次最多浪费一晚上。但如果团队协作别人拉你代码跑不起来那就要出大问题。我强烈建议用Docker把环境做进镜像里连CUDA、Python依赖、系统库一起打包做到“代码在哪跑环境都一样”。我常用的做法是这样的FROM pytorch/pytorch:2.0.1-cuda11.7-cudnn8-runtime WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . . CMD [python, train.py]构建镜像后本地跑、服务器跑甚至迁移到云上都是一样的行为。这不叫过度工程化这是对一个上线系统的底线要求。没有Docker的AI项目回溯问题的时候会非常痛苦尤其是模型表现莫名其妙变化时你根本不知道是代码变了、数据变了还是环境中某个库的版本被动了。3. 数据工程真正决定模型上限的环节3.1 数据从哪来开源数据集、爬取、还是自己标注模型的上限由数据决定这句话被说烂了但真正愿意在数据上花时间的人不多。工单分类任务里如果公司有历史工单优先用真实业务数据那是无价之宝没有的话只能先用手头能拿到的开源数据集做原型再逐步让业务方标注真实数据。爬虫和开源数据都存在一个共性问题分布和你的真实场景不一致。电商工单的语气、用词、上下文和后端处理流程开源数据里不会有。所以我的建议是第一步先用公开数据把整个流程跑通快速看到模型能work第二步花大力气做真实数据的清洗和标注这个才是项目能落地的关键。标注阶段有个很现实的坑直接让业务人员大量标注质量会非常不稳定。我的经验是设计一套双人标注加抽检的流程先让两个人分别标同一批数据算一致率Cohen‘s Kappa低于0.8就得回来对齐标准。这个前置工作看起来麻烦但能避免你用脏数据训出一堆垃圾模型。3.2 清洗与构造文本数据的通用处理流程文本数据清洗很多人以为就是去掉换行和HTML标签实际上远不止这些。站在业务角度看工单数据会发现几个高频问题重复提交、无意义字符、PII信息手机号、地址、广告噪声、网络用语。我给工单数据写过一个清洗管线大概是这样的import re import pandas as pd def clean_text(text: str) - str: # 去掉HTML标签 text re.sub(r[^], , text) # 去掉URL text re.sub(rhttp\S, , text) # 去掉手机号/邮箱保留占位符 text re.sub(r1[3-9]\d{9}, [PHONE], text) text re.sub(r\w\w\.\w, [EMAIL], text) # 压缩多余空白 text re.sub(r\s, , text).strip() return text df[clean_text] df[raw_text].apply(clean_text)这里有几个细节值得多说一句。正则去掉了手机号和邮箱是防止模型学到“包含手机号的就是投诉”这种表面模式以后真实场景里没有手机号的投诉就漏判。把“去重”放在清洗之前做基于原文去重的话效果会更好因为清洗后的文本可能丢失信息导致两条不同的工单被误判成重复。3.3 数据管线实现从CSV到可训练的Dataset数据管线的目标是让训练代码只关心“从Dataset取一个batch”而不关心数据从哪来、在哪清洗。PyTorch里的DatasetDataLoader已经是标准做法但很多人没把逻辑分层。我的习惯是让数据集类只负责“加载和索引”清洗和预处理放到单独的模块里。这样后续切换数据源、调整预处理流程都不用动模型代码。from torch.utils.data import Dataset from transformers import AutoTokenizer class TicketDataset(Dataset): def __init__(self, df, tokenizer_namehfl/chinese-bert-wwm-ext, max_len128): self.df df.reset_index(dropTrue) self.tokenizer AutoTokenizer.from_pretrained(tokenizer_name) self.max_len max_len self.label_map {label: idx for idx, label in enumerate(sorted(df[label].unique()))} def __len__(self): return len(self.df) def __getitem__(self, idx): row self.df.iloc[idx] encoded self.tokenizer( row[title] [SEP] row[content], truncationTrue, max_lengthself.max_len, paddingmax_length ) return { input_ids: torch.tensor(encoded[input_ids]), attention_mask: torch.tensor(encoded[attention_mask]), labels: torch.tensor(self.label_map[row[label]]) }注意我里面把标题和正文用[SEP]拼到一起这是因为工单的标题信息密度通常很高分类的关键词往往来自标题。序列长度设128而不是512因为绝大多数工单内容在一两百字以内截断处理不会损失太多信息却能显著降低显存占用和训练时间。4. 模型训练迁移学习与微调的实战细节4.1 为什么不从零训练模型从零训练一个语言模型需要海量数据和算力个人和小团队完全不具备这个条件。就像你不会从炼铁开始造一辆车而是买来发动机组装整车。预训练模型就是这个“发动机”。所以“from scratch”指的从来不是从零训练模型而是从零把一个AI项目完整搭建起来。用现成的预训练模型做下游任务微调是这个时代做AI工程最成熟的做法。对工单分类来说选择一个中文预训练模型比如BERT类在几千条标注数据上做微调往往就已经能到95%以上的准确率。4.2 用LoRA低成本微调分类模型全参数微调对大模型来说是显存噩梦近年来的LoRA技术则只训练一小部分低秩矩阵参数训练参数量减少95%以上效果却接近全参微调。我用的方式是用HuggingFace的PEFT库from transformers import AutoModelForSequenceClassification, TrainingArguments, Trainer from peft import LoraConfig, get_peft_model, TaskType model AutoModelForSequenceClassification.from_pretrained( hfl/chinese-bert-wwm-ext, num_labelslen(train_dataset.label_map) ) lora_config LoraConfig( task_typeTaskType.SEQ_CLS, r8, # 秩的大小常见取8或16 lora_alpha32, # 缩放系数 lora_dropout0.1, target_modules[query, value] # 只对attention的q、v矩阵加LoRA ) model get_peft_model(model, lora_config) training_args TrainingArguments( output_dir./checkpoints, learning_rate2e-5, per_device_train_batch_size16, num_train_epochs5, evaluation_strategyepoch, save_strategyepoch, load_best_model_at_endTrue, metric_for_best_modelaccuracy ) trainer Trainer( modelmodel, argstraining_args, train_datasettrain_dataset, eval_datasetval_dataset, compute_metricscompute_metrics ) trainer.train()LoRA的r值怎么定像工单分类这种中等规模任务r8就够用更大的r不一定会带来明显提升反而增加显存和过拟合风险。target_modules只作用于query和value直觉上这两者承载了更多任务相关的语义关联信息实际效果也确实是这个组合最稳。4.3 训练循环里最容易翻车的几个点训练过程的绝大多数坑集中在loss和过拟合上。先说loss变成NaN遇到这个我第一反应不是调代码而是去查数据的标签有没有问题是不是存在NaN的浮点数特征、是不是某个类别的样本数量极端到让损失爆炸。检查完数据再调学习率用warmup配合梯度裁剪基本能解决问题。过拟合在小样本上很容易出现。工单分类数据集可能只有几千条模型训练三五轮就开始在验证集上退化。解决方案很简单早停加交叉验证。上面代码里已经写了load_best_model_at_endTrue让Trainer自己跟踪验证集指标选最好的那轮checkpoint。另一个我实际遇到的问题是类别不均衡。假设投诉类只占5%其他类占95%一个“全部分成其他类”的模型就能拿到95%的准确率但业务完全不可用。这时候建议用加权交叉熵或者干脆在评估时主要看各类别的F1而不是整体准确率。我在评估指标里同时算了Macro-F1和准确率以Macro-F1作为选模型的依据。5. 部署上线工程化的最后一公里5.1 模型导出与推理优化训练出的模型是.bin格式的PyTorch权重不能直接对外提供HTTP服务需要用TorchScript或ONNX导出成推理格式。对于这套工单分类系统ONNX是性价比很高的选择。导出前先把模型转成FP16显存和延迟能同时降一半。不过要注意某些CPU推理环境对FP16并不友好反而更慢需要量化成INT8。在实践里我设了一个对比表方便根据不同部署环境选方案部署方式平均延迟单条内存占用适用场景PyTorch FP32 CPU约30ms高开发调试PyTorch FP16 GPU约8ms中正式服务ONNX FP16 GPU约5ms中正式服务ONNX INT8 CPU约12ms低低成本CPU部署ONNX导出的代码大致这样import torch from transformers import AutoModelForSequenceClassification model AutoModelForSequenceClassification.from_pretrained(best_checkpoint) model.eval() dummy_input { input_ids: torch.ones(1, 128, dtypetorch.long), attention_mask: torch.ones(1, 128, dtypetorch.long) } torch.onnx.export( model, (dummy_input[input_ids], dummy_input[attention_mask]), ticket_classifier.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{input_ids: {0: batch}, attention_mask: {0: batch}, logits: {0: batch}}, opset_version13 )这里有个经验dynamic_axes一定得加。不加的话导出的模型只能接受固定batch size的输入线上服务一遇到并发就难受。加了batch维度动态化之后单条和批量推理都能灵活处理。5.2 用FastAPI把模型封装成服务模型导出好了就该让它进入业务系统。FastAPI是我一直推荐的选择性能好、自带OpenAPI文档代码量也少。部署这一层最容易犯的错是每来一次请求就做一次tokenizer编码和模型推理不仅慢而且tokenizer在并发下会成为瓶颈。正确做法是服务启动时一次性加载模型和tokenizer放到全局变量或者依赖注入里推理时只做必要的计算。代码大致是这样的from fastapi import FastAPI from transformers import AutoTokenizer import onnxruntime as ort app FastAPI() tokenizer AutoTokenizer.from_pretrained(hfl/chinese-bert-wwm-ext) session ort.InferenceSession(ticket_classifier.onnx, providers[CUDAExecutionProvider, CPUExecutionProvider]) app.post(/predict) def predict(req: dict): title req[title] content req.get(content, ) encoded tokenizer( title [SEP] content, truncationTrue, max_length128, paddingmax_length, return_tensorsnp ) logits session.run(None, { input_ids: encoded[input_ids], attention_mask: encoded[attention_mask] })[0] prob softmax(logits[0]) top_idx prob.argmax() return { label: id2label[top_idx], confidence: float(prob[top_idx]) }这里要注意ONNX的输入名必须和导出时保持一致不然会报错。我踩过的一个坑是导出时用了input_ids推理时也用了input_ids但导出时没有把attention_mask设为动态维度服务里一旦batch size不是1就报shape不匹配这个问题在开发环境根本测不出来上线压测才暴露。5.3 性能、监控与迭代闭环模型服务上线后你还要回答几个问题延迟是否达标错误率是否飙升输入数据的分布是否在变化这三个问题分别对应性能、可靠性和监控。性能方面如果单条推理10ms看起来很快但假如一次要批量处理1000条工单就要考虑批量推理和异步队列。我后来给系统加了批量预测能力接口收到一条就入队列攒够32条或等50ms再一次性推理吞吐提升明显成本却很有限。监控方面我至少会记录三类指标指标含义告警阈值p99延迟99%请求耗时大于100ms预测置信度均值模型整体自信程度低于0.8且持续下降低置信度工单占比被抛弃的样本比例大于10%有一个容易被忽略但很重要的点置信度均值比单条置信度更能反映模型退化。如果发现置信度持续走低大概率不是模型出了问题而是业务输入变了比如出现新的话术、新的产品类型。这时候不能只靠模型去硬扛要给业务方人工兜底的通道。这已经属于MLOps的范畴但哪怕你只做一个最小版本也要把这个闭环跑起来。6. 复盘从零到一踩过的坑6.1 这些事比预想中多花三倍时间很多人对AI项目的预期是“训练模型很费时间”但实际体验后会发现训练本身反而是最顺的一段。我整个项目花的时间分配大概是这样环境与工具链搭建占了15%都是版本匹配和Docker镜像的小麻烦数据清洗与标注占了35%比预期多一倍标注质量反复返工模型训练与调参占了20%LoRA方案本身很顺难在评估指标对标部署与接口对接占了25%ONNX导出和动态shape的坑监控与告警占了5%写代码很快想清楚指标花时间数据清洗与标注是真正的重头戏这一点我后来跟很多同行聊都得到印证。模型不对你可以换模型数据不干净你换什么模型都没用。6.2 给后来者的建议清单以我个人做了几个AI项目、经历了若干次推倒重来后的体会来收尾有几点想明确分享出来第一先跑通最小闭环再追求模块化。很多初学者会把工程拆得很细但连一段数据都跑不通之前那些抽象都是纸上谈兵。我的习惯是先用脚本不管多丑跑通全部流程再回头把里面的步骤抽成函数、类、模块。第二训练脚本从一开始就固定随机种子否则你会在复现模型结果上浪费大量时间。第三别过度追求高精度业务上95%的准确率加一套低置信度兜底机制往往比99%的准确率但没有解释能力更受欢迎。我管这个项目叫“ai-engineering-from-scratch”是因为它确实是从一台干净系统开始一步一步走到线上服务的。这条路不算漫长但每一步都有实实在在的坑。希望这条分享能帮你少走几个弯路。
返回列表