iamai package#

Subpackages#

Submodules#

Module contents#

API Documentation

Comprehensive AI Toolkit for Multimodal Learning and Cross-Platform Robotics.

This Module imports the following contents from the sub-module.

class iamai.Adapter(bot: Bot)[source]#

Bases: Generic[EventT, ConfigT], ABC

协议适配器基类。

name#

适配器的名称。

Type:: str

bot#

当前的机器人对象。

Type:: Bot

Config: Type[ConfigT]#

bot: Bot#

property config: ConfigT#: 适配器配置。

获取满足指定条件的的事件，协程会等待直到适配器接收到满足条件的事件、超过最大事件数或超时。

类似 Bot 类的 get() 方法，但是隐含了判断产生事件的适配器是本适配器。等效于 Bot 类的 get() 方法传入 adapter_type 为本适配器类型。

Parameters:

func – 协程或者函数，函数会被自动包装为协程执行。要求接受一个事件作为参数，返回布尔值。当协程返回 True 时返回当前事件。当为 None 时相当于输入对于任何事件均返回真的协程，即返回适配器接收到的下一个事件。
event_type – 当指定时，只接受指定类型的事件，先于 func 条件生效。默认为 None。
max_try_times – 最大事件数。
timeout – 超时时间。

Returns:

返回满足 func 条件的事件。

Raises:

GetEventTimeout – 超过最大事件数或超时。

name: str#

abstract async run() → None[source]#

适配器运行方法，适配器开发者必须实现该方法。

适配器运行过程中保持保持运行，当此方法结束后， AliceBot 不会自动重新启动适配器。

final async safe_run() → None[source]#: 附带有异常处理地安全运行适配器。

async shutdown() → None[source]#

在适配器结束运行时运行的方法，用于安全地关闭适配器。

AliceBot 在接收到系统的结束信号后依次运行并等待所有适配器的 shutdown() 方法。当强制退出时此方法可能未被执行。

async startup() → None[source]#

在适配器开始运行前运行的方法，用于初始化适配器。

AliceBot 依次运行并等待所有适配器的 startup() 方法，待运行完毕后再创建 run() 任务。

class iamai.Bot(*, config_file: str | None = 'config.toml', config_dict: Dict[str, Any] | None = None, hot_reload: bool = False)[source]#

Bases: object

iamai 机器人对象，定义了机器人的基本方法。

读取并储存配置 Config，加载适配器 Adapter 和插件 Plugin，并进行事件分发。

config#

机器人配置。

Type:: iamai.config.MainConfig

should_exit#

机器人是否应该进入准备退出状态。

Type:: asyncio.locks.Event

adapters#

当前已经加载的适配器的列表。

Type:: List[iamai.adapter.Adapter[Any, Any]]

plugins_priority_dict#

插件优先级字典。

Type:: Dict[int, List[Type[iamai.plugin.Plugin[Any, Any, Any]]]]

plugin_state#

插件状态。

Type:: Dict[str, Any]

global_state#

全局状态。

Type:: Dict[Any, Any]

adapter_run_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器运行时的函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

adapter_shutdown_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器关闭时的函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

adapter_startup_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][source]#

注册一个适配器初始化时的函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

adapters: List[Adapter[Any, Any]]#

bot_exit_hook(func: Callable[[Bot], Awaitable[None]]) → Callable[[Bot], Awaitable[None]][source]#

注册一个 Bot 退出时的函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

bot_run_hook(func: Callable[[Bot], Awaitable[None]]) → Callable[[Bot], Awaitable[None]][source]#

注册一个 Bot 启动时的函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

config: MainConfig#

error_or_exception(message: str, exception: Exception) → None[source]#

根据当前 Bot 的配置输出 error 或者 exception 日志。

Parameters:

message – 消息。
exception – 异常。

event_postprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) → Callable[[Event[Any]], Awaitable[None]][source]#

注册一个事件后处理函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

event_preprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) → Callable[[Event[Any]], Awaitable[None]][source]#

