ComfyUI 工程化 · API · 自定义节点

一、ComfyUI 的架构:UI 是前端,Server 是核心

这意味着:只要你能 POST 工作流 JSON,ComfyUI 就能跑——不需要 UI。Python / Node.js / 任何语言都行。

二、导出 API 格式工作流

UI 里的 .json 有两种形式:

API 格式长什么样

{
  "3": {
    "inputs": {
      "seed": 42,
      "steps": 25,
      "cfg": 6.5,
      "sampler_name": "dpmpp_2m",
      "scheduler": "karras",
      "denoise": 1.0,
      "model": ["4", 0],       # 引用 node 4 的第 0 个 output
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    },
    "class_type": "KSampler"
  },
  "4": {
    "inputs": {"ckpt_name": "juggernautXL_v10.safetensors"},
    "class_type": "CheckpointLoaderSimple"
  },
  "5": {
    "inputs": {"width": 1024, "height": 1024, "batch_size": 1},
    "class_type": "EmptyLatentImage"
  },
  "6": {
    "inputs": {"text": "a cat in a spacesuit", "clip": ["4", 1]},
    "class_type": "CLIPTextEncode"
  },
  ...
}

键是节点 ID(字符串数字),class_type 对应节点类型,inputs 里既有静态参数也有对其他节点的引用(["节点ID", 输出索引])。

三、Python 客户端最小示例

import json, time, uuid, urllib.request
import websocket   # pip install websocket-client

SERVER = "127.0.0.1:8188"
CLIENT_ID = str(uuid.uuid4())

def queue_prompt(workflow):
    payload = {"prompt": workflow, "client_id": CLIENT_ID}
    req = urllib.request.Request(
        f"http://{SERVER}/prompt",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
    )
    return json.loads(urllib.request.urlopen(req).read())

def get_image(filename, subfolder, typ):
    url = f"http://{SERVER}/view?filename={filename}" \
          f"&subfolder={subfolder}&type={typ}"
    return urllib.request.urlopen(url).read()

def run_workflow(workflow):
    # 1) 订阅 WebSocket 看进度
    ws = websocket.WebSocket()
    ws.connect(f"ws://{SERVER}/ws?clientId={CLIENT_ID}")

    # 2) 提交
    prompt_id = queue_prompt(workflow)["prompt_id"]

    # 3) 等结束
    while True:
        msg = json.loads(ws.recv())
        if msg["type"] == "executing":
            data = msg["data"]
            if data["node"] is None and data["prompt_id"] == prompt_id:
                break   # 完成

    # 4) 取结果
    history = json.loads(urllib.request.urlopen(
        f"http://{SERVER}/history/{prompt_id}").read())
    images = []
    for node_output in history[prompt_id]["outputs"].values():
        if "images" in node_output:
            for img in node_output["images"]:
                images.append(get_image(img["filename"],
                                         img["subfolder"],
                                         img["type"]))
    return images

# 使用
workflow = json.load(open("workflow_api.json"))
workflow["3"]["inputs"]["seed"] = int(time.time())
workflow["6"]["inputs"]["text"] = "a cyberpunk cat, neon lights"
imgs = run_workflow(workflow)
open("out.png", "wb").write(imgs[0])

这 50 行代码就是"ComfyUI 作为后端"的完整骨架——所有 SaaS AI 绘画产品底层都长这样。

四、WebSocket 进度事件

/ws 会推送多种事件:

产品 UI 里"正在生成... 42%"就是订阅 progress 事件实时显示。

五、并发与队列

ComfyUI 自带队列——多个请求排队顺序执行。进阶部署常见模式:

取消任务

• POST /interrupt 取消正在跑的任务
• POST /queue 带 {"delete": [prompt_id]} 从队列删除未执行的任务

六、参数化工作流:模板 + 变量

产品场景:一个工作流跑多种变体,不可能每次改 JSON。常见做法:

def build_workflow(template, *, prompt, seed, ckpt, steps=25, cfg=6):
    wf = copy.deepcopy(template)
    wf["3"]["inputs"]["seed"] = seed
    wf["3"]["inputs"]["steps"] = steps
    wf["3"]["inputs"]["cfg"] = cfg
    wf["4"]["inputs"]["ckpt_name"] = ckpt
    wf["6"]["inputs"]["text"] = prompt
    return wf

# 用节点名而非节点 ID 更稳
def find_node(wf, class_type, title=None):
    for nid, node in wf.items():
        if node["class_type"] == class_type:
            if title is None or node.get("_meta", {}).get("title") == title:
                return nid, node
    raise KeyError(f"no {class_type} node")

