关键词: ComfyUI API, Stable Diffusion, 图像生成, WebSocket, Python自动化

目录

1、执行效果

2、环境准备

2.1 安装依赖库

2.2 启动ComfyUI服务器

3、请求代码编写

3.1 抓取请求信息

3.2 官方示例脚本

3.3 编写请求代码

4、执行流程解析

4.1 建立WebSocket连接

4.2 提交任务并获取结果

4.3 显示生成图像

1、Python执行效果

2、环境准备

2.1 安装依赖库

确保已安装以下Python库：

pip install websocket-client uuid Pillow
  • ​websocket-client: 用于与ComfyUI服务器建立WebSocket长连接
  • ​uuid: 生成唯一客户端标识符
  • ​Pillow: 图像处理库（用于显示生成的图片）

2.2 启动ComfyUI服务器

启动ComfyUI

在设置的服务器配置里查看网址和监听的端口。

浏览器访问服务地址。

3、请求代码编写

3.1 抓取请求信息

在网页端中，选择要运行的工作流，按下F12，运行一次，抓取请求信息。

我们只需要请求参数中的prompt字段。这里相当于每次请求使用的工作流。

3.2 官方示例脚本

参考官方代码仓库的请求脚本：ComfyUI/script_examples/websockets_api_example.py at master · comfyanonymous/ComfyUI · GitHub

3.3 编写请求代码

# 导入所需库
import websocket  # 用于WebSocket通信
import uuid       # 生成唯一客户端ID
import json       # 处理JSON数据
import urllib.request  # 发送HTTP请求
import urllib.parse    # 构造URL参数

# 定义服务器地址和生成客户端唯一ID
server_address = "127.0.0.1:8000"
client_id = str(uuid.uuid4())

def queue_prompt(prompt):
    """提交提示到服务器并返回执行ID"""
    # 构造包含提示和客户端ID的请求体
    p = {"prompt": prompt, "client_id": client_id}
    # 发送POST请求到/prompt端点
    data = json.dumps(p).encode('utf-8')
    req = urllib.request.Request(
        f"http://{server_address}/prompt", 
        data=data
    )
    return json.loads(urllib.request.urlopen(req).read())

def get_image(filename, subfolder, folder_type):
    """根据参数获取生成的图像二进制数据"""
    # 构造URL查询参数
    data = {"filename": filename, "subfolder": subfolder, "type": folder_type}
    url_values = urllib.parse.urlencode(data)
    # 发送GET请求获取图像
    with urllib.request.urlopen(f"http://{server_address}/view?{url_values}") as response:
        return response.read()

def get_history(prompt_id):
    """获取指定提示ID的执行历史记录"""
    with urllib.request.urlopen(f"http://{server_address}/history/{prompt_id}") as response:
        return json.loads(response.read())

def get_images(ws, prompt):
    """通过WebSocket监听执行过程并获取最终生成的图像"""
    # 提交提示并获取执行ID
    prompt_id = queue_prompt(prompt)['prompt_id']
    output_images = {}
    
    # 持续监听WebSocket消息
    while True:
        out = ws.recv()
        if isinstance(out, str):
            message = json.loads(out)
            # 检测执行完成信号
            if message['type'] == 'executing':
                data = message['data']
                if data['node'] is None and data['prompt_id'] == prompt_id:
                    break  # 执行完成时退出循环
        else:
            # 二进制数据流（预览图）处理逻辑（已跳过）
            continue
    
    # 获取执行历史并提取生成的图像
    history = get_history(prompt_id)[prompt_id]
    for node_id in history['outputs']:
        node_output = history['outputs'][node_id]
        images_output = []
        if 'images' in node_output:
            for image in node_output['images']:
                # 获取每个图像的二进制数据
                image_data = get_image(image['filename'], image['subfolder'], image['type'])
                images_output.append(image_data)
        output_images[node_id] = images_output
    
    return output_images

# 定义工作流提示模板
prompt = {
    "6": {
        "inputs": {
            "text": "cute anime girl with massive fluffy fennec ears and a big fluffy tail blonde messy long hair blue eyes wearing a maid outfit with a long black gold leaf pattern dress and a white apron mouth open placing a fancy black forest cake with candles on top of a dinner table of an old dark Victorian mansion lit by candlelight with a bright window to the foggy forest and very expensive stuff everywhere there are paintings on the walls",
            "clip": ["30", 1]
        },
        "class_type": "CLIPTextEncode",
        "_meta": {"title": "CLIP文本编码器"}
    },
    "8": {
        "inputs": {"samples": ["31", 0], "vae": ["30", 2]},
        "class_type": "VAEDecode",
        "_meta": {"title": "VAE解码"}
    },
    "9": {
        "inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]},
        "class_type": "SaveImage",
        "_meta": {"title": "保存图像"}
    },
    "27": {
        "inputs": {"width": 1024, "height": 1024, "batch_size": 1},
        "class_type": "EmptySD3LatentImage",
        "_meta": {"title": "空Latent_SD3"}
    },
    "30": {
        "inputs": {"ckpt_name": "flux1-dev-fp8.safetensors"},
        "class_type": "CheckpointLoaderSimple",
        "_meta": {"title": "Checkpoint加载器(简易)"}
    },
    "31": {
        "inputs": {
            "seed": 972054013131369,
            "steps": 20,
            "cfg": 1,
            "sampler_name": "euler",
            "scheduler": "simple",
            "denoise": 1,
            "model": ["30", 0],
            "positive": ["35", 0],
            "negative": ["33", 0],
            "latent_image": ["27", 0]
        },
        "class_type": "KSampler",
        "_meta": {"title": "K采样器"}
    },
    "33": {
        "inputs": {"text": "", "clip": ["30", 1]},
        "class_type": "CLIPTextEncode",
        "_meta": {"title": "CLIP文本编码器"}
    },
    "35": {
        "inputs": {"guidance": 3.5, "conditioning": ["6", 0]},
        "class_type": "FluxGuidance",
        "_meta": {"title": "Flux引导"}
    }
}

