简体中文 English 繁體中文 日本語 Русский
本文为静态镜像，内容以交互版为准 在交互式文档中心打开 →

name: erispulse-module-dev description: "适用于：要开发 ErisPulse 模块（业务逻辑载体）：命令、事件、多轮对话、ORM。覆盖场景：このドキュメントに沿って進めると、/hello コマンドに応答できる実行可能なモジュールが完成し、モジュールのプロジェクト構造、登録方法、ロード/アンロードフックについても理解できます。今後のすべてのモジュールドキュメントは、この骨組みの上に成り立っています。この文書では、モジュールのロードからアンロードまでの完全なライフサイクル、宣言的設定とストレージの使い方を解説します——フレームワークが自動で行うこと、あなたが書くべきことを、読み終える頃には把握できる内容です。プロセッサの event パラメータが直接 .get_text() / .reply() / .wait_reply() を呼べるのは、Event 包装クラスのおかげです。本記事では、Event 包装クラスの全メソッド一覧を紹介します。イベントハンドラを書く際には、このリストを参照してください。メソッドの署名と動作は、ここを基準とします。最初のモジュールを書き終えた後に読むことを推奨（エンジニアリングのベストプラクティス）。必要なければスキップしても、他の機能の利用には影響しません。この記事を読むタイミング：データモデル、自動テーブル作成、関連マッピングが必要な場合。不要であれば飛ばしても、他の機能の使用に影響はありません。このドキュメントを読む必要があるのはいつですか：多輪対話のステートマシンとチェックポイントの復元が必要な場合です。必要ない場合はスキップしても、他の機能の使用には影響しません。この記事を読むのはいつか：wait_reply、多輪対話、定時リマインダー、セッション排他、およびトレースIDが必要な場合。不要な場合は飛ばしても、他の機能の使用には影響しません。"

ErisPulse 模块开发

This skill is auto-built from the ErisPulse documentation. Repository: https://github.com/ErisPulse/ErisPulse Source paths are annotated per section; see docs/ in the repository for the latest content.

内容大纲

AI 按需读取：只读需要的章节即可，不必全文加载。

模块开发入门

来源: docs/ja/developer-guide/modules/getting-started.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/developer-guide/modules/getting-started.md）

モジュール開発入門

Note

L1 コア必読――トップページの「学習パス」にある5つのコアドキュメントの1つで、モジュール開発に必須です。

このドキュメントに沿って進めると、/hello コマンドに応答できる実行可能なモジュールが完成し、モジュールのプロジェクト構造、登録方法、ロード/アンロードフックについても理解できます。今後のすべてのモジュールドキュメントは、この骨組みの上に成り立っています。

プロジェクト構造

標準的なモジュールの構造：

MyModule/
├── pyproject.toml
├── README.md
├── LICENSE
└── MyModule/
    ├── __init__.py
    └── Core.py

pyproject.toml 設定

[project]
name = "ErisPulse-MyModule"
version = "1.0.0"
description = "モジュール機能の説明"
readme = "README.md"
requires-python = ">=3.10"
license = { file = "LICENSE" }
authors = [ { name = "yourname", email = "[email protected]" } ]
dependencies = []

[project.urls]
"homepage" = "https://github.com/yourname/MyModule"

[project.entry-points."erispulse.module"]
"MyModule" = "MyModule:Main"

__init__.py

from .Core import Main

Core.py - 基本モジュール

from ErisPulse import sdk
# ルートからのインポートを推奨（2.10+）：よく使うシンボルはルートパッケージから直接インポートでき、深いパスも引き続き互換
from ErisPulse import BaseModule, command

class Main(BaseModule):
    def __init__(self, sdk):
        self.sdk = sdk
        self.logger = sdk.logger.get_child("MyModule")
        self.storage = sdk.storage
    
    @staticmethod
    def get_load_strategy():
        """モジュールのロード戦略を返します"""
        from ErisPulse.loaders import ModuleLoadStrategy
        return ModuleLoadStrategy(
            lazy_load=True,
            priority=0,
            depends=[],  # 任意：依存する他のモジュールのリスト
            # 任意：イベント駆動の遅延アクティベーション――トリガーを宣言すると、最初に一致するイベント/コマンドが到着した時点で自動的にロードされます
            # activate_on=[{"command": {"name": "hello", "help": "挨拶を送る"}}],
        )
    
    async def on_load(self, event):
        """モジュールのロード時に呼び出されます"""
        @command("hello", help="挨拶を送る")
        async def hello_command(event):
            name = event.get_user_nickname() or "フレンド"
            await event.reply(f"こんにちは、{name}さん！")
        
        self.logger.info("モジュールがロードされました")
    
    async def on_unload(self, event):
        """モジュールのアンロード時に呼び出されます"""
        self.logger.info("モジュールがアンロードされました")

設定の読み取り：上記の基本例では設定を使用していません。設定を読み取る必要がある場合は、ネストした ConfigClass を宣言し、self.cfg 経由でリアルタイムに読み取る方法が推奨されます（詳細は モジュールのコア概念 を参照）。_load_config() を手動で呼び出す旧式の書き方は非推奨となりました。

モジュールのテスト

ローカルテスト

# プロジェクトディレクトリでモジュールをインストール
epsdk install ./MyModule

# プロジェクトを実行
epsdk run main.py --reload

コマンドのテスト

コマンドを送信してテストします：

/hello

コア概念

BaseModule 基底クラス

すべてのモジュールは BaseModule を継承する必要があり、以下のメソッドを提供します：

メソッド 説明 必須
__init__(self, sdk) コンストラクター（フレームワークが sdk インスタンスを渡します） いいえ
get_load_strategy() ロード戦略を返します いいえ
get_meta() モジュール紹介のメタ情報を返します（任意） いいえ
on_load(self, event) モジュールのロード時に呼び出されます はい
on_unload(self, event) モジュールのアンロード時に呼び出されます はい

モジュール紹介 meta

Note

この機能には ErisPulse 2.8.0+ が必要です。

get_meta() を通じて、モジュールの紹介メタ情報（このモジュールが何をするのか、どのカテゴリに属するのかなど）を宣言します。 メタ情報はモジュールの汎用紹介データであり、help モジュール、Dashboard のモジュールリスト、モジュールストアなど、さまざまなインターフェース/エコシステムモジュールから利用されます。

get_load_strategy() が ModuleLoadStrategy を返すのと同様に、**ModuleMeta 設定クラスのインスタンスを返すことが推奨されます**（属性の型付け、IDE 補完）。dict を直接返すことにも対応しています：

class MyModule(BaseModule):
    @staticmethod
    def get_meta() -> ModuleMeta:
        return ModuleMeta(
            name="天気",               # 表示名（デフォルトの登録名）
            description="都市の天気を照会",  # モジュールの概要
            version="1.0.0",
            author="ErisDev",
            group="ツール",               # 機能グループ
            tags=["天気", "照会"],
        )

互換的な書き方（dict）：

class MyModule(BaseModule):
    @staticmethod
    def get_meta() -> dict:
        return {
            "name": "天気",
            "description": "都市の天気を照会",
            "version": "1.0.0",
            "author": "ErisDev",
            "group": "ツール",
            "tags": ["天気", "照会"],
        }

meta フィールドの i18n サポート

メタ情報のフィールド値には、プレーンな文字列、または i18n 辞書 {"i18n": "key.path", "default": "フォールバックテキスト"}（設定の description と同じ規約）を使用できます。 翻訳キーは I18nClass で宣言・登録し、module.get_meta() が読み取る際に現在の言語のテキストへ自動的に解決されます：

class MyModule(BaseModule):
    class I18nClass(BaseI18n):
        meta_description: I18nKey = I18nKey(
            default="Weather lookup",
            zh_CN="查询城市天气",
            en="Weather lookup",
        )

    @staticmethod
    def get_meta() -> ModuleMeta:
        return ModuleMeta(
            name="天気",
            description={"i18n": "MyModule.meta_description", "default": "Weather lookup"},
        )

SDK オブジェクト

sdk オブジェクトを通じてコア機能にアクセスします：

from ErisPulse import sdk

sdk.storage    # ストレージシステム
sdk.config     # 設定システム
sdk.logger     # ログシステム
sdk.adapter    # アダプターシステム
sdk.router     # ルーティングシステム
sdk.lifecycle  # ライフサイクルシステム

次のステップ

学習パスの続き

作成後は

模块核心概念

来源: docs/ja/developer-guide/modules/core-concepts.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/developer-guide/modules/core-concepts.md）

モジュールのコア概念

Note

L1 コア必読——トップページの「学習パス」にある5つのコアドキュメントの1つで、モジュール開発に必須です。

この文書では、モジュールのロードからアンロードまでの完全なライフサイクル、宣言的設定とストレージの使い方を解説します——フレームワークが自動で行うこと、あなたが書くべきことを、読み終える頃には把握できる内容です。

モジュールのライフサイクル

ロード戦略

from ErisPulse.Core.Bases import BaseModule
from ErisPulse.loaders import ModuleLoadStrategy

class MyModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        """モジュールのロード戦略を返す"""
        return ModuleLoadStrategy(
            lazy_load=True,   # 遅延ロードするか即時ロードするか
            priority=0,       # ロード優先度（数値が大きいほど先にロードされる）
            depends=["OtherModule"]  # オプション：依存する他のモジュールを宣言
        )

depends で宣言されたモジュールが未登録の場合、現在のモジュールはスキップされ、警告が記録されます。ロード順序はトポロジカルソートによって決まり、同じ階層では priority の降順で処理されます。

Note

カスケードアンロード / カスケードリロード（ErisPulse 2.8.0+）：他のモジュールから依存されているモジュールをアンロードする場合、それに依存しているモジュールは先にカスケードアンロードされます（ログにカスケードチェーンが記録されます）。任意のモジュール（ローカルプラグイン / 実行時登録 / PyPI インストールパッケージ）をホットリロードする場合も、それに依存しているモジュールは同様にカスケードリロードされ、依存しているモジュールが無効になったインスタンス参照を保持したまま動作し続けることを防ぎます。循環依存が宣言されている場合は、ロード時に RuntimeError で拒否されます。

reload は全量リロード（ErisPulse **2.11.0+**）：reload() はデフォルトでトップレベルのパッケージ名に基づいて import キャッシュをクリアし、依存しているモジュールのコードを再インポートします。アンロード前に自動的に構文の事前チェックを行い（失敗時はアンロードを拒否し、旧インスタンスがサービスを継続）、成功後はコードの同一性を確認します（偽リロードの警告 / stale マーク）。ロード失敗時の状況は sdk.module.last_error(名前) で直接照会できます（traceback を含む）。

on_load メソッド

モジュールのロード時に呼び出され、リソースの初期化とイベントハンドラの登録に使用します：

