简介这是一份面向Python开发者与Bot入门学习者的多用途QQ群机器人实战项目基于NoneBot2框架实现群管理、自动回复、游戏互动等常见功能适用于社交平台智能应用开发与自动化运维场景。资源包共213个文件含141个核心Python源码插件逻辑与事件处理、34份Markdown文档含部署指南、API说明与开发规范、20份License协议文件辅以JPG/PNG示例图、JSON配置及Shell脚本等完整覆盖开发、调试与部署全流程压缩包仅3.04MB轻量易上手。已有633人学习下载资源结构清晰主目录mokabot2-master包含可直接运行的工程骨架、典型功能插件示例如公告、档案、二次元互动等及配套配置文件附带.gitignore与CI相关yml文件体现良好工程实践。读者可快速掌握NoneBot2插件开发范式、异步消息处理机制及QQ Bot Token接入流程并复用模块化代码拓展自定义功能。1. 为什么一个“多用途QQ群机器人”必须用 NoneBot2 而不是自己轮子重写你刚在群里看到一个能自动查天气、转发 RSS、抽签、统计发言热词、甚至对接内部 Jenkins 的 QQ 群机器人点开介绍页发现它只依赖nonebot2和几个 Python 包——没有 Web 框架胶水层、没手写长连接心跳、没硬编码消息解析逻辑。这不是巧合。NoneBot2 是目前唯一把「QQ 协议适配」和「插件化业务逻辑」彻底解耦的 Python 框架它不绑定具体协议支持 go-cqhttp、onebot v12、KOOK不强制 MVC 结构但通过事件驱动 依赖注入 插件生命周期管理让「加一个查汇率功能」变成pip install nonebot-plugin-exchange 两行配置。对运维人员它提供nb run一键启停和nb deploy容器化打包对开发者它用 Pydantic 模型校验每条入参用matcher.pause()实现多轮对话状态机用Depends()注入数据库连接或缓存客户端。如果你正被「每次加新功能都要改 main.py、消息解析总出 UnicodeDecodeError、群聊私聊逻辑混在一起」折磨这个标题不是教你搭玩具而是给出一套可随业务增长横向扩展的群机器人生产级架构。2. 从零初始化一个可热重载、带基础命令的 NoneBot2 项目2.1 初始化项目结构与核心依赖安装NoneBot2 不是单个包而是一套分层生态nonebot2是核心运行时nonebot-adapter-onebot提供 QQ 协议适配nonebot-plugin-apscheduler支持定时任务。我们跳过pip install nonebot2这种易出错的手动安装直接用官方推荐的nb-cli工具链# 全局安装 nb-cli需 Python 3.8 pip install nb-cli # 创建项目自动选择 onebot v11 适配器、生成标准目录结构 nb create my-qq-bot --adapter onebot-v11 # 进入项目并安装依赖会自动处理 nonebot2 适配器 uvicorn 等 cd my-qq-bot pip install -e .提示-e参数启用开发模式后续修改插件代码无需重新pip install。若使用 Conda 环境请先conda activate your-env再执行上述命令避免 pip 与 conda 混装导致依赖冲突。生成的目录结构中关键路径为bot.py主入口定义 Bot 实例和全局配置src/plugins/所有插件存放目录每个子目录是一个独立插件pyproject.toml声明项目元信息、插件入口点、依赖版本约束2.2 配置 go-cqhttp 作为底层消息桥接器NoneBot2 本身不直连 QQ 服务器必须通过go-cqhttp或其他 OneBot 实现接收/发送消息。下载对应系统版本的go-cqhttp二进制文件后执行首次启动# Linux/macOS 下赋予执行权限并启动Windows 直接双击 go-cqhttp.exe chmod x go-cqhttp ./go-cqhttp # 按提示扫码登录 QQ成功后 CtrlC 退出 # 编辑生成的 config.yml重点修改以下三处 # 1. 启用反向 WebSocketNoneBot2 默认监听此端口 # 2. 设置 access_token与 bot.py 中保持一致 # 3. 开放本地监听地址避免 Docker 网络问题config.yml关键片段# 反向 WebSocket 配置NoneBot2 将从此处拉取消息 servers: - ws-reverse: url: ws://127.0.0.1:8080/ws reverse-api-url: http://127.0.0.1:8080/api reverse-event-url: http://127.0.0.1:8080/event access-token: your_secure_token_here # 必须与 bot.py 中 token 一致2.3 在 bot.py 中声明适配器与插件加载逻辑bot.py是整个项目的调度中枢。它不包含业务逻辑只负责注册适配器、加载插件、设置全局中间件# bot.py from nonebot import init, load_plugins, get_driver from nonebot.adapters.onebot.v11 import Adapter as OneBotV11Adapter # 初始化 NoneBot2 核心读取 pyproject.toml 中的 [tool.nonebot] 配置 init() # 获取驱动器实例用于注册适配器 driver get_driver() # 注册 OneBot v11 适配器必须在 load_plugins 之前 driver.register_adapter(OneBotV11Adapter) # 加载 src/plugins/ 下所有插件支持子目录递归 load_plugins(src/plugins) # 可选添加全局异常处理器捕获未处理的插件异常 driver.on_exception async def handle_exception(event, exception): from nonebot.log import logger logger.error(f全局异常: {type(exception).__name__}: {exception})参数说明load_plugins(src/plugins)会扫描该路径下所有含__init__.py的子目录并执行其中的export函数由nonebot.plugin.load_plugins自动触发。若插件需禁用只需重命名其目录如weather_off无需注释代码。2.4 创建第一个插件响应/help命令的文本帮助系统在src/plugins/help/__init__.py中编写最简插件# src/plugins/help/__init__.py from nonebot import on_command from nonebot.adapters.onebot.v11 import Message, MessageEvent # 定义命令匹配器响应群聊和私聊中的 /help help_cmd on_command(help, aliases{帮助, /help}, priority10, blockTrue) help_cmd.handle() async def send_help(event: MessageEvent): # 判断消息来源群聊 or 私聊 if event.group_id: target f群 {event.group_id} else: target 私聊 # 构建帮助文本支持 Markdown 风格换行 help_text ( f {target} 机器人帮助\n ────────────────\n • /help — 显示本帮助\n • /status — 查看机器人运行状态\n • /ping — 测试响应延迟\n • /weather 上海 — 查询指定城市天气需额外安装 weather 插件\n ────────────────\n 提示所有命令均不区分大小写支持中文别名 ) await help_cmd.finish(Message(help_text))逻辑说明on_command创建的匹配器默认监听群聊和私聊事件priority10表示该命令优先级为 10数值越小优先级越高系统内置命令通常为 1~5blockTrue表示匹配成功后阻断后续同类型匹配器执行避免多个插件同时响应同一命令。验证方式启动go-cqhttp后在终端执行nb run然后在 QQ 群中发送/help应立即收到格式化帮助文本。3. 实现多用途能力天气查询、RSS 订阅与群内投票的插件化落地3.1 天气查询插件调用高德 API 并结构化渲染天气功能需外部 API我们选用免费额度充足的高德地图 API需申请 key。插件结构为src/plugins/weather/核心文件__init__.py# src/plugins/weather/__init__.py import httpx from nonebot import on_command from nonebot.adapters.onebot.v11 import Message, MessageEvent from nonebot.params import CommandArg from pydantic import BaseModel class WeatherResponse(BaseModel): city: str temperature: str weather: str humidity: str winddirection: str weather_cmd on_command(weather, aliases{天气, 查天气}, priority5) weather_cmd.handle() async def query_weather(event: MessageEvent, city_name: Message CommandArg()): city city_name.extract_plain_text().strip() if not city: await weather_cmd.finish(请指定城市名称例如/weather 北京) # 调用高德 API替换 YOUR_AMAP_KEY async with httpx.AsyncClient() as client: try: resp await client.get( https://restapi.amap.com/v3/weather/weatherInfo, params{ key: YOUR_AMAP_KEY, city: await _get_city_code(client, city), extensions: base }, timeout10.0 ) data resp.json() if data[status] ! 1: raise ValueError(data.get(info, API 请求失败)) w WeatherResponse(**data[lives][0]) msg ( f️ {w.city} 天气\n f温度{w.temperature}℃\n f天气{w.weather}\n f湿度{w.humidity}\n f风向{w.winddirection} ) await weather_cmd.finish(Message(msg)) except httpx.TimeoutException: await weather_cmd.finish(⚠️ 请求超时请稍后重试) except Exception as e: await weather_cmd.finish(f❌ 查询失败{str(e)}) async def _get_city_code(client: httpx.AsyncClient, city_name: str) - str: 根据城市名获取高德 citycode resp await client.get( https://restapi.amap.com/v3/config/district, params{key: YOUR_AMAP_KEY, keywords: city_name, subdistrict: 0} ) districts resp.json().get(districts, []) return districts[0][adcode] if districts else 110000 # 默认北京参数说明CommandArg()自动提取命令后跟随的纯文本参数httpx.AsyncClient支持异步 HTTP 请求避免阻塞事件循环timeout10.0防止 API 响应慢拖垮整个机器人_get_city_code是辅助函数将城市名转为高德要求的adcode行政区划编码。3.2 RSS 订阅插件用 APScheduler 实现定时抓取与去重推送RSS 功能需定时轮询源站NoneBot2 官方插件nonebot-plugin-apscheduler提供无缝集成# 安装定时任务插件 pip install nonebot-plugin-apscheduler在src/plugins/rss/__init__.py中# src/plugins/rss/__init__.py from nonebot import require, on_command from nonebot.adapters.onebot.v11 import Message, MessageEvent, GroupMessageEvent from nonebot.plugin import PluginMetadata from nonebot_plugin_apscheduler import scheduler import feedparser import sqlite3 from datetime import datetime require(nonebot_plugin_apscheduler) rss_cmd on_command(rss, aliases{订阅, RSS}, priority5) # SQLite 存储已推送条目轻量级避免引入 Redis DB_PATH data/rss.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS rss_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, feed_url TEXT NOT NULL, entry_id TEXT UNIQUE NOT NULL, title TEXT NOT NULL, link TEXT NOT NULL, published TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() init_db() rss_cmd.handle() async def add_rss(event: GroupMessageEvent, url: Message CommandArg()): feed_url url.extract_plain_text().strip() if not feed_url.startswith((http://, https://)): await rss_cmd.finish(请输入有效的 RSS 地址例如/rss https://example.com/feed.xml) # 添加到数据库此处简化实际应校验 URL 可访问性 conn sqlite3.connect(DB_PATH) conn.execute(INSERT OR IGNORE INTO rss_items (feed_url, entry_id) VALUES (?, ?), (feed_url, placeholder)) conn.commit() conn.close() await rss_cmd.finish(f✅ 已订阅{feed_url}) # 每 15 分钟检查一次所有 RSS 源 scheduler.scheduled_job(interval, minutes15, idcheck_rss_feeds) async def check_rss_feeds(): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute(SELECT DISTINCT feed_url FROM rss_items) feeds cursor.fetchall() for (feed_url,) in feeds: try: feed feedparser.parse(feed_url) for entry in feed.entries[:3]: # 只检查最新 3 条 if not cursor.execute( SELECT 1 FROM rss_items WHERE entry_id ?, (entry.id,) ).fetchone(): # 新条目推送到所有已订阅群此处简化为固定群号 from nonebot import get_bot bot get_bot() msg f {entry.title}\n{entry.link} await bot.send_group_msg(group_id123456789, messagemsg) cursor.execute( INSERT INTO rss_items (feed_url, entry_id, title, link) VALUES (?, ?, ?, ?), (feed_url, entry.id, entry.title, entry.link) ) except Exception as e: print(fRSS 抓取失败 {feed_url}: {e}) conn.commit() conn.close()关键设计scheduler.scheduled_job装饰器将函数注册为定时任务interval类型按固定间隔执行SQLite 表rss_items以entry_id为主键实现天然去重get_bot()获取当前 Bot 实例调用send_group_msg主动推送消息需确保机器人已在目标群中。3.3 群内投票插件用 Matcher 状态机实现多轮交互投票功能需用户输入选项、确认、实时统计NoneBot2 的Matcher状态管理比手动维护字典更可靠# src/plugins/vote/__init__.py from nonebot import on_command from nonebot.adapters.onebot.v11 import Message, MessageEvent, GroupMessageEvent from nonebot.matcher import Matcher from nonebot.params import Arg, ArgPlainText, CommandArg from nonebot.rule import to_me vote_cmd on_command(vote, aliases{投票, 发起投票}, ruleto_me(), priority5) # 全局存储投票状态生产环境建议用 Redis _VOTE_STATE {} vote_cmd.handle() async def start_vote(matcher: Matcher, event: GroupMessageEvent, arg: Message CommandArg()): text arg.extract_plain_text().strip() if not text: await vote_cmd.finish(请用空格分隔选项例如/vote 吃火锅 吃烧烤 吃寿司) options [opt.strip() for opt in text.split() if opt.strip()] if len(options) 2: await vote_cmd.finish(至少需要 2 个选项) # 存储当前群的投票状态 group_id event.group_id _VOTE_STATE[group_id] { options: options, votes: {opt: 0 for opt in options}, voters: set() # 记录已投票用户 ID防重复 } await vote_cmd.send( f 投票已开启\n f选项{ | .join(f[{i1}] {opt} for i, opt in enumerate(options))}\n f请回复数字如 1选择或发送“结束”终止投票 ) # 设置下一步等待用户输入 matcher.set_arg(vote_choice, Arg()) vote_cmd.got(vote_choice, prompt请选择序号1,2,3...) async def handle_choice( matcher: Matcher, event: GroupMessageEvent, choice: str ArgPlainText(vote_choice) ): group_id event.group_id state _VOTE_STATE.get(group_id) if not state: await vote_cmd.finish(当前无进行中的投票请先发起。) try: idx int(choice) - 1 if 0 idx len(state[options]): option state[options][idx] user_id event.user_id if user_id not in state[voters]: state[voters].add(user_id) state[votes][option] 1 await vote_cmd.send(f✅ 您选择了{option}) else: await vote_cmd.send(⚠️ 您已投过票不能重复投票) else: await vote_cmd.send(❌ 选项序号超出范围请重新输入。) except ValueError: if choice 结束: await _show_result(matcher, group_id) _VOTE_STATE.pop(group_id, None) return await vote_cmd.send(❌ 请输入有效数字或“结束”。) async def _show_result(matcher: Matcher, group_id: int): state _VOTE_STATE.get(group_id) if not state: return total sum(state[votes].values()) result_lines [️ 投票结果] for opt, count in sorted(state[votes].items(), keylambda x: x[1], reverseTrue): pct f{count/total*100:.1f}% if total 0 else 0% result_lines.append(f • {opt}: {count} 票 ({pct})) result_lines.append(f 总票数{total}) await matcher.send(\n.join(result_lines))状态机说明matcher.set_arg()触发got事件vote_cmd.got()捕获用户下一条消息_VOTE_STATE字典按group_id隔离不同群的投票状态to_me()规则确保只有 机器人时才触发避免刷屏voters集合防止同一用户多次投票。4. 解决 nonebot2 插件冲突依赖隔离、加载顺序与调试技巧4.1 插件冲突的三大典型场景与定位方法NoneBot2 插件冲突并非框架 Bug而是模块间隐式耦合导致。常见场景包括场景表现定位命令同名命令覆盖执行/status时只响应某个插件另一个插件的同名命令失效nb plugin list查看所有已加载插件及其命令Pydantic 模型冲突启动时报ValidationError提示字段重复定义nb plugin show plugin-name检查插件依赖树全局中间件干扰某插件的日志突然消失或所有命令都返回空响应nb run --log-level DEBUG开启调试日志搜索matcher和event注意nb plugin list输出中Priority列显示命令匹配优先级数值越小越先匹配若两个插件都注册了on_command(status)且 priority 相同则按插件加载顺序目录字母序决定谁生效。4.2 用插件入口点entrypoint机制实现依赖隔离NoneBot2 推荐通过pyproject.toml声明插件入口点而非在bot.py中硬编码load_plugins。在pyproject.toml中添加[project.entry-points.nonebot.plugins] weather src.plugins.weather rss src.plugins.rss vote src.plugins.vote # 若某插件需条件加载如仅限特定群可设为可选 [tool.nonebot.plugins] optional [src.plugins.admin] # 此插件不会自动加载需手动 load_plugin这样做的好处nb plugin list可精确控制启用/禁用插件nb plugin disable weather插件间依赖关系显式化src/plugins/weather/pyproject.toml中声明requires [httpx]避免load_plugins(src/plugins)扫描到测试代码或废弃插件4.3 调试插件加载失败的四步法当nb run启动后插件未生效按此顺序排查检查插件目录结构确认src/plugins/name/__init__.py存在且无语法错误python -m py_compile src/plugins/weather/__init__.py验证插件入口点运行python -c import nonebot; print(nonebot.load_plugins(src/plugins))观察是否返回[]查看 import 错误在__init__.py开头添加print(Loading weather plugin)启动时观察是否打印检查 Pydantic 版本兼容性NoneBot2 v2.2 要求 Pydantic v2.x若插件依赖旧版pydantic2需升级插件或降级 NoneBot2不推荐4.4 生产环境插件热重载的边界与替代方案NoneBot2 的nb run --reload支持代码修改后自动重启但存在限制仅监控.py文件变化不监控pyproject.toml或config.yml修改bot.py或适配器配置需手动重启热重载期间go-cqhttp连接可能中断导致消息丢失推荐生产部署方案# 使用 systemd 管理进程Linux # /etc/systemd/system/qq-bot.service [Unit] DescriptionQQ Bot Service Afternetwork.target [Service] Typesimple Userbotuser WorkingDirectory/opt/my-qq-bot ExecStart/opt/my-qq-bot/venv/bin/nb run Restartalways RestartSec10 [Install] WantedBymulti-user.target关键参数Restartalways确保崩溃后自动恢复RestartSec10避免频繁重启WorkingDirectory必须指向项目根目录否则load_plugins无法定位插件。5. 高级技巧用自定义 Matcher 实现跨插件上下文感知与敏感词过滤5.1 构建跨插件共享的上下文管理器当多个插件需共享用户状态如语言偏好、所在群权限等级不应各自维护字典。NoneBot2 提供Matcher.state但更健壮的方式是创建全局 ContextManager# src/utils/context.py from typing import Dict, Any, Optional from nonebot import get_driver from nonebot.adapters.onebot.v11 import Event class BotContext: _storage: Dict[str, Dict[str, Any]] {} classmethod def get(cls, key: str, defaultNone) - Any: return cls._storage.get(key, default) classmethod def set(cls, key: str, value: Any): cls._storage[key] value classmethod def clear(cls, key: str): cls._storage.pop(key, None) # 在 bot.py 中初始化确保早于插件加载 driver get_driver() driver.on_startup async def init_context(): BotContext.set(plugin_config, {weather_api_key: xxx})在任意插件中使用# src/plugins/admin/__init__.py from src.utils.context import BotContext admin_cmd.handle() async def set_api_key(event: MessageEvent, arg: Message CommandArg()): key arg.extract_plain_text().strip() BotContext.set(weather_api_key, key) # 全局生效 await admin_cmd.finish(✅ API Key 已更新)5.2 用全局 Rule 实现敏感词实时拦截在src/plugins/sensitive/__init__.py中不注册命令而是注入全局 Rule# src/plugins/sensitive/__init__.py from nonebot import get_driver, on_message from nonebot.adapters.onebot.v11 import MessageEvent, Message from nonebot.rule import Rule from nonebot.matcher import Matcher # 敏感词列表生产环境应从数据库或远程配置中心加载 SENSITIVE_WORDS [违禁词1, 违禁词2] async def sensitive_rule(event: MessageEvent) - bool: msg event.get_message().extract_plain_text() return any(word in msg for word in SENSITIVE_WORDS) # 创建全局拦截器priority1最高优先级 sensitive_blocker on_message(ruleRule(sensitive_rule), priority1, blockTrue) sensitive_blocker.handle() async def handle_sensitive(event: MessageEvent): await sensitive_blocker.send(⚠️ 检测到不适宜内容已自动拦截) # 可选记录日志、通知管理员、调用审核 API # from nonebot.log import logger # logger.warning(fSensitive content from {event.user_id}: {event.get_message()})原理说明on_messageRule组合会在所有命令匹配前执行priority1确保它最先被检查blockTrue阻断后续所有匹配器使敏感消息完全不进入业务逻辑层。5.3 验证插件是否真正生效的三类检查命令不要仅靠 QQ 群里发命令测试用以下命令快速验证命令作用示例输出nb plugin list --enabled列出所有启用插件及命令weather: /weather, /天气nb run --log-level INFO --debug启动时输出详细加载日志INFO: Loaded plugin weathercurl -X POST http://127.0.0.1:8080/api/get_status直接调用 go-cqhttp API 检查连接{status:ok,retcode:0,data:{good:true}}最后一步在go-cqhttp日志中搜索ws connected确认 WebSocket 连接建立成功若出现connection refused检查bot.py中access-token是否与config.yml一致以及go-cqhttp是否在运行。本文还有配套的精品资源点击获取