プラグインインスタンスとライフサイクル

すべてのNekro Agentプラグインのコアは、NekroPluginクラスのインスタンスです。このインスタンスは、プラグインの基本情報を定義するだけでなく、さまざまな機能やイベントコールバックのマウントポイントとしても機能します。プラグインインスタンスの作成とその完全なライフサイクルを理解することは、安定した機能豊富なプラグインを開発するために重要です。

NekroPluginインスタンス

プラグインのメインファイル（例：plugin.py）で、まずNekroPluginインスタンスをインポートして作成する必要があります：

from nekro_agent.api.plugin import NekroPlugin

plugin = NekroPlugin(
    name="My Cool Plugin",
    module_name="my_cool_plugin",
    description="This plugin can do many cool things.",
    version="1.0.0",
    author="DeveloperName",
    url="https://github.com/developer/my_cool_plugin",
    support_adapter=["onebot_v11", "telegram"],  # オプション、サポートするアダプターのリスト
    is_builtin=False,  # サードパーティプラグインの場合は通常False
    is_package=False   # プラグインに複雑なパッケージ構造がある場合は調整が必要な場合があります
)

コアパラメータ分析：

• name (str): プラグインの表示名で、Nekro Agentユーザーインターフェースに表示されます。
• module_name (str): プラグインのモジュール名で、Pythonモジュールの有効な名前である必要があり、すべてのプラグイン間で一意である必要があります。通常、プラグインのディレクトリ名と一致します。プラグインの内部識別と管理に使用されます。
• description (str): プラグインの簡単な説明で、その主な機能を説明します。
• version (str): プラグインのバージョン番号で、セマンティックバージョニング（例：1.0.0）に従うことが推奨されます。
• author (str): プラグインの作者または開発チームの名前で、英字、数字、アンダースコアで構成され、数字で始めることはできません。
• url (str): プラグインのプロジェクトリポジトリ、ドキュメント、またはホームページを指すURL。
• support_adapter (List[str], オプション): プラグインがサポートするアダプターのリストで、例：["onebot_v11", "telegram"]。指定しない場合、デフォルトですべてのアダプターがサポートされます。
• is_builtin (bool, オプション, デフォルトFalse): プラグインがNekro Agentの組み込みプラグインであるかどうかをマークします。
• is_package (bool, オプション, デフォルトFalse): プラグインがPythonパッケージとして管理されるかどうかをマークします。

NekroPluginインスタンス（plugin）は、すべての後続のデコレータの呼び出しオブジェクトとして機能し、サンドボックスメソッド、設定、ライフサイクルコールバックなどをマウントするために使用されます。

プラグインライフサイクル

プラグインライフサイクルは、プラグインがシステムによって検出され、ロードされ、実行され、最終的にアンロードされるまでのプロセス全体を説明します。Nekro Agentは一連のデコレータを提供し、ライフサイクルの特定の段階でカスタムロジックを実行できるようにします。

1. 初期化（@plugin.mount_init_method()）

Nekro Agentが最初にプラグインをロードするとき（通常、エージェントの起動時またはプラグインが動的に有効にされたとき）、@plugin.mount_init_method()デコレータを通じて登録された非同期関数が実行されます。この段階は、一度だけ行う必要があるセットアップ作業を実行するのに理想的です。

使用例：

• リソース準備：データベースを作成または接続し、ファイルシステム（プラグインが必要とするデータディレクトリの作成など）を初期化します。
• 状態初期化：プラグインの初期状態を設定し、デフォルトの設定またはデータをロードします。
• 外部システム接続：プラグインが外部サービスと対話する必要がある場合、ここで接続を確立できます。
• 環境検証：必要な依存関係または環境変数が満たされているかどうかを確認します。

例：

from nekro_agent.api import core
from qdrant_client import models as qdrant_models # qdrantの使用を想定

