三个月前,我被一个整天刷屏的签到机器人搞得有点烦。每到整点,它就准时把几十条抽奖结果刷满整个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_level="member", cooldown=10) 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_id="test-001", group_id="grp-dev", user_id="user-test", raw_message="/运势", sender_role="member", ) 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_LEVEL=INFO 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 排查技巧小结
这几个问题总结下来,核心就一条:把每个环节都记录下来。不要只记异常日志,也要记正常路径的关键节点。我习惯在消息处理流水线里加node=1/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群聊天机器人开始的记录,也能给你一点启发。