ComfyUI API 批量出图自动化：脚本接入工作流实战

你有一套稳定的 ComfyUI 工作流，想给 200 个商品各出 4 张主图，不想守着界面一张张改 prompt 和 seed。第一次用脚本 POST 到 /prompt，返回的却是 node_errors，才发现导出的不是 API format。任务跑完了，/history 里只看到 filename 和 subfolder，不知道还要用 /view 才能下载图片。

ComfyUI 从 GUI 到脚本量产，要过三道坎：导出正确的 API format workflow、用 WebSocket 等待任务结束而不是盲目轮询、控制并发和显存避免 OOM。后面会给出最小脚本骨架、参数化批量示例、队列策略和后端工程化清单，帮你避开格式错误、等待混乱、显存不足、结果丢失这些坑点。

本地 ComfyUI Server API 入口

CLI 启动与默认端口

ComfyUI 本地服务默认监听 127.0.0.1:8188。只在本机调用时直接启动即可；确实需要局域网设备访问时，再显式指定监听地址：

# 仅本机访问
python main.py --port 8188

# 允许局域网访问；同时配置防火墙、鉴权或反向代理
python main.py --listen 0.0.0.0 --port 8188

终端输出 Starting server 后，用浏览器访问 http://127.0.0.1:8188 能看到 ComfyUI 前端界面，说明服务已就绪。

显存不足时可按当前版本支持情况使用 --lowvram、--novram 或最后再考虑很慢的 --cpu；显存余量充足时也可评估 --highvram。不要按固定显存容量机械选择参数，因为模型、精度、VAE 和工作流结构都会改变峰值。启动参数会更新，使用前以 python main.py --help 和当前官方排障文档为准。

核心 API Routes 速查表

本地 ComfyUI Server 的主要路由都在 server.py 中定义，以下是脚本调用会用到的端点：

路由	用途	参数/返回
/prompt	验证 workflow 并加入队列	POST {"prompt": workflow_dict, "client_id": "..."}；成功返回 prompt_id，失败返回 node_errors
/history/{prompt_id}	获取任务执行历史与输出 metadata	GET；返回 outputs 数组，包含 filename、subfolder、type
/view?filename=...&subfolder=...&type=...	下载输出文件	GET；返回二进制文件
/ws	WebSocket 连接，接收执行状态	ws://127.0.0.1:8188/ws?clientId=...
/queue	查看当前队列	GET；返回队列中任务列表
/interrupt	中断当前执行	POST；用于超时取消
/upload/image	上传输入图片（图生图）	POST multipart/form-data
/object_info	查询所有可用节点类型与参数	GET；调试用，可检查节点是否存在

路由列表可能随 ComfyUI 版本增加或修改，建议用当前官方文档或直接读 server.py 复核。

WebSocket 消息类型

通过 WebSocket 等待任务结束时，需要监听以下消息类型：

• status：队列状态，包含 queue_remaining 数量
• execution_start：执行开始，携带 prompt_id
• execution_cached：缓存节点复用，哪些节点被跳过
• executing：当前执行节点；当 node is None 且 prompt_id 匹配时，表示任务结束
• progress：进度条信息，包含当前步数和总步数
• executed：节点执行完成，携带输出 metadata

脚本判断任务结束的核心逻辑：收到 type == "executing" 消息，检查 data.node 是否为 None，同时确认 data.prompt_id 是否和提交任务时返回的 prompt_id 一致。

导出 API Format Workflow

Export Workflow (API) 步骤

ComfyUI 前端保存的 workflow JSON 和 API 调用所需的格式不同。直接 POST 普通 workflow 会返回 node_errors 验证失败，必须导出 API format。

导出流程：

1. 在 ComfyUI 前端加载已验证能正常生成的工作流
2. 点击菜单 File -> Export Workflow (API)（部分版本显示为 Save (API Format)，以当前 UI 为准）
3. 保存为 .json 文件（如 workflow_api.json）
4. 打开 JSON 文件，确认节点键为数字 ID（如 "3"、"6"），每个节点包含 class_type 和 inputs 字段

