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))

缓存工作方式：

• 执行前，会使用与 execute() 相同的参数调用 fingerprint_inputs()
• 将返回值与上一次运行的返回值进行比较
• 如果相同 → 跳过执行，使用缓存输出
• 如果不同 → 重新执行节点
• 如果未定义 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.internal_logging，而不是 comfy.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 输出和预览