Stable Diffusion | ComfyUI API 工作流自动优化

ComfyUI 可直接保存生图工作流为 API 格式，但该 API 格式文本行数较多且节点顺序与逻辑执行顺序不一致，不利于编写或修改 API 的调用代码。

在上一篇文章 Stable Cascade | ComfyUI API 工作流格式优化 中介绍了 API 工作流的结构以及手工优化 API 工作流的方法。

本文主要介绍自动优化 ComfyUI API 工作流的 Python 代码。

1. 拓扑排序函数

ComfyUI API 工作流可以看作一张有向图，因此可使用 collections 库的 deque 模块对工作流节点进行拓扑排序。拓扑排序函数如下：

from collections import deque

def topological_sort(graph):
    in_degree = {v: 0 for v in graph}
    for node in graph:
        for neighbor in graph[node]:
            in_degree[neighbor] += 1
    queue = deque([node for node in in_degree if in_degree[node] == 0])
    result = []
    while queue:
        node = queue.popleft()
        result.append(node)
        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    if len(result) != len(graph):
        return None
    return result

拓扑排序函数参考自文章：Python深度学习-有向图合并、排序、最长路径计算 。

2. 工作流转有向图

import json

# 将工作流转为有向图
link_list = {}
for node in workflow:
    node_link = []
    for input in workflow[node]["inputs"]:
        if isinstance(workflow[node]["inputs"][input], list):
            node_link.append(workflow[node]["inputs"][input][0])
    link_list[node] = node_link

print(link_list)

示例工作流内容（workflow_api.json 文件内容）：

{
  "3": {
    "inputs": {
      "seed": 314307448448003,
      "steps": 20,
      "cfg": 4,
      "sampler_name": "euler_ancestral",
      "scheduler": "simple",
      "denoise": 1,
      "model": [
        "41",
        0
      ],
      "positive": [
        "6",
        0
      ],
      "negative": [
        "7",
        0
      ],
      "latent_image": [
        "34",
        0
      ]
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "KSampler"
    }
  },
  "6": {
    "inputs": {
      "text": "evening sunset scenery blue sky nature, glass bottle with a fizzy ice cold freezing rainbow liquid in it",
      "clip": [
        "41",
        1
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Positive Prompt)"
    }
  },
  "7": {
    "inputs": {
      "text": "text, watermark",
      "clip": [
        "41",
        1
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Negative Prompt)"
    }
  },
  "8": {
    "inputs": {
      "samples": [
        "33",
        0
      ],
      "vae": [
        "42",
        2
      ]
    },
    "class_type": "VAEDecode",
    "_meta": {
      "title": "VAE Decode"
    }
  },
  "9": {
    "inputs": {
      "filename_prefix": "ComfyUI",
      "images": [
        "8",
        0
      ]
    },
    "class_type": "SaveImage",
    "_meta": {
      "title": "Save Image"
    }
  },
  "33": {
    "inputs": {
      "seed": 183495397600639,
      "steps": 10,
      "cfg": 1.1,
      "sampler_name": "euler_ancestral",
      "scheduler": "simple",
      "denoise": 1,
      "model": [
        "42",
        0
      ],
      "positive": [
        "36",
        0
      ],
      "negative": [
        "7",
        0
      ],
      "latent_image": [
        "34",
        1
      ]
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "KSampler"
    }
  },
  "34": {
    "inputs": {
      "width": 1024,
      "height": 1024,
      "compression": 42,
      "batch_size": 1
    },
    "class_type": "StableCascade_EmptyLatentImage",
    "_meta": {
      "title": "StableCascade_EmptyLatentImage"
    }
  },
  "36": {
    "inputs": {
      "conditioning": [
        "6",
        0
      ],
      "stage_c": [
        "3",
        0
      ]
    },
    "class_type": "StableCascade_StageB_Conditioning",
    "_meta": {
      "title": "StableCascade_StageB_Conditioning"
    }
  },
  "41": {
    "inputs": {
      "ckpt_name": "stable_cascade_stage_c.safetensors"
    },
    "class_type": "CheckpointLoaderSimple",
    "_meta": {
      "title": "Load Checkpoint"
    }
  },
  "42": {
    "inputs": {
      "ckpt_name": "stable_cascade_stage_b.safetensors"
    },
    "class_type": "CheckpointLoaderSimple",
    "_meta": {
      "title": "Load Checkpoint"
    }
  }
}

工作流转有向图结果：

{'3': ['41', '6', '7', '34'], '6': ['41'], '7': ['41'], '8': ['33', '42'], '9': ['8'], '33': ['42', '36', '7', '34'], '34': [], '36': ['6', '3'], '41': [], '42': []}

