1
# 节点的核心组成部分
2
"""
3
1. INPUT_TYPES: 定义节点的输入参数
4
2. RETURN_TYPES: 定义节点的输出类型
5
3. FUNCTION: 执行逻辑的函数
6
4. CATEGORY: 节点在菜单中的分类
7
5. 可选方法: VALIDATE_INPUTS、IS_CHANGED 等
8
"""

1
ComfyUI/
2
├── custom_nodes/           # 自定义节点存放目录
3
│   ├── __init__.py
4
│   ├── your_node.py       # 你的节点文件
5
│   └── web/               # 前端文件（可选）
6
├── nodes.py               # 内置节点（参考学习）
7
├── comfy/
8
│   └── comfy_types/       # 类型定义（重要参考）
9
└── main.py

1
class NumberNode:
2
    """展示各种输入参数约束的节点"""
3

4
    RETURN_TYPES = ("INT", "FLOAT", "STRING")
5
    RETURN_NAMES = ("integer", "float", "validation_info")
6
    FUNCTION = "process"
7
    CATEGORY = "My Nodes/Numbers"
8

9
    @classmethod
10
    def INPUT_TYPES(cls):
11
        return {
12
            "required": {
13
                # 整数输入，带范围约束
14
                "integer_input": ("INT", {
15
                    "default": 10,
16
                    "min": 0,           # 最小值
17
                    "max": 100,         # 最大值
18
                    "step": 1,          # 步长
19
                }),
20
                # 浮点数输入
21
                "float_input": ("FLOAT", {
22
                    "default": 0.5,
23
                    "min": 0.0,
24
                    "max": 1.0,
25
                    "step": 0.01,
26
                    "round": 0.001,     # 保留小数位数
27
                }),
28
                # 带有提示选项的输入
29
                "preset": (["option_a", "option_b", "option_c"], {
30
                    "default": "option_a"
31
                }),
32
                # 多行文本输入
33
                "text_input": ("STRING", {
34
                    "default": "Hello",
35
                    "multiline": True,  # 允许换行输入
36
                }),
37
            },
38
            "optional": {
39
                # 可选输入（可以为空）
40
                "optional_value": ("INT", {
41
                    "default": 0,
42
                }),
43
            },
44
            "hidden": {
45
                # 隐藏输入，通常用于内部使用
46
                "node_id": "UNIQUE_ID",
47
            }
48
        }
49

50
    def process(self, integer_input, float_input, preset, text_input,
51
                optional_value=None, node_id=None):
52
        info = f"INT: {integer_input}, FLOAT: {float_input}, OPT: {preset}"
53
        return (int(integer_input), float(float_input), info)

1
from comfy.cli_args import NormalMode
2

3
class DynamicInputNode:
4
    """展示动态输入类型的节点"""
5

6
    RETURN_TYPES = ("*",)  # 任意类型输出
7
    FUNCTION = "passthrough"
8
    CATEGORY = "My Nodes/Advanced"
9

10
    @classmethod
11
    def INPUT_TYPES(cls):
12
        return {
13
            "required": {
14
                "anything": ("*", {}),  # 接受任意类型
15
            }
16
        }
17

18
    def passthrough(self, anything):
19
        # 原样返回输入
20
        return (anything,)

1
class MultiOutputNode:
2
    """一个节点输出多个结果的示例"""
3

4
    RETURN_TYPES = ("INT", "FLOAT", "STRING", "BOOLEAN")
5
    RETURN_NAMES = ("count", "ratio", "status", "success")
6
    FUNCTION = "analyze"
7
    CATEGORY = "My Nodes/Analysis"
8

9
    @classmethod
10
    def INPUT_TYPES(cls):
11
        return {
12
            "required": {
13
                "data_size": ("INT", {"default": 100, "min": 1, "max": 1000}),
14
                "threshold": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0}),
15
            }
16
        }
17

18
    def analyze(self, data_size, threshold):
19
        count = data_size
20
        ratio = threshold * 2
21
        status = f"Processed {count} items"
22
        success = ratio <= 1.0
23
        return (count, ratio, status, success)

1
import torch
2
import numpy as np
3
from PIL import Image
4

