Skip to main content
Version: Next

会话控制

指南weather 插件中,我们使用依赖注入获取了机器人用户发送的地名参数,并根据地名参数进行相应的回复。但是,一问一答的对话模式仅仅适用于简单的对话场景,如果我们想要实现更复杂的对话模式,就需要使用会话控制。

询问并获取用户输入

weather 插件中,我们对于用户未输入地名参数的情况直接回复了 请输入地名 并结束了事件流程。但是,这样用户体验并不好,需要重新输入指令和地名参数才能获取天气回复。我们现在来实现询问并获取用户地名参数的功能。

询问用户

我们可以使用事件响应器操作中的 got 装饰器来表示当前事件处理流程需要询问并获取用户输入的消息:

@weather.handle()
async def handle_function(args: Message = CommandArg()):
if location := args.extract_plain_text():
await weather.finish(f"今天{location}的天气是...")

@weather.got("location", prompt="请输入地名")
async def got_location():
...

在上面的代码中,我们使用 got 事件响应器操作来向用户发送 prompt 消息,并等待用户的回复。用户的回复消息将会被作为 location 参数存储于事件响应器状态中。

提示

事件处理函数根据定义的顺序依次执行。

获取用户输入

在询问以及用户回复之后,我们就可以获取到我们需要的 location 参数了。我们使用 ArgPlainText 依赖注入来获取参数纯文本信息:

from nonebot.params import ArgPlainText

@weather.handle()
async def handle_function(args: Message = CommandArg()):
if location := args.extract_plain_text():
await weather.finish(f"今天{location}的天气是...")

@weather.got("location", prompt="请输入地名")
async def got_location(location: str = ArgPlainText()):
await weather.finish(f"今天{location}的天气是...")
NoneBot
/天气
请输入地名
北京
今天北京的天气是...

在上面的代码中,我们在 got_location 函数中定义了一个依赖注入参数 location,他的值将会是用户回复的消息纯文本信息。获取到用户输入的地名参数后,我们就可以进行天气查询并回复了。

提示

如果想要获取用户回复的消息对象 Message ,可以使用 Arg 依赖注入。

跳过询问

在上面的代码中,如果用户在输入天气指令时,同时提供了地名参数,我们直接回复了天气信息,这部分的逻辑是和询问用户地名参数之后的逻辑一致的。如果在复杂的业务场景下,我们希望这部分代码应该复用以减少代码冗余。我们可以使用事件响应器操作中的 set_arg 来主动设置一个参数:

from nonebot.matcher import Matcher

@weather.handle()
async def handle_function(matcher: Matcher, args: Message = CommandArg()):
if args.extract_plain_text():
matcher.set_arg("location", args)

@weather.got("location", prompt="请输入地名")
async def got_location(location: str = ArgPlainText()):
await weather.finish(f"今天{location}的天气是...")

请注意,设置参数需要使用依赖注入来获取 Matcher 实例以确保上下文正确,且参数值应为 Message 对象。

location 参数被设置之后,got 事件响应器操作将不再会询问并等待用户的回复,而是直接进入 got_location 函数。

请求重新输入

在实际的业务场景中,用户的输入很有可能并非是我们所期望的,而结束事件处理流程让用户重新发送指令也不是一个好的体验。这时我们可以使用 reject 事件响应器操作来请求用户重新输入:

@weather.handle()
async def handle_function(matcher: Matcher, args: Message = CommandArg()):
if args.extract_plain_text():
matcher.set_arg("location", args)

@weather.got("location", prompt="请输入地名")
async def got_location(location: str = ArgPlainText()):
if location not in ["北京", "上海", "广州", "深圳"]:
await weather.reject(f"你想查询的城市 {location} 暂不支持,请重新输入!")
await weather.finish(f"今天{location}的天气是...")
NoneBot
/天气
请输入地名
南京
你想查询的城市 南京 暂不支持,请重新输入!
北京
今天北京的天气是...

在上面的代码中,我们在 got_location 函数中判断用户输入的地名是否在支持的城市列表中,如果不在,则使用 reject 事件响应器操作。操作将会向用户发送 reject 参数中的消息,并等待用户回复后,重新执行 got_location 函数。通过 gotreject 事件响应器操作,我们实现了类似于循环的执行方式。

