パターンと実践例 (Patterns and Examples)

高度な設計パターンとリファレンス資料のコレクションです：API 連携アプローチ、本番ノードから抽出した UI/UX パターン、柔軟なアーティファクト処理、およびインポート、ユーティリティ、型定義のクイックリファレンステーブルを解説します。

高度なトピック (Advanced Topics)

REST API 連携 vs SDK 連携

課題: Python SDK は新機能のサポートにおいて REST API よりも遅れることがよくあります。REST API ドキュメントに記載されているパラメータが、SDK ライブラリではまだ公開されていない場合があります。

REST API を直接使用すべきケース:

• SDK で文書化された API 機能（例: Gemini の image_config など）が欠落している場合
• 新しい API パラメータに即座にアクセスする必要がある場合
• SDK にバグや制限が存在する場合
• 依存関係をより軽量に保ちたい場合

REST API 実装パターン:

import base64
import requests
from google.oauth2 import service_account
from google.auth.transport.requests import Request

# 認証
credentials = service_account.Credentials.from_service_account_file(
    service_account_file, scopes=["https://www.googleapis.com/auth/cloud-platform"]
)


def _get_access_token(self, credentials) -> str:
    """認証情報からアクセストークンを取得します。"""
    if not credentials.valid:
        credentials.refresh(Request())
    return credentials.token


# REST API 仕様に合致する JSON ペイロードを構築
payload = {
    "contents": {
        "role": "USER",
        "parts": [
            {"text": prompt},
            {"inline_data": {"mime_type": "image/jpeg", "data": base64.b64encode(image_bytes).decode("utf-8")}},
        ],
    },
    "generation_config": {
        "temperature": 1.0,
        "topP": 0.95,
        "candidateCount": 1,
        "response_modalities": ["TEXT", "IMAGE"],
        "image_config": {  # SDK には存在しない機能！
            "aspect_ratio": "16:9"
        },
    },
}

# 認証済みリクエストを実行
access_token = self._get_access_token(credentials)
headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}

api_endpoint = f"https://{location}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{location}/publishers/google/models/{model}:generateContent"

response = requests.post(api_endpoint, headers=headers, json=payload, timeout=120)
response.raise_for_status()
response_data = response.json()

# JSON レスポンスを解析（camelCase と snake_case の両方に対応）
candidates = response_data.get("candidates", [])
for cand in candidates:
    parts_list = cand.get("content", {}).get("parts", [])
    for part in parts_list:
        if "inlineData" in part or "inline_data" in part:
            inline_data = part.get("inlineData") or part.get("inline_data", {})
            mime = inline_data.get("mimeType") or inline_data.get("mime_type")
            data_b64 = inline_data.get("data", "")
            data_bytes = base64.b64decode(data_b64)

重要な考慮事項:

1. 依存関係: 完全な SDK（google-cloud-aiplatform、google-genai）の代わりに google-auth を使用します
2. リージョンの可用性: 一部のモデルは global ではなく特定のリージョン（例: us-central1）でのみ動作します
3. モデル名: プレビューモデルと安定版モデルの間の -preview サフィックスの違いを確認します
4. 認証スコープ: Vertex AI には https://www.googleapis.com/auth/cloud-platform を使用します
5. レスポンス形式: camelCase（API）と snake_case（一部の SDK）の両方のフィールド名を処理できるようにします
6. Base64 エンコード: REST API はバイナリデータに対して Base64 エンコードされた文字列を期待します
7. エラー処理: 詳細なエラーメッセージを取得するために JSON エラーレスポンスを解析します

トレードオフ:

• ✅ すべての API 機能への即時アクセスが可能
• ✅ 依存関係が軽量
• ✅ リクエストの完全な制御が可能
• ❌ 実装作業量が増加
• ❌ 認証やトークン管理を手動で行う必要がある
• ❌ API の仕様変更を自身で追跡する必要がある

複雑な型管理システム (Complex Type Management Systems)

複数のパラメータ間で高度な型ネゴシエーションが必要なノードの場合：