async def on_load(self, event):
    # イベントハンドラを登録
    @command("hello", help="挨拶コマンド")
    async def hello_handler(event):
        await event.reply("こんにちは！")
    
    # SDK 内蔵の HTTP クライアントを使用（接続プールを自動管理、session を手動で作成する必要なし）
    # sdk.client を通じてリクエストを送信できます

on_unload メソッド

モジュールのアンロード時に呼び出され、リソースのクリーンアップに使用します：

async def on_unload(self, event):
    # カスタムリソースをクリーンアップ
    # sdk.client はフレームワークが管理するため、手動で閉じる必要なし
    
    # イベントハンドラをキャンセル（フレームワークが自動処理）
    self.logger.info("モジュールはアンロードされました")

バックグラウンドタスクの作成とクリーンアップ（self.spawn() / フレームワークによるフォールバックキャンセル）の詳細は、ライフサイクル管理 を参照してください。

on_before_unload メソッド

Note

このフックには ErisPulse 2.11.0+ が必要です。

アンロードパイプラインの最初のステップ（on_unload とリソース回収の前に実行）でトリガーされるオプションのフックです——この時点ではモジュールのすべてのリソースがまだ無傷のため、アンロード / スリープ / リロードの前に状態を sdk.storage に保存し、次回の on_load で復元するのに適しています（遅延ロードのスリープにおける状態の引き継ぎはまさにこれに基づいています）：

async def on_before_unload(self, event):
    # event["reason"] は目的を示す：unload / reload / disable / cascade / dormant
    if event["reason"] in ("reload", "dormant"):
        await self.sdk.storage.aset(f"{self._module_name}:state", self.collect_state())

フックが失敗した場合は警告が出るだけで、アンロードフローはブロックされません。このフックを宣言していないモジュールの動作は変わりません。スリープ / ウェイクのセマンティクスの詳細は、 遅延ロードのスリープ を参照してください。

アンロードと完全アンロード（purge）

Note

この機能には ErisPulse 2.8.0+ が必要です。

unload() はデフォルトでロードの取り消しのみを行い（インスタンスとリソースをアンロード）、登録スタブ（モジュールクラスとメタ情報）は保持します——モジュールは discover で再発見でき、register() をやり直すことなく load() で再インスタンス化できます。

完全アンロード（モジュールクラスへの参照を解放し、sys.modules をクリーンアップして、プラグインとその専用の依存関係を GC で回収できるようにする）が必要な場合は、purge=True を渡します：

# ロードの取り消しのみ：登録スタブは保持され、いつでも再 load() 可能
await sdk.module.unload("MyModule")

# 完全アンロード：登録スタブを削除 + sys.modules をクリーンアップ（プラグインソース）
await sdk.module.unload("MyModule", purge=True)
セマンティクス unload() デフォルト unload(purge=True)
インスタンスとリソースのアンロード（イベント/task/ルーティング/lifecycle/i18n） ✅ ✅
登録スタブの保持（モジュールクラスとメタ情報） ✅ ❌ 削除
sys.modules のクリーンアップ（プラグインフォルダソースのみ） ❌ ✅
モジュールクラスが GC で回収可能 ❌ ✅
再ロード load() でそのまま可能 先に register() + load() が必要

purge=True の場合、カスケードアンロードされる依存先のモジュールも同様に purge されます。アンロード後、フレームワークは gc.collect() を実行し、モジュールクラス/インスタンスが回収可能かどうかをチェックします。残存する参照はログに警告として記録されます（参照元を含む、DEBUG レベル）。

ライフサイクルの全体像

上記のメソッドを組み合わせると、フレームワークがモジュールのロードとアンロード時に裏で自動的に行うすべての処理は以下の通りです：

flowchart TD
    subgraph Load["ロード（register → load）"]
        L1["register：モジュールクラスとメタ情報を登録"] --> L2["依存関係チェック<br/>欠落していればスキップ"]
        L2 --> L3["トポロジカルソート（Kahn + priority）"]
        L3 --> L4["owner として current_owner を注入"]
        L4 --> L5["設定テンプレートの生成 + i18n 翻訳キーの登録"]
        L5 --> L6["モジュールのインスタンス化（sdk を注入）"]
        L6 --> L7["on_load() を呼び出し"]
        L7 --> L8["sdk 属性へのマウント + emit module.load"]
    end

    subgraph Unload["アンロード（unload）"]
        U1["on_unload() を呼び出し"] --> U2["フォールバックでバックグラウンドタスクをキャンセル（self.spawn の帰属）"]
        U2 --> U3["i18n 翻訳キーのクリーンアップ"]
        U3 --> U4["ルーティング / コマンド / イベントハンドラの削除（owner 別）"]
        U4 --> U5["lifecycle フックのクリーンアップ（owner 別）"]
        U5 --> U6["SDK 属性 + 遅延ロードプロキシの削除"]
        U6 --> U7["emit module.unload"]
    end

    Load --> Unload

ロード時にフレームワークが自動で行うこと（あなたは on_load を書くだけで、残りは自動的に完了します）：

段階 フレームワークが自動で行うこと
owner 注入 インスタンス化の間、owner_scope でモジュール名を囲みます——あなたが on_load 内で登録したコマンド/イベント/フック/バックグラウンドタスクは自動的にこのモジュールに帰属し、アンロード時に owner 別に一括クリーンアップされます
設定テンプレート ConfigClass を宣言したモジュールでは、フレームワークが ErisPulse.<ModuleName> 設定セクションを自動生成/自動入力します
i18n 翻訳キー I18nClass を宣言したモジュールでは、翻訳キーが自動登録されます（アンロード時に自動的に登録解除）
依存トポロジ depends の宣言に基づいてソートし、依存されるモジュールが先にロードされることを保証します。循環依存は RuntimeError で拒否されます
SDK マウント インスタンス化後に sdk.<ModuleName> にマウントされ、sdk.MyModule.xxx でアクセスできるようになります

アンロード時にフレームワークがクリーンアップするもの（上記の U1→U7 に対応）：on_unload の実行後にフォールバックとしてクリーンアップを行います——バックグラウンドタスクの強制キャンセル（self.spawn で作成されたもの。適切な終了処理は on_unload 内で自前に行ってください）、i18n キー、ルーティング、コマンド/イベントハンドラ、lifecycle フック、最後に SDK 属性を削除します。purge=True の場合はさらに登録スタブの削除 + sys.modules のクリーンアップが行われます。

この自動クリーンアップこそが、「on_load/on_unload だけ書けば、手動で unregister する必要はない」の根拠です——フレームワークは owner による帰属を利用して、「登録した者がクリーンアップする」仕組みをワンアクションで実現しています。

SDK オブジェクト

コアモジュールへのアクセス

from ErisPulse import sdk

# sdk オブジェクトを通じてすべてのコアモジュールにアクセス
sdk.logger.info("ログ")
sdk.storage.set("key", "value")
config = sdk.config.getConfig("MyModule")

モジュール間通信

# 他のモジュールにアクセス
other_module = sdk.OtherModule
result = await other_module.some_method()

アダプター送信メソッドの照会

新しい標準仕様では、__getattr__ メソッドをオーバーライドしてフォールバック送信メカニズムを実装することが求められているため、hasattr メソッドでメソッドの存在を確認することができなくなりました。2.3.5 から、送信メソッドを照会する機能が新たに追加されました。

サポートされている送信メソッドの一覧

# プラットフォームがサポートするすべての送信メソッドを一覧表示
methods = sdk.adapter.list_sends("onebot11")
# 戻り値: ["Text", "Image", "Voice", "Markdown", ...]

メソッドの詳細情報の取得

# 特定のメソッドの詳細情報を取得
info = sdk.adapter.send_info("onebot11", "Text")
# 戻り値:
# {
#     "name": "Text",
#     "parameters": [
#         {"name": "text", "type": "str", "default": null, "annotation": "str"}
#     ],
#     "return_type": "Awaitable[Any]",
#     "docstring": "テキストメッセージを送信します..."
# }

設定管理

宣言的設定（推奨）

v2.5.2 から、モジュールは ConfigClass で設定クラスを宣言でき、アダプターと同じ設定 Schema システムを使用できます。設定は self.cfg でリアルタイムに読み取られ、変更後すぐに反映されます：

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseModule
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_key: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_module.api_key", "default": "API キー"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 1},
        },
    )
    timeout: int = field(
        default=30,
        metadata={
            "description": {"i18n": "my_module.timeout", "default": "タイムアウト時間（秒）"},
            "ui": {"widget": "number", "group": "advanced", "order": 2},
        },
    )

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    def __init__(self, sdk):
        self.sdk = sdk
        self.logger = sdk.logger.get_child("MyModule")

    async def on_load(self, event):
        self.logger.info("モジュールがロードされました")

    async def on_unload(self, event):
        pass

    async def do_something(self):
        cfg = self.cfg  # リアルタイム読み取り、型安全
        api_key = cfg.api_key
        timeout = cfg.timeout

BaseConfig は汎用の設定基底クラスで、アダプター、モジュール、外部プロジェクトなどあらゆるシーンに適用できます。設定フィールドは i18n による多言語説明に対応しています（詳細は i18n ドキュメント を参照）。

Note

モジュール（アダプターを含む）のロード、アンロード、起動・停止などのライフサイクルは、フレームワークが統一的に管理します（module manager / scope が照会・制御 API を提供）。ConfigClass はビジネスパラメータのみを担うべきであり、モジュールが内部で「自分を起動するかどうか」を判断するロジックを再度ネストする必要はありません。

設定 Schema システムはさらに以下にも対応しています（v2.8.0+、詳細は アダプター core-concepts を参照）：

宣言的翻訳キー（v2.7.0+）

v2.7.0 から、モジュールは ConfigClass を宣言するのと同じように、ネストクラス I18nClass を通じて翻訳キーを一括宣言できるようになりました。フレームワークはロード時に宣言されたすべての翻訳キーを自動登録し、手動で i18n.register() を呼び出す必要はありません。また、登録のタイミングは設定テンプレートの生成より前であるため、設定説明で参照される i18n キーが確実に利用可能になります。

from ErisPulse.Core.Bases import BaseConfig, BaseI18n, I18nKey

class MyModule(BaseModule):
    # 設定クラス（オプション）
    @dataclass
    class ConfigClass(BaseConfig):
        welcome_msg: str = field(
            default="ようこそ",
            metadata={
                "description": {"i18n": "mymodule.welcome_msg", "default": "ウェルカムメッセージ"},
            },
        )

    # 翻訳キーコレクションクラス（オプション）
    class I18nClass(BaseI18n):
        # 属性名から完全なキーパスが自動的に組み立てられます：<モジュール名>.<属性名>
        welcome_msg: I18nKey = I18nKey(
            default="Welcome Message",   # 言語に依存しないフォールバック
            zh_CN="欢迎消息",
            zh_TW="歡迎訊息",
            en="Welcome Message",
            ja="ウェルカムメッセージ",
            ru="Приветственное сообщение",
        )
        hello: I18nKey = I18nKey(
            default="Hello, {name}!",
            zh_CN="你好，{name}！",
            zh_TW="你好，{name}！",
            en="Hello, {name}!",
            ja="こんにちは、{name}！",
            ru="Привет, {name}!",
        )

