Nonebot+轻量机器学习构建可追溯QQ群日报

简介：这是一份面向Python开发者与AI初学者的实战型QQ群机器人项目资源，聚焦人工智能在社交场景中的落地应用，解决群聊信息过载、关键内容难提炼的痛点。资源基于Nonebot框架构建，集成机器学习文本分析能力，可自动解析每日聊天记录并生成结构化总结，适用于技术交流群、学习社群等需高效信息沉淀的场景。压缩包共28个文件，含18个核心Python脚本（如bot.py、MyTextRankDemo.py、各类Utils工具模块）、3个说明类txt文件、3张功能示意图png、1个配置json、1份PDF版TextRank算法原理文档、1个LICENSE及1个README.md，整体2.05MB，结构清晰，模块职责分明，便于理解机器人架构与ML流程。目前已有274人学习下载，读者可直接复用完整代码工程、掌握聊天文本预处理—关键词提取—主题建模—摘要生成的全链路实现，并参考配套算法文档与配置范例快速部署调试。

1. 为什么你的QQ群每日复盘总像“AI写作文”？——用Nonebot+轻量机器学习做真正可读、可追溯、可干预的群聊日报

你有没有试过让机器人每天在群里发一条“今日群聊精华摘要”，结果发现：要么是关键词堆砌（“今天大家聊了Python、面试、租房、游戏”），要么是模板化套话（“今日讨论热烈，氛围融洽，感谢各位参与”），更糟的是——它根本没读懂谁在和谁争论、哪条消息触发了集体刷屏、哪个新人问的问题被反复解答了三次却没人贴代码。这不是NLP没用，而是把“摘要生成”当成了终点，而忽略了群聊数据的时序性、角色性、意图模糊性和上下文碎片化这四大黑匣子。本方案不追求端到端大模型，而是用Nonebot作为稳定消息管道，把原始聊天记录按天切片、结构化清洗、特征工程后，喂给一个可解释、可调试、可回滚的轻量级机器学习流水线——最终输出的不是“总结”，而是带证据链的日报：比如“张三在14:23提出‘pip install失败’，随后李四、王五分别给出conda/pip镜像/重装Python三种解法，该问题共被提及7次，解决率100%”。它不替代人工运营，但能让你一眼抓住群健康度拐点。适合技术群主、开源项目维护者、课程助教——只要你想从“看消息”升级到“懂对话流”。

---

2. 搭建Nonebot消息捕获基座：从QQ群到结构化日志的最小闭环

2.1 为什么选Nonebot而非酷Q或Mirai原生SDK？

Nonebot的核心优势不在“能连QQ”，而在事件驱动架构与插件生态的解耦设计。酷Q已停更，Mirai SDK虽强大但需自行管理连接保活、消息去重、反撤回逻辑；而Nonebot v2（推荐）通过Adapter抽象层屏蔽协议差异，其Event对象天然携带user_id、group_id、time、message_id、raw_message等字段，且支持on_message、on_notice、on_request三级事件监听——这意味着你能精准捕获“撤回消息”“入群通知”“红包提醒”等非文本信号，这些恰恰是判断群活跃拐点的关键线索。更重要的是，它的MessageEvent自带get_plaintext()方法，能自动剥离CQ码（如[CQ:image,file=xxx.jpg]），避免后续NLP处理被乱码干扰。我一般会禁用echo插件，改用自定义中间件做消息预处理，这样既保留原始时间戳精度（毫秒级），又能在入库前完成敏感词过滤、链接标准化（如把https://t.cn/abc还原为https://github.com/xxx）。

2.2 配置Nonebot接收并落盘每日原始消息

先确保环境：Python 3.9+，nonebot2>=2.3.0，nonebot-adapter-onebot-v11>=2.3.0（适配OneBot v11协议）。关键配置在.env中：

# .env ENVIRONMENT=prod LOG_LEVEL=WARNING SUPERUSERS=["123456789"] # 你的QQ号，用于管理指令 ONEBOT_ACCESS_TOKEN="your_token" # 反向WebSocket需token校验

核心插件代码（src/plugins/daily_log.py）：

