iamai package¶

Subpackages¶

Submodules¶

Module contents¶

API Documentation

Comprehensive AI Toolkit for Multi Modal 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 Bot class object, defines the basic functionality for interacting.

Read and save configuration Config, load adapters Adapter and plugins Plugin, then distribute the events.

config¶

Bot configuration.

Type:: iamai.config.MainConfig

should_exit¶

Whether the bot should enter the ready-to-exit state.

Type:: asyncio.locks.Event

adapters¶

List of currently loaded adapters.

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

plugins_priority_dict¶

Plugin priority dictionary.

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

plugin_state¶

Plugin status.

Type:: Dict[str, Any]

global_state¶

Global status.

Type:: Dict[Any, Any]

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

Parameters:: func – the registered function.
Returns:: The registered function.

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

Parameters:: func – the registered function.
Returns:: The registered function.

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

Parameters:: func – the registered function.
Returns:: The registered function.

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

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

Parameters:: func – the registered function.
Returns:: The registered function.

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

Parameters:: func – the registered function.
Returns:: The registered function.

config: MainConfig¶

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

Output error or exception logs based on the current Bot configuration.

Parameters:

message – message.
exception – Exception.

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

Parameters:: func – the registered function.
Returns:: The registered function.

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

Parameters:: func – the registered function.
Returns:: The registered function.

Get events that meet the specified conditions. The coroutine will wait until the adapter receives events that meet the conditions, exceeds the maximum number of events, or times out.

Parameters:

func – Coroutine or function, the function will be automatically packaged as a coroutine for execution. Requires an event to be accepted as a parameter and returns a Boolean value. Returns the current event when the coroutine returns True. When None is equivalent to the input coroutine returning true for any event, that is, returning the next event received by the adapter.
event_type – When specified, only events of the specified type are accepted, taking effect before the func condition. Defaults to None.
adapter_type – When specified, only events generated by the specified adapter will be accepted, taking effect before the func condition. Defaults to None.
max_try_times – Maximum number of events.
timeout – timeout period.

Returns:

Returns events that satisfy the condition of func.

Raises:

GetEventTimeout – Maximum number of events exceeded or timeout.

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

get_adapter(adapter: Type[AdapterT]) → AdapterT

Get the loaded adapter by name or adapter class.

Parameters:: adapter – adapter name or adapter class.
Returns:: The obtained adapter object.
Raises:: LookupError – No adapter object with this name found.

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

Get the loaded plugin class by name.

Parameters:: name – plugin name
Returns:: The obtained plug-in class.
Raises:: LookupError – The plugin class with this name cannot be found.

global_state: Dict[Any, Any]¶

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

Called by the adapter object, distributes events to all plug-ins according to priority, and handles plug-in signals such as stop and skip.

This method should not be called manually by the user.

Parameters:

current_event – The currently pending Event.
handle_get – Whether the current event can be captured by the get method, the default is True.
show_log – Whether to display in the log, the default is True.

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

Load adapter.

Parameters:

*adapters –

Adapter class or adapter name, type can be Type[Adapter] or str. If it is of type Type[Adapter], it will be loaded as an adapter class. If it is of type str, it will be loaded as the adapter module name, and the format is the same as the Python import statement.

For example: path.of.adapter.

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

Load the plugin.

Parameters:

*plugins –

Plugin class, plugin module name or plug-in module file path. Type can be Type[Plugin], str or pathlib.Path. If it is Type[Plugin], it will be loaded as a plug-in class. If it is of type str, it will be loaded as the plug-in module name, and the format is the same as the Python import statement.

For example: path.of.plugin.

If it is of type pathlib.Path, it will be loaded as the plug-in module file path.: For example: pathlib.Path("path/of/plugin").

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

Load plug-ins from the directory. Plug-ins in modules starting with _ will not be imported. The path can be a relative path or an absolute path.

Parameters:: *dirs – Module paths that store modules containing plugins. For example: pathlib.Path("path/of/plugins/") .