詳細は i18n 推奨記法 を参照してください。

手動での設定読み取り（非推奨）

非推奨：代わりに 宣言的設定 + self.cfg によるリアルタイム読み取りを使用してください。

class MyModule(BaseModule):
    def __init__(self, sdk):
        self.sdk = sdk

    def _load_config(self):
        config = self.sdk.config.getConfig("MyModule")
        if not config:
            self.sdk.config.setConfig("MyModule", {"api_key": "", "timeout": 30})
            return {"api_key": "", "timeout": 30}
        return config

ストレージシステム

基本的な使い方

# データを保存
sdk.storage.set("user:123", {"name": "山田太郎"})

# データを取得
user = sdk.storage.get("user:123", {})

# データを削除
sdk.storage.delete("user:123")

トランザクションの使用

# トランザクションを使用してデータの一貫性を保証
with sdk.storage.transaction():
    sdk.storage.set("key1", "value1")
    sdk.storage.set("key2", "value2")
    # いずれかの操作が失敗した場合、すべての変更がロールバックされます

イベント処理

イベントハンドラの登録

from ErisPulse.Core.Event import command, message

# コマンドを登録
@command("info", help="情報を取得")
async def info_handler(event):
    await event.reply("これは情報です")

# メッセージハンドラを登録
@message.on_group_message()
async def group_handler(event):
    sdk.logger.info(f"グループメッセージを受信: {event.get_text()}")

イベントハンドラのライフサイクル

フレームワークがイベントハンドラの登録と登録解除を自動的に管理するため、あなたは on_load 内で登録するだけで済みます。

遅延ロード機構

仕組み

# モジュールは初めてアクセスされた時に初期化されます
result = await sdk.my_module.some_method()
# ↑ ここでモジュールの初期化がトリガーされます

即時ロード

即時初期化が必要なモジュール（リスナー、タイマーなど）の場合：

@staticmethod
def get_load_strategy():
    return ModuleLoadStrategy(
        lazy_load=False,  # 即時ロード
        priority=100
    )

エラー処理

例外のキャッチ

async def handle_event(self, event):
    try:
        # ビジネスロジック
        await self.process_event(event)
    except ValueError as e:
        self.logger.warning(f"パラメータエラー: {e}")
        await event.reply(f"パラメータエラー: {e}")
    except Exception as e:
        self.logger.error(f"処理に失敗しました: {e}")
        raise

ログ記録

# 異なるログレベルを使用
self.logger.debug("デバッグ情報")    # 詳細なデバッグ情報
self.logger.info("実行状態")      # 通常の実行情報
self.logger.warning("警告情報")  # 警告情報
self.logger.error("エラー情報")    # エラー情報
self.logger.critical("致命的エラー") # 致命的エラー

関連ドキュメント

学習パスの続き

深掘りとトラブルシューティング

Event 包装类

来源: docs/ja/developer-guide/modules/event-wrapper.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/developer-guide/modules/event-wrapper.md）

Event 包装类详解

Note

L1 核心必読——「読むべき道筋」の5つの核心ドキュメントの1つ、モジュール開発に必須。

プロセッサの event パラメータが直接 .get_text() / .reply() / .wait_reply() を呼べるのは、Event 包装クラスのおかげです。本記事では、Event 包装クラスの全メソッド一覧を紹介します。イベントハンドラを書く際には、このリストを参照してください。メソッドの署名と動作は、ここを基準とします。

event パラメータに型注釈を追加する

イベントハンドラの event パラメータは Event 包装クラス（dict のサブクラス）です。型注釈を付けることを強く推奨します：

from ErisPulse.Core.Event import Event

@message.on_private_message()
async def handler(event: Event):
    text = event.get_text()   # IDE が便利なメソッドを自動補完
    await event.reply(text)   # 拼写ミスは静的チェック時に発見可能

型注釈を付けないと、IDE は Event 上のメソッド（get_text() / reply() / wait_reply() / プラットフォーム拡張メソッド）を認識できず、すべて手動で記述する必要があります。

注意：イベントハンドラのコールバックの event は Event 包装クラス（注釈は Event）です。一方、モジュールのライフサイクルメソッド on_load / on_unload の event は普通の dict（注釈は dict）です。これらを混同しないでください。

核心特性

核心フィールドメソッド

from ErisPulse.Core.Event import command

@command("info")
async def info_command(event: Event):
    event_id = event.get_id()
    platform = event.get_platform()
    time = event.get_time()
    print(f"ID: {event_id}, プラットフォーム: {platform}, 時間: {time}")

メッセージイベントメソッド

from ErisPulse.Core.Event import message

@message.on_private_message()
async def private_handler(event: Event):
    text = event.get_text()
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"こんにちは、{nickname}！")

メッセージタイプの判断

from ErisPulse.Core.Event import message

@message.on_group_message()
async def group_handler(event: Event):
    is_private = event.is_private_message()
    is_group = event.is_group_message()
    is_at = event.is_at_message()
    await event.reply(f"タイプ: {'プライベートチャット' if is_private else 'グループチャット'}")

回答機能

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event: Event):
    await event.reply("名前を入力してください:")
    reply = await event.wait_reply(timeout=30)
    if reply:
        name = reply.get_text()
        await event.reply(f"こんにちは、{name}！")

@command("price")
async def price_command(event: Event):
    await event.reply("金額を入力してください（例：5元）:")
    # 回答は正規表現に一致する必要があります。一致しない場合は、タイムアウトまで待機します。
    reply = await event.wait_reply(timeout=30, regex=r"\d+\s*元")
    if reply:
        await event.reply(f"受信金額: {reply.get_text()}")

会話の高度な機能

Note

本節の機能は ErisPulse 2.8.0+ が必要です。

# 会話のリマインダー：5分間返信がなければリマインダーを送信し、ユーザーが返信すると自動的にキャンセル
reminder = event.remind(300, "まだですか？話したくない場合は「退出」を送信してください")
reminder.cancel()  # 手動でキャンセルすることも可能です

# タイムアウトの昇格：時間内に必ず到達（返信でキャンセルされない）、長時間処理がなければ管理者に通知
event.escalate(1800, lambda e: notify_master("工事件がタイムアウトしました"))

# 複数ルートの待機：「同意」および「拒否」を同時に待機し、先着順
which, reply = await event.select(
    event.expect(pattern="同意*", user="10001"),
    event.expect(pattern="拒否*", user="10002"),
    timeout=60,
)
if which is None:
    await event.reply("承認が受け取られませんでした")

# 会話レベルの待機：同じグループ内の誰からの返信でも対象になります（グループ協力）
reply = await event.wait_reply(session=True, prompt="誰かが回答してくれませんか？")

# 会話の受信箱：現在の会話の最近20件のメッセージ（ロボット含む、AIコンテキスト/再送防止のベース）
messages = await event.history(20)

# メッセージトランザクション：例外が発生した場合、トランザクション内で送信されたメッセージは自動的に撤回されます
async with event.message_tx():
    await event.reply("処理中、少々お待ちください...")
    result = await do_something()
    await event.reply(f"完了: {result}")

コマンド情報の取得

from ErisPulse.Core.Event import command

@command("cmdinfo")
async def cmdinfo_command(event: Event):
    cmd_name = event.get_command_name()
    cmd_args = event.get_command_args()
    await event.reply(f"コマンド: {cmd_name}, 引数: {cmd_args}")

通知イベントメソッド

from ErisPulse.Core.Event import notice

@notice.on_friend_add()
async def friend_add_handler(event: Event):
    await event.reply("友達追加ありがとうございます！")

メソッドの速見表

核心メソッド

イベントの基本情報

ロボット情報

会話識別子

メッセージイベントメソッド

メッセージ内容

送信者情報

グループ/チャンネル情報

@メッセージ関連

メッセージタイプの判断

基本判断

通知イベントメソッド

通知操作者

通知タイプの判断

リクエストイベントメソッド

リクエスト情報

リクエストタイプの判断

回答機能

基本回答

プラットフォーム機能の確認

転送機能

注意：転送機能はアダプタの Send DSL によって実現されるため、Event 包装クラス自体には直接の転送メソッドは提供されていません。

# メッセージをグループに転送
adapter = sdk.adapter.get(event.get_platform())
target_id = event.get_group_id()  # または他のグループIDを指定
await adapter.Send.To("group", target_id).Text(event.get_text())

返信を待つ機能

会話機能

会話機能の例

confirm() - 確認会話：

@command("delete", help="データを削除")
async def delete_handler(event: Event):
    if await event.confirm("すべてのデータを削除しますか？"):
        sdk.storage.delete("all_data")
        await event.reply("データを削除しました")
    else:
        await event.reply("キャンセルしました")

confirm() - ヒント付き：

# hint=True はプロンプト末尾に "（はい/いいえ）" を追加します
if await event.confirm("続行しますか？", hint=True):
    await event.reply("続行しました")
# ユーザーが見るのは：続行しますか？（はい/いいえ）

choose() - 選択メニュー：

@command("color", help="色を選択")
async def color_handler(event: Event):
    choice = await event.choose("色を選択してください：", ["赤", "緑", "青"])
    if choice is not None:
        colors = ["赤", "緑", "青"]
        await event.reply(f"選択した色は：{colors[choice]}")

choose() - 選択肢のフォーマットとメッセージのマージ：

# inline 形式：選択肢を1行に表示
choice = await event.choose("選択してください：", ["A", "B", "C"], options_format="inline")
# 出力：1.A | 2.B | 3.C

# 自定義フォーマット
choice = await event.choose("選択してください：", ["猫", "犬"],
    options_format=lambda opts: " / ".join(opts))
# 出力：猫 / 犬

# options_format="auto"（デフォルト）：method に応じて自動的に組み込みスタイルを選択
# Markdown → 無序リスト
choice = await event.choose(
    "## 選択してください", ["猫", "犬"],
    method="Markdown",  # auto は md 列表を自動認識
)
# 出力：
# ## 選択してください
# - 1. 猫
# - 2. 犬

# Html → 有序リスト
choice = await event.choose(
    "<h2>選択してください</h2>", ["猫", "犬"],
    method="Html", merge_prompt=True,  # auto は html 列表を自動認識
)
# 出力：
# <h2>選択してください</h2>
# <ol><li>1. 猫</li><li>2. 犬</li></ol>

# マージモード + 置換文字
choice = await event.choose(
    "## 選択してください\n{options}\n番号を返信してください",
    ["猫", "犬"],
    method="Markdown", merge_prompt=True,
)

