- 后端
- 即时通讯
【免费下载链接】nonebot2
跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python
NoneBot2 的nonebot.matcher模块是整套框架的事件调度核心：它定义了"事件响应器"（Matcher）的创建、注册、匹配与运行机制，并提供了send、finish、pause、reject、got、receive等快捷方法，帮助开发者以极低的成本实现与用户的多轮对话交互。阅读本文后，你将掌握事件响应器的完整生命周期——从通过Matcher.new()或on_*辅助函数创建响应器、配置类型/规则/权限/优先级，到理解事件在响应器之间传播、处理函数按序执行以及暂停/拒绝/结束会话等控制流的底层实现。
模块定位：nonebot.matcher 是什么
根据 API 文档 的说明，nonebot.matcher模块实现事件响应器的创建与运行，并提供一些快捷方法来帮助用户更好地与机器人进行对话。
在 NoneBot2 中，事件响应器是对接收到的事件进行响应的基本单元，所有事件响应器都继承自Matcher基类。开发者可以通过一系列规则从事件流中筛选出具有某种特征的事件，再按照预定义的处理函数列表（handlers）对事件进行处理。
模块对外导出的核心对象在 nonebot/matcher.py 中定义，包括：
Matcher：事件响应器基类MatcherManager：事件响应器管理器（全局matchers对象即其实例）MatcherProvider：事件响应器存储器基类DEFAULT_PROVIDER_CLASS：默认存储器类型matchers：全局事件响应器存储current_bot、current_event、current_matcher、current_handler：四个ContextVar上下文变量，分别保存当前运行的 Bot、事件、事件响应器与处理函数，是send/finish等快捷方法取用上下文的基础
提示：上述
current_*上下文变量与MatcherSource类型也在 nonebot/internal/matcher/init.py 中被再次导出，作为内部实现与公开 API 的衔接层。
全局存储：matchers 与 DEFAULT_PROVIDER_CLASS
matchers是一个全局的 MatcherManager 实例，它实现了常用的字典操作，用于按优先级管理全部事件响应器：
- 键（key）：
int类型的优先级 - 值（value）：
list[type[Matcher]]类型，即该优先级下注册的全部事件响应器类
DEFAULT_PROVIDER_CLASS是默认存储器类型，实际指向_DictProvider（见 provider.py），它是一个继承自defaultdict[int, list[type[Matcher]]]与MatcherProvider的实现，默认以list作为工厂，因此访问不存在的优先级键时会自动得到空列表。优先级数值越小越先被匹配，这是理解事件调度顺序的关键。
MatcherManager完整实现了MutableMapping协议，支持keys()、values()、items()、get(key, default=None)、pop(key)、popitem()、clear()、update(m, /)、setdefault(key, default)等标准字典操作，并额外提供set_provider(provider_class)方法用于更换底层存储器（详见下文"自定义事件响应器存储器"一节）。
Matcher 类的类变量：事件响应器的静态属性
事件响应器本质上是一个类，其匹配特性全部由类变量（ClassVar）描述，在 nonebot/internal/matcher/matcher.py 中定义如下：
| 类变量 | 类型 | 默认值 | 说明 |
|---|---|---|---|
type | ClassVar[str] | "" | 事件响应器类型，与event.get_type()一致时触发，空字符串表示任意类型 |
rule | ClassVar[Rule] | Rule() | 事件响应器匹配规则 |
permission | ClassVar[Permission] | Permission() | 事件响应器触发权限 |
handlers | ClassVar[list[Dependent[Any]]] | [] | 事件响应器拥有的事件处理函数列表 |
priority | ClassVar[int] | 1 | 事件响应器优先级（越小越先匹配） |
block | bool | False | 事件响应器是否阻止事件向更低优先级传播 |
temp | ClassVar[bool] | False | 事件响应器是否为临时（触发一次后删除） |
expire_time | ClassVar[datetime \| None] | None | 事件响应器过期时间点，过时即被删除 |
其中block虽然声明为普通实例属性，但在运行期通过stop_propagation()或捕获StopPropagation异常时被置为True，从而阻止事件继续向更低优先级响应器传播（相关处理见simple_run与 exception.py 中的StopPropagation说明）。
此外，Matcher还维护了_default_state（默认状态state）、_default_type_updater（默认事件类型更新函数）、_default_permission_updater（默认会话权限更新函数）等内部类变量，以及MatcherSource源代码上下文信息（包含plugin_id、module_name、lineno，可用于追溯响应器定义所在插件与行号）。
创建事件响应器：Matcher.new()
Matcher.new()是创建事件响应器的底层方法，签名如下：
@classmethod def new( cls, type_: str = "", rule: Rule | None = None, permission: Permission | None = None, handlers: list[T_Handler | Dependent[Any]] | None = None, temp: bool = False, priority: int = 1, block: bool = False, *, plugin: Plugin | None = None, # Deprecated，请改用 source module: ModuleType | None = None, # Deprecated，请改用 source source: MatcherSource | None = None, expire_time: datetime | timedelta | None = None, default_state: T_State | None = None, default_type_updater: T_TypeUpdater | Dependent[str] | None = None, default_permission_updater: T_PermissionUpdater | Dependent[Permission] | None = None, ) -> type[Matcher]各参数含义（与 API 文档 一致）：
type_：事件响应器类型，与event.get_type()一致时触发，空字符串表示任意；rule：匹配规则，未提供时默认Rule()（空规则）；permission：触发权限，未提供时默认Permission()（空权限，即不限制）；handlers：事件处理函数列表，元素可以是普通函数（T_Handler）或已解析的Dependent对象，普通函数会被Dependent.parse()包装；temp：是否临时事件响应器，True时触发一次即被删除；priority：响应优先级，数值越小越先匹配；block：是否阻止事件向更低优先级响应器传播；plugin/module：已弃用，传入会触发DeprecationWarning，请改用source；source：MatcherSource类型的源代码上下文信息（插件 ID、模块名、行号）；expire_time：接受datetime（绝对时间点）或timedelta（相对时长，内部会转换为datetime.now() + expire_time），过时即被删除；default_state：默认状态state，供处理函数通过T_State依赖读取；default_type_updater/default_permission_updater：默认事件类型更新函数与默认会话权限更新函数，普通函数同样会被解析为Dependent。
从源码实现看（matcher.py），new()会通过type()动态创建一个继承自当前Matcher子类的新类，将上述属性写入新类的命名空间，然后执行matchers[priority].append(NewMatcher)将其注册到全局存储中，最后返回这个新的事件响应器类。每次调用new()都会生成一个全新的类，因此每个事件响应器都拥有独立的rule、permission、handlers等属性，互不干扰。
对应地，Matcher.destroy()类方法从matchers[cls.priority]中移除当前响应器，实现销毁操作。
辅助函数：更优雅的创建方式
直接调用Matcher.new()过于繁琐且不能自动记录插件信息，因此 NoneBot2 在 nonebot/plugin/on.py 中提供了一系列"事件响应器辅助函数"，以on()或on_<type/rule>()形式出现，调用后返回一个新的type[Matcher]：
on(type="", rule=None, permission=None, ...)：注册基础事件响应器，可自定义类型；on_message/on_notice/on_request/on_metaevent：分别注册消息、通知、请求、元事件响应器；on_startswith(msg, ignorecase=False)/on_endswith/on_fullmatch/on_keyword(keywords)：按消息文本前缀、后缀、全匹配、关键词匹配注册消息响应器；on_command(cmd, aliases=None, force_whitespace=None)：按命令形式匹配注册响应器；on_shell_command(cmd, aliases=None, parser=None)：注册支持 shell 风格参数解析的命令响应器；on_regex(pattern, flags=0)：按正则匹配注册响应器；on_type(types, ...)：按事件类型注册响应器。
以官方教程中的用法为例（见 教程文档）：
from nonebot import on_command from nonebot.rule import to_me weather = on_command( "天气", rule=to_me(), aliases={"weather", "查天气"}, priority=10, block=True )这创建了一个可以响应天气、weather、查天气三个命令、要求私聊或 @ 机器人（to_me()规则）才会响应、优先级为 10 且阻断事件向后续优先级传播的响应器。从源码看，on_command内部会调用on_message(command(*commands, force_whitespace=force_whitespace) & rule, **kwargs)，on_message又默认block=True并调用on("message", ...)，最终由on()调用Matcher.new()并通过get_matcher_source()捕获定义位置生成MatcherSource，再通过store_matcher()将响应器记录到当前加载插件中。这也解释了为什么使用辅助函数创建的响应器能被插件系统正确追踪（plugin/on.py）。
此外还提供CommandGroup（具有共同命令名称前缀的命令组，支持prefix_aliases参数为别名自动加前缀）与MatcherGroup（具有共同参数的响应器组）两个组合类，均支持链式注册一组响应器。
匹配检查：check_perm 与 check_rule
事件到达后，NoneBot2 会先检查事件响应器是否"有权"响应、是否"匹配"。两个类方法的实现逻辑（matcher.py）都包含两步：
event_type = event.get_type() return event_type == (cls.type or event_type) and await ...即：先比较事件类型——若cls.type非空则要求event.get_type()与之相等，若为空字符串（任意类型）则直接通过；再执行权限/规则检查。
check_perm(bot, event, stack=None, dependency_cache=None) -> bool：调用cls.permission(bot, event, stack, dependency_cache)检查是否满足触发权限；check_rule(bot, event, state, stack=None, dependency_cache=None) -> bool：调用cls.rule(bot, event, state, stack, dependency_cache)检查是否满足匹配规则。
Rule与Permission的具体定义与组合方式见 rule 模块 与 permission 模块；stack是AsyncExitStack异步上下文栈，dependency_cache是依赖缓存（T_DependencyCache），二者用于在依赖解析中复用已初始化的资源。
添加处理函数：handle、append_handler、receive、got
事件响应器的"行为"由一系列处理函数（handler）定义，每个处理函数会被解析为Dependent对象存入handlers列表，运行时按顺序执行。
append_handler(handler, parameterless=None) -> Dependent[Any]：直接向handlers追加一个处理函数，parameterless是非参数类型依赖列表（如Depends(...)、Event.is_tome()等无需参数注入的对象），返回解析后的Dependent；handle(parameterless=None)：装饰器，等价于"装饰一个函数来向事件响应器直接添加一个处理函数"：
@weather.handle() async def handle_first(bot: Bot, event: Event): await weather.send("收到你的消息了")receive(id="", parameterless=None)：装饰器，指示 NoneBot 在接收用户新的一条消息后继续运行该函数，id为消息 ID（默认空字符串）。其内部注入了一个Depends(_receive)前置依赖：先通过set_target记录目标，若_receive_{id}已存在则直接复用，否则调用reject()等待下一条消息（matcher.py）；got(key, prompt=None, parameterless=None)：装饰器，指示 NoneBot 获取一个参数key。当key不存在时发送prompt提示消息并接收用户新的一条消息后再运行该函数；若key已存在则直接继续运行。其内部同样通过_key_getter依赖实现：命中目标后调用set_arg(key, event.get_message())将用户消息存入状态，未命中则调用reject(prompt)等待输入（matcher.py）：
from nonebot import on_command from nonebot.adapters import Message weather = on_command("天气") @weather.got("city", prompt="你想查询哪个城市的天气？") async def handle_city(city: Message = Arg()): await weather.finish(f"正在查询 {city} 的天气……")细节：
receive与got在装饰时若发现handlers中最后一个处理函数与当前函数相同（即连续装饰同一函数），会将新依赖合并进已有Dependent而非重复追加，这是@matcher.got("key")与@matcher.receive()叠加使用的实现基础。
对话快捷方法：send、finish、pause、reject、skip
Matcher提供了一组以"发消息 + 控制会话流程"为核心的快捷方法，它们都依赖current_bot、current_event等上下文变量取用当前交互对象：
| 方法 | 说明 | 底层行为 |
|---|---|---|
send(message, **kwargs) -> Any | 发送一条消息给当前交互用户 | 调用bot.send(event=event, message=...)；若message是MessageTemplate，会先用当前state渲染模板（message.format(**state)）。**kwargs为Bot.send的参数，具体参考对应 adapter 的 bot 对象 API |
finish(message=None, **kwargs) -> NoReturn | 发送消息并结束当前事件响应器 | 发送消息后抛出FinishedException，run()捕获后不再继续任何处理函数 |
pause(prompt=None, **kwargs) -> NoReturn | 发送消息并暂停响应器，接收用户新消息后继续下一个处理函数 | 发送prompt后将发送结果存入state[PAUSE_PROMPT_RESULT_KEY]，然后抛出PausedException |
reject(prompt=None, **kwargs) -> NoReturn | 最近用got/receive接收的消息不符合预期，发送消息并中断在当前位置，接收新事件后从头重新执行当前处理函数 | 发送prompt后将结果存入state[REJECT_PROMPT_RESULT_KEY.format(key=target)]，抛出RejectedException |
reject_arg(key, prompt=None, **kwargs) -> NoReturn | 针对got的定向拒绝 | 设置ARG_KEY.format(key=key)为目标后抛出RejectedException |
reject_receive(id="", prompt=None, **kwargs) -> NoReturn | 针对receive的定向拒绝 | 设置RECEIVE_KEY.format(id=id)为目标后抛出RejectedException |
skip() -> NoReturn | 跳过当前事件处理函数，继续下一个处理函数 | 抛出SkippedException，通常在事件处理函数的依赖中使用（如Matcher.skip()） |
finish/pause/reject抛出的三个异常分别对应 nonebot/exception.py 中的FinishedException、PausedException、RejectedException，它们都是MatcherException的子类，在run()中被特殊捕获处理（见下文"运行机制"）。官方文档对它们的描述分别是：
PausedException：指示 NoneBot 结束当前Handler并等待下一条消息后继续下一个Handler；RejectedException：指示 NoneBot 结束当前Handler并等待下一条消息后重新运行当前Handler；FinishedException：指示 NoneBot 结束当前Handler且后续Handler不再被运行，可用于结束用户会话。
典型多轮对话示例：pause 与 reject 的区别
from nonebot import on_command weather = on_command("天气") @weather.handle() async def handle_first(): await weather.pause("请发送城市名") # 暂停，等用户回复后运行下一个处理函数 @weather.handle() async def handle_second(city: Message = Arg()): await weather.finish(f"正在查询 {city} 的天气……")如果希望用户输入"不符合预期"时原地重试当前函数，则应使用reject：
@weather.handle() async def handle_city(city: Message = Arg("city", prompt="请输入城市名")): if city.extract_plain_text() == "不查了": await weather.finish("好的，再见！") await weather.reject("城市名无效，请重新输入") # 重新执行本函数会话状态存取：get/set 系列方法
Matcher实例（__init__中self.state = self._default_state.copy()）维护一份独立的会话状态字典，以下方法用于读写其中的关键数据（存储键名见 nonebot/consts.py）：
get_receive(id, default=None)/set_receive(id, event)：按_receive_{id}键读取/写入某个receive事件；set_receive同时会更新_last_receive；get_last_receive(default=None)：读取最近一次receive事件（_last_receive键），没有时返回default；get_arg(key, default=None)/set_arg(key, message)：按{key}键读取/写入某个got消息，get_arg返回Message或default；set_target(target, cache=True)/get_target(default=None)：管理"当前 reject 目标"。cache=True时写入_next_target（缓存目标），否则写入_current_target；get_target读取_current_target。当reject触发后，resolve_reject()会将缓存的_next_target提升为_current_target，从而让下一次got/receive正确命中用户新消息对应的处理位置；stop_propagation()：将self.block置为True，阻止事件传播（等价于创建时block=True）。
这些键的设计与receive/got装饰器内部的_key_getter、_receive依赖完全对应，理解它们有助于排查多轮对话中的状态问题。
运行机制：simple_run 与 run
当事件被匹配后，NoneBot2 会调用Matcher.run(bot, event, state, stack=None, dependency_cache=None)执行整个处理流程，其核心逻辑分为两层（matcher.py）：
simple_run（实际执行层）：
- 通过
ensure_context(bot, event)上下文管理器设置current_bot、current_event、current_matcher三个上下文变量（结束时自动 reset），使send等快捷方法能够取用当前交互对象； self.state.update(state)刷新预处理状态；- 循环从
self.remain_handlers（实例属性，初始为handlers.copy()）中逐个弹出处理函数，通过current_handler.set(handler)记录当前处理函数，然后调用Dependent执行； - 使用
exceptiongroup.catch捕获SkippedException（跳过当前处理函数）与StopPropagation（将self.block置为True，终止事件向下层传播）。
run（会话控制层）：包裹simple_run，通过catch捕获FinishedException、RejectedException、PausedException三种"会话控制异常"（若同时抛出多个会记录警告并按Finished > Rejected > Paused的顺序择优处理），然后分别处理：
FinishedException：直接结束，不做任何后续操作；RejectedException：先resolve_reject()（将当前处理函数插回remain_handlers头部、把缓存的_next_target提升为当前目标），再通过update_type()/update_permission()计算新的类型与权限，最后以temp=True、priority=0、block=True、expire_time=bot.config.session_expire_timeout的参数调用self.new(...)创建一个新的临时响应器来接续会话——这就是多轮会话"暂停后续航"的实现方式；PausedException：同样创建一个新的临时响应器（remain_handlers中剩余的处理函数会继续执行），但不会把当前处理函数插回头部（即"继续下一个处理函数"而非"重跑当前函数"）。
update_type(bot, event, stack, dependency_cache) -> str会调用_default_type_updater（未设置时默认返回"message"），update_permission(...) -> Permission会调用_default_permission_updater（未设置时默认返回Permission(User.from_event(event, perm=self.permission))，即把当前会话用户纳入权限范围）。这两个方法分别对应装饰器@matcher.type_updater(func)与@matcher.permission_updater(func)，可自定义会话续接时的类型与权限规则。
事件响应器管理器：MatcherManager 与自定义存储器
MatcherManager是 nonebot/internal/matcher/manager.py 中定义的响应器管理器，它继承MutableMapping[int, list[type[Matcher]]]，将字典操作全部委托给内部的provider（MatcherProvider实例）。全局的matchers对象即为MatcherManager()实例，其provider默认为DEFAULT_PROVIDER_CLASS({})。
MatcherProvider是事件响应器存储器基类（provider.py），抽象方法为__init__(self, matchers: Mapping[int, list[type[Matcher]]])，其中matchers是当前存储器中已有的事件响应器。默认实现_DictProvider基于defaultdict(list)。
通过matchers.set_provider(provider_class)可以更换底层存储器（如切换为基于数据库或 Redis 的持久化存储），新 provider 初始化时会被传入当前已存在的全部响应器，从而完成迁移。更换后，Matcher.new()注册与Matcher.destroy()注销操作将自动作用于新存储器。
结语
nonebot.matcher模块是 NoneBot2 事件驱动架构的中枢：Matcher.new()与on_*辅助函数负责响应器的创建与注册，check_perm/check_rule决定事件能否被响应，handlers链表与run/simple_run驱动处理函数按序执行，而send/finish/pause/reject/got/receive则把多轮会话控制封装成了极简的 API。无论是编写第一个插件，还是深度定制会话行为，理解本模块的类变量、方法语义与异常控制流，都是掌握 NoneBot2 开发的关键一步。更多配套概念可继续阅读 事件响应器进阶、规则模块、权限模块 与 依赖注入模块。
- 后端
- 即时通讯
【免费下载链接】nonebot2
跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python
相关推荐
NoneBot2 事件响应器（Matcher）进阶指南：组成机制、内置响应规则与响应器组实战
NoneBot2 事件响应器（Matcher）进阶指南：组成机制、内置响应规则与响应器组实战 本篇进阶指南聚焦 NoneBot2 事件响应器（Matcher）的
后端即时通讯3个关键步骤，让xiaomusic在Windows上流畅运行小爱音箱音乐
3个关键步骤，让xiaomusic在Windows上流畅运行小爱音箱音乐 xiaomusic是一个开源音乐播放项目，专为小爱音箱用户设计，通过yt dlp技术实
后端智能硬件音视频NoneBot2 事件响应器(Matcher)详解与实战指南
NoneBot2 事件响应器 Matcher 详解与实战指南 什么是事件响应器 在 NoneBot2 框架中，事件响应器 Matcher 是构建机器人功能的核心
后端即时通讯
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
