后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载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_CLASSmatchers是一个全局的 MatcherManager 实例它实现了常用的字典操作用于按优先级管理全部事件响应器键keyint类型的优先级值valuelist[type[Matcher]]类型即该优先级下注册的全部事件响应器类DEFAULT_PROVIDER_CLASS是默认存储器类型实际指向_DictProvider见 provider.py它是一个继承自defaultdict[int, list[type[Matcher]]]与MatcherProvider的实现默认以list作为工厂因此访问不存在的优先级键时会自动得到空列表。优先级数值越小越先被匹配这是理解事件调度顺序的关键。MatcherManager完整实现了MutableMapping协议支持keys()、values()、items()、get(key, defaultNone)、pop(key)、popitem()、clear()、update(m, /)、setdefault(key, default)等标准字典操作并额外提供set_provider(provider_class)方法用于更换底层存储器详见下文自定义事件响应器存储器一节。Matcher 类的类变量事件响应器的静态属性事件响应器本质上是一个类其匹配特性全部由类变量ClassVar描述在 nonebot/internal/matcher/matcher.py 中定义如下类变量类型默认值说明typeClassVar[str]事件响应器类型与event.get_type()一致时触发空字符串表示任意类型ruleClassVar[Rule]Rule()事件响应器匹配规则permissionClassVar[Permission]Permission()事件响应器触发权限handlersClassVar[list[Dependent[Any]]][]事件响应器拥有的事件处理函数列表priorityClassVar[int]1事件响应器优先级越小越先匹配blockboolFalse事件响应器是否阻止事件向更低优先级传播tempClassVar[bool]False事件响应器是否为临时触发一次后删除expire_timeClassVar[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请改用sourcesourceMatcherSource类型的源代码上下文信息插件 ID、模块名、行号expire_time接受datetime绝对时间点或timedelta相对时长内部会转换为datetime.now() expire_time过时即被删除default_state默认状态state供处理函数通过T_State依赖读取default_type_updater/default_permission_updater默认事件类型更新函数与默认会话权限更新函数普通函数同样会被解析为Dependent。从源码实现看matcher.pynew()会通过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, ruleNone, permissionNone, ...)注册基础事件响应器可自定义类型on_message/on_notice/on_request/on_metaevent分别注册消息、通知、请求、元事件响应器on_startswith(msg, ignorecaseFalse)/on_endswith/on_fullmatch/on_keyword(keywords)按消息文本前缀、后缀、全匹配、关键词匹配注册消息响应器on_command(cmd, aliasesNone, force_whitespaceNone)按命令形式匹配注册响应器on_shell_command(cmd, aliasesNone, parserNone)注册支持 shell 风格参数解析的命令响应器on_regex(pattern, flags0)按正则匹配注册响应器on_type(types, ...)按事件类型注册响应器。以官方教程中的用法为例见 教程文档from nonebot import on_command from nonebot.rule import to_me weather on_command( 天气, ruleto_me(), aliases{weather, 查天气}, priority10, blockTrue )这创建了一个可以响应天气、weather、查天气三个命令、要求私聊或 机器人to_me()规则才会响应、优先级为 10 且阻断事件向后续优先级传播的响应器。从源码看on_command内部会调用on_message(command(*commands, force_whitespaceforce_whitespace) rule, **kwargs)on_message又默认blockTrue并调用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, stackNone, dependency_cacheNone) - bool调用cls.permission(bot, event, stack, dependency_cache)检查是否满足触发权限check_rule(bot, event, state, stackNone, dependency_cacheNone) - 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, parameterlessNone) - Dependent[Any]直接向handlers追加一个处理函数parameterless是非参数类型依赖列表如Depends(...)、Event.is_tome()等无需参数注入的对象返回解析后的Dependenthandle(parameterlessNone)装饰器等价于装饰一个函数来向事件响应器直接添加一个处理函数weather.handle() async def handle_first(bot: Bot, event: Event): await weather.send(收到你的消息了)receive(id, parameterlessNone)装饰器指示 NoneBot 在接收用户新的一条消息后继续运行该函数id为消息 ID默认空字符串。其内部注入了一个Depends(_receive)前置依赖先通过set_target记录目标若_receive_{id}已存在则直接复用否则调用reject()等待下一条消息matcher.pygot(key, promptNone, parameterlessNone)装饰器指示 NoneBot 获取一个参数key。当key不存在时发送prompt提示消息并接收用户新的一条消息后再运行该函数若key已存在则直接继续运行。其内部同样通过_key_getter依赖实现命中目标后调用set_arg(key, event.get_message())将用户消息存入状态未命中则调用reject(prompt)等待输入matcher.pyfrom 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、skipMatcher提供了一组以发消息 控制会话流程为核心的快捷方法它们都依赖current_bot、current_event等上下文变量取用当前交互对象方法说明底层行为send(message, **kwargs) - Any发送一条消息给当前交互用户调用bot.send(eventevent, message...)若message是MessageTemplate会先用当前state渲染模板message.format(**state)。**kwargs为Bot.send的参数具体参考对应 adapter 的 bot 对象 APIfinish(messageNone, **kwargs) - NoReturn发送消息并结束当前事件响应器发送消息后抛出FinishedExceptionrun()捕获后不再继续任何处理函数pause(promptNone, **kwargs) - NoReturn发送消息并暂停响应器接收用户新消息后继续下一个处理函数发送prompt后将发送结果存入state[PAUSE_PROMPT_RESULT_KEY]然后抛出PausedExceptionreject(promptNone, **kwargs) - NoReturn最近用got/receive接收的消息不符合预期发送消息并中断在当前位置接收新事件后从头重新执行当前处理函数发送prompt后将结果存入state[REJECT_PROMPT_RESULT_KEY.format(keytarget)]抛出RejectedExceptionreject_arg(key, promptNone, **kwargs) - NoReturn针对got的定向拒绝设置ARG_KEY.format(keykey)为目标后抛出RejectedExceptionreject_receive(id, promptNone, **kwargs) - NoReturn针对receive的定向拒绝设置RECEIVE_KEY.format(idid)为目标后抛出RejectedExceptionskip() - NoReturn跳过当前事件处理函数继续下一个处理函数抛出SkippedException通常在事件处理函数的依赖中使用如Matcher.skip()finish/pause/reject抛出的三个异常分别对应 nonebot/exception.py 中的FinishedException、PausedException、RejectedException它们都是MatcherException的子类在run()中被特殊捕获处理见下文运行机制。官方文档对它们的描述分别是PausedException指示 NoneBot 结束当前Handler并等待下一条消息后继续下一个HandlerRejectedException指示 NoneBot 结束当前Handler并等待下一条消息后重新运行当前HandlerFinishedException指示 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} 的天气……)如果希望用户输入不符合预期时原地重试当前函数则应使用rejectweather.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.pyget_receive(id, defaultNone)/set_receive(id, event)按_receive_{id}键读取/写入某个receive事件set_receive同时会更新_last_receiveget_last_receive(defaultNone)读取最近一次receive事件_last_receive键没有时返回defaultget_arg(key, defaultNone)/set_arg(key, message)按{key}键读取/写入某个got消息get_arg返回Message或defaultset_target(target, cacheTrue)/get_target(defaultNone)管理当前 reject 目标。cacheTrue时写入_next_target缓存目标否则写入_current_targetget_target读取_current_target。当reject触发后resolve_reject()会将缓存的_next_target提升为_current_target从而让下一次got/receive正确命中用户新消息对应的处理位置stop_propagation()将self.block置为True阻止事件传播等价于创建时blockTrue。这些键的设计与receive/got装饰器内部的_key_getter、_receive依赖完全对应理解它们有助于排查多轮对话中的状态问题。运行机制simple_run 与 run当事件被匹配后NoneBot2 会调用Matcher.run(bot, event, state, stackNone, dependency_cacheNone)执行整个处理流程其核心逻辑分为两层matcher.pysimple_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()计算新的类型与权限最后以tempTrue、priority0、blockTrue、expire_timebot.config.session_expire_timeout的参数调用self.new(...)创建一个新的临时响应器来接续会话——这就是多轮会话暂停后续航的实现方式PausedException同样创建一个新的临时响应器remain_handlers中剩余的处理函数会继续执行但不会把当前处理函数插回头部即继续下一个处理函数而非重跑当前函数。update_type(bot, event, stack, dependency_cache) - str会调用_default_type_updater未设置时默认返回messageupdate_permission(...) - Permission会调用_default_permission_updater未设置时默认返回Permission(User.from_event(event, permself.permission))即把当前会话用户纳入权限范围。这两个方法分别对应装饰器matcher.type_updater(func)与matcher.permission_updater(func)可自定义会话续接时的类型与权限规则。事件响应器管理器MatcherManager 与自定义存储器MatcherManager是 nonebot/internal/matcher/manager.py 中定义的响应器管理器它继承MutableMapping[int, list[type[Matcher]]]将字典操作全部委托给内部的providerMatcherProvider实例。全局的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项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载相关推荐NoneBot2 事件响应器Matcher进阶指南组成机制、内置响应规则与响应器组实战NoneBot2 事件响应器Matcher进阶指南组成机制、内置响应规则与响应器组实战 本篇进阶指南聚焦 NoneBot2 事件响应器Matcher的后端即时通讯3个关键步骤让xiaomusic在Windows上流畅运行小爱音箱音乐3个关键步骤让xiaomusic在Windows上流畅运行小爱音箱音乐 xiaomusic是一个开源音乐播放项目专为小爱音箱用户设计通过yt dlp技术实后端智能硬件音视频NoneBot2 事件响应器(Matcher)详解与实战指南NoneBot2 事件响应器 Matcher 详解与实战指南 什么是事件响应器 在 NoneBot2 框架中事件响应器 Matcher 是构建机器人功能的核心后端即时通讯上一篇Kubernetes Cluster Autoscaler ProvisioningRequest 详解为 Pod 群体按需、原子化预置集群容量下一篇BBDown免费开源B站下载器完整指南从零开始掌握命令行视频下载创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考