背景痛点：传统客服的“答非所问”现场
去年做外包项目时，我接手过一套老客服系统：关键词匹配 + 正则硬编码，意图识别准确率不到 60%，用户多问一句“那怎么办”，机器人就原地打转。
痛点归纳起来就两条：
- 意图识别准确率差：没有语义泛化能力，换个说法就失效。
- 多轮对话无状态：每次请求都是“陌生人”，无法结合上文给出连贯回复。
于是老板拍桌子：两周内给我上“智能体”。本文就是我踩坑两周后，写给同样被拍桌子的你。
技术选型：Rasa、Dialogflow、LangChain 怎么挑
先把三兄弟拉出来遛一遛，结论先行：
- 要完全免费、可私有部署 → Rasa
- 想最快上线、不差钱 → Dialogflow
- 已准备拥抱 LLM、需要链式推理 → LangChain
| 维度 | Rasa 3.x | Dialogflow ES | LangChain |
|---|---|---|---|
| 意图识别 | 自带 DIET，可自定义 | 谷歌预训练，黑盒 | 靠 LLM Prompt，灵活 |
| 实体抽取 | 支持正则、CRF、DIET | 内置，但不可微调 | 需自己写 Parser |
| 对话管理 | 基于 Story & Rule，可写代码 | 可视化流程图 | 用 Chain 硬编码 |
| 数据隐私 | 本地训练，完全私有 | 上传云端，合规需评估 | 取决于 LLM 提供商 |
| 中文体验 | 社区活跃，语料多 | 中文支持一般 | 依赖 LLM 本身能力 |
我最后选了 Rasa：免费、可改源码、中文社区活跃，下面所有代码均基于 Rasa 3.7 版本。
核心实现：30 分钟跑通最小可用系统
1. 环境初始化
python -m venv rasa-env source rasa-env/bin/activate pip install rasa==3.7 redis aioredis fastapi uvloop2. NLU 流水线配置（config.yml）
language: zh pipeline: - name: JiebaTokenizer - name: RegexFeaturizer - name: LexicalSyntacticFeaturizer - name: CountVectorsFeaturizer analyzer: char_wb min_ngram: 1 max_ngram: 4 - name: DIETClassifier epochs: 100 constrain_similarities: true - name: EntitySynonymMapper - name: ResponseSelector epochs: 1003. 最小训练数据（nlu.yml 片段）
nlu: - intent: greet examples: | - 你好 - 嗨 - intent: apply_refund examples: | - 我要退款 - 怎么退钱 - intent: deny examples: | - 不用了 - 算了4. 故事与规则（stories.yml & rules.yml）
# rules.yml rules: - rule: 退款流程 steps: - intent: apply_refund - action: utter_ask_order# stories.yml stories: - story: 退款多轮 steps: - intent: apply_refund - action: utter_ask_order - intent: inform entities: - order_id: "123456" - action: action_apply_refund5. 自定义 Action（Python 端）
# actions.py from typing import Any, Dict, List, Text from rasa_sdk import Action, Tracker from rasa_sdk.executor import CollectingDispatcher import aioredis, json, os REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/1") class ActionApplyRefund(Action): def name(self) -> Text: return "action_apply_refund" async def run( self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any], ) -> List[Dict[Text, Any]]: order_id = next(tracker.get_latest_entity_values("order_id"), None) if not order_id: dispatcher.utter_message("没查到订单号，请检查") return [] # 持久化状态 redis = aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True) await redis.hset(f"refund:{tracker.sender_id}", mapping={ "order_id": order_id, "status": "processing" }) dispatcher.utter_message(f"订单 {order_id} 退款已受理，预计 1-3 个工作日") return []6. 启动顺序
rasa train rasa run --actions actions --port 5055 --cors "*" rasa shell # 另一个终端，验证多轮生产级加固：别让机器人“掉线”
1. 对话状态持久化（Redis 连接池）
# redis_pool.py import aioredis from typing import Optional class RedisPool: _pool: Optional[aioredis.ConnectionPool] = None @classmethod async def get_conn(cls): if cls._pool is None: cls._pool = aioredis.ConnectionPool.from_url( "redis://localhost:6379/1", max_connections=20 ) return aioredis.Redis(connection_pool=cls._pool) # 使用处 redis = await RedisPool.get_conn() await redis.expire(f"session:{sender_id}", 3600) # 1h 过期2. 异步消息队列（asyncio + asyncio.Queue）
# queue_worker.py import asyncio, time QUEUE: asyncio.Queue = asyncio.Queue(maxsize=1000) async def producer(message: dict): await QUEUE.put(message) async def consumer(): while True: msg = await QUEUE.get() await handle_message(msg) # 你的业务协程 QUEUE.task_done() async def handle_message(msg: dict) -> None: await asyncio.sleep(0.01) # 模拟 IO3. 对话超时机制
# timeout.py import time SESSION_TTL = 300 # 5 分钟 async def is_timeout(sender_id: str) -> bool: redis = await RedisPool.get_conn() last = await redis.get(f"last_ts:{sender_id}") if last and time.time() - float(last) > SESSION_TTL: return True await redis.set(f"last_ts:{sender_id}", time.time(), ex=SESSION_TTL) return False4. 敏感词过滤（正则优化）
# sensitive.py import re PATTERN = re.compile( r"(?:退款|发票|人工)", # 把业务词放白名单，其余命中则** re.IGNORECASE | re.UNICODE, ) def mask_sensitive(text: str) -> str: return PATTERN.sub("***", text)5. 负载测试指标
本地 4 核 8 G 笔记本 + Redis，单进程压测结果：
- QPS ≈ 120
- P99 延迟 220 ms
- CPU 占用 65 %
若上 K8s，多副本横向扩展即可。
避坑指南：数据、中断、熔断
1. 训练数据偏差
- 负样本必须给：把“不属于任何意图”的语料单独标成
out_of_scope，占比 ≥ 15 %。 - 时间衰减采样：对高频出现但无意义的词（“嗯”、“好的”）降采样，防止模型偷懒。
2. 对话中断恢复 3 策略
- 本地缓存：浏览器端存
session_id，刷新后带回来。 - Redis 续命：前端心跳包每 30 s 调一次
/keepalive延长 TTL。 - 兜底引导：超时后机器人先道歉，再提供“快捷按钮”让用户一键回到主菜单。
3. 第三方 API 熔断
# circuit.py from typing import Callable, Any import asyncio, time class CircuitBreaker: def __init__(self, fail_max: int = 5, timeout: int = 60): self.fail_max = fail_max self.timeout = timeout self.fail_cnt = 0 self.last_fail = 0.0 async def call(self, func: Callable[..., Any], *args, **kw): if self.fail_cnt >= self.fail_max: if time.time() - self.last_fail < self.timeout: raise RuntimeError("Circuit open") self.fail_cnt = 0 try: res = await func(*args, **kw) self.fail_cnt = 0 return res except Exception as e: self.fail_cnt += 1 self.last_fail = time.time() raise e代码规范小结
- 全项目强制
mypy --strict过检，类型注解覆盖率 100 %。 - 关键算法时间复杂度：
- DIET 内部 transformer 推理 O(n²d) ，n=token 数，d=hidden。
- Redis 读写 O(1) ，队列插入 O(1) 。
- 异常处理：所有
await包try...except并写日志，防止事件循环崩溃。
延伸思考：LLM 增强方案
- 意图召回：先用小模型（DIET）做粗排，Top-3 候选再送 LLM 做精排，延迟可接受。
- 多轮改写：把历史 3 轮对话拼成一段“上下文”，让 LLM 生成“用户真实诉求”再喂给 Rasa，解决口语省略。
- 知识问答：把 FAQ 向量化存 Milvus，用户问题 embedding 后检索 Top-5，让 LLM 按检索结果生成答案， fallback 回 Rasa 故事流。
写在最后
整套流程跑下来，我最大的感受是：别把“智能”全押在模型上，工程细节（状态、超时、熔断）才是生产环境不哭的关键。
如果你也在被老板催着上线客服机器人，希望这份避坑笔记能让你少熬两个夜。下一步，我准备把 LLM 召回模块做成可插拔组件，等踩完新坑再来汇报。祝编码愉快，机器人不再已读乱回。