@plugin.mount_init_method()
async def initialize_plugin():
    core.logger.info(f"Plugin '{plugin.name}' is initializing...")
    # 例：ベクトルデータベースコレクションを初期化
    try:
        client = await core.get_qdrant_client() # Qdrantクライアントを取得
        collection_name = plugin.get_vector_collection_name("my_data") # プラグインの一意のコレクション名を生成

        collections = await client.get_collections()
        collection_names = [collection.name for collection in collections.collections]

        if collection_name not in collection_names:
            core.logger.info(f"Creating vector database collection: {collection_name}")
            await client.create_collection(
                collection_name=collection_name,
                vectors_config=qdrant_models.VectorParams(
                    size=768,  # 例の次元、埋め込みモデルに応じて調整
                    distance=qdrant_models.Distance.COSINE,
                ),
            )
            core.logger.success(f"Collection {collection_name} created successfully")
        else:
            core.logger.info(f"Collection {collection_name} already exists")
    except Exception as e:
        core.logger.error(f"Plugin '{plugin.name}' failed to initialize vector database: {e}")
    core.logger.success(f"Plugin '{plugin.name}' initialization complete.")

注：初期化メソッドは非同期（async def）である必要があります。初期化中にエラーが発生した場合、エージェントが実行を継続できるように、適切に処理し、ログに記録する必要があります。

2. セッションリセットコールバック（@plugin.mount_on_channel_reset()）

特定のチャットセッションがリセットされたとき（例：ユーザーがチャット履歴またはセッション状態をクリアしたとき）、@plugin.mount_on_channel_reset()デコレータを通じて登録された非同期関数が呼び出されます。このコールバックは、リセットされたセッションに関する情報を含むAgentCtxオブジェクトを受け取ります。

使用例：

• 特定のセッションに関連するプラグインデータ（キャッシュ、状態など）をクリーンアップします。
• そのセッションのプラグイン固有の設定をリセットします。

例：

from nekro_agent.api.schemas import AgentCtx

@plugin.mount_on_channel_reset()
async def handle_channel_reset(_ctx: AgentCtx):
    core.logger.info(f"Plugin '{plugin.name}' received session {_ctx.from_chat_key} reset event.")
    # 例：このセッションのプラグイン固有のキャッシュをクリア
    # await plugin.store.delete(chat_key=ctx.from_chat_key, store_key="session_cache")
    core.logger.info(f"Plugin-specific data for session {_ctx.from_chat_key} has been cleaned up.")

注：このコールバックの誤用は、エージェントのパフォーマンスまたは正常な会話フローに影響を与える可能性があります。ほとんどのプラグイン機能は、サンドボックスメソッドを通じて実装する必要があります。

3. ユーザーメッセージコールバック（@plugin.mount_on_user_message()）

ユーザーがメッセージを送信すると、@plugin.mount_on_user_message()デコレータを通じて登録された非同期関数が呼び出されます。このコールバックは、ユーザーメッセージをリッスン、前処理、またはインターセプトするために使用できます。

使用例：

• メッセージ前処理：AI処理の前にユーザーメッセージを処理、フィルタリング、または強化します。
• アクセス制御：ユーザーIDまたはメッセージ内容に基づいてアクセス制御を実行します。
• メッセージ統計：ユーザーメッセージの使用量または行動分析を記録します。
• 自動返信：特定のタイプのメッセージに即座に応答します。

例：

from nekro_agent.api.message import ChatMessage
from nekro_agent.api.signal import MsgSignal

@plugin.mount_on_user_message()
async def handle_user_message(_ctx: AgentCtx, message: ChatMessage) -> MsgSignal | None:
    """ユーザーメッセージを処理するためのコールバック関数"""

    # 例：メッセージ内容フィルタリング
    if "forbidden_word" in message.content_text:
        core.logger.warning(f"User {_ctx.from_user_id} sent a message containing forbidden words")
        return MsgSignal.BLOCK_ALL  # メッセージが履歴に記録されるのをブロックし、後続の処理を防止します

    # 例：特殊コマンド処理
    if message.content_text.startswith("!plugin:"):
        command = message.content_text[8:].strip()
        await _ctx.send_text(f"Plugin command received: {command}")
        return MsgSignal.BLOCK_TRIGGER  # メッセージが記録されることを許可しますが、AI処理のトリガーを防止します

    # 例：メッセージ統計
    await plugin.store.set(
        user_key=_ctx.from_user_id,
        store_key="message_count",
        value=str(int(await plugin.store.get(user_key=_ctx.from_user_id, store_key="message_count") or "0") + 1)
    )

    # 通常のメッセージ処理を示すためにNoneを返します
    return None

