ダイナミックパッケージインポート

Nekro Agentはダイナミックパッケージインポート機能を提供しており、プラグインが実行時にオンデマンドでPythonサードパーティパッケージをインストール・インポートできるようにします。この機能により、プラグインはシステム環境を変更することなく、Pythonエコシステムの様々なツールやフレームワークを柔軟に利用できます。

機能概要

ダイナミックパッケージインポート機能には以下の特徴があります：

オンデマンドインストール: 必要な時だけパッケージをダウンロード・インストールし、システムリソースを節約
バージョン管理: 正確なバージョン指定と依存関係の制約をサポート
分離ストレージ: パッケージはプラグイン固有のディレクトリにインストールされ、システムPython環境に影響を与えない
ミラーサポート: 国内ダウンロードを高速化するためのPyPIミラーソース設定をサポート
エラーハンドリング: フレンドリーなエラープロンプトと例外処理メカニズムを提供

基本的な使用方法

インポート関数

nekro_agent.api.pluginからダイナミックパッケージインポート関数をインポートします：

python

from nekro_agent.api.plugin import dynamic_import_pkg
from nekro_agent.api.schemas import AgentCtx

簡単なインポート例

最も簡単な使用方法は、パッケージ名を直接指定することです：

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "use_requests", "requestsライブラリを使用してHTTPリクエストを送信")
async def fetch_url(_ctx: AgentCtx, url: str) -> str:
    """動的にインポートしたrequestsライブラリを使用してWebページコンテンツを取得"""

    # requestsパッケージを動的にインポート
    requests = dynamic_import_pkg("requests")

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return f"ページの取得に成功、ステータスコード: {response.status_code}、コンテンツ長: {len(response.text)}文字"
    except Exception as e:
        return f"リクエスト失敗: {e}"

バージョン指定

バージョン範囲の指定

標準的なPythonパッケージバージョン指定構文をサポートします：

python

# 正確なバージョン
pandas = dynamic_import_pkg("pandas==2.0.0")

# 最小バージョン要件
numpy = dynamic_import_pkg("numpy>=1.24.0")

# バージョン範囲
requests = dynamic_import_pkg("requests>=2.25.0,<3.0.0")

# 特定バージョンを除外
flask = dynamic_import_pkg("flask>=2.0,!=2.0.1")

バージョン指定の例

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "data_analysis", "データ分析を実行")
async def analyze_data(_ctx: AgentCtx, data: list) -> str:
    """pandasを使用してデータ分析を実行"""

    # 特定バージョンのpandasをインポート
    pd = dynamic_import_pkg(
        "pandas>=2.0.0,<3.0.0",
        import_name="pandas"  # インポート用のモジュール名を指定
    )

    try:
        df = pd.DataFrame(data)
        summary = df.describe().to_string()
        return f"データ分析結果:\n{summary}"
    except Exception as e:
        return f"分析失敗: {e}"

高度なパラメータ設定

完全なパラメータ説明

python

dynamic_import_pkg(
    package_spec: str,                    # パッケージ指定、例: "requests>=2.25.0"
    import_name: Optional[str] = None,    # インポート名、デフォルトはパッケージ名
    mirror: Optional[str] = "https://pypi.tuna.tsinghua.edu.cn/simple",  # PyPIミラーソース
    trusted_host: bool = True,            # ミラーソースホストを信頼するかどうか
    timeout: int = 300,                   # インストールタイムアウト（秒）
    repo_dir: Optional[Path] = None       # カスタムインストールディレクトリ
)

国内ミラーを使用した高速化

デフォルトで清華大学PyPIミラーソースを使用しますが、他のミラーも指定できます：

python

# アリババクラウドミラーを使用
beautifulsoup4 = dynamic_import_pkg(
    "beautifulsoup4",
    mirror="https://mirrors.aliyun.com/pypi/simple/",
    trusted_host=True
)

# テンセントクラウドミラーを使用
pillow = dynamic_import_pkg(
    "Pillow>=10.0.0",
    mirror="https://mirrors.cloud.tencent.com/pypi/simple",
    trusted_host=True
)

# 公式PyPIを使用（遅い）
scipy = dynamic_import_pkg(
    "scipy",
    mirror="https://pypi.org/simple",
    trusted_host=False
)

実用的な使用例