以节点3为例，从有向图中可以看到，节点3同时指向节点41，6，7，34。需注意此时节点指向是朝向输入节点的，而工作流应该是输入指向输出方向，所以在拓扑排序后节点序列需要反向排序一次。

3. 工作流拓扑排序

调用拓扑排序函数对工作流有向图进行拓扑排序，并将排序后的节点序列反向排序。

# 工作流（有向图）拓扑排序
order_list = topological_sort(link_list)[::-1]

print(f'原始工作流拓扑排序结果：{order_list}')

排序结果：

原始工作流拓扑排序结果：['41', '34', '7', '6', '3', '36', '42', '33', '8', '9']

4. 工作流节点重排序

重排序流程：

# 工作流节点临时编号（为了避免覆盖已有编号，从最大节点总数 * 10 + 1开始编号，同时可避免下一步排序时出现1，10，11，..., 2的情况）
max_nodes = 100
new_node_id = max_nodes * 10 + 1
workflow_string = json.dumps(workflow)
for node in order_list:
    workflow_string = workflow_string.replace(f'"{node}"', f'"{new_node_id}"')
    new_node_id += 1
workflow = json.loads(workflow_string)

# 工作流排序
sorted_data = OrderedDict(sorted(workflow.items()))
sorted_data = json.dumps(sorted_data)
workflow = json.loads(sorted_data)

# 工作流节点最终编号（从1开始编号）
new_node_id = 1
workflow_string = json.dumps(workflow)
for node in workflow:
    workflow_string = workflow_string.replace(f'"{node}"', f'"{new_node_id}"')
    new_node_id += 1
workflow = json.loads(workflow_string)

print(workflow)

输出结果：

{'1': {'inputs': {'ckpt_name': 'stable_cascade_stage_c.safetensors'}, 'class_type': 'CheckpointLoaderSimple', '_meta': {'title': 'Load Checkpoint'}}, '2': {'inputs': {'width': 1024, 'height': 1024, 'compression': 42, 'batch_size': 1}, 'class_type': 'StableCascade_EmptyLatentImage', '_meta': {'title': 'StableCascade_EmptyLatentImage'}}, '3': {'inputs': {'text': 'text, watermark', 'clip': ['1', 1]}, 'class_type': 'CLIPTextEncode', '_meta': {'title': 'CLIP Text Encode (Negative Prompt)'}}, '4': {'inputs': {'text': 'evening sunset scenery blue sky nature, glass bottle with a fizzy ice cold freezing rainbow liquid in it', 'clip': ['1', 1]}, 'class_type': 'CLIPTextEncode', '_meta': {'title': 'CLIP Text Encode (Positive Prompt)'}}, '5': {'inputs': {'seed': 314307448448003, 'steps': 20, 'cfg': 4, 'sampler_name': 'euler_ancestral', 'scheduler': 'simple', 'denoise': 1, 'model': ['1', 0], 'positive': ['4', 0], 'negative': ['3', 0], 'latent_image': ['2', 0]}, 'class_type': 'KSampler', '_meta': {'title': 'KSampler'}}, '6': {'inputs': {'conditioning': ['4', 0], 'stage_c': ['5', 0]}, 'class_type': 'StableCascade_StageB_Conditioning', '_meta': {'title': 'StableCascade_StageB_Conditioning'}}, '7': {'inputs': {'ckpt_name': 'stable_cascade_stage_b.safetensors'}, 'class_type': 'CheckpointLoaderSimple', '_meta': {'title': 'Load Checkpoint'}}, '8': {'inputs': {'seed': 183495397600639, 'steps': 10, 'cfg': 1.1, 'sampler_name': 'euler_ancestral', 'scheduler': 'simple', 'denoise': 1, 'model': ['7', 0], 'positive': ['6', 0], 'negative': ['3', 0], 'latent_image': ['2', 1]}, 'class_type': 'KSampler', '_meta': {'title': 'KSampler'}}, '9': {'inputs': {'samples': ['8', 0], 'vae': ['7', 2]}, 'class_type': 'VAEDecode', '_meta': {'title': 'VAE Decode'}}, '10': {'inputs': {'filename_prefix': 'ComfyUI', 'images': ['9', 0]}, 'class_type': 'SaveImage', '_meta': {'title': 'Save Image'}}}

5. 工作流简化

工作流中 '_meta': {'title': '...'} 字段在调用 API 过程中无实际意义，可以删除以简化工作流。

# 删除子键"_meta": "..."
for node in workflow:
    del workflow[node]["_meta"]

print(workflow)

输出结果：