注册一个事件预处理函数。

Parameters:: func – 被注册的函数。
Returns:: 被注册的函数。

获取满足指定条件的的事件，协程会等待直到适配器接收到满足条件的事件、超过最大事件数或超时。

Parameters:

func – 协程或者函数，函数会被自动包装为协程执行。要求接受一个事件作为参数，返回布尔值。当协程返回 True 时返回当前事件。当为 None 时相当于输入对于任何事件均返回真的协程，即返回适配器接收到的下一个事件。
event_type – 当指定时，只接受指定类型的事件，先于 func 条件生效。默认为 None。
adapter_type – 当指定时，只接受指定适配器产生的事件，先于 func 条件生效。默认为 None。
max_try_times – 最大事件数。
timeout – 超时时间。

Returns:

返回满足 func 条件的事件。

Raises:

GetEventTimeout – 超过最大事件数或超时。

get_adapter(adapter: str) → Adapter[Any, Any][source]#

get_adapter(adapter: Type[AdapterT]) → AdapterT

按照名称或适配器类获取已经加载的适配器。

Parameters:: adapter – 适配器名称或适配器类。
Returns:: 获取到的适配器对象。
Raises:: LookupError – 找不到此名称的适配器对象。

get_plugin(name: str) → Type[Plugin[Any, Any, Any]][source]#

按照名称获取已经加载的插件类。

Parameters:: name – 插件名称
Returns:: 获取到的插件类。
Raises:: LookupError – 找不到此名称的插件类。

global_state: Dict[Any, Any]#

async handle_event(current_event: Event[Any], *, handle_get: bool = True, show_log: bool = True) → None[source]#

被适配器对象调用，根据优先级分发事件给所有插件，并处理插件的 stop 、 skip 等信号。

此方法不应该被用户手动调用。

Parameters:

current_event – 当前待处理的 Event。
handle_get – 当前事件是否可以被 get 方法捕获，默认为 True。
show_log – 是否在日志中显示，默认为 True。

load_adapters(*adapters: Type[Adapter[Any, Any]] | str) → None[source]#

加载适配器。

Parameters:

*adapters –

适配器类或适配器名称，类型可以是 Type[Adapter] 或 str。如果为 Type[Adapter] 类型时，将作为适配器类进行加载。如果为 str 类型时，将作为适配器模块名称进行加载，格式和 Python import 语句相同。

例如：path.of.adapter。

load_plugins(*plugins: Type[Plugin[Any, Any, Any]] | str | Path) → None[source]#

加载插件。

Parameters:

*plugins –

插件类、插件模块名称或者插件模块文件路径。类型可以是 Type[Plugin], str 或 pathlib.Path。如果为 Type[Plugin] 类型时，将作为插件类进行加载。如果为 str 类型时，将作为插件模块名称进行加载，格式和 Python import 语句相同。

例如：path.of.plugin。

如果为 pathlib.Path 类型时，将作为插件模块文件路径进行加载。: 例如：pathlib.Path(“path/of/plugin”)。

load_plugins_from_dirs(*dirs: Path) → None[source]#

从目录中加载插件，以 _ 开头的模块中的插件不会被导入。路径可以是相对路径或绝对路径。

Parameters:: *dirs – 储存包含插件的模块的模块路径。例如：pathlib.Path(“path/of/plugins/”) 。

plugin_state: Dict[str, Any]#

property plugins: List[Type[Plugin[Any, Any, Any]]]#: 当前已经加载的插件的列表。

plugins_priority_dict: Dict[int, List[Type[Plugin[Any, Any, Any]]]]#

reload_plugins() → None[source]#: 手动重新加载所有插件。

restart() → None[source]#: 退出并重新运行 iamai。

run() → None[source]#: 运行 iamai，监听并拦截系统退出信号，更新机器人配置。

should_exit: Event#

class iamai.ConfigModel(**extra_data: Any)[source]#

Bases: BaseModel

iamai 配置模型。

__config_name__#

配置名称。

Type:: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