class IfElse(BaseNode):
    def __init__(self, name: str, metadata: dict[Any, Any] | None = None) -> None:
        super().__init__(name, metadata)

        # 型管理のための高度な接続追跡
        self._possibility_space: list[str] = []  # 出力先で許容可能な型の空間
        self._locked_type: str | None = None  # 入力によってロックされた特定の型
        self._connected_inputs: set[str] = set()  # 入力接続の追跡
        self._output_connected: bool = False  # 出力接続の追跡

    def _update_parameter_types(self) -> None:
        """現在の状態に基づいてすべてのパラメータ型を更新します。"""
        if self._locked_type:
            # 特定の型にロックされている場合 - すべてがその型を使用
            self.output_if_true.input_types = [self._locked_type]
            self.output_if_false.input_types = [self._locked_type]
            self.output.output_type = self._locked_type
        elif self._possibility_space:
            # 許容空間内で柔軟に対応
            self.output_if_true.input_types = self._possibility_space.copy()
            self.output_if_false.input_types = self._possibility_space.copy()
            self.output.output_type = ParameterTypeBuiltin.ALL.value
        else:
            # デフォルト状態 - 任意の型を受け入れ
            self.output_if_true.input_types = ["any"]
            self.output_if_false.input_types = ["any"]
            self.output.output_type = ParameterTypeBuiltin.ALL.value

ベストプラクティス: 複数の入力と出力の間でデータをルーティングするノードには、高度な型管理ロジックを使用してください。

エージェントノード (Agentic Nodes)

エージェント管理には ControlNode を継承します：

from griptape.structures import Agent
from griptape_nodes.exe_types.node_types import ControlNode


class MyAgentNode(ControlNode):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.add_parameter(Parameter(name="agent_in", input_types=["Agent"], type="Agent"))
        self.add_parameter(Parameter(name="agent_out", output_type="Agent"))

    def process(self) -> None:
        agent_state = self.get_parameter_value("agent_in")
        agent = Agent.from_dict(agent_state) if agent_state else Agent()
        # エージェントによる処理
        self.parameter_output_values["agent_out"] = agent.to_dict()

ノードファミリー向けの抽象基底クラス

関連するノード間で共通機能を共有するための抽象基底クラスを作成します：

from abc import abstractmethod
from typing import Any

from griptape_nodes.exe_types.base_iterative_nodes import BaseIterativeStartNode


class BaseCustomIterativeStartNode(BaseIterativeStartNode):
    """カスタム反復 Start ノードファミリーのための基底クラス。"""

    @abstractmethod
    def _get_compatible_end_classes(self) -> set[type]:
        """この Start ノードが接続可能な End ノードクラスのセットを返します。"""

    @abstractmethod
    def _get_parameter_group_name(self) -> str:
        """反復データを含むパラメータグループの名前を返します。"""

    @abstractmethod
    def _get_exec_out_display_name(self) -> str:
        """exec_out パラメータの表示名を返します。"""

    @abstractmethod
    def _get_exec_out_tooltip(self) -> str:
        """exec_out パラメータのツールチップを返します。"""

    @abstractmethod
    def _get_iteration_items(self) -> list[Any]:
        """反復処理するアイテムのリストを取得します。"""

    @abstractmethod
    def is_loop_finished(self) -> bool:
        """ループのすべての反復が完了した場合に True を返します。"""

ベストプラクティス: 特定メソッドの実装を強制しながら、ノードファミリー間で共通ロジックを共有するために抽象基底クラスを使用してください。

キャッシュ (Caching)

共有リソースには ClassVar を使用します：

from typing import ClassVar, Any


class CachedModelNode(DataNode):
    _cache: ClassVar[dict[str, Any]] = {}

    def get_model(self, model_id: str) -> Any:
        if model_id not in self._cache:
            self._cache[model_id] = load_model(model_id)
        return self._cache[model_id]

モデルハブ連携 (例: Hugging Face)

# ゲート付き（利用申請が必要な）モデルの検出
is_gated = getattr(model, "gated", False)
model_dict["gated"] = is_gated

# ゲート付きモデルのステータス更新
if getattr(model_info, "gated", False):
    self.publish_update_to_parameter("status", "🔒 ゲート付きモデル - アクセス承認が必要な場合があります")

外部リンクおよびステータス更新用の ParameterMessage

from griptape_nodes.exe_types.core_types import ParameterMessage

# 外部リンクの例
ParameterMessage(
    name="model_card_link",
    title="モデルカード",
    variant="info",
    value="モデルのドキュメントを表示",
    button_link=f"https://huggingface.co/{model_id}",
    button_text="HuggingFace で表示",
)


# 動的ステータスメッセージの例
class MyIterativeNode(BaseIterativeStartNode):
    def __init__(self, name: str, metadata: dict[Any, Any] | None = None) -> None:
        super().__init__(name, metadata)

        # リアルタイム更新用のステータスメッセージパラメータ
        self.status_message = ParameterMessage(
            name="status_message",
            variant="info",
            value="",
        )
        self.add_node_element(self.status_message)

    def _update_status_message(self, status_type: str = "normal") -> None:
        """現在の状態に基づいてステータスメッセージを更新します。"""
        if self._total_iterations == 0:
            status = "完了 0 (全 0 件)"
        elif status_type == "break":
            status = f"{self._current_iteration_count} 件目で停止 (全 {self._total_iterations} 件) - Break"
        elif self.is_loop_finished():
            status = f"完了 {self._total_iterations} (全 {self._total_iterations} 件)"
        else:
            status = f"処理中 {self._current_iteration_count} (全 {self._total_iterations} 件)"

        self.status_message.value = status

