背景痛点：传统客服的“三座大山”
去年第一次接手客服自动化项目时，老板只给了两句话：“把重复问题干掉，把夜班人力砍半。”
干之前，得先搞清楚旧系统到底卡在哪：
- 响应速度：人工坐席平均首响 45 秒，高峰期 2 分钟起步，用户直接暴走。
- 人力成本：三线轮班，20 人/班，光夜班补贴一年就烧掉 60 万。
- 7×24 服务：节假日排班永远凑不齐人，凌晨 3 点电话一响，值班同事睡眼惺忪，答非所问，投诉率飙升。
一句话，传统客服是“人拉肩扛”，RPA 智能客服要的是“机器人不下班”。可真正动手搭建才发现，坑比想象的多：对话流程一复杂就绕成麻花；微信、网页、App 各玩各的协议；异常没兜底，机器人一句“我没听懂”就被用户截图挂微博。本文记录我用 Python + Dialogflow 从零撸出一套可落地的轻量化方案，踩过的坑都摊开，能抄就抄。
技术选型：NLU 框架“三选一”
先给 NLU（自然语言理解）引擎打样，我对比了市面上最主流的三家：
| 维度 | Dialogflow (Google) | Lex (AWS) | Rasa (开源) |
|---|---|---|---|
| 意图识别准确率（中文） | 92% | 89% | 87% |
| 多语言支持 | 120+ 语言 | 英语、日语、西班牙语 | 全语言，需自训 |
| 本地化部署 | 不支持，必须走外网 | 不支持 | 完全本地 |
| 费用 | 月 1000 次免费，超出 0.002 美元/次 | 按语音时长计费，贵 | 免费 |
| 开发效率 | 5 分钟出 Demo | 与 AWS 生态绑定，配置多 | 训练、标注、调参全套自己干 |
结论：公司既要快，又没钱买 GPU，Dialogflow 是最折中的选择；如果数据必须留在内网，再考虑 Rasa。下文代码全部基于 Dialogflow ES 版，Python 客户端库google-cloud-dialogflow==2.28.0。
核心实现：三板斧搞定机器人骨架
1. 对话状态机：让机器人“有记忆脑袋”
Dialogflow 自带上下文，可一旦多轮对话跨接口，它就“失忆”。我用 Python 自己包一层状态机，把会话状态抢回本地。
# state_machine.py import time from enum import Enum, auto class State(Enum): IDLE = auto() # 刚接入 ASK_NAME = auto() # 问姓名 ASK_ORDER = auto() # 问订单号 CONFIRM = auto() # 确认信息 END = auto() # 结束 class Session: def __init__(self, uid: str, timeout: int = 300): self.uid = uid self.state = State.IDLE self.data = {} # 放槽位数据 self.last_active = time.time() self.timeout = timeout def is_expired(self) -> bool: return time.time() - self.last_active > self.timeout def update(self, state: State, key: str = None, value: str = None): self.state = state self.last_active = time.time() if key: self.data[key] = value状态转移逻辑：
def next_state(session: Session, intent: str, params: dict) -> str: """根据意图和当前状态决定下一步，返回机器人回复文本""" if session.is_expired(): session.update(State.IDLE) return "会话已超时，请重新提问" if session.state == State.IDLE: if intent == "greeting": session.update(State.ASK_NAME) return "你好，请问怎么称呼？" # 其他意图略... if session.state == State.ASK_NAME: session.update(State.ASK_ORDER, "name", params.get("name")) return f"收到，{params.get('name')}，请提供订单号" if session.state == State.ASK_ORDER: if len(params.get("order_id", "")) < 8: return "订单号不足 8 位，请重新输入" session.update(State.CONFIRM, "order_id", params.get("order_id")) return f"订单 {params.get('order_id')} 对吗？回复“确认”提交" if session.state == State.CONFIRM and intent == "affirm": session.update(State.END) return "已提交，稍后为您处理，感谢使用" # 兜底 return "没听懂，请再说一遍"超时处理：每次收到用户消息先is_expired()判断，超时直接拉回IDLE，避免僵尸会话占内存。
2. 消息中间件：微信、网页端一把梭
为了让机器人在微信和网页同时上岗，我做了个极轻量的 Webhook，统一收消息，再分渠道回包。
# webhook.py from flask import Flask, request, jsonify import hashlib, xml.etree.ElementTree as ET from state_machine import Session, next_state import dialogflow app = Flask(__name__) redis = get_redis() # 后面给出 @app.route("/wechat", methods=["GET", "POST"]) def wechat_entry(): # 微信签名校验 if request.method == "GET": return request.args.get("echostr", "") # 解析 XML xml = ET.fromstring(request.data) uid = xml.find("FromUserName").text msg = xml.find("Content").text # 查会话 session = get_session(uid) # 调 Dialogflow 识别意图 intent, params = detect_intent_text(uid, msg) reply = next_state(session, intent, params) # 回包 return f""" <xml> <ToUserName><![CDATA[{uid}]]></ToUserName> <FromUserName><![CDATA[gh_xxx]]></FromUserName> <CreateTime>{int(time.time())}</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[{reply}]]></Content> </xml> """ def detect_intent_text(uid: str, text: str) -> (str, dict): session_client = dialogflow.SessionsClient() session_path = session_client.session_path("项目ID", uid) text_input = dialogflow.TextInput(text=text, language_code="zh-CN") query = dialogflow.QueryInput(text=text_input) response = session_client.detect_intent(session=session_path, query_input=query) intent = response.query_result.intent.display_name params = {p.key: p.value for p in response.query_result.parameters} return intent, params网页端同理，把/web路由换成 JSON 入 JSON 出即可，核心代码复用 90%。
3. 会话持久化：Redis 结构这样搭
内存里蹦跶的Session对象，服务一重启就全灭。用 Redis 做持久化，结构足够简单：
key: session:<uid> value: json.dumps({"state": "ASK_NAME", "data": ..., "last_active": 123456}) expire: 300 秒存取封装：
import json, redis r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True) def get_session(uid: str) -> Session: raw = r.get(f"session:{uid}") if raw: d = json.loads(raw) s = Session(uid) s.state = State[d["state"]] s.data = d["data"] s.last_active = d["last_active"] return s return Session(uid) def save_session(session: Session): r.setex( f"session:{session.uid}", 300, json.dumps({ "state": session.state.name, "data": session.data, "last_active": session.last_active }) )300 秒过期，既省内存，又能自动清掉“幽灵会话”。
生产考量：让老板敢签字上线
1. 压力测试指标
我用 locust 在本机 4 核 8 G 笔记本跑出的数据：
- 单节点 Flask + gunicorn（4 workers）稳态并发 180 会话/秒
- 平均响应延迟 120 ms，P99 280 ms
- CPU 占用 65%，内存 450 M
再往上顶，瓶颈卡在 Dialogflow 的 600 QPM 配额；想冲 1000+ 并发，得做配额扩容或切 Rasa 本地。
2. 敏感词过滤 & 审计日志
客服场景最怕“说错话”。我加了两道闸：
- 敏感词 Trie 树过滤：维护 3000 条敏感词，每次用户消息先过树，命中直接返回“亲亲，咱们换个说法吧”。
- 审计日志：把用户原句、机器人回复、意图、状态、时间戳写进 MongoDB，字段加索引，后台运营可模糊搜索，30 天自动归档。
代码片段：
def audit(uid: str, inbound: str, outbound: str, intent: str): mongo.audit.insert_one({ "uid": uid, "in": inbound, "out": outbound, "intent": intent, "ts": datetime.utcnow() })避坑指南：血与泪的总结
多轮对话上下文丢失
现象：用户说“确认”，机器人问“确认啥？”
根因：状态机没持久化或 Redis 过期太短。
解法：把过期延长到 600 秒，并在每次收到消息后save_session()刷新 TTL。第三方 API 调用失败
现象：查订单接口 5 秒超时，机器人原地尬住。
解法：用tenacity库做指数退避重试，最大 3 次，总耗时 ≤ 8 秒，仍失败就返回“订单查询繁忙，稍后再试”。冷启动默认应答太蠢
现象：刚上线没训练数据，机器人全程“我没听懂”。
解法：预置 50 条高频 FAQ 意图，把置信度阈值降到 0.3，先“泛答”再后台标注，逐步收紧。
写在最后的开放问题
文本对话的 RPA 客服已经能替掉 60% 夜班人力，但语音交互才是“终极懒人福利”。如果要再进一步，让机器人听得懂、说得出，还能把语音文件转成结构化指令，你会怎么设计架构？欢迎一起头脑风暴。