# 自定義置換文字
choice = await event.choose(
    "選択してください: [choices]",
    ["猫", "犬"],
    placeholder="[choices]",
)

collect() - フォーム収集：

@command("register", help="登録")
async def register_handler(event: Event):
    data = await event.collect([
        {"key": "name", "prompt": "名前を入力してください："},
        {"key": "age", "prompt": "年齢を入力してください：",
         "validator": lambda e: e.get_text().isdigit()},
    ])
    if data:
        await event.reply(f"登録完了！{data['name']}、{data['age']}歳")

非 Text メソッドの reply：

await event.reply("http://example.com/img.jpg", method="Image")
await event.reply("http://example.com/audio.mp3", method="Voice")

from ErisPulse.Core.Event import MessageBuilder
segments = MessageBuilder.text("画像を見てください：").image("http://example.com/img.jpg").build()
await event.reply_ob12(segments)

完全な Conversation 多ラウンド会話の使い方は Conversation 多ラウンド会話 を参照してください。

コマンド情報

コマンドの基本

元のデータ

プラットフォーム拡張メソッド

アダプタは Event 包装クラスにプラットフォーム固有のメソッドを登録できます。メソッドは対応するプラットフォームの Event インスタンスでのみ利用可能で、他のプラットフォームでアクセスすると AttributeError が発生します。

プラットフォームメソッドは Event.__getattribute__ によって、組み込みメソッドよりも優先的に有効になるため、confirm、choose、collect、wait_reply などの組み込みインタラクティブメソッドを覆い、プラットフォーム固有の実装（例: ボタン、カードなど）を提供できます。組み込み実装は覆い書きする側に _builtin_* 関数として導出されます。

# メールイベント - メールメソッドのみ
event = Event({"platform": "email", "email_raw": {"subject": "Hello"}})
event.get_subject()      # ✅ "Hello" を返す
event.get_chat_type()    # ❌ AttributeError

# Telegram イベント - Telegram メソッドのみ
event = Event({"platform": "telegram", "telegram_raw": {"chat": {"type": "private"}}})
event.get_chat_type()    # ✅ "private" を返す
event.get_subject()      # ❌ AttributeError

# 組み込みメソッドは常に利用可能
event.get_text()         # ✅ どのプラットフォームでも
event.reply("hi")        # ✅ どのプラットフォームでも

登録されたメソッドの確認

from ErisPulse.Core.Event import get_platform_event_methods

methods = get_platform_event_methods("email")
# ["get_subject", "get_from", ...]

hasattr と dir のサポート

hasattr(event, "get_subject")   # platform="email" のみで True を返す
"get_subject" in dir(event)     # 同上

跨プラットフォーム拡張（ワイルドカード）

register_event_method と register_event_mixin は "*" をプラットフォーム名として渡すことができ、登録されたメソッドはすべてのプラットフォームの Event インスタンスで利用可能になります。AI会話、コンテキスト管理など、跨プラットフォームで再利用可能な機能に適しています。

from ErisPulse.Core.Event.wrapper import register_event_method

@register_event_method("*")
async def ai_chat(self, prompt: str):
    # self は Event インスタンス、イベントデータと組み込みメソッドにアクセス可能
    await self.reply(f"AI: {prompt}")

登録後、どのプラットフォームのイベントハンドラでも event.ai_chat(...) を呼び出すことができます。

メソッドの優先順位（高い順）：プラットフォーム固有メソッド → ワイルドカードメソッド → 組み込みメソッド → dict キーのアクセス。

アダプタ開発者が拡張メソッドを登録する方法については イベントシステム API - 跨プラットフォーム拡張ワイルドカード を参照してください。

関連ドキュメント

モジュール開発の継続

インタラクティブ機能の深掘り

模块最佳实践

来源: docs/ja/developer-guide/modules/best-practices.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/developer-guide/modules/best-practices.md）

モジュール開発ベストプラクティス

Tip

いつこのドキュメントを読むべきか：最初のモジュールを書き終えた後に読むことを推奨（エンジニアリングのベストプラクティス）。必要なければスキップしても、他の機能の利用には影響しません。

ここでは、モジュール開発におけるエンジニアリング上の作法をまとめています：非同期操作はイベントループをブロックしない、設定は宣言的に、他のモジュールのリソースを管理する場合はアンロード時に解放する——これに従って書けば、メモリリーク、バックグラウンドタスクの残存、ホットリロードの失効といったモジュール開発で最もよくある落とし穴を回避できます。

モジュール設計

1. 単一責任の原則

各モジュールは1つのコア機能のみを担当すべきです：

# 良い設計：各モジュールは1つの機能のみを担当
class WeatherModule(BaseModule):
    """天気照会モジュール"""
    pass

class NewsModule(BaseModule):
    """ニュース照会モジュール"""
    pass

# 悪い設計：1つのモジュールが複数の無関係な機能を担当
class UtilityModule(BaseModule):
    """天気、ニュース、ジョークなど複数の機能を含む"""
    pass

2. モジュール命名規則

[project]
name = "ErisPulse-ModuleName"  # ErisPulse- プレフィックスを使用

3. 明確な設定管理

宣言的設定（ConfigClass + BaseConfig）の使用を推奨します。型安全性、自動テンプレート生成、WebUI フォーム対応などの機能が得られます：

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_url: str = field(default="https://api.example.com", metadata={
        "description": {"i18n": "my_module.api_url", "default": "API アドレス"},
    })
    timeout: int = field(default=30, metadata={
        "description": {"i18n": "my_module.timeout", "default": "タイムアウト時間（秒）"},
    })
    cache_ttl: int = field(default=3600, metadata={
        "description": {"i18n": "my_module.cache_ttl", "default": "キャッシュ生存時間（秒）"},
    })

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    async def do_something(self):
        cfg = self.cfg  # 型安全、リアルタイム読み取り
        await self._fetch(cfg.api_url, timeout=cfg.timeout)

手動方式で設定ストレージの読み書きを行うことも引き続き可能です（モジュールのコア概念を参照）。

宣言的翻訳キー（v2.7.0+）

モジュールは I18nClass を通じて翻訳キーを一元的に宣言でき、フレームワークが自動的に i18n システムへ登録するため、手動で i18n.register() を呼び出す必要はありません。

from ErisPulse.Core.Bases import BaseI18n, I18nKey

class MyModule(BaseModule):
    class I18nClass(BaseI18n):
        # プレースホルダー付きのビジネス翻訳キー
        welcome: I18nKey = I18nKey(
            default="Welcome, {name}!",
            zh_CN="欢迎你，{name}！",
            zh_TW="歡迎你，{name}！",
            en="Welcome, {name}!",
            ja="ようこそ、{name}！",
            ru="Добро пожаловать, {name}!",
        )
        # 設定フィールド説明の翻訳
        api_url: I18nKey = I18nKey(
            default="API URL",
            zh_CN="API 地址",
            zh_TW="API 位址",
            en="API URL",
            ja="API URL",
            ru="API URL",
        )

詳細な使い方は i18n ドキュメント を参照してください。

非同期プログラミング

1. 非同期ライブラリの使用

# SDK 組み込み HTTP クライアントの使用を推奨（非同期、ログと統計を自動取得）
from ErisPulse.Core import client

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await client.get(url)
        return await resp.json()

# sdk.client 経由でも使用可能（効果は同じ）
from ErisPulse import sdk

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await sdk.client.get(url)
        return await resp.json()

# aiohttp を直接インポートしない（フレームワークによる一元管理ができない）
import aiohttp

class MyModule(BaseModule):
    async def fetch_data(self, url):
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()

# requests は使用しない（同期のためイベントループをブロックする）
import requests

class MyModule(BaseModule):
    def fetch_data(self, url):
        return requests.get(url).json()  # イベントループをブロックする

2. 正しい非同期操作

from ErisPulse.Core.Event import Event  # event: Event アノテーションで IDE 補完が得られる

async def handle_command(self, event: Event):
    # 結果を待つ必要のある時間のかかる操作：直接 await（ライフサイクルが明確）
    result = await self._long_operation()

async def on_load(self, event: dict):
    # バックグラウンドタスク（ポーリング/タイマー/fire-and-forget）：self.spawn() を使用。
    # モジュールのアンロード時、フレームワークが on_unload の後にフォールバックでキャンセルするため、
    # self を保持することによるリークを回避できる
    self.spawn(self._poll())

Note

バックグラウンドタスクには self.spawn() を推奨します（ErisPulse 2.8.0+）。2.8.3 以降は、素の asyncio.create_task も自動的にモジュールへ暗黙的に帰属します（Task Factory が自動登録し、アンロード時にフォールバックでキャンセル、self 参照はリークしなくなります）。 self.spawn() が引き続き推奨の書き方です——非メインループのスレッドからメインループへのスケジューリングや、明示的な owner= 指定をサポートします。 2.8.3 より前のバージョンでは、素のタスクは帰属せず self 参照を保持し続けるためモジュールインスタンスが回収できなくなり （ホットリロード時のリーク）、self.spawn() を使う必要があります。詳細は ライフサイクル管理 を参照。

3. リソース管理

async def on_load(self, event):
    # SDK クライアントは接続プールを自動管理するため、手動で session を作成する必要なし
    pass
    
async def on_unload(self, event):
    # カスタムクライアントが必要な場合は、リソースのクリーンアップを忘れずに
    pass

イベント処理

1. Event ラッパークラスの使用

# Event ラッパークラスの便利メソッドを使用
@command("info")
async def info_command(event: Event):
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"こんにちは、{nickname}！")

# 辞書へ直接アクセスするのではなく
@command("info")
async def info_command(event: Event):
    user_id = event["user_id"]  # わかりにくく、間違いやすい

2. 遅延ロードの適切な使用

# 低頻度コマンドモジュール：activate_on トリガーを宣言し、最初に一致するコマンドが到着した時点で自動アクティブ化（遅延ロードを維持）
class CommandModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"command": {"name": "dice", "help": "サイコロを1つ振る", "aliases": ["d"]}},
        ])

# 低頻度リスナーモジュール：イベントトリガーを宣言し、イベント到着時に自動アクティブ化
class ListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"notice": "group_member_increase"},
        ])

# 高頻度トリガー（すべてのメッセージを処理する必要がある）や起動時に即時準備が必要なモジュール：即時ロード
class HotListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=False)

# ユーティリティモジュールは遅延ロードが適している
class UtilityModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True)

activate_on の完全な構文（イベントの3形式 / コマンド略記と dict 宣言 / help フォールバックチェーン）については 遅延ロードモジュールシステム を参照してください。

3. イベントハンドラーの登録

async def on_load(self, event):
    # on_load 内でイベントハンドラーを登録
    @command("hello")
    async def hello_handler(event: Event):
        await event.reply("こんにちは！")
    
    @message.on_group_message()
    async def group_handler(event: Event):
        self.logger.info("グループメッセージを受信しました")
    
    # 手動で登録解除する必要はなく、フレームワークが自動的に処理する

