背景痛点：规则引擎的“三板斧”失灵了
做智能客服之前，我先用 if-else 写了一套“关键词+正则”应答逻辑，上线第一天就翻车：
- 冷启动没数据，运营同事一口气录了 200 条 FAQ，结果用户换种问法就匹配不到，命中率不到 30%。
- 多轮对话根本玩不转，用户说“我要改地址”，系统回“好的，请提供新地址”，结果用户接着说“算了，先查物流”，状态机直接懵圈，把“查物流”当成新地址存进库。
- 最尴尬的是并发一上来，全局变量乱窜，A 用户把 B 用户的订单号背走了，客服后台炸锅。
痛定思痛，决定用 Python 生态搞一套 AI 辅助的对话系统，把“意图识别”和“对话状态”这两座大山一起搬掉。
技术选型：Rasa + BERT 为何胜出
我拉了个对比表，把团队能 hold 住的技术栈都扔了进去：
| 方案 | 优点 | 缺点 | 结论 |
|---|---|---|---|
| Dialogflow | 谷歌全家桶，免运维 | 中文支持一般、按次收费、数据出境 | 被财务一票否决 |
| 自建 BERT 服务 | 精度高，完全可控 | 多轮状态机自己写，工期爆炸 | 当“枪”可以，当“炮”不行 |
| Rasa + BERT | 社区活跃、可本地部署、Python 源码级可控 | 需要自搭训练管线 | 折中后最香 |
一句话：Rasa 负责对话管理，BERT 负责语义理解，两者用 Redis 做上下文桥梁，既解耦又省 GPU 预算。
核心实现：30 行代码跑通“意图+状态”
1. 系统架构速览
- 用户消息 → Rasa Core → 调用自定义 NLU（BERT）→ 更新对话状态 → Redis → 返回回复
- 所有状态以
sender_id为 key，TTL 30 min，自动过期防膨胀。
2. Rasa Core 状态机
故事文件data/stories.yml片段：
version: "3.1" stories: - story: 修改地址 steps: - intent: request_change_address - action: utter_ask_new_address - intent: inform_address - action: action_change_address自定义 Action 示例（PEP8 + 类型注解）：
from typing import Any, Dict, List, Text from rasa_sdk import Action, Tracker from rasa_sdk.executor import Collectfully from rasa_sdk.events import SlotSet import redis class ActionChangeAddress(Action): """修改收货地址并清空缓存""" def __init__(self) -> None: self.r = redis.Redis(host="127.0.0.1", decode_responses=True) def name(self) -> Text: return "action_change_address" async def run( self, dispatcher: Collectfully, tracker: Tracker, domain: Dict[Text, Any], ) -> List[Dict[Text, Any]]: sender_id: Text = tracker.sender_id new_addr: Text = tracker.latest_message.get("text", "") # 这里调用业务 API 省略 self.r.hset(sender_id, "address", new_addr) dispatcher.utter_message(text="地址已更新～") return [SlotSet("address", new_addr)]3. BERT 意图分类组件
把 Rasa 的DIETClassifier换成轻量 BERT 服务，降低 40% 的 GPU 显存。
模型加载与预处理代码（含注释）：
# nlu/bert_intent.py from typing import List import torch, redis, json from transformers import BertTokenizer, BertForSequenceClassification class BertIntentPredictor: """轻量级意图预测器，线程隔离""" def __init__(self, model_dir: str, num_labels: int = 18): self.tokenizer = BertTokenizer.from_pretrained(model_dir) self.model = BertForSequenceClassification.from_pretrained(model_dir) self.model.eval() self.r = redis.Redis(decode_responses=True) def predict(self, text: str, sender_id: str) -> List[float]: """ 返回各意图概率分布 """ inputs = self.tokenizer( text, return_tensors="pt", truncation=True, max_length=64 ) with torch.no_grad(): logits = self.model(**inputs).logits probs = torch.softmax(logits, dim=-1).cpu().numpy().tolist()[0] # 缓存结果，供后面规则兜底 self.r.hset(sender_id, "last_intent_probs", json.dumps(probs)) return probs4. 对话上下文存储方案
- 使用 Redis Hash：
HSET <sender_id> intent xxx slot_xxx yyy - 设置 30 min TTL，Lua 脚本批量过期，防止内存爆炸。
- 关键：每个
sender_id对应独立 Hash，天然线程隔离，避免状态串台。
生产考量：并发、安全、灰度一个都不能少
1. 并发压测
Locust 脚本示例（单核 QPS 跑到 180 无报错）：
# locustfile.py from locust import HttpUser, task class ChatUser(HttpUser): @task def ask(self): self.client.post( "/webhooks/rest/webhook", json={"sender": "test_user", "message": "我要改地址"}, )启动：
locust -f locustfile.py -u 200 -r 20 --host=http://127.0.0.1:5005瓶颈最先卡在 BERT 推理，加一层 TorchServe + batch 推理，QPS 提到 420。
2. 敏感词与数据脱敏
- 敏感词树（DFA）一次性加载到内存，拦截 < 2 ms。
- 地址、手机号用正则脱敏：
(?P<phone>1[3-9]\d{9})→*** - 日志打印前统一走
json.dumps(sensitive_filter(data), ensure_ascii=False)，防止明文落盘。
避坑指南：线程隔离 + 模型热更新
1. 线程隔离
Rasa 默认asyncio，但自定义 Action 里如果访问全局变量极易踩坑。方案：
- 每个 Action 实例化新的 Redis 连接，禁止单例共享。
- 用
contextvars保存sender_id，确保日志链路不串话。
2. 模型热更新灰度
- 新模型放
models/bert_v2，旧模型保持models/bert_v1 - 在 Redis 写
model_version: v2，预测脚本读取该 key，按流量比例切换 - 灰度 10% 观察 30 min，准确率无下降再全量，回滚只需改一行配置
代码规范小结
- 全项目强制
black + isort，CI 阶段检查 PEP8 - 所有函数写
docstring，复杂参数加TypeDict或dataclass - 单元测试覆盖 NLU 组件 > 90%，核心 Action 用
pytest-asyncio测 Redis 超时场景
下一步思考：怎么让模型听懂方言？
目前 BERT 只在标准普通话语料上训练，遇到“粤普”或“川普”就抓瞎。留一道开放题：
如果让你设计一个支持方言的意图识别模块，你会如何收集语料、做数据增强，并在不增发 GPU 的前提下保证推理速度？欢迎留言聊聊你的思路，一起把智能客服做成真正的“智能”。
踩坑、填坑、再踩坑，这就是工程师的日常。。希望这份实战笔记能帮你少熬几个夜，早日上线不宕机的 Python 智能客服。祝你编码愉快，Bug 退散！