{'1': {'inputs': {'ckpt_name': 'stable_cascade_stage_c.safetensors'}, 'class_type': 'CheckpointLoaderSimple'}, '2': {'inputs': {'width': 1024, 'height': 1024, 'compression': 42, 'batch_size': 1}, 'class_type': 'StableCascade_EmptyLatentImage'}, '3': {'inputs': {'text': 'text, watermark', 'clip': ['1', 1]}, 'class_type': 'CLIPTextEncode'}, '4': {'inputs': {'text': 'evening sunset scenery blue sky nature, glass bottle with a fizzy ice cold freezing rainbow liquid in it', 'clip': ['1', 1]}, 'class_type': 'CLIPTextEncode'}, '5': {'inputs': {'seed': 314307448448003, 'steps': 20, 'cfg': 4, 'sampler_name': 'euler_ancestral', 'scheduler': 'simple', 'denoise': 1, 'model': ['1', 0], 'positive': ['4', 0], 'negative': ['3', 0], 'latent_image': ['2', 0]}, 'class_type': 'KSampler'}, '6': {'inputs': {'conditioning': ['4', 0], 'stage_c': ['5', 0]}, 'class_type': 'StableCascade_StageB_Conditioning'}, '7': {'inputs': {'ckpt_name': 'stable_cascade_stage_b.safetensors'}, 'class_type': 'CheckpointLoaderSimple'}, '8': {'inputs': {'seed': 183495397600639, 'steps': 10, 'cfg': 1.1, 'sampler_name': 'euler_ancestral', 'scheduler': 'simple', 'denoise': 1, 'model': ['7', 0], 'positive': ['6', 0], 'negative': ['3', 0], 'latent_image': ['2', 1]}, 'class_type': 'KSampler'}, '9': {'inputs': {'samples': ['8', 0], 'vae': ['7', 2]}, 'class_type': 'VAEDecode'}, '10': {'inputs': {'filename_prefix': 'ComfyUI', 'images': ['9', 0]}, 'class_type': 'SaveImage'}}

6. 工作流保存

这里按每节点一行输出至 workflow_api_ordered.json 文件。

# 保存处理好的工作流
with open('workflow_api_ordered.json', 'w') as f:
    f.write('{')
    line = 1
    for key in workflow:
        f.write('\n')
        node_info = str(workflow[key]).replace("'", '"')
        if not line == len(workflow):
            f.writelines('"' + str(key) + '": ' + node_info + ",")
        else:
            f.writelines('"' + str(key) + '": ' + node_info)
        line += 1
    f.write('\n'+'}')

workflow_api_ordered.json 文件内容：

{
"1": {"inputs": {"ckpt_name": "stable_cascade_stage_c.safetensors"}, "class_type": "CheckpointLoaderSimple"},
"2": {"inputs": {"width": 1024, "height": 1024, "compression": 42, "batch_size": 1}, "class_type": "StableCascade_EmptyLatentImage"},
"3": {"inputs": {"text": "text, watermark", "clip": ["1", 1]}, "class_type": "CLIPTextEncode"},
"4": {"inputs": {"text": "evening sunset scenery blue sky nature, glass bottle with a fizzy ice cold freezing rainbow liquid in it", "clip": ["1", 1]}, "class_type": "CLIPTextEncode"},
"5": {"inputs": {"seed": 314307448448003, "steps": 20, "cfg": 4, "sampler_name": "euler_ancestral", "scheduler": "simple", "denoise": 1, "model": ["1", 0], "positive": ["4", 0], "negative": ["3", 0], "latent_image": ["2", 0]}, "class_type": "KSampler"},
"6": {"inputs": {"conditioning": ["4", 0], "stage_c": ["5", 0]}, "class_type": "StableCascade_StageB_Conditioning"},
"7": {"inputs": {"ckpt_name": "stable_cascade_stage_b.safetensors"}, "class_type": "CheckpointLoaderSimple"},
"8": {"inputs": {"seed": 183495397600639, "steps": 10, "cfg": 1.1, "sampler_name": "euler_ancestral", "scheduler": "simple", "denoise": 1, "model": ["7", 0], "positive": ["6", 0], "negative": ["3", 0], "latent_image": ["2", 1]}, "class_type": "KSampler"},
"9": {"inputs": {"samples": ["8", 0], "vae": ["7", 2]}, "class_type": "VAEDecode"},
"10": {"inputs": {"filename_prefix": "ComfyUI", "images": ["9", 0]}, "class_type": "SaveImage"}
}

至此，ComfyUI API 工作流优化结束。

7. 完整Python代码

以下为 ComfyUI API 工作流自动优化 完整代码（仅打印最终工作流内容）。