前端菜单命名可能随版本更新变化，建议以当前 ComfyUI 界面为准。

API format vs Save format 差异对照

普通保存的 workflow JSON（Save format）包含大量 UI 布局信息，API format 则去掉这些 metadata，只保留执行所需的核心数据：

格式	包含内容	用途
Save format	节点位置坐标、颜色、分组、尺寸、链接可视化信息	前端导入复用工作流布局
API format	数字 node ID、class_type、inputs、可选 _meta	脚本/API 提交执行

直接用 Save format JSON POST 到 /prompt，验证会失败，返回 node_errors 或 error。正确做法是在前端加载 Save format workflow，再导出为 API format。

节点 ID 管理建议

脚本参数化时需要定位节点：prompt 节点、seed 节点、尺寸节点、输出节点。导出 API format 后，打开 JSON 文件，找到以下节点 ID：

• Prompt node：CLIPTextEncode，常见 ID 是 "6"
• Seed node：KSampler，常见 ID 是 "3"
• Width/Height node：在 EmptyLatentImage 或 KSampler 的 inputs 中
• Output node：SaveImage 或 SaveImageWebsocket

建议在前端用 Note 节点或 Group 标注节点用途，方便后续在 JSON 中查找。重新导出 workflow 后，节点 ID 可能变化，需要重新记录。节点 ID 是数字，不是节点名称，所以必须在 JSON 中确认。

最小脚本骨架：提交、等待、下载

ComfyUI 调用流程可以拆成三步：提交 workflow 到 /prompt、等待任务结束、通过 /history 和 /view 下载图片。

HTTP 提交流程（submit-and-forget）

最简单的提交方式：POST workflow 到 /prompt，不等待结果。适合批量提交后轮询检查的场景。

import json
import requests

# 加载 API format workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# 构造请求数据
payload = {
    "prompt": workflow,
    "client_id": "my-script-client"  # 可选，用于 WebSocket 关联
}

# 提交到队列
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)

if response.status_code == 200:
    result = response.json()
    prompt_id = result["prompt_id"]
    print(f"任务已提交，prompt_id: {prompt_id}")
else:
    error = response.json()
    print(f"提交失败: {error}")
    # node_errors 包含具体节点验证错误

成功返回 {"prompt_id": "...", "number": ...}，失败返回 {"error": {...}, "node_errors": {...}}。任务提交后进入队列。脚本不会自动等待完成。

WebSocket 等待完成（推荐）

用 WebSocket 监听执行状态，比轮询高效。连接 WebSocket → 提交任务 → 监听 executing 消息判断结束。

import json
import uuid
import requests
import websocket

# 加载 workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# 生成 client_id
client_id = str(uuid.uuid4())

# 连接 WebSocket
ws = websocket.create_connection(f"ws://127.0.0.1:8188/ws?clientId={client_id}")

# 提交任务
payload = {"prompt": workflow, "client_id": client_id}
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
prompt_id = response.json()["prompt_id"]

# 等待执行完成
while True:
    message = ws.recv()
    data = json.loads(message)

    if data["type"] == "executing":
        # node is None 且 prompt_id 匹配表示任务结束
        if data["data"]["node"] is None and data["data"]["prompt_id"] == prompt_id:
            print("任务完成")
            break

ws.close()

建议加超时保护：设置最长等待时间（如 300 秒），超时后调用 /interrupt 中断任务。

History/View 下载图片

任务结束后，用 /history/{prompt_id} 获取输出 metadata，再通过 /view 下载实际图片文件。

import requests

# 获取任务历史
history_url = f"http://127.0.0.1:8188/history/{prompt_id}"
history = requests.get(history_url).json()