戻り値の説明：

• NoneまたはMsgSignal.CONTINUE：メッセージは正常に処理され、後続のプロセスに渡され続けます
• MsgSignal.BLOCK_TRIGGER：メッセージが履歴に記録されることを許可しますが、メッセージが後続の処理をトリガーするのを防止します
• MsgSignal.BLOCK_ALL：メッセージが履歴に記録されるのをブロックし、メッセージが後続の処理をトリガーするのも防止します

4. システムメッセージコールバック（@plugin.mount_on_system_message()）

システムがメッセージを送信するとき（管理者コマンド、システム通知など）、@plugin.mount_on_system_message()デコレータを通じて登録された非同期関数が呼び出されます。

使用例：

• システムイベント監視：システムレベルのメッセージとイベントを監視します。
• 管理機能統合：管理者コマンドまたはシステム状態の変更に応答します。
• ロギング：重要なシステムメッセージを記録します。

例：

@plugin.mount_on_system_message()
async def handle_system_message(_ctx: AgentCtx, message: str) -> MsgSignal | None:
    """システムメッセージを処理するためのコールバック関数"""

    core.logger.info(f"Plugin '{plugin.name}' received admin command: {admin_command}")

    return None

5. プロンプトインジェクション（@plugin.mount_prompt_inject_method()）

プロンプトインジェクションにより、プラグインはAIがユーザーメッセージを処理する前にシステムプロンプトに動的にコンテンツを追加できます。これは、AIの動作に影響を与え、コンテキスト情報を提供するための重要なメカニズムです。

使用例：

• コンテキスト強化：現在のセッション状態情報、ユーザー設定などをAIに提供します。
• 役割定義：AIに特定の役割または動作パターンを設定します。
• 機能プロンプト：AIにより詳細なプラグイン使用ガイダンスを提供します。

例：

@plugin.mount_prompt_inject_method(
    name="user_preference_injection",
    description="Inject user preferences and status information to AI"
)
async def inject_user_context(_ctx: AgentCtx) -> str:
    """ユーザーコンテキスト情報をAIプロンプトに注入する"""

    # ユーザー設定を取得
    user_preferences = await plugin.store.get(
        user_key=_ctx.from_user_id,
        store_key="preferences"
    )

    # セッション状態を取得
    session_mode = await plugin.store.get(
        chat_key=_ctx.from_chat_key,
        store_key="mode"
    )

    prompt_parts = []

    if user_preferences:
        prompt_parts.append(f"User preference settings: {user_preferences}")

    if session_mode:
        prompt_parts.append(f"Current session mode: {session_mode}")

    return "\n".join(prompt_parts)

詳細なプロンプトインジェクションメカニズムについては、プロンプトインジェクションの章を参照してください。

6. クリーンアップ（@plugin.mount_cleanup_method()）

プラグインがアンロードされるか、Nekro Agentがシャットダウンするとき、@plugin.mount_cleanup_method()デコレータを通じて登録された非同期関数が実行されます。この段階は、プラグインが占有するすべてのリソースを解放するために使用され、システムがクリーンにシャットダウンするか、プラグインをリロードすることを確実にします。

使用例：

• リソース解放：データベース接続を閉じ、ファイルハンドルを解放し、バックグラウンドタスクを停止します。
• 状態リセット：次回のロード時に新鮮な状態を確実にするために、グローバル変数またはキャッシュをクリアします。
• 外部システム切断：プラグインが外部サービスとの長い接続を維持している場合、ここで切断する必要があります。

例：

@plugin.mount_cleanup_method()
async def cleanup_plugin():
    core.logger.info(f"Plugin '{plugin.name}' is cleaning up...")
    # 例：可能なデータベース接続プールを閉じるか、一時ファイルをクリアする
    # await close_database_connections()
    # await clear_temporary_files()
    core.logger.success(f"Plugin '{plugin.name}' cleanup complete.")

