通过 API 使用托管 ComfyUI 工作流

ComfyUI 是一套生图领域常用的工作流（workflow）编排工具，可以方便灵活的设计图片处理工作流。无问芯穹 GenStudio 提供了 ComfyUI 工作流托管服务，用户托管工作流后，可以在业务中通过 API 调用该工作流。平台负责维护工作流的运行环境，优化算力资源、推理效率。

NOTE

关于 API 端点的路径、参数等细节，详见 ComfyUI HTTP API 参考文档。
为了展示如何通过 API 使用传入工作流 ID 和节点参数，我们提供了一个文生图 API 请求体示例。

简介

本文介绍了与 GenStuio ComfyUI 服务通信的 API 端点结构，并结合 Python 代码展示了托管、数据预处理和 API 对接的详细流程。

您将了解：

如何在 GenStudio 中创建托管的 ComfyUI 工作流。
如何搭建一个文生图的 API 流程。
如何搭建一个图生图的 API 流程。
如何在获取托管工作流 API 的输入参数类型和范围。
如何在业务流程中修改工作流参数。

托管 ComfyUI 工作流

托管工作流是指将本地以开发好的 ComfyUI 工作流上传至 GenStudio 平台。

导出 API Format 工作流

请在 ComfyUI 中使用 "Save（API Format）" 保存工作流。若无此选项，需在设置中启用 "Enable Dev Mode options"。

alt text

上传 API Format 工作流

前往 GenStudio 托管工作流的管理界面，上传从 ComfyUI 导出的 API JSON 格式文件以创建工作流。

ComfyUI Workflow 管理

工作流管理界面

平台会校验您的工作流中的节点参数，通过校验表示托管成功，平台会生成工作流 ID。

NOTE

如果校验不通过，请确保是通过 ComfyUI 的 Save（API Format）导出的工作流文件。

查看入参范围

在通过 API 调用工作流时，需要获取 API 各个输入参数的类型、范围等。您可以在 GenStudio 托管工作流的管理界面，点击工作流右侧的查看参数。

alt text

平台会在每个可传入的参数的 spec 键下返回可接受的入参范围和默认值。

    "spec": {
        "default": 20,
        "min": 1,
        "max": 10000
    }

通过上述步骤，我们已完成在 ComfyUI 中的处理。后续在搭建 API 流程时，我们将使用 GenStudio 平台，获取工作流 ID 和工作流参数。

API 鉴权设置

使用 API 服务，首先需要完成身份验证。请按照以下步骤获取您的 API 密钥：

API 密钥管理

复制已有的 API 密钥，或自助创建 API Key。
点击复制按钮获取密钥。您可能需要完成二次验证。

API 密钥管理界面

Jupyter Notebook 设置

在 Jupyter Notebook 中，首先设置全局变量和导入必要库。

python

import os
import requests
import time
import json
import random
from PIL import Image
import io
from urllib.parse import urlparse

# 设置全局变量
# 设置 API 密钥和服务器地址
API_KEY = os.getenv('COMFYUI_API_KEY')
SERVER_ADDRESS = os.getenv('COMFYUI_SERVER_ADDRESS')
# 设置默认工作流 ID
DEFAULT_WORKFLOW_ID = os.getenv('DEFAULT_WORKFLOW_ID')
# 设置默认输出路径
DEFAULT_OUTPUT_PATH = os.getenv('DEFAULT_OUTPUT_PATH')
# 设置图生图默认上传路径
DEFAULT_INPUT_PATH = os.getenv('DEFAULT_INPUT_PATH')
# 设置 ComfyUI API JSON 文生图 Workflow 文件路径
COMFY_T2I_WORKFLOW_API_JSON_PATH = os.getenv('COMFY_T2I_WORKFLOW_API_JSON_PATH')
# 设置 ComfyUI API JSON 图生图 Workflow 文件路径
COMFY_I2I_WORKFLOW_API_JSON_PATH = os.getenv('COMFY_I2I_WORKFLOW_API_JSON_PATH')
# 设置 ComfyUI 使用的 Checkpoint 名称
CHECKPOINT_NAME = os.getenv('CHECKPOINT_NAME')

# 如果环境变量未设置，你可以手动赋值：
API_KEY = ""
SERVER_ADDRESS = "cloud.infini-ai.com"
DEFAULT_WORKFLOW_ID = ""
DEFAULT_OUTPUT_PATH = './output'
COMFY_T2I_WORKFLOW_API_JSON_PATH='./comfy_exported_t2i_workflow_api.json'
# COMFY_I2I_WORKFLOW_API_JSON_PATH='./comfy_exported_i2i_workflow_api.json'
# Checkpoint 取值请以 GenStudio 工作流【查看参数】的中可选值为准
CHECKPOINT_NAME = 'common/sd1.5/majicMIX realistic 麦橘写实_v7.safetensors'

# 验证所有环境变量是否设置
environment_variables = [API_KEY, SERVER_ADDRESS, DEFAULT_WORKFLOW_ID, DEFAULT_OUTPUT_PATH, COMFY_T2I_WORKFLOW_API_JSON_PATH, CHECKPOINT_NAME]
if not all(environment_variables):
    missing_variables = [var_name for var_name, var_value in zip(['API_KEY', 'SERVER_ADDRESS', 'DEFAULT_WORKFLOW_ID', 'DEFAULT_OUTPUT_PATH', 'COMFY_t2i_WORKFLOW_API_JSON_PATH', 'CHECKPOINT_NAME'], environment_variables) if not var_value]
    raise ValueError(f"以下环境变量未设置：{', '.join(missing_variables)}")

