这个机器人叫 LuckyTri是我自己花了两三周时间做完并在几个群里跑了半年多的一个 QQ 群聊天机器人。可能有人问现在群里已经有很多现成 Bot 了为什么还要自己写一个这个问题我一开始也问过自己。后来发现现成的 Bot 要么功能太固定要么数据不在自己手里想改一个功能要等别人更新你还不知道他哪天停止维护。既然自己会用一点代码不如自己动手做一个。这半年踩了不少坑也攒了一些心得。这篇文章我不打算写成一板正经的说明书而是想按照我实际的开发顺序把这个项目从动机、设计、实现到部署、排查完整捋一遍。无论你是想给班级群、朋友群做个签到提醒机器人还是单纯想学习消息机器人的开发套路应该都能从中找到直接能用的东西。1. 从为什么做说起LuckyTri 是怎么诞生的1.1 最初的动机一个被海量消息淹没的群我最早产生想法是因为自己管理的一个同好群。群不大几十个人但消息刷得飞快。每天有人发签到、发资源、发问题重要的信息比如活动链接、公告、文件往往几分钟就被顶上去了。有人晚进群想找之前的资料根本翻不到管理员想统计一下谁经常冒泡、谁长期潜水也完全没头绪。其实这些需求并不复杂本质上就是把重复劳动自动化但群里没有现成工具能做到。市面上确实有功能很全的机器人我也试过几个。它们大多能做到关键词回复、入群欢迎、定时水群但问题也很明显一是功能是别人设计的我想要的连续签到天数群友活跃度排行自定义幸运签文案这些东西要么没有要么做得很粗糙二是数据完全不在我手里想导出群友的签到记录做个简单分析都做不到。这时候我才意识到与其等一个永远不存在的完美 Bot不如自己写一个足够贴合需求的。LuckyTri 这个名字就是这么来的Lucky 取幸运的意思Tri 代表我最先想做的三个核心模块签到、抽奖、每日运势合在一起就是一个能给群聊带来一点小幸运的机器人。1.2 定位做一个工具箱而不是话痨动手之前我先给自己定了几条规矩避免项目做到一半失控。第一条是克制机器人不应该过度发言除非有人在群里主动唤起它或者定时任务真的有必要推送否则保持沉默。很多机器人死掉就是因为太吵天天广告式推送群主不烦群友也烦最后只能被踢出去。第二条是模块化把签到、抽奖、运势、群管理辅助做成彼此独立的模块每个模块只负责一件事以后想加新功能不用动老代码。第三条是数据可追溯所有关键操作都留记录签到记录、抽奖记录、指令调用日志都要能查这样出问题的时候能快速定位。定位成工具箱还有个好处就是不会陷入和群友无限聊天的泥潭。真正好用、能长期存活在群里的机器人往往是那种你需要它的时候喊一声它马上出现的工具而不是一个话痨式 AI。LuckyTri 的核心价值是让人少做重复事想查规则发个指令就出来想知道今天运势发个指令就返回管理员想清理关键词刷屏它可以自动提醒。做到这些机器人就已经在群里站稳了脚跟。2. 技术选型与整体搭建思路2.1 接入方案选型我为什么选 WebSocket 而不是 Webhook做 QQ 群机器人第一步是解决消息怎么收发的问题。我当时主要对比了两种常见接入方式WebSocket 长连接和 Webhook 回调。它们都能把群里的消息事件实时推给我的服务端但差别还挺明显。我这里用一张表格说明当时的选择依据。对比维度WebSocketWebhook连接方向服务端主动连平台保持长连接平台回调我的公网 HTTP 接口服务器要求不需要公网 IP出站连接即可需要公网可访问的 HTTPS 地址实时性高消息秒级到达依赖回调整体也算即时稳定性断线后我可以自动重连平台回调失败时依赖平台重试策略开发复杂度只需处理一个连接 心跳需要额外处理签名、鉴权、响应超时我最后选了 WebSocket。原因很实际我不打算把机器人服务部署在一台有固定公网 IP 的服务器上更不想去折腾域名证书。用 WebSocket 的话只要服务器能主动访问外网就行断了它还能自动重连运维成本低很多。如果你本身就有公网服务器而且喜欢 REST 风格选 Webhook 也完全可行只是要把回调接口的鉴权和幂等做得更仔细一些。LuckyTri 的服务端我用了 Python 来写主要图它生态好、写起来快。异步这块我最初考虑用原生 asyncio后来发现要管的事情不少干脆用了 FastAPI 作为 Web 框架通过它的 WebSocket 能力来实现长连接管理。如果你对 Node.js 更熟用 Node 写也是完全没问题的语言不是关键思路才是。2.2 消息处理链路从群消息到指令执行无论用哪种语言消息机器人的核心都是一个处理链路。我画了一下 LuckyTri 的骨架大概是这样的收到事件 - 解析事件类型 - 基础过滤 - 指令匹配 - 执行业务逻辑 - 构造回复 - 检查限流 - 发送消息。别看链路不长每一步都可能出幺蛾子。基础过滤这一步特别容易忽略。群里有很多机器人之间互动的场景或者某个群友发了一长串文本里面碰巧包含了触发前缀如果不做过滤机器人就会莫名其妙被唤醒。我的做法是先判断事件类型是不是群消息再看消息是否以指令前缀开头还要排除机器人自己发出的消息避免自触发。只有全部通过了才进入指令匹配阶段。指令匹配实质上是把文本解析成命令名 参数列表LuckyTri 统一用斜杠开头比如 /sign、/lucky、/draw这样群友容易记代码里也好判断。还有一个很重要的细节是回复的异步处理。早期写 LuckyTri 的时候我在事件回调里直接同步发消息结果遇到数据库查询稍慢整个连接就卡住了体验很差。后来我把收到事件和发送回复解耦了收到事件后立刻返回业务逻辑放到消息队列里异步执行这样就算某个功能崩溃了也不会影响消息接收。实践下来这一改动对稳定性提升特别大。2.3 数据存储SQLite 起步量产再考虑换库LuckyTri 的数据量其实不大几百人的群每天签到几千条记录抽奖、运势日志再多一些撑死了也就几万条。这种规模完全不需要一上来就搞一套 MySQL 集群我更推荐 SQLite 起步。它零配置、单文件、备份就是一个文件名拷走对个人项目来说极其友好。不过用 SQLite 有几个注意点。第一是要开 WAL 模式不然并发读写的时候很容易出现 database is locked 错误。第二是不要把所有逻辑都塞进一个巨大的表里能分表就分表比如签到记录表、抽奖记录表、配置表分开查询会清晰很多。第三是给常用的查询字段建好索引比如签到表里的 user_id 和 sign_date没有索引的话连续签到判断会随着数据量增长越跑越慢。等哪天数据量真的大到 SQLite 扛不住了再平滑迁移到 MySQL 或者 PostgreSQL 也不迟到时候改一下数据库连接层就行。配置管理我也简单粗暴一份 YAML 配置文件存机器人 Token、指令前缀、群白名单这些静态信息数据库里只存用户状态和业务数据。Token 这种敏感信息不要硬编码在代码里也不建议直接写进 YAML 提交到仓库最好用环境变量注入我实际部署的时候就是这么做的安全上有保障。2.4 整体目录结构我的 LuckyTri 目录结构大概长这样维护了半年多觉得还挺顺手的luckytri/ ├── bot.py # 入口启动事件监听 ├── config.yaml # 静态配置Token、前缀、白名单 ├── handlers/ │ ├── message.py # 消息预处理与指令路由 │ ├── sign.py # 签到模块 │ ├── lucky.py # 每日运势模块 │ ├── draw.py # 抽奖模块 │ └── admin.py # 群管理辅助 ├── services/ │ ├── db.py # 数据库连接与建表 │ ├── sender.py # 消息发送封装 │ └── scheduler.py # 定时任务 ├── utils/ │ ├── filters.py # 消息过滤 │ └── logger.py # 日志配置 └── tests/ └── test_handlers.py # 简单单元测试一开始我没分这么细全部揉在 bot.py 里后来每加一个功能就要动主文件改出 bug 的概率直线上升。如果你也想做类似项目我建议从第一天就按功能分目录哪怕每个文件只有几十行也比一坨强。3. 从零搭建 LuckyTri核心功能是怎么一步步实现的3.1 第一步建立连接先让机器人活着接入平台这一步本质上就是一个 WebSocket 客户端连上去然后在一个循环里处理服务端推送的事件。用 Python 伪代码表示大概是这样的具体字段名称和接口地址以你接入的平台文档为准import asyncio import json import websockets TOKEN 从开放平台后台获取的令牌 async def on_event(event): print(收到事件:, json.dumps(event, ensure_asciiFalse)) # 这里先不做事只证明通了 async def main(): headers {Authorization: fBearer {TOKEN}} # 实际接入地址以官方文档为准这里是示意 async with websockets.connect( wss://robot.example.com/ws, extra_headersheaders, ) as ws: while True: message await ws.recv() event json.loads(message) await on_event(event)这一步最重要的不是代码而是验证连接是否稳定。我只做了一件事把所有收到的事件 JSON 原样打日志连着跑了一个小时看有没有断连、有没有重复推送、有没有异常数据。不要小看这个步骤很多问题早发现一个小时后面能省一整天的排查时间。3.2 第二步指令路由机器人开始听懂人话连接通之后我做的第一件事不是写签到而是实现一个干净的指令路由。指令路由的核心是把字符串解析成命令和参数然后分发给对应的处理函数。示意代码如下PREFIX / def parse_command(text: str): if not text.startswith(PREFIX): return None body text[len(PREFIX):].strip() parts body.split() if not parts: return None return parts[0].lower(), parts[1:] async def route_message(event): text event.get(text, ) result parse_command(text) if result is None: return cmd, args result if cmd in (sign, 签到): await handle_sign(event) elif cmd in (lucky, 运势): await handle_lucky(event) elif cmd in (draw, 抽奖): await handle_draw(event, args) # 其他命令走默认处理这里我特意让签到支持别名比如既支持 /sign 也支持 /签到。很多群友并不是程序员他们记不住英文命令中文别名能显著降低使用门槛。还有一个小经验是命令名一律转小写避免群友大小写混用导致匹配失败。另外我还做了一个群白名单过滤只允许配置好的群 ID 使用机器人避免机器人被人拉到莫名其妙的群里。这个在早期测试阶段特别重要不然沙箱环境里乱拉可能触发平台方风控。3.3 第三步签到、运势、抽奖三个模块的实现细节签到是 LuckyTri 的门面也是数据量最大的模块。需求很简单群友每天发 /sign机器人回复今日签到成功这是你连续第 N 天签到如果已经签过就回复今天已经签过了。实现逻辑可以分为三步import datetime def handle_sign(event): user_id event[user_id] today datetime.date.today().isoformat() # 1. 查询今天有没有签到记录 if already_signed(user_id, today): return 今天已经签到过了明天再来吧 # 2. 查昨天的记录判断是否连续 yestoday (datetime.date.today() - datetime.timedelta(days1)).isoformat() if already_signed(user_id, yestoday): streak get_streak(user_id) 1 else: streak 1 # 3. 写入签到记录并回复 save_sign_record(user_id, today) return f签到成功你已经连续签到 {streak} 天看起来很简单但有一个坑连续签到判断要小心时区。机器人服务器如果用的不是北京时间那昨天和今天的边界就完全乱了。我因为在海外服务器上跑过一段时间吃过这个亏后来统一把数据库里的日期字段存成UTC 日期 北京时区日期双份签到判断全部用北京时区才算彻底稳定。每日运势模块思路是伪随机加分级文案。我会准备一批文案分成大吉、中吉、小吉、平平、注意五个等级每个等级有若干条不同的祝福语。然后用用户 ID 加当天日期拼出一个稳定的哈希再把哈希映射到等级上。这样做的巧妙之处在于同一个群友在同一天内无论发多少次 /lucky结果都是一样的想刷都刷不出不同结果而不同群友之间的运势又各不相同。import hashlib def get_daily_lucky(user_id: str, date_str: str): seed f{user_id}:{date_str} # 这里用哈希而不是 random是为了保证一天内结果稳定 digest hashlib.md5(seed.encode()).hexdigest() num int(digest[:8], 16) % 100 if num 15: return 大吉今天会有意外之喜多留意身边小事 elif num 35: return 中吉适合推进搁置已久的事情 elif num 60: return 小吉心情好运气就不会太差 elif num 85: return 平平平稳度过别折腾大事 else: return 注意今天做决定前先喝口水冷静三分钟抽奖模块是这三个里面唯一带概率和库存的。我设计了一个权重池每个奖品有不同的权重比如普通奖励权重高、稀有奖励权重低。从实现上避免用先 random 判断稀有然后再随机选奖品这种两次随机而是直接把所有奖品放一个池子里按权重抽取保证数学上一致且实现简单import random PRIZE_POOL [ {name: 谢谢参与, weight: 70}, {name: 群红包 1 元, weight: 20}, {name: 定制头衔 1 天, weight: 8}, {name: 神秘大奖, weight: 2}, ] def draw_prize(): total sum(item[weight] for item in PRIZE_POOL) r random.uniform(0, total) upto 0 for item in PRIZE_POOL: upto item[weight] if r upto: return item[name]为了保证抽奖不过度刷屏我给每个群友设置了一个 10 分钟冷却时间。这个冷却时间我用了一张独立的冷却记录表记录 user_id 和上次抽奖时间戳判断时只查最近一条就行。如果你不想用表用内存里的字典加过期时间也可以但进程重启就会丢建议还是落库。3.4 第四步定时任务与群管理辅助定时任务是让机器人活起来的关键。我用的是 Python 的调度库每天早上 8 点定时推送一句群提醒每周日晚汇总本周签到排行榜。配置定时任务时最简单的方式是直接写上 cron 表达式我在配置文件里预留了一个 cron 列表后续加定时任务不需要改代码只改配置就行。定时任务的推送策略值得单独说一下。QQ 群里如果机器人每天都主动发消息很容易让群友产生这个机器人话真多的厌烦感。我在调定时任务时做过几次减法从每天三次改成每天一次最终只保留一次早间问候和一次周报。实践下来群友的接受度明显更高。主动消息一定要克制这是机器人能不能活过三个月的关键。群管理辅助方面我做了一个新人入群欢迎 关键规则发送的小功能。机器人收到入群事件后只在群里发一条欢迎语并附上群规关键词。这个功能逻辑不难但要注意和签到模块一样的问题事件处理必须幂等如果入群事件重复推送不能发两次欢迎语否则群友会觉得这机器人脑子不太灵光。4. 部署运维实录把机器人放进生产环境4.1 服务器选型与进程守护LuckyTri 的负载非常轻一台 1 核 2G 的云主机就绰绰有余。我实际部署用的就是这种最低配机器跑 Python 进程加上 SQLiteCPU 日常连 5% 都用不到。选服务器唯一要注意的是网络稳定性因为 WebSocket 长连接对网络的抖动比较敏感。进程守护工具我强烈推荐用 systemd。很多人喜欢用 nohup 或 screen 挂后台但进程一崩就没人知道了。systemd 可以自动重启、写标准日志、随开机启动是 Linux 上的正道。我的 systemd 服务文件长这样[Unit] DescriptionLuckyTri Bot Service Afternetwork.target [Service] Userluckytri WorkingDirectory/opt/luckytri ExecStart/opt/luckytri/venv/bin/python bot.py Restartalways RestartSec5 EnvironmentFile/etc/luckytri.env [Install] WantedBymulti-user.target几个关键配置说下Restartalways 表示进程退出就自动拉起RestartSec5 防止频繁重启把日志刷爆EnvironmentFile 用来注入 Token 等环境变量。如果你习惯容器化也可以用 Docker 加 docker-compose但要注意把数据目录挂载到宿主机不然容器一删数据全没了。我用 systemd 是因为这个项目足够轻再加一层容器反而显得重。4.2 日志排查问题的唯一线索日志做得好不好直接决定半夜被叫起来排查时的心态。我最初只在代码里零星打印几个 print出了 bug 基本靠猜。后来专门给 LuckyTri 配了一套结构化日志每条日志包含时间、级别、模块、消息内容。关键事件至少要打三条日志收到指令、执行结果、回复内容以及异常栈。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, ) logger logging.getLogger(luckytri) async def handle_sign(event): logger.info(收到签到指令 user%s group%s, event[user_id], event[group_id]) try: reply do_sign(event) logger.info(签到执行成功 user%s reply%s, event[user_id], reply) except Exception: logger.exception(签到执行异常 user%s, event[user_id])日志文件我会做轮转用 Python logging 自带的 RotatingFileHandler单个文件达到 5MB 就自动转存保留最近 5 份。这样日志不会无限膨胀出问题时又有足够的历史记录可以翻。4.3 数据备份与一致性SQLite 备份特别简单但也不能直接把正在写库的文件拷走。我写了一个每天凌晨的备份脚本先用 SQLite 的在线备份接口把当前数据库备份到一个临时文件再压缩成带日期的 gz 包最后按保留策略删掉 30 天前的旧备份。整套流程用一个 server 级 crontab 定时跑稳如老狗。备份最容易被忽略的是备份数据是否可用。我隔一段时间会拿一个备份文件恢复到临时目录跑一遍查询确认数据不是坏的。别笑话这种备份了但恢复不了的悲剧在运维圈真的不少见。另外如果有条件把备份文件再同步一份到另一台机器上防止服务器整个挂掉。4.4 资源占用与基础调优跑了一段时间后我发现LuckyTri 有两个地方消耗资源比较明显。一个是日志模块如果开启 DEBUG 级别高频群的消息日志量会非常大刷得磁盘 IO 飙升另一个是数据库连接我一开始每次查询都新建连接后来改用连接复用性能提升很明显。还有一个调优点是在回复发送前做限流。比如抽奖冷却已经限制了单个用户但群友可能同时触发多个指令机器人会同时发好几条消息观感很差。我在发送模块里加了一个简单的令牌桶每秒最多发 5 条消息超出就排队。这样即使遇到指令风暴群聊也不会被刷屏。这个限流在上线后非常有用强烈建议加上。5. 常见问题与排查技巧实录5.1 收不到消息、断连了怎么办我遇到过最诡异的问题就是 WebSocket 时不时断开客户端并没有报错但消息就是收不到。排查后发现是平台侧的连接空闲超时需要客户端定期发心跳包。多数平台支持客户端发送 ping 帧或者业务层心跳消息我在代码里用一个定时任务每 30 秒发一次心跳之后再也没有无故断连过。如果你也遇到连接不稳定建议先看日志里有没有重连记录然后把重连间隔做指数退避从 1 秒、2 秒、4 秒逐步拉长避免断连后集中重连把服务器打挂。还有一点本地开发调试时记得检查防火墙和代理有些网络环境会拦截 WebSocket 出站连接当时我用公司网络测试死活连不上换手机热点一下就通了。5.2 消息重复推送幂等处理消息重复这个坑我是在做新人欢迎功能时踩到的。运营群里出现了一次欢迎语发了两遍的情况排查之后发现是平台在回调超时后自动重试了一次同一事件被推送了两遍。解决办法很简单给每个事件建一个 ID维护一张已处理事件表收到事件后先查重、再处理、再记录保障幂等。def is_duplicate(event_id: str) - bool: if exists_event_id(event_id): return True save_event_id(event_id) return False事件 ID 表不用留太多数据我保留最近 1 万条就够用了定期清理旧的。另外这个表的大小一定要控制不然会变成一个持续膨胀的隐患。5.3 昵称解析、消息与权限判断群里消息经过平台推送到我这里之后文本里出现的群友昵称和 符号往往不是你在客户端看到的那种原始文本而是带协议格式的编码。我在做关键词功能时就需要解析出真实昵称早期没搞清楚直接把带格式的字符串拿来做匹配导致关键词永远对不上。这个问题没有捷径只能靠打日志观察实际推送的原始 JSON再针对格式做解析。权限判断也是个容易出问题的地方。LuckyTri 里有一些管理指令比如设置群规、清空签到记录这些只有管理员可以用。我的做法是在配置里维护一个管理员 QQ 号列表指令执行前先校验发送者是否在列表内不在就礼貌地拒绝。不要用客户端是不是管理员的字段来判断因为机器人收到的权限字段可能和你想象的不一样白名单最可靠。5.4 内容安全与平台合规这是所有做消息机器人的人都必须重视的一环。我在 LuckyTri 里做了一层文本过滤所有群友发来的消息在进入业务逻辑之前都会过一遍敏感词表命中就直接忽略不处理、不回复。这既是为了群聊环境也是为了避免机器人被恶意利用发不合适的内容。发送侧同样要注意内容合规。机器人的回复文案我全部写在配置文件或代码里不使用任何动态拼接的敏感语言。主动推送的频率也做了严格限制防止出现骚扰。还有一点容易被忽视不要在多个群里用同一个机器人发一模一样的高频内容这种行为在平台侧极易触发风控导致机器人直接被限制。我见过有人的机器人因为疯狂群发最后账号都被处理了这个教训要记住。为了帮助排查我把常见问题整理成了一个速查表方便读者对照症状可能原因处理建议收不到群消息WebSocket 断连/超时检查心跳与重连机制看网络出站是否正常消息重复处理平台回调重试用事件 ID 做幂等去重回复延迟高同步发送阻塞改成异步队列处理先收后发数据库 lockedSQLite 并发写开启 WAL 模式减少连接数指令偶尔失灵文本格式含协议编码打日志查看原始消息做格式解析机器人被风控高频主动消息降低推送频率控制指令频次5.5 后续扩展变成插件化、接入大模型LuckyTri 目前的架构虽然分了模块但还不够插件化。我下一步的理想是把每个功能做成一个独立插件包通过配置文件动态加载这样别人拿到手之后直接丢一个目录进 plugins 就能扩展新功能不需要改主程序。这个设计其实不复杂就是每个插件约定好接口一个 handler 函数、一份元信息命令名、描述、权限。如果你也想做建议从设计插件接口开始而不是等代码写多了再重构。另外最近群里很多人希望机器人能聊天而不是只会执行命令。我打算在 LuckyTri 里接入一个大模型接口做成一个专门的对话插件并且做严格的触发条件只有群友用特定的指令唤起时才会进入对话模式返回的内容也会经过一层安全过滤。这样既满足趣味性又不会让机器人变成一个不受控的聊天狂魔。6. 一路测试下来我的一些体会半年的运维让我明白一件事做聊天机器人技术上的难点其实不多难的是克制。什么时候不回复、什么时候主动说话、每天推送几条内容这些产品层面的决策比任何技术实现都更影响机器人的存活率。LuckyTri 中间有一版加了每日早报功能推送了一周之后就有群友私聊我说太吵了我第二天就把它下线了。做工具最重要的不是功能多而是好用且不烦人。如果你也想从零做一个自己的群聊机器人我的建议是第一先跑通最小闭环哪怕只有一个签到功能也要完整经历收消息、处理、回消息全流程第二日志一定要从第一天就好好打不要等出了 bug 再补第三主动消息永远要往少了调先默默无闻地活下来再考虑增彩。这个项目给我最大的收获不是代码写了多少行而是亲手把一个想法从零变成能长期运行的东西那个成就感很难被替代。