简介这是一份面向Python开发者与AI初学者的实战型QQ群机器人项目资源聚焦机器学习在社交文本分析中的落地应用。项目基于Nonebot框架构建核心能力是自动解析每日群聊记录运用TextRank等算法提取关键词、识别热点话题并生成结构化总结有效解决群信息过载与关键内容难追溯的问题。资源包共28个文件含18个Python源码涵盖bot主程序、ML模型模块、工具类Utils及TextRank实现、3个说明类txt与pdf文档、3张效果示意图、1个配置文件config.py和1个README.md整体2.05MB结构清晰、模块解耦便于理解机器人架构与文本处理全流程。已有274人学习下载读者可直接复用完整代码工程、参考TextRank算法实现细节、借鉴聊天记录清洗与特征提取实践并结合JSON/IO工具类快速适配自有群场景。1. 这不是“自动发消息”的QQ群机器人而是能读完一整天聊天后自己写日报的AI协作者你见过每天凌晨自动在群公告里贴出「今日关键词项目进度、外包报价、咖啡续命高频提问服务器怎么重启Git冲突怎么解异常波动张工 的发言量比上周320%」的机器人吗这不是规则引擎拼凑的关键词统计而是用真实聊天文本训练轻量级文本分类与摘要模型后输出具备语义连贯性和业务指向性的自然语言总结。它不依赖预设关键词库能识别“改需求”和“加个按钮”背后的真实意图差异也不靠人工写正则而是让模型从历史对话中自主发现“技术讨论”“闲聊摸鱼”“紧急求助”三类语义簇。适合中小团队技术群、高校实验室群、开源项目协作群——尤其当你发现群消息已多到没人再翻记录但关键决策、阻塞点、待办项却散落在几百条消息里时这个基于 Nonebot 框架、嵌入机器学习能力的每日总结系统就是那个沉默但可靠的“群消息审计员”。2. 用 Nonebot 3 搭建可扩展的群消息捕获管道从监听到结构化存储Nonebot 3 是当前主流 QQ 群机器人开发框架其事件驱动架构天然适配“消息采集→清洗→入库→触发分析”的流水线。与旧版相比v3 的Event类型体系更清晰MessageEvent子类如GroupMessageEvent明确分离了群 ID、发送者、消息内容、时间戳等字段避免手动解析 JSON 的脆弱性。我们不采用on_message()全局监听这种粗粒度方式而是通过自定义Matcher绑定特定群组并启用rule过滤非文本消息如图片、语音、表情包确保后续机器学习模块只处理可解析的纯文本。2.1 初始化 Nonebot 项目并配置群消息持久化入口首先创建最小可行项目结构mkdir daily-qq-summary cd daily-qq-summary python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install nonebot2[all] nonebot-adapter-onebot sqlalchemy aiosqlite jieba在bot.py中初始化适配器并注册消息处理器# bot.py from nonebot import init, load_plugins, run from nonebot.adapters.onebot.v11 import Adapter as OneBotAdapter init() # 启用 OneBot v11 协议适配器适配 go-cqhttp 或 Lagrange driver get_driver() driver.register_adapter(OneBotAdapter) # 加载插件目录 load_plugins(src/plugins) if __name__ __main__: run()关键在于消息捕获逻辑需解耦为独立插件。在src/plugins/record/下新建__init__.py# src/plugins/record/__init__.py from nonebot import on_message, require from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message from nonebot.rule import Rule from nonebot.matcher import Matcher from nonebot.internal.adapter import Event import asyncio import time # 仅处理指定群组ID 列表可从配置文件读取 TARGET_GROUPS [123456789, 987654321] async def is_target_group(event: Event) - bool: if not isinstance(event, GroupMessageEvent): return False return event.group_id in TARGET_GROUPS # 定义规则仅文本、非系统消息、非自身发送 record_rule Rule(is_target_group) # 使用 matcher 而非 on_message便于后续添加中间件 record_matcher on_message(rulerecord_rule, priority10, blockFalse) record_matcher.handle() async def handle_group_message(event: GroupMessageEvent): # 提取结构化字段 record { group_id: event.group_id, user_id: event.user_id, message_id: event.message_id, raw_text: event.get_plaintext().strip(), timestamp: event.time, is_at_bot: any(seg.type at and str(seg.data.get(qq)) self for seg in event.message) } # 异步写入 SQLite避免阻塞事件循环 await save_to_db(record)提示blockFalse是关键——它允许消息继续被其他插件如命令响应、关键词回复处理而非被本插件独占。priority10保证其在多数业务逻辑前执行但又低于权限校验类插件通常 priority1–5。2.2 构建轻量级 SQLite 消息仓库支持按日分区与快速检索消息需按日期归档否则单表膨胀将导致SELECT * FROM messages WHERE date2024-06-15查询变慢。我们采用“表名动态生成”策略每日新建一张表如messages_20240615并通过视图统一查询接口# src/plugins/record/db.py import aiosqlite from datetime import datetime import os DB_PATH data/messages.db async def init_db(): os.makedirs(data, exist_okTrue) async with aiosqlite.connect(DB_PATH) as db: # 创建元数据表记录各日期表是否存在 await db.execute( CREATE TABLE IF NOT EXISTS date_index ( date TEXT PRIMARY KEY, table_name TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) await db.commit() async def get_daily_table_name(date_str: str) - str: return fmessages_{date_str.replace(-, )} async def ensure_daily_table(date_str: str): table_name await get_daily_table_name(date_str) async with aiosqlite.connect(DB_PATH) as db: # 检查表是否存在 async with db.execute(fSELECT name FROM sqlite_master WHERE typetable AND name{table_name}) as cursor: if not await cursor.fetchone(): # 创建新表 await db.execute(f CREATE TABLE {table_name} ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL, user_id INTEGER NOT NULL, message_id TEXT UNIQUE, raw_text TEXT NOT NULL, timestamp INTEGER NOT NULL, is_at_bot INTEGER DEFAULT 0 ) ) # 添加索引加速按群、按时间查询 await db.execute(fCREATE INDEX IF NOT EXISTS idx_{table_name}_group_time ON {table_name}(group_id, timestamp)) await db.execute(fINSERT OR IGNORE INTO date_index (date, table_name) VALUES (?, ?), (date_str, table_name)) await db.commit() async def save_to_db(record: dict): date_str datetime.fromtimestamp(record[timestamp]).strftime(%Y-%m-%d) await ensure_daily_table(date_str) table_name await get_daily_table_name(date_str) async with aiosqlite.connect(DB_PATH) as db: await db.execute(f INSERT OR IGNORE INTO {table_name} (group_id, user_id, message_id, raw_text, timestamp, is_at_bot) VALUES (?, ?, ?, ?, ?, ?) , ( record[group_id], record[user_id], record[message_id], record[raw_text], record[timestamp], 1 if record[is_at_bot] else 0 )) await db.commit()注意SQLite 的INSERT OR IGNORE避免重复消息写入同一消息可能因网络重传被多次接收。is_at_bot字段虽不参与机器学习但为后续“机器人触发总结”功能预留钩子。3. 用轻量级机器学习 pipeline 实现每日聊天语义解析从分词到主题聚类每日总结的核心不是统计词频而是理解对话的语义结构哪些是技术问题讨论哪些是项目进度同步哪些是临时协调这需要一套端到端可部署的 NLP 流水线且必须满足三个硬约束① 单次推理耗时 5 秒避免阻塞定时任务② 模型体积 50MB适配低配 VPS③ 不依赖 GPU纯 CPU 推理。我们放弃 BERT 类大模型选择jiebaTfidfVectorizerMiniBatchKMeans的组合方案——它在中文短文本聚类上表现稳定且全部组件均支持增量训练与热更新。3.1 构建中文消息预处理链去噪、标准化、停用词过滤QQ 群消息充满噪声[图片]、[链接]、[红包]、xxx、/face123表情码、大量语气词“啊”“哦”“嗯嗯”。预处理目标是保留语义主干同时不丢失关键实体如人名、项目代号、错误码。我们采用分层清洗策略# src/ml/preprocess.py import jieba import re from typing import List, Set # 加载自定义词典提升“git merge”“Redis缓存”等技术词切分准确率 jieba.load_userdict(src/ml/custom_dict.txt) # 常见停用词精简版避免过度删除 STOPWORDS { 的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这, 那, 它, 他, 她 } def clean_message(text: str) - str: # 1. 移除特殊标记 text re.sub(r\[.*?\], , text) # 移除 [图片][链接] 等 text re.sub(r[\u4e00-\u9fa5a-zA-Z0-9_], , text) # 移除 提及 text re.sub(r/face\d, , text) # 移除 QQ 表情码 # 2. 标准化空白符 text re.sub(r\s, , text).strip() # 3. 保留中文、英文字母、数字、常见标点用于保留代码片段如 error 500 text re.sub(r[^\u4e00-\u9fa5a-zA-Z0-9\u3000-\u303f\uff00-\uffef。“”‘’【】《》、\s], , text) return text def segment_and_filter(text: str) - List[str]: if not text: return [] words jieba.lcut(text) # 过滤停用词、单字词除非是技术缩写如 API、过短词 filtered [ w for w in words if w not in STOPWORDS and len(w) 1 and not (len(w) 2 and w.isalpha()) # 保留 Git Redis过滤 是 有 ] return filtered参数说明custom_dict.txt应包含团队常用术语每行一个词例如git merge Redis缓存 Jenkins构建 404错误3.2 训练轻量级主题聚类模型用 MiniBatchKMeans 替代 LDALDA 主题模型在小样本单日 200–500 条消息下易过拟合且训练耗时高。MiniBatchKMeans在 CPU 上单次聚类 500 条消息仅需 1.2 秒且支持在线更新。我们设定n_clusters3对应最常见的三类语义主题technical_discussion技术讨论、project_update项目同步、coordination协调事务。聚类前需将文本向量化# src/ml/cluster.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import MiniBatchKMeans from sklearn.metrics import silhouette_score import joblib import numpy as np from typing import List, Tuple, Dict class DailyClusterModel: def __init__(self, model_path: str models/cluster_model.joblib): self.model_path model_path self.vectorizer TfidfVectorizer( max_features5000, # 控制向量维度平衡精度与内存 ngram_range(1, 2), # 加入二元词组捕捉 服务器宕机 这类短语 min_df2, # 忽略只出现 1 次的词降低噪声 stop_wordsNone # 停用词已在预处理阶段移除 ) self.kmeans MiniBatchKMeans( n_clusters3, random_state42, batch_size100, # 小批量更新节省内存 max_iter100 ) self.labels_map {0: technical_discussion, 1: project_update, 2: coordination} def fit_transform(self, texts: List[str]) - Tuple[np.ndarray, np.ndarray]: # 向量化 X self.vectorizer.fit_transform(texts) # 聚类 labels self.kmeans.fit_predict(X) # 计算轮廓系数评估聚类质量0.3 可接受 score silhouette_score(X, labels) if len(set(labels)) 1 else 0.0 return labels, score def predict(self, texts: List[str]) - np.ndarray: X self.vectorizer.transform(texts) return self.kmeans.predict(X) def save(self): joblib.dump({ vectorizer: self.vectorizer, kmeans: self.kmeans, labels_map: self.labels_map }, self.model_path) def load(self): data joblib.load(self.model_path) self.vectorizer data[vectorizer] self.kmeans data[kmeans] self.labels_map data[labels_map]关键参数解释max_features5000将 TF-IDF 向量压缩至 5000 维避免稀疏矩阵爆炸ngram_range(1,2)让模型识别“数据库连接”而非孤立的“数据库”“连接”min_df2过滤掉仅出现一次的拼写错误或随机字符。3.3 生成可读性日报从聚类标签到自然语言摘要聚类结果只是数字标签0/1/2需映射为人类可读的摘要段落。我们不使用模板填空易僵硬而是基于每个簇内高频词 代表性句子生成摘要# src/ml/generate_summary.py from collections import Counter, defaultdict from typing import List, Dict, Tuple import jieba def generate_daily_summary(clustered_messages: Dict[int, List[str]], cluster_labels: Dict[int, str]) - str: summary_parts [] for cluster_id, messages in clustered_messages.items(): if not messages: continue # 提取该簇高频词前 5 all_words [] for msg in messages: words jieba.lcut(msg) all_words.extend([w for w in words if len(w) 1]) top_words [w for w, _ in Counter(all_words).most_common(5)] # 找出最“典型”的一条消息长度适中、含高频词最多 representative_msg max_hit 0 for msg in messages: hits sum(1 for w in top_words if w in msg or msg in w) if hits max_hit and 10 len(msg) 80: max_hit hits representative_msg msg # 构建摘要句 label_zh cluster_labels.get(cluster_id, 未知主题) if top_words: keywords 、.join(top_words[:3]) summary_parts.append(f【{label_zh}】涉及关键词{keywords}。典型讨论「{representative_msg}」) else: summary_parts.append(f【{label_zh}】共 {len(messages)} 条消息未提取显著关键词。) return \n.join(summary_parts) # 示例调用 # clustered {0: [如何解决 MySQL 连接超时, Redis 缓存穿透怎么防], # 1: [前端联调完成, 后端 API 文档已更新], # 2: [会议室预约明天下午, 测试环境账号发我]} # summary generate_daily_summary(clustered, {0:技术讨论, 1:项目同步, 2:协调事务})逻辑说明representative_msg的筛选逻辑兼顾信息量含高频词与可读性长度 10–80 字避免截断或过长。top_words限制取前 3 个展示防止摘要冗长。4. 实现定时触发与总结推送用 APScheduler 调度模型推理与消息发送每日总结必须准时、可靠、可监控。Nonebot 内置的nonebot.scheduler功能有限我们选用工业级调度库APScheduler并将其深度集成进 Nonebot 生命周期——启动时加载关闭时优雅退出。4.1 注册定时任务每日凌晨 2:00 执行分析与推送在src/plugins/summary/__init__.py中定义任务# src/plugins/summary/__init__.py from nonebot import get_driver, logger from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from src.ml.cluster import DailyClusterModel from src.ml.generate_summary import generate_daily_summary from src.plugins.record.db import fetch_daily_messages from nonebot.adapters.onebot.v11 import Bot, Message, MessageSegment import asyncio from datetime import datetime, timedelta driver get_driver() scheduler AsyncIOScheduler() driver.on_startup async def start_scheduler(): scheduler.start() logger.info(Daily summary scheduler started.) driver.on_shutdown async def shutdown_scheduler(): scheduler.shutdown() logger.info(Daily summary scheduler shutdown.) # 定义每日任务 scheduler.scheduled_job(CronTrigger(hour2, minute0), iddaily_summary) async def daily_summary_job(): today datetime.now().date() yesterday today - timedelta(days1) date_str yesterday.strftime(%Y-%m-%d) logger.info(fStarting daily summary for {date_str}) try: # 1. 从 DB 获取昨日所有消息 messages await fetch_daily_messages(date_str) if not messages: logger.warning(fNo messages found for {date_str}) return # 2. 预处理 cleaned_texts [clean_message(msg[raw_text]) for msg in messages] cleaned_texts [t for t in cleaned_texts if t] # 过滤空字符串 if len(cleaned_texts) 5: logger.warning(fToo few messages ({len(cleaned_texts)}) for clustering on {date_str}) return # 3. 加载/训练模型 model DailyClusterModel() try: model.load() except FileNotFoundError: logger.info(No existing cluster model, training new one...) labels, score model.fit_transform(cleaned_texts) model.save() logger.info(fNew model trained, silhouette score: {score:.3f}) else: labels model.predict(cleaned_texts) # 4. 按标签分组消息 clustered defaultdict(list) for msg, label in zip(messages, labels): clustered[label].append(msg[raw_text]) # 5. 生成摘要 summary generate_daily_summary(clustered, model.labels_map) # 6. 推送至指定群 bot: Bot nonebot.get_bots().get(onebot) # 假设只有一个 Bot 实例 if bot: for group_id in [123456789, 987654321]: # 可配置 await bot.send_group_msg( group_idgroup_id, messageMessage(f {date_str} 群消息摘要\n\n{summary}) ) logger.info(fDaily summary sent for {date_str}) except Exception as e: logger.error(fFailed to generate daily summary for {date_str}: {e}, exc_infoTrue)注意fetch_daily_messages(date_str)需在db.py中补充实现根据date_str动态查询对应日期表返回List[Dict]。4.2 关键参数调优避免调度冲突与资源争抢APScheduler 默认使用ThreadPoolExecutor但在高并发场景下可能与 Nonebot 的异步事件循环冲突。我们显式配置为AsyncIOExecutor并限制并发数# 在 driver.on_startup 中添加 from apscheduler.executors.asyncio import AsyncIOExecutor scheduler AsyncIOScheduler( executors{ default: AsyncIOExecutor() }, job_defaults{ coalesce: True, # 同一任务若错过执行时间只执行一次避免堆积 max_instances: 1, # 同一任务最多 1 个实例运行防止重复推送 misfire_grace_time: 300 # 任务延迟 ≤300 秒仍执行超时则丢弃 } )提示coalesceTrue和max_instances1是防止“昨日总结因网络故障未发送今日又触发导致两天总结合并发送”的核心配置。5. 模型持续优化与效果验证用人工反馈闭环改进聚类质量机器学习模型上线后最大的风险不是初始不准而是无法感知业务语义漂移——比如团队开始频繁讨论“Rust 重构”但模型仍把相关消息归入technical_discussion而非新主题。我们设计三层验证机制自动化指标监控、人工标注反馈、模型热更新。5.1 自动化监控每日记录聚类质量与主题分布每次daily_summary_job执行后将关键指标写入metrics.db供后续分析# src/ml/metrics.py import aiosqlite from datetime import datetime async def log_metrics(date_str: str, silhouette_score: float, cluster_counts: Dict[int, int], total_messages: int): async with aiosqlite.connect(data/metrics.db) as db: await db.execute( CREATE TABLE IF NOT EXISTS daily_metrics ( date TEXT PRIMARY KEY, silhouette_score REAL, cluster_0_count INTEGER, cluster_1_count INTEGER, cluster_2_count INTEGER, total_messages INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) await db.execute( INSERT OR REPLACE INTO daily_metrics (date, silhouette_score, cluster_0_count, cluster_1_count, cluster_2_count, total_messages) VALUES (?, ?, ?, ?, ?, ?) , ( date_str, silhouette_score, cluster_counts.get(0, 0), cluster_counts.get(1, 0), cluster_counts.get(2, 0), total_messages )) await db.commit()为什么监控轮廓系数它衡量聚类内紧密度与簇间分离度值域 [-1,1]0.5 表示良好聚类0.25 表示可能需调整n_clusters或重新清洗数据。5.2 人工反馈通道用/feedback命令收集标注样本当用户觉得某日总结不准可发送/feedback 2024-06-15 technical_discussion将昨日消息中属于technical_discussion的原始文本提交给管理员审核# src/plugins/feedback/__init__.py from nonebot import on_command from nonebot.adapters.onebot.v11 import MessageEvent, Message from nonebot.params import CommandArg import re feedback_cmd on_command(feedback, priority5) feedback_cmd.handle() async def handle_feedback(event: MessageEvent, args: Message CommandArg()): text args.extract_plain_text().strip() # 匹配 /feedback 2024-06-15 project_update match re.match(r(\d{4}-\d{2}-\d{2})\s(technical_discussion|project_update|coordination), text) if not match: await feedback_cmd.finish(格式错误/feedback YYYY-MM-DD topic_name) date_str, topic match.groups() # 将 event.message_id 对应的消息标记为 topic并存入 feedback_queue 表 # 具体实现略核心是建立人工标注样本池 await feedback_cmd.finish(f已收到 {date_str} 的 {topic} 标注请求管理员将在 24 小时内处理。)5.3 模型热更新每周自动融合新标注数据重训每周日凌晨脚本检查feedback_queue表中是否有 ≥10 条有效标注若有则执行增量训练# scripts/retrain_model.sh #!/bin/bash cd /path/to/daily-qq-summary source .venv/bin/activate python -c from src.ml.cluster import DailyClusterModel from src.plugins.feedback.db import fetch_feedback_samples import numpy as np # 获取新标注数据 texts, labels fetch_feedback_samples() if len(texts) 10: model DailyClusterModel() model.load() # 使用新数据微调 KMeans实际中可重训整个 pipeline # 此处简化为用新数据替换部分旧训练集重新 fit print(Retraining model with new feedback...) # ... 具体重训逻辑 model.save() print(Model updated.) 关键技巧不完全抛弃旧模型而是用新标注数据做partial_fitMiniBatchKMeans 支持既保留历史知识又吸收最新语义。fetch_feedback_samples()需实现去重、清洗、映射到0/1/2标签的逻辑。最终这个系统不是一次性交付的工具而是一个会随团队语言习惯演进的“活”组件——它不承诺 100% 准确但保证每次偏差都被看见、被记录、被修正。本文还有配套的精品资源点击获取