パイプラインイベントとAPI

LangBotプラグインはパイプラインイベントを登録して処理できます。使用方法については、コンポーネント: イベントリスナーを参照してください。

イベント一覧

一部のイベントには設定可能な属性があり、プラグインコードで変更して、その後のLangBot処理で使用できます。

*MessageReceived

グループチャットまたはプライベートチャットでメッセージを受信したときにトリガーされます。

class PersonMessageReceived(BaseEventModel):
    """プライベートチャットメッセージを受信したとき"""

    event_name: str = "PersonMessageReceived"

    launcher_type: str
    """起動元オブジェクトタイプ(person)"""

    launcher_id: typing.Union[int, str]
    """送信者ID"""

    sender_id: typing.Union[int, str]
    """送信者ID、プライベートチャットの場合はlauncher_idと同じ"""

    message_event: platform_events.PersonMessage
    """元のメッセージイベントオブジェクト。送信者情報を含みます。"""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """メッセージチェーン"""


class GroupMessageReceived(BaseEventModel):
    """グループチャットメッセージを受信したとき"""

    event_name: str = "GroupMessageReceived"

    launcher_type: str
    """起動元オブジェクトタイプ(group)"""

    launcher_id: typing.Union[int, str]
    """グループID"""

    sender_id: typing.Union[int, str]
    """送信者ID"""

    message_event: platform_events.GroupMessage
    """元のメッセージイベントオブジェクト。グループと送信者情報を含みます。"""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """メッセージチェーン"""

*NormalMessageReceived

グループチャットまたはプライベートチャットメッセージを受信し、LLMで処理する必要があるメッセージ(コマンドでないメッセージ)であると判定されたときにトリガーされます。

class PersonNormalMessageReceived(BaseEventModel):
    """処理すべきプライベートチャット通常メッセージが判定されたときにトリガーされる"""

    event_name: str = "PersonNormalMessageReceived"

    launcher_type: str
    """起動元オブジェクトタイプ(person)"""

    launcher_id: typing.Union[int, str]
    """起動元オブジェクトID"""

    sender_id: typing.Union[int, str]
    """送信者ID、プライベートチャットの場合はlauncher_idと同じ"""

    text_message: str
    """メッセージテキスト"""

    message_event: platform_events.PersonMessage
    """元のメッセージイベントオブジェクト。送信者情報を含みます。"""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """メッセージチェーン"""

    # ========== 設定可能な属性 ==========

    user_message_alter: typing.Optional[provider_message.ContentElement] = None
    """変更されたメッセージテキスト、langbot_plugin.api.entities.builtin.provider.message.ContentElement型"""

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """直接返信メッセージチェーン、デフォルト動作を防止した場合のみ有効"""


class GroupNormalMessageReceived(BaseEventModel):
    """処理すべきグループチャット通常メッセージが判定されたときにトリガーされる"""

    event_name: str = "GroupNormalMessageReceived"

    launcher_type: str
    """起動元オブジェクトタイプ(group)"""

    launcher_id: typing.Union[int, str]
    """グループID"""

    sender_id: typing.Union[int, str]
    """送信者ID"""

    text_message: str
    """メッセージテキスト"""

    message_event: platform_events.GroupMessage
    """元のメッセージイベントオブジェクト。グループと送信者情報を含みます。"""

    message_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """メッセージチェーン"""

    # ========== 設定可能な属性 ==========

    user_message_alter: typing.Optional[provider_message.ContentElement] = None
    """変更されたメッセージテキスト、langbot_plugin.api.entities.builtin.provider.message.ContentElement型"""

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """直接返信メッセージチェーン、デフォルト動作を防止した場合のみ有効"""

*CommandSent

使用は推奨されなくなりました。代わりにコンポーネント: Commandを使用してください。

グループチャットまたはプライベートチャットでコマンドを受信したときにトリガーされます。

class PersonCommandSent(BaseEventModel):
    """処理すべきプライベートチャットコマンドが判定されたときにトリガーされる"""

    event_name: str = "PersonCommandSent"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    command: str
    """コマンドテキスト"""

    params: list[str]
    """コマンドパラメータ"""

    text_message: str
    """メッセージテキスト"""

    is_admin: bool
    """管理者かどうか"""