子依赖装饰器。

Parameters:

dependency – 依赖类。如果不指定则根据字段的类型注释自动判断。
use_cache – 是否使用缓存。默认为 True。

Returns:

返回内部子依赖对象。

class iamai.Event(*, adapter: Any, type: str | None, **extra_data: Any)[source]#

Bases: ABC, BaseModel, Generic[AdapterT]

事件类的基类。

adapter#

产生当前事件的适配器对象。

Type:: Any

type#

事件类型。

Type:: str | None

__handled__#

表示事件是否被处理过了，用于适配器处理。警告：请勿手动更改此属性的值。

Type:: bool

adapter: Any#

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'adapter': FieldInfo(annotation=Any, required=True), 'type': FieldInfo(annotation=Union[str, NoneType], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str | None#

class iamai.MessageEvent(*, adapter: Any, type: str | None, **extra_data: Any)[source]#

Bases: Event, Generic[AdapterT]

通用的消息事件类的基类。

async ask(message: str, max_try_times: int | None = None, timeout: int | float | None = None) → Self[source]#

询问消息。

表示回复一个消息后获取用户的回复。相当于 reply() 后执行 get()。

Parameters:

message – 回复消息的内容。
max_try_times – 最大事件数。
timeout – 超时时间。

Returns:

用户回复的消息事件。

async get(*, max_try_times: int | None = None, timeout: int | float | None = None) → Self[source]#

获取用户回复消息。

相当于 Bot 的 get()，条件为适配器、事件类型、发送人相同。

Parameters:

max_try_times – 最大事件数。
timeout – 超时时间。

Returns:

用户回复的消息事件。

Raises:

GetEventTimeout – 超过最大事件数或超时。

abstract get_plain_text() → str[source]#

获取消息的纯文本内容。

Returns:: 消息的纯文本内容。

abstract async is_same_sender(other: Self) → bool[source]#

判断自身和另一个事件是否是同一个发送者。

Parameters:: other – 另一个事件。
Returns:: 是否是同一个发送者。

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'adapter': FieldInfo(annotation=Any, required=True), 'type': FieldInfo(annotation=Union[str, NoneType], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

abstract async reply(message: str) → Any[source]#

回复消息。

Parameters:: message – 回复消息的内容。
Returns:: 回复消息动作的响应。

class iamai.Plugin[source]#

Bases: ABC, Generic[EventT, StateT, ConfigT]

所有 iamai 插件的基类。

event#: 当前正在被此插件处理的事件。

priority#

插件的优先级，数字越小表示优先级越高，默认为 0。

Type:: ClassVar[int]

block#

插件执行结束后是否阻止事件的传播。True 表示阻止。

Type:: ClassVar[bool]

__plugin_load_type__#

插件加载类型，由 iamai 自动设置，反映了此插件是如何被加载的。

Type:: ClassVar[iamai.plugin.PluginLoadType]

__plugin_file_path__#

当插件加载类型为 PluginLoadType.CLASS 时为 None，否则为定义插件在的 Python 模块的位置。

Type:: ClassVar[str | None]

Config: Type[ConfigT]#

block: ClassVar[bool] = False#

property bot: Bot#: 机器人对象。

property config: ConfigT#: 插件配置。

event: EventT = InnerDepends(Event)#

abstract async handle() → None[source]#: 处理事件的方法。当 rule() 方法返回 True 时 iamai 会调用此方法。每个插件必须实现此方法。

property name: str#: 插件类名称。

priority: ClassVar[int] = 0#

abstract async rule() → bool[source]#

匹配事件的方法。事件处理时，会按照插件的优先级依次调用此方法，当此方法返回 True 时将事件交由此插件处理。每个插件必须实现此方法。

注意：不建议直接在此方法内实现对事件的处理，事件的具体处理请交由 handle() 方法。

final skip() → NoReturn[source]#: 跳过自身继续当前事件传播。

property state: StateT#: 插件状态。

final stop() → NoReturn[source]#: 停止当前事件传播。