封面图：./workbuddy-wasender-blog-20260713-pm-cover.png
日期：2026-07-13（周一）｜主题：WhatsApp 聊天机器人的消息路由、状态机与上下文持久化

见过太多团队把 WhatsApp 机器人当成“关键字回复”的高级版——上线三天，要么把用户聊到反复说“人工”，要么把多轮对话状态搞成一团浆糊。今天这篇不聊 LLM，不聊 prompt 技巧，只聊一个被低估的工程问题：当用户从“你好”变成“我要退单”，你的机器人怎么知道现在该说什么？

这背后是消息路由、状态机、上下文持久化三件事。把这三件事做对，一个普通的规则型机器人也能撑住 80% 的出海客服场景；做错，再聪明的模型也只是金玉其外。

---

一、先看清消息长什么样

WhatsApp 机器人接收的消息有两种来源：走官方 WhatsApp Cloud API 的 Webhook，或者走 WhatsApp Web 本地协议 的消息副本。它们的字段命名不同，但本质是一样的：你拿到的是一条带发送者、时间戳、类型、正文的消息。

字段	Cloud API Webhook	Web 本地消息	工程含义
发送方	messages[0].from	key.remoteJid	用户或群聊的 JID
消息 ID	messages[0].id	key.id	去重、ACK、回复引用
时间戳	messages[0].timestamp	messageTimestamp	秒级，用于超时和排序
消息类型	messages[0].type	message.*Message	text / image / button / list_reply
文本正文	messages[0].text.body	message.conversation	纯文本内容

拿到消息后，第一件事永远是解析成统一的数据结构。别在业务代码里直接 payload["entry"][0]["changes"]，否则总有一天 Webhook 结构变了，你要改 30 个文件。

from dataclasses import dataclass
from typing import Literal, Optional

@dataclass
class InboundMessage:
    msg_id: str
    sender: str          # 手机号，如 86138xxxx
    chat_jid: str        # 用户或群的 JID
    ts: int
    type: Literal["text", "image", "button", "list_reply"]
    body: str
    is_group: bool = False

    @classmethod
    def from_cloud_webhook(cls, payload: dict) -> "InboundMessage":
        msg = payload["entry"][0]["changes"][0]["value"]["messages"][0]
        return cls(
            msg_id=msg["id"],
            sender=msg["from"],
            chat_jid=f"{msg['from']}@s.whatsapp.net",
            ts=int(msg["timestamp"]),
            type=msg["type"],
            body=msg.get("text", {}).get("body", ""),
        )

这里有一个小坑：同一个用户连续发三条消息时，Meta 可能发三条独立的 Webhook。如果你的 handler 是并发的，顺序会乱。后面会讲怎么解决。

---

二、消息路由：别用 if-else 硬堆

路由器的任务是判断“这条消息现在应该交给哪个 handler”。最简单的方式是 if-else，但业务一多就崩盘。一个能长期维护的路由器至少应该支持：

• 关键字 / 正则匹配
• 状态过滤（同一句话，在“idle”和“await_order”状态下响应不同）
• 优先级（退款投诉 > 通用问候）
• 兜底回复

import re
from dataclasses import dataclass
from typing import Callable

@dataclass
class Handler:
    match: Callable[[InboundMessage, "Session"], bool]
    action: Callable[[InboundMessage, "Session"], dict]
    priority: int = 0

class MessageRouter:
    def __init__(self):
        self.handlers: list[Handler] = []

    def register(self, handler: Handler):
        self.handlers.append(handler)
        self.handlers.sort(key=lambda h: h.priority, reverse=True)

    def dispatch(self, msg: InboundMessage, session: "Session") -> dict:
        for h in self.handlers:
            if h.match(msg, session):
                return h.action(msg, session)
        return {"type": "text", "body": "没听懂，输入 '人工' 转接客服。"}

注册几个具体 handler：

router = MessageRouter()

router.register(Handler(
    match=lambda m, s: any(k in m.body for k in ["退", "refund", "return"]),
    action=lambda m, s: (
        s.update(state="refund:ask_order"),
        {"type": "text", "body": "请发送订单号，我帮你查退款进度。"}
    )[1],
    priority=10,
))

router.register(Handler(
    match=lambda m, s: s.state == "refund:ask_order" and m.type == "text",
    action=lambda m, s: (
        s.update(state="refund:confirming", order_id=m.body),
        {"type": "text", "body": f"收到订单号 {m.body}，正在提交退款确认。"}
    )[1],
    priority=20,
))

router.register(Handler(
    match=lambda m, s: "人工" in m.body,
    action=lambda m, s: (
        s.update(state="handoff:queued"),
        {"type": "text", "body": "已为你排队，客服接入前请勿关闭聊天。"}
    )[1],
    priority=100,
))

这种写法把匹配规则和业务动作拆开，比一坨 if-else 好测试得多。测试时你不需要调 WhatsApp，只需要构造 InboundMessage 和 Session。

---