# 遍历输出节点
outputs = history[prompt_id]["outputs"]
for node_id, node_output in outputs.items():
    if "images" in node_output:
        for image in node_output["images"]:
            filename = image["filename"]
            subfolder = image.get("subfolder", "")
            type = image.get("type", "output")

            # 构造下载 URL
            view_url = f"http://127.0.0.1:8188/view?filename={filename}&subfolder={subfolder}&type={type}"

            # 下载图片
            img_data = requests.get(view_url).content
            with open(f"output_{filename}", "wb") as f:
                f.write(img_data)
            print(f"已保存: output_{filename}")

/history 返回的 outputs 结构：{node_id: {"images": [{"filename": "...", "subfolder": "...", "type": "..."}]}}。图片通过 /view 下载为二进制数据。

批量参数化脚本

脚本参数化是批量出图的核心：循环提交任务、每次修改 prompt、seed、尺寸等参数，避免手动改 GUI。

Prompt/Seed/尺寸参数化

修改 workflow JSON 中的节点 inputs，可以实现参数化。以下示例修改 prompt、seed、width/height：

import json
import random

# 加载 workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# 修改 prompt（节点 ID 需根据实际工作流调整）
workflow["6"]["inputs"]["text"] = "a beautiful landscape, sunset, mountains"

# 修改 seed（随机生成）
workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

# 修改尺寸
workflow["5"]["inputs"]["width"] = 1024
workflow["5"]["inputs"]["height"] = 768

# 提交任务...

节点 ID（"6"、"3"、"5"）需要根据实际导出的 API format workflow 确认。批量生成时，用 for 循环遍历 prompt 列表，每次修改 seed 和参数后提交：

prompts = [
    "product photo, white background",
    "product photo, outdoor scene",
    "product photo, studio lighting"
]

for i, prompt_text in enumerate(prompts):
    workflow["6"]["inputs"]["text"] = prompt_text
    workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

    # 提交任务
    payload = {"prompt": workflow, "client_id": client_id}
    response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
    prompt_id = response.json()["prompt_id"]

    # 等待完成（WebSocket）...
    # 下载图片...

官方示例中也会修改 KSampler.seed 和 CLIPTextEncode.text，可作为参数化入口参考。

批量队列策略

批量生成 200 张图，有两个策略选择：调整 batch size 还是循环提交 prompt。显存、模型大小和工作流复杂度会影响选择。

策略	特点	适用场景
单张单次提交	每次提交一个 prompt，显存占用较易控制	显存余量有限、模型较大、工作流复杂
Batch size	单个 prompt 内批量生成多张，显存峰值通常更高	显存余量充足、模型较小、简单工作流
并发请求	多个 prompt 同时提交，需控制并发数	需要快速吞吐，显存充足

并发请求的风险：同时提交 20 个 prompt，队列堆积，显存峰值可能超过 GPU 容量，触发 OOM 或服务崩溃。

推荐策略：顺序提交 + WebSocket 等待完成 + 显存监控。每次提交一个任务，等待结束后再提交下一个，避免队列堆积。显存不足时调整启动参数（--lowvram）或使用 Tiled VAE，具体细节参考《ComfyUI 低显存与提速实战》。

并发控制代码示例

如果确实需要并发提交（例如 8 卡服务器或云端多实例），可以用信号量控制并发数：

import copy
import threading

# 并发控制：最多同时运行 2 个任务
semaphore = threading.Semaphore(2)

def submit_and_wait(prompt_text, seed):
    with semaphore:  # 获取许可
        # 每个任务使用独立副本，避免线程同时修改同一个 workflow
        job_workflow = copy.deepcopy(workflow)
        job_workflow["6"]["inputs"]["text"] = prompt_text
        job_workflow["3"]["inputs"]["seed"] = seed

        # 提交并等待完成...
        # WebSocket 循环...

    # 许可归还，下一个任务可以开始

# 批量提交
threads = []
for i in range(50):
    t = threading.Thread(target=submit_and_wait, args=(prompts[i], seeds[i]))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

