从零开发QQ群聊天机器人LuckyTri：WebSocket架构与部署实战

这个机器人叫 LuckyTri，是我自己花了两三周时间做完并在几个群里跑了半年多的一个 QQ 群聊天机器人。可能有人问，现在群里已经有很多现成 Bot 了，为什么还要自己写一个？这个问题我一开始也问过自己。后来发现，现成的 Bot 要么功能太固定，要么数据不在自己手里，想改一个功能要等别人更新，你还不知道他哪天停止维护。既然自己会用一点代码，不如自己动手做一个。

这半年踩了不少坑，也攒了一些心得。这篇文章我不打算写成一板正经的说明书，而是想按照我实际的开发顺序，把这个项目从动机、设计、实现到部署、排查，完整捋一遍。无论你是想给班级群、朋友群做个签到提醒机器人，还是单纯想学习消息机器人的开发套路，应该都能从中找到直接能用的东西。

1. 从"为什么做"说起：LuckyTri 是怎么诞生的

1.1 最初的动机：一个被海量消息淹没的群

我最早产生想法是因为自己管理的一个同好群。群不大，几十个人，但消息刷得飞快。每天有人发签到、发资源、发问题，重要的信息比如活动链接、公告、文件，往往几分钟就被顶上去了。有人晚进群想找之前的资料，根本翻不到；管理员想统计一下谁经常冒泡、谁长期潜水，也完全没头绪。其实这些需求并不复杂，本质上就是"把重复劳动自动化"，但群里没有现成工具能做到。

市面上确实有功能很全的机器人，我也试过几个。它们大多能做到关键词回复、入群欢迎、定时水群，但问题也很明显：一是功能是别人设计的，我想要的"连续签到天数""群友活跃度排行""自定义幸运签文案"这些东西，要么没有，要么做得很粗糙；二是数据完全不在我手里，想导出群友的签到记录做个简单分析都做不到。这时候我才意识到，与其等一个永远不存在的完美 Bot，不如自己写一个足够贴合需求的。LuckyTri 这个名字就是这么来的，Lucky 取"幸运"的意思，Tri 代表我最先想做的三个核心模块：签到、抽奖、每日运势，合在一起就是一个能给群聊带来一点小幸运的机器人。

1.2 定位：做一个"工具箱"，而不是"话痨"

动手之前，我先给自己定了几条规矩，避免项目做到一半失控。第一条是克制：机器人不应该过度发言，除非有人在群里主动唤起它，或者定时任务真的有必要推送，否则保持沉默。很多机器人死掉就是因为太吵，天天广告式推送，群主不烦群友也烦，最后只能被踢出去。第二条是模块化：把签到、抽奖、运势、群管理辅助做成彼此独立的模块，每个模块只负责一件事，以后想加新功能不用动老代码。第三条是数据可追溯：所有关键操作都留记录，签到记录、抽奖记录、指令调用日志都要能查，这样出问题的时候能快速定位。

定位成"工具箱"还有个好处，就是不会陷入和群友无限聊天的泥潭。真正好用、能长期存活在群里的机器人，往往是那种"你需要它的时候喊一声，它马上出现"的工具，而不是一个话痨式 AI。LuckyTri 的核心价值是让人少做重复事：想查规则，发个指令就出来；想知道今天运势，发个指令就返回；管理员想清理关键词刷屏，它可以自动提醒。做到这些，机器人就已经在群里站稳了脚跟。

2. 技术选型与整体搭建思路

2.1 接入方案选型：我为什么选 WebSocket 而不是 Webhook

做 QQ 群机器人，第一步是解决"消息怎么收发"的问题。我当时主要对比了两种常见接入方式：WebSocket 长连接和 Webhook 回调。它们都能把群里的消息事件实时推给我的服务端，但差别还挺明显。我这里用一张表格说明当时的选择依据。

我最后选了 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_ascii=False)) # 这里先不做事，只证明通了 async def main(): headers = {"Authorization": f"Bearer {TOKEN}"} # 实际接入地址以官方文档为准，这里是示意 async with websockets.connect( "wss://robot.example.com/ws", extra_headers=headers, ) 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(days=1)).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 进程加上 SQLite，CPU 日常连 5% 都用不到。选服务器唯一要注意的是网络稳定性，因为 WebSocket 长连接对网络的抖动比较敏感。

