通过 API 托管 ComfyUI 工作流(译文)

动机

本文重点介绍如何利用 ComfyUI 的基本工作流功能。您已经创建了一个出色的工作流，并希望与世界分享它或围绕它构建应用程序。通过托管您的项目并利用此 WebSocket API 概念，您可以动态处理用户输入以创建令人难以置信的风格转换或令人惊叹的照片效果。

介绍

这篇博文介绍了与 ComfyUI 通信的 WebSocket API 的基本结构。通过 ComfyUI 生成图像通常需要几秒钟，并且根据工作流程的复杂性，此时间可能会增加。我们利用 WebSocket 连接来跟踪进度并允许我们向用户提供实时反馈。可以在没有 WebSocket 连接的情况下使用这些端点，但这会让您失去实时更新的好处。

基本 WebSocket API 结构的代码可以在这里找到：基本 WebSocket API。

利用 ComfyUI 端点

ComfyUI 已经预定义了端点ComfyUI 端点，我们可以将其作为目标。此外，ComfyUI 还提供了 WebSocket 接口。对于本博文后面描述的 API，我们不需要修改此文件，因为它已经提供了我们需要的一切。

@routes.get('/ws') 返回 WebSocket 对象，发送状态和执行消息

@routes.post("/prompt") 将提示排队到工作流，返回 prompt_id 或错误

@routes.get("/history/{prompt_id}") 返回给定 prompt_id 的队列或输出

@routes.get("/view") 返回给定文件名、子文件夹和类型（“输入”、“输出”、“临时”）的图像

@routes.post("/upload/image") 将图像上传到 ComfyUI，给定 image_data 和类型（“input”、“output”、“temp”）

API 设置

为了实现这一点，我们首先在 API 中创建与 ComfyUI 的 WebSocket 路径的连接。这是通过"/ws"路由完成的。我们可以为我们的基本 API 用例自动生成带有 uuid 的 client_id。并且当我们使用本地运行的 ComfyUI 时server_address,。

Pythondef open_websocket_connection():

server_address='127.0.0.1:8188'

client_id=str(uuid.uuid4())

ws = websocket.WebSocket()

ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id))

return ws, server_address, client_id

API 端点

队列提示

通过 ComfyUI 提供的端点向 ComfyUI 发送提示，将其放入工作流队列"/prompt"。参数是，即我们生成的prompt整个工作流 JSON；以及正在运行的 ComfyUI 实例的client_id,server_address

Pythondef queue_prompt(prompt, client_id, server_address):

p = {"prompt": prompt, "client_id": client_id}

headers = {'Content-Type': 'application/json'}

data = json.dumps(p).encode('utf-8')

req = urllib.request.Request("http://{}/prompt".format(server_address), data=data, headers=headers)

return json.loads(urllib.request.urlopen(req).read())获取历史记录

通过 ndpoint 从 ComfyUI 获取给定提示 ID 的历史记录"/history/{prompt_id}" e。它接收提示的 ID 和server_address正在运行的 ComfyUI 服务器的 ID 作为参数。ComfyUI 返回包含相关输出数据的 JSON，例如带有文件名和目录的图像，然后我们可以使用它来获取这些图像。

Pythondef get_history(prompt_id, server_address):

with urllib.request.urlopen("http://{}/history/{}".format(server_address, prompt_id)) as response:

return json.loads(response.read())获取图像

根据路径、文件名和类型通过"/view"端点从 ComfyUI 检索图像。ComfyUI 返回原始图像数据。

上传图片

multipart/form-data通过端点使用编码将图像上传到 ComfyUI"/upload/image"。此函数从指定路径打开图像文件并将其上传到 ComfyUI。作为参数，我们指定input_path，即我们要上传的图像文件的路径，name，即上传图像的文件名和server_address运行 ComfyUI 的文件名。作为可选参数，我们可以定义是否要将图像上传到不同的目录（“输出”、“临时”），以及是否要覆盖该路径和文件名的现有图像。图像上传为'image/png'。

Pythondef upload_image(input_path, name, server_address, image_type="input", overwrite=False):

with open(input_path, 'rb') as file:

multipart_data = MultipartEncoder(

fields= {

'image': (name, file, 'image/png'),

'type': image_type,

'overwrite': str(overwrite).lower()

}

)

data = multipart_data

headers = { 'Content-Type': multipart_data.content_type }

