背景痛点：大促凌晨的“客服雪崩”
去年双11，我们团队守着监控大屏，眼睁睁看着客服接口 RT 从 200 ms 飙到 4 s，队列里 3 w+ 消息在“排队跳楼”。
传统规则引擎（if-else 树 + 正则词典）在并发一上来就原形毕露：
- 上下文靠 Redis String 粗暴拼接，多轮对话一半丢失；
- 意图规则 1 w+ 条，加载一次 7 s，改一条规则要全量重启；
- 高峰期 CPU 占满，只因每条消息都要遍历整棵决策树。
痛定思痛，我们决定把“扣子物客服智能体”搬上战场，用 AI 模型 + 微服务架构重新洗牌。
技术选型：规则引擎 vs AI 智能体
先放一张对比表，数据来自去年 Q4 的 A/B 期真实流量：
| 维度 | 规则引擎 | RPA 脚本 | 扣子物 AI 智能体 |
|---|---|---|---|
| 峰值 QPS | 800 | 500 | 3500 |
| Top-1 意图准确率 | 78 % | 72 % | 93 % |
| 新意图上线 | 2 d | 1 d | 30 min（热更新） |
| 维护人日/月 | 18 | 15 | 4 |
| 硬件成本（32C128G） | 12 台 | 18 台 | 4 台 + 2 张 T4 |
结论很现实：
- 规则引擎适合冷启动，但“边际效应”为负——规则越多，冲突越多；
- RPA 只能做“点击搬运”，无法处理多轮语义；
- AI 智能体一次性投入高，却在并发、准确率、迭代速度上全面碾压。
架构设计：三层管线，让 NLU→DM→KG 各司其职
- 接入层：统一网关做租户路由、鉴权、脱敏。
- NLU 引擎：
- 意图分类（BERT+FC）
- 槽填充（BiLSTM+CRF）
- 对话管理（DM）：
- 状态机（见下节代码）
- 策略中心（DQN 选槽/反问/转人工）
- 知识图谱：商品属性、活动规则、售后 SOP 三元组，支持毫秒级子图查询。
- 基础设施：
- 微服务：K8s + Istio，单服务灰度；
- 缓存：Redis+Kafka 做事件溯源；
- 监控：Prometheus + Grafana，核心指标——P99 延迟、意图置信度分布。
核心实现
1. 对话状态机（Python 3.11）
状态机要解决“刷新页面对话丢”的顽疾，必须把状态落到 DB，同时保证高并发读写。
# state_machine.py import json from enum import Enum, auto from datetime import datetime from sqlalchemy import Column, String, DateTime, Text from sqlalchemy.orm import declarative_base Base = declarative_base() class State(Enum): INIT = auto() AWAIT_ITEM = auto() AWAIT_SIZE = auto() CONFIRM_ORDER = auto() END = auto() class DialogState(Base): __tablename__ = "dialog_state" session_id = Column(String(64), primary_key=True) state = Column(String(20), default="INIT") slots = Column(Text, default="{}") # JSON 字符串 updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class StateMachine: """时间复杂度：O(1) 状态转移；空间：O(1) 每会话""" def __init__(self, repo): self.repo = repo # DAO 对象，封装 DB 读写 def transit(self, session_id: str, intent: str, slots: dict): row = self.repo.get(session_id) or DialogState(session_id=session_id) cur = State[row.state] new_slots = {**json.loads(row.slots), **slots} # 简单示例：状态转移表 table = { State.INIT: { "ask_item": State.AWAIT_ITEM, "greeting": State.INIT }, State.AWAIT_ITEM: { "provide_item": State.AWAIT_SIZE }, State.AWAIT_SIZE: { "provide_size": State.CONFIRM_ORDER } } nxt = table.get(cur{}).get(intent, cur) row.state = nxt.name row.slots = json.dumps(new_slots, ensure_ascii=False) self.repo.save(row) return nxt, new_slots要点
- 状态与槽位同表，单行写保证原子；
- 对外暴露
transit()无锁，DB 层用ON CONFLICT UPDATE做 Upsert； - 灰度发布时，状态枚举新增字段可向后兼容。
2. 意图识别优化：BERT Fine-tuning
预训练模型我们选bert-base-chinese，训练集 12 w 条客服语料，平均长度 28 token。
# bert_intent.py from datasets import load_dataset from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments tokenizer = BertTokenizer.from_pretrained("bert-base-chinese") model = BertForSequenceClassification.from_pretrained("bert-base-chinese", num_labels=47) def encode(examples): return tokenizer(examples["text"], truncation=True, padding="max_length", max_length=32) train_ds = load_dataset("csv", data_files="intent_train.csv", split="train") train_ds = train_ds.map(encode, batched=True) args = TrainingArguments( output_dir="./bert_intent", per_device_train_batch_size=128, learning_rate=2e-5, num_train_epochs=3, fp16=True, logging_steps=100, save_total_limit=2, load_best_model_at_end=True, metric_for_best_model="accuracy", ) trainer = Trainer(model=model, args=args, train_dataset=train_ds, tokenizer=tokenizer) trainer.train()tricks
- 32 token 截断，经统计可覆盖 97 % 句子，减少 30 % 计算量；
- FP16 + 动态批填充，训练时间从 4 h 降到 1.5 h；
- 推理阶段用
TorchScript导出，CPU 延迟 18 ms→7 ms。
生产考量
1. 压测曲线
| 并发 | 平均 RT | P99 RT | CPU 占用 |
|---|---|---|---|
| 500 | 28 ms | 45 ms | 35 % |
| 1000 | 32 ms | 55 ms | 55 % |
| 2000 | 41 ms | 78 ms | 78 % |
| 3000 | 120 ms | 250 ms | 92 % |
拐点在 2500 QPS 左右，后面 RT 陡升，原因是 Kafka 线程池打满，批量改同步导致毛刺。
2. 熔断策略（Sentinel 1.8）
# flow-rule.yaml rules: - resource: "nlu_predict" grade: QPS count: 2500 strategy: 0 # 直接拒绝 controlBehavior: 0 maxQueueingTimeMs: 0 - resource: "kg_query" grade: RT count: 100 # 单位 ms timeWindow: 5 minRequestAmount: 50效果：
- 超过 2500 QPS 直接抛
BlockException，前端降级到“人工客服排队”页面； - KG 查询 RT 突刺 100 ms 以上，连续 5 次即熔断 5 s，保护图数据库不被拖化。
避坑指南
冷启动语料预处理
- 先把历史会话做“滑窗”清洗，去掉客服敏感信息；
- 用 TF-IDF + 聚类去重，相似度 >0.9 的句子只留 1 条，减少 40 % 标注量；
- 低资源场景下，用
bert-mini蒸馏，准确率掉 1.2 %，推理提速 2.7 倍。
多租户资源隔离
- 网关层按租户 ID 打 Tag，K8s 侧用
ResourceQuota限制 CPU/Mem； - 模型侧共享 GPU，但通过
cuda-mps按比例切分显存，防止大租户独占； - 日志索引按租户拆分，避免 ES 热点写爆。
- 网关层按租户 ID 打 Tag，K8s 侧用
写在最后
扣子物客服智能体上线三个月，累计拦截 82 % 重复咨询，人工坐席缩减 35 %，P99 延迟稳定在 80 ms 以内。
但新问题也来了：当业务想把 47 个意图扩展到 200+ 时，BERT 的推理延迟开始线性增长。
如何平衡模型精度与推理延迟？
是继续蒸馏？还是上双塔架构（粗排+精排）？欢迎留言一起拆坑。
