NekroAgent ドキュメント

日本語

简体中文
English

日本語

简体中文
English

Appearance

ファイルインタラクション ​

プラグイン開発では、ユーザーがアップロードしたファイルの読み取り、AIに返す画像やドキュメントの生成、マルチメディアコンテンツの処理など、ファイル処理が必要になることがよくあります。Nekro AgentはAgentCtxのfs属性を通じて新しいファイルシステムAPIを提供し、プラグインとAIサンドボックス間のファイル転送をシンプルかつ効率的にします。

ファイルシステム概要 ​

Nekro Agentでは、AIは分離されたサンドボックス環境で実行され、プラグインはメインサービスプロセスで実行されます。新しいファイルシステムAPIは_ctx.fsを通じて2つのコアメソッドを提供し、ファイル転送を処理します：

  • mixed_forward_file: プラグインがファイルをAIに転送（URL、バイトデータ、ローカルファイルなど → サンドボックスパス）
  • get_file: AIがファイルをプラグインに転送（サンドボックスパス → ホストマシン上の実際のパス）

コアコンセプト ​

サンドボックスパス vs. ホストパス ​

  • サンドボックスパス: AIが認識するファイルパスで、通常/app/uploads/または/app/shared/で始まります
  • ホストパス: プラグインが実際にファイルを操作するホストマシン上の実際のパス

ファイルシステムAPIはこれら2つのパスタイプ間の変換を自動的に処理するため、開発者は基礎となる実装の詳細を気にする必要がありません。

プラグインからAIへのファイル転送 ​

mixed_forward_file メソッド ​

プラグインがファイル（画像、ドキュメント、データなど）を生成し、それをAIに返す必要がある場合、このメソッドを使用します：

python
async def mixed_forward_file(
    file_source: Union[str, bytes, Path, BinaryIO],
    file_name: Optional[str] = None,
    **kwargs
) -> str

サポートされるファイルソースタイプ:

  • HTTP/HTTPS URL文字列
  • バイトデータ (bytes)
  • ローカルファイルパス (Path)
  • ファイルオブジェクト (BinaryIO)

使用例 ​

例1: URLから画像を転送 ​

python
from nekro_agent.api.plugin import SandboxMethodType
from nekro_agent.api.schemas import AgentCtx

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_logo", "プロジェクトロゴを取得")
async def get_logo(_ctx: AgentCtx) -> str:
    """URLからロゴ画像を取得し、AIに返す"""

    # 外部URLから画像を取得
    image_url = "https://nekro-agent-dev.oss-cn-beijing.aliyuncs.com/images/NA_logo.png"

    # AIが使用できるサンドボックスパスに変換
    sandbox_path = await _ctx.fs.mixed_forward_file(image_url, file_name="logo.png")

    # サンドボックスパスをAIに返す
    return sandbox_path  # "/app/uploads/logo.png"

例2: データファイルを生成 ​

python
import json

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_data_file", "データファイルを生成")
async def generate_data_file(_ctx: AgentCtx, data: dict) -> str:
    """JSONデータファイルを生成し、AIに返す"""

    # データをJSONバイトにシリアライズ
    json_data = json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8')

    # AIが使用できるサンドボックスパスに変換
    sandbox_path = await _ctx.fs.mixed_forward_file(json_data, file_name="data.json")

    return sandbox_path  # "/app/uploads/data.json"

例3: プラグイン共有ディレクトリを使用 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_file", "一時ファイルを作成")
async def create_temp_file(_ctx: AgentCtx, content: str) -> str:
    """プラグイン共有ディレクトリに一時ファイルを作成"""

    # 共有ディレクトリにファイルを作成
    temp_file = _ctx.fs.shared_path / "temp.txt"
    with open(temp_file, "w", encoding="utf-8") as f:
        f.write(content)

    # AIが使用できるサンドボックスパスに変換
    sandbox_path = _ctx.fs.forward_file(temp_file)

    return sandbox_path  # "/app/shared/temp.txt"

AIからプラグインへのファイル転送 ​

get_file メソッド ​

AIがプラグインを呼び出し、ファイルパスパラメータを渡す場合、プラグインはこのメソッドを使用してホストマシン上の実際のパスを取得します：

python
def get_file(sandbox_path: Union[str, Path]) -> Path

使用例 ​

例1: 画像ファイルを分析 ​

python
import aiofiles
from PIL import Image

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "画像ファイルを分析")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
    """AIが提供した画像ファイルを分析"""

    # AIが提供したサンドボックスパス: "/app/uploads/photo.jpg"
    # ホストマシンがアクセスできる実際のパスに変換
    host_path = _ctx.fs.get_file(image_path)

    try:
        # PILを使用して画像を開いて分析
        with Image.open(host_path) as img:
            width, height = img.size
            format_name = img.format
            mode = img.mode

        return f"画像分析結果: サイズ {width}x{height}, フォーマット {format_name}, モード {mode}"

    except Exception as e:
        return f"画像分析に失敗しました: {e}"

例2: テキストファイルを処理 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_text_file", "テキストファイルを処理")
async def process_text_file(_ctx: AgentCtx, file_path: str) -> str:
    """AIが提供したテキストファイルを処理"""

    # ホストパスに変換
    host_path = _ctx.fs.get_file(file_path)

    try:
        # ファイルコンテンツを非同期で読み取り
        async with aiofiles.open(host_path, "r", encoding="utf-8") as f:
            content = await f.read()

        # 簡単なテキスト分析
        lines = len(content.splitlines())
        words = len(content.split())
        chars = len(content)

        return f"ファイル分析: {lines}行, {words}単語, {chars}文字"

    except Exception as e:
        return f"ファイル処理に失敗しました: {e}"

