ベストプラクティスとエラー処理 (Best Practices and Error Handling)

本番環境に耐えうる高品質なノードを構築するための一般的なベストプラクティス：シークレット管理、インポート、コード品質、エラー処理、バリデーション、およびロギングについて解説します。

ベストプラクティス (Best Practices)

コア原則 (Core Principles)

• わかりやすい名前とツールチップ (Descriptive names and tooltips)
• バリデータを用いた堅牢なエラー処理 (Robust error handling with validators)
• ノードごとの単一責任の原則 (Single responsibility per node)
• API キーや機密情報には GetSecretValueRequest を使用 (Use GetSecretValueRequest for API keys and secrets)
• すべての依存関係をモジュールレベルでインポート (Import all dependencies at module level)
• 冪等性のある process メソッド (Idempotent process methods)

シークレット管理 (Secrets Management)

シークレットは GetSecretValueRequest を通じて読み込みます：

from griptape_nodes.retained_mode.events.secrets_events import (
    GetSecretValueRequest,
    GetSecretValueResultSuccess,
)
from griptape_nodes.retained_mode.griptape_nodes import GriptapeNodes


class MyNode(DataNode):
    SERVICE_NAME = "MyService"
    API_KEY_NAME = "MY_SERVICE_API_KEY"

    def _validate_api_key(self) -> str:
        result = GriptapeNodes.handle_request(GetSecretValueRequest(key=self.API_KEY_NAME))
        if not isinstance(result, GetSecretValueResultSuccess) or not result.value:
            raise ValueError(f"Missing {self.API_KEY_NAME}")
        return result.value

重要なポイント:

• 関数内ではなく、モジュールレベルでインポートします
• GriptapeNodes.SecretsManager() ではなくリクエストハンドラーを使用してください。マネージャーのアクセサーは ノードがワーカー内で実行されている間はアクセスが拒否されます。また、このようなヘルパーは process（ワーカー内）とバリデーション（オーケストレーター内）の両方から呼び出される可能性があるため、マネージャー方式だと一方の呼び出し元では動作し、もう一方では例外が発生します。リクエストハンドラー方式であれば両方で正常に動作します。
• 一貫性を保つため、API_KEY_NAME はクラス定数として定義します
• シークレットを使用する前に、必ず存在することを検証してください

パラメータのペイロードサイズ (Parameter Payload Size)

パラメータ値は、以下の 2 つの場所に完全に埋め込まれる可能性があります：

• 保存されたワークフローファイル。 ワークフローのシリアライザーは、一意のパラメータ値をサイズ制限なしで保存された .py ファイル内にインラインで書き込みます — 値が保持しているすべての Python 状態がそのまま書き出されます。
• WebSocket イベント。 リクエスト/レスポンスイベント内で送受信されるパラメータ値はシリアライズされ、接続されているすべてのクライアント（エディタ UI、MCP サーバー）にブロードキャスト送信されます。

どちらのパスも事前に値のサイズをチェックしないため、デフォルトでは大きな値によって保存ワークフローファイルとすべての接続クライアント宛ての通信トラフィックの両方が肥大化してしまいます。ノードが使用する基盤 API が許す限り、大容量のバイナリデータ（画像、音声、動画、3D アセット、モデルの重みなど）はバイト列をインライン化するのではなく、ファイルパスや URL などの参照形式で保持してください。

Parameter(serializable=False)（パラメータ属性 を参照）は 最初のパスのみ をカバーします。これは、ドライバー、ファイルハンドル、一時的な大容量バッファなど、永続化すべきでない値がワークフローファイルに保存されるのを防ぐ適切な選択肢ですが、2 番目のパスには何の影響も及ぼしません（接続されているすべてのクライアントに値が送信され続けます）。WebSocket パスをパラメータごとにオプトアウトする設定は存在しないため、値のサイズ自体を小さく保つことが唯一の制御手段となります。なお、出力パラメータにおいてこの宣言を行うと、値を生成したプロセス内に保持し、ワーカープロセスの境界を越えてキーのみを送信するようになります — 詳細は シリアライズできない値の受け渡し を参照してください。

