パイプラインイベントと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"""
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で処理する必要があるメッセージ(コマンドでないメッセージ)であると判定されたときにトリガーされます。
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を使用してください。
グループチャットまたはプライベートチャットでコマンドを受信したときにトリガーされます。
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
# ========== 設定可能な属性 ==========
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()このメソッドを呼び出すと、このイベントのデフォルトの後続動作が防止され、パイプラインは直接終了します。
INFO
次のイベントのみがデフォルト動作を防止できます:
- PersonMessageReceived
- GroupMessageReceived
- PersonNormalMessageReceived
- GroupNormalMessageReceived
- PersonCommandSent
- GroupCommandSent
- NormalMessageResponded
後続実行を防止
def prevent_postorder(self):
"""後続実行を防止する"""
# 使用例
event_context.prevent_postorder()このメソッドを呼び出すと、後続のプラグインがこの時点で実行されなくなります。
