简介基于Nonebot2的QQ群机器人插件核心亮点是引入TextRank等机器学习算法对群内每日聊天记录进行自动摘要与关键词总结可大幅节省人工回顾聊天内容的时间。面向计算机、人工智能、通信工程等专业的在校学生及开发者适合用于毕设、课设或功能扩展。压缩包共19个文件以11个Python源码为主覆盖插件加载、文本分割、停用词过滤、群组规则判定、数据记录等完整模块另含算法讲解PDF、运行效果图、停用词表及说明文档整体约1.91MB。代码全部测试通过可直接部署运行TextRank算法演示部分能直观展示关键词提取与摘要生成的流程便于二次开发比如自定义数据源、调整摘要长度或接入更多聊天平台。已有545人学习浏览适合希望快速上手Nonebot2插件开发或借鉴聊天记录分析思路的读者。1. 先别急着训模型Nonebot2 的每日总结插件难点其实在数据侧标题挂着“机器学习算法”但你把 QQ 群机器人插件真正跑起来就会发现模型选型永远不是第一个瓶颈。基于 Nonebot2 的 QQ 群机器人插件核心链路是“监听群消息 → 清洗落库 → 定时聚合 → 生成每日总结 → 推回群里”机器学习算法只占其中一小段。真正让新人翻车的是事件监听被其他插件抢走、消息里混着 CQ 码和复读、定时任务时区不对以及总结发出去被群友嫌“不像人话”。这篇文章按我自己的落地路径写先搭最小插件骨架收数据再选 TF-IDF 加时间窗切话题最后聊推送和排障。目标不是堆模型是让你用最小成本把“每日总结”做成一个群友愿意看的群日报。适合正在用 Nonebot2 写插件、或者准备从零接 QQ 群机器人的开发者。2. 用 Nonebot2 搭 QQ 群机器人插件骨架事件响应、文本过滤与 SQLite 落库2.1 插件的最小结构on_message 事件响应器与 priority/block 参数Nonebot2 的插件本质上就是一个 Python 包放在项目配置好的插件目录下。我一般习惯把单个插件做成一个目录目录里放__init__.py这样后续加配置、加数据文件都方便不会把代码全堆在一个文件里。最基础的事件响应器是on_message它监听所有会话消息群消息和私聊消息都能收到所以要先在参数里决定自己要不要区分场景。# plugins/daily_summary/__init__.py from nonebot import on_message from nonebot.adapters.onebot.v11 import GroupMessageEvent, Bot, Message msg_recorder on_message(priority10, blockFalse) msg_recorder.handle() async def record_msg(bot: Bot, event: GroupMessageEvent): # 只处理群消息且不做任何回复记录完就放行 print(f群 {event.group_id} 收到来自 {event.user_id} 的消息)这里有两个参数必须调明白priority和block。Nonebot2 的事件响应机制是按优先级从低到高依次调用处理器默认priority1、blockTrue的插件会把消息完全拦住后面优先级更低的插件就收不到这条消息了。想做“旁路记录”类的插件正确姿势是把priority设得靠后一点比如 10同时blockFalse这样自己处理完消息事件流还能继续往后面的插件走。这个参数组合不调对就会出现两个插件打架你的总结插件把消息全拦了群里其他功能全部哑火。实际开发里我见过太多次“我的插件不生效”“别的插件突然不响应”最后查出来是priority和block配错。这个插件因为是纯记录、不做任何回复所以 handler 里不需要调用send相关的方法记完就结束。跑通这一步你已经有了一个不会打扰任何人的“群监听器”后续所有数据都从这里来。2.2 清洗与落库把 CQ 码、图片与指令词挡在统计之外监听消息只是第一步真正决定每日总结质量的是落库这一步。QQ 群消息里只有一部分是真正的文本内容图片会以 CQ 码形式出现如[CQ:image,filexxxx]]也可以不是连续出现而是以“图片文字表情”形式混在一条消息里。所以清洗时要把非 text 段去掉再把 text 段拼接起来。import sqlite3 from pathlib import Path DB_PATH Path(__file__).parent / daily_summary.db def to_plain_text(event: GroupMessageEvent) - str: # 遍历消息段只保留 text 段丢弃图片/表情/回复等 CQ 码 parts [] for seg in event.message: if seg.type text: text seg.data.get(text, ).strip() if text: parts.append(text) return .join(parts) def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute( CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL, user_id INTEGER NOT NULL, msg_time INTEGER NOT NULL, content TEXT NOT NULL, UNIQUE (group_id, user_id, msg_time, content) ) ) conn.execute(PRAGMA journal_modeWAL)代码里的seg.type text是 OneBot V11 适配器里判断文本段的方式图片、语音、回复等类型分别对应image、record、reply。content字段存的是清洗后的纯文本msg_time存 Unix 时间戳方便后面按天窗口切数据。表结构里刻意加了四个字段的 UNIQUE 约束防止协议端重连后重复推送同一条消息导致重复入库。落库函数我单独写不塞进事件响应器这样后面离线回放历史数据时也能复用import asyncio def insert_message_sync(group_id: int, user_id: int, msg_time: int, content: str): with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT OR IGNORE INTO messages (group_id, user_id, msg_time, content) VALUES (?, ?, ?, ?), (group_id, user_id, msg_time, content), ) async def insert_message(group_id: int, user_id: int, msg_time: int, content: str): # 用 asyncio.to_thread 避免 SQLite 同步操作阻塞事件循环 await asyncio.to_thread(insert_message_sync, group_id, user_id, msg_time, content)asyncio.to_thread在这里是必要的。SQLite 是同步 I/O如果直接在异步事件循环里高频写入群消息一多就会拖慢整个 Nonebot2 进程导致连接超时甚至掉线。量小的时候感觉不出来遇到一天几千条消息的活跃群就原形毕露了。2.3 启动钩子建表为什么初始化要放进 on_startup 而不是模块顶部初次接触 Nonebot2 的人很容易在模块顶部直接调用init_db()想着“反正这个函数建表很快”。这个做法在本地没问题但在 Nonebot2 的加载流程里可能跑在事件循环启动前某些环境下会导致数据库连接在异步上下文里被复用出现莫名的锁错误。更稳的做法是把初始化挂到启动钩子上。from nonebot import get_driver driver get_driver() driver.on_startup async def do_startup(): init_db() # 可以在这里补一句日志确认插件已加载 print(daily_summary 插件数据库初始化完成)get_driver().on_startup是 Nonebot2 标准生命周期钩子程序进程启动完成后执行一次。把建表和后续的配置加载都放到这里顺序是确定的协议端连接建立之前数据库一定已经就绪。等到机器人真正开始收消息INSERT语句就不会因为表还没建好而报错。这一步完成后你的插件已经把群消息全部收进 SQLite 了。此时先别急着写算法把机器人挂在群里跑半天然后打开数据库看一眼messages表的行数和内容质量。你会惊讶地发现一个活跃群半天就能产生几百条消息而其中真正有信息量的可能只有三成剩下的全是“哈哈哈哈哈哈”“收到”“1”和表情包轰炸。下一章要解决的就是怎么从这三成里提炼出像样的总结。3. 每日总结的机器学习算法选型TF-IDF 关键词、TextRank 边界与话题时间窗3.1 TF-IDF 关键词抽取jieba.analyse.extract_tags 的参数与停用词群聊文本和新闻、博客不一样句子极短、口语化严重、错别字多、大量人名和梗词。上来就上 BERT 或者大语言模型成本高、部署麻烦。我一般用 TF-IDF 先抽关键词理由是它不要求长文本对短消息集合也能给出可解释的权重。Python 里最省事的实现是jieba.analyse.extract_tags它把 TF-IDF 封装好了还内置了一个基础 idf 词典直接用就行。import jieba.analyse from functools import lru_cache # 停用词把高频无意义词、机器人指令词、以及“总结”本身都排除掉 STOP_WORDS { 机器人, 每日, 总结, 这个, 那个, 什么, 怎么, 哈哈, 哈哈哈, 收到, 1, 每日总结, } lru_cache(maxsize64) def extract_keywords(texts: tuple[str, ...], top_k: int 20): # 把消息列表拼成一个大文本交给 jieba 做关键词抽取 corpus \n.join(texts) tags jieba.analyse.extract_tags( corpus, topKtop_k, withWeightTrue, allowPOS(n, vn, v, nr, nt, nz), ) return [(word, float(weight)) for word, weight in tags if word not in STOP_WORDS]关键参数是allowPOS和topK。allowPOS限制词性名词、动词、人名、组织名、专有名词是群聊总结里信息量最大的几类形容词和副词基本是情绪表达对“今天聊了什么”没帮助。topK控制返回数量默认 20 够用。lru_cache是常用的性能优化手段每天定时任务只跑一次但测试时可能反复调用加缓存避免重复分词拖慢调试。这里有个容易被忽略的点extract_tags的输入是一个大字符串多消息之间用换行隔开。TF-IDF 的 IDF 部分依赖语料统计如果把多条消息拼成一个句子词频会被稀释换行分隔则每行当作独立文本统计口径更合理。群聊里表情、英文、数字混杂也不会报错jieba 对未知 token 会直接跳过。3.2 话题分组为什么短聊天记录不适合直接上 KMeans拿到一堆高权重关键词后下一步是生成“今天讨论了哪几个话题”。很多人第一反应是上聚类算法比如 KMeans。但我必须说在群聊短文本场景KMeans 是典型的“看起来对、用起来崩”的方案。短文本向量化后极度稀疏fit_transform出来的矩阵几乎全是 0聚类结果取决于离群点和 emoji而不是真正的话题结构而且 K 值没法预先确定今天聊 3 个话题明天聊 7 个K 写死就翻车。我实际用的是一种更保守的时间窗聚合群聊话题天然带有连续性同一个话题的消息通常集中在几分钟内。把一天的消息按时间排序相邻两条间隔超过阈值就切开形成若干“话题段”。WINDOW_SECONDS 10 * 60 # 10 分钟无消息视为话题切换 def split_into_topics(rows: list[dict]) - list[list[dict]]: topics [] current [] for row in rows: if current and row[msg_time] - current[-1][msg_time] WINDOW_SECONDS: topics.append(current) current [] current.append(row) if current: topics.append(current) return topics时间窗的取值直接决定话题粒度。我在活跃群试过 3 分钟话题被切得稀碎一个话题能裂成四五个段在死水群试过 30 分钟把早上的讨论和晚上的讨论粘在一起。10 分钟是比较稳的默认值你可以根据群的活跃度调整消息越密的群窗口越小消息越稀的群窗口越大。话题段切好后给每个段算一个“话题分”把段内所有消息拼起来和全局关键词做交集命中的关键词权重之和就是该段的话题分。最后按话题分从高到低取前几个段作为今天的重点讨论内容。这个逻辑比聚类可解释得多而且后续想加新维度比如统计发言人、识别链接都很容易。3.3 摘要怎么给按关键词密度挑重点消息不硬套 TextRank关键词有了话题段有了最后一步是“把总结写出来”。这里最大的误区是硬套 TextRank 做抽取式摘要。TextRank 在长文本上有效但群聊消息普遍只有几个字到十几个字句子和句子之间的语义重叠几乎为零TextRank 出来的结果往往就是“收到”“哈哈”这种高频短句参考价值不大。我的做法是对每个高话题分段保留其中“关键词密度最高”的若干条原文消息。因为群聊里真正有价值的消息通常就是那几条有人甩了个链接、有人发了段代码、有人讲了结论。把这些原文按时间顺序拎出来比任何改写都真实。def pick_highlights(topics, keywords, max_items3): # keywords 是 (词, 权重) 列表转换成词到权重的字典 weight_map dict(keywords) result [] for topic in topics: scored [] for row in topic: score 0 for word, weight in weight_map.items(): if word in row[content]: score weight scored.append((score, row)) scored.sort(keylambda x: x[0], reverseTrue) result.extend([row for score, row in scored[:max_items] if score 0]) return result这里挑选的是“整条消息包含多个高权重词”的句子直觉上就是话题的核心陈述。max_items按话题控制条数一天总结压到 5~8 条原文就够了再多群友看不下去。整条链路跑完之后你手里有当天高频关键词、按时间切出的话题分段、每个分段里最值得读的原文。这三样东西组合起来就是一份及格线以上的每日总结。4. 生成每日总结的完整工作流从读库聚合、过滤复读到定时推送4.1 读库与清洗一天几千条消息怎么压成候选行定时任务到点后第一件事是从 SQLite 里把当天的消息捞出来。这里有个很多人会漏掉的细节如果你的机器人是中途上线的数据库里可能只有最近几个小时的数据而定时任务跑在 0 点附近要拿的是“昨天”的数据。所以时间窗口必须算清楚不能简单地用date(now)切日要按你设定的推送时区来。import time import datetime def load_daily_messages(group_id: int, day_offset: int 1) - list[dict]: # 默认取昨天的消息day_offset0 时取今天 now datetime.datetime.now() target now - datetime.timedelta(daysday_offset) start_ts int(time.mktime(target.replace(hour0, minute0, second0).timetuple())) end_ts start_ts 86400 with sqlite3.connect(DB_PATH) as conn: conn.row_factory sqlite3.Row rows conn.execute( SELECT user_id, msg_time, content FROM messages WHERE group_id ? AND msg_time ? AND msg_time ? ORDER BY msg_time ASC , (group_id, start_ts, end_ts), ).fetchall() return [dict(row) for row in rows]day_offset这个参数在调试时特别好用上线初期不放心可以用day_offset0在晚上手动跑一遍“今天的总结”观察算法效果。定时推送则固定用day_offset1取昨天完整一天的数据避免今天还没结束就急着总结。清洗这一步除了过滤 CQ 码还要处理两类脏数据机器人自己发的消息以及群里的指令消息如/今日总结。机器人的消息来自event.self_id对应的 QQ 号判断方法很简单比对消息行的user_id和当前 bot 的self_id。指令消息则以command_start配置的符号开头通常默认是/。def clean_rows(rows: list[dict], bot_id: int) - list[dict]: out [] last_content None for row in rows: if row[user_id] bot_id: continue content row[content].strip() # 指令消息不进统计 if content.startswith(/): continue # 连续复读只保留第一条避免“哈哈”刷屏 if content last_content: continue out.append(row) last_content content return out连续复读过滤我用的是相邻去重一条“哈哈哈哈”后面跟十条一样的“哈哈哈哈”只保留第一条。这个规则对“1”“复读机式玩梗”特别有效但也存在误伤——如果群友在不同时间点说了同样的内容中间隔了几条别的消息不会被过滤掉。所以这里的判断标准是“是否紧挨着重复”而不是“当天是否出现过”。保留多条独立但相似的消息反而能反映这个梗今天出现了多次是有信息量的信号。4.2 总结文本的组装模板、长度控制与 emoji 使用策略清洗完的数据直接丢进第 3 章的关键词抽取和时间窗切分产出今天的重点消息后就该拼总结文本了。这个环节最考验产品感文本格式直接决定群友愿不愿意看完。我的模板分三块开头给当天的整体热度指标中间列话题和重点原文结尾放一到两个参与度最高的群友昵称可选。整段控制在 300 字内。def build_summary_text(rows: list[dict], bot_id: int) - str: rows clean_rows(rows, bot_id) if len(rows) 30: return 今天群聊样本不足明天再总结。 texts [row[content] for row in rows] keywords extract_keywords(tuple(texts), top_k20) topics split_into_topics(rows) highlights pick_highlights(topics, keywords, max_items3) lines [] lines.append(今日群聊小报) lines.append(参与消息 str(len(rows)) 条关键词热度:) kw_text 、.join(word for word, _ in keywords[:8]) lines.append( kw_text) lines.append(最有话聊的几个片段:) for hl in highlights[:5]: # 只展示原文不伪造任何结论 lines.append(· hl[content][:50]) return \n.join(lines)模板里刻意不用大段形容词也不写“今天大家讨论了”这种废话因为算法不保证对话题的理解完全正确与其写错结论被群友挑刺不如只列事实关键词、原句、条数。emoji 在总结里要克制我最多在标题行加一个不占语义的符号正文不加表情。QQ 群环境里表情渲染会导致文本长度计算偏差消息一长被服务端截断反而翻车。拼完的文本先别急着发先print()到控制台或者用logger.info打一条日志人工扫一眼再放量。这一步能筛掉九成以上的弱智输出。4.3 定时任务与手动触发用 Apscheduler 和命令各留一条路每日总结必须有一条定时推送路径同时必须有一条手动触发路径。我在实践中发现百分之百依赖定时任务会让你调试极其痛苦你想测一下算法效果得等到第二天凌晨而手动命令可以让算法和模板随时验证。手动触发的命令用 SUPERUSER 权限限制避免群友疯狂刷总结把机器人搞崩。from nonebot import require, get_bots require(nonebot_plugin_apscheduler) from nonebot_plugin_apscheduler import scheduler async def push_daily_summary(group_id: int, day_offset: int 1): rows load_daily_messages(group_id, day_offset) text build_summary_text(rows, bot_id) if not text: return bots get_bots() # 单 bot 场景直接取第一个多 bot 场景按 group_id 路由 bot next(iter(bots.values())) await bot.send_group_msg(group_idgroup_id, messageMessage(text)) scheduler.add_job( push_daily_summary, triggercron, hour23, minute50, iddaily_summary_task, args[MY_GROUP_ID, 1], timezoneAsia/Shanghai, misfire_grace_time3600, )定时参数里的timezone是重中之重。很多 Nonebot2 部署在云服务器上系统时区是 UTC如果你漏掉timezonehour23, minute50跑出来的实际是北京时间早上 7 点 50。misfire_grace_time是容错 : 如果任务因重启等原因错过了触发时间在一小时的宽限期内会补跑一次避免某天总结直接消失。手动触发我习惯用on_commandfrom nonebot.permission import SUPERUSER from nonebot.rule import to_me my_cmd on_command(今日总结, permissionSUPERUSER, priority1, blockTrue) my_cmd.handle() async def handle_cmd(): rows load_daily_messages(GROUP_ID, day_offset0) text build_summary_text(rows, BOT_ID) await my_cmd.finish(Message(text))手动命令和定时任务共用load_daily_messages和build_summary_text只在day_offset上不同定时任务取昨天手动命令取今天到目前为止。这两个路径都跑通后你的插件已经从“监听器”变成了一个真正会写日报的群功能。接下来的一大半时间会花在排障上。5. QQ 群机器人插件上线后的 5 个典型坑风控限流、自我污染与任务不触发5.1 定时总结发不出去QQ 频控与分片发送现象手动触发能正常收到总结定时任务日志也显示执行完了但群里就是没有消息。或者是连续推了两天之后突然总结发不出去机器人其他功能正常。原因单条消息内容过长或发送频率过高时QQ 服务端会触发频控导致send_group_msg返回失败。每日总结拼出来经常超过 300 字再加上群消息本身的频率很容易踩线。另一个原因是有些协议端对长文本有默认上限超长消息会直接被丢弃。解决把总结文本按行拆成几段每段不超过 150 字依次发送段与段之间asyncio.sleep一秒以上。如果当天总结特别长宁可按“关键词 重点消息 参与度”拆成三条发也不要一次性塞一条长消息。另外把推送时间定在晚上 11 点 50 而不是中午低峰期频控风险小得多。5.2 机器人把自己说过的内容统计进关键词现象每日总结里出现“收到机器人指令后的回复语”或者关键词权重最高的词是“今日总结”“小报”之类。原因on_message响应器不区分消息来源它同样会收到机器人自己发出的消息。如果脚本没过滤event.self_id机器人发的“今日总结”也会被当成群友内容写进数据库跑 TF-IDF 时“总结”这个词当然权重爆炸。解决落库时直接忽略user_id bot.self_id的消息清洗时再补一道过滤clean_rows里把bot_id参数传进去凡是row[user_id] bot_id的行一律跳过。两道过滤是因为有些历史数据已经污染只靠落库过滤救不回来清洗时再兜一次底。5.3 群复读把“哈”顶成第一关键词现象每天的关键词热度第一名永远是“哈哈”“哈哈哈哈”“笑死”真正讨论的内容全被淹掉。群成员看到总结的反应是“这总结怎么跟个憨憨一样”。原因TF-IDF 的词频统计逻辑天然偏向高频短词。“哈哈”这种词在群聊里一天出现几十次IDF 权重被拉低后仍然排在最前面因为它们占据了绝对的次数优势。这就是群聊短文本和长文档最大的差别。解决我的方案分为三层。第一层停用词表里把“哈哈”“哈哈哈”“笑死”“1”“收到”“好活”这类情绪词和复读词全部塞进去第二层用allowPOS过滤只保留名词、动名词、人名从词性上排除语气词第三层把 4.1 节的连续复读过滤加在 TF-IDF 之前先减少同类噪声的占比。三个动作叠加关键词列表才能落到实质内容上。这里没有一劳永逸的办法停用词表需要你连看三天的输出每次看到莫名高频的词就补进去属于血泪经验。5.4 事件被别的插件拦截落库静默失败现象群消息的记录功能偶尔不生效数据库里某几个时间段出现明显空洞或者插件启动后没有任何报错但messages表就是没有新增行。同时群里其他功能也能正常回复看起来一切正常。原因Nonebot2 里如果先加载的某个插件用了priority1, blockTrue的on_message后面所有低优先级插件都收不到这条消息。你的落库插件设置了priority10, blockFalse是排在后面的一旦前面的插件把消息拦了你的 handler 根本没被调用。没报错是因为事件根本没到你这里黑匣子一样。解决看两个地方。第一检查 Nonebot2 启动日志里插件加载顺序第二在record_msg.handle()第一行加一个print确认是否每次消息都触发。如果发现被拦截要么把你的插件priority调到比它更前要么跟那个插件的作者约定blockFalse。我自己长期维护的机器人里所有纯记录类插件统一用priority90, blockFalse专门避开其他插件的正常范围。5.5 Apscheduler 到点不跑时区与 misfire 排查现象日志里没有任何报错scheduler 也显示 job 已添加但等到 23:50 群里就是没有总结。手动用命令触发又正常。原因最常见的是时区不对。Nonebot2 部署在 UTC 时区的服务器上scheduler.add_job不传timezone时按服务器本地时间算hour23就成了北京时间上午 7 点。第二个常见原因是misfire_grace_time没设置机器人当时恰好断线重连错过触发点后任务直接跳过。解决所有add_job必须显式传timezoneAsia/Shanghai不要依赖服务器系统时区。misfire_grace_time给 3600 秒错过任务后一小时内补跑。排查时先看 Nonebot2 启动日志里打印的 job 触发时间确认是不是差 8 小时再在任务函数第一行加日志确认函数有没有被调用。把这两步做完定时问题基本水落石出。6. 上线前先做离线回放让总结算法先背着历史数据跑三天6.1 离线回放是给算法的后悔药新插件上线最怕的不是代码报错而是算法输出的总结被群友截图嘲讽。所以在机器人正式推送第一篇每日总结之前我会先做离线回放把数据库里的历史数据当成“昨天的消息”手动调用同一套load_daily_messages和build_summary_text连跑三天看输出。这个动作不依赖定时任务、不打扰群友纯粹给自己吃后悔药。回放时把day_offset调成 2、3、4就是三天前、四天前、五天前的历史。你会发现同一个算法在不同日期表现差距很大有的日期话题清晰、关键词精准有的日期全部被玩梗消息淹没。针对后者调整停用词表和时间窗再回放直到连续三天的输出都能看。6.2 关键词冷启动与持续打磨刚部署时数据库里没有历史数据当天文本量少于 30 条时算法会退化成“样本不足”这是有意的保护机制。新功能的机器人建议先跑三天数据再开总结推送。持续打磨的抓手是每日的关键词列表把“今日”“总结”“机器人”这类词补进停用词把群特有的黑话比如项目代号、人名外号加进 jieba 词典TF-IDF 权重会越来越准。我自己的习惯是每周扫一眼关键词输出顺手补几个停用词比反复调模型参数划算得多。6.3 从日总结扩展到周报日总结跑顺后周报几乎零成本定时任务把triggercron的hour/minute换成day_of_weekmonload_daily_messages的时间窗从 1 天改成 7 天关键词和话题分复用同一套函数。周报的样本量更大pick_highlights的max_items建议调到 5消息条数上限放宽到 80 字。我最后提醒一句每日总结是给人看的不是给指标看的。群友真正认可的总结永远是“它记住了我们昨天聊的那个梗”而不是“它统计出了 328 条消息”。这套方案不完美但它让我第一次觉得Nonebot2 插件里的机器学习算法是真的在服务活人。希望帮到你。本文还有配套的精品资源点击获取