print("所有全局变量已设置。")

API 端点

GenStudio ComfyUI 服务提供了封装原始 ComfyUI HTTP 端点的新端点。以下是端点映射：

原始 ComfyUI 端点	描述	我们的 API 端点
POST /prompt	提交任务（请求示例）	POST /api/maas/comfy_task_api/prompt
GET /history/	获取任务排队信息和生图结果	POST /api/maas/comfy_task_api/get_task_info
POST /upload/image	上传图像	POST /api/maas/comfy_task_api/upload/image

GenStuio 的 API Base URL 为 https://cloud.infini-ai.com。

提交任务

使用 POST /api/maas/comfy_task_api/prompt 端点提交任务，将其放入工作流队列。参数 workflow_id 用来引用托管的 ComfyUI 工作流。参数 prompt 是当前请求的提示词，可用于修改托管工作流中的提示文本和配置参数。

python

import requests
import json

def queue_prompt(workflow_id=DEFAULT_WORKFLOW_ID, prompt=None):
    """
    将提示排队处理。

    参数：
        workflow_id (str): 工作流ID。必需参数，但默认使用环境变量中的值。
        prompt (dict, 可选): 要处理的提示。默认为None。

    返回：
        dict: API的JSON响应，如果发生错误则返回None。

    异常：
        ValueError: 如果workflow_id为None或空字符串。
    """
    if not workflow_id:
        raise ValueError("workflow_id 是必需的，不能为空")

    url = f"https://{SERVER_ADDRESS}/api/maas/comfy_task_api/prompt"
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {API_KEY}'
    }
    payload = {
        "workflow_id": workflow_id,
        "prompt": prompt
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"queue_prompt 出错，未能将生图请求加入队列: {e}")
        return None

获取任务信息

使用 POST /api/maas/comfy_task_api/get_task_info 端点获取 ComfyUI 生图任务的排队状态和生图结果。

python