ベストプラクティス: 静的な外部リンク、動的なステータス更新、および非推奨通知には ParameterMessage を使用してください。非推奨通知パターンの完全な解説（非表示メッセージ + before_value_set 自動移行）については、非推奨モデルの移行とユーザー通知 を参照してください。

モダンな UI/UX パターン (Modern UI/UX Patterns)

UI オプション (UI Options)

サポートされている ui_options キー、各パラメータ型がレンダリングするウィジェット、およびそれらを管理するトレイトについては、パラメータ UI リファレンス で正式に文書化されています。そこに記載されていないキーはエディタ内部専用であり、予告なく変更される可能性があります。

非表示パラメータのベストプラクティス

UI からパラメータを非表示にするには "hide": True を使用します（高度/エキスパート向け設定用）：

num_images_param = Parameter(
    name="num_images",
    input_types=["int"],
    type="int",
    default_value=1,
    tooltip="生成する画像数 (1-9)",
    allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
    ui_options={
        "display_name": "画像数",
        "hide": True,
    },
)
num_images_param.add_trait(Slider(min_val=1, max_val=9))
self.add_parameter(num_images_param)

一部の古いノードにはレガシーな "hidden": True キーも存在しますが、"hide": True を優先してください。

非表示パラメータの一般的なユースケース:

• 高度/エキスパート向けの設定オプション
• 内部制御シグナル
• デバッグパラメータ
• オプションの高度な機能
• プログラムからのみ設定されるべきパラメータ

Success/Failure ノードパターン

成功または失敗する可能性のある処理を行うノードには、SuccessFailureNode を継承します：

from griptape_nodes.exe_types.node_types import SuccessFailureNode


class LoadImage(SuccessFailureNode):
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)

        # ヘルパーメソッドを使用してステータスパラメータを追加
        self._create_status_parameters(
            result_details_tooltip="画像読み込み操作の結果に関する詳細",
            result_details_placeholder="読み込み試行の詳細がここに表示されます。",
        )

    def process(self) -> None:
        # 開始時に実行状態をリセット
        self._clear_execution_status()

        # エラー発生時に古いデータが残らないよう出力値をクリア
        self.parameter_output_values["image"] = None

        try:
            # ここに処理ロジックを実装
            result = load_image()
            self.parameter_output_values["image"] = result

            # 成功時
            success_details = f"{source} から画像が正常に読み込まれました"
            self._set_status_results(was_successful=True, result_details=f"SUCCESS: {success_details}")

        except Exception as e:
            error_details = f"画像の読み込みに失敗しました: {e}"
            self._set_status_results(was_successful=False, result_details=f"FAILURE: {error_details}")
            self._handle_failure_exception(e)

ベストプラクティス: 失敗する可能性があり、ユーザーにステータスを報告する必要がある操作には SuccessFailureNode を使用してください。

パラメータの初期化

ノード作成時にパラメータの表示状態を初期化します：

def _initialize_parameter_visibility(self) -> None:
    """デフォルト値に基づいてパラメータの表示状態を初期化します。"""
    default_model = self.get_parameter_value("model") or "default"
    if default_model == "text-only":
        self.hide_parameter_by_name("image_input")
    else:
        self.show_parameter_by_name("image_input")

アーティファクトパスのテザリングパターン

ファイルを扱うノードでは、パスパラメータとアーティファクトパラメータの同期を保つためにアーティファクトテザリングパターンを使用します：

from griptape_nodes_library.utils.artifact_path_tethering import (
    ArtifactPathTethering,
    ArtifactTetheringConfig,
)