三、状态机：对话的灵魂

路由只解决“谁处理”，状态机解决“现在处在哪一步”。一个多轮对话如果没有状态机，用户体验就是：每发一条消息都像在跟陌生人重新打招呼。

from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class Session:
    chat_jid: str
    state: str = "idle"
    data: dict = field(default_factory=dict)
    updated_at: float = 0.0

    def update(self, state: Optional[str] = None, **kwargs):
        if state:
            self.state = state
        self.data.update(kwargs)
        self.updated_at = time.time()

一个典型的退款/查单状态转移表如下：

当前状态	用户输入	下一状态	动作
idle	“查订单” / “order”	order:await_id	请求订单号
order:await_id	订单号	order:fetching	查询并返回状态
idle	“退款” / “refund”	refund:ask_order	请求订单号
refund:ask_order	订单号	refund:confirming	提交退款确认
*	“人工” / “human”	handoff:queued	进入人工队列

状态机的价值在于：它把业务流程显式化。运营人员、产品经理、后端开发看到的是同一张状态表，而不是藏在某个 prompt 里的“你看着办”。

---

四、上下文持久化：别让进程重启吃掉对话

生产环境不是单机 demo。容器重启、部署更新、负载均衡切换，都会让内存里的会话消失。所以状态必须外置，通常是 Redis。

我的 Redis key 设计很简单：

Key	类型	示例	作用
wa:session:{jid}	Hash	state, data, updated_at	当前会话状态
wa:history:{jid}	List	最近 20 条消息	上下文与审计
wa:rate:{jid}	String	60 秒窗口计数	单聊限流，防止刷屏
wa:lock:{jid}	String	UUID	同聊消息串行，避免竞态

会话存取代码：

import json
import redis

class RedisSessionStore:
    def __init__(self, r: redis.Redis, ttl: int = 3600):
        self.r = r
        self.ttl = ttl

    def _key(self, jid: str) -> str:
        return f"wa:session:{jid}"

    def load(self, jid: str) -> Session:
        raw = self.r.hgetall(self._key(jid))
        if not raw:
            return Session(chat_jid=jid)
        return Session(
            chat_jid=jid,
            state=raw.get(b"state", b"idle").decode(),
            data=json.loads(raw.get(b"data", b"{}")),
            updated_at=float(raw.get(b"updated_at", b"0")),
        )

    def save(self, session: Session):
        key = self._key(session.chat_jid)
        self.r.hset(key, mapping={
            "state": session.state,
            "data": json.dumps(session.data),
            "updated_at": str(session.updated_at),
        })
        self.r.expire(key, self.ttl)

三个细节：

1. TTL 不要太长：一般 30-60 分钟足够。超过这个时间用户没说话，默认回到 idle 是对话机器人最好的礼仪。
2. data 里只放业务需要的字段：不要塞进整条聊天记录，那是 history 的职责。
3. 同 JID 加锁：Webhook 是并发的，同一用户连发三条消息时，必须用 SET wa:lock:{jid} NX EX 5 串行处理，否则状态机会翻车。

---

五、把用户拉回轨道：列表与按钮

纯文本对话最危险的地方是用户“乱输入”。理想情况下，关键节点应该给用户提供有限选项。WhatsApp Cloud API 支持 interactive 消息，包括按钮和列表。

def build_list_message(title: str, body: str, options: list[tuple[str, str]]) -> dict:
    rows = [
        {"id": opt_id, "title": opt_title, "description": ""}
        for opt_id, opt_title in options
    ]
    return {
        "type": "interactive",
        "interactive": {
            "type": "list",
            "body": {"text": body},
            "action": {
                "button": "选择服务",
                "sections": [{"title": title, "rows": rows}],
            },
        },
    }

# 使用示例
menu = build_list_message(
    title="客服服务",
    body="请选择你需要的服务：",
    options=[
        ("opt_order", "查订单"),
        ("opt_refund", "退款/售后"),
        ("opt_human", "转人工"),
    ],
)

用户点击列表项后，Webhook 会发来 type=list_reply 的消息，body 是 interactive.list_reply.id，比如 opt_order。你的路由器只要按 ID 分发，不需要再做 NLP。这种“结构化引导”比让用户自由输入可靠十倍。

---

六、人工接管：什么时候该认输

再强的机器人也有边界。敏感问题、重复失败、用户明确要求，必须能平滑交给人工。关键是不要只是在群里艾特老板。

from collections import deque

class HumanQueue:
    def __init__(self):
        self.waiting: deque[str] = deque()

    def enqueue(self, jid: str):
        if jid not in self.waiting:
            self.waiting.append(jid)

    def next(self) -> str | None:
        return self.waiting.popleft() if self.waiting else None

def should_handoff(msg: InboundMessage, session: Session) -> bool:
    if "人工" in msg.body or "human" in msg.body.lower():
        return True
    if session.state.startswith("fallback") and session.data.get("fallback_count", 0) >= 2:
        return True
    if session.state in {"payment", "refund"} and session.data.get("amount", 0) > 1000:
        return True
    return False