メッセージ送信と組み合わせたファイル処理 ​

便利なメソッドを使用してファイルを送信 ​

AgentCtxはユーザーに直接ファイルを送信できる便利なメッセージ送信メソッドを提供します：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "レポートを生成して送信")
async def send_report(_ctx: AgentCtx, title: str, content: str) -> str:
    """レポートファイルを生成し、ユーザーに送信"""

    # レポートコンテンツを作成
    report_content = f"# {title}\n\n{content}\n\n生成時間: {datetime.now()}"
    report_bytes = report_content.encode('utf-8')

    # サンドボックスパスに変換
    sandbox_path = await _ctx.fs.mixed_forward_file(report_bytes, file_name=f"{title}.md")

    # ユーザーにファイルを送信
    await _ctx.send_file(sandbox_path)

    return f"レポート '{title}' を生成し、送信しました"

処理された画像を送信 ​

python
from PIL import Image, ImageFilter

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "apply_blur_filter", "ぼかしフィルタを適用")
async def apply_blur_filter(_ctx: AgentCtx, image_path: str, radius: float = 2.0) -> str:
    """画像にぼかしフィルタを適用し、結果を送信"""

    # 元の画像のホストパスを取得
    host_path = _ctx.fs.get_file(image_path)

    try:
        # 画像を開いて処理
        with Image.open(host_path) as img:
            # ぼかしフィルタを適用
            blurred = img.filter(ImageFilter.GaussianBlur(radius=radius))

            # 一時的な場所に保存
            output_path = _ctx.fs.shared_path / "blurred_image.jpg"
            blurred.save(output_path, "JPEG")

        # サンドボックスパスに変換して送信
        sandbox_path = _ctx.fs.forward_file(output_path)
        await _ctx.send_image(sandbox_path)

        return f"ぼかしフィルタ（半径 {radius}）を適用し、結果画像を送信しました"

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

ファイルシステムプロパティ ​

共通パスプロパティ ​

_ctx.fsはいくつかの便利なパスプロパティを提供します：

python
# プラグイン共有ディレクトリ（一時ファイル用）
shared_path = _ctx.fs.shared_path

# ユーザーアップロードディレクトリ
uploads_path = _ctx.fs.uploads_path

# チャットチャネル識別子
chat_key = _ctx.fs.chat_key

# コンテナ識別子
container_key = _ctx.fs.container_key

ユーティリティメソッド ​

python
# ホストファイルをサンドボックスパスに変換
sandbox_path = _ctx.fs.forward_file(host_file_path)

# ファイルが存在するかチェック
if _ctx.fs.get_file(sandbox_path).exists():
    # ファイルが存在
    pass

ベストプラクティス ​

1. エラーハンドリング ​

ファイル操作には常に適切なエラーハンドリングを実行してください：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "safe_file_process", "安全なファイル処理")
async def safe_file_process(_ctx: AgentCtx, file_path: str) -> str:
    try:
        host_path = _ctx.fs.get_file(file_path)

        if not host_path.exists():
            return "エラー: 指定されたファイルが存在しません"

        # ファイル処理ロジック...
        return "処理が成功しました"

    except Exception as e:
        return f"ファイル処理に失敗しました: {e}"

2. ファイルタイプ検証 ​

特定のタイプのファイルの場合、タイプ検証を実行することをお勧めします：

python
import mimetypes

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image_only", "画像ファイルのみ処理")
async def process_image_only(_ctx: AgentCtx, file_path: str) -> str:
    host_path = _ctx.fs.get_file(file_path)

    # ファイルタイプをチェック
    mime_type, _ = mimetypes.guess_type(str(host_path))
    if not mime_type or not mime_type.startswith('image/'):
        return "エラー: 画像ファイルを提供してください"

    # 画像処理ロジック...
    return "画像処理が完了しました"

3. 大きなファイルの処理 ​

大きなファイルの場合、ストリーミング処理を使用します：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "hash_large_file", "大きなファイルのハッシュを計算")
async def hash_large_file(_ctx: AgentCtx, file_path: str) -> str:
    import hashlib

    host_path = _ctx.fs.get_file(file_path)

    sha256_hash = hashlib.sha256()

    # 大きなファイルをストリーム読み取り
    async with aiofiles.open(host_path, "rb") as f:
        while chunk := await f.read(8192):  # 8KBチャンク
            sha256_hash.update(chunk)

    return f"ファイルSHA256: {sha256_hash.hexdigest()}"

4. 一時ファイルのクリーンアップ ​

コンテキストマネージャを使用するか、適切なタイミングで一時ファイルをクリーンアップします：

python
import tempfile
import os

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_processed_file", "一時処理ファイルを作成")
async def create_temp_processed_file(_ctx: AgentCtx, input_data: str) -> str:
    try:
        # 一時ファイルを作成
        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as temp_file:
            temp_file.write(f"処理結果: {input_data}")
            temp_path = temp_file.name

        # サンドボックスパスに変換
        sandbox_path = await _ctx.fs.mixed_forward_file(temp_path, file_name="processed.txt")

        # 一時ファイルをクリーンアップ
        os.unlink(temp_path)

        return sandbox_path

    except Exception as e:
        return f"一時ファイルの作成に失敗しました: {e}"

新しいファイルシステムAPIを使用することで、プラグイン開発者は基礎となるパス変換や環境の違いを気にすることなく、複雑なファイル処理機能を簡単に実装できます。

ドキュメントに問題がありますか？問題を報告