from nonebot import on_message, require from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent from nonebot.matcher import Matcher from nonebot.message import event_preprocessor import json import os from datetime import datetime from pathlib import Path # 每日日志目录按年月日创建 LOG_ROOT = Path("data/daily_logs") LOG_ROOT.mkdir(exist_ok=True) @event_preprocessor async def log_message(event: MessageEvent): """全局消息预处理器：所有消息在此统一落盘""" if not isinstance(event, GroupMessageEvent): return # 构建日志路径：data/daily_logs/2024/06/15/100001.json date_str = datetime.fromtimestamp(event.time).strftime("%Y/%m/%d") log_dir = LOG_ROOT / date_str log_dir.mkdir(parents=True, exist_ok=True) # 结构化日志字段（关键！后续ML特征全靠这里） log_entry = { "timestamp": event.time, # Unix时间戳，非字符串 "group_id": event.group_id, "user_id": event.user_id, "nickname": event.sender.card or event.sender.nickname, "raw_message": str(event.get_message()), # 保留CQ码原始形态 "plain_text": event.get_plaintext().strip(), "message_id": event.message_id, "is_image": bool(event.get_message().has("image")), "is_at_all": "[CQ:at,qq=all]" in str(event.get_message()), "reply_to": event.reply.message_id if event.reply else None } # 写入文件（注意：用追加模式，避免并发冲突） log_file = log_dir / f"{event.group_id}.jsonl" with open(log_file, "a", encoding="utf-8") as f: f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") # 注册一个指令用于手动触发当日汇总（调试用） summary_cmd = on_command("daily_summary", priority=10) @summary_cmd.handle() async def _(matcher: Matcher, event: GroupMessageEvent): today = datetime.now().strftime("%Y/%m/%d") log_file = LOG_ROOT / today / f"{event.group_id}.jsonl" if not log_file.exists(): await matcher.finish("今日暂无聊天记录") # 统计基础指标（快速验证管道通路） lines = sum(1 for _ in open(log_file, "r", encoding="utf-8")) await matcher.finish(f"今日群[{event.group_id}]共{lines}条消息")

提示：jsonl格式（每行一个JSON）比单个大JSON更易流式处理，且pandas.read_json(..., lines=True)可直接加载。不要用SQLite存原始消息——高频写入易锁表，且JSONL天然支持按日期分片归档。

2.3 验证消息捕获是否可靠：三个必查点

1. 时间戳一致性：对比机器人收到消息的时间戳（event.time）与你手机QQ客户端显示的发送时间，误差应<3秒。若偏差大，检查服务器时区是否为Asia/Shanghai（timedatectl status确认）。
2. 撤回消息捕获：让测试号发消息后立即撤回，观察data/daily_logs/.../xxx.jsonl中是否出现"notice_type":"group_recall"事件（需额外监听on_notice事件，此处略）。
3. 长消息截断：发送超2000字符的消息，检查plain_text字段是否完整。OneBot v11默认不限制，但某些反向WS代理会截断，此时需在config.yml中调大max_content_length。

---

3. 构建可解释的机器学习流水线：从原始日志到带证据链的日报

3.1 为什么不用BERT微调做摘要？——轻量级方案的选型逻辑

看到“机器学习”就上Transformer是最大误区。群聊日报有三大硬约束：

• 低延迟需求：日报需在凌晨2点前生成，留给ML的窗口<30分钟；
• 可审计性要求：运营者必须能查到“某条结论来自哪几条原始消息”；
• 冷启动友好：新群第一天就需产出有效摘要，不能等积累1000条样本再训练。

因此我们放弃端到端生成，采用三阶段特征工程+规则增强的集成模型：

1. 消息级特征提取（无监督）：用Sentence-BERT计算每条消息与“常见问题模板”的语义距离（如“怎么安装”“报错xxx”“求资源”）；
2. 会话级聚类（半监督）：基于时间窗口+回复链构建会话图，用社区发现算法（Louvain）识别讨论主题簇；
3. 日报生成（规则引擎）：对每个主题簇，提取发言频次TOP3用户、首次提问时间、最终解决方案消息ID，并拼接成自然语言句子。

这套方案在100人规模群聊中，单日处理耗时<8分钟（i5-10210U），模型体积<50MB，且所有中间结果（特征向量、会话图、簇标签）均可导出查验。

3.2 实现消息语义特征提取：用Sentence-BERT做轻量相似度打分

我们不用训练新模型，直接加载paraphrase-multilingual-MiniLM-L12-v2（多语言、12层、仅230MB），它在中文短文本相似度任务上F1达0.82，且推理速度是BERT-base的3倍：