def get_task_info(prompt_ids, url_expire_period=3600):
    url = f"https://{SERVER_ADDRESS}/api/maas/comfy_task_api/get_task_info"
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {API_KEY}'
    }
    payload = {
        "comfy_task_ids": prompt_ids,
        "url_expire_period": url_expire_period
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

示例响应体（排队状态）：

json

{
    "code":0,
    "msg":"Success",
    "data":{
        "comfy_task_info":[
            {
                "comfy_task_id":"cft-c7yg23q4ao5wkcsa",
                "status":1,
                "queue_size":1,
                "current_position":1,
                "errMsg":"",
                "files":"None"
            }
        ]
    }
}

示例响应体（生图成功）：

json

{
    "code":0,
    "msg":"Success",
    "data":{
        "comfy_task_info":[
            {
                "comfy_task_id":"cft-c7yg23q4ao5wkcsa",
                "status":3,
                "queue_size":0,
                "current_position":0,
                "errMsg":"",
                "files":{
                    "9": "https://infini-imagegen-dev.oss-cn-beijing.aliyuncs.com/te-c7e3pay4xevcmqxc%2Fac-c7huz2vofrgclbrx%2Fcft-c7yg23q4ao5wkcsa%2Fec3800ed-eafa-4757-9d83-275f7bcb8df4.png?Expires=1725002942&OSSAccessKeyId=LTAI5tC2tDVy5NzETEJRqQ88&Signature=9LPFCwSFdgoOXMrcdj0AiaGur4o%3D"
                }
            }
        ]
    }
}

在这个响应中，status 字段表示任务的当前状态：

1: 排队中
2: 处理中
3: 已完成
4: 失败
5: 权限不足

上传图像

使用 POST /api/maas/comfy_task_api/upload/image 端点上传图像。

python

def upload_image(file_path):
    url = f"https://{server_address}/api/maas/comfy_task_api/upload/image"
    headers = {
        'Accept': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }
    with open(file_path, 'rb') as file:
        files = {'source_file': file}
        response = requests.post(url, headers=headers, files=files)
    response.raise_for_status()
    return response.json()

# 使用示例
file_path = "path/to/your/image.png"
result = upload_image(file_path)
print(result)  # 打印上传结果，包含 image_id

上传图像的响应示例：

json

{
    "image_id": "te-c7e3pay4xevcmqxc/ac-c7e4ab5hkyjfvmp6/sui-c7yb56jtr6mz7ewp.png"
}

搭建文生图 API 工作流

在下面的流程中，我们将描述如何搭建一个可以修改生图参数的 API 工作流。

获取 Base Promt

如果不修改托管工作流的任何参数，在提交任务时，仅需传入 workflow_id。如果需要修改托管工作流参数，则必须传入 promt。我们将从GenStudio 托管的工作流获取全部参数，经过特殊处理后生成 Base Promt，便于后续传值和修改参数。

WARNING

prompt 参数的输入值并非 ComfyUI 导出的 Workflow API JSON 文件。

从 GenStudio 托管的工作流获取全部参数：

alt text

将复制的全部参数作为 get_prompt_json_body 函数输入，生成 Base Promt。

python

def get_prompt_json_body(api_params):
    """
    提取 GenStudio 托管工作流参数，制作 API 的 Promt 输入。

    参数：
        api_params (dict): 从 GenStudio 托管工作流复制的工作流参数。

    """
    prompt_json_body = {}
    
    for key, value in api_params.items():
        # Initialize the new structure with top-level keys
        prompt_json_body[key] = {'inputs': {}}
        
        # Focus on 'inputs' only
        if 'inputs' in value:
            for input_key, input_value in value['inputs'].items():
                # Swap the value to 'current' in the sub-level objects
                prompt_json_body[key]['inputs'][input_key] = input_value.get('current')

    return prompt_json_body

# 调用函数，制作 API 的 Promt 输入
api_params_by_genstudio = {}  # 你复制的 GenStudio 托管工作流参数
prompt_from_api_params = get_prompt_json_body(api_params_by_genstudio)
# 打印结果
print(json.dumps(prompt_from_api_params, indent=2, ensure_ascii=False))

查看示例：复制的 GenStudio 托管工作流参数

# Example usage
api_params_by_genstudio = {
  "3": {
    "inputs": {
      "seed": {
        "type": "INT",
        "current": 1,
        "spec": {
          "default": 0,
          "min": 0,
          "max": 18446744073709552000
        }
      },
      "steps": {
        "type": "INT",
        "current": 20,
        "spec": {
          "default": 20,
          "min": 1,
          "max": 10000
        }
      },
      "cfg": {
        "type": "FLOAT",
        "current": 7,
        "spec": {
          "default": 8,
          "min": 0,
          "max": 100,
          "step": 0.1,
          "round": 0.01
        }
      },
      "sampler_name": {
        "type": "LIST",
        "current": "ddim",
        "spec": {
          "choices": [
            "euler",
            "euler_cfg_pp",
            "euler_ancestral",
            "euler_ancestral_cfg_pp",
            "heun",
            "heunpp2",
            "dpm_2",
            "dpm_2_ancestral",
            "lms",
            "dpm_fast",
            "dpm_adaptive",
            "dpmpp_2s_ancestral",
            "dpmpp_sde",
            "dpmpp_sde_gpu",
            "dpmpp_2m",
            "dpmpp_2m_alt",
            "dpmpp_2m_sde",
            "dpmpp_2m_sde_gpu",
            "dpmpp_3m_sde",
            "dpmpp_3m_sde_gpu",
            "ddpm",
            "lcm",
            "ipndm",
            "ipndm_v",
            "deis",
            "ddim",
            "uni_pc",
            "uni_pc_bh2"
          ]
        }
      },
      "scheduler": {
        "type": "LIST",
        "current": "karras",
        "spec": {
          "choices": [
            "normal",
            "karras",
            "exponential",
            "sgm_uniform",
            "simple",
            "ddim_uniform",
            "beta"
          ]
        }
      },
      "denoise": {
        "type": "FLOAT",
        "current": 1,
        "spec": {
          "default": 1,
          "min": 0,
          "max": 1,
          "step": 0.01
        }
      }
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "KSampler"
    }
  },
  "4": {
    "inputs": {
      "ckpt_name": {
        "type": "MODEL",
        "current": "majicMIX realistic 麦橘写实_v7.safetensors",
        "spec": {
          "choices": [
            "SUPIR_pruned/SUPIR-v0F_fp16.safetensors",
            "SUPIR_pruned/SUPIR-v0Q_fp16.safetensors",
            "SVD/svd_xt.safetensors",
            "common/flux/flux1-dev-fp8.safetensors",
            "common/flux/flux1-schnell-fp8.safetensors",
            "common/sd1.5/AWPainting_v1.3.safetensors",
            "common/sd1.5/GhostMix鬼混_V2.0.safetensors",
            "common/sd1.5/juggernaut_reborn.safetensors",
            "common/sd1.5/majicMIX realistic 麦橘写实_v7.safetensors",
            "common/sd1.5/v1-5-pruned-emaonly.safetensors",
            "common/sd1.5/美女AnyCharacterMix (Baked Vae)_v2.0 baked vae.safetensors",
            "common/sd3/sd3_medium.safetensors",
            "common/sd3/sd3_medium_incl_clips.safetensors",
            "common/sd3/sd3_medium_incl_clips_t5xxlfp16.safetensors",
            "common/sd3/sd3_medium_incl_clips_t5xxlfp8.safetensors",
            "common/sdxl/ArtiWaifu Diffusion 1.0 vae_0.1.safetensors",
            "common/sdxl/LEOSAM AIArt 兔狲插画 SDXL大模型_v1.safetensors",
            "common/sdxl/LEOSAM HelloWorld 新世界 | SDXL大模型_v2.safetensors",
            "common/sdxl/SDXL-Anime | 天空之境 _v3.1.safetensors",
            "common/sdxl/SDXL_ArienMixXL_V4.5.safetensors",
            "common/sdxl/SHMILY绚丽多彩_V1.0.safetensors",
            "common/sdxl/promissingRealistic_v35.safetensors",
            "common/sdxl/sd_xl_base_1.0.safetensors",
            "common/sdxl/sd_xl_refiner_1.0.safetensors",
            "flux/lcm_lora_sd15.safetensors",
            "sd1.5/majicMIX realistic 麦橘写实_v7.safetensors",
          ]
        }
      }
    },
    "class_type": "CheckpointLoaderSimple",
    "_meta": {
      "title": "Load Checkpoint"
    }
  },
  "5": {
    "inputs": {
      "width": {
        "type": "INT",
        "current": 512,
        "spec": {
          "default": 512,
          "min": 16,
          "max": 16384,
          "step": 8
        }
      },
      "height": {
        "type": "INT",
        "current": 512,
        "spec": {
          "default": 512,
          "min": 16,
          "max": 16384,
          "step": 8
        }
      },
      "batch_size": {
        "type": "INT",
        "current": 1,
        "spec": {
          "default": 1,
          "min": 1,
          "max": 4096
        }
      }
    },
    "class_type": "EmptyLatentImage",
    "_meta": {
      "title": "Empty Latent Image"
    }
  },
  "6": {
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "a girl",
        "spec": {}
      }
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  },
  "7": {
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "",
        "spec": {}
      }
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  }
}

