使用 Python 接入 GenStudio 托管 ComfyUI 工作流 API

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

NOTE

如需总体了解云端全托管 ComfyUI 工作流的解决方案，请参考迁移到云端全托管 ComfyUI 工作流。
关于 API 端点的路径、参数等细节，详见 ComfyUI HTTP API 参考文档。
为了展示如何通过 API 使用传入工作流 ID 和节点参数，我们提供了一个文生图 API 请求体示例。
本文档仅作为说明，可配合使用本教程提供的 Jupyter Notebook 文件。

简介

本文介绍了与 GenStudio 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', 'cloud.infini-ai.com')
# 设置默认工作流 ID
DEFAULT_WORKFLOW_ID = os.getenv('DEFAULT_WORKFLOW_ID', 'wf-c735bjbsu65qcyfw')
# 设置默认输出路径
DEFAULT_OUTPUT_PATH = os.getenv('DEFAULT_OUTPUT_PATH', './output')
# 设置图生图默认上传路径
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', './comfy_exported_i2i_workflow_api.json')
# 设置 ComfyUI API JSON 图生图 Workflow 文件路径
COMFY_I2I_WORKFLOW_API_JSON_PATH = os.getenv('COMFY_I2I_WORKFLOW_API_JSON_PATH', './comfy_exported_i2i_workflow_api.json')
# 设置 ComfyUI 使用的 Checkpoint 名称
CHECKPOINT_NAME = os.getenv('CHECKPOINT_NAME', '/public/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

GenStudio 的 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

在提交任务时，需要传入工作流 ID（workflow_id）和待修改的工作流参数（promt）。API 的 promt 参数为一个 JSON 对象，其中包含了 ComfyUI 工作流所有节点的输入参数。

为了构造 prompt 参数的值，我们需要访问 GenStudio 托管的工作流，通过查看参数按钮复制全部参数，经过特殊处理后生成 Base Promt，便于后续传值和修改。

WARNING

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

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

alt text

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

python

import json

def get_prompt_json_body(api_params_json):
    """
    处理API参数JSON字符串并返回处理后的JSON主体。

    :param api_params_json: 包含API参数的JSON字符串
    :type api_params_json: str
    :return: 处理后的JSON主体
    :rtype: dict

    这个函数执行以下操作：
    1. 将输入的JSON字符串解析为Python字典
    2. 处理解析后的API参数
    3. 创建一个新的JSON结构，只保留'inputs'中的'current'值
    """
    # 将JSON字符串解析为Python字典
    api_params = json.loads(api_params_json)
    
    prompt_json_body = {}
    
    for key, value in api_params.items():
        # 用顶层键初始化新结构
        prompt_json_body[key] = {'inputs': {}}
        
        # 仅关注'inputs'
        if 'inputs' in value:
            for input_key, input_value in value['inputs'].items():
                # 在子级对象中将值替换为'current'
                prompt_json_body[key]['inputs'][input_key] = input_value.get('current')

    return prompt_json_body

# 调用函数，制作 API 的 Promt 输入
# 使用多行字符串（用三引号"""括起来）包裹从"查看参数" 拷贝的JSON数据。
# 允许我们保持JSON的原始格式，避免处理 true/false/null 等问题
t2i_api_params_by_genstudio = """

"""
t2i_prompt_from_api_params = get_prompt_json_body(t2i_api_params_by_genstudio)# 打印结果
print(json.dumps(t2i_prompt_from_api_params, indent=2, ensure_ascii=False))

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

python

# 使用多行字符串（用三引号"""括起来）包裹从"查看参数" 拷贝的JSON数据。
# 允许我们保持JSON的原始格式，避免处理 true/false/null 等问题
t2i_api_params_by_genstudio = """
{
  "3": {
    "_meta": {
      "title": "KSampler"
    },
    "class_type": "KSampler",
    "inputs": {
      "cfg": {
        "type": "FLOAT",
        "current": 7,
        "spec": {
          "default": 8,
          "min": 0,
          "max": 100,
          "step": 0.1,
          "round": 0.01
        },
        "current_valid": true,
        "subtype": null
      },
      "denoise": {
        "type": "FLOAT",
        "current": 1,
        "spec": {
          "default": 1,
          "min": 0,
          "max": 1,
          "step": 0.01
        },
        "current_valid": true,
        "subtype": null
      },
      "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"
          ]
        },
        "current_valid": true,
        "subtype": null
      },
      "scheduler": {
        "type": "LIST",
        "current": "karras",
        "spec": {
          "choices": [
            "normal",
            "karras",
            "exponential",
            "sgm_uniform",
            "simple",
            "ddim_uniform",
            "beta"
          ]
        },
        "current_valid": true,
        "subtype": null
      },
      "seed": {
        "type": "INT",
        "current": 1,
        "spec": {
          "default": 0,
          "min": 0,
          "max": 18446744073709552000
        },
        "current_valid": true,
        "subtype": null
      },
      "steps": {
        "type": "INT",
        "current": 20,
        "spec": {
          "default": 20,
          "min": 1,
          "max": 10000
        },
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "4": {
    "_meta": {
      "title": "Load Checkpoint"
    },
    "class_type": "CheckpointLoaderSimple",
    "inputs": {
      "ckpt_name": {
        "type": "MODEL",
        "current": "majicMIX realistic 麦橘写实_v7.safetensors",
        "spec": {
          "choices": [
            "/private/majicMIX realistic 麦橘写实_v7.safetensors",
            "AWPainting_v1.3.safetensors",
            "juggernaut_reborn.safetensors",
            "v1-5-pruned-emaonly.safetensors",
            "GhostMix鬼混_V2.0.safetensors",
            "majicMIX realistic 麦橘写实_v7.safetensors",
            "美女AnyCharacterMix (Baked Vae)_v2.0 baked vae.safetensors",
            "flux1-dev-fp8.safetensors",
            "sd3_medium_incl_clips.safetensors",
            "sd3_medium_incl_clips_t5xxlfp16.safetensors",
            "sd3_medium_incl_clips_t5xxlfp8.safetensors",
            "sd3_medium.safetensors",
            "ArtiWaifu Diffusion 1.0 vae_0.1.safetensors",
            "LEOSAM AIArt 兔狲插画 SDXL大模型_v1.safetensors",
            "LEOSAM HelloWorld 新世界 | SDXL大模型_v2.safetensors",
            "promissingRealistic_v35.safetensors",
            "SDXL-Anime | 天空之境 _v3.1.safetensors",
            "SDXL_ArienMixXL_V4.5.safetensors",
            "sd_xl_base_1.0.safetensors",
            "sd_xl_refiner_1.0.safetensors",
            "SHMILY绚丽多彩_V1.0.safetensors"
          ]
        },
        "current_valid": false,
        "subtype": "CHECKPOINT"
      }
    }
  },
  "5": {
    "_meta": {
      "title": "Empty Latent Image"
    },
    "class_type": "EmptyLatentImage",
    "inputs": {
      "batch_size": {
        "type": "INT",
        "current": 1,
        "spec": {
          "default": 1,
          "min": 1,
          "max": 4096
        },
        "current_valid": true,
        "subtype": null
      },
      "height": {
        "type": "INT",
        "current": 512,
        "spec": {
          "default": 512,
          "min": 16,
          "max": 16384,
          "step": 8
        },
        "current_valid": true,
        "subtype": null
      },
      "width": {
        "type": "INT",
        "current": 512,
        "spec": {
          "default": 512,
          "min": 16,
          "max": 16384,
          "step": 8
        },
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "6": {
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    },
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "a girl",
        "spec": {},
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "7": {
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    },
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "",
        "spec": {},
        "current_valid": true,
        "subtype": null
      }
    }
  }
}
"""

制作节点 ID 映射表

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

NOTE

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

python

# import json
# import json

def get_id_to_class_type_map(comfy_workflow_api_json_path):
    """
    从 ComfyUI 导出原始 API Format JSON 文件中读取工作流数据，并生成节点 ID 到类型的映射。

    :param comfy_workflow_api_json_path: Comfy工作流 API JSON 文件的路径
    :type comfy_workflow_api_json_path: str
    :return: 节点ID到类型的映射字典，如果出现错误则返回None
    :rtype: dict or None

    此函数执行以下操作：
    1. 尝试读取并解析指定的JSON文件
    2. 为每个节点创建ID到类型的基本映射
    3. 对KSampler节点进行特殊处理，识别其正面和负面输入
    4. 返回生成的映射字典

    如果文件不存在或JSON无效，函数将打印错误消息并返回None。
    """
    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():
        # 默认映射
        id_to_class_type[node_id] = node_data['class_type']

    # 对KSampler节点进行特殊处理
    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 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 为子目录，将图像保存到本地。

修改工作流参数

为了重复使用托管的特定工作流（workflow_id），同时能够更改正面提示，负面提示，seed，cfg 等参数以生成不同的图像，我们以 Base Prompt 为基础，从中找到相关节点并调整参数值，最后将修改后的 Base Prompt 作为 API prompt 参数的值提交给 GenStudio 服务端。

NOTE

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

TIP

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

我们利用前面制作的节点 ID 映射表（id_to_class_type）作为地图，在 Base Promt 中寻找节点、修改参数。

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 工作流

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

python

# 前面已生成了 Base Prompt，此处直接获取即可
base_workflow_prompt = t2i_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

文生图 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

# 调用函数，使用复制的 GenStudio 托管工作流参数制作 API 的 Promt 输入
i2i_api_params_by_genstudio = """
{}
"""
i2i_prompt_from_api_params = get_prompt_json_body(i2i_api_params_by_genstudio)

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

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

python

# 使用多行字符串（用三引号"""括起来）包裹从"查看参数" 拷贝的JSON数据。
# 允许我们保持JSON的原始格式，避免处理 true/false/null 等问题
i2i_api_params_by_genstudio = """
{
  "3": {
    "_meta": {
      "title": "KSampler"
    },
    "class_type": "KSampler",
    "inputs": {
      "cfg": {
        "type": "FLOAT",
        "current": 8,
        "spec": {
          "default": 8,
          "min": 0,
          "max": 100,
          "step": 0.1,
          "round": 0.01
        },
        "current_valid": true,
        "subtype": null
      },
      "denoise": {
        "type": "FLOAT",
        "current": 0.6,
        "spec": {
          "default": 1,
          "min": 0,
          "max": 1,
          "step": 0.01
        },
        "current_valid": true,
        "subtype": null
      },
      "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"
          ]
        },
        "current_valid": true,
        "subtype": null
      },
      "scheduler": {
        "type": "LIST",
        "current": "karras",
        "spec": {
          "choices": [
            "normal",
            "karras",
            "exponential",
            "sgm_uniform",
            "simple",
            "ddim_uniform",
            "beta"
          ]
        },
        "current_valid": true,
        "subtype": null
      },
      "seed": {
        "type": "INT",
        "current": 156680208700286,
        "spec": {
          "default": 0,
          "min": 0,
          "max": 18446744073709552000
        },
        "current_valid": true,
        "subtype": null
      },
      "steps": {
        "type": "INT",
        "current": 22,
        "spec": {
          "default": 20,
          "min": 1,
          "max": 10000
        },
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "4": {
    "_meta": {
      "title": "Load Checkpoint"
    },
    "class_type": "CheckpointLoaderSimple",
    "inputs": {
      "ckpt_name": {
        "type": "MODEL",
        "current": "sdXL_v10VAEFix.safetensors",
        "spec": {
          "choices": [
            "/private/majicMIX realistic 麦橘写实_v7.safetensors",
            "AWPainting_v1.3.safetensors",
            "juggernaut_reborn.safetensors",
            "v1-5-pruned-emaonly.safetensors",
            "GhostMix鬼混_V2.0.safetensors",
            "majicMIX realistic 麦橘写实_v7.safetensors",
            "美女AnyCharacterMix (Baked Vae)_v2.0 baked vae.safetensors",
            "flux1-dev-fp8.safetensors",
            "sd3_medium_incl_clips.safetensors",
            "sd3_medium_incl_clips_t5xxlfp16.safetensors",
            "sd3_medium_incl_clips_t5xxlfp8.safetensors",
            "sd3_medium.safetensors",
            "ArtiWaifu Diffusion 1.0 vae_0.1.safetensors",
            "LEOSAM AIArt 兔狲插画 SDXL大模型_v1.safetensors",
            "LEOSAM HelloWorld 新世界 | SDXL大模型_v2.safetensors",
            "promissingRealistic_v35.safetensors",
            "SDXL-Anime | 天空之境 _v3.1.safetensors",
            "SDXL_ArienMixXL_V4.5.safetensors",
            "sd_xl_base_1.0.safetensors",
            "sd_xl_refiner_1.0.safetensors",
            "SHMILY绚丽多彩_V1.0.safetensors"
          ]
        },
        "current_valid": false,
        "subtype": "CHECKPOINT"
      }
    }
  },
  "6": {
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    },
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
        "spec": {},
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "7": {
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    },
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": {
        "type": "STRING",
        "current": "text, watermark",
        "spec": {},
        "current_valid": true,
        "subtype": null
      }
    }
  },
  "10": {
    "_meta": {
      "title": "Load Image"
    },
    "class_type": "LoadImage",
    "inputs": {
      "image": {
        "type": "IMAGE",
        "current": "example.png",
        "spec": {
          "choices": []
        },
        "current_valid": false,
        "subtype": null
      }
    }
  }
}
"""

修改工作流参数

在图生图的流程中，在 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 中的 image 输入为 OSS 图片 ID
            prompt[load_image]['inputs']['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)

            # 为 KSampler 节点设置一个 sampler，cfg， steps（硬编码）
            prompt[k_sampler]['inputs']['sampler_name'] = 'ddim'
            prompt[k_sampler]['inputs']['cfg'] = 9
            prompt[k_sampler]['inputs']['steps'] = 30

            # 获取 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 为子目录，将图像保存到本地。

使用图生图 API 工作流

使用图像到图像的 API 工作流：

python

# 前面已生成了 Base Prompt，此处直接获取即可
base_workflow_prompt = i2i_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_I2I_WORKFLOW_API_JSON_PATH)

# 修改 Base Prompt 中的提示词
positive_prompt = "Spiderwoman dressed entirely in monochromatic, pure black suit standing in middle of a crowded place, skyscrapers in the background, cinematic, neon colors, realistic look"
negative_prompt = "ugly, deformed, red, reddish, blue"

# 调用函数生成图像
modified_prompt, prompt_id = prompt_image_to_image(
    prompt=base_workflow_prompt,
    positive_prompt=positive_prompt,
    negative_prompt=negative_prompt,
    input_path=DEFAULT_INPUT_PATH,
    output_path=DEFAULT_OUTPUT_PATH,
    save_previews=True
)

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

使用 Python 接入 GenStudio 托管 ComfyUI 工作流 API ​

简介 ​

托管 ComfyUI 工作流 ​

导出 API Format 工作流 ​

上传 API Format 工作流 ​

查看入参范围 ​

API 鉴权设置 ​

Jupyter Notebook 设置 ​

API 端点 ​

提交任务 ​

获取任务信息 ​

上传图像 ​

搭建文生图 API 工作流 ​

获取 Base Promt ​

制作节点 ID 映射表 ​

获取和保存图像 ​

处理与 API 的交互 ​

修改工作流参数 ​

使用文生图 API 工作流 ​

示例介绍 ​

文生图 API 请求体示例 ​

搭建图生图 API 工作流 ​

获取 Base Promt ​

修改工作流参数 ​

处理与 API 的交互 ​

使用图生图 API 工作流 ​

使用 Python 接入 GenStudio 托管 ComfyUI 工作流 API

简介

托管 ComfyUI 工作流

导出 API Format 工作流

上传 API Format 工作流

查看入参范围

API 鉴权设置

Jupyter Notebook 设置

API 端点

提交任务

获取任务信息

上传图像

搭建文生图 API 工作流

获取 Base Promt

制作节点 ID 映射表

获取和保存图像

处理与 API 的交互

修改工作流参数

使用文生图 API 工作流

示例介绍

文生图 API 请求体示例

搭建图生图 API 工作流

获取 Base Promt

修改工作流参数

处理与 API 的交互

使用图生图 API 工作流