# 建立WebSocket连接并获取图像
ws = websocket.WebSocket()
ws.connect(f"ws://{server_address}/ws?clientId={client_id}")
images = get_images(ws, prompt)
ws.close()  # 显式关闭连接

# 以下代码可用于显示生成的图像（需取消注释）
for node_id in images:
    for image_data in images[node_id]:
        from PIL import Image
        import io
        image = Image.open(io.BytesIO(image_data))
        image.save("output.jpg")  # 保存到文件
        image.show()

注意，代码中的请求地址和端口要和你ComfyUI的地址要保持一致。通过Python调用api接口的方式，可以实现批量，自动化等操作，使用起来更灵活。

3.4 代码详细解析

1. 导入库及作用
  • ​websocket​：用于与服务器建立WebSocket通信，实时接收生成状态。
  • ​uuid​：生成唯一客户端ID，标识当前会话。
  • ​json​：处理请求和响应中的JSON数据。
  • ​urllib.request​：发送HTTP请求（GET/POST）。
  • ​urllib.parse​：构造URL查询参数。
2. 全局变量定义
  • ​**server_address = "127.0.0.1:8000"**​：本地服务器地址，指向ComfyUI或其他AI服务。
  • ​**client_id = str(uuid.uuid4())**​：生成唯一客户端ID，确保会话隔离。
3. 核心函数解析
(1) queue_prompt(prompt)
  • ​功能​：提交生成任务到服务器，返回任务ID。
  • ​步骤​：
    1. 构造包含提示和客户端ID的请求体。
    2. 发送POST请求到/prompt端点。
    3. 解析响应中的prompt_id，用于后续查询。
(2) get_image(filename, subfolder, folder_type)
  • ​功能​：通过文件名、子文件夹和类型获取图像二进制数据。
  • ​步骤​：
    1. 构造URL查询参数（如filename）。
    2. 发送GET请求到/view端点。
    3. 返回图像二进制数据。
(3) get_history(prompt_id)
  • ​功能​：获取指定任务ID的历史执行记录。
  • ​步骤​：
    1. 请求/history/{prompt_id}端点。
    2. 返回包含输出节点数据的JSON。
(4) get_images(ws, prompt)
  • ​功能​：主流程函数，监听任务完成并提取图像。
  • ​步骤​：
    1. 提交提示，获取prompt_id。
    2. 通过WebSocket循环监听消息，直到任务完成。
    3. 查询历史记录，遍历所有输出节点。
    4. 调用get_image下载图像数据，按节点ID分类存储。
4. 工作流提示（prompt变量）
  • ​结构​：字典定义多个节点，每个节点包含输入、类型和元数据。
  • ​关键节点​：
    • ​节点30​：加载模型检查点（CheckpointLoaderSimple）。
    • ​节点6/33​：CLIP文本编码器，处理正/负面提示。
    • ​节点27​：生成空白潜在图像（分辨率1024x1024）。
    • ​节点31​：K采样器（KSampler），配置生成参数（如步数、采样器）。
    • ​节点35​：Flux引导，调整生成条件。
    • ​节点8​：VAE解码，将潜在数据转为图像。
    • ​节点9​：保存图像到文件。
5. WebSocket通信流程
  • ​连接建立​：ws.connect()使用客户端ID连接WebSocket端点。
  • ​消息监听​：循环接收消息，检测executing类型消息判断任务完成。
  • ​资源释放​：显式调用ws.close()关闭连接。
6. 图像保存与展示（注释部分）
  • ​步骤​：
    1. 使用PIL.Image读取二进制数据。
    2. 保存为output.jpg并显示图像（需取消注释）。
7. 整体执行流程
  1. 提交提示，启动生成任务。
  2. 通过WebSocket监听任务状态。
  3. 任务完成后，从历史记录提取图像信息。
  4. 下载图像数据并保存到本地文件。

代码总结

该脚本实现了一个完整的图像生成流程：

  1. ​配置工作流​：通过节点定义生成步骤（模型加载、文本编码、采样、解码等）。
  2. ​任务提交与监听​：使用HTTP和WebSocket与服务器交互。
  3. ​结果获取​：按节点提取生成的图像数据，最终保存到本地。

4、执行流程解析

4.1 建立WebSocket连接

ws = websocket.WebSocket()
ws.connect(f"ws://{server_address}/ws?clientId={client_id}")

4.2 提交任务并获取结果

images = get_images(ws, prompt)

4.3 显示生成图像

for node_id in images:
    for image_data in images[node_id]:
        from PIL import Image
        import io
        image = Image.open(io.BytesIO(image_data))
        image.save("output.jpg")  # 保存到文件
        image.show()             # 弹出预览窗口

# python # AIGC # ComfyUI
Logo
天启AI社区

GitCode 天启AI是一款由 GitCode 团队打造的智能助手，基于先进的LLM（大语言模型）与多智能体 Agent 技术构建，致力于为用户提供高效、智能、多模态的创作与开发支持。它不仅支持自然语言对话，还具备处理文件、生成 PPT、撰写分析报告、开发 Web 应用等多项能力，真正做到“一句话，让 Al帮你完成复杂任务”。

更多推荐

cover

【感知机】感知机(perceptron)学习算法例题及详解

avatar 天启AI社区
cover

AI云原生：数智化时代AI Agent加速落地的技术基石与实践指南（THS）

avatar 天启AI社区
cover

【AI论文】设计实验室：通过迭代检测与修正来设计幻灯片

avatar 天启AI社区