人工客服智能体工作流：从零搭建高可用对话系统的实战指南-编程实验室

背景痛点：传统客服系统到底卡在哪？

刚接手客服项目时，我以为“能跑就行”，结果上线第一天就被用户吐槽“答非所问”。总结下来，传统脚本式客服有三座大山：

意图识别准确率/Intent Recognition Accuracy 低：关键词+正则的组合，用户换个说法就翻车，比如“我钱包丢了”和“卡片遗失”被当成两件事。
多轮对话状态维护/Dialogue State Tracking 困难：HTTP 无状态，每次请求都得把前面所有信息再传一遍，字段一多就爆炸。
异常流程处理缺失：用户突然说“等等，我接个电话”，系统继续追问“请输入验证码”，体验瞬间归零。

痛点明确后，目标就一句话：让机器像人一样“记得住、听得懂、不跑偏”。

架构设计：FSM+规则引擎双轮驱动

我最后选的是“有限状态机（Finite State Machine/FSM）+轻量规则引擎”的混合架构：FSM 管状态流转，规则引擎管细粒度判断，两者互补，复杂度可控。

核心状态只有 5 个：

greeting
collecting_info
confirming
handover（转人工）
end

状态跳转图如下：

规则引擎侧只干两件事：

识别置信度低于阈值时，直接回落到规则兜底；
命中敏感词或高危意图时，强制切入 handover 状态。

这样就算 NLU 模型抽风，也不至于把用户带沟里。

核心实现一：Rasa NLU 让意图识别稳一点

下面给出最小可运行片段，已按 PEP8 排版，关键函数带类型注解与异常捕获。

# nlu_engine.py from typing import Dict, Any import asyncio from rasa.nlu.model import Interpreter import logging logging.basicConfig(level=logging.INFO) log = logging.getLogger(__name__) class RasaNLUEngine: def __init__(self, model_path: str) -> None: try: self.interpreter = Interpreter.load(model_path) log.info("Rasa NLU 模型加载完成") except Exception as e: log.exception("模型加载失败") raise RuntimeError("NLU 初始化异常") from e async def parse(self, text: str) -> Dict[str, Any]: """返回结构化意图与实体，带置信度""" if not text or not text.strip(): raise ValueError("输入文本为空") loop = asyncio.get_event_loop() # Rasa 同步接口，用线程池防止阻塞主事件循环 result = await loop.run_in_executor(None, self.interpreter.parse, text) intent: Dict[str, Any] = result["intent"] entities: list = result.get("entities", []) # 置信度低于 0.3 直接标为未知，防止误触发 if intent.get("confidence", 0.0) < 0.3: intent["name"] = "unknown" return { "intent": intent["name"], "confidence": intent["confidence"], "entities": entities, }

实体提取/Entity Extraction 结果直接丢给下游状态机，省得再解析一遍。

核心实现二：Redis 做对话上下文“备忘录”

状态机必须无状态，上下文全部丢到 Redis，TTL 自动清掉，省内存也省代码。

# context_repo.py import json import redis from typing import Optional, Dict, Any from datetime import timedelta class DialogueContextRepo: def __init__(self, redis_url: str = "redis://localhost:6379/0"): self.rdb = redis.from_url(redis_url, decode_responses=False) def _key(self, session_id: str) -> str: return f"ctx:{session_id}" def save(self, session_id: str, data: Dict[str, Any], ttl: int = 600) -> None: """序列化后写入，默认 10 分钟过期""" value = json.dumps(data, ensure_ascii=False) self.rdb.setex(self._key(session_id), timedelta(seconds=ttl), value) def load(self, session_id: str) -> Optional[Dict[str, Any]]: raw = self.rdb.get(self._key(session_id)) return json.loads(raw) if raw else None def renew_ttl(self, session_id: str, ttl: int = 600) -> None: """用户活跃时续期""" self.rdb.expire(self._key(session_id), timedelta(seconds=ttl))

序列化直接用 JSON，字段少于 2 KB 时性能差距可忽略；若后续量大可换成 MessagePack。

生产考量：让系统扛得住 9k QPS

对话超时机制：
在 Redis 键过期回调里抛TimeoutEvent，FSM 捕获后自动转入end状态，前端收到后弹提示“会话已结束”，避免用户对着空气说话。
敏感词过滤：
采用异步队列方案——用户消息先落库，立即返回“正在输入”，再由后台 Celery 任务做敏感扫描；命中则回滚状态到handover，全程无阻塞。
负载测试指标：
4C8G 容器*3，单实例 300 QPS，P99 延迟 450 ms，CPU 65%。瓶颈主要在 NLU 模型推理，后续可转 TensorRT 或远程 gRPC 推理池。