背景：多轮对话的“三座大山”
做智能客服最怕什么？最怕用户说“我刚才不是已经告诉你了吗”。
�上下文丢失、意图漂移、槽位反复确认，这三座大山把多少项目卡死在上线前夜。
- 上下文丢失：HTTP无状态，每次请求都是“陌生人”，对话历史存在、取回、拼接一步错步步错。
- 意图漂移：用户中途插一句“那运费多少”，模型立刻把话题带偏，结果前面填好的订单全作废。
- 槽位反复确认：地址、手机号来回问，用户体验直接负分。
技术选型：规则、状态机、深度学习怎么挑
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 规则引擎 | 上线快、可解释 | 硬编码爆炸、难维护 | 活动临时客服、FAQ |
| 有限状态机（FSM） | 结构清晰、易调试、可图形化 | 状态爆炸、难表达并行 | 订单、工单、退换货等线性流程 |
| 深度端到端 | 泛化强、体验拟人 | 数据饥渴、黑盒、难排错 | 闲聊、开放域 |
一句话总结：先把“能固化的业务”用FSM兜住，再让模型在FSM的边界外“自由发挥”。
我们最终选型：Python 3.11 + FSM（transitions库）+ BERT意图分类器，兼顾可控与泛化。
核心实现：FSM + 上下文仓库
1. 状态机骨架
# chatflow/fsm.py from transitions import Machine import json, time, hashlib class ChatContext: # 充当内存数据库 def __init__(self, uid): self.uid = uid self.slot = {} # 槽位 self.turns = [] # 原始对话 self.ts = int(time.time()) class OrderBot(Machine): states = [ 'start', # 初始 'await_phone', # 等手机号 'await_addr', # 等地址 'confirm', # 确认订单 'end' # 结束 ] def __init__(self, ctx: ChatContext): self.ctx = ctx super().__init__( states=OrderBot.states, initial='start', auto_transitions=False, send_event=True ) self._add_transitions() def _add_transitions(self): # 触发器、源、目标、条件、before/after回调 self.add_transition('trigger_phone', 'start', 'await_phone', conditions=[self._need_phone]) self.add_transition('submit_phone', 'await_phone', 'await_addr', conditions=[self._valid_phone], before=[self._save_phone]) self.add_transition('submit_addr', 'await_addr', 'confirm', before=[self._save_addr]) self.add_transition('confirm_yes', 'confirm', 'end', after=[self._submit_order]) self.add_transition('confirm_no', 'confirm', 'await_addr') # ---------- 条件函数 ---------- def _need_phone(self, event): # event.data 里是本轮用户原文 return 'phone' not in self.ctx.slot def _valid_phone(self, event): phone = event.data.get('phone') return bool(re.fullmatch(r'1[3-9]\d{9}', phone or '')) # ---------- 动作函数 ---------- def _save_phone(self, event): self.ctx.slot['phone'] = event.data['phone'] def _save_addr(self, event): self.ctx.slot['addr'] = event.data['addr']2. 上下文仓库（线程安全）
# chatflow/repo.py from threading import Lock from collections import OrderedDict class ContextRepo: def __init__(self, max_size=10000, ttl=1800): self._data = OrderedDict() self.lock = Lock() self.ttl = ttl self.max_size = max_size def get_or_create(self, uid: str) -> ChatContext: with self.lock: now = int(time.time()) # LRU + TTL 双清 while self._data and (now - next(iter(self._data.values())).ts > self.ttl or len(self._data) >= self.max_size): self._data.popitem(last=False) if uid not in self._data: self._data[uid] = ChatContext(uid) ctx = self._data[uid] ctx.ts = now self._data.move_to_end(uid) return ctx3. 意图识别与FSM的“握手”
# chatflow/nlu.py from transformers import pipeline cls = pipeline('text-classification', model='bert-base-chinese', top_k=None) def predict_intent(text: str) -> dict: # 返回 {'label': 'inform_phone', 'score': 0.92} return cls(text)[0] # chatflow/main.py def handle_message(uid: str, text: str) -> str: ctx = repo.get_or_create(uid) bot = OrderBot(ctx) intent = predict_intent(text)['label'] if intent == 'inform_phone' and bot.state == 'await_phone': bot.submit_phone({'phone': extract_phone(text)}) elif intent == 'inform_addr' and bot.state == 'await_addr': bot.submit_addr({'addr': text}) elif intent == 'affirm': bot.confirm_yes() else: return fallback_answer(text) # 兜底回复 return next_prompt(bot.state)要点：
- 意图只负责“把用户话翻译成事件”，状态机决定“要不要收”。
- 事件数据用正则/实体抽取提前洗净，FSM里只做校验，不耦合NLP。
性能优化：并发与长对话
- 进程模型：uvicorn + gunicorn gevent，worker数=CPU*2，I/O耗时（模型推理）全走异步。
的 - 上下文冷热分层：
- 热数据放内存（ContextRepo）。
- 冷数据每5分钟异步刷到Redis，重启可恢复。
- 长对话裁剪：turns>30轮时，仅保留系统槽位与最近5轮用户原句，压缩80%内存。
- 状态机缓存：OrderBot实例化代价低，但把“transition”对象预生成单例，QPS 1k→3k。
避坑指南：生产踩出来的血书
- 槽位字段被运营改名字：把“addr”改成“address”，结果条件函数全挂。
→ 方案：槽位用枚举常量，配置中心统一收口，发版前diff。 - 状态图循环引用：confirm_no回到await_addr，用户反复点“修改”，栈溢出。
→ 方案：给transition加计数器，单会话最多3次循环，第四次转人工。 - 意图模型热更新：新模型把“好的”判成affirm，老模型是neutral，灰度时用户懵。
→ 方案：意图版本号写进日志，支持按UID百分比灰度，可秒级回滚。 - 高并发下OrderedDict线程锁成为瓶颈：
→ 方案：分片锁（uid尾号%16），冲突概率下降90%。 - 敏感词日志泄露：手机号、地址被打到ELK。
→ 方案：日志脱敏器，正则替换中间4位，落盘前再审计。
安全：敏感信息的三板斧
- 传输：全程TLS1.3，HSTS强制。
- 存储：槽位落Redis全走AES-256-GCM，密钥放KMS，轮换周期90天。
- 审计：任何客服人工拉取原始对话需二次授权，日志留痕6个月自动粉碎。
下一步：多语言怎么扩展
- 状态机与语言无关，复用同一套图。
- 意图模型做多语言fine-tune，或lang-id→路由到单语模型。
- 槽位正则、实体词典按语言分包，启动时懒加载。
- 右向文字（阿拉伯、希伯来）前端镜像，状态机里加字段
rtl=True，提示模板自动翻转。
写在结尾
把FSM当“铁轨”，模型当“司机”，让司机在轨道上跑，既不会冲出悬崖，也能欣赏沿途风景。
这套方案已在日活30W的电商客服跑稳，平均对话轮次从5.2降到2.7，意图准确率从81%提到93%。
如果你也在做多轮对话，不妨先画状态图，再写条件函数，把复杂留给自己，把简单留给用户。
下一步想试试把强化inter的“改写”能力融进来，让机器自己补全用户省略的槽位——路还长，慢慢聊。