class LoadImage(SuccessFailureNode):
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)

        # アーティファクトテザリングの設定
        self._tethering_config = ArtifactTetheringConfig(
            dict_to_artifact_func=dict_to_image_url_artifact,
            extract_url_func=self._extract_url_from_image_value,
            supported_extensions=self.SUPPORTED_EXTENSIONS,
            default_extension="png",
            url_content_type_prefix="image/",
        )

        # アーティファクトパラメータを作成
        self.image_parameter = Parameter(
            name="image",
            input_types=["ImageUrlArtifact", "ImageArtifact", "str"],
            type="ImageUrlArtifact",
            output_type="ImageUrlArtifact",
            ui_options={"clickable_file_browser": True},
        )

        # テザリングユーティリティを使用してパスパラメータを作成
        self.path_parameter = ArtifactPathTethering.create_path_parameter(
            name="path",
            config=self._tethering_config,
            display_name="ファイルパスまたは URL",
        )

        # テザリングヘルパーがパラメータの同期を維持
        self._tethering = ArtifactPathTethering(
            node=self,
            artifact_parameter=self.image_parameter,
            path_parameter=self.path_parameter,
            config=self._tethering_config,
        )

    def after_value_set(self, parameter: Parameter, value: Any) -> None:
        # テザリングロジックをヘルパーに委譲
        self._tethering.on_after_value_set(parameter, value)
        return super().after_value_set(parameter, value)

ベストプラクティス: ファイル/URL パラメータのシームレスな同期にはアーティファクトテザリングを使用してください。

2 モード UI パターン（シンプル + カスタム）

ユースケース: 初心者に親しみやすい UI を提供しつつ、パワーユーザー向けに高度な制御を可能にします。

例: 音楽/動画生成 API では、「シンプルな説明」モードと「詳細な制御」モードが用意されていることがよくあります。

実装パターン

class GenerativeNode(DataNode):
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)

        # モードセレクター
        mode_param = Parameter(
            name="custom_mode",
            input_types=["bool"],
            type="bool",
            default_value=False,
            tooltip="カスタムモード: 完全な制御。シンプルモード: プロンプトからの自動生成。",
            allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
            ui_options={"display_name": "カスタムモード"},
        )
        self.add_parameter(mode_param)

        # プロンプト（モードによって意味が変わる）
        prompt_param = Parameter(
            name="prompt",
            input_types=["str"],
            type="str",
            default_value="",
            tooltip=[
                {"type": "text", "text": "カスタムモード: 正確な歌詞やスクリプト"},
                {"type": "text", "text": "シンプルモード: 一般的な説明"},
            ],
            allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
            ui_options={"multiline": True, "display_name": "プロンプト"},
        )
        self.add_parameter(prompt_param)

        # 高度なパラメータ（カスタムモードのみ）
        style_param = Parameter(
            name="style",
            input_types=["str"],
            type="str",
            default_value="",
            tooltip="スタイル/ジャンル（カスタムモードのみ）",
            allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
            ui_options={"hide": True},  # デフォルトで非表示
        )
        self.add_parameter(style_param)

        title_param = Parameter(
            name="title",
            input_types=["str"],
            type="str",
            default_value="",
            tooltip="タイトル（カスタムモードのみ）",
            allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
            ui_options={"hide": True},  # デフォルトで非表示
        )
        self.add_parameter(title_param)

        # 表示状態の初期化
        self._initialize_parameter_visibility()

    def _initialize_parameter_visibility(self) -> None:
        """デフォルトモードに基づいてパラメータの表示状態を初期化します。"""
        custom_mode = self.get_parameter_value("custom_mode") or False
        if custom_mode:
            self.show_parameter_by_name("style")
            self.show_parameter_by_name("title")
        else:
            self.hide_parameter_by_name("style")
            self.hide_parameter_by_name("title")

    def after_value_set(self, parameter: Parameter, value: Any) -> None:
        """モード選択に基づいて UI を更新します。"""
        if parameter.name == "custom_mode":
            if value:
                self.show_parameter_by_name("style")
                self.show_parameter_by_name("title")
            else:
                self.hide_parameter_by_name("style")
                self.hide_parameter_by_name("title")

        return super().after_value_set(parameter, value)

    def validate_before_node_run(self) -> list[Exception] | None:
        """選択されたモードに基づいて検証を行います。"""
        exceptions = []
        custom_mode = self.get_parameter_value("custom_mode")

        if custom_mode:
            # カスタムモードではスタイルとタイトルが必須
            style = self.get_parameter_value("style") or ""
            title = self.get_parameter_value("title") or ""

            if not style.strip():
                exceptions.append(ValueError(f"{self.name}: カスタムモードではスタイルが必須です"))
            if not title.strip():
                exceptions.append(ValueError(f"{self.name}: カスタムモードではタイトルが必須です"))
        else:
            # シンプルモードではプロンプトのみが必要
            prompt = self.get_parameter_value("prompt") or ""
            if not prompt.strip():
                exceptions.append(ValueError(f"{self.name}: シンプルモードではプロンプトが必須です"))

        return exceptions if exceptions else None