reject 事件响应器操作与 finish 类似,NoneBot 会在向机器人用户发送消息内容后抛出 RejectedException 异常来暂停事件响应流程以等待用户输入。也就是说,在 reject 被执行后,后续的程序同样是不会被执行的。

更多事件响应器操作

在之前的章节中,我们已经大致了解了五个事件响应器操作:handlegotfinishsendreject。现在我们来完整地介绍一下这些操作。

事件响应器操作可以分为两大类:交互操作流程控制操作。我们可以通过交互操作来与用户进行交互,而流程控制操作则可以用来控制事件处理流程的执行。

提示

事件处理流程按照事件处理函数添加顺序执行,已经结束的事件处理函数不可能被恢复执行。

handle

handle 事件响应器操作是一个装饰器,用于向事件处理流程添加一个事件处理函数。

@matcher.handle()
async def handle_func():
...

handle 装饰器支持嵌套操作,即一个事件处理函数可以被添加多次:

@matcher.handle()
@matcher.handle()
async def handle_func():
# 这个函数会被执行两次
...

got

got 事件响应器操作也是一个装饰器,它会在当前装饰的事件处理函数运行之前,中断当前事件处理流程,等待接收一个新的事件。它可以通过 prompt 参数来向用户发送询问消息,然后等待用户的回复消息,贴近对话形式会话。

got 装饰器接受一个参数 key 和一个可选参数 prompt。当会话状态中不存在 key 对应的消息时,会向用户发送 prompt 参数的消息,并等待用户回复。prompt 参数的类型和 send 事件响应器操作的参数类型一致。

在事件处理函数中,可以通过依赖注入的方式来获取接收到的消息,参考:ArgArgStrArgPlainText

@matcher.got("key", prompt="请输入...")
async def got_func(key: Message = Arg()):
...

got 装饰器支持与 gotreceive 装饰器嵌套操作,即一个事件处理函数可以在接收多个事件或消息后执行:

@matcher.got("key1", prompt="请输入key1...")
@matcher.got("key2", prompt="请输入key2...")
@matcher.receive("key3")
async def got_func(key1: Message = Arg(), key2: Message = Arg(), key3: Event = Received("key3")):
...

receive

receive 事件响应器操作也是一个装饰器,它会在当前装饰的事件处理函数运行之前,中断当前事件处理流程,等待接收一个新的事件。与 got 不同的是,receive 不会向用户发送询问消息,并且等待一个用户事件。可以接收的事件类型取决于会话更新

receive 装饰器接受一个可选参数 id,用于标识当前需要接收的事件,如果不指定,则默认为空 ""

在事件处理函数中,可以通过依赖注入的方式来获取接收到的事件,参考:ReceivedLastReceived

@matcher.receive("id")
async def receive_func(event: Event = Received("id")):
...

receive 装饰器支持与 gotreceive 装饰器嵌套操作,即一个事件处理函数可以在接收多个事件或消息后执行:

@matcher.receive("key1")
@matcher.got("key2", prompt="请输入key2...")
@matcher.got("key3", prompt="请输入key3...")
async def receive_func(key1: Event = Received("key1"), key2: Message = Arg(), key3: Message = Arg()):
...

send

send 事件响应器操作用于向用户回复一条消息。协议适配器会根据当前 event 选择回复的途径。

send 操作接受一个参数 message 和其他任何协议适配器接受的参数。message 参数类型可以是字符串、消息序列、消息段或者消息模板。消息模板将会使用会话状态字典进行渲染后发送。

这个操作等同于使用 bot.send(event, message, **kwargs),但不需要自行传入 event

@matcher.handle()
async def _():
await matcher.send("Hello world!")

finish

向用户回复一条消息(可选),并立即结束整个处理流程

参数与 send 相同。

@matcher.handle()
async def _():
await matcher.finish("Hello world!")
# 下面的代码不会被执行

pause

向用户回复一条消息(可选),立即结束当前事件处理函数,等待接收一个新的事件后进入下一个事件处理函数。