转人工后，建议把当前会话状态冻结，并给客服展示一条摘要：用户 JID、当前状态、已收集的数据、最近 3 轮对话。这能省掉客服 30% 的重复提问时间。

---

七、出海实战：从“数据提取”到“对话承接”的本地闭环

前面都是技术骨架，落地到业务里，通常是一条私域运营流水线。下面这条链路是一个做东南亚家居市场的 3 人小团队的真实打法，可以直接复刻。

痛点：行业 WhatsApp 群里每天有潜在采购商标价、问库存，但手动抄号码太慢，群发又容易被封。

闭环：先用号码提取与备份工具把群成员和聊天记录导出，经清洗验证后，用多账号群发触达工具做个性化开发信；用户回复后，由自研机器人按状态机承接，会话上下文和聊天记录本地归档。

阶段	动作	关键控制点
1. 提取	从目标行业群导出成员号码与聊天记录	按国家码、日期过滤，拒绝无效数据
2. 验证	校验号码是否已注册 WhatsApp	去重、无效号剔除
3. 触达	多账号轮换发送个性化开发信	发送延迟、账号轮换、内容变量
4. 承接	用户回复后进入机器人状态机	查单、留资、转人工
5. 归档	聊天记录导出为本地 Excel/HTML	防止账号异常导致客户失联

WAExport 在阶段一和阶段五承担“号码提取与备份”角色：从群组一键拉取成员清单并导出 XLSX/CSV，聊天记录实时同步到本地，全程不上云。

WASender 在阶段三承担“多账号批量触达”角色：支持最高 50 个账号并行轮换、插入姓名/国家变量与防封策略，让开发信在合规节奏内跑完。

这条链路的工程含义是：把 WhatsApp 上非结构化的线索，变成可清洗、可编排、可对话、可归档的数据资产。机器人做的不是“自动回复”，而是承接群发后的用户意图，把一次性触达变成可持续运营的对话。

---

八、生产自检清单

上线前建议逐项过一遍：

检查项	是否完成	说明
用户已 opt-in 或属于公开行业群	□	降低被举报和投诉风险
Webhook 已做消息 ID 去重	□	防止重复处理和重复回复
同 JID 已加分布式锁	□	避免并发导致状态机错乱
会话状态已外置并设 TTL	□	Redis 失效后回到 idle 是合理行为
单聊限流已配置	□	防止用户刷屏或误触发
人工接管规则已明确	□	退款、投诉、敏感词必须能转人工
异常状态已记录日志	□	方便排查“为什么没回复”
聊天记录与线索已本地备份	□	防止账号异常导致客户失联

---

九、FAQ

Q：WhatsApp 机器人和“关键字自动回复”有什么区别？

A：关键字回复只根据当前消息匹配固定答案；机器人有状态机，能根据对话历史决定下一步动作。前者是“单轮”，后者是“多轮”。

Q：状态机是不是只适合做规则型机器人？

A：不是。LLM 也可以被状态机约束——比如用状态定义当前流程，用 LLM 生成状态内的回复文案。状态机保证流程不跑偏，LLM 提升文案自然度，两者并不矛盾。

Q：同用户连续发多条消息，顺序乱了怎么办？

A：按 chat_jid 加锁，处理完一条再拿下一条。锁的超时建议 5-10 秒，既能防止竞态，也不会因为某条消息卡住导致后续消息长时间等待。

Q：会话状态应该存多久？

A：一般 30-60 分钟。超过 TTL 后，用户再次发消息视为新会话，回到 idle。这是合理的，因为用户通常不会隔几个小时继续上一轮对话。

Q：WhatsApp Cloud API 和本地 Web 方案都能做机器人吗？

A：都可以。Cloud API 适合企业级、有技术团队的场景，需要 WABA 和模板审批；本地 Web 方案更轻量，适合小团队快速验证，但需要自己处理浏览器环境、防封和会话隔离。

---

参考与延伸阅读

• Meta WhatsApp Business Platform 官方文档：https://developers.facebook.com/docs/whatsapp
• Meta WhatsApp Cloud API Webhook 消息格式：https://developers.facebook.com/docs/whatsapp/cloud-api/webhooks/components
• Green API whatsapp-chatbot-python 状态管理：https://pypi.org/project/whatsapp-chatbot-python/
• wa-agent：基于 JID/Group 路由的 WhatsApp Agent 框架：https://github.com/ibrahimhajjaj/wa-agent
• Wappa：Python 异步 WhatsApp 机器人框架（含 Redis 状态缓存）：https://pypi.org/project/wappa/

---

写在最后：WhatsApp 聊天机器人的技术难点，从来都不是“把消息发出去”，而是“知道现在该发什么”。消息路由负责分拣，状态机负责记忆，上下文持久化负责在进程重启后依然记得用户。把这三层做扎实，出海私域的客服和转化才有了可扩展的底座。