comfyui-node-lifecycle

ComfyUI ノード実行ライフサイクル

実行ライフサイクルを理解すると、効率的で正しいノードを構築できます。

実行フローの概要

1. Prompt received from frontend
2. Validation phase
   ├── Look up each node class
   ├── Call INPUT_TYPES() / define_schema() for input specs
   ├── Validate connections and types
   ├── Call validate_inputs() for each node
   └── Validate loop boundaries (StartLoop/EndLoop pairing, v0.36+)
3. Build execution order (topological sort from output nodes)
4. For each node in order:
   ├── Cache check (fingerprint_inputs)
   ├── Input resolution (get upstream values)
   ├── Lazy evaluation (check_lazy_status)
   ├── Execute function
   └── Store outputs in cache
5. Return results to frontend

実行順序

ComfyUI は出力ノードから逆向きに実行します。

1. 出力ノード（is_output_node=True）を特定します
2. 依存関係グラフを構築します
3. トポロジカルソートで実行順序を決定します
4. 出力ノードに接続されたノードのみを実行します

キャッシュ制御：fingerprint_inputs（V3）/ IS_CHANGED（V1）

ノードを再実行するか、キャッシュされた結果を使用するかを制御します。

class RandomNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="RandomNode",
            display_name="Random Value",
            category="utils",
            inputs=[
                io.Float.Input("min_val", default=0.0),
                io.Float.Input("max_val", default=1.0),
            ],
            outputs=[io.Float.Output("FLOAT")],
        )

    @classmethod
    def fingerprint_inputs(cls, min_val, max_val):
        """Return value compared to last run. Different value = re-execute."""
        # Return unique value each time to always re-execute
        import time
        return time.time()

    @classmethod
    def execute(cls, min_val, max_val):
        import random
        return io.NodeOutput(random.uniform(min_val, max_val))

キャッシュの仕組み：

• 実行前に、fingerprint_inputs() が execute() と同じ引数で呼び出されます
• 戻り値が前回実行時の戻り値と比較されます
• 同じ場合 → 実行をスキップし、キャッシュされた出力を使用します
• 異なる場合 → ノードを再実行します
• fingerprint_inputs が定義されていない場合 → 入力値に基づいてキャッシュします

V1 相当（IS_CHANGED）：

@classmethod
def IS_CHANGED(s, min_val, max_val):
    return time.time()  # always re-execute

not_idempotent フラグ

キャッシュを一切行わないノードの場合：

io.Schema(
    node_id="AlwaysRunNode",
    not_idempotent=True,  # prevents cache sharing between instances of the same node
    # ...
)

重要： not_idempotent=True を指定しても、後続の実行でノード自身のキャッシュ済み出力が再利用されるのを防ぐことはできません。この指定で防げるのは、同じ入力を持つ同じノードタイプの異なるインスタンス間でのキャッシュ共有のみです。実行のたびに再実行させるには（ファイル書き込みノードなど）、毎回一意の値を返す fingerprint_inputs（V3）または IS_CHANGED（V1）も実装する必要があります。

has_intermediate_output フラグ

中間出力を生成するインタラクティブ UI を持つノード（Image Crop、Painter など）の場合に使用します。これらは出力ノードのように動作します（UI の結果がキャッシュされ、ページの再読み込み時にフロントエンドへ再送されます）が、自動的に実行リストへ追加されることはありません。実際の出力ノードへの依存経路上にある場合にのみ実行されます。

io.Schema(
    node_id="InteractiveCropNode",
    has_intermediate_output=True,
    # ...
)

外部キャッシュプロバイダー

CacheProvider を登録すると、ComfyUI インスタンス間でノードのキャッシュ済み出力を共有できます（共有ネットワークキャッシュなど）。

from comfy_api.latest import Caching

class MyCacheProvider(Caching.CacheProvider):
    async def on_lookup(self, context):   # context: node_id, class_type, cache_key_hash
        ...  # return Caching.CacheValue(outputs=[...], ui={...}) or None on miss

    async def on_store(self, context, value):
        ...  # store to external storage (dispatched as a background task)

    def should_cache(self, context, value=None) -> bool:
        return True  # return False to skip external caching for a node

    def on_prompt_start(self, prompt_id): ...
    def on_prompt_end(self, prompt_id): ...