5
class ImageBrightnessNode:
6
    """调整图像亮度的节点"""
7

8
    @classmethod
9
    def INPUT_TYPES(cls):
10
        return {
11
            "required": {
12
                "image": ("IMAGE",),              # 图像输入
13
                "brightness": ("FLOAT", {
14
                    "default": 1.0,
15
                    "min": 0.0,
16
                    "max": 3.0,
17
                    "step": 0.1,
18
                    "round": 0.01
19
                }),
20
            }
21
        }
22

23
    RETURN_TYPES = ("IMAGE",)
24
    FUNCTION = "adjust_brightness"
25
    CATEGORY = "My Nodes/Image Processing"
26

27
    def adjust_brightness(self, image, brightness):
28
        """
29
        image 格式: torch.Tensor, shape 为 (batch, height, width, channels)
30
        值范围: [0, 1]
31
        """
32
        # 将 tensor 转换为 numpy 进行处理
33
        batch_size = image.shape[0]
34
        results = []
35

36
        for i in range(batch_size):
37
            img = image[i].cpu().numpy()
38

39
            # 转换为 PIL Image
40
            pil_img = Image.fromarray((img * 255).astype(np.uint8))
41

42
            # 调整亮度
43
            from PIL import ImageEnhance
44
            enhancer = ImageEnhance.Brightness(pil_img)
45
            adjusted = enhancer.enhance(brightness)
46

47
            # 转换回 tensor
48
            adjusted_array = np.array(adjusted).astype(np.float32) / 255.0
49
            results.append(adjusted_array)
50

51
        # 堆叠回 batch 格式
52
        output = torch.from_numpy(np.stack(results, axis=0))
53
        return (output,)

1
class ImageResizeNode:
2
    """调整图像尺寸的节点"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "image": ("IMAGE",),
9
                "width": ("INT", {"default": 512, "min": 1, "max": 4096}),
10
                "height": ("INT", {"default": 512, "min": 1, "max": 4096}),
11
            },
12
            "optional": {
13
                "maintain_ratio": ("BOOLEAN", {"default": True}),
14
                "resize_mode": (["crop", "stretch", "pad"], {"default": "stretch"}),
15
            }
16
        }
17

18
    RETURN_TYPES = ("IMAGE",)
19
    FUNCTION = "resize_image"
20
    CATEGORY = "My Nodes/Image Processing"
21

22
    def resize_image(self, image, width, height, maintain_ratio=True, resize_mode="stretch"):
23
        batch_size = image.shape[0]
24
        results = []
25

26
        for i in range(batch_size):
27
            img = image[i].cpu().numpy()
28
            pil_img = Image.fromarray((img * 255).astype(np.uint8))
29

30
            if maintain_ratio:
31
                # 保持宽高比
32
                orig_width, orig_height = pil_img.size
33
                ratio = min(width / orig_width, height / orig_height)
34
                new_width = int(orig_width * ratio)
35
                new_height = int(orig_height * ratio)
36

37
                if resize_mode == "stretch":
38
                    resized = pil_img.resize((width, height), Image.LANCZOS)
39
                elif resize_mode == "crop":
40
                    # 居中裁剪
41
                    resized = self._crop_to_size(pil_img, width, height)
42
                else:  # pad
43
                    resized = self._pad_to_size(pil_img, width, height)
44
            else:
45
                resized = pil_img.resize((width, height), Image.LANCZOS)
46

47
            resized_array = np.array(resized).astype(np.float32) / 255.0
48
            results.append(resized_array)
49

50
        output = torch.from_numpy(np.stack(results, axis=0))
51
        return (output,)
52

53
    def _crop_to_size(self, pil_img, target_width, target_height):
54
        """裁剪到目标尺寸"""
55
        orig_width, orig_height = pil_img.size
56
        left = (orig_width - target_width) // 2
57
        top = (orig_height - target_height) // 2
58
        right = left + target_width
59
        bottom = top + target_height
60
        return pil_img.crop((left, top, right, bottom))
61

62
    def _pad_to_size(self, pil_img, target_width, target_height):
63
        """填充到目标尺寸"""
64
        padded = Image.new("RGB", (target_width, target_height), (0, 0, 0))
65
        paste_pos = ((target_width - pil_img.width) // 2,
66
                     (target_height - pil_img.height) // 2)
67
        padded.paste(pil_img, paste_pos)
68
        return padded

1
class ImageCompositeNode:
2
    """将两张图像合成为一张（Alpha 混合）"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "base_image": ("IMAGE",),          # 基础图像