初めて利用するユーザー向けのベストプラクティス: ドキュメントの推奨に従い、デフォルトをシンプルモードに設定します：

### はじめに

#### シンプルモード（初回利用時に推奨）

1. 「カスタムモード」のチェックを外したままにします
2. 説明を入力します: "穏やかなピアノのメロディ"
3. 実行します！

#### カスタムモード（上級者向け）

1. 「カスタムモード」にチェックを入れます
2. スタイル、タイトル、詳細なプロンプトを入力します
3. 高度なパラメータを微調整します

音楽/音声生成 API パターン (Music/Audio Generation API Patterns)

モデル別の文字数制限

多くの生成 API では、モデル固有の文字数制限が設けられています。制限値はクラス定数として保持します：

class MusicGenerationNode(DataNode):
    # モデルごとのプロンプト文字数制限
    PROMPT_LIMITS_CUSTOM = {
        "V3_5": 3000,
        "V4": 3000,
        "V4_5": 5000,
        "V5": 5000,
    }
    PROMPT_LIMIT_SIMPLE = 500

    # モデルごとのスタイル文字数制限
    STYLE_LIMITS = {
        "V3_5": 200,
        "V4": 200,
        "V4_5": 1000,
        "V5": 1000,
    }

    TITLE_LIMIT = 80

    def validate_before_node_run(self) -> list[Exception] | None:
        """モデル固有の制限値で検証します。"""
        exceptions = []
        model = self.get_parameter_value("model")
        custom_mode = self.get_parameter_value("custom_mode")

        if custom_mode:
            prompt = self.get_parameter_value("prompt") or ""
            prompt_limit = self.PROMPT_LIMITS_CUSTOM.get(model, 3000)
            if len(prompt) > prompt_limit:
                exceptions.append(
                    ValueError(
                        f"{self.name}: プロンプトが {model} の文字数制限（{prompt_limit} 文字）を超えています "
                        f"（現在: {len(prompt)} 文字）"
                    )
                )

            style = self.get_parameter_value("style") or ""
            style_limit = self.STYLE_LIMITS.get(model, 200)
            if len(style) > style_limit:
                exceptions.append(
                    ValueError(
                        f"{self.name}: スタイルが {model} の文字数制限（{style_limit} 文字）を超えています "
                        f"（現在: {len(style)} 文字）"
                    )
                )

        return exceptions if exceptions else None

詳細なツールチップ付きのモデル選択

モデルの比較には、辞書のリスト形式のツールチップを使用します：

model_param = Parameter(
    name="model",
    input_types=["str"],
    type="str",
    default_value="V5",
    tooltip=[
        {"type": "text", "text": "生成用モデルバージョン:"},
        {"type": "text", "text": "• V5: 最高品質、最速 (最大4分)"},
        {"type": "text", "text": "• V4_5PLUS: 最も豊かなサウンド、最大8分"},
        {"type": "text", "text": "• V4_5: 優れた調和、最大8分"},
        {"type": "text", "text": "• V4: 最高品質、洗練された構成 (4分)"},
        {"type": "text", "text": "• V3_5: 創造的な多様性 (4分)"},
    ],
    allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY},
)
model_param.add_trait(Options(choices=["V5", "V4_5PLUS", "V4_5", "V4", "V3_5"]))

デュアルトラック出力パターン

複数のバリエーションを生成する API の場合：

# 複数トラック用の出力パラメータ
music_urls_param = Parameter(
    name="music_urls",
    output_type="list[str]",
    type="list[str]",
    tooltip="生成されたトラックのダウンロード URL (2 バリエーション)",
    allowed_modes={ParameterMode.OUTPUT},
    settable=False,
    ui_options={"is_full_width": True, "display_name": "音楽 URL"},
)
self.add_parameter(music_urls_param)


def process(self) -> None:
    # ... 生成ロジック ...
    urls = self._extract_music_urls(response_data)
    self.parameter_output_values["music_urls"] = urls

    # 詳細な結果を構築
    result_lines = [
        f"✓ {len(urls)} 件のトラックバリエーションを生成しました",
        "",
        "音楽 URL:",
    ]
    for i, url in enumerate(urls, 1):
        result_lines.append(f"{i}. {url}")

    self.parameter_output_values["result_details"] = "\n".join(result_lines)

長時間処理中のステータス更新

ポーリング中にステータスパラメータをリアルタイムに更新します：