# features/semantic_features.py from sentence_transformers import SentenceTransformer import numpy as np from sklearn.metrics.pairwise import cosine_similarity # 加载模型（首次运行会自动下载） model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') # 预定义问题模板（运营者可随时增删） QUESTION_TEMPLATES = [ "怎么安装", "如何配置", "报错", "错误提示", "无法启动", "下载地址", "求资源", "教程", "文档", "API怎么用", "兼容性", "版本冲突", "依赖问题", "权限不足", "网络超时" ] # 向量化模板（只做一次） template_embeddings = model.encode(QUESTION_TEMPLATES, show_progress_bar=False) def extract_semantic_features(plain_text: str) -> dict: """返回该消息与各模板的最大相似度及对应模板索引""" if not plain_text.strip(): return {"max_sim": 0.0, "template_idx": -1, "template": ""} # 单条消息编码（避免batch推理的内存开销） msg_embedding = model.encode([plain_text], show_progress_bar=False)[0] sims = cosine_similarity([msg_embedding], template_embeddings)[0] max_idx = int(np.argmax(sims)) return { "max_sim": float(sims[max_idx]), "template_idx": max_idx, "template": QUESTION_TEMPLATES[max_idx], "all_sims": [float(s) for s in sims] # 保留全量供调试 } # 示例调用 # feat = extract_semantic_features("pip install torch总是timeout") # print(feat) # {'max_sim': 0.72, 'template_idx': 14, 'template': '网络超时', ...}

参数说明：cosine_similarity返回值范围[0,1]，>0.6视为强相关；template_idx用于后续统计“网络超时”类问题占比；all_sims数组可做热力图分析群话题分布。

3.3 构建会话图并聚类：用回复关系+时间衰减识别真实讨论簇

群聊中“同一话题”常被多轮刷屏打断，单纯按时间滑窗会割裂上下文。我们利用OneBot提供的reply_to字段（撤回消息也含此字段）构建有向图：

# features/session_clustering.py import networkx as nx from datetime import datetime, timedelta import pandas as pd def build_conversation_graph(log_df: pd.DataFrame) -> nx.DiGraph: """基于回复链和时间邻近性构建会话图""" G = nx.DiGraph() # 步骤1：添加所有消息节点（id为message_id） for _, row in log_df.iterrows(): G.add_node(row["message_id"], timestamp=row["timestamp"], user_id=row["user_id"], plain_text=row["plain_text"][:50] + "...") # 截断存文本 # 步骤2：添加回复边（A回复B，则B->A） for _, row in log_df.iterrows(): if pd.notna(row["reply_to"]): if row["reply_to"] in G.nodes(): G.add_edge(row["reply_to"], row["message_id"]) # 步骤3：添加时间邻近边（同用户10分钟内连续发言视为延续） log_df = log_df.sort_values("timestamp") for i in range(len(log_df)-1): curr = log_df.iloc[i] next_msg = log_df.iloc[i+1] if (curr["user_id"] == next_msg["user_id"] and next_msg["timestamp"] - curr["timestamp"] < 600): # 10分钟 G.add_edge(curr["message_id"], next_msg["message_id"]) return G def cluster_sessions(G: nx.DiGraph, min_cluster_size=3) -> dict: """用Louvain算法聚类，返回{cluster_id: [msg_ids]}""" if len(G.nodes()) < min_cluster_size: return {} # 计算边权重：回复边权重=2.0，时间边权重=1.0 for u, v, d in G.edges(data=True): if "reply_to" in str(d): # 简化判据，实际需存原始属性 G[u][v]["weight"] = 2.0 else: G[u][v]["weight"] = 1.0 # Louvain聚类（需安装python-louvain） import community as community_louvain partition = community_louvain.best_partition(G, weight="weight") # 按簇大小过滤 clusters = {} for msg_id, cluster_id in partition.items(): if cluster_id not in clusters: clusters[cluster_id] = [] clusters[cluster_id].append(msg_id) return {k: v for k, v in clusters.items() if len(v) >= min_cluster_size}

关键参数：min_cluster_size=3避免噪声单点；600秒时间窗口经实测，在技术群中能覆盖92%的连续问答链；weight=2.0确保回复关系主导聚类结果。

---

4. 生成带证据链的日报：规则引擎如何让机器学习结论可追溯

4.1 日报模板设计原则：拒绝“AI腔”，拥抱运营语言

日报不是论文摘要，必须满足三个运营刚需：

• 可行动：看到“XX问题未解决”立刻能定位到最新提问消息；
• 可归因：知道“讨论最活跃”是因为张三发起3个话题+李四回应5次；
• 可验证：每句话都能反查到原始消息ID和时间戳。

因此模板采用“事实陈述+证据锚点”结构：

【问题解决】

• “pip install失败”问题共出现4次，首次提问于14:23（消息ID:123456），最终由王五提供pip install --trusted-host pypi.tuna.tsinghua.edu.cn方案解决（消息ID:123489）。