class GroupCommandSent(BaseEventModel):
    """処理すべきグループチャットコマンドが判定されたときにトリガーされる"""

    event_name: str = "GroupCommandSent"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    command: str
    """コマンドテキスト"""

    params: list[str]
    """コマンドパラメータ"""

    text_message: str
    """メッセージテキスト"""

    is_admin: bool
    """管理者かどうか"""

NormalMessageResponded

メッセージがLLMレスポンスを受信したときにトリガーされます。

class NormalMessageResponded(BaseEventModel):
    """通常メッセージに返信するときにトリガーされる"""

    event_name: str = "NormalMessageResponded"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

    session: provider_session.Session
    """セッションオブジェクト"""

    prefix: str
    """返信メッセージプレフィックス"""

    response_text: str
    """返信メッセージテキスト"""

    finish_reason: str
    """レスポンス終了理由"""

    funcs_called: list[str]
    """呼び出された関数のリスト"""

    # ========== 設定可能な属性 ==========

    reply_message_chain: typing.Optional[platform_message.MessageChain] = None
    """返信メッセージコンポーネントリスト、デフォルト動作を防止した場合のみ有効"""

PromptPreProcessing

LLMレスポンスコンテキスト(プロンプト)を構築するときにトリガーされます。

class PromptPreProcessing(BaseEventModel):
    """セッション内のプロンプトを前処理するときにトリガーされる"""

    event_name: str = "PromptPreProcessing"

    session_name: str
    """セッション名、形式は person_1234567890 または group_1234567890"""

    # ========== 設定可能な属性 ==========

    default_prompt: list[typing.Union[provider_message.Message, provider_message.MessageChunk]]
    """この会話のシナリオプリセット(システムプロンプト)、変更可能、langbot_plugin.api.entities.builtin.provider.message.Messageまたはlangbot_plugin.api.entities.builtin.provider.message.MessageChunk型"""

    prompt: list[typing.Union[provider_message.Message, provider_message.MessageChunk]]
    """この会話の既存メッセージレコード、変更可能、langbot_plugin.api.entities.builtin.provider.message.Messageまたはlangbot_plugin.api.entities.builtin.provider.message.MessageChunk型"""

イベントコンテキストAPI

...
        @self.handler(events.PersonMessageReceived)
        async def handler(event_context: context.EventContext):
            ...

イベント処理メソッドにはEventContextオブジェクトが渡され、イベントコンテキスト情報が含まれます。このオブジェクトにはリクエストAPIとイベントコンテキスト固有のAPIの両方があります。以下は、イベントコンテキスト固有のAPIのリストです:

元のイベント属性の取得

event_context.event属性は元のイベントオブジェクトであり、元のイベントの属性を取得できます。型はリッスンされているイベントの型です。例えば、リッスンされているイベントがGroupMessageReceivedの場合:

event = event_context.event
print(event.launcher_type)  # 起動元オブジェクトタイプ(group)
print(event.launcher_id)    # グループID
print(event.sender_id)      # 送信者アカウントID
print(event.message_chain)  # メッセージチェーン

デフォルト動作を防止

def prevent_default(self):
    """デフォルト動作を防止する"""

# 使用例
event_context.prevent_default()

このメソッドを呼び出すと、このイベントのデフォルトの後続動作が防止され、パイプラインは直接終了します。

以下のイベントのみデフォルト動作を阻止できます：

PersonMessageReceived
GroupMessageReceived
PersonNormalMessageReceived
GroupNormalMessageReceived
PersonCommandSent
GroupCommandSent
NormalMessageResponded

後続実行を防止

def prevent_postorder(self):
    """後続実行を防止する"""

# 使用例
event_context.prevent_postorder()

このメソッドを呼び出すと、後続のプラグインがこの時点で実行されなくなります。

​イベント一覧

​*MessageReceived

​*NormalMessageReceived

​*CommandSent

​NormalMessageResponded

​PromptPreProcessing

​イベントコンテキストAPI

​元のイベント属性の取得

​デフォルト動作を防止

​後続実行を防止

イベント一覧

*MessageReceived

*NormalMessageReceived

*CommandSent

NormalMessageResponded

PromptPreProcessing

イベントコンテキストAPI

元のイベント属性の取得

デフォルト動作を防止

後続実行を防止