ユーティリティモジュール：他のモジュールのものを預かる時は「アンロード通知」を受け止める

いつ必要か：あなたのモジュールが他のモジュールの代わりにものを管理している場合（タイマーコールバック、サブスクライバー、接続、キャッシュエントリなど）。これらの参照を、相手モジュールのアンロード後も捨てずに保持し続けると、相手のインスタンスは永遠に回収されません——これがユーティリティモジュールで最もよくあるメモリリークの原因です。

from ErisPulse.Core.Bases import BaseModule
from ErisPulse.runtime import off_cleanup, on_cleanup

class MyToolModule(BaseModule):
    def __init__(self):
        self._entries = {}  # {モジュール名: 管理しているもの}

    def register(self, entry):
        owner = on_cleanup(self._drop)   # ① 登録時にクリーンアップチェーンへ登録し、呼び出し元を自動識別
        self._entries.setdefault(owner, []).append(entry)

    def _drop(self, owner: str):
        self._entries.pop(owner, None)   # ② 相手のアンロード時にフレームワークが自動呼び出し：相手のものを破棄

    async def on_unload(self, event):
        off_cleanup(self._drop)          # ③ 自身のアンロード前にフックを解除

たったこれだけで、フレームワークが保証します：

組み込まない場合の結果：相手が purge で完全アンロードされるときにインスタンスを回収できず（リーク診断が「回収不可」と報告）、相手自身も on_unload 内であなたへの登録解除を行わなければ、リークは恒久的なものになります。

通常のモジュール（他のモジュールのものを管理しない）はこれを気にする必要ありません——フレームワークリソース（コマンド / ハンドラー / ルート / バックグラウンドタスクなど）のアンロード時のクリーンアップは全自動です。

トリガータイミング、呼び出し元識別ルール、タイムアウトとフォールトトレランスなどの詳細は 所有権システム · ユーティリティモジュールガイド を参照してください。

エラー処理

1. 例外の分類処理

from ErisPulse.Core.Bases.errors import ClientError

async def handle_event(self, event: Event):
    try:
        result = await self._process(event)
    except ValueError as e:
        # 予期されるビジネスエラー
        self.logger.warning(f"ビジネス警告: {e}")
        await event.reply(f"パラメータエラー: {e}")
    except ClientError as e:
        # ネットワークエラー（sdk.client の基盤となる aiohttp 例外は自動変換済み）
        self.logger.error(f"ネットワークエラー {e.method} {e.url}: {e}")
        await event.reply("ネットワークリクエストに失敗しました。しばらくしてから再試行してください")
    except Exception as e:
        # 予期しないエラー
        self.logger.error(f"不明なエラー: {e}", exc_info=True)
        await event.reply("処理に失敗しました。管理者に連絡してください")
        raise

2. タイムアウト処理

# SDK 組み込みクライアントの使用を推奨（タイムアウトとリトライを内蔵）
from ErisPulse.Core import client
from ErisPulse.Core.Bases.errors import ClientTimeoutError

async def fetch_with_timeout(self, url, timeout=30):
    try:
        resp = await client.get(url, timeout=timeout)
        return await resp.json()
    except ClientTimeoutError:
        self.logger.warning(f"リクエストタイムアウト: {url}")
        raise

ストレージシステム

1. トランザクションの使用

# トランザクションを使用してデータの一貫性を確保
async def update_user(self, user_id, data):
    with self.sdk.storage.transaction():
        self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
        self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

# ❌ トランザクションを使わないとデータ不整合が起こる可能性
async def update_user(self, user_id, data):
    self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
    # ここでエラーが発生すると、上の設定はロールバックできない
    self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

2. バッチ操作

# バッチ操作でパフォーマンスを向上
def cache_multiple_items(self, items):
    self.sdk.storage.set_multi({
        f"item:{k}": v for k, v in items.items()
    })

# ❌ 何度も呼び出すと効率が悪い
def cache_multiple_items(self, items):
    for k, v in items.items():
        self.sdk.storage.set(f"item:{k}", v)

ロギング

1. ログレベルの適切な使用

# DEBUG: 詳細なデバッグ情報（開発時のみ）
self.logger.debug(f"入力パラメータ: {params}")

# INFO: 正常動作の情報
self.logger.info("モジュールがロードされました")
self.logger.info(f"リクエストを処理中: {request_id}")

# WARNING: 警告情報、主要機能には影響しない
self.logger.warning(f"設定項目 {key} が未設定のため、デフォルト値を使用します")
self.logger.warning("API の応答が遅く、最適化が必要な可能性があります")

# ERROR: エラー情報
self.logger.error(f"API リクエスト失敗: {e}")
self.logger.error(f"イベント処理失敗: {e}", exc_info=True)

# CRITICAL: 致命的エラー、即時対応が必要
self.logger.critical("データベース接続に失敗し、ボットが正常に動作できません")

2. 構造化ログ

# 構造化ログを使用すると解析しやすい
self.logger.info(f"リクエスト処理: request_id={request_id}, user_id={user_id}, duration={duration}ms")

# ❌ 非構造化ログを使用
self.logger.info(f"ユーザー {user_id} からのリクエストを処理しました。所要時間 {duration} ミリ秒")

パフォーマンス最適化

1. キャッシュの使用

class MyModule(BaseModule):
    def __init__(self):
        self._cache = {}
        self._cache_lock = asyncio.Lock()
    
    async def get_data(self, key):
        async with self._cache_lock:
            if key in self._cache:
                return self._cache[key]
            
            # データベースから取得
            data = await self._fetch_from_db(key)
            
            # データをキャッシュ
            self._cache[key] = data
            return data

2. ブロッキング操作の回避

# 非同期操作を使用
async def process_message(self, event: Event):
    # 非同期処理
    await self._async_process(event)

# ❌ ブロッキング操作
async def process_message(self, event: Event):
    # 同期操作、イベントループをブロックする
    result = self._sync_process(event)

セキュリティ

1. 機密データの保護

# 機密データは設定に保存（宣言的 ConfigClass、secret フィールドはログ/エクスポートに入らない）
from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseModule, BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_key: str = field(
        default="",
        metadata={"description": "API キー", "secret": True},
    )

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    def check_api_key(self):
        if not self.cfg.api_key or self.cfg.api_key == "YOUR_API_KEY_HERE":
            raise ValueError("config.toml で有効な API キーを設定してください")

# ❌ 機密データのハードコード
class MyModule(BaseModule):
    API_KEY = "sk-1234567890"  # こんなことをしてはいけない！

2. 入力検証

# ユーザー入力を検証
async def process_command(self, event: Event):
    user_input = event.get_text()
    
    # 入力長を検証
    if len(user_input) > 1000:
        await event.reply("入力が長すぎます。入力し直してください")
        return
    
    # 入力形式を検証
    if not re.match(r'^[a-zA-Z0-9]+$', user_input):
        await event.reply("入力形式が正しくありません")
        return

テスト

1. ユニットテスト

import pytest
from ErisPulse.Core.Bases import BaseModule

class TestMyModule:
    def test_config_defaults(self):
        """設定のデフォルト値をテスト"""
        config = MyModule.ConfigClass()
        assert config.timeout == 30

2. 結合テスト

@pytest.mark.asyncio
async def test_command_handling():
    """コマンド処理をテスト"""
    module = MyModule()
    await module.on_load({})
    
    # コマンドイベントをシミュレート
    event = create_test_command_event("hello")
    await module.handle_command(event)

デプロイ

1. バージョン管理

[project]
name = "ErisPulse-MyModule"
version = "1.0.0"

セマンティックバージョニングに従う：

2. README ヘッダー

epsdk create が生成する README には ErisPulse ヘッダー識別（ロゴ + バッジ行）が組み込まれています。推奨される2つのパターン：

パターン A — ErisPulse ロゴのみ（デフォルト）：

<div align="center">

<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" width="180" alt="MyModule" />

# MyModule

**一言説明**

<p>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/pypi/v/ErisPulse-MyModule?style=for-the-badge&logo=pypi&logoColor=white" alt="PyPI"></a>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/badge/Python-3.10+-FFD43B?style=for-the-badge&logo=python&logoColor=blue" alt="Python"></a>
  <a href="./LICENSE"><img src="https://img.shields.io/badge/License-MIT-blue?style=for-the-badge" alt="License"></a>
  <a href="https://github.com/ErisPulse/ErisPulse"><img src="https://img.shields.io/badge/Powered_by-ErisPulse-FF6B9D?style=for-the-badge&logo=bookstack&logoColor=white" alt="ErisPulse"></a>
</p>

</div>

パターン B — モジュールアイコン × ErisPulse ロゴ（カスタムアイコンがある場合）：

<div align="center">

<img src=".github/assets/MyModuleIcon.svg" width="120" alt="MyModule" />
<span style="font-size:44px;color:#c8c8c8;margin:0 18px;vertical-align:middle;">×</span>
<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" height="120" alt="ErisPulse" />

# MyModule
（バッジ行は上に同じ）
</div>

必要に応じて GitHub Stars、Downloads などのバッジを追加できます。ロゴをプロジェクトのローカル（.github/assets/ErisPulseLogo.png）にダウンロードして、相対パス参照に変更することもできます。

関連ドキュメント

学習パスに戻る

シナリオ別に詳しく

ORM 数据模型

来源: docs/ja/developer-guide/orm.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/developer-guide/orm.md）

データモデル層（ORM）

Tip

この記事を読むタイミング：データモデル、自動テーブル作成、関連マッピングが必要な場合。不要であれば飛ばしても、他の機能の使用に影響はありません。

2.9.0 から、フレームワークに宣言型のデータモデル層が内蔵されました。Model を継承し、Field を使ってフィールドを宣言するだけで、自動テーブル作成と CRUD 機能が得られます。モデルは、内蔵されたストレージ層（SQLite / MySQL / PostgreSQL）の上に直接構築されています。ストレージのバックエンドは ErisPulse.storage.backend で設定され、バックエンドの切り替えに伴いモデルのコードを変更する必要はありません。

モデルの宣言

from ErisPulse.Core.Bases import Model, Field

class User(Model):
    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0, ge=0, le=150)
    role: str = Field(default="user", choices=["user", "admin"])
    tags: list = Field(default_factory=list)          # JSON 列、読み書き時に自動的にシリアライズ
    bio: str = Field(default="", description={"i18n": "user.bio", "default": "自己紹介"})

フィールドパラメータ

パラメータ 説明
default 既定値（提供されていない場合、かつ自動増分でない → NOT NULL が必須）
default_factory 変更可能な既定値の工場（例: list）
primary_key 主キー
autoincrement 自動増分主キー（暗黙の主キー、create 後に自動で戻り値を埋め込む）
max_length 文字列の最大長さ（生成される VARCHAR(n)、書き込み時の検証）
nullable NULL を許可するかどうか（デフォルトは False）
index 普通のインデックスを生成
unique 一意制約
choices / ge / le 列挙 / 数値範囲（書き込み時の検証）
description 説明（i18n ディクショナリ、構成クラスと同じ形式）
column_type SQL 列の型定義を上書き