【新人引导】

• 新成员@小明（2024-06-15 09:12加入）提问“怎么提交PR”，张三在10:05发送GitHub官方指南链接（消息ID:123467），该链接被点赞12次。

所有括号内的消息ID都是真实存在的，点击即可跳转到QQ客户端对应消息。

4.2 实现证据链注入：从聚类结果到可点击消息ID

核心是构建message_id到QQ客户端URL的映射。OneBot v11不提供直接跳转链接，但我们可用/api/get_msg接口反查（需开启enable_msg_log）：

# report/generator.py import requests from nonebot import get_driver from nonebot.adapters.onebot.v11 import Bot driver = get_driver() @driver.on_startup async def init_bot_cache(): """缓存Bot实例供异步调用""" global cached_bot bots = list(driver.bots.values()) cached_bot = bots[0] if bots else None def get_qq_message_url(group_id: int, message_id: int) -> str: """生成可点击的QQ消息跳转链接（需配合PC版QQ）""" # PC版QQ支持：tencent://privatemsg/?g=GROUP_ID&u=USER_ID&msg=MESSAGE_ID # 但OneBot不暴露USER_ID，故退而求其次：用群号+消息ID构造搜索关键词 return f"https://web.qq.com/search?keyword={message_id}&type=group&gid={group_id}" def generate_daily_report(cluster_dict: dict, log_df: pd.DataFrame) -> str: """主日报生成函数""" report_lines = ["【群聊日报】" + datetime.now().strftime("%Y-%m-%d")] # 按簇分析每个主题 for cluster_id, msg_ids in cluster_dict.items(): cluster_msgs = log_df[log_df["message_id"].isin(msg_ids)].copy() # 提取关键信息 first_msg = cluster_msgs.loc[cluster_msgs["timestamp"].idxmin()] last_msg = cluster_msgs.loc[cluster_msgs["timestamp"].idxmax()] top_users = cluster_msgs["user_id"].value_counts().head(3) # 构建证据链句子 first_url = get_qq_message_url(first_msg["group_id"], first_msg["message_id"]) last_url = get_qq_message_url(last_msg["group_id"], last_msg["message_id"]) topic_summary = ( f"- “{first_msg['plain_text'][:20]}...”话题共{len(msg_ids)}条消息，" f"首问于{datetime.fromtimestamp(first_msg['timestamp']).strftime('%H:%M')} " f"[跳转]({first_url})，终解于{datetime.fromtimestamp(last_msg['timestamp']).strftime('%H:%M')} " f"[跳转]({last_url})。" ) report_lines.append(topic_summary) return "\n".join(report_lines)

注意：tencent://协议仅PC QQ支持，移动端需用https://web.qq.com搜索。此处用Markdown链接是为适配Nonebot的send_group_msg（支持部分HTML渲染）。

4.3 自动化日报推送：定时任务与失败熔断

用APScheduler实现凌晨1:30触发，但必须加熔断——避免某天日志损坏导致机器人卡死：

# scheduler.py from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger import asyncio from src.plugins.daily_log import LOG_ROOT scheduler = AsyncIOScheduler() @scheduler.scheduled_job(CronTrigger(hour=1, minute=30)) async def daily_report_job(): today = datetime.now().strftime("%Y/%m/%d") log_dir = LOG_ROOT / today # 熔断1：检查日志文件是否存在 if not log_dir.exists(): return # 熔断2：检查当日消息量（少于5条不生成） total_msgs = 0 for f in log_dir.glob("*.jsonl"): total_msgs += sum(1 for _ in open(f, "r", encoding="utf-8")) if total_msgs < 5: return try: # 执行ML流水线（此处省略调用细节） report = generate_report_for_date(today) # 推送至指定群（可配置） await bot.send_group_msg(group_id=10001, message=report) except Exception as e: # 熔断3：记录错误并推送告警 error_msg = f"日报生成失败：{type(e).__name__} - {str(e)[:100]}" await bot.send_group_msg(group_id=10001, message=error_msg) # 同时写入error.log供排查 with open("logs/error.log", "a") as f: f.write(f"{datetime.now()} {error_msg}\n")

---

5. 避坑指南：Nonebot+机器学习落地中的5个血泪经验

5.1 现象：日报中频繁出现“用户A说XXX，用户B说YYY”，但实际是不同话题的拼接

原因：会话图构建时未过滤广告/表情包消息，导致无关消息被时间邻近边强行连接。
解决：在build_conversation_graph中增加过滤逻辑——对plain_text为空、或含[CQ:image]且len(plain_text)<5的消息，不参与时间边构建。实测可降低误聚类率67%。

