LuckyTri:三层架构打造不刷屏的QQ群机器人 三个月前我被一个整天刷屏的签到机器人搞得有点烦。每到整点它就准时把几十条抽奖结果刷满整个QQ群群友提过好几次管理员又拿它没办法。后来我实在忍不住决定自己写一个QQ群聊天机器人不吵、不闹但能真正帮群友解决点小需求。于是就有了LuckyTri。LuckyTri不是那种一拉起来就跑的成品机器人而是我从这次需求里拆出来的一个小框架。它核心想解决的不是“怎么发消息”而是“一条消息进来之后该怎么走完一段可控的流程”。现在这套东西已经在几个技术交流群里跑了小半年日常消息量不算大但胜在稳。写这篇博文是想把从需求到落地的完整过程摊开为什么叫LuckyTri三层结构怎么画中间哪些地方会翻车以及如果你也要做类似的事情可以怎么少走弯路。1. 内容整体设计与思路拆解1.1 机器人要解决的不是“聊天”是“热闹和秩序”群里要机器人多数时候不是为了找人聊天。你看群里最受欢迎的同款机器人无非是签到、每日运势、抽奖、群管提醒这几类。它们本质上是工具签到制造黏性运势制造话题抽奖制造刺激。既然是工具就得讲可控。最烦人的场景我也经历过某个机器人被群友玩出花样“再来一次”“我也要”“出个红色”一刷就是几十条整个群都没法看。所以我把需求拆成一句话让机器人提供乐趣但绝不允许它成为噪声来源。怎么做到两者兼顾我的答案是分层。触发层先判断“这条消息到底是不是给我的”路由层再去匹配“这条消息应该交给谁处理”执行层最后调用具体功能并负责冷却、去重和回复模板。这三层拆开之后每个层的逻辑都变得很简单也方便单独加限制。打个比方一个厨房里不能每个人随便上灶台触发层是迎宾员只负责把你领到对应窗口路由层是点菜单写下你要的菜执行层才是后厨按菜单出菜。以前那种把所有if判断堆在同一个函数里的写法相当于让迎宾员、传菜员、厨师挤在同一个灶台前人一多必乱。1.2 为什么叫 LuckyTri三层语义三层结构项目名是同事某天开玩笑起的当时我正在写每日运势功能他说“你这机器人能不能让我今天转运一次”。名字里有 Lucky又因为整体结构是三个层次再加一个 Tri拼起来就是 LuckyTri。后来我把名字的含义干脆固化成“Lucky Triple”触发、路由、执行三层都顺了机器人才会显得“好运”——不误判、不重复、不卡死。三层结构用下表就能看清边界触发层只回答三类问题是不是群消息、是不是以指令前缀开头、有没有命中的关键词路由层只回答命令注册表里有没有这个命令、当前用户有没有权限、冷却是否通过执行层只做调用具体功能、生成回复、写日志。每一层都不越界。实际改动中我收益最大的是新增一个指令时只需要在 plugins 目录里加一个文件注册装饰器即可触发层和路由层几乎不用动。如果一开始不拆层加一个指令就要去改主循环和权限判断改十个指令之后几乎没人能看懂整个流程了。层次核心职责典型输出触发层判断消息是否属于机器人应处理的范畴命中 / 忽略路由层匹配指令、校验权限、执行冷却可执行的 handler 或拒绝原因执行层运行业务逻辑、生成回复、记录结果回复内容 日志2. 技术选型与工程准备2.1 接入协议OneBot 风格事件流第一个要决定的是机器人怎么和QQ群建立连接。我选的方式是走开放的事件上报标准也就是 OneBot 风格的接口。简单说接入端负责维护和IM服务器的长连接然后把群消息、群成员变动、请求事件等统一包装成 JSON 事件通过 WebSocket 推给我自己的服务端。LuckyTri 的业务层不关心底层是 HTTP 回调还是 WebSocket 长连只拿到一个结构化的事件对象。这样最大的好处是以后想换接入端或者换平台不需要动业务代码。事件对象我直接定义一个轻量数据类。写代码的时候可以按下面的最小字段来class GroupMessageEvent: event_id: str group_id: str user_id: str raw_message: str sender_role: str # owner / admin / memberevent_id 特别重要后面会提到防重复回复全靠它sender_role 则是权限判断的可信来源不要自己去消息文本里猜。很多人刚开始只取 raw_message然后就开始 if 判断往往漏掉这两个字段问题过两天才暴露。2.2 编程语言与异步模型LuckyTri 用的 Python 3.10 和 asyncio。选 Python不是因为它性能强而是群聊机器人本来就是 IO 密集型大部分时间在等网络返回CPU 占用很低。asyncio 的好处是一条消息触发的外部请求在等待时事件循环可以继续处理其他消息。如果只用同步脚本某条指令调一次慢接口群里其余人的消息都得排队体感就是“机器人变笨了”。我实际测试过一个同步版本在处理每日一言时如果外部接口扛了 3 秒后续所有消息都会卡住改成异步之后同样场景下机器人在等待期间能继续响应其他指令。对于小群来说可能无所谓但是当机器人加进多个活跃群之后差距一下就出来了。所以即便你的功能很简单我也建议从第一天就写异步省得日后返工。2.3 配置与状态存储配置文件我用 YAML 而不是 JSON。原因很实际YAML 支持注释我可以在配置里写给未来自己看的话嵌套结构也更好读。状态存储选了 SQLite文件型数据库不需要单独部署服务备份就是拷走一个文件。对于 LuckyTri 这种量级一天几千条事件、几百次抽奖记录SQLite 绰绰有余没必要上中间件。等真到了需要 Redis 做订阅和缓存的规模你已经不是看这篇博文的人了。如果要做高并发当然缓存很重要。但刚开始部署一个群机器人核心目标是快速跑通和稳定运行。我见过有人一上来就搭 Redis结果数据一致性没理清反而比 SQLite 更容易出错。我的建议是先让SQLite跑起来把日志和数据目录挂在宿主机上将来如果性能不够再在接入端前面加缓冲层。渐进式演进比一步到位靠谱。3. 核心逻辑拆解与实操实现3.1 消息进入后的完整处理链路一条群消息进来LuckyTri 的处理链路是固定的接入端把原始消息包装成 GroupMessageEvent。预处理过滤空字符串、统一大小写、去掉首尾空白、解析命令前缀。触发判断前缀命中 / 或自定义关键词否则直接忽略。权限检查读取 sender_role与指令要求的 required_level 比较。冷却检查在进程内字典查该指令在“群用户”维度上的最近调用时间。路由到注册表中的 handler。handler 执行具体逻辑返回 Reply 对象。模板渲染发送并写入日志。这个流程刻意固定是为了让每个环节都能独立插桩。比如我想知道“指令触发了多少次”只需要在第 6 步加一个统计装饰器想封禁某个群的指令在第 4 步加一条群黑名单判断即可。如果逻辑混在一个大函数里任何改动都得通读全部代码。实测下来固定链路虽然多几个函数跳转但排查问题的时候思路特别清晰从日志的节点ID就能看出消息走到哪一步卡住了。3.2 用注册表代替 if-else 大山早期版本里我是这样处理的if msg.startswith(/运势): await fortune(ctx) elif msg.startswith(/抽奖): await roll(ctx)三个功能的时候还能忍功能到十几个之后这个函数就是一座大山每个分支里还要再判断权限、冷却、参数长度逻辑互相纠缠。后来我改成注册表模式每个功能一个文件文件里用装饰器注册指令。示例# plugins/fortune.py from core.router import router router.register(运势, required_levelmember, cooldown10) async def fortune(ctx): result await fortune_service.draw(ctx.user_id) return ctx.reply(result)router.register 做的事很简单把指令名和 handler 存进字典启动时扫描 plugins 目录自动 import。这样新增功能不需要修改路由主文件也不会影响已有功能。权限、冷却都能作为元数据写在装饰器里代码即配置读起来非常直观。3.3 权限模型别相信消息里的“我是管理”权限是群机器人最容易翻车的点。很多机器人会判断消息文本里有没有 机器人、或者昵称叫“管理员”就放行这是不可靠的。LuckyTri 的信任链只建立在事件对象里的 sender_role这个字段由接入端从IM服务器拿到的真实身份信息填充。我定义了三个级别owner、admin、member任何指令注册时都可以指定required_level。权限检查在路由之前不通过就直接忽略甚至不需要告诉用户“没权限”这样也能避免向非管理员暴露指令的存在。有一次我亲眼见过一个机器人翻车普通群友发现管理员指令是/ban xxx于是自己也发一遍结果机器人居然认了因为它只检测了消息前缀里有没有 机器人没有看发送者身份。这个教训让我把权限逻辑抽成单独模块并用配置声明哪些指令属于管理组。再次提醒所有涉及敏感操作的指令必须同时校验 group_id 白名单和 sender_role 字段缺一不可。3.4 概率引擎LuckyTri 的“运气”不是纯随机名字带 Lucky自然要讲讲扔骰子、抽奖、每日运势这些功能。最开始的抽奖实现非常简单result random.choice([SSR, SR, R, N])跑了两天群友反馈“出SSR的概率不对”“我连续十次都是N是不是针对我”。这其实暴露了一个陷阱全局 random 在多协程下如果被多个 handler 共用并发时会互相干扰结果分布看上去会偏离预期。另一方面没有保底机制极端非洲酋长确实可能很长一段时间什么都不出体验很差。我后来的做法是抽奖单独使用一个secrets.SystemRandom实例避免全局随机种子同时为了实现“非酋保护”在数据库里记录每个用户的连续失败次数连续 N 次未中奖时下一次强制命中中奖后计数器清零。对于每日运势我用的是可复现随机以日期和用户ID作为种子同一个用户在同一天内无论问多少次得到的结果都一样这样既公平又省数据库查询。import hashlib, random seed int(hashlib.md5(f{today}:{user_id}.encode()).hexdigest()[:8], 16) daily_rng random.Random(seed) level daily_rng.choices([大吉, 吉, 平, 小凶], weights[1, 3, 5, 2])[0]群友后来评价“这机器人有点玄学”因为同一天反复查结果都一样反而成了话题。3.5 冷却、去重和回复模板冷却机制是防止刷屏的硬要求。我在指令装饰器里加了cooldown参数单位是秒。实现上就是把(group_id, user_id, command_name)作为 key最近一次调用时间存在进程内字典里。单机单进程场景下完全够用如果以后要跑多个实例就得换成 Redis 这类共享存储但目前没必要。需要注意冷却粒度我建议按“群用户”来限而不是按群全局限否则两个群友同时想玩后面那个人会被莫名其妙地冷却到体验很糟。去重则是另一类问题。接入端基于 WebSocket 长连时偶尔发生断线重连过程中可能把若干条消息重复推给服务端。如果不处理就会看到机器人把一条回复发两遍。LuckyTri 维护了一个最近 event_id 的环形缓冲收到已经处理过的 event_id 就直接丢弃。回复模板我统一放在 YAML 里允许不同群覆盖模板让群管理员可以定制机器人语气这也是后来“好运”感知的来源之一。4. 实操部署从本地调试到稳定在线4.1 目录结构与启动入口先看我当时的目录结构luckytri/ ├── app.py ├── config.yml ├── core/ │ ├── bot.py │ ├── router.py │ ├── permissions.py │ └── databases.py ├── plugins/ │ ├── __init__.py │ ├── fortune.py │ ├── roll.py │ └── manage.py └── data/ └── luckytri.dbapp.py 只负责创建事件循环、加载配置、启动接入端core 目录放框架代码plugins 目录放具体功能插件。这个结构很朴素但足够撑到几十个指令。启动入口不要写太复杂。核心逻辑是读配置 - 初始化数据库 - 实例化 Bot - 注册所有插件 - 连接接入端 - 进入事件循环。我习惯把核心对象都挂在bot上这样 handler 里通过 ctx 就能访问 bot.db、bot.router、bot.config。不推荐用真正的全局单例测试时会很痛苦依赖注入虽然多写一点代码但让每个 handler 都变成了可独立调用的小函数。4.2 本地模拟事件先不连真实群写代码的时候最怕每次都要去真实QQ群里发一条消息验证。后来我发现直接在本地构造 GroupMessageEvent 喂给 bot 处理调试效率高得多。比如async def test_run(): event GroupMessageEvent( event_idtest-001, group_idgrp-dev, user_iduser-test, raw_message/运势, sender_rolemember, ) result await bot.handle(event) print(result.to_dict())这其实就是一个最简单的单元测试。再配合 pytest把各种输入都跑一遍能拦住很多回归问题。真实群聊环境适合做最后的验收不适合每写一个功能都拉进去验证。4.3 用容器化部署把它变成 7×24 小时服务本地调试通过之后我部署在一台小主机上。这里不得不用到容器化因为服务器上环境很干净我既不想污染宿主机也希望restart策略能自动拉起进程。Docker Compose 大致如下services: luckytri: build: . restart: unless-stopped environment: - LUCKYTRI_LOG_LEVELINFO volumes: - ./config.yml:/app/config.yml:ro - ./data:/app/datarestart: unless-stopped表示除非手动停止否则容器退出会自动重启配合日志收集基本能做到掉线后不需要人工干预。配置用只读挂载是因为我不想在容器内部改配置所有变更都在宿主机上完成重启容器生效。data 目录必须写挂载否则 SQLite 文件存在容器层里容器重建就全没了。我踩过这个坑某次更新镜像忘了挂数据目录结果所有群友的抽奖记录被清空只能一遍遍道歉。后来我在部署脚本里加了检查启动前先确认data/luckytri.db存在否则拒绝启动。4.4 接真实群聊白名单和日志级别连接真实QQ群之前最重要的一件事是配置群白名单。LuckyTri 只会处理配置里出现的 group_id其他群的消息一律忽略。这样即便有人把机器人邀请到陌生群里机器人也不会乱说话。白名单可以通过管理指令动态增加但初始配置必须是 YAML 里的硬编码安全兜底。日志级别我平时用 INFO只记录消息ID、群ID、命中的指令名、处理耗时不记录完整消息内容。DEBUG 日志虽然非常详细但开着跑一天会产生海量日志还会带来隐私风险。想看某个具体问题的时候再临时把某个群调整到 DEBUG 级别问题解决后立刻关回去。这些小细节决定了你这个机器人能不能长期稳定待在群里。5. 常见问题与排查技巧实录5.1 机器人偶尔不回消息排查顺序我建议固定先看接入端日志确认事件有没有到服务端再看 LuckyTri 的节点日志确认走到第几步最后检查冷却时间和权限。很多时候问题都出在冷却还在生效群友以为机器人挂了其实它只是不想理人。我们可以把冷却失败也写成一条 DEBUG 日志而不是完全静默这样后期排查方便很多。5.2 某条回复被重复发送这个问题的根源通常是 WebSocket 重连之前提到过。解决方案本身不复杂LuckyTri 用一个set保存最近 5000 个 event_id超过上限就清掉旧的。为什么是 5000因为正常情况下一个活跃群的单日消息量也就几千保留最近 5000 个足够覆盖一次断线重连期间的所有消息内存也完全可接受。要注意去重集合必须和事件处理在同一进程内如果拆成多实例就要把 event_id 放到共享存储里。5.3 群友说抽奖概率不对这个问题概率算法本身可能没错但“概率不对”往往是感知问题。人会对坏结果记忆更深刻所以需要保底机制。我在数据库里记录每个用户的连续未中奖次数达到阈值后强制命中。同时把每个抽奖的权重和概率都写进配置并且开放给群友查询比如/sr指令返回当前各档位的准确概率。透明能解决绝大多数“你是不是暗改概率”的质疑。5.4 管理指令被普通成员触发如果出现这种情况第一个要检查 sender_role 是从哪里来的。如果它来自消息文本或者用户昵称那等于没有权限系统。第二个要检查的是权限检查有没有被放在路由之前。我建议权限检查的日志单独一个字段权限不足时记为blocked而不是ignored这样审计管理指令时一眼能看出有没有人尝试越权。同一段逻辑不要重复写在各个 handler 里一旦某个功能漏写又会变成漏洞。5.5 容器重启后数据没了前面已经提过这是 volume 挂载的问题。除了挂载还要养成备份习惯。一个小技巧每次更新部署前用sqlite3 luckytri.db .backup backup.db备份一次再去 build 镜像。如果没有备份我那次丢记录之后只能靠群友自己截图回忆中奖历史很尴尬。数据库备份文件不要放在 data 目录里防止被容器进程写坏放到宿主机另一个路径。5.6 排查技巧小结这几个问题总结下来核心就一条把每个环节都记录下来。不要只记异常日志也要记正常路径的关键节点。我习惯在消息处理流水线里加node1/2/3这样的日志字段一条消息只输出一行结构化日志格式是时间、event_id、node、action、耗时。这样日志既不会太多又能快速定位。多花十分钟加日志能省下后面数小时的排查时间这笔账非常划算。6. 从机器人到 LuckyTri我提炼出的三个通用能力6.1 配置驱动的指令注册可以迁移到任意 IM整个项目做完最让我有成就感的是框架部分可以抽出来继续用。LuckyTri 定义了一个MessageEvent抽象QQ群只是第一个接入端。只要另一个平台的事件能映射成同样的结构触发层、路由层、执行层完全不用改。换句话说今天我做的是一个QQ群聊天机器人但底层已经是一套通用的消息处理框架。以后接别的入口只需要在接入层做适配成本会小很多。6.2 事件去重、冷却、权限三件套可以直接复用这三个功能不只在群聊里有用。做自动提醒、定时播报、监控报警机器人时同样需要解决重复通知、频率限制、操作者授权的问题。我在 LuckyTri 里把它们做成了三个独立的小模块没有跟任何QQ专用逻辑绑定。下次另一个项目要写消息推送直接把这三个模块拿过去用就行不用重新造轮子。6.3 给业余项目写测试收益比想象中大写测试最大的障碍是“没有真实群聊环境”但用好 mock 事件之后这个障碍就消失了。我现在的习惯是每个新功能必须带两条测试一条验证正常输入一条验证越权输入。这样即使几个月不去动代码再提笔时跑一遍 pytest 就能知道有没有改坏。真群里的反馈永远是最慢的不能作为回归测试手段。最后再分享一个小细节LuckyTri 的版本号一直停留在 0.4.0因为我始终把它当作一个在真实环境里不断打磨的实验品。每当群友反馈一个新需求我会先问一句这是不是必须由机器人解决的如果不是就不加。这比任何架构设计都更能让一个个人项目持续活下去。希望这篇从QQ群聊天机器人开始的记录也能给你一点启发。