---

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

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

1. 数据稀疏：真实对话里，80% 的咨询集中在 20% 的意图，长尾意图样本寥寥几条，模型直接“脸盲”。
2. 标注不一致：三位外包同学同时标“退货”意图，有人写成“退款”，有人写成“退换货”，训练时标签噪声直接拉低 F1。
3. 多轮对话断档：用户先问“你们家鞋偏大吗？”再追问“那 42 码要不要买小一号？”两句话合起来才是真实意图，可日志里常被切成两条独立样本，模型学到一半一脸懵。
4. 隐私与合规：客服记录里姓名、地址、手机号应有尽有，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'))

生产建议：让数据集“自我生长”

1. 数据增强

• 同义词替换：维护客服领域同义词典，“退款→退货”“尺码→码数”，随机 15% 替换，意图标签不变。
• 回译：中→英→中，调用 DeepL 免费接口，生成语义等价的新句子，实测 BLEU 与原句 0.82，可带来 1.8% F1 提升。
• 模板生成：把高频句式“我的订单还没收到”做槽位替换，自动生成 5 k 新句，低成本补长尾。

2. 冷启动技巧

• FAQ 先跑：把官方 200 条 FAQ 人工标注成黄金数据，训练 1.0 版模型上线，灰度 10% 流量，收集用户真实问法，再迭代。
• 规则兜底：关键词+正则做“兜底意图”，当模型置信度<0.7 时走规则，保证首日体验不翻车。
• 主动学习：每轮挑出模型最不确定的 5% 日志，人工标注后快速微调，3 周就把覆盖率从 65% 拉到 87%。

3. 质量监控看板

• 标注一致性：每周随机抽 500 条双标，α<0.8 立即回炉。
• 模型漂移：对比昨日 vs 上周的预测分布，KL 散度>0.05 触发再训练。
• 用户反馈：点踩率>5% 的意图自动进入高优标注池，保证“用户教模型”。

开放问题：数据量级 vs 标注成本，这道天平你怎么摆？

做 100 万条标注，预算立刻飙到七位数；不标，模型又学不精。有人用半监督+主动学习小步快跑，有人直接上众包“人海战术”，也有人干脆买现成数据集再微调。你的团队会怎么选？欢迎留言聊聊最贴合业务的那条“省钱又好用”的路线。

---