Nodes are the fundamental building blocks of ComfyUI workflows. Each node represents a specific operation in the image generation pipeline, from loading models to applying effects. Nodes communicate through typed connections and execute in dependency order.
From execution.py:152-184, inputs are resolved at runtime:
def get_input_data(inputs, class_def, unique_id, execution_list=None): input_data_all = {} for x in inputs: input_data = inputs[x] if is_link(input_data): input_unique_id = input_data[0] output_index = input_data[1] # Get cached output from connected node cached = execution_list.get_cache(input_unique_id, unique_id) if cached is None or cached.outputs is None: # Input not yet available continue obj = cached.outputs[output_index] input_data_all[x] = obj else: # Direct value (not a link) input_data_all[x] = [input_data] return input_data_all
Lazy Evaluation: Nodes can defer input evaluation until needed using the check_lazy_status method.
Nodes can process batched inputs using the map-over-list pattern:
class MyNode: INPUT_IS_LIST = False # Process items individually OUTPUT_IS_LIST = [False, True] # Second output is a list def process(self, image, strength): # Automatically called for each item result = apply_effect(image, strength) return (result, [metadata])
From execution.py:232-308, the execution system handles batching:
async def _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, func): input_is_list = getattr(obj, "INPUT_IS_LIST", False) if input_is_list: # Pass all inputs as lists await process_inputs(input_data_all, 0, input_is_list=True) else: # Process each item individually max_len = max(len(x) for x in input_data_all.values()) for i in range(max_len): input_dict = slice_dict(input_data_all, i) await process_inputs(input_dict, i)
class MyNode: @classmethod def VALIDATE_INPUTS(s, **kwargs): if 'image' in kwargs: img = kwargs['image'] if img.shape[2] < 512: return "Image width must be at least 512px" return True
From execution.py:971-998, validation is called before execution:
if len(validate_function_inputs) > 0: input_data_all = get_input_data(inputs, obj_class, unique_id) input_filtered = {} for x in input_data_all: if x in validate_function_inputs: input_filtered[x] = input_data_all[x] ret = await _async_map_node_over_list( prompt_id, unique_id, obj_class, input_filtered, validate_function_name ) for r in ret: if r is not True: errors.append({ "type": "custom_validation_failed", "message": "Custom validation failed", "details": str(r) })
if "hidden" in valid_inputs: h = valid_inputs["hidden"] if h.get("UNIQUE_ID"): input_data_all["unique_id"] = [unique_id] if h.get("PROMPT"): input_data_all["prompt"] = [dynprompt.get_original_prompt()]
class CheckpointLoader: @classmethod def IS_CHANGED(s, ckpt_name): # Return different value when file changes ckpt_path = folder_paths.get_full_path("checkpoints", ckpt_name) m = hashlib.sha256() with open(ckpt_path, 'rb') as f: m.update(f.read()) return m.digest().hex()