comfyui-node-advanced

ComfyUI の高度なノードパターン（V3）

V3 では、動的で型安全かつ柔軟なノードを設計するための高度な入力パターンを利用できます。

MatchType - ジェネリック型接続

MatchType は、同じテンプレートを共有する入力と出力が、接続時に同じ型になることを保証します。型付き言語のジェネリクスと同様です。

class PassThrough(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        # Template(template_id, allowed_types=AnyType) - optional type constraint
        template = io.MatchType.Template("T")
        return io.Schema(
            node_id="PassThrough",
            display_name="Pass Through",
            category="utils",
            inputs=[
                io.MatchType.Input("value", template=template),
            ],
            outputs=[
                io.MatchType.Output(template=template, display_name="output"),
            ],
        )

    @classmethod
    def execute(cls, value):
        return io.NodeOutput(value)

ユーザーが IMAGE を入力に接続すると、出力は自動的に IMAGE 型になります。

スイッチノードのパターン

Core Switch（v0.35 以降）では、どちらか一方を未接続のままにできるよう、両方の分岐がオプションになっています。未接続のオプション入力は kwargs に含まれませんが、接続済みで未評価の遅延入力は None になります。センチネルを使って区別します。

MISSING = object()

class Switch(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        template = io.MatchType.Template("switch")
        return io.Schema(
            node_id="Switch",
            display_name="Switch",
            category="logic",
            inputs=[
                io.Boolean.Input("switch"),
                io.MatchType.Input("on_false", template=template, lazy=True, optional=True),
                io.MatchType.Input("on_true", template=template, lazy=True, optional=True),
            ],
            outputs=[
                io.MatchType.Output(template=template, display_name="output"),
            ],
        )

    @classmethod
    def check_lazy_status(cls, switch, on_false=MISSING, on_true=MISSING):
        if switch and on_true is None:      # connected, not yet evaluated
            return ["on_true"]
        if not switch and on_false is None:
            return ["on_false"]

    @classmethod
    def execute(cls, switch, on_true=MISSING, on_false=MISSING):
        selected = on_true if switch else on_false
        return io.NodeOutput(None if selected is MISSING else selected)

MultiType - 複数の型を受け付ける

複数の異なる型を受け付ける単一の入力です。

io.MultiType.Input("data",
    types=[io.Image, io.Mask, io.Latent],
    optional=True,
)

Autogrow - 動的に増える入力

ユーザーが入力スロットに接続すると、スロットが自動的に追加されます。テンプレートモードは 2 種類です。

TemplatePrefix（番号付きスロット）

class ConcatImages(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="ConcatImages",
            display_name="Concat Images",
            category="image",
            inputs=[
                io.Autogrow.Input("images",
                    template=io.Autogrow.TemplatePrefix(
                        input=io.Image.Input("img"),  # template for each slot
                        prefix="image_",              # slot names: image_0, image_1, ...
                        min=2,                        # minimum visible slots (default 1)
                        max=16,                       # maximum slots (default 10, hard limit 100)
                    ),
                ),
            ],
            outputs=[io.Image.Output("IMAGE")],
        )

    @classmethod
    def execute(cls, images: io.Autogrow.Type):
        # images is a dict: {"image_0": tensor, "image_1": tensor, ...}
        tensors = [v for v in images.values() if v is not None]
        return io.NodeOutput(torch.cat(tensors, dim=0))

TemplateNames（名前付きスロット）

io.Autogrow.Input("inputs",
    template=io.Autogrow.TemplateNames(
        input=io.Float.Input("val"),
        names=["red", "green", "blue", "alpha"],  # specific slot names
        min=3,  # first 3 are required
    ),
)
# Creates slots: "red" (required), "green" (required), "blue" (required), "alpha" (optional)

主な動作：

• テンプレート内のウィジェット入力は、接続専用（force_input=True）に強制されます
• min 未満のスロットは必須で、min を超えるスロットはオプションです
• 名前は合計 100 個までです

DynamicCombo - 条件付き入力

各オプションで異なるサブ入力が表示されるコンボドロップダウンです。

class ProcessNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="ProcessNode",
            display_name="Process Node",
            category="processing",
            is_output_node=True,
            inputs=[
                io.DynamicCombo.Input("mode", options=[
                    io.DynamicCombo.Option("resize", [
                        io.Int.Input("width", default=512, min=1, max=8192),
                        io.Int.Input("height", default=512, min=1, max=8192),
                    ]),
                    io.DynamicCombo.Option("blur", [
                        io.Float.Input("radius", default=5.0, min=0.1, max=100.0),
                    ]),
                    io.DynamicCombo.Option("sharpen", [
                        io.Float.Input("amount", default=1.0, min=0.0, max=10.0),
                    ]),
                ]),
                io.Image.Input("image"),
            ],
            outputs=[io.Image.Output("IMAGE")],
        )

    @classmethod
    def execute(cls, mode: io.DynamicCombo.Type, image, **kwargs):
        # mode is a dict with the combo value + sub-inputs
        # key for selected option matches the DynamicCombo input ID
        if mode["mode"] == "resize":
            width = mode["width"]
            height = mode["height"]
            # ... resize logic
        return io.NodeOutput(image)

ネストされた DynamicCombo：

io.DynamicCombo.Input("outer", options=[
    io.DynamicCombo.Option("option1", [
        io.DynamicCombo.Input("inner", options=[
            io.DynamicCombo.Option("sub1", [io.Float.Input("val")]),
            io.DynamicCombo.Option("sub2", [io.Int.Input("count")]),
        ])
    ]),
])

DynamicSlot - 接続をトリガーとする入力

DynamicCombo と同様ですが、コンボオプションが選択されたときではなく、スロットに接続されたときにサブ入力が表示されます。

io.DynamicSlot.Input(
    slot=io.Image.Input("image"),        # the trigger slot (widget inputs are forced to connection-only)
    inputs=[                              # revealed when the slot is connected
        io.Float.Input("blend", default=0.5),
        io.Boolean.Input("invert", default=False),
    ],
)
# Value type: dict containing the slot value + sub-input values

注意：DynamicSlot は comfy_api.latest に登録されたインフラストラクチャですが、コアノードではまだ使用されていません。実験的な機能として扱ってください。

_DynamicGroup（COMFY_DYNAMICGROUP_V3、ウィジェットの繰り返し行で、コアの LoRA スタックノードで使用）は内部実装で不安定です。io からエクスポートされていないため、comfy_api.latest._io からインポートしないでください。フロントエンドのサポートは frontend v1.57.0 で追加されました。

動的入力内の遅延入力

Autogrow / DynamicCombo のサブ入力には lazy=True を指定できます（スケジューリングは v0.40 で修正済みです）。check_lazy_status 内でのみ、すべてのネストされた値は (value, original_key) タプルとして渡されます。入力を要求するには original_key を返してください。execute には通常の値が渡されます。

io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(
    io.Image.Input("img", lazy=True), prefix="item", min=1))