パラメータ値は小さく保つこと (Keep parameter values small)

griptape.artifacts.BlobArtifact は生のバイト列を保持し、ImageArtifact と AudioArtifact はどちらもそのサブクラスです — したがって、これらをパラメータ型として使用するノードは、接続されているすべてのクライアントにバイトペイロード全体を送信し、パラメータが serializable=False と宣言されていない限り、保存されたワークフローにも書き出します。代わりに ImageUrlArtifact / AudioUrlArtifact を使用してください（ParameterImage / ParameterAudio ヘルパークラスがこれらを強制します — パラメータ を参照）。これらは参照先のファイルがどれほど巨大であっても、短い URL 文字列のみを保持します。

生のバイト列を保持するのは BlobArtifact とそのサブクラスに限りません — ThreeDArtifact もバイト列を保持します（ただし、WebSocket 経由でバイト列を送信することはありません）。型名が安全そうに見えるかどうかではなく、そのパラメータが実際に保持する値のサイズによって判断してください。

インポートのベストプラクティス (Import Best Practices)

依存関係は常に関数内ではなくモジュールレベルでインポートしてください：

❌ アンチパターン - 条件付き/遅延インポート:

def _get_image_data(self, image_artifact):
    try:
        from PIL import Image  # これを行わないでください
        from io import BytesIO
        img = Image.open(BytesIO(image_bytes))

✅ 推奨 - モジュールレベルのインポート:

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


def _get_image_data(self, image_artifact):
    img = Image.open(BytesIO(image_bytes))

理由:

• 依存関係が明確かつ可視化される
• ファイル全体での冗長なインポートを回避できる
• Python のベストプラクティス（PEP 8）に準拠
• 欠落している依存関係を早期に検出しやすい
• IDE のサポートやコード補完が向上する

例外: インストールされていない可能性のある、真にオプションな依存関係に対してのみ条件付きインポートを使用します：

def process(self) -> None:
    try:
        from huggingface_hub import HfApi
    except ImportError:
        error_msg = "huggingface_hub ライブラリがインストールされていません"
        self.parameter_output_values["output"] = None
        raise ImportError(error_msg)

インポートの整理 (Import Organization)

標準的な順序でインポートを整理し、グループ間に空行を挟みます：

# 標準ライブラリのインポート
import base64
import logging
from typing import Any

# サードパーティライブラリのインポート
import requests
from PIL import Image

# ローカル/Griptape のインポート
from griptape_nodes.exe_types.core_types import Parameter, ParameterMode
from griptape_nodes.exe_types.node_types import DataNode
from griptape_nodes.retained_mode.griptape_nodes import GriptapeNodes

サードパーティライブラリの型チェック (Type Checking for Third-Party Libraries)

サードパーティライブラリをインポートする際、型チェックエラーが発生する場合があります。状況に応じて適切な type: ignore コメントを使用してください：

シナリオ 1: ライブラリはインストールされているが型スタブが存在しない場合

インストールされているものの型アノテーションが不足しているライブラリ（sklearn、ultralytics、supervision など）：

# ✅ ライブラリは存在するが型スタブがない
from sklearn.cluster import KMeans  # type: ignore[import-untyped]
from ultralytics import YOLO  # type: ignore[import-untyped]
from supervision import Detections  # type: ignore[import-untyped]

シナリオ 2: CI 型チェック環境にライブラリがインストールされていない場合

実行時の依存関係ではあるものの、CI の型チェック環境にはインストールされていないライブラリ（color-matcher、特殊な処理ライブラリなど）：

# ✅ 型チェック環境にライブラリがインストールされていない
from color_matcher import ColorMatcher  # type: ignore[reportMissingImports]
from color_matcher.normalizations import norm_img_to_uint8  # type: ignore[reportMissingImports]

使い分けの基準