def _poll_for_completion(self, task_id: str, api_key: str) -> dict[str, Any]:
    """リアルタイムのステータス更新を伴う API ポーリング。"""
    for attempt in range(self.MAX_POLLING_ATTEMPTS):
        time.sleep(self.POLLING_INTERVAL)

        # 進捗状況でステータスパラメータを更新
        status_msg = f"生成中... ({attempt + 1}/{self.MAX_POLLING_ATTEMPTS})"
        self.set_parameter_value("status", status_msg)

        response = requests.get(query_url, headers=headers, params={"ids": task_id})
        # ... 完了を確認 ...

ベストプラクティス: 10 秒を超える処理には、必ず進捗状況のフィードバックを提供してください。

柔軟なアーティファクト処理 (Flexible Artifact Processing)

以下のパターンは、古いワークフローや上流ノードから ImageArtifact として渡される可能性のある値を処理するためのものです — これらは実行時の入力処理用であり、新しいパラメータを宣言するためのテンプレートではありません。新しいパラメータでは ImageUrlArtifact を宣言してください（パラメータのペイロードサイズ を参照）。

アーティファクトのダックタイピング

複数のアーティファクト形式を柔軟に処理します：

def _extract_image_value(self, image_input: Any) -> str | None:
    """さまざまな画像入力型から文字列値を抽出します。"""
    if isinstance(image_input, str):
        return image_input

    try:
        # ImageUrlArtifact: .value に URL 文字列が保持されている
        if hasattr(image_input, "value"):
            value = getattr(image_input, "value", None)
            if isinstance(value, str):
                return value

        # ImageArtifact: .base64 に生の文字列またはデータ URI が保持されている
        if hasattr(image_input, "base64"):
            b64 = getattr(image_input, "base64", None)
            if isinstance(b64, str) and b64:
                return b64
    except Exception as e:
        self._log(f"画像値の抽出に失敗しました: {e}")

    return None

外部 API 向けの画像形式変換

課題: 外部 API は形式の要件が厳格である（例: JPEG、PNG、WebP のみ許可）ことが多いのに対し、カメラは 3D/連写写真用に MPO（Multi Picture Object）のようなサポート外の形式で画像を保存する場合があります。

解決策: サポートされていない形式を自動的に検出して変換します：

# ファイルの先頭でインポート
from PIL import Image
from io import BytesIO


def _get_image_data(self, image_artifact: ImageArtifact | ImageUrlArtifact) -> str:
    """画像を API 互換の形式に変換します。"""
    # ... image_bytes の抽出 ...

    try:
        img = Image.open(BytesIO(image_bytes))

        # サポートされていない形式（MPO、TIFF、BMP 等）を JPEG に変換
        if img.format not in ["JPEG", "PNG", "WEBP"]:
            self._log(f"API 互換性のため {img.format} を JPEG に変換中")
            # 必要に応じて RGB に変換（MPO などの形式用）
            if img.mode not in ["RGB", "L"]:
                img = img.convert("RGB")
            # バイト列として JPEG 保存
            output = BytesIO()
            img.save(output, format="JPEG", quality=95)
            image_bytes = output.getvalue()
            mime_type = "image/jpeg"
        else:
            format_to_mime = {"JPEG": "image/jpeg", "PNG": "image/png", "WEBP": "image/webp"}
            mime_type = format_to_mime.get(img.format, "image/jpeg")
    except Exception as e:
        self._log(f"画像形式を検出できませんでした: {e}")
        mime_type = "image/jpeg"

    # Base64 データ URI としてエンコード
    base64_data = base64.b64encode(image_bytes).decode("utf-8")
    return f"data:{mime_type};base64,{base64_data}"

重要なポイント:

• すべての画像入力ポイント（ImageArtifact、ImageUrlArtifact、localhost URL）で変換を適用します
• 画像の品質を保つため、高い品質設定（95%）を使用します
• カラーモードの変換を処理します（MPO は非 RGB モードを使用することが多いです）
• デバッグのために変換処理をログに出力します
• 検出に失敗した場合は安全に JPEG にフォールバックします

ユーティリティ関数パターン

一般的な操作のための再利用可能なユーティリティ関数を作成します：

# 接続確認ユーティリティ。
#
# これらは GriptapeNodes.FlowManager() にアクセスするのではなく、リクエストを通じて問い合わせます。
# マネージャーのアクセサーはノードがワーカー内で実行されている間は拒否されます（接続情報はオーケストレーターに属するため）。
# リクエスト方式であれば、接続を所有する適切なプロセスによって回答されます。
def _connections_for(node_name: str) -> ListConnectionsForNodeResultSuccess | None:
    from griptape_nodes.retained_mode.events.connection_events import (
        ListConnectionsForNodeRequest,
        ListConnectionsForNodeResultSuccess,
    )
    from griptape_nodes.retained_mode.griptape_nodes import GriptapeNodes

    result = GriptapeNodes.handle_request(ListConnectionsForNodeRequest(node_name=node_name))
    if isinstance(result, ListConnectionsForNodeResultSuccess):
        return result
    return None


