智能客服对话分析实战：基于NLP与规则引擎的混合架构设计-平芜编程栈

背景：纯NLP方案在客服场景下的“水土不服”

去年双十一，我们给电商客服系统上线了一套“全神经网络”对话分析模块，想着终于可以把人工标注团队“省掉一半”。结果凌晨两点，老板在群里疯狂艾特：“为什么‘我要退款’被识别成‘我要退货’？用户都炸锅了！”

复盘发现，纯NLP方案在客服场景确实容易踩坑：

长尾意图淹没：训练集里“开发票”出现 3 次，模型直接把它归到“报销”大类，结果用户真实意图是“补开发票”，准确率 0%。
业务规则漂移：大促期间“价保”政策临时调整，模型重新训练至少 6 小时，线上一直在误答。
可解释性为零：一旦识别错误，运营只能把日志甩给算法同学，排障链路极长。

一句话：精度看着高，落地却处处“玻璃心”。

技术路线对比：规则、模型、混合，谁更适合扛流量？

我们把三种方案放在同一批 20 万条线上日志里跑了 7 天，核心指标如下：

方案	意图准确率	P99 延迟	维护人日/月	备注
纯规则引擎	92.3%	18 ms	0.5	规则>7000 条后冲突爆炸
纯 NLP（BERT base）	96.1%	56 ms	2	长尾漂移需周级重训
混合架构（BERT + 规则）	98.0%	29 ms	1	规则热加载，模型半年一训

结论：混合架构用规则“兜住”业务变更，用模型“吃掉”泛化语义，整体 ROI 最高。

架构总览：双层过滤，一个流程图就能讲清

用户 query 进来后，先过 BERT 意图分类，拿到 Top-3 候选；再把候选丢进 Drools 规则引擎，由业务规则做“二次裁判”；最终输出唯一意图并返回答案。

文字流程图如下：

用户输入 → 预处理（分词、归一化） → BERT 意图 Top-3 → Drools 规则过滤 → 返回意图/答案

核心实现：Python 代码直接搬

1. BERT 意图分类层

安装依赖：

pip install transformers==4.30.0 torch1.11 onnxruntime-gpu

代码（带类型标注，GPU/CPU 自动回退）：

# intent_model.py from transformers import BertTokenizer, BertForSequenceClassification import torch from typing import List, Tuple class BertIntent: def __init__(self, model_dir: str, device: str = None): self.tokenizer = BertTokenizer.from_pretrained(model_dir) self.model = BertForSequenceClassification.from_pretrained(model_dir) self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") self.model.to(self.device) self.model.eval() @torch.no_grad() def predict_topk(self, text: str, k: int = 3) -> List[Tuple[str, float]]: inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=64) inputs = {k: v.to(self.device) for k, v in inputs.items()} logits = self.model(**inputs).logits[0] probs = torch.softmax(logits, dim=-1) topk = torch.topk(probs, k) id2label = self.model.config.id2label return [(id2label[i], s.item()) for i, s in zip(topk.indices, topk.values)]

2. Drools 规则引擎层

Python 侧通过durable-rules做轻量包装，规则文件放在rules/目录，支持动态热加载。

规则示例rules/discount.drl：

rule "618大促折扣" when $m: IntentCandidate(label == "discount", score > 0.85) User(tags contains "vip") then $m.setPass(true); end

Python 封装：

# rule_engine.py import os, json, typing as t from durable.lang import ruleset, rule, m, post class RuleEngine: def __init__(self, drl_path: str): self.rs = ruleset("intent_filter") self._load_drl(drl_path) def _load_drl(self, path: str): # 简化：把 drl 转成 json 再注册 with open(path, encoding="utf8") as f: rule_json = json.load(f) for r in rule_json: rule(self.rs, r["name"], **r["cond"])(lambda c: self._set_pass(c, r["action"])) def _set_pass(self, c, action: dict): c.m.passed = action.get("pass", False) def filter(self, candidates: t.List[t.Dict], user: t.Dict) -> t.Dict: facts = [{"label": c["label"], "score": c["score"], "user": user}] post(self.rs, "intent_filter", facts) # 返回第一个 passed 的候选 return next((c for c in candidates if c.get("passed")), candidates[0])

3. 把两层串起来

# main.py from intent_model import BertIntent from rule_engine import RuleEngine class HybridBot: def __性能优化：让 29 ms 延迟再砍一半 1. 模型量化：把 BERT base 转 ONNX INT8，体积 380 M → 110 M，GPU 延迟 29 ms → 17 ms。 ```bash python -m transformers.onnx --model=./bert_intent --quantize --opset=13 intent.onnx