例1: Webスクレイピング

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scrape_webpage", "Webページコンテンツをスクレイピング")
async def scrape_webpage(_ctx: AgentCtx, url: str, selector: str) -> str:
    """BeautifulSoupを使用してWebページから指定された要素をスクレイピング"""

    # 必要なパッケージを動的にインポート
    requests = dynamic_import_pkg("requests>=2.25.0")
    bs4 = dynamic_import_pkg("beautifulsoup4>=4.9.0", import_name="bs4")

    try:
        # Webページコンテンツを取得
        response = requests.get(url, timeout=10)
        response.raise_for_status()

        # HTMLを解析
        soup = bs4.BeautifulSoup(response.text, 'html.parser')
        elements = soup.select(selector)

        if not elements:
            return f"セレクタ '{selector}' に一致する要素が見つかりません"

        # テキストコンテンツを抽出
        results = [elem.get_text(strip=True) for elem in elements[:5]]  # 最大5つまで返す
        return f"{len(elements)}個の要素が見つかりました、最初の{len(results)}個の内容:\n" + "\n".join(results)

    except Exception as e:
        return f"スクレイピング失敗: {e}"

例2: 画像処理

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image", "画像ファイルを処理")
async def process_image(_ctx: AgentCtx, image_path: str, operation: str) -> str:
    """Pillowを使用して画像を処理"""

    # Pillowを動的にインポート
    PIL = dynamic_import_pkg("Pillow>=10.0.0", import_name="PIL")
    from PIL import Image, ImageFilter

    try:
        # 実際のファイルパスを取得
        real_path = _ctx.fs.get_file(image_path)

        # 画像を開く
        img = Image.open(real_path)

        # 操作タイプに基づいて処理
        if operation == "blur":
            processed = img.filter(ImageFilter.BLUR)
        elif operation == "grayscale":
            processed = img.convert('L')
        elif operation == "resize":
            processed = img.resize((800, 600))
        else:
            return f"サポートされていない操作: {operation}"

        # 処理した画像を保存
        output_path = _ctx.fs.shared_path / f"processed_{operation}.jpg"
        processed.save(output_path)

        # AIにサンドボックスパスを返す
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"画像処理完了、結果: {sandbox_path}"

    except Exception as e:
        return f"画像処理失敗: {e}"

例3: データ処理と可視化

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_chart", "データチャートを作成")
async def create_data_chart(_ctx: AgentCtx, data: dict, chart_type: str) -> str:
    """matplotlibを使用してデータチャートを作成"""

    # データ処理と可視化ライブラリを動的にインポート
    pd = dynamic_import_pkg("pandas>=2.0.0", import_name="pandas")
    plt = dynamic_import_pkg("matplotlib>=3.5.0", import_name="matplotlib.pyplot")

    try:
        # DataFrameを作成
        df = pd.DataFrame(data)

        # チャートを作成
        fig, ax = plt.subplots(figsize=(10, 6))

        if chart_type == "bar":
            df.plot(kind='bar', ax=ax)
        elif chart_type == "line":
            df.plot(kind='line', ax=ax)
        elif chart_type == "pie":
            df.plot(kind='pie', y=df.columns[0], ax=ax)
        else:
            return f"サポートされていないチャートタイプ: {chart_type}"

        # チャートを保存
        output_path = _ctx.fs.shared_path / f"chart_{chart_type}.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close(fig)

        # サンドボックスパスを返す
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"チャート生成完了: {sandbox_path}"

    except Exception as e:
        return f"チャート生成失敗: {e}"

例4: 科学技術計算

python

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scientific_compute", "科学技術計算を実行")
async def scientific_compute(_ctx: AgentCtx, operation: str, values: list) -> str:
    """NumPyとSciPyを使用して科学技術計算を実行"""

    # 科学技術計算ライブラリを動的にインポート
    np = dynamic_import_pkg("numpy>=1.24.0", import_name="numpy")
    scipy = dynamic_import_pkg("scipy>=1.10.0")

    try:
        data = np.array(values)

        if operation == "fft":
            # 高速フーリエ変換
            result = np.fft.fft(data)
            return f"FFT結果（最初の5個）: {result[:5]}"

        elif operation == "stats":
            # 統計分析
            from scipy import stats
            mean = np.mean(data)
            std = np.std(data)
            skew = stats.skew(data)
            kurtosis = stats.kurtosis(data)

            return f"""統計結果:
- 平均値: {mean:.4f}
- 標準偏差: {std:.4f}
- 歪度: {skew:.4f}
- 尖度: {kurtosis:.4f}"""

        elif operation == "interpolate":
            # データ補間
            from scipy import interpolate
            x = np.arange(len(data))
            f = interpolate.interp1d(x, data, kind='cubic')
            x_new = np.linspace(0, len(data)-1, len(data)*2)
            y_new = f(x_new)
            return f"補間完了、元のデータポイント: {len(data)}、補間後: {len(y_new)}"

        else:
            return f"サポートされていない操作: {operation}"

    except Exception as e:
        return f"計算失敗: {e}"