request = urllib.request.Request("http://{}/upload/image".format(server_address), data=data, headers=headers)

with urllib.request.urlopen(request) as response:

return response.read()

API 工作流程

要通过 API 使用 ComfyUI 工作流，请使用“保存（API 格式）”保存工作流。如果没有此按钮，则必须通过单击右上角的“设置”按钮（齿轮图标）来启用“开发模式选项”。选中设置选项“启用开发模式选项”。之后，应该会出现“保存（API 格式）”按钮。

图像生成的基本提示

第一步，我们必须加载工作流 JSON。在我的例子中，我在 API 的根级别有一个文件夹，用于保存我的工作流。

Pythondef load_workflow(workflow_path):

try:

with open(workflow_path, 'r') as file:

workflow = json.load(file)

return json.dumps(workflow)

except FileNotFoundError:

print(f"The file {workflow_path} was not found.")

return None

except json.JSONDecodeError:

print(f"The file {workflow_path} contains invalid JSON.")

return None操作工作流程

目标是反复重复使用已定义的工作流程，同时还能够更改正向和负向提示以生成不同的图像。该技术还可用于根据您的需要更改工作流程的其他方面，例如“cfg”或“步骤”。您需要在工作流程中找到相关节点并调整值。一个重要的方面是每次都必须重新生成种子，因为如果种子没有改变，ComfyUI 将不会生成新的图像。

Pythondef prompt_to_image(workflow, positve_prompt, negative_prompt='', save_previews=False):

prompt = json.loads(workflow)

id_to_class_type = {id: details['class_type'] for id, details in prompt.items()}

k_sampler = [key for key, value in id_to_class_type.items() if value == 'KSampler'][0]

prompt.get(k_sampler)['inputs']['seed'] = random.randint(10**14, 10**15 - 1)

postive_input_id = prompt.get(k_sampler)['inputs']['positive'][0]

prompt.get(postive_input_id)['inputs']['text'] = positve_prompt

if negative_prompt != '':

negative_input_id = prompt.get(k_sampler)['inputs']['negative'][0]

prompt.get(negative_input_id)['inputs']['text'] = negative_prompt

generate_image_by_prompt(prompt, './output/', save_previews)调用 API

Pythondef generate_image_by_prompt(prompt, output_path, save_previews=False):

try:

ws, server_address, client_id = open_websocket_connection()

prompt_id = queue_prompt(prompt, client_id, server_address)['prompt_id']

track_progress(prompt, ws, prompt_id)

images = get_images(prompt_id, server_address, save_previews)

save_image(images, output_path, save_previews)

finally:

ws.close()

与 ComfyUI 建立 WebSocket 连接

通过 API 调用对提示进行排队

使用 WebSocket 连接跟踪提示的进度

获取提示所生成的图像

将图像保存在本地

追踪进度

WebSocket 连接使我们能够跟踪工作流的进度。我们想知道我们当前处于哪个节点以及工作流中有多少个节点，以便了解最新进展。我们还想详细跟踪 K-Sampler 的进度，因为这是需要花费大部分时间才能完成的节点，我们想知道 K-Sampler 当前处于哪个步骤以及通常有多少个步骤。

Pythondef track_progress(prompt, ws, prompt_id):

node_ids = list(prompt.keys())

finished_nodes = []

while True:

out = ws.recv()

if isinstance(out, str):

message = json.loads(out)

if message['type'] == 'progress':

data = message['data']

current_step = data['value']

print('In K-Sampler -> Step: ', current_step, ' of: ', data['max'])

if message['type'] == 'execution_cached':

data = message['data']

for itm in data['nodes']:

if itm not in finished_nodes:

finished_nodes.append(itm)

print('Progess: ', len(finished_nodes), '/', len(node_ids), ' Tasks done')

if message['type'] == 'executing':

data = message['data']

if data['node'] not in finished_nodes:

finished_nodes.append(data['node'])

print('Progess: ', len(finished_nodes), '/', len(node_ids), ' Tasks done')

if data['node'] is None and data['prompt_id'] == prompt_id:

break #Execution is done

else:

continue

return处理图像

我们首先获取给定提示 ID 的历史记录，然后获取并保存图像。之后，我们在历史记录端点提供的输出中搜索相关图像。来自“保存图像”节点的图像列在“输出”中。预览图像列为“临时”。我们可以指定是否也要保存预览。在最后一步中，我们将图像本地保存在指定的输出目录中。

