---

背景痛点：原始对话数据到底“脏”在哪？

做智能客服的同学都懂，业务方把原始日志一股脑丢过来时，数据往往长这样：

• 用户口语放飞：“啊？你们那个啥…上次我买的那个能退不？”
• 多轮对话断片：第3轮突然提到“它”，却找不到“它”指代的实体。
• 中英混排：“APP闪退了，help！”
• 标注标准漂移：A同学把“查询订单”标成order_query，B同学标成check_order，C同学干脆标order。

结果模型一训练，意图识别准确率连80%都不到。为了把噪声降下来，我们干脆搭了一套“清洗+标注”双轮驱动的流水线，三个月跑下来，标注效率提升300%，意图识别准确率稳在98%以上。下面把踩过的坑、撸过的代码一次性摊开。

技术方案总览：规则引擎 + 主动学习

整个系统分三层：

1. 数据清洗层：正则+NLP多级过滤，把明显脏数据挡在门外。
2. 标注平台层：React+Flask，B/S架构，浏览器里点两下就能标，后台主动学习模型实时挑“最值钱”的样本送标。
3. 质量监控层：每小时算一次Cohen's Kappa，掉到0.8以下自动报警。

数据清洗层实战：正则与模型组合拳

1. 多级过滤架构

• 一级：正则速筛（URL、手机号、脏词）
• 二级：FastText语种识别，把非中英双语的全部丢掉
• 三级：BERT微调模型做语义断句，把“啊哦哈哈”这种无意义片段干掉

2. 核心代码片段

下面这段clean_pipeline.py可直接搬到Airflow里跑，异常、日志、并发全齐活。

# clean_pipeline.py
import re, json, logging, emoji, fasttext
from pathlib import Path
from joblib import Parallel, delayed
from transformers import pipeline
from typing import List, Dict

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
ft_model = fasttext.load_model("lid.176.bin")
sem_filter = pipeline("text-classification", model="your-semantics-model")

REGEXES = {
    "url": re.compile(r"https?://\S+"),
    "phone": re.compile(r"\b1[3-9]\d{9}\b"),
    "emoji": emoji.get_emoji_regexp(),
}

def regex_filter(text: str) -> str:
    """一级：正则速扫"""
    for name, reg in REGEXES.items():
        text = reg.sub(f"<{name}>", text)
    return text.strip()

def lang_filter(text: str, thr: float = 0.85) -> bool:
    """二级：语种过滤"""
    pred, score = ft_model.predict(text.replace("<url>", ""))
    return pred[0] in ("__label__en", "__label__zh") and score[0] > thr

def semantic_filter(text: str, thr: float = 0.9) -> bool:
    """三级：语义过滤"""
    if len(text) < 5:
        return False
    res = sem_filter(text)[0]
    return res["label"] == "valid" and res["score"] > thr

def clean_one(sample: Dict[str, str]) -> Dict[str, str]:
    try:
        raw = sample["raw"]
        txt = regex_filter(raw)
        if not lang_filter(txt):
            return {"id": sample["id"], "clean": None, "reason": "lang"}
        if not semantic_filter(txt):
            return {"id": sample["id"], "clean": None, "reason": "semantic"}
        return {"id": sample["id"], "clean": txt, "reason": "ok"}
    except Exception as e:
        logging.exception(f"clean_one error: {e}")
        return {"id": sample["id"], "clean": None, "reason": "exception"}

def run_batch(file: Path, out: Path, n_jobs: int = 8):
    data = json.loads(file.read_text(encoding="utf8"))
    res = Parallel(n_jobs=n_jobs)(delayed(clean_one)(s) for s in data)
    out.write_text(json.dumps(res, ensure_ascii=False, indent=2), encoding="utf8")
    logging.info(f"finished {file} -> {out}")

if __name__ == "__main__":
    run_batch(Path("raw.json"), Path("clean.json"))

跑完以后，脏数据占比从28%降到6%，后面模型训练省了一半GPU时间。

标注系统设计：React+Flask+主动学习

1. 前后端分离要点

• 前端React：用react-query轮询后端，拿到高优先级样本，标注完一条POST /api/annotate实时回写。

• 后端Flask：暴露三个核心接口

  • GET /api/next：主动学习挑样本，策略用“熵+多样性”双指标。
  • POST /api/annotate：回写标签，同时把样本写进“已标池”。
  • GET /api/agreement：实时算Kappa，给管理员看板。

2. 主动学习反馈循环

每新增500条已标数据，就触发一次轻量微调（LoRA，10分钟级），重新打分排序，保证送标样本永远是最“不确定”的那一批。实测同样2000条标注量，主动学习版本比随机采样在测试集上F1高7个点。

性能优化：规则匹配 vs 模型推断

单条正则过滤平均0.2 ms，BERT推断20 ms，差了两个数量级。为了不给业务方拖节奏，我们把“重模型”拆到分布式队列：

1. 清洗阶段用Celery+Redis，把semantic_filter任务路由到GPU节点，CPU节点只跑正则+FastText。
2. 标注阶段用RabbitMQ做优先级队列，高不确定样本插标到队头。
3. 结果：日增量500万条日志，4张T4卡+20台CPU机，4小时全跑完。

避坑指南：标签漂移与质量监控

标注小哥连续标两小时，标签分布会肉眼可见地往“默认类别”飘。我们搞了三板斧：

• 每标100条插5条“金标”答错就弹窗提醒。
• 每小时算一次Cohen's Kappa，低于0.8自动冻结账号，回炉培训。
• 每周随机抽10%双标，A/B标完再算一致性，Kappa>0.85才给过。

自从上了这套，标签漂移从12%降到2%，模型迭代再也没出现“莫名掉点”。

延伸思考：小语种数据增强

如果业务拓展到越南语、印尼语，样本少怎么办？我们内部实验了两条路线：

1. 翻译回译：英→小语种→英→中文，中间加同义词替换，保证意图不变。
2. 语音-文本双通道：先用TTS生成语音，再加背景噪声，ASR转回文本，变相拿到口语化数据。

在越南语场景下，用回译+噪声ASR，把训练集从3千扩到2万，意图识别F1提升9个点，成本只有人工标注的1/5。

写在最后

这套清洗+标注流水线已经在我们内部跑了大半年，从最早“数据没法看”到现在“模型上线只要一周”，核心就是：先让规则跑得快，再让模型学得准，最后让人标得少。代码全部模块化，丢到GitLab CI里能一键部署。如果你也在为客服数据头疼，不妨挑一段脚本先跑起来，相信你会回来点赞的。

---