聊机器人能力容易说空，换个角度切入：先看能收到哪些事件。触发源决定能力上限——能接住的事件越多，自动化覆盖面越大。这篇先把触发源列全，再讲三层处理结构，最后说自动化该分几档。

一、八类触发源盘点

私聊消息、群消息（含 @ 提及）、好友申请、好友删除或拉黑、群成员进群退群、群公告变更、消息撤回、账号状态变化。八类事件的 payload 结构各不相同，字段名、嵌套层级都不一样。

实际做的时候，这八类里使用频率最高的是前三类，撤回事件主要用于纠正已归档的错误数据，群成员变更常用于维护群档案。账号状态做监控告警，别等客户投诉才发现号掉了。

二、三层结构：感知、决策、执行

感知层做规范化——把八种不同的 payload 转成一个统一的内部事件结构，字段固定为 type、session、actor、payload、ts。上层逻辑只处理这一种结构，新增事件类型只改适配器。

决策层分两条路：明确规则走关键词和白名单，模糊判断走模型或相似度。两条路并行，先规则后模型，规则命中就不必再调模型，省钱也快。

执行层汇总所有动作：发消息、打标签、拉群、撤回、通知内部人。执行层统一走队列控频，别让业务代码里到处直接 POST。

三、自动化分三档，别一步到位

L1 只提醒：把事件发到内部群，人来做决定。适合刚上线的场景，零风险。

L2 半自动：机器人生成草稿，人工确认后发。适合回复内容需要斟酌的场景。

L3 全自动：直接回复，事后抽样复盘。只留给高频且低风险的场景——FAQ、营业时间、地址查询这类。

大多数团队的正确顺序是 L1 跑两周看准确率，L2 再跑两周，最后只把最稳的一两类放到 L3。

触发源与能力对照

触发源	可做的自动化	典型档位	注意点
私聊消息	FAQ 应答、意向识别	L2/L3	回复别千篇一律
群消息	关键词监控、@提醒	L1/L2	误报要人工兜底
好友申请	自动通过、打来源标签	L3	通过后别立刻群发
群成员变更	刷新群档案、欢迎语	L2	高峰时段限速
消息撤回	同步下架已归档内容	L3	按 msg_id 精准匹配

事件规范化与分档路由

import time, threading, queue

class BotCore:
    """三层结构：规范化 → 决策 → 执行，按档位分流"""
    SEND_GAP = 3.0
    L1_NOTIFY_ROOM = "internal_alert_room"   # 内部提醒群

    def __init__(self, api, store, auto_level="L1"):
        self.api, self.store = api, store
        self.auto_level = auto_level
        self.exec_queue = queue.Queue()
        threading.Thread(target=self._executor, daemon=True).start()

    # ---------- 感知层：八类事件统一结构 ----------
    def normalize(self, raw):
        t = raw.get("type")
        msg = raw.get("message", {})
        base = {"actor": msg.get("from"),
                "ts": msg.get("ts", int(time.time())),
                "raw_type": t}
        if t == "message":
            base.update(type="private_msg",
                        session=msg.get("from"),
                        payload={"text": msg.get("content", ""),
                                 "file_id": msg.get("file_id")})
        elif t == "group_message":
            base.update(type="group_msg",
                        session=msg.get("group_id"),
                        payload={"text": msg.get("content", ""),
                                 "at": msg.get("at_list", [])})
        elif t == "friend_request":
            base.update(type="friend_req",
                        session=msg.get("from"),
                        payload={"req_id": msg.get("req_id"),
                                 "verify": msg.get("content", "")})
        elif t == "friend_deleted":
            base.update(type="friend_del", session=msg.get("from"),
                        payload={})
        elif t in ("group_member_join", "group_member_leave"):
            base.update(type=t, session=msg.get("group_id"),
                        payload={"members": msg.get("members", [])})
        elif t == "group_notice_change":
            base.update(type="notice", session=msg.get("group_id"),
                        payload={"notice": msg.get("content", "")})
        elif t == "message_recalled":
            base.update(type="recall", session=msg.get("from"),
                        payload={"msg_id": msg.get("recalled_msg_id")})
        elif t == "account_status":
            base.update(type="acc_status", session="",
                        payload={"online": msg.get("online")})
        else:
            base.update(type="unknown", session="", payload={})
        return base

    # ---------- 决策层：规则优先，模型兜底 ----------
    def decide(self, ev):
        etype = ev["type"]
        if etype == "private_msg":
            rule = self.store.match_rule(ev["payload"]["text"])
            if rule:
                return {"action": "reply", "text": rule["reply"],
                        "conf": 0.95, "by": "rule"}
            if self.auto_level in ("L2", "L3"):
                draft = self.store.draft_reply(ev["payload"]["text"])
                return {"action": "reply" if self.auto_level == "L3"
                        else "draft", "text": draft,
                        "conf": 0.6, "by": "model"}
        if etype == "group_msg":
            for kw in self.store.group_keywords(ev["session"]):
                if kw in ev["payload"]["text"]:
                    return {"action": "alert", "text": f"命中[{kw}]",
                            "conf": 1.0, "by": "rule"}
        if etype == "friend_req":
            return {"action": "accept", "text": "", "conf": 0.9,
                    "by": "rule"}
        if etype == "recall":
            return {"action": "unpublish", "text": "", "conf": 1.0,
                    "by": "rule"}
        return None

    # ---------- 执行层：统一入队，统一控频 ----------
    def handle_event(self, raw):
        ev = self.normalize(raw)
        if ev["type"] == "acc_status":
            if not ev["payload"].get("online"):
                self._enqueue("alert", self.L1_NOTIFY_ROOM, "账号离线")
            return
        decision = self.decide(ev)
        if not decision:
            return
        act = decision["action"]
        if act == "alert" or self.auto_level == "L1":
            self._enqueue("send", self.L1_NOTIFY_ROOM,
                          f"{ev['type']} 命中：{decision['text']}")
        elif act == "draft":
            self.store.save_draft(ev, decision["text"])   # 等人确认
        elif act == "reply":
            self._enqueue("send", ev["session"], decision["text"])
        elif act == "accept":
            self._enqueue("accept", ev["payload"]["req_id"], "")
        elif act == "unpublish":
            self.store.unpublish_by_msg(ev["payload"]["msg_id"])

    def _enqueue(self, kind, target, text):
        self.exec_queue.put({"kind": kind, "target": target, "text": text})

    def _executor(self):
        while True:
            job = self.exec_queue.get()
            try:
                if job["kind"] == "send":
                    self.api.post("/send", {"receiver": job["target"],
                                            "content": job["text"][:500]})
                elif job["kind"] == "accept":
                    self.api.post("/friend/accept",
                                  {"req_id": job["target"]})
            except Exception as e:
                self.store.log_fail("exec", str(e))
            time.sleep(self.SEND_GAP)

落地建议

机器人的能力上限由触发源决定，先把八类事件接全再做功能，别只盯着私聊消息。结构上分感知、决策、执行三层，八种 payload 在感知层统一成一种结构，上层逻辑就永远不会因为"多了一种事件"而重写。自动化千万别一步到 L3，L1 提醒跑两周看准确率、L2 草稿跑两周看改稿率，最后只把最稳的场景放开到全自动。群关键词、好友申请这类动作容易触发风控，全部走统一队列控频。事件推送格式与撤回、通过申请这些接口，字段以 Eyun 这类个人微信API平台的开发文档为准；分档策略、关键词库、草稿路由都是你的业务规则。