Pythondef get_images(prompt_id, server_address, allow_preview = False):

output_images = []

history = get_history(prompt_id, server_address)[prompt_id]

for node_id in history['outputs']:

node_output = history['outputs'][node_id]

output_data = {}

if 'images' in node_output:

for image in node_output['images']:

if allow_preview and image['type'] == 'temp':

preview_data = get_image(image['filename'], image['subfolder'], image['type'], server_address)

output_data['image_data'] = preview_data

if image['type'] == 'output':

image_data = get_image(image['filename'], image['subfolder'], image['type'], server_address)

output_data['image_data'] = image_data

output_data['file_name'] = image['filename']

output_data['type'] = image['type']

output_images.append(output_data)

return output_images

def save_image(images, output_path, save_previews):

for itm in images:

directory = os.path.join(output_path, 'temp/') if itm['type'] == 'temp' and save_previews else output_path

os.makedirs(directory, exist_ok=True)

try:

image = Image.open(io.BytesIO(itm['image_data']))

image.save(os.path.join(directory, itm['file_name']))

except Exception as e:

print(f"Failed to save image {itm['file_name']}: {e}") 例子

本示例使用的工作流程：基本提示到图像工作流程

型号：sdXL_v10VAEFix.safetensors

采样器名称：dpmpp_3m_sde

调度员：karras

步骤：22

配置文件：6.2

API 将正向提示改为：身穿红色连衣裙的女子站在人群中央，背景是摩天大楼，电影感十足，霓虹色彩，逼真的外观

基本图像到图像生成

首先，您必须在 ComfyUI 中构建一个基本的图像到图像工作流程，使用加载图像和 VEA 编码，如下所示：

操作工作流程

在大多数情况下，我们以与提示到图像工作流程相同的方式操作工作流程，但我们还希望能够更改我们使用的输入图像。因此，我们需要在图像加载器中修改图像的名称，并在排队提示之前上传图像。

Pythondef prompt_image_to_image(workflow, input_path, positve_prompt, negative_prompt='', save_previews=False):

prompt = json.loads(workflow)

id_to_class_type = {id: details['class_type'] for id, details in prompt.items()}

k_sampler = [key for key, value in id_to_class_type.items() if value == 'KSampler'][0]

prompt.get(k_sampler)['inputs']['seed'] = random.randint(10**14, 10**15 - 1)

postive_input_id = prompt.get(k_sampler)['inputs']['positive'][0]

prompt.get(postive_input_id)['inputs']['text'] = positve_prompt

if negative_prompt != '':

negative_input_id = prompt.get(k_sampler)['inputs']['negative'][0]

prompt.get(negative_input_id)['inputs']['text'] = negative_prompt

image_loader = [key for key, value in id_to_class_type.items() if value == 'LoadImage'][0]

filename = input_path.split('/')[-1]

prompt.get(image_loader)['inputs']['image'] = filename

generate_image_by_prompt_and_image(prompt, './output/', input_path, filename, save_previews)调用 API

Pythondef generate_image_by_prompt_and_image(prompt, output_path, input_path, filename, save_previews=False):

try:

ws, server_address, client_id = open_websocket_connection()

upload_image(input_path, filename, server_address)

prompt_id = queue_prompt(prompt, client_id, server_address)['prompt_id']

track_progress(prompt, ws, prompt_id)

images = get_images(prompt_id, server_address, save_previews)

save_image(images, output_path, save_previews)

finally:

ws.close()

与 ComfyUI 建立 WebSocket 连接

通过 API 调用对提示进行排队

使用 WebSocket 连接跟踪提示的进度

获取提示所生成的图像

将图像保存在本地

例子

让我们对第一个示例的图像进行一些细微的更改。更改衣服的颜色，但保持图像的整体构图不变。

本示例使用的工作流程：基本图像到图像工作流程

采样器名称：dpmpp_3m_sde

调度员：karras

步骤：22

cfg：8

扩散：0.6

API 将正面提示改为：身穿白色连衣裙的女子站在人群中央，背景是摩天大楼，电影、单色、黑暗、反乌托邦

结果

如您所见，托管您的 ComfyUI 工作流并将其置于 API 后面是一项简单的任务。细节决定成败，例如当您想要使用自定义节点和模型时，但一切都是可行的。