建表と CRUD

await User.create_table()                      # 幂等（IF NOT EXISTS）

user = await User.create(name="Alice", age=20) # 挿入（自動増分主キーは自動的に戻り値として返される）
got = await User.get(id=user.id)               # 等値検索で最初の1件を取得

users = await User.where(User.age > 18).order_by("-age").limit(10).all()
first = await User.where(User.name == "Alice").first()
total = await User.count(User.age > 18)

user.age = 21
await user.save()                              # 主キーに基づいて更新（事前制約検証）；主キーが存在しない場合は挿入に降格（自動増分主キーもインスタンスに戻り値として返される）
await user.delete()                            # 主キーに基づいて削除

await User.update_all(User.age > 18, role="adult")  # バッチ更新
await User.delete_all(User.age > 100)               # バッチ削除

クエリ式は > >= < <= == !=、in_([...])、&（AND）/ |（OR）による組み合わせをサポートします：

await User.where((User.age > 18) & User.name.in_(["Alice", "Bob"])).all()

トランザクション

ORM の読み書きとフレームワークのストレージ層は、同じトランザクションルーティングを使用します。storage.atransaction() 環境のトランザクション内では、create / save / delete とクエリは自動的にトランザクション接続を再利用し、トランザクションに合わせて一括でコミットまたはロールバックされます。個別にコミットされることはありません。

async with storage.atransaction():
    await User.create(name="Alice")
    ...  # ブロック内で例外が発生した場合、上記の INSERT もロールバックされます

宣言式構成クラスとの関係

モデルの宣言と ConfigClass（@dataclass + field(metadata=...)）は同じ基底を共有しています。制約辞書、検証エンジン（validate_field_constraints）、型カテゴリ登録表（python_type_category）などです。ただし、クラスの基盤は意図的に分離されています。構成フィールドは普通の値（TOMLの往復、一度のロードによるホット更新）であり、モデルフィールドは列記述子（クラスアクセス = クエリ式、行ごとのインスタンス）です。同じ宣言構文を用い、それぞれ異なる役割を持つ2つの基盤です。

どちらの宣言を使うべきか

維度 構成クラス field(metadata=...) モデルフィールド Field()
適用場面 モジュールの動作パラメータ（少数、人間が読みやすい、ホット更新が必要） 業務データのレコード（複数行、プログラムによる読み書き、クエリが必要）
ストレージ形態 config.toml（コメントを保持、TOMLの往復） データベーステーブル（自動作成、SQL方言）
値の形態 普通の値（dataclassの属性として直接アクセス） 記述子（クラスアクセス = 列式、インスタンスアクセス = 行値）
制約の宣言 metadata={"choices": ..., "min": ..., "max": ...} Field(choices=..., ge=..., le=..., max_length=...)
共有基底 検証エンジン + 制約辞書 + 型カテゴリ登録表（同一のもの） 同左

経験則として：「モジュールがどのように動作するか」は構成クラスを使い、「ユーザーがどのようなデータを生成したか」はモデルを使う。

辺界と注意事項

自動マイグレーション（フェーズ2）

create_table() は、テーブルが既に存在する場合、既存の列とモデルフィールドを自動的に比較します。追加されたフィールドは、ALTER TABLE ADD COLUMN（マイグレーションで NOT NULL 制約を削除し、既存の行に NULL を挿入）を自動的に実行し、index=True を宣言した新しいフィールドは同期的にインデックスが作成されます。手動でマイグレーションスクリプトを書く必要はありません。

# v1リリース後にモデルを進化させ、email / bioフィールドを追加
class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0)
    email: str = Field(default="")     # 新規追加：次回の create_table() で自動的に ADD COLUMN が実行される
    bio: str = Field(default="")

await User.create_table()              # 再帰的：追加された列のみマイグレーションが実行される

制限事項：追加列のみをサポートしています。主キーの変更、列の型の変更、列の削除は手動で処理する必要があります（破壊的な ALTER による誤操作を避けるため）。

外部キー（リレーションマッピングの基礎）

foreign_key="テーブル.列" は、列レベルの外部キー制約を宣言し、DDL 生成時に REFERENCES 子句を生成します。

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")

    content: str = Field(max_length=255)

関係マッピング（relationship）

モデルクラス内で relationship() をクラス属性として宣言することで、外キー列がどのテーブルに宣言されているかによって自動的に方向が判定されます：

from ErisPulse.Core.Bases import Model, Field, relationship

class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)

    posts = relationship("Post", foreign_key="author")   # 外キーが相手のテーブルにある → has-many

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")
    content: str = Field(max_length=255)

    writer = relationship("User", foreign_key="author")  # 外キーがこのテーブルにある → belongs-to

has-many：インスタンス属性はクエリセットを返し、QuerySet の全チェイン機能が利用可能で、create は自動的に本テーブルの主キーを相手の外キー列に埋め込みます：

alice = await User.get(name="Alice")

posts = await alice.posts.all()                          # このユーザーの全記事
latest = await alice.posts.order_by("-id").first()
total = await alice.posts.count()
hot = await alice.posts.where(Post.content != "").all()  # 相手のテーブルのフィールド条件を追加
await alice.posts.delete()                               # このユーザーのものだけ削除

new_post = await alice.posts.create(content="hi")        # author は自動的に alice.id に設定

belongs-to：await を直接実行すると相手のインスタンスが得られます（一致するものがない場合や外キーが NULL の場合は None を返します）：

post = await Post.get(id=1)
writer = await post.writer          # User インスタンスまたは None
await writer.posts.count()          # 双方向にアクセス可能

ポイント：

次のステップ

データベースの変更やクラウドデータベースへのデプロイが必要な場合

テーブル構造は不要で、いくつかのキーと値だけを保存したい場合

多轮对话

来源: docs/ja/advanced/conversation.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/advanced/conversation.md）

Conversation 多輪対話

Tip

このドキュメントを読む必要があるのはいつですか：多輪対話のステートマシンとチェックポイントの復元が必要な場合です。必要ない場合はスキップしても、他の機能の使用には影響しません。

1 つのコマンドで複数の質問をユーザーに連続して求める必要がある場合（登録時の情報収集、アンケート、設定のガイドなど）、Conversation を使用して、同じセッション内で「質問 → 回答を待つ → 検証 → 次のステップ」をつなげます。タイムアウトや再起動後も、チェックポイントから復元できます。

対話の作成

Event オブジェクトの conversation() メソッドを使用して作成します：

from ErisPulse.Core.Event import command

@command("quiz")
async def quiz_handler(event):
    conv = event.conversation(timeout=30)

    await conv.say("🎮 知識クイズへようこそ！")

    answer = await conv.choose("第1問：Pythonの生みの親は誰ですか？", [
        "Guido van Rossum",
        "James Gosling",
        "Dennis Ritchie",
    ])

    if answer is None:
        await conv.say("タイムアウトしました。また次回お試しください！")
        return

    if answer == 0:
        await conv.say("正解です！")
    else:
        await conv.say("不正解です。正解は Guido van Rossum です")

    conv.stop()

コア API

say(content, **kwargs)

メッセージを送信し、self を返してメソッドチェーンを可能にします：

await conv.say("1行目").say("2行目").say("3行目")

送信方法を指定することもできます：

await conv.say("https://example.com/image.jpg", method="Image")

wait(prompt=None, timeout=None)

ユーザーからの返信を待機し、Event オブジェクトまたは None（タイムアウト）を返します：

# 単純な待機
resp = await conv.wait()
if resp:
    text = resp.get_text()

# プロンプトを送信して待機
resp = await conv.wait(prompt="名前を入力してください：")

# カスタムタイムアウトを使用（対話のデフォルトタイムアウトを上書き）
resp = await conv.wait(prompt="10秒以内に返信してください：", timeout=10)

confirm(prompt=None, **kwargs)

ユーザーの確認（はい/いいえ）を待機し、True / False / None（タイムアウト）を返します：

result = await conv.confirm("すべてのデータを削除してもよろしいですか？")
if result is True:
    await conv.say("削除しました")
elif result is False:
    await conv.say("キャンセルしました")
else:
    await conv.say("タイムアウトしました")

内蔵の確認用語：はい/yes/y/確認/確定/好/ok/true/対/うん/行/同意/問題ない/可能/当然...

内蔵の否定用語：否/no/n/キャンセル/不/不要/行かない/cancel/false/間違った/間違った/別/拒否...

choose(prompt, options, **kwargs)

ユーザーが選択肢から選ぶのを待機し、選択肢のインデックス（0ベース）または None を返します：

choice = await conv.choose("色を選択してください：", ["赤", "緑", "青"])
if choice is not None:
    colors = ["赤", "緑", "青"]
    await conv.say(f"選択した色は {colors[choice]} です")

ユーザーは番号（1/2/3）または選択肢のテキスト（赤）を入力して選択できます。

options_format="auto"（デフォルト）は、method に応じて自動的に内蔵のスタイルを選択します：Markdown→無序リスト、Html→順序付きリスト、その他→純粋なテキストリスト。 また、"list"、"inline"、"md"、"html"、またはカスタム関数もサポートします。

merge_prompt=True を使用して、プロンプトと選択肢を1つのメッセージに統合することもできます。また、オプション挿入位置を制御するプレースホルダ（デフォルトは {options}、placeholder でカスタマイズ可能）もサポートします：

choice = await conv.choose(
    "## 選択してください\n{options}",
    ["オプションA", "オプションB"],
    method="Markdown",
    merge_prompt=True,
)

# カスタムプレースホルダ
choice = await conv.choose(
    "選択してください: [choices]",
    ["オプションA", "オプションB"],
    placeholder="[choices]",
)

collect(fields, **kwargs)

複数ステップで情報を収集し、データ辞書または None を返します：

data = await conv.collect([
    {"key": "name", "prompt": "名前を入力してください"},
    {"key": "age", "prompt": "年齢を入力してください",
     "validator": lambda e: e.get("alt_message", "").strip().isdigit(),
     "retry_prompt": "年齢は数字でなければなりません。再度入力してください"},
    {"key": "city", "prompt": "都市を入力してください"},
])

if data:
    await conv.say(f"登録完了！\n名前: {data['name']}\n年齢: {data['age']}\n都市: {data['city']}")
else:
    await conv.say("登録が中断されました")

フィールドの設定：

パラメータ 説明 デフォルト値
key フィールドのキー名（必須） -
prompt プロンプトメッセージ "{key} を入力してください"
validator 関数、Event を受け取り、bool を返す なし
retry_prompt 検証失敗時の再試行プロンプト "入力が無効です。再度入力してください"
max_retries 最大再試行回数 3
condition 関数、既に収集されたデータの辞書を受け取り、bool を返す なし