property locale: List[str]¶: Get the current language of the bot.

property locale_dirs: List[Path]¶: Get the locale directories of the bot.

plugin_state: Dict[str, Any]¶

property plugins: List[Type[Plugin[Any, Any, Any]]]¶: List of currently loaded plugins.

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

reload_plugins() → None[source]¶: Manually reload all plugins.

restart() → None[source]¶: Exit and rerun iamai.

run() → None[source]¶: Run iamai, monitor and intercept system exit signals, and update robot configuration.

should_exit: Event¶

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

Bases: BaseModel

iamai configuration model.

__config_name__¶

Configuration name.

Type:: str

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

Subdependency decorator.

Parameters:

dependency – dependency class. If not specified, it will be automatically determined based on the type annotation of the field.
use_cache – whether to use cache. Defaults to True.

Returns:

Returns the internal child dependency object.

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

Bases: ABC, BaseModel, Generic[AdapterT]

The base class of event classes.

adapter¶

The adapter object that generated the current event.

Type:: Any

type¶

event type.

Type:: str | None

__handled__¶

Indicates whether the event has been handled, used for adapter processing. Warning: Do not change the value of this property manually.

Type:: bool

adapter: Any¶

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

type: str | None¶

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

Bases: Event, Generic[AdapterT]

Base class for general message event classes.

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

Ask for news.

Indicates getting the user’s reply after replying to a message. Equivalent to executing get() after reply().

Parameters:

message – The content of the reply message.
max_try_times – Maximum number of events.
timeout – timeout period.

Returns:

Message event that the user replies to.

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

Get the user’s reply message.

Equivalent to get() of Bot, the condition is that the adapter, event type and sender are the same.

Parameters:

max_try_times – Maximum number of events.
timeout – timeout period.

Returns:

Message event that the user replies to.

Raises:

GetEventTimeout – Maximum number of events exceeded or timeout.

abstract get_plain_text() → str[source]¶

Get the plain text content of the message.

Returns:: The plain text content of the message.

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

Determine whether itself and another event are the same sender.

Parameters:: other – another event.
Returns:: Is it the same sender?

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

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

Reply message.

Parameters:: message – The content of the reply message.
Returns:: The response to the reply message action.

class iamai.Plugin[source]¶

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

Base class for all iamai plugins.

event¶: The event currently being processed by this plugin.

priority¶

The priority of the plugin. The smaller the number, the higher the priority. The default is 0.

Type:: ClassVar[int]

block¶

Whether to prevent the propagation of events after the plug-in is executed. True means blocking.

Type:: ClassVar[bool]

__plugin_load_type__¶

Plugin load type, automatically set by iamai, reflects how this plugin is loaded.

Type:: ClassVar[iamai.plugin.PluginLoadType]

__plugin_file_path__¶

None when the plugin loading type is PluginLoadType.CLASS, Otherwise, the location of the Python module in which the plugin is defined.

Type:: ClassVar[str | None]

Config: Type[ConfigT]¶

block: ClassVar[bool] = False¶

property bot: Bot¶: bot object.

property config: ConfigT¶: plugin configuration.

event: EventT = InnerDepends(Event)¶

abstract async handle() → None[source]¶: Method to handle events. iamai will call this method when the rule() method returns True. Each plugin must implement this method.

property name: str¶: plugin class name.

priority: ClassVar[int] = 0¶

abstract async rule() → bool[source]¶: Method to match the event. When the event is processed, this method will be called in sequence according to the priority of the plugin. When this method returns True, the event will be handed over to this plugin for processing. Each plugin must implement this method.

Note

It is not recommended to implement event processing directly in this method. Please leave the specific processing of events to the handle() method.

final skip() → NoReturn[source]¶: Skips itself and continues propagation of the current event.

property state: StateT¶: plugin status.

final stop() → NoReturn[source]¶: Stop propagation of current events.