# Register in ComfyExtension.on_load():
api = ComfyAPI()
await api.caching.register_provider(MyCacheProvider())

ローカルキャッシュにヒットしなかった場合、登録順にプロバイダーが参照されます。プロバイダーで例外が発生しても、実行が妨げられることはありません。

キャッシュにヒットすると、キャッシュ済みの UI が再送されます（UI エントリから id／preview_id が取り除かれます。詳しくは comfyui-node-outputs を参照してください）。カスタム出力オブジェクトの RAM キャッシュ容量については、comfyui-node-datatypes の _comfy_cache_tensors を参照してください。

入力の検証：validate_inputs（V3）/ VALIDATE_INPUTS（V1）

実行前に入力を検証します。検証フェーズ中に実行されます。

class ValidatedNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="ValidatedNode",
            display_name="Validated Node",
            category="utils",
            inputs=[
                io.Int.Input("width", default=512, min=1, max=8192),
                io.Int.Input("height", default=512, min=1, max=8192),
            ],
            outputs=[io.Image.Output("IMAGE")],
        )

    @classmethod
    def validate_inputs(cls, width, height):
        """Return True if valid, or error string if invalid."""
        if width % 8 != 0 or height % 8 != 0:
            return "Width and height must be multiples of 8"
        if width * height > 4096 * 4096:
            return "Total pixels exceed maximum (4096x4096)"
        return True

    @classmethod
    def execute(cls, width, height):
        import torch
        return io.NodeOutput(torch.zeros(1, height, width, 3))

V1 相当：

@classmethod
def VALIDATE_INPUTS(s, width, height):
    if width % 8 != 0:
        return "Width must be a multiple of 8"
    return True

validate_inputs の動的入力（v0.40+）

Autogrow／DynamicCombo 入力のルート ID を指定すると、再構築されたネスト値を受け取れます。指定していない動的ルートは渡されません。**kwargs にはネストされたすべての値が渡されます。

@classmethod
def validate_inputs(cls, rows):          # rows = {"item0": 0.5, ...} (Autogrow)
    return True if rows else "Connect at least one row"

• 入力が渡されていない場合でも、False／文字列を返すと報告されます。
• 不正な動的キーがあると、エラータイプ invalid_dynamic_input で検証に失敗します。

型検証のスキップ

任意の型（ワイルドカード入力）を受け入れるには、input_types パラメーターを含めます。

@classmethod
def validate_inputs(cls, input_types: dict = None, **kwargs):
    # input_types contains the actual types of connected inputs
    # Returning True skips the default type checking
    return True

遅延評価：check_lazy_status

実際に評価する必要がある遅延入力を制御します。詳しくは comfyui-node-inputs を参照してください。

@classmethod
def check_lazy_status(cls, condition, value_a=None, value_b=None):
    """Called before execute. Return names of inputs that need evaluation."""
    if condition and value_a is None:
        return ["value_a"]
    if not condition and value_b is None:
        return ["value_b"]
    return []

主な動作：

• ノードに遅延入力がある場合にのみ呼び出されます
• 入力が利用可能になるたびに、複数回呼び出されることがあります
• 未評価の遅延入力は None です
• 実行可能になったら、空のリスト（または None）を返します
• 評価済み入力は呼び出しをまたいで値を保持します

出力ノード

is_output_node=True を持つノードは実行の起点です。ComfyUI はこれらのノードから逆向きにたどります。

class SaveMyData(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="SaveMyData",
            display_name="Save Data",
            category="output",
            is_output_node=True,  # marks as output node
            inputs=[
                io.String.Input("data"),
                io.String.Input("filename", default="output.txt"),
            ],
            outputs=[],  # output nodes may have no outputs
            hidden=[io.Hidden.prompt, io.Hidden.extra_pnginfo],
        )

    @classmethod
    def execute(cls, data, filename):
        import folder_paths, os
        output_dir = folder_paths.get_output_directory()
        with open(os.path.join(output_dir, filename), 'w') as f:
            f.write(data)
        return io.NodeOutput()

リスト処理

リストの受け取り