import json
from collections import deque, OrderedDict

# 有向图拓扑排序函数
def topological_sort(graph):
    in_degree = {v: 0 for v in graph}
    for node in graph:
        for neighbor in graph[node]:
            in_degree[neighbor] += 1
    queue = deque([node for node in in_degree if in_degree[node] == 0])
    result = []
    while queue:
        node = queue.popleft()
        result.append(node)
        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    if len(result) != len(graph):
        return None
    return result

# 加载工作流文件
with open('workflow_api.json', 'r', encoding='utf-8', errors='ignore') as file:
    workflow = json.loads(file.read())

# 将工作流转为有向图
link_list = {}
for node in workflow:
    node_link = []
    for input in workflow[node]["inputs"]:
        if isinstance(workflow[node]["inputs"][input], list):
            node_link.append(workflow[node]["inputs"][input][0])
    link_list[node] = node_link

# 工作流（有向图）拓扑排序
order_list = topological_sort(link_list)[::-1]

# 工作流节点临时编号（为了避免覆盖已有编号，从最大节点总数 * 10 + 1开始编号，同时可避免下一步排序时出现1，10，11，..., 2的情况）
max_nodes = 100
new_node_id = max_nodes * 10 + 1
workflow_string = json.dumps(workflow)
for node in order_list:
    workflow_string = workflow_string.replace(f'"{node}"', f'"{new_node_id}"')
    new_node_id += 1
workflow = json.loads(workflow_string)

# 工作流节点排序
sorted_data = OrderedDict(sorted(workflow.items()))
sorted_data = json.dumps(sorted_data)
workflow = json.loads(sorted_data)

# 工作流节点最终编号（从1开始编号）
new_node_id = 1
workflow_string = json.dumps(workflow)
for node in workflow:
    workflow_string = workflow_string.replace(f'"{node}"', f'"{new_node_id}"')
    new_node_id += 1
workflow = json.loads(workflow_string)

# 删除子键"_meta": "..."
for node in workflow:
    del workflow[node]["_meta"]

# 保存处理好的工作流
with open('workflow_api_ordered.json', 'w') as f:
    f.write('{')
    line = 1
    for key in workflow:
        f.write('\n')
        node_info = str(workflow[key]).replace("'", '"')
        if not line == len(workflow):
            f.writelines('"' + str(key) + '": ' + node_info + ",")
        else:
            f.writelines('"' + str(key) + '": ' + node_info)
        line += 1
    f.write('\n'+'}')

# 打印处理好的工作流
with open('workflow_api_ordered.json', 'r', encoding='utf-8', errors='ignore') as file:
    print("处理好的工作流：")
    for line in file:
        print(line)

输出结果：

处理好的工作流：
{

"1": {"inputs": {"ckpt_name": "stable_cascade_stage_c.safetensors"}, "class_type": "CheckpointLoaderSimple"},

"2": {"inputs": {"width": 1024, "height": 1024, "compression": 42, "batch_size": 1}, "class_type": "StableCascade_EmptyLatentImage"},

"3": {"inputs": {"text": "text, watermark", "clip": ["1", 1]}, "class_type": "CLIPTextEncode"},

"4": {"inputs": {"text": "evening sunset scenery blue sky nature, glass bottle with a fizzy ice cold freezing rainbow liquid in it", "clip": ["1", 1]}, "class_type": "CLIPTextEncode"},

"5": {"inputs": {"seed": 314307448448003, "steps": 20, "cfg": 4, "sampler_name": "euler_ancestral", "scheduler": "simple", "denoise": 1, "model": ["1", 0], "positive": ["4", 0], "negative": ["3", 0], "latent_image": ["2", 0]}, "class_type": "KSampler"},

"6": {"inputs": {"conditioning": ["4", 0], "stage_c": ["5", 0]}, "class_type": "StableCascade_StageB_Conditioning"},

"7": {"inputs": {"ckpt_name": "stable_cascade_stage_b.safetensors"}, "class_type": "CheckpointLoaderSimple"},

"8": {"inputs": {"seed": 183495397600639, "steps": 10, "cfg": 1.1, "sampler_name": "euler_ancestral", "scheduler": "simple", "denoise": 1, "model": ["7", 0], "positive": ["6", 0], "negative": ["3", 0], "latent_image": ["2", 1]}, "class_type": "KSampler"},

"9": {"inputs": {"samples": ["8", 0], "vae": ["7", 2]}, "class_type": "VAEDecode"},

"10": {"inputs": {"filename_prefix": "ComfyUI", "images": ["9", 0]}, "class_type": "SaveImage"}

}