场景痛点：规则引擎的“最后一公里”
去年双十一，公司客服系统被“这件衣服有没有S码”和“这件衣服有S号吗”两句话彻底打败。人工维护的 3000+ 正则规则在 48 小时内膨胀到 5000+，仍然无法覆盖同义词、语序变换、口语省略。更尴尬的是，当意图被误分类后，后续固定 FAQ 列表直接“鸡同鸭讲”，转化率掉到谷底。传统方案的三座大山：
- 规则维护成本指数级增长
- 生成问题与上文无关，用户需要重复输入背景信息
- 意图漂移导致多轮对话状态丢失，无法形成有效漏斗
痛定思痛，我们决定用 AI 把“识别+生成”这两步全部接管，让客服系统自己“听懂”并“追问”。
技术选型：RNN 已老，Transformer 当立
意图识别赛道
| 模型 | 训练速度 | 长程依赖 | 线上延迟 | 准确率(自建 32 类) |
|---|---|---|---|---|
| Bi-LSTM | 慢 | 衰减严重 | 18 ms | 87.2 % |
| TextCNN | 快 | 3-gram 局限 | 6 ms | 89.4 % |
| BERT-base | 中等 | Self-Attention | 11 ms | 94.7 % |
结论：BERT 在 F1 与延迟之间取得最佳平衡，且微调代码量最小，工程师心智负担最低。
问题生成赛道
- GPT-3：API 方便，但 ① 输出不可控，容易“满嘴跑火车”；② 数据出境合规风险；③ 按 Token 计费，高并发下成本爆炸。
- T5-small：可私有部署，Beam Search + Repeat Penalty 后，事实一致性>96%，单机 QPS≈120，成本趋近于零。
最终方案：BERT 做意图判别，T5 做条件生成，两者均基于 Transformer，参数共享与量化策略统一，运维只有一套 GPU 池。
资产库：
核心实现
1. 意图分类模型（BERT + PyTorch）
目录结构
intent/ ├── data │ └── raw.csv # 原始语料：text,label ├── train.py ├── infer.py └── tests └── test_model.pytrain.py 关键片段（含注释，符合 PEP8）
# -*- coding: utf-8 -*- import torch, random, numpy as np, pandas as pd from transformers import BertTokenizerFast, BertForSequenceClassification from sklearn.model_selection import train_test_split from torch.utils.data import Dataset, DataLoader from sklearn.preprocessing import LabelEncoder RANDOM_SEED = 42 random.seed(RANDOM_SEED) np.random.seed(RANDOM_SEED) torch.manual_seed(RANDOM_SEED) MAX_LEN = 64 BATCH = 128 EPOCHS = 4 LR = 2e-5 class IntentDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_len): self.texts, self.labels = texts, labels self.tokenizer, self.max_len = tokenizer, 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' ) item = {k: v.squeeze(0) for k, v in encoding.items()} item['labels'] = torch.tensor(self.labels[idx], dtype=torch.long) return item def main(): df = pd.read_csv('data/raw.csv') le = LabelEncoder() labels = le.fit_transform(df['label']) train_texts, val_texts, train_labels, val_labels = train_test_split( df['text'], labels, test_size=0.1, random_state=RANDOM_SEED ) tokenizer = BertTokenizerFast.from_pretrained('bert-base-chinese') train_ds = IntentDataset(train_texts, train_labels, tokenizer, MAX_LEN) val_ds = IntentDataset(val_texts, val_labels, tokenizer, MAX_LEN) train_loader = DataLoader(train_ds, batch_size=BATCH, shuffle=True) val_loader = DataLoader(val_ds批 , batch_size=BATCH) model = BertForSequenceClassification.from_pretrained( 'bert-base-chinese', num_labels=len(le.classes_) ).cuda() optimizer = torch.optim.AdamW(model.parameters(), lr=LR) criterion = torch.nn.CrossEntropyLoss() for epoch in range(EPOCHS): model.train() for batch in train_loader: batch = {k: v.cuda() for k, v in batch.items()} outputs = model(**batch) loss = outputs.loss loss.backward() optimizer.step(); optimizer.zero_grad() # 验证 & 保存略 tokenizer.save_pretrained('output/') model.save_pretrained('output/') if __name__ == '__main__': main()单元测试示例（tests/test_model.py）
import pytest, torch from transformers import BertTokenizer, BertForSequenceClassification @pytest.fixture def setup(): tok = BertTokenizer.from_pretrained('output/') model = BertForSequenceClassification.from_pretrained('output/') return tok, model def test_predict_return_shape(setup): tok, model = setup text = "这件可以退货吗" inputs = tok(text, return_tensors='pt') with torch.no_grad(): logits = model(**inputs).logits assert logits.shape[1] == 32 # 32 类2. 问题生成 pipeline（T5）
采用“上下文 + 意图”作为条件 prompt，强制模型生成后续澄清问句。
from transformers import T5Tokenizer, T5ForConditionalGeneration import torch tok = T5Tokenizer.from_pretrained('ClueAI/T5-small-chinese') model = T5ForConditionalGeneration.from_pretrained('ClueAI/T5-small-chinese') def build_prompt(context, intent): # 用[INTENT]做显式控制，降低幻觉 return f"上下文：{context} [INTENT]{intent} 追问：" def generate_question(context, intent, max_len=64, num_beams=4, repetition_penalty=2.0): prompt = build_prompt(context, intent) inputs = tok(prompt, return_tensors='pt') outputs = model.generate( **inputs, max_length=max_len, num_beams=num_beams, repetition_penalty=repetition_penalty, early_stopping=True ) return tok.decode(outputs[0], skip_special_tokens=True) # 示例 print(generate_question("用户问：这件衣服有红色吗", "询问库存")) # 输出：请问您需要多大尺码？我们库存实时变动，可帮您确认。性能优化三板斧
1. 模型量化部署
- 意图模型：BERT 采用 ONNX Runtime + 动态量化（INT8），延迟从 11 ms 降到 4.3 ms，F1 下降 0.4%，可接受。
- 生成模型：T5 使用 PyTorch JIT + 半精度（FP16），显存占用减半，Beam Search 4 路并发 QPS≈120→230。
2. 对话状态管理
自研轻量级状态机，以 Redis Hash 存储 session：
key: cx:{uid} field: context, intent, slot, ttl每次请求只 O(1) 读写，TTL 300 s 自动淘汰，防止僵尸 key 堆积。
3. 异步处理架构
网关层 → Kafka → 推理微服务（FastAPI + Uvicorn）→ 结果写回 WebSocket。
通过 Prefect 做队列削峰，单卡 A10 可抗 8000 并发，99 线 180 ms。
生产环境避坑指南
- 冷启动：意图模型初始无数据时，先用规则兜底，同时把用户日志实时回流到“待标注池”。每日凌晨主动学习（Active Learning），优先挑选熵最高的 5% 样本，两周后线上准确率即可从 68% 提升到 91%。
- 安全过滤：T5 生成后过两层正则 + 敏感词 DFA + 情绪模型（中文 RoBERTa 微调），出现辱骂/过度承诺直接丢弃并触发兜底回复。
- 高并发缓存：意图结果缓存 key 设计为
hash(text)并加入“最近 7 天模型版本号”，防止模型更新后旧缓存误用；生成问题侧因上下文差异大，缓存命中率仅 12%，故采用 GPU 级缓存（FasterTransformer）而非业务缓存，降低序列重复编译开销。
开放性问题：如何评估生成问题的质量？
目前我们离线用 BLEU-4 与人工“相关+友好”双指标，线上用“用户是否继续回复”作为弱监督。
但 BLEU 高≠体验好，人工贵且慢。是否可以用：
- 强化学习，把“最终是否转人工”作为 reward？
- 引入对话级连贯性（Coherence）自动评分？
- 用对抗样本检测模型是否过度“安全”而生成无效问题？
欢迎读者动手实验：
- 在开源 Multi-Domain-WOZ 上训练 T5，对比 Beam Search vs. Diverse Beam Search 的 Coherence 分数；
- 构建轻量级 rank 模型，把“生成问题+真实用户回复”作为正例，“生成问题+用户沉默 10 s”作为负例，看 AUC 能否逼近人工标注。
期待你的 PR 与实验报告。