制作节点 ID 映射表

从 ComfyUI 导出的 Workflow API JSON 文件中获取节点 ID 与 Class type 的映射表。这个映射表在修改工作流参数时用于查询节点 ID。

NOTE

这里再次用到了 ComfyUI 导出的 Workflow API JSON 文件，即之前上传至 GenStudio 的 Workflow API JSON 文件。

python

# import json

def get_id_to_class_type_map(COMFY_WORKFLOW_API_JSON_PATH):
    try:
        with open(COMFY_WORKFLOW_API_JSON_PATH, 'r', encoding='utf-8') as f:
            data = json.load(f)
    except FileNotFoundError:
        print(f"未找到文件 {COMFY_WORKFLOW_API_JSON_PATH}")
        return None
    except json.JSONDecodeError:
        print(f"文件 {COMFY_WORKFLOW_API_JSON_PATH} 包含无效的 JSON。")
        return None

    id_to_class_type = {}

    for node_id, node_data in data.items():
        # Default mapping
        id_to_class_type[node_id] = node_data['class_type']

    # 特殊处理 KSampler 节点，从中获取正负提示词的节点 ID
    for node_id, node_data in data.items():
        if node_data['class_type'] == 'KSampler':
            positive_input_id = node_data['inputs'].get('positive', [None])[0]
            negative_input_id = node_data['inputs'].get('negative', [None])[0]

            if positive_input_id is not None:
                id_to_class_type[positive_input_id] = 'positive_input_id'
            if negative_input_id is not None:
                id_to_class_type[negative_input_id] = 'negative_input_id'

    # Return the dictionary
    return id_to_class_type

获取和保存图像

定义两个工具函数，用于获取和保存生成的图像。下一步的 generate_image_by_prompt 函数中会调用这些函数。

python

# import os
# import image
# from urllib.parse import urlparse


def get_images(files_dict):
    """
    从给定的文件字典中获取图像。

    参数:
        files_dict (dict): 包含图像URL的字典。键是图像类型标识符，值是URL列表。

    返回:
        list: 包含图像信息的字典列表。每个字典包含以下键：
            - 'file_name': 图像文件名
            - 'image_data': 图像二进制数据
            - 'type': 图像类型 ('output')
            - 'key': 原字典中的键

    异常:
        打印获取图像失败的错误信息，但不中断程序执行。
    """
    output_images = []
    for key, urls in files_dict.items():
        for url in urls:
            try:
                response = requests.get(url)
                response.raise_for_status()
                file_name = os.path.basename(urlparse(url).path.split('?')[0])  # Remove query parameters from filename
                output_images.append({
                    'file_name': file_name,
                    'image_data': response.content,
                    'type': 'output',
                    'key': key
                })
            except Exception as e:
                print(f"获取图像失败 {url}: {e}")
    return output_images

def save_images(images, output_path, save_previews):
    """
    将图像保存到指定路径。

    参数:
        images (list): 包含图像信息的字典列表。每个字典应包含以下键：
            - 'file_name': 图像文件名
            - 'image_data': 图像二进制数据
            - 'type': 图像类型 ('output' 或 'temp')
            - 'key': 用于创建子目录的键，以出图节点 ID 为子目录名称
        output_path (str): 保存图像的基本路径
        save_previews (bool): 是否保存预览图像（类型为'temp'的图像）

    返回:
        None

    异常:
        打印保存图像失败的错误信息，但不中断程序执行。
    """
    for item in images:
        directory = os.path.join(output_path, item['key'])
        if item['type'] == 'temp' and save_previews:
            directory = os.path.join(directory, 'temp')
        os.makedirs(directory, exist_ok=True)
        try:
            if 'image_data' not in item or item['image_data'] is None:
                print(f"跳过保存图像 {item.get('file_name', 'unknown')}: 没有图像数据")
                continue
            image = Image.open(io.BytesIO(item['image_data']))
            image.save(os.path.join(directory, item['file_name']))
            print(f"成功保存图像: {os.path.join(directory, item['file_name'])}")
        except Exception as e:
            print(f"保存图像 {item.get('file_name', 'unknown')} 失败: {e}")

处理与 API 的交互

generate_image_by_prompt 函数负责实际处理与 GenStudio ComfyUI API 的交互，包括提交请求、等待任务完成、获取图像。

python

# import json
# import time