进程守护工具我强烈推荐用 systemd。很多人喜欢用 nohup 或 screen 挂后台，但进程一崩就没人知道了。systemd 可以自动重启、写标准日志、随开机启动，是 Linux 上的正道。我的 systemd 服务文件长这样：

[Unit] Description=LuckyTri Bot Service After=network.target [Service] User=luckytri WorkingDirectory=/opt/luckytri ExecStart=/opt/luckytri/venv/bin/python bot.py Restart=always RestartSec=5 EnvironmentFile=/etc/luckytri.env [Install] WantedBy=multi-user.target

几个关键配置说下：Restart=always 表示进程退出就自动拉起，RestartSec=5 防止频繁重启把日志刷爆，EnvironmentFile 用来注入 Token 等环境变量。如果你习惯容器化，也可以用 Docker 加 docker-compose，但要注意把数据目录挂载到宿主机，不然容器一删数据全没了。我用 systemd 是因为这个项目足够轻，再加一层容器反而显得重。

4.2 日志：排查问题的唯一线索

日志做得好不好，直接决定半夜被叫起来排查时的心态。我最初只在代码里零星打印几个 print，出了 bug 基本靠猜。后来专门给 LuckyTri 配了一套结构化日志：每条日志包含时间、级别、模块、消息内容。关键事件至少要打三条日志：收到指令、执行结果、回复内容，以及异常栈。

import logging logging.basicConfig( level=logging.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 里做了一层文本过滤：所有群友发来的消息在进入业务逻辑之前，都会过一遍敏感词表，命中就直接忽略，不处理、不回复。这既是为了群聊环境，也是为了避免机器人被恶意利用发不合适的内容。

发送侧同样要注意内容合规。机器人的回复文案我全部写在配置文件或代码里，不使用任何动态拼接的敏感语言。主动推送的频率也做了严格限制，防止出现骚扰。还有一点容易被忽视：不要在多个群里用同一个机器人发一模一样的高频内容，这种行为在平台侧极易触发风控，导致机器人直接被限制。我见过有人的机器人因为疯狂群发，最后账号都被处理了，这个教训要记住。

为了帮助排查，我把常见问题整理成了一个速查表，方便读者对照：

5.5 后续扩展：变成插件化、接入大模型

LuckyTri 目前的架构虽然分了模块，但还不够"插件化"。我下一步的理想是把每个功能做成一个独立插件包，通过配置文件动态加载，这样别人拿到手之后直接丢一个目录进 plugins 就能扩展新功能，不需要改主程序。这个设计其实不复杂，就是每个插件约定好接口：一个 handler 函数、一份元信息（命令名、描述、权限）。如果你也想做，建议从设计插件接口开始，而不是等代码写多了再重构。

另外，最近群里很多人希望机器人能"聊天"，而不是只会执行命令。我打算在 LuckyTri 里接入一个大模型接口，做成一个专门的对话插件，并且做严格的触发条件：只有群友用特定的指令唤起时，才会进入对话模式，返回的内容也会经过一层安全过滤。这样既满足趣味性，又不会让机器人变成一个不受控的聊天狂魔。

6. 一路测试下来，我的一些体会

半年的运维让我明白一件事：做聊天机器人，技术上的难点其实不多，难的是"克制"。什么时候不回复、什么时候主动说话、每天推送几条内容，这些产品层面的决策，比任何技术实现都更影响机器人的存活率。LuckyTri 中间有一版加了每日早报功能，推送了一周之后，就有群友私聊我说太吵了，我第二天就把它下线了。做工具，最重要的不是功能多，而是好用且不烦人。

如果你也想从零做一个自己的群聊机器人，我的建议是：第一，先跑通最小闭环，哪怕只有一个签到功能，也要完整经历"收消息、处理、回消息"全流程；第二，日志一定要从第一天就好好打，不要等出了 bug 再补；第三，主动消息永远要往少了调，先默默无闻地活下来，再考虑增彩。这个项目给我最大的收获，不是代码写了多少行，而是亲手把一个想法从零变成能长期运行的东西，那个成就感很难被替代。