9
                "overlay_image": ("IMAGE",),       # 叠加图像
10
                "opacity": ("FLOAT", {
11
                    "default": 0.5,
12
                    "min": 0.0,
13
                    "max": 1.0,
14
                    "step": 0.01
15
                }),
16
            },
17
            "optional": {
18
                "x_offset": ("INT", {"default": 0}),
19
                "y_offset": ("INT", {"default": 0}),
20
            }
21
        }
22

23
    RETURN_TYPES = ("IMAGE",)
24
    FUNCTION = "composite"
25
    CATEGORY = "My Nodes/Image Processing"
26

27
    def composite(self, base_image, overlay_image, opacity, x_offset=0, y_offset=0):
28
        batch = max(base_image.shape[0], overlay_image.shape[0])
29
        results = []
30

31
        for i in range(batch):
32
            base = base_image[i % base_image.shape[0]].cpu().numpy()
33
            overlay = overlay_image[i % overlay_image.shape[0]].cpu().numpy()
34

35
            h, w = base.shape[:2]
36
            oh, ow = overlay.shape[:2]
37

38
            # 确保尺寸一致
39
            if h != oh or w != ow:
40
                overlay = np.array(Image.fromarray((overlay * 255).astype(np.uint8))
41
                                  .resize((w, h), Image.LANCZOS)) / 255.0
42

43
            # Alpha 混合
44
            result = base * (1 - opacity) + overlay * opacity
45
            results.append(result)
46

47
        return (torch.from_numpy(np.stack(results, axis=0)),)

1
class IfElseNode:
2
    """根据条件选择不同输入的节点"""
3

4
    RETURN_TYPES = ("*",)
5
    RETURN_NAMES = ("output",)
6
    FUNCTION = "execute"
7
    CATEGORY = "My Nodes/Control Flow"
8

9
    @classmethod
10
    def INPUT_TYPES(cls):
11
        return {
12
            "required": {
13
                "condition": ("BOOLEAN", {"default": True}),
14
                "if_true": ("*", {}),
15
                "if_false": ("*", {}),
16
            }
17
        }
18

19
    def execute(self, condition, if_true, if_false):
20
        return (if_true if condition else if_false,)

1
class CompareNode:
2
    """比较两个数值并输出结果"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "value_a": ("FLOAT", {"default": 0.0}),
9
                "value_b": ("FLOAT", {"default": 0.0}),
10
                "operation": (["equal", "greater", "less", "greater_equal", "less_equal"],
11
                             {"default": "equal"}),
12
            }
13
        }
14

15
    RETURN_TYPES = ("FLOAT", "FLOAT", "BOOLEAN", "STRING")
16
    RETURN_NAMES = ("value_a", "value_b", "result", "description")
17
    FUNCTION = "compare"
18
    CATEGORY = "My Nodes/Control Flow"
19

20
    def compare(self, value_a, value_b, operation):
21
        if operation == "equal":
22
            result = abs(value_a - value_b) < 1e-6
23
            desc = f"{value_a} == {value_b}"
24
        elif operation == "greater":
25
            result = value_a > value_b
26
            desc = f"{value_a} > {value_b}"
27
        elif operation == "less":
28
            result = value_a < value_b
29
            desc = f"{value_a} < {value_b}"
30
        elif operation == "greater_equal":
31
            result = value_a >= value_b
32
            desc = f"{value_a} >= {value_b}"
33
        else:  # less_equal
34
            result = value_a <= value_b
35
            desc = f"{value_a} <= {value_b}"
36

37
        return (float(value_a), float(value_b), result, desc)

1
class CounterNode:
2
    """带计数功能的触发节点"""
3

4
    def __init__(self):
5
        self.count = 0
6

7
    @classmethod
8
    def INPUT_TYPES(cls):
9
        return {
10
            "required": {
11
                "trigger": ("BOOLEAN", {"default": False}),
12
                "reset": ("BOOLEAN", {"default": False}),
13
                "increment": ("INT", {"default": 1, "min": 1, "max": 100}),
14
            }
15
        }
16

17
    RETURN_TYPES = ("INT", "BOOLEAN", "STRING")
18
    RETURN_NAMES = ("count", "triggered", "status")
19
    FUNCTION = "count"
20
    CATEGORY = "My Nodes/Control Flow"
21

22
    def count(self, trigger, reset, increment):
23
        if reset:
24
            self.count = 0
25

26
        if trigger:
27
            self.count += increment
28

29
        status = f"Count: {self.count}, Triggered: {trigger}, Reset: {reset}"
30
        return (self.count, trigger, status)

1
class ValidatedInputNode:
2
    """使用 VALIDATE_INPUTS 进行输入验证"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "filename": ("STRING", {"default": "image.png"}),