def generate_image_by_prompt(workflow_id=DEFAULT_WORKFLOW_ID, prompt=None, output_path=DEFAULT_OUTPUT_PATH, save_previews=False):
    """
    根据提示生成图像并保存到指定路径。

    参数：
        workflow_id (str): 工作流ID。默认使用环境变量中的值。
        prompt (dict, 可选): 用于生成图像的提示。默认为None。
        output_path (str): 保存生成图像的路径。此参数是必需的。
        save_previews (bool): 是否保存预览图像。默认为False。

    返回：
        dict: 包含prompt_id和生成的图像数据的字典。失败时返回None。

    异常：
        ValueError: 如果workflow_id为None或空字符串，或者output_path为None或空字符串。
    """
    if not workflow_id:
        raise ValueError("workflow_id 是必需的，不能为空")
    if not output_path:
        raise ValueError("output_path 是必需的，不能为空")

    try:
        print(f"提交图像请求，使用 Prompt: {prompt}")
        response = queue_prompt(workflow_id, prompt)
        prompt_id = response['data']['prompt_id']
        
        while True:
            task_info = get_task_info([prompt_id])
            if task_info['code'] != 0:
                raise Exception(f"获取任务信息失败: {task_info['msg']}")
            
            comfy_task_info = task_info['data']['comfy_task_info'][0]
            status = comfy_task_info['status']
            if status == 3:  # 已完成
                break
            elif status == 4:  # 失败
                raise Exception(f"任务失败: {comfy_task_info['errMsg']}")
            time.sleep(5)  # 等待 5 秒后再次检查
        
        files_dict = comfy_task_info.get('files', {})
        images = get_images(files_dict)
        save_images(images, output_path, save_previews)
        
        return prompt_id
    except Exception as e:
        print(f"生成图像时发生错误: {e}")
        return None

通过 API 调用，将文生图请求加入队列。
通过 API 调用，轮询任务状态，直到任务完成或失败。状态 3 表示全部图片生成结束。
下载当前 Workflow 生成的所有图像。
以出图节点 ID 为子目录，将图像保存到本地。

修改工作流参数

在 Base Prompt 中找到相关节点并调整参数值，例如正面提示，负面提示，seed，cfg 等。

NOTE

您可以通过修改参数复用托管的特定工作流（workflow_id）。如果不希望传 prompt，也可以直接创建多个工作流。

TIP

上传工作流并经平台校验通过后，平台会自动生成入参范围。获取方式详见查看入参范围。

查询前面制作的节点 ID 映射表（id_to_class_type），定位目标节点、修改参数。

python

# import json
# import random

def prompt_to_image(workflow_id=DEFAULT_WORKFLOW_ID, prompt=None, positive_prompt=None, negative_prompt='', output_path=None, save_previews=False):
    """
    根据给定的提示修改工作流并生成图像。

    参数：
        workflow_id (str): 工作流ID。默认使用环境变量中的值。
        prompt (dict 或 str): 工作流提示。如果为None，则直接使用原始工作流。
        positive_prompt (str, 可选): 正面提示文本。
        negative_prompt (str): 负面提示文本。默认为空字符串。
        output_path (str): 保存生成图像的路径。
        save_previews (bool): 是否保存预览图像。默认为False。

    返回：
        tuple: (修改后的工作流JSON字符串, prompt_id)。如果出错，则返回 (None, None)。

    异常：
        ValueError: 如果workflow_id为None或空字符串，或者output_path为None或空字符串。
    """
    if not workflow_id:
        raise ValueError("workflow_id是必需的，不能为空")
    if not output_path:
        raise ValueError("output_path是必需的，不能为空")

    try:
        if prompt is None:
            print("工作流的 Base Prompt 为空，直接调用生成函数")
            prompt_id = generate_image_by_prompt(workflow_id, None, output_path, save_previews)
            return None, prompt_id
        else:

            prompt = json.loads(prompt) if isinstance(prompt, str) else prompt
            print("加载工作流 Base Prompt: ", prompt)

            # 使用从节点 ID 到类型的映射 id_to_class_type，找到 CheckpointLoaderSimple 节点数字
            ckpt_name = next((key for key, value in id_to_class_type.items() if value == 'CheckpointLoaderSimple'), None)
            if ckpt_name is None:
                print("找不到 ckpt_name 节点")
                return None

            # 设置正确的 checkpoint 名称，必须以 GenStudio 返回的 checkpoint 名称为准
            prompt[ckpt_name]['inputs']['ckpt_name'] = CHECKPOINT_NAME

            # 使用从节点 ID 到类型的映射 id_to_class_type，找到 KSampler 节点数字
            k_sampler = next((key for key, value in id_to_class_type.items() if value == 'KSampler'), None)
            if k_sampler is None:
                print("找不到 KSampler 节点")
                return None

            # 为 KSampler 节点设置一个随机种子
            prompt[k_sampler]['inputs']['seed'] = random.randint(10**14, 10**15 - 1)

            # 获取 positive_input_id 和 negative_input_id
            positive_input_id = next((key for key, value in id_to_class_type.items() if value == 'positive_input_id'), None)
            negative_input_id = next((key for key, value in id_to_class_type.items() if value == 'negative_input_id'), None)

            # 更新正面提示
            if positive_input_id is not None:
                prompt[positive_input_id]['inputs']['text'] = positive_prompt

            # 如果提供了负面提示，则更新负面提示
            if negative_prompt != '' and negative_input_id is not None:
                prompt[negative_input_id]['inputs']['text'] = negative_prompt

        prompt_id = generate_image_by_prompt(workflow_id, prompt, output_path, save_previews)
        return json.dumps(prompt), prompt_id

    except Exception as e:
        print(f"处理提示时发生错误: {e}")
        return None, None