エラー型	コメント	使用する場面
import-untyped	# type: ignore[import-untyped]	ライブラリはインストール済みだが、型スタブがない
reportMissingImports	# type: ignore[reportMissingImports]	CI 環境に対象ライブラリが存在しない

一般的なガイドライン:

• 両方が機能する場合は、より精緻な import-untyped を推奨します
• 型チェック中にライブラリが利用できない場合は reportMissingImports が必要です
• 実際にどのエラーが発生しているかを確認するには、CI ログをチェックしてください

関数パラメータの管理 (Function Parameter Management)

データクラス（dataclass）を活用して、関数の引数の数を少なく（6個未満に）抑えます：

❌ アンチパターン - 多すぎるパラメータ:

def process_bbox(self, x: int, y: int, width: int, height: int,
                 dilation_percent: float, img_width: int, img_height: int):
    # バウンディングボックスの処理

✅ 推奨 - dataclass の利用:

from dataclasses import dataclass

@dataclass
class BoundingBox:
    x: int
    y: int
    width: int
    height: int
    dilation_percent: float
    img_width: int
    img_height: int

def process_bbox(self, bbox: BoundingBox):
    # bbox.x, bbox.y などを使ってバウンディングボックスを処理

メリット:

• 可読性の向上
• 型安全性の確保
• 保守性の向上
• 自己文書化されたコード

コード品質 (Code Quality)

追加の静的解析（リンティング）ベストプラクティス:

• すべての行（空行を含む）から行末の余分な空白（trailing whitespace）を削除する
• 一貫したインデントを使用する（タブではなくスペースのみ）
• 可能な限り行の長さを 120 文字未満に抑える
• 分かりやすい変数名を使用する
• 不要な Python パッケージボイラープレートの追加を避ける。実際にパッケージ化したい場合（または採用したパッケージ方式で必要な場合）にのみ __init__.py ファイルを作成してください。

コミット前チェック（必須）

griptape-nodes でコミットを行う前に、フォーマットとチェックを実行し、エラーを修正してください：

make format
make check/lint
make check/types

ノードドキュメントとナビゲーション

コアライブラリに新しいノードを追加する際は、ノードリファレンスドキュメントも追加してください：

• ドキュメントページを作成: docs/nodes/<category>/<node>.md
• mkdocs.yml に追加: nav -> Nodes Reference -> <Category> 配下

よくある落とし穴

• リポジトリ全体の lint/型チェックは、Git 管理外（untracked） のファイルに対しても問題を検出することがあります。チェック実行時や PR 作成時に、リポジトリ内に追跡されていないフォルダやファイル（コピーした一時フォルダなど）を残さないようにしてください。
• ruff が関数の複雑度（例: C901, PLR0912）を警告した場合、警告を抑制するのではなく、小さなヘルパー関数にリファクタリングすることを推奨します。
• parent_container_name ≠ parent_element_name: これら 2 つの Parameter 属性は似ていますが、まったく異なる目的で使用されます。parent_container_name は ParameterContainer（リストや辞書の所有権）用であり、parent_element_name は ParameterGroup（UI 上の視覚的グルーピング）用です。これらを混同すると、パラメータがノードのルートに配置されたり、実行間のクリーンアップが行われなくなったり、保存/再読み込み時に通知なしで消失したりします。完全な違いについては コンテナ セクションを参照してください。

本番環境向けエラー処理 (Production Error Handling)

包括的なバリデーション

複雑な検証には validate_before_node_run() を使用します：

def validate_before_node_run(self) -> list[Exception] | None:
    """ノード実行前にパラメータを検証します。"""
    exceptions = []

    model = self.get_parameter_value("model")
    if model == "advanced":
        images = self.get_parameter_list_value("images") or []
        if len(images) > MAX_IMAGES:
            exceptions.append(ValueError(f"{self.name}: 最大 {MAX_IMAGES} 枚の画像のみ許可されています（現在: {len(images)} 枚）"))

    return exceptions if exceptions else None