def _outgoing_connection_exists(source_node: str, source_param: str) -> bool:
    """送信元ノード/パラメータに送信接続が存在するかを確認します。"""
    connections = _connections_for(source_node)
    if connections is None:
        return False

    return any(c.source_parameter_name == source_param for c in connections.outgoing_connections)


def _incoming_connection_exists(target_node: str, target_param: str) -> bool:
    """接続先ノード/パラメータに着信接続が存在するかを確認します。"""
    connections = _connections_for(target_node)
    if connections is None:
        return False

    return any(c.target_parameter_name == target_param for c in connections.incoming_connections)

ベストプラクティス: 接続確認、検証、データ処理などの一般的な操作にはユーティリティ関数を作成してください。

柔軟な画像処理

def _image_to_bytes(self, image_artifact) -> bytes:
    """さまざまな画像アーティファクト型をバイト列に変換します。"""
    if not image_artifact:
        raise ValueError("入力画像が提供されていません")

    try:
        # 辞書形式を処理（シリアライズされたアーティファクト）
        if isinstance(image_artifact, dict):
            image_url_artifact = self._dict_to_image_url_artifact(image_artifact)
            image_bytes = image_url_artifact.to_bytes()
        # アーティファクトオブジェクトを直接処理
        elif isinstance(image_artifact, (ImageArtifact, ImageUrlArtifact)):
            image_bytes = image_artifact.to_bytes()
        else:
            # 汎用の to_bytes メソッドを試行
            image_bytes = image_artifact.to_bytes()

        # 有効な画像データであることを確認
        if not image_bytes or len(image_bytes) < 100:
            raise ValueError("画像データが空か小さすぎます")

        return image_bytes

    except Exception as e:
        raise ValueError(f"画像データの抽出に失敗しました: {str(e)}")

付録 (Appendix)

インポート一覧 (Imports)

# コアインポート
from griptape_nodes.exe_types.core_types import (
    Parameter,
    ParameterList,
    ParameterMode,
    ParameterTypeBuiltin,
    ParameterGroup,
    ParameterMessage,
    ControlParameterInput,
    ControlParameterOutput,
)
from griptape_nodes.exe_types.node_types import DataNode, ControlNode, BaseNode, SuccessFailureNode, StartNode, EndNode
from griptape_nodes.exe_types.base_iterative_nodes import BaseIterativeStartNode, BaseIterativeEndNode
from griptape_nodes.traits.options import Options
from griptape_nodes.traits.slider import Slider
from griptape_nodes.traits.color_picker import ColorPicker
from griptape_nodes.traits.file_system_picker import FileSystemPicker

# アーティファクト
from griptape.artifacts import ImageArtifact, ImageUrlArtifact, TextArtifact

# ユーティリティ
from griptape_nodes_library.utils.artifact_path_tethering import ArtifactPathTethering, ArtifactTetheringConfig
from griptape_nodes_library.utils.image_utils import (
    dict_to_image_url_artifact,
    load_pil_from_url,
    save_pil_image_with_named_filename,
)
from griptape_nodes_library.utils.file_utils import generate_filename

ユーティリティ関数リファレンス

画像ユーティリティ (griptape_nodes_library.utils.image_utils)

関数	役割	戻り値
dict_to_image_url_artifact(d)	辞書表現を ImageUrlArtifact に変換	ImageUrlArtifact
load_pil_from_url(url)	URL から PIL Image をロード（localhost を処理）	PIL.Image.Image
save_pil_image_with_named_filename(img, filename)	プロジェクトシステムを使用して PIL Image を保存（下記注記参照）	ImageUrlArtifact

使用例:

from griptape_nodes_library.utils.image_utils import (
    dict_to_image_url_artifact,
    load_pil_from_url,
    save_pil_image_with_named_filename,
)

# パラメータ値をアーティファクトに変換
value = self.get_parameter_value("image")
if isinstance(value, dict):
    artifact = dict_to_image_url_artifact(value)
else:
    artifact = value

# 処理のために PIL としてロード
pil_image = load_pil_from_url(artifact.value)

# 画像を処理...
processed = pil_image.filter(...)

# 保存して出力アーティファクトを取得
output_artifact = save_pil_image_with_named_filename(processed, "result.png")
self.parameter_output_values["output"] = output_artifact

ファイルユーティリティ (griptape_nodes_library.utils.file_utils)