使用文生图 API 工作流

alt text

使用文本提示到图像的 API 工作流：

python

# 前面已生成了 Base Prompt，此处直接获取即可
base_workflow_prompt = prompt_from_api_params

# 为了方便修改 Base Promt，从 ComfyUI 导出的 Workflow API JSON 中
# 获取 node id 到 class type 的映射
id_to_class_type=get_id_to_class_type_map(COMFY_t2i_WORKFLOW_API_JSON_PATH)

# 修改 Base Prompt 中的提示词
positive_prompt = "Spiderman in a red suit standing in middle of a crowded place, skyscrapers in the background, cinematic, neon colors, realistic look"
negative_prompt = "ugly, deformed"

# 调用函数生成图像
modified_prompt, prompt_id = prompt_to_image(
    prompt=base_workflow_prompt,
    positive_prompt=positive_prompt,
    negative_prompt=negative_prompt,
    output_path="./output",
    save_previews=True
)

if prompt_id:
    print(f"修改后的 Prompt: {modified_prompt}")
    print(f"生图成功 Prompt ID: {prompt_id}")
else:
    print("生图失败")

示例介绍

示例使用以下设置：

工作流： ComfyUI 导出的文生图工作流
模型：majicMIX realistic 麦橘写实_v7.safetensors
采样器名称：ddim
调度器：karras
步骤：20
cfg：7
API 更改后的正面提示：Spiderman in a red suit standing in middle of a crowded place, skyscrapers in the background, cinematic, neon colors, realistic look
API 更改后负面提示：ugly, deformed

本示例提供了一个 Jupyter Notebook 文件

文生图 API 请求体示例

提交生图任务接口是最核心的接口，以下上述代码实际生成的 POST /api/maas/comfy_task_api/prompt 接口的 JSON 请求正文。

json

{
  "workflow_id": "wf-c72qozydy6suxesa",
  "prompt": {
    "3": {
      "inputs": {
        "seed": 423710630168223,
        "steps": 20,
        "cfg": 7,
        "sampler_name": "ddim",
        "scheduler": "karras",
        "denoise": 1
      }
    },
    "4": {
      "inputs": {
        "ckpt_name": "common/sd1.5/majicMIX realistic 麦橘写实_v7.safetensors"
      }
    },
    "5": {
      "inputs": {
        "width": 512,
        "height": 512,
        "batch_size": 1
      }
    },
    "6": {
      "inputs": {
        "text": "Spiderman in a red suit standing in middle of a crowded place, skyscrapers in the background, cinematic, neon colors, realistic look"
      }
    },
    "7": {
      "inputs": {
        "text": "ugly, deformed"
      }
    }
  }
}

搭建图生图 API 工作流

首先，您必须在 ComfyUI 中构建基本的图像到图像工作流程，使用加载图像（Load Image）和 VEA 编码（VEA Encode）节点。

获取 Base Promt

与文生图的流程相同，获取 Base Prompt。

python

# 调用函数，制作 API 的 Promt 输入
api_params_by_genstudio = {}  # 你复制的 GenStudio 托管工作流参数
prompt_from_api_params = get_prompt_json_body(api_params_by_genstudio)

# 打印结果
print(json.dumps(prompt_from_api_params, indent=2, ensure_ascii=False))

查看示例：复制的 GenStudio 托管图生图工作流参数

python