# V3: is_input_list=True in Schema (same as V1 INPUT_IS_LIST)
# All inputs arrive as lists — including widget values like batch_size
# Widget values: use widget_value[0] to get the scalar
# Shorter lists are padded by repeating the last value

# V1: INPUT_IS_LIST = True to receive full lists
class ListNode:
    INPUT_IS_LIST = True
    # Now execute() receives lists instead of individual items

リストの出力

# V3
io.Image.Output("IMAGE", is_output_list=True)

# V1
OUTPUT_IS_LIST = (True,)  # tuple matching RETURN_TYPES

エラー処理

@classmethod
def execute(cls, image, model):
    try:
        result = model.process(image)
    except RuntimeError as e:
        if "out of memory" in str(e):
            import torch
            torch.cuda.empty_cache()
            # Try with smaller batch
            result = process_in_chunks(image, model)
        else:
            raise
    return io.NodeOutput(result)

サーバー通信

実行中にフロントエンドへメッセージを送信します。

from server import PromptServer

@classmethod
def execute(cls, data):
    PromptServer.instance.send_sync(
        "my_extension.status",
        {"message": "Processing complete", "progress": 100}
    )
    # Status text shown on the node (binary TEXT event; sid=None broadcasts)
    PromptServer.instance.send_progress_text("Step 3 / 10", cls.hidden.unique_id)
    return io.NodeOutput(data)

• "prompt_id" を含む辞書ペイロードには、プロンプトの workflow_metadata（workflow_id など）がマージされます（v0.40+）。クライアントがメッセージを該当するワークフローへルーティングできるよう、prompt_id を含めてください。

ロギング

loggingを使用します。INFO より詳細な診断ログには、DETAIL レベル（v0.30 以降）を使用します：from comfy.internal_logging import detail; detail("msg %s", x)。--verbose DETAILで表示するか、--verbose DETAIL path.logで書き込みます。（モジュールはcomfy.loggingではなく、comfy.internal_loggingです。）

ライフサイクル全体の例

import time
import torch
from comfy_api.latest import ComfyExtension, io, ComfyAPISync

class FullLifecycleNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="FullLifecycleNode",
            display_name="Full Lifecycle Demo",
            category="example",
            inputs=[
                io.Image.Input("image"),
                io.Float.Input("threshold", default=0.5, min=0.0, max=1.0),
                io.Image.Input("optional_ref", optional=True, lazy=True),
            ],
            outputs=[
                io.Image.Output("IMAGE"),
                io.Mask.Output("MASK"),
            ],
            hidden=[io.Hidden.unique_id],
        )

    @classmethod
    def validate_inputs(cls, image, threshold, optional_ref=None):
        if threshold == 0.0:
            return "Threshold cannot be exactly 0"
        return True

    @classmethod
    def fingerprint_inputs(cls, image, threshold, optional_ref=None):
        # Re-execute if threshold changed; cache otherwise
        return threshold

    @classmethod
    def check_lazy_status(cls, image, threshold, optional_ref=None):
        # Only request optional_ref if threshold is high
        if threshold > 0.8 and optional_ref is None:
            return ["optional_ref"]
        return []

    @classmethod
    def execute(cls, image, threshold, optional_ref=None):
        node_id = cls.hidden.unique_id

        api = ComfyAPISync()  # use ComfyAPISync in sync execute; ComfyAPI in async
        api.execution.set_progress(0, 2)

        # Generate mask from threshold
        gray = image[:, :, :, 0] * 0.299 + image[:, :, :, 1] * 0.587 + image[:, :, :, 2] * 0.114
        mask = (gray > threshold).float()

        api.execution.set_progress(1, 2)

        # Apply mask
        result = image * mask.unsqueeze(-1)
        if optional_ref is not None:
            result = result + optional_ref * (1 - mask.unsqueeze(-1))

        api.execution.set_progress(2, 2)
        return io.NodeOutput(result, mask)

関連項目

• comfyui-node-basics - ノード構造の基礎
• comfyui-node-inputs - 入力タイプと遅延評価
• comfyui-node-advanced - 展開、MatchType、DynamicCombo
• comfyui-node-outputs - UI 出力とプレビュー