関数	役割	戻り値
generate_filename(node_name, suffix, ext)	一貫性のあるファイル名を生成	str

使用例:

from griptape_nodes_library.utils.file_utils import generate_filename

# "ColorMatch_processed_abc123.png" のようなファイル名を生成
filename = generate_filename(self.name, suffix="processed", ext="png")

プロジェクトシステム (griptape_nodes.files.project_file)

クラス/関数	役割	戻り値
ProjectFileDestination.from_situation()	指定されたシチュエーションでファイル保存先を作成	ProjectFileDestination
ProjectFileParameter	設定可能なファイル出力のためのパラメータコンポーネント	-

使用例:

from griptape_nodes.files.project_file import ProjectFileDestination

# プロジェクトシステムを使用してファイルを保存
dest = ProjectFileDestination.from_situation(filename="output.mp4", situation="save_node_output")
saved = dest.write_bytes(file_bytes)
artifact = VideoUrlArtifact(saved.location)

注記: save_pil_image_with_named_filename() などのヘルパー関数は、内部的にプロジェクトシステムを使用しています。新しいコードでは、ファイルの取り扱いを完全に制御するために ProjectFileParameter または ProjectFileDestination を直接使用することを推奨します。包括的なドキュメントについては プロジェクトシステムとの連携 を参照してください。

高度なパラメータ型 (Advanced Parameter Types)

• ControlParameterInput/Output: 実行フロー制御用
• ParameterGroup: 折りたたみ可能な UI で関連パラメータを整理
• ParameterMessage: ステータス更新および外部リンク用
• ParameterList: 同じ型の複数の入力を受け入れる用

列挙型 (Enumerations)

• NodeResolutionState: UNRESOLVED, RESOLVING, RESOLVED
• ParameterMode: INPUT, OUTPUT, PROPERTY
• ParameterTypeBuiltin: STR("str"), BOOL("bool"), INT("int"), FLOAT("float"), ANY("any"), NONE("none"), CONTROL_TYPE("parametercontroltype"), ALL("all")

高度なノード型 (Advanced Node Types)

• BaseNode: カスタム実装のための最も基本的なノード型
• DataNode: 実行フローを伴わないデータ処理用
• ControlNode: 実行フローを制御するノード用
• SuccessFailureNode: 成功または失敗する可能性のある処理用
• BaseIterativeStartNode/BaseIterativeEndNode: 反復処理（ループ）用
• AsyncResult: ブロッキング処理をバックグラウンドスレッドに委託するためのジェネレータ型（真に非同期な処理には async def aprocess() のオーバーライドを推奨）

カスタムアーティファクト (Custom Artifacts)

BaseArtifact を継承し、必要に応じてメソッドをオーバーライドします：

from griptape.artifacts import BaseArtifact


class CustomArtifact(BaseArtifact):
    def __init__(self, value: Any, **kwargs):
        super().__init__(value, **kwargs)

    def to_text(self) -> str:
        return str(self.value)

高度なライフサイクルメソッド (Advanced Lifecycle Methods)

スポットライト制御 (Spotlight Control)

条件付きの依存関係解決用：

def initialize_spotlight(self) -> None:
    """カスタムスポットライト初期化 - 初期状態では evaluate パラメータのみを含める。"""
    evaluate_param = self.get_parameter_by_name("evaluate")
    if evaluate_param and ParameterMode.INPUT in evaluate_param.get_mode():
        self.current_spotlight_parameter = evaluate_param


def advance_parameter(self) -> bool:
    """条件付き依存関係解決を伴うカスタムパラメータ進行。"""
    if self.current_spotlight_parameter is None:
        return False

    # 条件付きパラメータの特別な処理
    if self.current_spotlight_parameter is self.evaluate:
        try:
            evaluation_result = self.check_evaluation()
            next_param = self.output_if_true if evaluation_result else self.output_if_false

            if ParameterMode.INPUT in next_param.get_mode():
                self.current_spotlight_parameter.next = next_param
                next_param.prev = self.current_spotlight_parameter
                self.current_spotlight_parameter = next_param
                return True
        except Exception:
            self.current_spotlight_parameter = None
            return False

    return super().advance_parameter()

制御フロー管理 (Control Flow Management)

def get_next_control_output(self) -> Parameter | None:
    """評価結果に基づいて適切な制御出力を返します。"""
    if "evaluate" not in self.parameter_output_values:
        self.stop_flow = True
        return None

    if self.parameter_output_values["evaluate"]:
        return self.get_parameter_by_name("Then")
    return self.get_parameter_by_name("Else")