api_params_by_genstudio = {
  "3": {
    "inputs": {
      "seed": {
        "type": "INT",
        "current": 156680208700286,
        "spec": {
          "default": 0,
          "min": 0,
          "max": 18446744073709552000
        }
      },
      "steps": {
        "type": "INT",
        "current": 22,
        "spec": {
          "default": 20,
          "min": 1,
          "max": 10000
        }
      },
      "cfg": {
        "type": "FLOAT",
        "current": 8,
        "spec": {
          "default": 8,
          "min": 0,
          "max": 100,
          "step": 0.1,
          "round": 0.01
        }
      },
      "sampler_name": {
        "type": "LIST",
        "current": "dpmpp_3m_sde",
        "spec": {
          "choices": [
            "euler",
            "euler_cfg_pp",
            "euler_ancestral",
            "euler_ancestral_cfg_pp",
            "heun",
            "heunpp2",
            "dpm_2",
            "dpm_2_ancestral",
            "lms",
            "dpm_fast",
            "dpm_adaptive",
            "dpmpp_2s_ancestral",
            "dpmpp_sde",
            "dpmpp_sde_gpu",
            "dpmpp_2m",
            "dpmpp_2m_alt",
            "dpmpp_2m_sde",
            "dpmpp_2m_sde_gpu",
            "dpmpp_3m_sde",
            "dpmpp_3m_sde_gpu",
            "ddpm",
            "lcm",
            "ipndm",
            "ipndm_v",
            "deis",
            "ddim",
            "uni_pc",
            "uni_pc_bh2"
          ]
        }
      },
      "scheduler": {
        "type": "LIST",
        "current": "karras",
        "spec": {
          "choices": [
            "normal",
            "karras",
            "exponential",
            "sgm_uniform",
            "simple",
            "ddim_uniform",
            "beta"
          ]
        }
      },
      "denoise": {
        "type": "FLOAT",
        "current": 0.6,
        "spec": {
          "default": 1,
          "min": 0,
          "max": 1,
          "step": 0.01
        }
      }
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "KSampler"
    }
  },
  "4": {
    "inputs": {
      "ckpt_name": {
        "type": "MODEL",
        "current": "sdXL_v10VAEFix.safetensors",
        "spec": {
          "choices": [
            "AWPainting_v1.3.safetensors",
            "ArtiWaifu Diffusion 1.0 vae_0.1.safetensors",
            "GhostMix鬼混_V2.0.safetensors",
            "LEOSAM AIArt 兔狲插画 SDXL大模型_v1.safetensors",
            "LEOSAM HelloWorld 新世界 | SDXL大模型_v2.safetensors",
            "SDXL-Anime | 天空之境 _v3.1.safetensors",
            "SDXL_ArienMixXL_V4.5.safetensors",
            "SHMILY绚丽多彩_V1.0.safetensors",
            "SUPIR_pruned/SUPIR-v0F_fp16.safetensors",
            "SUPIR_pruned/SUPIR-v0Q_fp16.safetensors",
            "SVD/svd_xt.safetensors",
            "majicMIX realistic 麦橘写实_v7.safetensors",
            "sd3_medium.safetensors",
            "sd_xl_base_1.0.safetensors",
            "sd_xl_refiner_1.0.safetensors",
            "v1-5-pruned-emaonly.safetensors",
            "美女AnyCharacterMix (Baked Vae)_v2.0 baked vae.safetensors"
          ]
        }
      }
    },
    "class_type": "CheckpointLoaderSimple",
    "_meta": {
      "title": "Load Checkpoint"
    }
  },
  "6": {
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
        "spec": {}
      }
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  },
  "7": {
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "text, watermark",
        "spec": {}
      }
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  },
  "10": {
    "inputs": {
      "image": {
        "type": "IMAGE",
        "current": "example.png",
        "spec": {
          "choices": []
        }
      }
    },
    "class_type": "LoadImage",
    "_meta": {
      "title": "Load Image"
    }
  }
}

修改工作流参数

在图生图的流程中，在 Base Prompt 中找到相关节点并调整参数值的逻辑与文生图的逻辑相同。主要区别在于修改 LoadImage 节点的输入，将输入改为 "POST /api/maas/comfy_task_api/upload/image" 返回的 image_id。

查询节点 ID 映射表（id_to_class_type），在给定的的 Base Promt 中寻找节点、修改参数。

python

# import json
# import random

def prompt_image_to_image(workflow_id=DEFAULT_WORKFLOW_ID, prompt=None, positive_prompt=None, negative_prompt='', input_path=DEFAULT_INPUT_PATH,output_path=None, save_previews=False):
    """
    根据给定的提示修改工作流并生成图像。

    参数：
        workflow_id (str): 工作流ID。默认使用环境变量中的值。
        prompt (dict 或 str): 工作流提示。如果为None，则直接使用原始工作流。
        positive_prompt (str, 可选): 正面提示文本。
        negative_prompt (str): 负面提示文本。默认为空字符串。
        input_path (str): 图生图的输入图片的路径。此参数是必需的。
        output_path (str): 保存生成图像的路径。
        save_previews (bool): 是否保存预览图像。默认为False。

    返回：
        tuple: (修改后的工作流JSON字符串, prompt_id)。如果出错，则返回 (None, None)。

    异常：
        ValueError: 如果workflow_id为None或空字符串，或者output_path为None或空字符串。
    """
    if not workflow_id:
        raise ValueError("workflow_id是必需的，不能为空")
    if not output_path:
        raise ValueError("output_path是必需的，不能为空")

    try:
        if prompt is None:
            print("工作流的 Base Prompt 为空，直接调用生成函数")
            prompt_id = generate_image_by_prompt(workflow_id, None, output_path, save_previews)
            return None, prompt_id
        else:

            prompt = json.loads(prompt) if isinstance(prompt, str) else prompt
            print("加载工作流 Base Prompt: ", prompt)

            # 使用从节点 ID 到类型的映射 id_to_class_type
            # 找到 LoadImage 节点数字
            load_image = next((key for key, value in id_to_class_type.items() if value == 'LoadImage'), None)
            if load_image is None:
                print("找不到 LoadImage 节点")

            # 上传图片到 OSS
            try:
                oss_image_id = upload_image(input_path)
                print(f"文件上传成功，image_id: {oss_image_id}")
            except requests.RequestException as e:
                print(f"上传请求失败: {e}")
            except KeyError as e:
                print(f"响应中未找到预期的数据结构: {e}")
            except Exception as e:
                print(f"发生未预期的错误: {e}")
            
            prompt[load_image]['image'] = oss_image_id

            # 使用从节点 ID 到类型的映射 id_to_class_type，找到 CheckpointLoaderSimple 节点数字
            ckpt_name = next((key for key, value in id_to_class_type.items() if value == 'CheckpointLoaderSimple'), None)
            if ckpt_name is None:
                print("找不到 ckpt_name 节点")
                return None

            # 设置正确的 checkpoint 名称，必须以 GenStudio 返回的 checkpoint 名称为准
            prompt[ckpt_name]['inputs']['ckpt_name'] = CHECKPOINT_NAME

            # 找到 KSampler 节点数字
            k_sampler = next((key for key, value in id_to_class_type.items() if value == 'KSampler'), None)
            if k_sampler is None:
                print("找不到 KSampler 节点")
                return None

            # 为 KSampler 节点设置一个随机种子
            prompt[k_sampler]['inputs']['seed'] = random.randint(10**14, 10**15 - 1)

            # 获取 positive_input_id 和 negative_input_id
            positive_input_id = next((key for key, value in id_to_class_type.items() if value == 'positive_input_id'), None)
            negative_input_id = next((key for key, value in id_to_class_type.items() if value == 'negative_input_id'), None)

            # 更新正面提示
            if positive_input_id is not None:
                prompt[positive_input_id]['inputs']['text'] = positive_prompt

            # 如果提供了负面提示，则更新负面提示
            if negative_prompt != '' and negative_input_id is not None:
                prompt[negative_input_id]['inputs']['text'] = negative_prompt

        prompt_id = generate_image_by_prompt(workflow_id, prompt, output_path, save_previews)
        return json.dumps(prompt), prompt_id

    except Exception as e:
        print(f"处理提示时发生错误: {e}")
        return None, None