エラーハンドリング

動的インポートはネットワーク問題、存在しないパッケージ、その他の理由で失敗する可能性があります

一般的なエラータイプ

python

# RuntimeError: インストール失敗
# - ネットワーク接続問題
# - パッケージが存在しないか、バージョンが利用不可
# - SSL証明書検証失敗
# - ミラーソースへのアクセス拒否

# ImportError: インポート失敗
# - パッケージはインストールされたがモジュールが見つからない
# - インポート名が実際のモジュール名と一致しない

# ValueError: パッケージ指定形式エラー
# - 不正なバージョン番号形式
# - サポートされていないバージョン演算子を使用

# subprocess.TimeoutExpired: インストールタイムアウト
# - パッケージサイズが大きすぎる
# - ネットワーク速度が遅すぎる

注意事項

セキュリティ考慮事項

信頼できるパッケージのみインポート: 公式PyPIまたは信頼できるミラーソースからのみパッケージをインストール
バージョンロック: 本番環境では正確なバージョン番号を使用し、予期せぬパッケージ更新を回避
依存関係の確認: パッケージの推移的依存関係を理解し、不要なリスクの導入を回避
ドキュメント宣言: 使用するパッケージをプラグインドキュメントで宣言し、バージョン番号を記載

パフォーマンス考慮事項

初回インストールオーバーヘッド: パッケージの初回使用時にダウンロードとインストール時間がかかる
ディスク容量: 各パッケージはディスク容量を占有するため、パッケージの数とサイズを制御することに注意

互換性考慮事項

Pythonバージョン: パッケージがNekro Agentを実行しているPythonバージョンをサポートしていることを確認
システム依存関係: 一部のパッケージはシステムレベルの依存ライブラリを必要とする場合がある（例: OpenCVは特定のC++ライブラリを必要）
パッケージ競合: 異なるパッケージ間の潜在的な依存関係競合に注意

トラブルシューティング

インストール成功後にインポート失敗

エラー: ImportError: インストールに成功しましたがmodule_nameをインポートできません

解決策:
1. import_nameパラメータが正しいか確認
2. 一部のパッケージはインポート名がパッケージ名と異なる場合がある（例: Pillow -> PIL）
3. パッケージが追加のシステム依存関係を必要とする可能性がある

ダイナミックパッケージインポート機能を適切に活用することで、プラグインはPythonの豊富なエコシステムを柔軟に活用し、より強力で多様な機能を実装できます。

コアコンセプト

高度な機能

ダイナミックパッケージインポート

機能概要

基本的な使用方法

インポート関数

簡単なインポート例

バージョン指定

バージョン範囲の指定

バージョン指定の例

高度なパラメータ設定

完全なパラメータ説明

国内ミラーを使用した高速化

実用的な使用例

例1: Webスクレイピング

例2: 画像処理

例3: データ処理と可視化

例4: 科学技術計算

エラーハンドリング

一般的なエラータイプ

注意事項

セキュリティ考慮事項

パフォーマンス考慮事項

互換性考慮事項

トラブルシューティング

インストール成功後にインポート失敗

ダイナミックパッケージインポート ​

機能概要 ​

基本的な使用方法 ​

インポート関数 ​

簡単なインポート例 ​

バージョン指定 ​

バージョン範囲の指定 ​

バージョン指定の例 ​

高度なパラメータ設定 ​

完全なパラメータ説明 ​

国内ミラーを使用した高速化 ​

実用的な使用例 ​

例1: Webスクレイピング ​

例2: 画像処理 ​

例3: データ処理と可視化 ​

例4: 科学技術計算 ​

エラーハンドリング ​

一般的なエラータイプ ​

注意事項 ​

セキュリティ考慮事項 ​

パフォーマンス考慮事項 ​

互換性考慮事項 ​

トラブルシューティング ​

インストール成功後にインポート失敗 ​

ダイナミックパッケージインポート

機能概要

基本的な使用方法

インポート関数

簡単なインポート例

バージョン指定

バージョン範囲の指定

バージョン指定の例

高度なパラメータ設定

完全なパラメータ説明

国内ミラーを使用した高速化

実用的な使用例

例1: Webスクレイピング

例2: 画像処理

例3: データ処理と可視化

例4: 科学技術計算

エラーハンドリング

一般的なエラータイプ

注意事項

セキュリティ考慮事項

パフォーマンス考慮事項

互換性考慮事項

トラブルシューティング

インストール成功後にインポート失敗