重要性： 適切なクリーンアップは、プラグインの安定性と保守性にとって重要です。特に、プラグインが頻繁にリロードされる開発とデバッグ段階では。リソースが適切にクリーンアップされない場合、以下の問題が発生する可能性があります：

• リソースリーク：メモリリーク、解放されないファイルハンドルなど。
• 状態の競合：リロードされたプラグインが古いプラグインから残留状態を継承する可能性があります。
• 異常な動作：プラグインのパフォーマンスが期待を満たしません。

初期化メソッドとクリーンアップメソッドがペアで表示され、プラグインリソースと状態を適切に管理することを確認してください。

高度な機能

動的メソッドコレクションフィルター（@plugin.mount_collect_methods()）

動的メソッドコレクションフィルターにより、プラグインはコンテキスト（ユーザー権限、セッション状態など）に基づいて、現在の状態で利用可能なサンドボックスメソッドを動的に決定できます。これは、権限制御とパーソナライズされた機能を実装するのに非常に便利です。

使用例：

• 権限制御：ユーザーIDに基づいて利用可能な機能を制限します。
• 条件付き機能：セッション状態または設定に基づいて機能を動的に有効/無効にします。
• パーソナライズされた体験：異なるユーザーに異なる機能セットを提供します。

例：

@plugin.mount_collect_methods()
async def collect_available_methods(_ctx: AgentCtx) -> List[SandboxMethod]:
    """コンテキストに基づいて利用可能なメソッドを動的に収集する"""

    # セッションタイプを確認
    if _ctx.channel_type == "group":
        # グループチャット固有の機能
        return [group_method_xxx, group_method_yyy]
    else:
        return [private_method_xxx, private_method_yyy]

プラグインドキュメントサポート

NekroPluginは、プラグインドキュメントの自動読み込みと表示をサポートし、ユーザー体験を向上させます：

ドキュメント読み込みの優先順位：

1. プラグインディレクトリ内のREADME.mdファイル
2. プラグインモジュールの__doc__文字列

例：

"""
My Cool Plugin

This plugin provides the following features:
- User preference management
- Intelligent recommendation system
- Data analysis tools

## Usage

1. Configure API keys
2. Set user preferences
3. Start using plugin features

## Notes

- Ensure network connection is normal
- API keys need to be updated regularly
"""

from nekro_agent.api.plugin import NekroPlugin

plugin = NekroPlugin(
    name="My Cool Plugin",
    # ... other parameters
)

# Plugin implementation code...

ベクトルデータベースコレクションの命名

プラグインは、異なるプラグイン間のデータ分離を確保するための便利なベクトルデータベースコレクション命名メソッドを提供します：

# プラグイン固有のコレクション名を取得
collection_name = plugin.get_vector_collection_name("user_embeddings")
# Returns: "author.module_name-user_embeddings"

# プラグインのデフォルトコレクション名を取得
default_collection = plugin.get_vector_collection_name()
# Returns: "author.module_name"

プラグインストレージアクセス

各プラグインには独立した永続KVストレージがあり、plugin.store属性を通じてアクセスされます。

# 任意のプラグインメソッドでストレージを使用
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "save_data", "Save data")
async def save_data(_ctx: AgentCtx, key: str, value: str) -> str:
    await plugin.store.set(
        chat_key=_ctx.from_chat_key,
        store_key=key,
        value=value
    )
    return f"Data saved: {key} = {value}"

詳細なストレージ使用については、データ保存の章を参照してください。

プラグイン設定アクセス

プラグインは、plugin.config属性を通じて設定にアクセスできます：

@plugin.mount_sandbox_method(SandboxMethodType.AGENT, "get_status", "Get plugin status")
async def get_plugin_status(_ctx: AgentCtx) -> str:
    config = plugin.get_config()

    return f"""
    Plugin Status:
    - Name: {plugin.name}
    - Version: {plugin.version}
    - Enabled: {plugin.is_enabled}
    - API Endpoint: {config.api_endpoint if hasattr(config, 'api_endpoint') else 'Not configured'}
    """

これらの機能を適切に使用することで、強力でユーザーフレンドリーなNekro Agentプラグインを作成できます。