Skip to content

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

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

イベント一覧

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

*MessageReceived

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

python
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"""

    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_chain: platform_message.MessageChain = pydantic.Field(
        serialization_alias="message_chain"
    )
    """生のメッセージチェーン"""

*NormalMessageReceived

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

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

    event_name: str = "PersonNormalMessageReceived"

    launcher_type: str

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

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

    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

    launcher_id: typing.Union[int, str]

    sender_id: typing.Union[int, str]

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

    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を使用してください。

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

python
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レスポンスを受信したときにトリガーされます。

python
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レスポンスコンテキスト(プロンプト)を構築するときにトリガーされます。

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

    event_name: str = "PromptPreProcessing"

    session_name: str

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

    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

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

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

元のイベント属性の取得

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

例えば、リッスンされているイベントがGroupMessageReceivedの場合:

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

デフォルト動作を防止

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

# 使用例
event_context.prevent_default()

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

INFO

次のイベントのみがデフォルト動作を防止できます:

  • PersonMessageReceived
  • GroupMessageReceived
  • PersonNormalMessageReceived
  • GroupNormalMessageReceived
  • PersonCommandSent
  • GroupCommandSent
  • NormalMessageResponded

後続実行を防止

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

# 使用例
event_context.prevent_postorder()

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