5.2 现象：语义相似度打分始终为0.0

原因：Sentence-BERT模型加载时未指定device='cpu'，在无GPU环境自动fallback失败，但异常被静默吞掉。
解决：显式声明设备，并加日志：

model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2', device='cpu') print(f"Model loaded on {model._target_device}") # 必须看到'cpu'输出

5.3 现象：机器人突然停止接收消息，日志显示WebSocket connection closed

原因：OneBot反向WS服务（如go-cqhttp）的heartbeat-interval默认30秒，而Nonebot心跳检测超时设为25秒，网络抖动时频繁断连。
解决：在go-cqhttp配置中将heartbeat-interval改为15秒，并在Nonebot配置中加：

# config.yml adapters: - nonebot.adapters.onebot.v11.Adapter: heartbeat_interval: 10 # 必须<服务端interval

5.4 现象：日报中“消息ID跳转链接”点击后找不到消息

原因：QQ客户端消息ID是64位整数，但OneBot v11返回的message_id是字符串类型（如"1234567890123456789"），直接拼接URL时被截断。
解决：在get_qq_message_url中强制转为整数再转回字符串：

def get_qq_message_url(group_id: int, message_id: str) -> str: try: msg_id_int = int(message_id) # 强制转整数去零 return f"https://web.qq.com/search?keyword={msg_id_int}&type=group&gid={group_id}" except: return f"https://web.qq.com/search?keyword={message_id}&type=group&gid={group_id}"

5.5 现象：聚类结果每天变化剧烈，运营者无法建立认知惯性

原因：Louvain算法随机种子未固定，导致相同输入产生不同分区。
解决：在cluster_sessions中设置随机种子：

import random random.seed(42) # 固定种子 import numpy as np np.random.seed(42) # 调用community_louvain前加这两行

---

6. 进阶技巧：用日报数据反哺群运营决策的3个真实场景

6.1 识别“沉默专家”：从发言频次到问题解决率的升维分析

日报只统计“谁发言多”，但真正的价值在“谁解决问题多”。我们在特征工程中额外计算每个用户的问题解决率：

• 定义“问题消息”：semantic_features["max_sim"] > 0.6 and semantic_features["template"] in ["报错","怎么安装","求资源"]
• 定义“解决消息”：reply_to指向问题消息ID，且plain_text含“已解决”“试试这个”“成功了”等关键词
• 解决率 = 用户解决消息数 / 该用户总发言数

然后在日报末尾加一栏：

【专家榜】

• 张三：解决率82%（12/15），主导解决“CUDA版本冲突”“Docker镜像拉取失败”
• 李四：解决率65%（7/11），专精“前端部署问题”

提示：这个指标需要至少3天数据才能稳定，首日可标注“数据积累中”。我习惯在群公告置顶本周专家榜，新人入群第一眼就知该@谁。

6.2 构建群健康度仪表盘：用日报字段生成可量化指标

把日报中的结构化字段导出为CSV，用Grafana搭简易看板：

指标	计算方式	健康阈值
问题解决率	sum(解决消息)/sum(问题消息)	>70%
新人留存率	(入群后3天内发言新人数)/(当日入群新人数)	>40%
平均响应时长	mean(解决消息.timestamp - 问题消息.timestamp)	<30分钟
话题多样性	len(聚类簇数量)/当日总消息数	0.05~0.15（太低=刷屏，太高=碎片化）

关键动作：当“问题解决率”连续3天<50%，自动触发机器人私聊群主：“检测到问题解决率下降，建议检查近期FAQ文档更新情况”。

6.3 实现“日报可编辑”：让运营者修正机器学习的误判

再好的模型也会错。我们在日报末尾加一行：
【人工修正】发送“/fix 123456 问题类型”可修正消息123456的分类（例：/fix 123456 求资源）

后端监听该指令，将修正记录写入corrections.csv：

message_id,corrected_template,operator,timestamp 123456,求资源,123456789,1718432100

下次生成日报时，优先读取corrections.csv覆盖原始语义特征。这个设计让机器学习从“黑匣子”变成“可协作工作台”——运营者每次修正都在训练模型，且无需碰代码。

最后说句实在的：这个方案上线三个月后，我负责的两个技术群平均问题解决时长从47分钟降到21分钟，新人3日留存率从31%升到58%。但它真正的价值不是数字，而是当我深夜看到日报里写着“张三在02:15解决了小明的CUDA问题”，我知道这个群正在自己生长。希望帮到你。

本文还有配套的精品资源，点击获取