并发数从 1 开始，用真实工作流实测峰值、延迟和失败率后再逐步增加。即使显存容量相同，不同模型、精度、VAE、后处理节点和 GPU worker 数量也会给出不同上限。

显存实时监控

批量任务运行期间，定时检查显存使用率，超过阈值就暂停提交新任务：

import time

def check_vram(threshold=0.8):
    stats = requests.get("http://127.0.0.1:8188/system_stats").json()
    vram_used = stats["system_stats"]["devices"][0]["vram_used"]
    vram_total = stats["system_stats"]["devices"][0]["vram_total"]
    return (vram_used / vram_total) > threshold

# 批量提交时检查显存
for prompt_text in prompts:
    while check_vram(0.85):  # 显存超过 85%，等待释放
        print("显存不足，等待 30 秒...")
        time.sleep(30)

    # 提交下一个任务...
    # WebSocket 等待完成...

显存峰值通常出现在 KSampler 执行阶段，任务结束后显存占用会降低。监控周期建议 10-30 秒，避免频繁请求。

批量输出归档

批量生成后需要有序管理输出文件，避免混在一起找不到。

文件命名建议：{prompt_id}_{seed}_{timestamp}.png，包含任务 ID、种子和时间戳。

元数据记录建议：

• prompt_id：ComfyUI 任务 ID
• seed：随机种子
• prompt_text：prompt 内容
• width/height：尺寸
• timestamp：生成时间
• business_id：业务 ID（如商品 SKU、订单号）

目录结构建议：按日期或任务批次归档，如 outputs/20260624/batch_001/。可选：数据库记录（如 SQLite 或 PostgreSQL）存储元数据与图片路径的关联。

后端工程化清单

把 ComfyUI 接入后端服务时，需要补充请求 ID、超时、队列限流、显存监控等工程化设计，避免生产环境出问题。

请求 ID 与幂等性

每个业务请求生成唯一 request_id（如 UUID 或业务订单号），记录 request_id 与 ComfyUI prompt_id 的映射。重复请求检查：已完成的 request_id 直接返回结果，不重新生成。

import uuid

# 业务请求 ID
request_id = str(uuid.uuid4())

# 记录映射（数据库或缓存）
request_prompt_map[request_id] = prompt_id

# 重复请求检查
if request_id in completed_requests:
    return get_cached_result(request_id)

注意：ComfyUI 的 prompt_id 由服务端生成，不等同于业务 request_id。需要自己维护映射关系。

超时与取消

设置单任务最长执行时间（如 300 秒），超时后调用 /interrupt 中断当前任务，让队列继续执行下一个任务。

import time

timeout = 300  # 5 分钟
start_time = time.time()

# 等待 WebSocket 消息...
while True:
    elapsed = time.time() - start_time
    if elapsed > timeout:
        # 超时中断
        requests.post("http://127.0.0.1:8188/interrupt")
        print("任务超时，已中断")
        break

    # 正常消息处理...

WebSocket 超时处理：连接断开或无响应超过阈值（如 30 秒），触发重连或取消。中断后需检查 /queue，清理队列中剩余任务。

队列限流与显存监控

队列长度限制：如最多 5 个任务排队，超出拒绝新请求。并发请求限制：如同时最多 2 个 prompt 执行，避免显存峰值。

显存监控：定时查询 /system_stats，检查 VRAM 使用率。

stats = requests.get("http://127.0.0.1:8188/system_stats").json()
vram_used = stats["system_stats"]["devices"][0]["vram_used"]
vram_total = stats["system_stats"]["devices"][0]["vram_total"]
vram_percent = vram_used / vram_total

if vram_percent > 0.8:
    print("显存不足，拒绝新请求")

显存不足时拒绝新请求或等待队列清空。低显存启动参数、Tiled VAE、智能内存管理的细节参考《ComfyUI 低显存与提速实战》。

错误分类与重试

不同错误类型有不同的处理策略，盲目重试可能让问题更严重：

