简介：这是一份面向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(rule=record_rule, priority=10, block=False) @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)提示：
block=False是关键——它允许消息继续被其他插件（如命令响应、关键词回复）处理，而非被本插件独占。priority=10保证其在多数业务逻辑前执行，但又低于权限校验类插件（通常 priority=1–5）。
2.2 构建轻量级 SQLite 消息仓库：支持按日分区与快速检索
消息需按日期归档，否则单表膨胀将导致SELECT * FROM messages WHERE date='2024-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_ok=True) 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 f"messages_{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(f"SELECT name FROM sqlite_master WHERE type='table' 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(f"CREATE INDEX IF NOT EXISTS idx_{table_name}_group_time ON {table_name}(group_id, timestamp)") await db.execute(f"INSERT 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 类大模型，选择jieba+TfidfVectorizer+MiniBatchKMeans的组合方案——它在中文短文本聚类上表现稳定，且全部组件均支持增量训练与热更新。
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 替代 LDA
LDA 主题模型在小样本（单日 200–500 条消息）下易过拟合，且训练耗时高。MiniBatchKMeans在 CPU 上单次聚类 500 条消息仅需 1.2 秒，且支持在线更新。我们设定n_clusters=3，对应最常见的三类语义主题：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_features=5000, # 控制向量维度，平衡精度与内存 ngram_range=(1, 2), # 加入二元词组，捕捉 "服务器宕机" 这类短语 min_df=2, # 忽略只出现 1 次的词（降低噪声） stop_words=None # 停用词已在预处理阶段移除 ) self.kmeans = MiniBatchKMeans( n_clusters=3, random_state=42, batch_size=100, # 小批量更新，节省内存 max_iter=100 ) 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_features=5000将 TF-IDF 向量压缩至 5000 维，避免稀疏矩阵爆炸；ngram_range=(1,2)让模型识别“数据库连接”而非孤立的“数据库”“连接”；min_df=2过滤掉仅出现一次的拼写错误或随机字符。
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(hour="2", minute="0"), id="daily_summary") async def daily_summary_job(): today = datetime.now().date() yesterday = today - timedelta(days=1) date_str = yesterday.strftime("%Y-%m-%d") logger.info(f"Starting daily summary for {date_str}") try: # 1. 从 DB 获取昨日所有消息 messages = await fetch_daily_messages(date_str) if not messages: logger.warning(f"No 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(f"Too 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(f"New 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_id=group_id, message=Message(f"📅 {date_str} 群消息摘要\n\n{summary}") ) logger.info(f"Daily summary sent for {date_str}") except Exception as e: logger.error(f"Failed to generate daily summary for {date_str}: {e}", exc_info=True)注意：
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 秒仍执行，超时则丢弃 } )提示：
coalesce=True和max_instances=1是防止“昨日总结因网络故障未发送，今日又触发，导致两天总结合并发送”的核心配置。
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", priority=5) @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_fit（MiniBatchKMeans 支持），既保留历史知识，又吸收最新语义。fetch_feedback_samples()需实现去重、清洗、映射到0/1/2标签的逻辑。
最终，这个系统不是一次性交付的工具，而是一个会随团队语言习惯演进的“活”组件——它不承诺 100% 准确，但保证每次偏差都被看见、被记录、被修正。
本文还有配套的精品资源，点击获取