9
                "quality": ("INT", {"default": 90, "min": 1, "max": 100}),
10
            }
11
        }
12

13
    RETURN_TYPES = ("STRING", "INT", "BOOLEAN")
14
    FUNCTION = "process"
15
    CATEGORY = "My Nodes/Advanced"
16

17
    @classmethod
18
    def VALIDATE_INPUTS(cls, filename, quality):
19
        """验证输入参数"""
20
        # 验证文件名
21
        if not filename.endswith(('.png', '.jpg', '.jpeg', '.gif')):
22
            return f"Invalid file extension. Supported: png, jpg, jpeg, gif"
23

24
        # 验证质量参数
25
        if quality < 1 or quality > 100:
26
            return "Quality must be between 1 and 100"
27

28
        return True  # 验证通过
29

30
    def process(self, filename, quality):
31
        valid = filename.endswith(('.png', '.jpg', '.jpeg', '.gif'))
32
        return (filename, quality, valid)

1
class ExpensiveComputationNode:
2
    """只在新输入变化时才重新计算的节点"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "data": ("STRING", {"multiline": True}),
9
            }
10
        }
11

12
    RETURN_TYPES = ("STRING",)
13
    FUNCTION = "compute"
14
    CATEGORY = "My Nodes/Advanced"
15

16
    @staticmethod
17
    def IS_CHANGED(data):
18
        """返回用于判断是否需要重新计算的标识"""
19
        # 可以返回哈希值、修改时间等
20
        return hash(data)
21

22
    def compute(self, data):
23
        # 这里模拟一个耗时的计算
24
        # 只有当 IS_CHANGED 返回不同值时才会执行
25
        import time
26
        time.sleep(1)  # 模拟耗时操作
27
        processed = f"Processed: {data}"
28
        return (processed,)

1
class DebugNode:
2
    """调试输出节点，打印输入到控制台"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "any_input": ("*", {}),
9
            },
10
            "optional": {
11
                "prefix": ("STRING", {"default": "DEBUG:"}),
12
            }
13
        }
14

15
    RETURN_TYPES = ()
16
    OUTPUT_NODE = True  # 标记为输出节点
17
    FUNCTION = "debug"
18
    CATEGORY = "My Nodes/Debug"
19

20
    def debug(self, any_input, prefix="DEBUG:"):
21
        print(f"\n{prefix}")
22
        print(f"Type: {type(any_input)}")
23
        if hasattr(any_input, 'shape'):
24
            print(f"Shape: {any_input.shape}")
25
        if hasattr(any_input, 'dtype'):
26
            print(f"Dtype: {any_input.dtype}")
27
        print(f"Value: {any_input}")
28
        print("-" * 50)
29
        return ()

1
class PipeBuilderNode:
2
    """构建数据管道"""
3

4
    @classmethod
5
    def INPUT_TYPES(cls):
6
        return {
7
            "required": {
8
                "value1": ("*", {}),
9
            },
10
            "optional": {
11
                "value2": ("*", {}),
12
                "value3": ("*", {}),
13
                "value4": ("*", {}),
14
            }
15
        }