错误类型	原因	处理方式
node_errors	缺模型、缺节点、输入参数错误	检查 workflow 和环境；不重试
OOM	显存不足	降低 batch size、调整启动参数（--lowvram）；不重试
WebSocket 断开	网络问题	重连并查询 /history/{prompt_id}
任务超时	模型加载慢或复杂工作流	增加超时时间或简化工作流；可重试一次
服务崩溃	内存耗尽、GPU 故障	检查日志、重启服务；重试需谨慎

node_errors 包含具体节点验证错误，如缺失节点类型、参数类型不匹配。需要检查 workflow JSON 和 ComfyUI 环境中的 custom nodes、模型文件。OOM 错误需要调整显存策略，不能简单重试。

Cloud API 与本地 API 对比

Comfy Cloud API 提供云端托管 ComfyUI 的程序化访问，但鉴权、任务状态、WebSocket 地址和并发限制与本地 Server API 不同。可以复用 API format workflow 的思路，迁移代码前仍要按当前 Cloud API Reference 调整端点。

路由前缀差异

本地 Server API 与 Cloud API 的主要路由对照：

功能	本地 API	Cloud API
提交任务	/prompt	/api/prompt
查询状态/结果	/history/{prompt_id}	/api/job/{prompt_id}/status 与 /api/jobs/{job_id}
下载输出	/view	/api/view
WebSocket	/ws?clientId=...	/ws?clientId=...&token=...

Cloud API 需要 X-API-Key header 和 active subscription；本地 API 默认只监听 127.0.0.1，但一旦通过 --listen、反向代理或端口映射开放网络，就必须自行补鉴权与访问控制。Cloud API 文档仍标注 experimental；旧的 /api/history_v2/{prompt_id} 已被标为 deprecated，当前建议使用 /api/jobs/{job_id}，迁移前应重新核对官方文档。

并发与订阅限制

Cloud API 并发受订阅层级限制，超出并发限制的 jobs 进入队列等待。输出保存在 cloud storage，通过 /api/view 返回临时签名 URL，有效期有限。

风险提醒：订阅层级、并发数、runtime cap、价格易变，本文不写具体数字，建议以 Comfy Cloud 官方 pricing 页面为准。云端托管的优势是无需本地 GPU，但成本、并发和 runtime cap 需要根据业务需求评估。本地 Server API 则没有并发层级限制，但显存、队列和稳定性需要自己维护。

下一步与延伸阅读

系列内链

本文是 ComfyUI 系列的工程化收口页，帮助从”会搭工作流”推进到”可编程调用/批量生产/后端接入”。前置知识和排障可以回系列其他文章：

• ComfyUI 工作流复用完整指南：缺节点、模型路径、workflow 导入与 missing nodes 排障
• ComfyUI 低显存与提速实战：显存优化、启动参数、OOM 排障、Tiled VAE
• ComfyUI 视频生成实战：视频工作流边界提醒
• ComfyUI 排障与维护：缺节点、启动卡死、版本冲突

跨主题内链

把 ComfyUI 接入后端服务或自动化流程，可以参考其他 AI 工具的自动化实践：

• n8n AI 工作流自动化：把 ComfyUI 接入 n8n 流程，实现多工具协同
• Ollama API 实践：LLM 服务的程序化调用，队列与结构化输出
• LLM 结构化输出：Prompt 工程与 API 集成，从 LLM 提取可靠数据

结论

ComfyUI 从 GUI 到脚本量产的核心流程：导出 API format workflow → 用 WebSocket 等待任务结束 → 通过 /history 和 /view 下载图片 → 参数化批量生成 → 补充工程化清单（请求 ID、超时、队列限流、显存监控）。

开始动手：导出你的第一个 API format workflow，用最小脚本骨架提交任务并等待完成。跑通单张后，尝试参数化批量生成 10 张图，观察显存占用和队列行为。批量稳定后，补充请求 ID、超时取消、显存监控等工程化设计，避免生产环境出问题。