条件付きフィールド：condition を使用して、条件が満たされた場合にのみフィールドを収集する動的フォームを実現できます：

data = await conv.collect([
    {"key": "has_car", "prompt": "車をお持ちですか？（はい/いいえ）"},
    {"key": "car_brand", "prompt": "車のブランドを入力してください",
     "condition": lambda d: d.get("has_car", "").lower() in ("はい", "yes", "y")},
])

stop()

手動で対話を終了し、is_active を False に設定します：

conv.stop()

is_active

対話がアクティブかどうかを返します：

if conv.is_active:
    await conv.say("対話はまだ進行中です")

アクティブ状態の管理

stateDiagram-v2
    state "アクティブ" as active
    state "非アクティブ" as inactive
    [*] --> active: event.conversation()
    active --> active: say / wait / confirm / choose / collect
    active --> inactive: stop()
    active --> inactive: wait() タイムアウト
    active --> inactive: collect() タイムアウトまたは再試行回数超過
    inactive --> [*]

以下の状況で対話は自動的に非アクティブになります：

  1. stop() メソッドを呼び出す
  2. wait() がタイムアウトして None を返す
  3. collect() がステップのタイムアウトまたは再試行回数超過で None を返す

非アクティブになった後、wait/confirm/choose/collect のすべてのインタラクションメソッドは即座に None を返し、ユーザーからの入力を待続しません。

分岐とジャンプ

@conv.branch(name) デコレータ

branch() を使用して対話の分岐を登録し、goto() で分岐間をジャンプします：

@command("menu")
async def menu_handler(event):
    conv = event.conversation(timeout=60)

    @conv.branch("main")
    async def main_menu():
        await conv.say("=== メインメニュー ===\n1. 本人情報\n2. 設定\n3. 終了")
        resp = await conv.wait()
        if resp is None:
            return
        text = resp.get_text().strip()
        if text == "1":
            await conv.goto("profile")
        elif text == "2":
            await conv.goto("settings")
        elif text == "3":
            await conv.say("さようなら！")
            conv.stop()

    @conv.branch("profile")
    async def profile():
        await conv.say("=== 本人情報 ===\n名前: Alice\n0. 戻る")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    @conv.branch("settings")
    async def settings():
        await conv.say("=== 設定 ===\n1. 通知のオン/オフ\n0. 戻る")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    await conv.start()  # 最初に登録された分岐から開始

conv.start(name=None)

対話を開始し、デフォルトでは最初に登録された分岐から開始します：

await conv.start()          # 最初の分岐から開始
await conv.start("settings") # 指定された分岐から開始

コンテキストと永続化

conv.context

各対話インスタンスには、分岐間で状態を共有するための context 辞書が内蔵されています：

@conv.branch("step1")
async def step1():
    conv.context["username"] = resp.get_text().strip()
    await conv.goto("step2")

@conv.branch("step2")
async def step2():
    name = conv.context.get("username", "未知")
    await conv.say(f"こんにちは、{name}！")

save() / resume() / clear_saved()

対話は永続化をサポートし、タイムアウトや中断後に復元できます：

# 対話状態を保存する（通常は手動で呼び出す必要はありません、下記の「自動チェックポイント」を参照）
await conv.save()

# ... その後、同じセッションで復元する ...
conv2 = event.conversation()
if await conv2.resume():
    await conv2.say("お帰りなさい！以前の対話を続けましょう")
else:
    await conv2.say("以前の対話は見つかりませんでした")

# 保存された対話を削除する
await conv.clear_saved()

ストレージキーにはターゲットの次元が含まれます（conversation:{platform}:{user_id}:{target_id}）、同じユーザーの異なるセッション間の対話は互いに上書きされません。resume() 時に、ターゲットを含まない旧形式のアーカイブは自動的に移行されます。

自動チェックポイントと再起動時の復元

自動アーカイブ

フレームワークは以下のタイミングで自動的にチェックポイントを管理します。通常、save() を手動で呼び出す必要はありません：

時機 行動
goto() / start() で分岐をジャンプする 自動保存（現在の分岐 + context）
stop() / wait() タイムアウト / collect() 失敗 自動削除（対話の終端状態）

チェックポイント TTL

アーカイブにはタイムスタンプが付いており、ErisPulse.interaction.checkpoint_ttl（デフォルト 24 時間）を超えるアーカイブはクリーンアップされます：

[ErisPulse.interaction]
checkpoint_ttl = 86400  # 秒

再起動時の自動復元

フレームワークが再起動した後、メモリ内の待機中のコルーチンは失われますが、チェックポイントは残ります。register_resume_handler を使用して復元ファクトリを登録することで、フレームワークは再起動後にその会話の最初のメッセージを受け取ったときに自動的に対話を継続します：

from ErisPulse.Core.Event.wrapper import Conversation

@Conversation.register_resume_handler()  # platform="onebot11" を渡すことでプラットフォームを限定することも可能
def make_conversation(event) -> Conversation:
    # ファクトリの役割：対話を再構築し、すべての分岐を再登録する
    conv = event.conversation(timeout=60)

    @conv.branch("menu")
    async def menu(conv, event):
        ...

    return conv

登録後、menu 分岐にあったユーザーが再起動前に最初のメッセージを送信した場合、フレームワークは自動的に：コンテキストを復元 → そのメッセージを認証 → 保存された分岐から対話を継続します。ファクトリを登録していない場合、このメカニズムは無駄なコストがかかりません。

復元は即座に引き継ぎ

resume() が成功した場合、フレームワークは自動的に2つのことを完了します：

  1. セッションの引き継ぎ：自動的にそのセッションの排他リースを取得します——他のモジュールは sdk.interaction.get_owner_of(event) を使用して「このユーザーが対話で占有されている」ことを感知できます；セッションが他のモジュールによって占有されている場合、復元は失敗（False を返す）し、2つの対話が競合することを防ぎます
  2. 履歴の持ち込み：会話の受信箱から最近の10件のメッセージを conv.recent_history に取り込みます（AI モジュールが復元された後、LLM のコンテキストが途切れません）；resume(with_history=0) でこの機能をオフにできます
if await conv.resume(with_history=20):
    for m in conv.recent_history:
        print(m["role"], ":", m["text"])

手動復元（自動メカニズムを使用しない場合）

@command("continue")
async def continue_handler(event):
    conv = event.conversation()
    # ... 分岐を登録する ...
    if await conv.resume():
        conv.goto(conv.get_current_branch())

代表的なフロー・パターン

ガイド付き登録

@command("register")
async def register_handler(event):
    conv = event.conversation(timeout=60)

    await conv.say("ようこそ、登録へ！")

    data = await conv.collect([
        {"key": "username", "prompt": "ユーザー名を入力してください（3-20文字）",
         "validator": lambda e: 3 <= len(e.get_text().strip()) <= 20},
        {"key": "email", "prompt": "メールアドレスを入力してください",
         "validator": lambda e: "@" in e.get_text() and "." in e.get_text(),
         "retry_prompt": "メールアドレスの形式が正しくありません。再度入力してください"},
    ])

    if not data:
        await event.reply("登録がキャンセルされました")
        return

    confirmed = await conv.confirm(
        f"登録情報を確認しますか？\nユーザー名: {data['username']}\nメールアドレス: {data['email']}"
    )

    if confirmed:
        await conv.say("✅ 登録完了！")
    else:
        await conv.say("❌ 登録がキャンセルされました")

ループ対話

@command("chat")
async def chat_handler(event):
    conv = event.conversation(timeout=120)
    await conv.say("対話モードに入りました。「終了」で終了します")

    while conv.is_active:
        resp = await conv.wait()
        if resp is None:
            await conv.say("タイムアウトしました。対話は終了します")
            break

        text = resp.get_text().strip()

        if text == "終了":
            await conv.say("さようなら！")
            conv.stop()
        elif text == "help":
            await conv.say("利用可能なコマンド：終了、help、status")
        elif text == "status":
            await conv.say("対話はアクティブです")
        else:
            await conv.say(f"あなたが言ったのは：{text}")

関連ドキュメント

交互会话

来源: docs/ja/advanced/interaction.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ja/advanced/interaction.md）

交互会話システム

Tip

この記事を読むのはいつか：wait_reply、多輪対話、定時リマインダー、セッション排他、およびトレースIDが必要な場合。不要な場合は飛ばしても、他の機能の使用には影響しません。

この記事を読んだ後でわかる：wait_replyの全パラメータの使い方とキャンセル理由、.got()による宣言的多輪対話チェーン、 セッションタイマー（remind / escalate）、多路待ち（expect / select）とセッション排他リース。

この記事で取り扱わない：多輪対話の分岐ステートマシンと再開/復元（Conversation 多輪対話を参照）； 受信箱の保持戦略のスイッチと実行時の上書き（会話受信箱を参照）。

Note

本章の内容はErisPulse 2.8.0+ が必要です。

ErisPulseは「ユーザーとの継続的な対話」をフレームワークレベルのインフラとして実装しています。1つのwait_replyから、定時リマインダー、多路待ち、セッション排他、再起動時の復元まで、すべては統一された インタラクティブセッションマネージャー（Core/Event/interaction.py、sdk.interaction）によってスケジュールされます。

{!--< tips >!--} 本文で取り扱う各機能には所有者（owner）が付いています：インタラクティブ待ち、リース、タイマーはすべて登録時のモジュール名を記録し、モジュールのアンロードまたはアダプターの停止時にフレームワークが自動的にクリーンアップし、待機側は即座に通知を受け取ります。これは、所有権システムがインタラクションの次元に拡張されたものです（所有権システムを参照）。 {!--< /tips >!--}

レプライ待ち：wait_reply

一言で：現在のコルーチンを中断して、対象ユーザーからの次のメッセージを待機し、一致すれば返信イベントを返し、タイムアウトまたはキャンセルされた場合はNoneを返します。pattern / regex / validatorで不適切な返信をフィルタリングできます。

wait_replyはインタラクティブセッションの基盤です。現在のコルーチンを中断し、ユーザーが次のメッセージで「返信」することを待ちます。

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event):
    reply = await event.wait_reply(prompt="あなたの名前を入力してください:", timeout=30)
    if reply is None:
        await event.reply("タイムアウトしました")
        return
    await event.reply(f"こんにちは、{reply.get_text()}さん！")

全パラメータ一覧