接続バリデーションパターン

複数の接続要件を持つ複雑なノードの場合：

def _validate_iterative_connections(self) -> list[Exception]:
    """必要なすべての接続が適切に確立されているかを検証します。"""
    errors = []
    node_type = self._get_base_node_type_name()

    # exec_out に送信接続が存在するか確認
    if not _outgoing_connection_exists(self.name, self.exec_out.name):
        errors.append(
            Exception(
                f"{self.name}: 'On Each Item' からの必要な接続が不足しています。"
                f"必要なアクション: {node_type} Start を内部のループノードに接続してください。"
                "ループ本体を実行するには、Start ノードを他のノードに接続する必要があります。"
            )
        )

    # ループに End への送信接続が存在するか確認
    if self.end_node is None:
        errors.append(
            Exception(
                f"{self.name}: 必要なテザリング接続が不足しています。"
                f"必要なアクション: {node_type} Start の 'Loop End Node' を {node_type} End の 'Loop Start Node' に接続してください。"
                "これにより、Start ノードと End ノードの間の明示的な関係が確立されます。"
            )
        )

    return errors

ベストプラクティス: どの接続が不足しており、どのように修正すればよいかをユーザーに正確に伝える、詳細で実用的なエラーメッセージを提供してください。

安全なデフォルト値パターン (Safe Defaults Pattern)

例外を送出する前に、必ず安全なデフォルト値を設定してください：

def _set_safe_defaults(self) -> None:
    """すべての出力に安全なデフォルト値を設定します。"""
    self.parameter_output_values["result"] = None
    self.parameter_output_values["status"] = "error"
    self.parameter_output_values["count"] = 0


def process(self) -> None:
    try:
        # 処理ロジック
        result = process_data()
        self.parameter_output_values["result"] = result
    except Exception as e:
        self._set_safe_defaults()
        raise RuntimeError(f"処理に失敗しました: {str(e)}") from e

安全な URL 構築

安全な URL 構築には urllib.parse.urljoin() を使用します：

from urllib.parse import urljoin
import os


def __init__(self, **kwargs):
    super().__init__(**kwargs)

    # 安全な URL 構築
    base = os.getenv("API_BASE_URL", "https://api.example.com")
    base_slash = base if base.endswith("/") else base + "/"
    api_base = urljoin(base_slash, "api/")
    self._endpoint = urljoin(api_base, "v1/process/")

ロギングのベストプラクティス (Logging Best Practices)

安全なロギングパターン

ロギングの失敗によって処理全体の実行が中断されるのを防ぎます：

from contextlib import suppress
import logging

logger = logging.getLogger(__name__)


def _log(self, message: str) -> None:
    """例外を抑制した安全なロギング。"""
    with suppress(Exception):
        logger.info(message)

リクエストのサニタイズ（機密データの保護）

ログ内の機密データをサニタイズ（マスキング）します：

from copy import deepcopy
import json

PROMPT_TRUNCATE_LENGTH = 100


def _log_request(self, payload: dict[str, Any]) -> None:
    """機密データをサニタイズしてリクエストをログ出力します。"""
    with suppress(Exception):
        sanitized_payload = deepcopy(payload)

        # 長いプロンプトを切り詰める
        prompt = sanitized_payload.get("prompt", "")
        if len(prompt) > PROMPT_TRUNCATE_LENGTH:
            sanitized_payload["prompt"] = prompt[:PROMPT_TRUNCATE_LENGTH] + "..."

        # Base64 画像データをマスクする
        if "image" in sanitized_payload:
            image_data = sanitized_payload["image"]
            if isinstance(image_data, str) and image_data.startswith("data:image/"):
                parts = image_data.split(",", 1)
                header = parts[0] if parts else "data:image/"
                b64_len = len(parts[1]) if len(parts) > 1 else 0
                sanitized_payload["image"] = f"{header},<base64 data length={b64_len}>"

        self._log(f"Request: {json.dumps(sanitized_payload, indent=2)}")