七、子工作流:Group Node & Subgraph

工作流超过 30 节点就开始眼花。ComfyUI 提供两种打包方式:

八、自定义节点开发

要集成业务逻辑(比如调公司内部 API、做特殊图像处理),就得写自定义节点。

最小节点骨架

# custom_nodes/my_pack/__init__.py
class MyPromptEnhancer:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "prompt": ("STRING", {"multiline": True}),
                "style": (["cinematic", "anime", "photorealistic"],),
                "strength": ("FLOAT", {"default": 1.0, "min": 0, "max": 2}),
            },
        }

    RETURN_TYPES = ("STRING",)
    RETURN_NAMES = ("enhanced_prompt",)
    FUNCTION = "enhance"
    CATEGORY = "my_pack/prompt"

    def enhance(self, prompt, style, strength):
        suffix = {
            "cinematic": ", cinematic lighting, 35mm film, dramatic composition",
            "anime": ", anime style, vibrant colors, detailed line art",
            "photorealistic": ", photorealistic, 8k, sharp focus",
        }[style]
        result = f"{prompt}{suffix} (weight:{strength})"
        return (result,)

NODE_CLASS_MAPPINGS = {"MyPromptEnhancer": MyPromptEnhancer}
NODE_DISPLAY_NAME_MAPPINGS = {"MyPromptEnhancer": "🎨 Prompt Enhancer"}

放到 ComfyUI/custom_nodes/my_pack/,重启 ComfyUI 就能在节点菜单 my_pack → prompt 下找到。

输入类型

• "STRING" / "INT" / "FLOAT" / "BOOLEAN"——基础类型
• ("STRING", {"multiline": True})——多行文本
• (["option1", "option2"],)——枚举下拉
• "IMAGE" / "LATENT" / "MODEL" / "CLIP" / "VAE" / "CONDITIONING"——ComfyUI 内置类型
• 自定义类型("MY_TYPE")可以,但前后节点必须都认

UI 增强(JS 端)

如果节点需要自定义 UI(比如动态刷新下拉选项),在目录下建 web/ 放 JS 文件:

// custom_nodes/my_pack/web/my_widget.js
import { app } from "../../scripts/app.js";

app.registerExtension({
  name: "MyPack.PromptEnhancer",
  async nodeCreated(node) {
    if (node.comfyClass === "MyPromptEnhancer") {
      // 加自定义按钮 / 颜色 / 回调
    }
  },
});

九、常用工程化 Custom Node 包

十、从 ComfyUI 到产品后端:架构模板

关键设计点:

• ComfyUI 不暴露给公网——只在内网,业务后端是唯一入口
• 任务状态机:pending → running → succeeded/failed,状态变更推 WebSocket 给用户
• 失败重试:ComfyUI 偶发 OOM,需要业务层重试,最大 2-3 次
• 图片 hash 去重:相同 prompt+seed+参数 的结果做缓存,降低成本

十一、调试技巧

1. ComfyUI 启动加 --verbose:看详细日志
2. 节点不熟时截图发 UI:右键 Copy (Clipspace) 或 Save (API Format) 比手写 JSON 稳
3. 画图输出 debug 文本:用 Show Text(rgthree)节点把中间 prompt / 条件 / 变量直接显示在画布上
4. 用 Preview Image 节点监控中间 latent:Hi-Res Fix 两阶段之间插一个 Preview 查第一阶段出图
5. /object_info:GET /object_info 列出所有可用节点 + 输入类型,写自动化脚本时查手册

十二、反模式

1. 用 workflow.json(UI 版)POST 到 /prompt:必然报错,要用 workflow_api.json。
2. 节点 ID 硬编码:换一台机器或加新节点就全崩,应按 class_type 查找。
3. ComfyUI 直接暴露到公网:UI 能上传任意文件 + 执行任意工作流,被入侵就是 RCE。
4. 没做队列超时:大图 OOM 导致任务挂死,后续全堵。业务层设超时 + /interrupt。
5. 自定义节点不写 requirements.txt:依赖装不全,别人部署直接报错。
6. 多 GPU 实例共享 models 目录:读 OK,但写(下载模型)会冲突。只读挂载或做锁。
7. 不用 Crystools 看 VRAM:OOM 经常突然发生,有监控才知道临界点在哪。
8. 自定义节点副作用:节点里开网络请求 / 写文件不加超时,整个图阻塞或卡死。

十三、本章小结