参数与 send 相同。

@matcher.handle()
async def _():
if need_confirm:
await matcher.pause("请在两分钟内确认执行")

@matcher.handle()
async def _():
...

reject

向用户回复一条消息(可选),立即结束当前事件处理函数,等待接收一个新的事件后再次执行当前事件处理函数。

reject 可以用于拒绝当前 receive 接收的事件或 got 接收的参数。通常在用户回复不符合格式或标准需要重新输入,或者用于循环进行用户交互。

参数与 send 相同。

@matcher.got("arg")
async def _(arg: str = ArgPlainText()):
if not is_valid(arg):
await matcher.reject("Invalid arg!")

reject_arg

向用户回复一条消息(可选),立即结束当前事件处理函数,等待接收一个新的消息后再次执行当前事件处理函数。

reject_arg 用于拒绝指定 got 接收的参数,通常在嵌套装饰器时使用。

reject_arg 操作接受一个 key 参数以及可选的 prompt 参数。prompt 参数与 send 相同。

@matcher.got("a")
@matcher.got("b")
async def _(a: str = ArgPlainText(), b: str = ArgPlainText()):
if a not in b:
await matcher.reject_arg("a", "Invalid a!")

reject_receive

向用户回复一条消息(可选),立即结束当前事件处理函数,等待接收一个新的事件后再次执行当前事件处理函数。

reject_receive 用于拒绝指定 receive 接收的事件,通常在嵌套装饰器时使用。

reject_receive 操作接受一个可选的 id 参数以及可选的 prompt 参数。id 参数默认为空 "",prompt 参数与 send 相同。

@matcher.receive("a")
@matcher.receive("b")
async def _(a: Event = Received("a"), b: Event = Received("b")):
if a.get_user_id() != b.get_user_id():
await matcher.reject_receive("a")

skip

立即结束当前事件处理函数,进入下一个事件处理函数。

通常在依赖注入中使用,用于跳过当前事件处理函数的执行。

from nonebot.params import Depends

async def dependency():
matcher.skip()

@matcher.handle()
async def _(check=Depends(dependency)):
# 这个函数不会被执行

stop_propagation

阻止事件向更低优先级的事件响应器传播。

from nonebot.matcher import Matcher

@foo.handle()
async def _(matcher: Matcher):
matcher.stop_propagation()
注意

stop_propagation 操作是实例方法,需要先通过依赖注入获取事件响应器实例再进行调用。

get_arg

获取一个 got 接收的参数。

get_arg 操作接受一个 key 参数和一个可选的 default 参数。当参数不存在时,将返回 default 或 None

from nonebot.matcher import Matcher

@matcher.handle()
async def _(matcher: Matcher):
key = matcher.get_arg("key", default=None)

set_arg

设置 / 覆盖一个 got 接收的参数。

set_arg 操作接受一个 key 参数和一个 value 参数。请注意,value 参数必须是消息序列对象,如需存储其他数据请使用会话状态

from nonebot.matcher import Matcher

@matcher.handle()
async def _(matcher: Matcher):
matcher.set_arg("key", Message("value"))

get_receive

获取一个 receive 接收的事件。

get_receive 操作接受一个 id 参数和一个可选的 default 参数。当事件不存在时,将返回 default 或 None

from nonebot.matcher import Matcher

@matcher.handle()
async def _(matcher: Matcher):
event = matcher.get_receive("id", default=None)

get_last_receive

获取最近的一个 receive 接收的事件。

get_last_receive 操作接受一个可选的 default 参数。当事件不存在时,将返回 default 或 None

from nonebot.matcher import Matcher

@matcher.handle()
async def _(matcher: Matcher):
event = matcher.get_last_receive(default=None)

set_receive

设置 / 覆盖一个 receive 接收的事件。

set_receive 操作接受一个 id 参数和一个 event 参数。请注意,event 参数必须是事件对象,如需存储其他数据请使用会话状态

from nonebot.matcher import Matcher

@matcher.handle()
async def _(matcher: Matcher):
matcher.set_receive("key", Event())