痛点分析:传统客服到底卡在哪?
刚接手客服系统改造时,我最大的感受是“慢”和“笨”。
- 响应速度:高峰期用户排队 30 秒以上,人工坐席一满,新咨询直接 404。
- 多轮对话:用户问“我订单到哪了→还要多久→能改地址吗”,传统关键词机器人只能逐条匹配,上下文一断就答非所问。
- 扩展性:每上新业务就要加一堆“if-else”,脚本膨胀到几千行,维护成本直线上升。
这三座大山,让我下定决心用 AI 方案重新造轮子。
技术选型:Rasa、Dialogflow、Lex 怎么挑?
我把自己踩过的坑做成一张对比表,先看数据:
| 维度 | Rasa(开源) | Dialogflow(谷歌) | Lex(AWS) | |----| 开源免费 | 按请求收费 | 按请求+语音时长 | | 训练数据 | 自己标,100 条也能起步 | 官方推荐 200+ 意图 | 同左 | | 多语言 | 社区模型,中文需自己训 | 官方支持中文 | 中文支持一般 | | 私有化 | 完全可行 | 不行 | 不行 | | 学习曲线 | 需懂 Python | 控制台拖拽即可 | 同左 |
结论:
- 想省钱、数据敏感、爱折腾 → Rasa
- 快速验证、无运维人力 → Dialogflow
- 已用 AWS 全家桶 → Lex
我最终选了 Rasa,因为老板一句“别把我用户数据放别人家服务器”。
核心实现一:意图分类最小可用模型
先别上 BERT,用传统机器学习就能跑通 MVP。下面这段 80 行代码,从 0 到意图识别只需 5 分钟。
# intent_train.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline import joblib, json, os from typing import List, Tuple def load_data(path: str) -> Tuple[List[str], List[str]]: """加载 JSONL:{"text":"查询订单","intent":"order_query"}""" texts, labels = [], [] with open(path, encoding="utf-8") as f: for line in f: item = json.loads(line) texts.append(item["text"]) labels.append(item["intent"]) return texts, labels def build_model() -> Pipeline: """TF-IDF + 逻辑回归,中文先简单用字符级""" return Pipeline([ ("tfidf", TfidfVectorizer(tokenizer=lambda s: list(s), ngram_range=(1,2))), ("clf", LogisticRegression(max_iter=1000)) ]) if __name__ == "__main__": # 1. 读数据 X, y = load_data("data/nlu.jsonl") # 2. 训练 model = build_model() model.fit(X, y) # 3. 保存 os.makedirs("models", exist_ok=True) joblib.dump(model, "models/intent.pkl") print("意图模型训练完成,本地 models/intent.pkl")跑通后,在终端输入python intent_train.py,看到准确率 0.92 就可以继续下一步。
核心实现二:对话状态机 JSON 示例
Rasa 的故事(story)文件本质是状态机,我把它简化成纯 JSON,方便前端同学一眼看懂:
{ "story": "订单查询流", "steps": [ {"intent": "order_query", "entities": cigar"}, {"action": "action_check_order"}, {"intent": "ask_delivery_time"}, {"action": "action_predict_eta"}, {"intent": "affirm", "action": "utter_anything_else"}, {"intent": "deny", "action": "utter_goodbye"} ] }把这段保存成stories.json,用 Rasa SDK 写对应的action_xxx.py就能跑通多轮。
生产考量:高并发与数据安全
- 异步 IO
官方 rasa server 默认同步,我用sanic做一层网关,把核心代码包成异步函数,QPS 从 80 提到 400。
# async_gateway.py from sanic import Sanic, response import joblib, asyncio model = joblib.load("models/intent.pkl") async def predict(text: str) -> str: # 模型预测 CPU 密集,放线程池防止阻塞 loop = asyncio.get_event_loop() return await loop.run_in_executor(None, model.predict, [text]) app = Sanic("bot") @app.post("/intent") async def intent(request): text = request.json.get("text", "") if not text: return response.json({"error": "text 为空"}, 400) try: label = await predict(text) return response.json({"intent": label[0]}) except Exception as e: return response.json({"error": str(e)}, 500) if __name__ == "__main__": app.run(host="0.0.0.0", port=8000)- 敏感词 & 加密
- 敏感词:把 2 万条词库放内存,用 DFA 构建前缀树,单次过滤 <1 ms。
- 加密:用户手机号、地址走 AES-256-CBC,密钥放 K8s Secret,定期轮转。
避坑指南:血泪经验 4 连击
- 过度迷信预训练模型
我直接拿中文 BERT 微调,结果业务语料只有 1 万句,过拟合到 99%,上线后意图准确率掉成 67%。后来回退 TF-IDF+LR,数据增强后再上 BERT,效果才稳。 - 对话超时
默认 300 秒 session 失效,用户去泡咖啡回来继续聊,bot 已失忆。我把 Redis 存 session 改成 30 分钟,并加“还在吗?”的兜底询问,超时自动 summary 写回数据库,用户体验好很多。 - 上下文丢失
多节点部署时 sticky session 没开,第二次请求打到另一台 pod,tracker 为空。用 Redis 共享 tracker 后解决。 - 日志打爆磁盘
开 debug 忘了关,一晚 80 G。现在只在出错时打印完整 feature,其余采样 1%,并加定时清理脚本。
代码规范小结
- 所有函数写类型注解,新人接手一眼看懂输入输出。
- 任何可能抛异常的地方都 try/except,并返回统一格式 `{"error": "可读中文"}。
- 关键步骤写中文注释,方便后来同事 grep 定位。
- 单元测试覆盖意图、实体、action 三大模块,CI 不通过禁止合并。
延伸思考:知识图谱让问答更聪明
当用户问“iPhone 14 和 13 的屏幕差异”时,纯意图匹配只能丢文档链接。如果把商品参数做成图谱,就能精准返回“14 用超瓷晶面板,13 用普通玻璃”,体验升级。
下一篇我准备写“如何把产品手册自动抽三元组,写入 Neo4j,再让 Rasa 调用 cypher 查询”,欢迎蹲更。
写在最后
整套流程跑下来,我最深的体会是:别急着上“高大上”模型,先把数据、流程、监控这些脏活累活干扎实,AI 客服才能真正省人力而不是添乱。希望这份避坑笔记,能让你少熬几次夜。祝部署顺利,有坑一起填!