智能客服系统数据集构建实战：从数据清洗到模型训练全流程解析-开发者社区

背景痛点：为什么客服数据总“喂不饱”模型？

做智能客服的同学都懂，最怕的不是算法，而是“没料”。
数据问题一箩筐：

数据稀疏：真实对话里，80% 的咨询集中在 20% 的意图，长尾意图样本寥寥几条，模型直接“脸盲”。
标注不一致：三位外包同学同时标“退货”意图，有人写成“退款”，有人写成“退换货”，训练时标签噪声直接拉低 F1。
多轮对话断档：用户先问“你们家鞋偏大吗？”再追问“那 42 码要不要买小一号？”两句话合起来才是真实意图，可日志里常被切成两条独立样本，模型学到一半一脸懵。
隐私与合规：客服记录里姓名、地址、手机号应有尽有，GDPR、个保法双重夹击，不脱敏直接上云，法务分分钟敲门。

技术方案：一条流水线把“脏数据”炼成“精粮”

1. 数据采集：爬虫策略与隐私合规要点

公开渠道：电商商品页“问大家”板块、微博评论、知乎问答，用 Scrapy 写定向爬虫，先 robots 协议白名单，再限速 1 req/s，避免 403。
授权日志：跟客服系统厂商签数据使用协议，把近 3 个月对话导出，脱敏脚本跑在最前面：正则挖手机号、地址，再用命名实体识别（NER）二次复查，双重保险“漏网之鱼”。
合规存储：落地到加密盘，AES-256 落盘，密钥放 KMS，谁用谁临时申请，审计日志 180 天自动清理。

2. 数据清洗：正则 + NLP 去噪三板斧

正则第一关：去掉“【系统自动回复】”“转人工中……”这类固定模板，一条re.sub(r'【.*?】', '', text)秒清。
语言模型第二关：用轻量 fastText 训练“客服 vs 垃圾”二分类，把广告、表情包、乱码丢进垃圾桶，准确率在 96% 左右，成本还低。
停用词第三关：维护客服领域停用词表，“亲”“哦”“呢”口语化虚词干掉，避免模型把语气词当主要特征。

3. 标注体系：意图与实体“两张表”

意图分类：先业务后算法，把客服 FAQ 拆成 3 级树，第一级“售前/售中/售后”，第二级“尺码/物流/退款”，第三级才到叶子意图，共 127 类，保证外包同学“有章可循”。
实体识别：采用 BIO 标注，人名、地址、订单号、商品 ID 四大类，订单号统一正则[A-Z]{2}\d{9}，减少标注歧义。
一致性校验：同一份数据 20% 交叉双标，Krippendorff α≥0.8 才算合格，否则打回重标。

代码实战：从“脏对话”到“能跑模型”的 Python 旅程

以下代码均跑通 Python 3.9，transformers 4.30，按顺序复制即可复现。

1. 对话清洗脚本（clean_dialogue.py）

# -*- coding: utf-8 -*- import re import json import jieba from pathlib import Path # 1. 自定义停用词表 STOP_WORDS = set(open('stopwords_cn.txt', encoding='utf8').read().split()) # 2. 正则合集 RE_PATTERNS = { 'sys_reply': re.compile(r'【.*?】'), 'emoji': re.compile(r'\[.*?\]'), 'phone': re.compile(r'1[3]\d{9}'), # 简易手机 'order_sn': re.compile(r'[A-Z]{2}\d{9}'), # 订单号脱敏 } def mask_private(text: str) -> str: """脱敏+去噪""" for k, v in RE_PATTERNS.items(): text = v.sub(f'[{k}]', text) return text def clean_text(text: str) -> str: """清洗+分词+去停用词""" text = mask_private(text) words = [w for w in jieba.lcut(text) if w not in STOP_WORDS and len(w) > 1] return ' '.join(words) def process_file(input_jsonl: str, output_jsonl: str): """逐行处理""" with open(input_jsonl, encoding='utf8') as fin, \ open(output_jsonl, 'w', encoding='utf8') as fout: for line in fin: item = json.loads(line) item['cleaned'] = clean_text(item['raw']) fout.write(json.dumps(item, ensure_ascii=False) + '\n') if __name__ == '__main__': process_file('raw_dialogues.jsonl', 'cleaned_dialogues.jsonl')

跑完示例：

原始：亲，我的订单AB123456789还没收到呢？【系统自动回复】正在查询…… 清洗：订单 收到 查询

2. 轻量意图分类（intent_model.py）

# -*- coding: utf-8 -*- import torch from torch.utils.data import Dataset, DataLoader from transformers import (AutoTokenizer, AutoModelForSequenceClassification, AdamW, get_linear_schedule_with_warmup) from sklearn.preprocessing import LabelEncoder import pandas as pd import json MAX_LEN = 128 BATCH = 32 EPOCHS = 3 MODEL_NAME = 'bert-base-chinese' class IntentDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_len=MAX_LEN): 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): enc = 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 enc.items()} item['labels'] = torch.tensor(self.labels[idx], dtype=torch.long) return item def load_data(path): texts, labels = [], [] with open(path, encoding='utf8') as f: for line in f: d = json.loads(line) texts.append(d['cleaned']) labels.append(d['intent']) return texts, LabelEncoder().fit_transform(labels) def train(): tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) texts, labels = load_data('cleaned_dialogues.jsonl') ds = IntentDataset(texts, labels, tokenizer) dl = DataLoader(ds, batch_size=BATCH, shuffle=True) model = AutoModelForSequenceClassification.from_pretrained( MODEL_NAME, num_labels=len(set(labels))) model.cuda() opt = AdamW(model.parameters(), lr=2e-5) sched = get_linear_schedule_with_warmup( opt, num_warmup_steps=0, num_training_steps=len(dl) * EPOCHS) model.train() for epoch in range(EPOCHS): for batch in dl: opt.zero_grad() outs = model(**{k: v.cuda() for k, v in batch.items()}) loss = outs.loss loss.backward() opt.step() sched.step() torch.save(model.state_dict(), f'epoch_{epoch}.pt') print(f'epoch {epoch} loss={loss.item():.4f}') if __name__ == '__main__': train()

验证阶段用 F1-score：

from sklearn.metrics import f1_score preds = torch.cat([model(b)[1].argmax(1).cpu() for b in val_dl]) true = torch.cat([b['labels'] for b in val_dl]) print(f1_score(true, preds, average='weighted'))