技术支持 wechatapi.net

很多人做微信机器人，第一反应都是先写一个能跑通的脚本：

@app.post("/wechat/callback")
async def handle_wechat(request: Request):
    data = await request.json()
    text = data["Data"]["Content"]["string"]
    result = call_ai(text)
    send_text(result)
    return {"ok": True}

这种写法最大的优点是快，当天就能跑起来；但它也有一个致命问题：你得到的只是一个脚本，不是一个能承载场景的入口层。

我最近在做的事情，是把微信接成 OpenClaw 的一个入口。做下来以后我最大的认知变化就是：
如果你的目标是长期做入口能力，就不要把项目理解成机器人脚本，而要把它理解成消息网关。

为什么“脚本思维”很快会碰到天花板

一开始，脚本思维的问题不明显，因为你的目标只有一个：
收消息，调模型，发回去。

但项目一旦继续往前走，就会很快碰到下面这些问题：

1. 私聊和群聊逻辑混在一起

私聊天然是单用户上下文。
群聊则至少有两种模式：

• 整个群共享一个上下文
• 群里每个人独立上下文

如果一开始没有 session 设计，后面会非常难补。

2. 消息回调和 AI 调用耦合太紧

如果回调一进来就直接等模型返回，慢一点就会把整个回调压住。
尤其是用 CLI 调 OpenClaw 时，每次消息都要重新起进程，延迟更明显。

3. 没有状态层

没有状态层，后面这些都很痛：

• 去重
• 限流
• 排队
• 日志
• 调试
• 会话积压控制

4. 无法自然扩展到企微、TG、飞书

如果从第一天就写死成“微信机器人脚本”，那以后每接一个入口都得重来一遍。

我后来是怎么改成“入口网关”的

我现在把整体链路拆成了更像网关的结构：

微信回调
   ↓
回调解析层
   ↓
消息标准化
   ↓
Session 路由
   ↓
Worker 队列
   ↓
OpenClaw 调用层
   ↓
微信发送层

核心思想很简单：

• 微信只是入口
• OpenClaw 是能力内核
• 中间的网关层负责把两边真正接起来

关键设计一：按 session 分片，而不是简单起线程

一个很常见的误区是：消息来了就开线程。
这样做在早期看起来很快，但很快就会遇到上下文乱序问题。

我最后用的是按 session 分片：

def shard_index_for_session(session_id: str, worker_count: int) -> int:
    h = int(hashlib.md5(session_id.encode("utf-8")).hexdigest(), 16)
    return h % worker_count

这样做的好处是：

• 不同会话可以并行
• 同一个会话固定进同一个 worker
• 不会把同一段上下文跑乱

这个设计在群聊场景尤其重要。

关键设计二：先把消息标准化，再谈业务逻辑

微信回调结构看起来很简单，实际上真正难的是不同消息场景的判断。

比如我后来按回调文档做了这些判断：

def parse_wechat_payload(data):
    wxid = str(data.get("Wxid", "") or "").strip()
    msg_data = data.get("Data", {}) or {}

    from_user = (msg_data.get("FromUserName", {}) or {}).get("string", "")
    to_user = (msg_data.get("ToUserName", {}) or {}).get("string", "")
    raw_content = (msg_data.get("Content", {}) or {}).get("string", "")
    msg_type = msg_data.get("MsgType")

    is_self = bool(wxid and from_user == wxid)
    is_group = from_user.endswith("@chatroom") or to_user.endswith("@chatroom")

    if from_user.endswith("@chatroom"):
        chat_id = from_user
    elif to_user.endswith("@chatroom"):
        chat_id = to_user
    else:
        chat_id = from_user if not is_self else to_user

    sender_wxid = from_user
    actual_text = (raw_content or "").strip()

    if is_group and raw_content and ":\n" in raw_content:
        possible_sender, possible_text = raw_content.split(":\n", 1)
        if possible_sender.startswith("wxid_"):
            sender_wxid = possible_sender.strip()
            actual_text = possible_text.strip()

    return {
        "wxid": wxid,
        "msg_type": msg_type,
        "chat_id": chat_id,
        "sender_wxid": sender_wxid,
        "actual_text": actual_text,
        "is_group": is_group,
        "is_self": is_self,
    }

这一步做对以后，后面的业务逻辑才会稳定。

关键设计三：先只处理文本，把链路跑稳

微信支持的消息类型很多：

• 文本
• 图片
• 语音
• 视频
• 文件
• 引用
• 名片
• 系统通知

如果一开始全部都接，系统很快会变得复杂。
所以我现在的策略很明确：

第一阶段只处理文本消息。

代码也非常直接：

if msg_type != 1:
    logger.info("暂时只处理文本消息，忽略 MsgType=%s", msg_type)
    return {"status": "ignored_msg_type"}

先把主链路做稳，比过早堆能力重要得多。

关键设计四：默认忽略自己发出去的消息

这是最容易忽略、但一定要做的防线。

if parsed.get("is_self"):
    logger.info("忽略自己发送的消息: %s", parsed.get("msg_id"))
    return {"status": "ignored_self"}

如果不做这一步，很容易出现：

• 机器人收到自己的回复
• 继续把自己的回复丢给 OpenClaw
• 最后形成自回环

这套思路对我最大的帮助

把项目从“脚本”改成“入口网关”以后，我最大的感受是：
很多之前混乱的问题，开始变得可以拆解了。

比如：

• 延迟问题，能明确判断是在微信层还是在 OpenClaw CLI 层
• 群聊问题，能明确知道是 chat_id 还是 sender_wxid 处理错了
• 会话问题，能通过 session 设计做策略切换
• 未来接企微，也不再需要完全重写

我现在的判断

如果你只是想做一个会回消息的 demo，脚本思维没问题。
但如果你想继续往下做：

• 私域入口
• 智能客服
• 群聊 AI 助手
• 多入口通道层

那就应该尽早把它往“入口网关”方向去设计。

一句话总结：

脚本解决的是“能不能跑”，网关解决的是“能不能长期承载场景”。