16

17
    RETURN_TYPES = ("PIPE_LINE",)
18
    FUNCTION = "build_pipe"
19
    CATEGORY = "My Nodes/Pipe"
20

21
    def build_pipe(self, value1, value2=None, value3=None, value4=None):
22
        pipe = (value1, value2, value3, value4)
23
        return (pipe,)
24

25

26
class PipeReaderNode:
27
    """从数据管道中读取数据"""
28

29
    @classmethod
30
    def INPUT_TYPES(cls):
31
        return {
32
            "required": {
33
                "pipe": ("PIPE_LINE",),
34
                "index": ("INT", {"default": 0, "min": 0, "max": 3}),
35
            }
36
        }
37

38
    RETURN_TYPES = ("*",)
39
    FUNCTION = "read_pipe"
40
    CATEGORY = "My Nodes/Pipe"
41

42
    def read_pipe(self, pipe, index):
43
        if pipe is None:
44
            return (None,)
45
        value = pipe[index] if index < len(pipe) else None
46
        return (value,)

1
app.registerExtension({
2
    name: "my_custom.settings",
3

4
    async setup() {
5
        // 注册自定义设置
6
        app.extensionManager.registerSetting({
7
            id: "myExtension.setting1",
8
            name: "My Setting",
9
            type: "text",
10
            defaultValue: "default",
11
            onChange: (newVal) => {
12
                console.log("Setting changed:", newVal);
13
            }
14
        });
15
    }
16
});

1
app.registerExtension({
2
    name: "my_custom.context_menu",
3

4
    async setup() {
5
        // 获取画布菜单
6
        app.registerExtension({
7
            name: "my_custom.canvas_menu",
8
            getCanvasMenuItems(canvas) {
9
                return [
10
                    {
11
                        content: "My Custom Action",
12
                        callback: () => {
13
                            console.log("Custom action triggered");
14
                            // 执行自定义操作
15
                        }
16
                    },
17
                    app.separator,
18
                    {
19
                        content: "Another Action",
20
                        callback: () => {
21
                            // 另一个操作
22
                        }
23
                    }
24
                ];
25
            }
26
        });
27
    }
28
});

1
custom_nodes/my_node_pack/
2
├── __init__.py                    # 模块入口，导出 NODE_CLASS_MAPPINGS
3
├── pyproject.toml                 # 项目配置
4
├── README.md                      # 说明文档
5
├── requirements.txt               # Python 依赖
6
├── LICENSE                        # 许可证
7
├── web/                           # 前端资源（可选）
8
│   ├── index.js
9
│   └── styles.css
10
├── nodes/                         # 节点模块
11
│   ├── __init__.py
12
│   ├── image_nodes.py            # 图像处理节点
13
│   ├── text_nodes.py             # 文本处理节点
14
│   └── utils.py                  # 工具函数
15
└── tests/                         # 测试文件
16
    ├── __init__.py
17
    └── test_nodes.py

1
# install.py（可选的安装脚本）
2
import subprocess
3
import sys
4

5
def install_requirements():
6
    """安装依赖"""
7
    subprocess.check_call([sys.executable, "-m", "pip", "install", "-r", "requirements.txt"])
8

9
if __name__ == "__main__":
10
    install_requirements()

1
import logging
2

3
# 设置日志
4
logging.basicConfig(level=logging.DEBUG)
5
logger = logging.getLogger(__name__)
6

7
class DebugExampleNode:
8
    def process(self, data):
9
        logger.debug(f"Input data: {data}")
10
        logger.info("Processing started")
11
        logger.warning("This is a warning")
12
        logger.error("This is an error")
13

14
        # 打印详细的张量信息
15
        if hasattr(data, 'shape'):
16
            logger.debug(f"Tensor shape: {data.shape}")
17
            logger.debug(f"Tensor dtype: {data.dtype}")
18
            logger.debug(f"Tensor min/max: {data.min()}/{data.max()}")
19

20
        return (data,)

1
ModuleNotFoundError: No module named 'xxx'

1
Output type mismatch: expected IMAGE, got <class 'numpy.ndarray'>