@classmethod
def check_lazy_status(cls, rows):
    value, key = rows["item0"]          # key == "rows.item0"
    return [key] if value is None else []

ループ（コアの StartLoop / EndLoop）

v0.36 以降、コアには StartLoop / EndLoop（utilities/looping、Simple / For / List モード、値の引き継ぎ、累積）が含まれています。カスタムのループノードよりも、これらを優先して使用してください。作成者に関係する注意点は次のとおりです。

• ループ本体はノード展開によって反復ごとに複製されるため、ループ内のノードは反復ごとに 1 回実行され、cls.hidden.unique_id は展開後の ID になります（表示 ID は元のノードに対応します）。
• io.Schema(loop_boundary="start" | "end") は、プロンプト検証用にループ境界を示します（ペアの整合性、本体からの逸脱がないこと、accumulate に本体の出力が入力されないこと）。コアのループ実装に依存するため、実験的な機能として扱ってください。

ノード展開 - サブグラフの挿入

ノードは、実行中に自分自身を置き換えるサブグラフを返せます。

from comfy_execution.graph_utils import GraphBuilder

class RepeatNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="RepeatNode",
            display_name="Repeat KSampler",
            category="sampling",
            enable_expand=True,
            inputs=[
                io.Model.Input("model"),
                io.Int.Input("repeat_count", default=2, min=1, max=10),
                io.Latent.Input("latent"),
            ],
            outputs=[io.Latent.Output("LATENT")],
        )

    @classmethod
    def execute(cls, model, repeat_count, latent):
        graph = GraphBuilder()
        current_latent = latent
        for i in range(repeat_count):
            sampler = graph.node("KSampler",
                model=model,
                latent_image=current_latent,
                # ... other params
            )
            current_latent = sampler.out(0)
        return io.NodeOutput(current_latent, expand=graph.finalize())

ノード展開の主なルール：

• Schema で enable_expand=True を設定します
• GraphBuilder を使って安全にサブグラフを構築します
• io.NodeOutput(output_ref, expand=graph.finalize()) を返します
• サブグラフ内のノード ID は決定的かつ一意である必要があります
• 各サブノードは個別にキャッシュされます

すべての入力を受け付ける

スキーマで定義されていない任意の入力を受け付けます。

class FlexibleNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="FlexibleNode",
            display_name="Flexible Node",
            category="utils",
            accept_all_inputs=True,
            inputs=[io.Combo.Input("mode", options=["a", "b"])],
            outputs=[io.String.Output()],
        )

    @classmethod
    def validate_inputs(cls, mode, **kwargs):
        return True  # skip validation for dynamic inputs

    @classmethod
    def execute(cls, mode, **kwargs):
        # kwargs contains all dynamic inputs
        return io.NodeOutput(str(kwargs))

実行のブロック

条件に応じて、下流の実行を防止します。

class GateNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="GateNode",
            display_name="Gate",
            category="logic",
            inputs=[
                io.Boolean.Input("allow"),
                io.Image.Input("image"),
            ],
            outputs=[io.Image.Output("IMAGE")],
        )

    @classmethod
    def execute(cls, allow, image):
        if not allow:
            return io.NodeOutput(block_execution="Gate is closed")
        return io.NodeOutput(image)

非同期実行

V3 は非同期実行をネイティブにサポートしています。

class AsyncNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="AsyncNode",
            display_name="Async Node",
            category="utils",
            inputs=[io.String.Input("url")],
            outputs=[io.String.Output()],
        )

    @classmethod
    async def execute(cls, url):
        import aiohttp
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                text = await response.text()
        return io.NodeOutput(text)

進捗状況の報告

時間のかかる処理中に進捗状況を報告します。

from comfy_api.latest import ComfyAPISync  # sync version; use ComfyAPI + await for async execute

class SlowNode(io.ComfyNode):
    @classmethod
    def define_schema(cls):
        return io.Schema(
            node_id="SlowNode",
            display_name="Slow Node",
            category="utils",
            inputs=[io.Int.Input("steps", default=100)],
            outputs=[io.String.Output()],
        )

    @classmethod
    def execute(cls, steps):
        api = ComfyAPISync()
        for i in range(steps):
            # ... do work ...
            api.execution.set_progress(i + 1, steps)
        return io.NodeOutput("done")

NodeReplace - ノード間の移行

置換を登録すると、古いワークフローが新しいノードに自動的に移行されます。

from typing_extensions import override
from comfy_api.latest import ComfyAPI, ComfyExtension, io

class MyExtension(ComfyExtension):
    @override
    async def on_load(self):
        api = ComfyAPI()
        await api.node_replacement.register(io.NodeReplace(
            new_node_id="MyNewNode_v2",
            old_node_id="MyOldNode",
            old_widget_ids=["width", "height", "mode"],  # positional widget order
            input_mapping=[
                {"new_id": "image_in", "old_id": "image"},     # rename input
                {"new_id": "size", "set_value": 512},           # set fixed value
            ],
            output_mapping=[
                {"new_idx": 0, "old_idx": 0},       # index-based, not name-based
            ],
        ))

    @override
    async def get_node_list(self):
        return [MyNewNodeV2]

InputMap の種類：

• InputMapOldId：{"new_id": str, "old_id": str} — 古い入力を新しい入力に対応付けます
• InputMapSetValue：{"new_id": str, "set_value": Any} — 新しい入力に固定値を設定します
• 自動拡張入力にはドット記法を使用します：{"new_id": "images.image0", "old_id": "image1"}

OutputMap（名前ではなくインデックスに基づきます）：

• {"new_idx": int, "old_idx": int} — 古い出力インデックスを新しい出力インデックスに対応付けます

old_widget_ids：ワークフローの JSON ではウィジェットの値が ID ではなく位置で保存されるため、必須です。このリストは位置インデックスを入力 ID に対応付け、正確な移行を可能にします。

ComfyAPI - ランタイム API

from comfy_api.latest import ComfyAPI, ComfyAPISync

# In sync execute(): use ComfyAPISync (no await)
api = ComfyAPISync()
api.execution.set_progress(value=50, max_value=100)
api.execution.set_progress(
    value=50, max_value=100,
    node_id=None,                   # optional: defaults to current node
    preview_image=pil_image,        # PIL Image or ImageInput tensor
    ignore_size_limit=False,
)

# In async execute(): use ComfyAPI (with await)
api = ComfyAPI()
await api.execution.set_progress(value=50, max_value=100)

# Node replacement registration (in async on_load)
await api.node_replacement.register(io.NodeReplace(...))

関連項目

• comfyui-node-basics - ノードの基本事項
• comfyui-node-inputs - 基本的な入力の種類
• comfyui-node-lifecycle - 実行ライフサイクルとキャッシュ
• comfyui-node-outputs - 出力の種類と UI ヘルパー