パラメータ 説明 デフォルト
prompt 待機前に送信するプロンプトメッセージ None
timeout 待機のタイムアウト（秒） 60
pattern globフィルター（* / ? / [seq]）、一致しない場合は待機を続ける None
regex 正規表現フィルター（patternと同時に指定する場合は両方一致する必要がある）、一致しない場合は待機を続ける None
validator 検証関数（Eventを受け取り、boolを返す）、失敗した場合は待機を続ける None
callback レプライを受け取ったときのコールバック（戻り値方式の代わりの書き方） None
method promptの送信方法 "Text"
session セッションレベルの待ち：同じセッション（グループ/チャンネル）内の誰の返信でも一致する False 
# 数値金額のみを受け入れ、そうでなければ待機を続ける
reply = await event.wait_reply("金額を入力してください:", regex=r"\d+\s*元", timeout=30)

# セッションレベルの待ち：グループ協力の場面で、グループ内の誰かが返信してもOK
reply = await event.wait_reply(session=True, prompt="誰かが答えてくれますか？")

待ちとコマンドの境界（コマンド透過）

待機中に登録済みのコマンド（例：/cancel）が送信された場合、デフォルトではコマンドディスパッチャーに渡され、 レプライとして消費されません（設定項目ErisPulse.event.wait_reply.cmdpass、Trueにすると逆転する— メッセージはすべてレプライとして扱われる）。判定のテキスト口径はコマンドディスパッチャーと同じ：alt_messageが優先され、 存在しない場合はメッセージセグメントの結合（2.10.1以降、純粋なmessageセグメントアダプターでも同様に有効）。

待ちがいつキャンセルされるか

待ちが「タイムアウトするまで待つ」だけではない—以下の状況で待ちが即座に終了（wait_replyがNoneを返す）し、 呼び出し側がタイムアウトまで待つ必要がなくなります。

触発 キャンセル理由（InteractionCancelled.reason） 説明
所有モジュールがアンロード / 禁止 owner_unload 所有権のクリーンアップ：誰が登録した待ちも、誰が消えた時に一緒に回収される
アダプターが停止 / 再起動 platform_stop そのプラットフォームで中断された待ちはすべてキャンセルされる
同セッションで新しい待ち / リースに置き換えられた conflict 下記「セッション仲裁」を参照
レプライ者がブロックされた / 所有モジュールが解除された revoked レプライが一致した権限再確認：scopeの身元次元 + モジュール次元
ユーザーのレプライが一致した —— 正常経路、レプライイベントを返す

下層の例外はInteractionCancelled（InteractionError例外体系に属する）で、 wait_replyはそれをNoneを返すように変換しています。理由が必要な呼び出し側は、 sdk.interaction.register()の低層APIを使用できます。

レプライが一致する完全な判定チェーン

レプライメッセージが到着した際、インタラクションマネージャーは以下の順序で判定します（コマンドマッチの前に実行され、 対話の連続性が優先される—たとえメッセージが他の高優先度のプロセッサーに認知されたとしても、中断された対話は完了します）：

セッションキーが一致する（正確なuser次元 → セッションレベルの戻りにフォールバック）
  → pattern / regexテキストフィルター（一致しない場合は待機を続ける）
  → validator検証（失敗した場合は待機を続ける）
  → 権限再確認（scope身元次元 + ownerモジュール次元、失敗した場合は待ちを終了）
  → 待ち側をウォーキング + イベントを認領（mark_processed）

宣言的多輪対話：.got()チェーン（2.11+）

@commandは多輪対話コントローラーのプロキシを返し、.got()を使って段階的に収集ステップを宣言します。ステップ関数はパラメータ名に従って収集された値を受け取ります（コマンドエントリのパラメータも含む）、型はtype= / 注釈に従って自動解析されます。下層ではConversationにコンパイルされ、各ステップが1つの分岐、ジャンプは自動的にチェックポイントを保存します。再起動後、ユーザーのレプライは保存されたステップから継続されます（復元工場は装飾時に自動的に登録され、モジュールのアンロードに伴って自動的に登録解除されます）。

@command("weather")
async def weather(event):
    pass  # エントリ：最初の.gotのプロンプトが自動的に送信され、手動で質問する必要はありません

@weather.got("city", prompt="都市名を直接返信してください")
async def weather_city(event, city: str):
    await event.reply(f"{city}：晴れ、25℃")

@weather.got("days", prompt="何日間の予報ですか？", type=int, retry_prompt="数字を入力してください")
async def weather_days(event, city: str, days: int):
    await event.reply(f"{city} 未来 {days} 日間……")

パラメータ説明

パラメータ 説明
name 収集値の名前（後続のステップ関数のパラメータ名として注入される；プロセス内で一意、重複宣言は登録時にエラーを出す）
prompt このステップに入るときに自動的に送信される質問
timeout 単一ステップの待機タイムアウト（秒数または"90s" / "1h30m"の時長文字列、デフォルトは60s）
type 値の型（int / float / str…；解析失敗した場合はretry_promptを再送信し、最大max_retries回まで）
validator / pattern / regex レプライのフィルタリングと検証（wait_replyと同様の意味）
session セッションレベルの待ち（同じセッション内の誰でも一致する）
cmdpass コマンド通過三態（デフォルトでは/cancelなどのコマンドは待機中に通常通り実行される）

意味と境界

セッションタイマー：remind / escalate

「タイムアウト」を戻り値から編集可能な原語にします。タイマーはインタラクティブセッションに紐づき、 モジュールのアンロード / アダプターの停止に伴って自動的にキャンセルされます。1セッションあたりのアクティブremindの上限は5個です。

remind：返信がなければリマインダー

@command("ticket")
async def ticket_command(event):
    await event.reply("チケットを提出しました。処理結果はここに通知されます")
    # 5分間返信がなければ優しくリマインダーを送信；ユーザーの返信はすべて自動的にキャンセルされます
    event.remind(300, "まだいますか？結果が出たらすぐにご連絡します")
    reply = await event.wait_reply(timeout=3600)
    ...

escalate：期限に必ず届くアップグレード

event.escalate(1800, lambda e: notify_master(f"チケット30分未処理：{event.get_command_args()}"))

remindとの唯一の違いは：ユーザーの返信でキャンセルされない—アップグレードアクション（管理者に通知、人間への転送） は「タイムアウトに必ず届く」約束であり、cancel() / モジュールのアンロード / アダプターの停止でのみキャンセルされます。

remind escalate
到期行動 テキストを送信 / callbackを実行 callbackを実行
ユーザーの返信 自動的にキャンセル 影響を受けない
所有権のクリーンアップ（アンロード / プラットフォーム停止） キャンセル キャンセル
セッションごとの上限 5 なし（所有権のクリーンアップで補完）

多路待ち：expect + select

一言で：expectは期待を記述するだけで登録はせず、selectは複数の期待を統一して待機し、いずれかが先に一致したら返す (インデックス, レプライイベント)、一致しなかったら自動的にクリーンアップ。

複数の期待を同時に待機し、先着順—典型的な場面：管理者の承認を待っている間にユーザーの撤回を待つ、 多人数の協力による投票。

which, reply = await event.select(
    event.expect(pattern="同意*", user="10001"),
    event.expect(pattern="拒否*", user="10002"),
    event.expect(validator=lambda e: e.get_text() == "搁置", session=True),
    timeout=60,
)
if which is None:
    await event.reply("60秒以内に承認結果がありませんでした")
elif which == 0:
    await event.reply("承認しました")
elif which == 1:
    await event.reply("拒否しました")

{!--< tips >!--} selectは多スレッドasyncio.waitによる手動編集と比べて：期待が一致しなかった場合の自動クリーンアップ、 一致したイベントの自動認領、権限再確認と所有権のクリーンアップがすべて有効—Futureを自分で管理する必要はありません。 {!--< /tips >!--}

Note

2.10.1以降、selectは集約待ち：すべての期待を1つの項目に統合し、独自の一致判定を行う、 同キーの多路待ち（同じuserに異なるpatternを2つ設定するなど）は互いにキャンセルされず、 いずれかのルートが先に一致したら返す。返信テキストの判定口径はalt_messageが優先され、 メッセージセグメントが戻る（コマンドディスパッチャーと同じ）。

境界説明：event.wait_for(...)はアダプターイベントの総線に直接接続され、中間件や detail_type条件フィルタリング（身元認証はディスパッチャーのエントリでカバー）は経由しない。 完全なパイプラインが必要な待ちの場合はwait_reply / selectを使用してください。

セッション排他：acquire / hold / get_owner_of

所有権が「リソース」から「セッション」へと移行—「このユーザーが現在誰に占有されているか」が第一のクエリになります。

# クエリ：このセッションは誰と対話中ですか？（空きならNoneを返す）
owner = sdk.interaction.get_owner_of(event)
if owner and owner != "MyModule":
    return  # 他のモジュールが対話中なので、干渉しない

# 排他リース：セッションを独占（deny戦略、占有中ならNoneを返す）
lease = sdk.interaction.acquire(event)          # デフォルトTTLは1時間、ttl=で渡すことも可能
if lease is None:
    return  # 既に占有されています
try:
    ...  # 独占対話
finally:
    lease.release()

コンテキストマネージャー形式（取得失敗時はSessionOccupiedErrorを送出）：

with sdk.interaction.hold(event) as lease:
    ...  # 終了時に自動リリース

リースはrenew(ttl)で更新可能；TTLは惰性で期限切れ—期限切れのリースは次回アクセス時に自動的にクリーンアップされます。

Conversation.resume()が対話を復元する際、フレームワークは自動的にリースを取得します（Conversation 多輪対話の「復元即占有」を参照）— 復元された対話は天然にセッションを保持し、他のモジュールは挿入できません。

セッション受信箱：event.history

保持戦略（7日間 / 条数上限）と実行時の上書きAPI、リスクと監査はセッション受信箱の専門記事を参照してください。

各セッションの最近のメッセージストリームを統一的に記録（ユーザーとロボットの両方）し、AIコンテキスト、 重複防止、行動分析系モジュールの共有事実のベースとして使用します—各モジュールは個別に履歴を保存する必要がありません。

messages = await event.history(20)   # 現在のセッションで最近20件、時系列昇順
for m in messages:
    print(m["role"], ":", m["text"])  # role: "user" / "bot"

メッセージトランザクション：message_tx

トランザクション内のすべての出力送信は自動的に記帳されます。例外が発生した場合、逆順に自動的に送信済みメッセージを撤回（アダプターがdelete_messageを実装していない場合はスキップし、トランザクションログは正常に記録される）。

async with event.message_tx():
    await event.reply("処理中です、少々お待ちください")
    result = await do_something()          # ここで例外が発生 →
    await event.reply(f"完了: {result}")   # 以前の「処理中」が自動的に撤回される

トランザクション外で送信されたものは記帳されません（ゼロオーバーヘッド）；get_send_receipts()で現在のトランザクションで送信されたレシートを確認できます。

トレースID：trace-id

各入力イベントには自動的にトレースID（event["id"]を再利用、存在しない場合は生成）が割り当てられ、以下を貫きます：

1つのメッセージが複数のモジュールによって処理された場合、全経路で同一のIDで連携できます（ログ / スロークエリ / 監査）。

他のシステムとの関係

関連文書