处理与 API 的交互

generate_image_by_prompt_and_image 函数负责实际处理与 GenStudio ComfyUI API 的交互，包括提交请求、等待任务完成、获取图像。

python

# import json
# import time

def generate_image_by_prompt_and_image(workflow_id=DEFAULT_WORKFLOW_ID, prompt=None, output_path=DEFAULT_OUTPUT_PATH, save_previews=False):
    """
    根据提示和输入图像生成新图像并保存到指定路径。

    参数：
        workflow_id (str): 工作流ID。默认使用环境变量中的值。
        prompt (dict, 可选): 用于生成图像的提示。默认为None。
        output_path (str): 保存生成图像的路径。此参数是必需的。
        save_previews (bool): 是否保存预览图像。默认为False。

    返回：
        str: 生成的prompt_id。失败时返回None。

    异常：
        ValueError: 如果workflow_id为None或空字符串，或者output_path为None或空字符串。
    """
    if not workflow_id:
        raise ValueError("workflow_id 是必需的，不能为空")
    if not output_path:
        raise ValueError("output_path 是必需的，不能为空")

    try:
        print(f"提交图像请求，使用 Prompt: {prompt}")
        response = queue_prompt(workflow_id, prompt)
        prompt_id = response['data']['prompt_id']
        
        while True:
            task_info = get_task_info([prompt_id])
            if task_info['code'] != 0:
                raise Exception(f"获取任务信息失败: {task_info['msg']}")
            
            comfy_task_info = task_info['data']['comfy_task_info'][0]
            status = comfy_task_info['status']
            if status == 3:  # 已完成
                break
            elif status == 4:  # 失败
                raise Exception(f"任务失败: {comfy_task_info['errMsg']}")
            time.sleep(5)  # 等待 5 秒后再次检查
        
        files_dict = comfy_task_info.get('files', {})
        images = get_images(files_dict)
        save_images(images, output_path, save_previews)
        
        return prompt_id
    except Exception as e:
        print(f"生成图像时发生错误: {e}")
        return None

通过 API 调用，将图生图请求加入队列。
通过 API 调用，轮询任务状态，直到任务完成或失败。状态 3 表示全部图片生成结束。
下载当前 Workflow 生成的所有图像。
以出图节点 ID 为子目录，将图像保存到本地。

常见问题

支持什么 ComfyUI 节点？

我们预置了部分常用节点。如果您的 Workflow 中使用了不受支持的 ComfyUI 节点，将无法通过 Workflow API JSON 校验。

查看已支持的 Checkpoint 和自定义节点： ComfyUI HTTP API 参考文档。

如果您需要更多节点，请联系 GenStudio 团队。

通过 API 使用托管 ComfyUI 工作流 ​

简介 ​

托管 ComfyUI 工作流 ​

导出 API Format 工作流 ​

上传 API Format 工作流 ​

查看入参范围 ​

API 鉴权设置 ​

Jupyter Notebook 设置 ​

API 端点 ​

提交任务 ​

获取任务信息 ​

上传图像 ​

搭建文生图 API 工作流 ​

获取 Base Promt ​

制作节点 ID 映射表 ​

获取和保存图像 ​

处理与 API 的交互 ​

修改工作流参数 ​

使用文生图 API 工作流 ​

示例介绍 ​

文生图 API 请求体示例 ​

搭建图生图 API 工作流 ​

获取 Base Promt ​

修改工作流参数 ​

处理与 API 的交互 ​

常见问题 ​

支持什么 ComfyUI 节点？ ​

通过 API 使用托管 ComfyUI 工作流

简介

托管 ComfyUI 工作流

导出 API Format 工作流

上传 API Format 工作流

查看入参范围

API 鉴权设置

Jupyter Notebook 设置

API 端点

提交任务

获取任务信息

上传图像

搭建文生图 API 工作流

获取 Base Promt

制作节点 ID 映射表

获取和保存图像

处理与 API 的交互

修改工作流参数

使用文生图 API 工作流

示例介绍

文生图 API 请求体示例

搭建图生图 API 工作流

获取 Base Promt

修改工作流参数

处理与 API 的交互

常见问题

支持什么 ComfyUI 节点？