客服在线会话智能体流程图：从设计到落地的工程实践

---

背景痛点：状态爆炸与“if-else 地狱”

做在线客服的同学都懂，用户一句“查订单”后面能跟出 20 多种分支：已登录/未登录、有订单/无订单、已发货/未发货、是否 VIP、是否黑名单……早期我们用“规则引擎 + 硬编码”硬怼：一层层 if-else，外加策略模式，结果代码行数指数级上涨。
最严重的时候，一个“退货”场景嵌套 7 层条件，Code Review 时小伙伴直接说：“这逻辑我读一遍得先压个栈。”

核心问题有三：

1. 状态爆炸：订单、用户、优惠券、活动，四维笛卡尔积，理论上可达上千状态。
2. 分支嵌套：规则引擎把“条件”和“动作”写在一起，改动一个分支要读 200 行 DSL。
3. 并发与异常：WebSocket 断线重连、用户狂点按钮、重复回调，都会导致“状态漂移”，客服界面出现“幽灵消息”。

技术方案：规则引擎 vs 有限状态机

| 维度 | 规则引擎 | 有限状态机（FSM） | |---|---|---|---|---| | 核心抽象 | 条件→动作 | 状态×事件→新状态 | | 可视化 | 决策树/表格 | 流程图（节点=状态，边=事件） | | 嵌套深度 | O(n) 层条件 | O(1) 跳转表 | | 并发控制 | 需额外加锁 | 单事件循环天然串行 | | 动态热更 | 规则文件重新加载 | 状态转移表可配置化 |

结论：客服会话以“生命周期”为主线，事件驱动明显，FSM 更贴合；规则引擎适合做“策略计算”——例如计算优惠金额，两者可共存。

Python 实现：一个最小可扩展的 FSM 引擎

代码基于 Python 3.8，利用dataclass与typing，方便后续接入 pydantic 做校验。

from __future__ import annotations import json import time from dataclasses import dataclass, field from typing import Dict, Callable, Optional, Any, List State = str Event = str Action = Callable[["Session", Event, Any], None] @dataclass class Transition: source: State event: Event target: State action: Optional[Action] = None @dataclass class Session: uid: str state: State = "INIT" ctx: Dict[str, Any] = field(default_factory=dict) ts: float = field(default_factory=time.time) class FSMSessionEngine: def dispatch(self, session: Session, event: Event, payload: Any = None): key = (session.state, event) trans = self._table.get(key) if not trans: raise ValueError(f"No transition for {key}") if trans.action: trans.action(session, event, payload) session.state = trans.target session.ts = time.time() def __init__(self, transitions: List[Transition]): self._table: Dict[tuple[State, Event], Transition] = { (t.source, t.event): t for t in transitions }

状态转移表（可直接放 JSON 给运营配置）：

[ {"source": "INIT", "event": "login", "target": "AUTHED"}, {"source": "AUTHED", "event": "ask_order", "target": "WAIT_ORDER"}, {"source": "WAIT_ORDER", "event": "provide_order", "target": "HAS_ORDER"}, {"source": "HAS_ORDER", "event": "refund", "target": "REFUND_ASK_REASON"}, {"source": "REFUND_ASK_REASON", "event": "input_reason", "target": "REFUND_DONE"} ]

时间复杂度：

• 单条事件分发O(1)（哈希表）；
• 状态快照序列化O(n)（n=上下文字段数）。

架构设计：流程图 + 分布式持久化

文字版流程图（按会话生命周期）：

INIT ──login────► AUTHED ▲ │ │ ├─ask_order─► WAIT_ORDER │ │ │ │ │ ├─provide_order─► HAS_ORDER │ │ │ │ │ ├─refund─► REFUND_ASK_REASON │ │ │ │ │ ├─input_reason─► REFUND_DONE │ │ │ │ └───────────────────────────────────────────────────────────┘（会话结束，自动回收）

分布式持久化：

1. 每个Session快照通过json.dumps压缩后写入 Redis Hashsession:{uid}，TTL=30 min。
2. 引擎无状态，横向扩容时任意 Pod 均可GET→处理→SETEX。
3. 采用 Redis Lua 脚本保证GET/SET原子性，避免并发写覆盖。

-- refresh_ttl.lua local key = KEYS[1] local ttl = ARGV[1] local snap = ARGV[2] redis.call('SETEX', key, ttl, snap)

生产考量

1. 超时会话自动回收
   • 启动定时轮询（或 Redis 的 keyspace notification）扫到过期键，发送内部Event.TIMEOUT把状态推到TIMEOUT节点，触发客服侧“已断开”提示。
2. 幂等性设计
   • 用户重试消息带msg_id，引擎维护processed_msg:Set(uid)，Lua 脚本里SISMEMBER→SADD原子判断，重复事件直接丢弃，复杂度O(1)。
3. 并发控制
   • 单用户维度的消息由 Kafka 分区键uid保证顺序消费；WebSocket 层仅做转发，不维护状态。

避坑指南

1. 循环状态转移检测
   把状态当节点、事件当边，构建有向图，上线前跑networkx.simple_cycles：

   import networkx as nx G = nx.DiGraph() for t in transitions: G.add_edge(t.source, t.target) assert not list(nx.simple_cycles(G)), "存在循环转移"

2. 敏感词过滤异步化
   命中敏感词时不应阻塞 FSM，把消息扔给旁路async_filter协程，结果通过Event.SENSITIVE_CHECKED再回调，FSM 继续流转，平均延迟降低 30%。
3. 大上下文雪崩
   上下文超过 8 KB 时改走对象存储，Redis 只保留oss_url，避免大 Key 阻塞单线程。

完整运行示例

if __name__ == "__main__": with open("transitions.json") as f: tbl = [Transition(**item) for item in json.load(f)] engine = FSMSessionEngine(tbl) s = Session(uid="u123") engine.dispatch(s, "login") engine.dispatch(s, "ask_order") engine.dispatch(s, "provide_order", {"order_id": "OID123"}) engine.dispatch(s, "refund") engine.dispatch(s, "input_reason", "七天无理由") print("最终状态 =>", s.state) # REFUND_DONE

思考题：如何设计支持“动态加载流程图”的 FSM？

目前转移表在进程内存，如果运营想临时加一条“VIP 用户跳过退款原因”规则，就得改 JSON 并重启。
留给大家一个开放问题：

• 怎样在不重启服务、不掉会话的前提下，让引擎感知“新增/修改”的转移边？
• 版本差异带来的“新旧会话”兼容你又打算如何处理？

欢迎在评论区分享你的思路，一起把客服智能体做得更丝滑。

---