Nunchaku FLUX.1-dev代码实例：Python调用ComfyUI API自动化生成

想用代码批量生成高质量图片，但每次都要手动点网页界面，太麻烦了？今天，我们就来解决这个问题。

Nunchaku FLUX.1-dev是一个强大的文生图模型，而ComfyUI则是它的一个流行图形化操作界面。但真正的效率，来自于自动化。本文将带你一步步实现用Python脚本，直接调用ComfyUI的API，实现Nunchaku FLUX.1-dev模型的自动化图片生成。从环境准备到代码实战，让你彻底告别手动操作，解放双手。

1. 环境准备与快速部署

在开始编写自动化脚本之前，我们需要确保ComfyUI服务已经正确运行起来。这是所有后续操作的基础。

1.1 基础环境检查

首先，确认你的系统已经满足以下基本要求：

• 操作系统：Linux (如Ubuntu 22.04) 或 Windows (WSL2环境更佳)。
• Python：版本3.10或更高。可以通过 python --version 命令检查。
• 显卡：支持CUDA的NVIDIA显卡。对于FLUX.1-dev模型，建议显存不低于24GB。如果显存较小，可以选择FP8或INT4量化版本的模型。
• Git：用于克隆代码仓库。

1.2 一键启动ComfyUI服务

假设你已经按照常规教程完成了ComfyUI以及Nunchaku插件的安装，模型文件也已就位。启动ComfyUI服务是API调用的前提。

进入你的ComfyUI项目根目录，使用以下命令启动服务：

# 进入ComfyUI目录
cd /path/to/your/ComfyUI

# 启动ComfyUI服务，默认监听127.0.0.1:8188
python main.py

启动成功后，你会在终端看到类似 Running on local URL: http://127.0.0.1:8188 的输出。此时，ComfyUI的Web界面和API服务都已经在后台运行了。

为了后续自动化脚本的稳定性，我们更推荐使用 --listen 参数启动，让服务监听所有网络接口，方便本地脚本调用：

python main.py --listen

这样，API地址就是 http://你的机器IP:8188，对于本机脚本，使用 http://127.0.0.1:8188 或 http://localhost:8188 即可。

2. 理解ComfyUI API与工作流

在写代码之前，我们需要搞清楚两件事：ComfyUI的API怎么用，以及我们要运行的“工作流”是什么。

2.1 ComfyUI API基础

ComfyUI提供了一套基于HTTP的API，核心是以下几个端点：

1. /prompt：这是最重要的API，用于提交一个工作流定义并执行它，生成图片。
2. /history 和 /view：用于查询任务执行历史和获取生成的图片。
3. /upload/image：用于上传图片（在图生图等场景有用）。

我们的自动化脚本主要与 /prompt 接口交互。你需要向它发送一个JSON数据，这个JSON完整描述了一个“工作流”。

2.2 解密工作流JSON文件

在ComfyUI的Web界面中，当你连接好各个节点并保存时，会生成一个 .json 文件。这个文件就是工作流的“配方”，它定义了：

• 使用了哪些节点（如加载模型、输入提示词、采样、保存图片等）。
• 节点之间的连接关系。
• 每个节点的具体参数（如模型名称、提示词内容、采样步数等）。

我们的Python脚本，本质上就是构造一个类似的JSON对象，然后通过API发送给ComfyUI服务去执行。

你可以通过Web界面加载之前配置好的 nunchaku-flux.1-dev.json 工作流，然后点击菜单栏的“保存”（Save）按钮，将其下载到本地，用文本编辑器打开查看其结构。这将是我们编写脚本的蓝图。

3. Python自动化脚本实战

理论说完了，现在开始写代码。我们将创建一个完整的Python脚本，实现从提示词到生成图片的全流程自动化。

3.1 安装必要的Python库

首先，确保安装了 requests 库，用于发送HTTP请求。

pip install requests

3.2 构建完整的自动化脚本

下面是一个功能完整的示例脚本 generate_with_api.py。我们将逐部分讲解。

import json
import requests
import time
import sys
import os

class ComfyUIAPIClient:
    """ComfyUI API 客户端类，封装核心操作"""
    
    def __init__(self, server_address="127.0.0.1", port=8188):
        """
        初始化客户端
        :param server_address: ComfyUI服务器地址
        :param port: 端口号，默认为8188
        """
        self.server_address = server_address
        self.port = port
        self.base_url = f"http://{server_address}:{port}"
        self.client_id = "auto_gen_script"  # 客户端标识，可自定义
        
    def get_workflow_template(self):
        """
        返回一个基础的Nunchaku FLUX.1-dev工作流模板。
        注意：这是一个简化示例，你需要根据你实际保存的 `nunchaku-flux.1-dev.json` 文件内容来填充。
        关键点：`class_type`, `inputs` 必须与你ComfyUI中安装的节点对应。
        """
        # 这是一个极其简化的结构，实际结构非常复杂。
        # 强烈建议：通过Web界面配置好工作流后，使用 `api/workflow` 接口获取或直接解析保存的json文件。
        workflow_template = {
            "3": {
                "class_type": "NunchakuLoader",
                "inputs": {
                    "ckpt_name": "svdq-int4_r32-flux.1-dev.safetensors", # 你的模型文件名
                    "vae_name": "ae.safetensors",
                    "clip_name1": "clip_l.safetensors",
                    "clip_name2": "t5xxl_fp16.safetensors"
                }
            },
            "5": {
                "class_type": "CLIPTextEncode",
                "inputs": {
                    "text": "masterpiece, best quality, 1girl, beautiful detailed eyes", # 正面提示词
                    "clip": ["3", 1] # 连接到NunchakuLoader节点的clip输出
                }
            },
            "6": {
                "class_type": "CLIPTextEncode",
                "inputs": {
                    "text": "worst quality, low quality, blurry", # 负面提示词
                    "clip": ["3", 1]
                }
            },
            "10": {
                "class_type": "EmptyLatentImage",
                "inputs": {
                    "width": 1024,
                    "height": 1024,
                    "batch_size": 1
                }
            },
            "11": {
                "class_type": "NunchakuSampler", # 采样器节点
                "inputs": {
                    "model": ["3", 0], # 连接到模型
                    "positive": ["5", 0],
                    "negative": ["6", 0],
                    "latent_image": ["10", 0],
                    "steps": 20,
                    "cfg": 7.0,
                    "sampler_name": "euler",
                    "scheduler": "simple"
                }
            },
            "12": {
                "class_type": "VAEDecode",
                "inputs": {
                    "samples": ["11", 0],
                    "vae": ["3", 2]
                }
            },
            "13": {
                "class_type": "SaveImage",
                "inputs": {
                    "filename_prefix": "ComfyUI",
                    "images": ["12", 0]
                }
            }
        }
        return workflow_template
    
    def _load_workflow_from_file(self, filepath):
        """从本地保存的JSON文件加载工作流定义"""
        try:
            with open(filepath, 'r', encoding='utf-8') as f:
                data = json.load(f)
            # ComfyUI的工作流数据通常在 `workflow` 键或直接是顶级对象
            if 'workflow' in data:
                return data['workflow']
            else:
                # 假设文件内容就是节点定义
                return data
        except FileNotFoundError:
            print(f"错误：工作流文件 {filepath} 未找到。")
            print("请先在ComfyUI Web界面配置并保存工作流。")
            sys.exit(1)
        except json.JSONDecodeError:
            print(f"错误：文件 {filepath} 不是有效的JSON格式。")
            sys.exit(1)
    
    def queue_prompt(self, prompt_workflow):
        """
        向ComfyUI服务器提交工作流执行请求
        :param prompt_workflow: 工作流定义字典
        :return: 包含任务ID的响应字典
        """
        api_url = f"{self.base_url}/prompt"
        payload = {"prompt": prompt_workflow, "client_id": self.client_id}
        
        try:
            response = requests.post(api_url, json=payload)
            response.raise_for_status()  # 检查HTTP错误
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API请求失败: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"服务器响应: {e.response.text}")
            sys.exit(1)
    
    def wait_for_completion(self, prompt_id, poll_interval=1):
        """
        轮询任务状态，直到完成
        :param prompt_id: 任务ID
        :param poll_interval: 轮询间隔(秒)
        :return: 任务历史信息
        """
        print(f"任务 {prompt_id} 已提交，等待生成...")
        history_url = f"{self.base_url}/history"
        
        while True:
            try:
                response = requests.get(history_url)
                history = response.json()
                # history 结构: { prompt_id: { "status": ..., "outputs": ... } }
                if prompt_id in history:
                    status_info = history[prompt_id]
                    # 检查状态，这里简化处理，实际可根据status字段判断
                    if status_info.get('outputs'):
                        print("图片生成完成！")
                        return status_info
                time.sleep(poll_interval)
            except requests.exceptions.RequestException as e:
                print(f"轮询历史记录失败: {e}")
                time.sleep(poll_interval)
    
    def get_generated_images(self, history_info, output_dir="./output"):
        """
        从历史信息中提取生成的图片并保存到本地
        :param history_info: wait_for_completion返回的历史信息
        :param output_dir: 本地保存目录
        """
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
            
        outputs = history_info.get('outputs', {})
        image_count = 0
        
        for node_id, node_output in outputs.items():
            if 'images' in node_output:
                for img_info in node_output['images']:
                    # 图片信息包含文件名和子文件夹
                    filename = img_info.get('filename')
                    subfolder = img_info.get('subfolder', '')
                    # ComfyUI默认图片在 `output` 目录下
                    server_image_path = os.path.join('output', subfolder, filename).replace('\\', '/')
                    
                    # 构建下载URL
                    view_url = f"{self.base_url}/view?filename={filename}&subfolder={subfolder}&type=output"
                    
                    try:
                        img_response = requests.get(view_url, stream=True)
                        if img_response.status_code == 200:
                            # 生成本地文件名
                            local_filename = f"generated_{int(time.time())}_{image_count}.png"
                            local_path = os.path.join(output_dir, local_filename)
                            
                            with open(local_path, 'wb') as f:
                                for chunk in img_response.iter_content(chunk_size=8192):
                                    f.write(chunk)
                            print(f"图片已保存至: {local_path}")
                            image_count += 1
                        else:
                            print(f"下载图片失败，状态码: {img_response.status_code}")
                    except Exception as e:
                        print(f"下载图片时出错: {e}")
        
        if image_count == 0:
            print("未在输出中找到图片信息。")
            print("提示：请确保工作流中包含 `SaveImage` 节点。")
    
    def generate_image(self, positive_prompt, negative_prompt="", workflow_file=None, **kwargs):
        """
        生成图片的主函数
        :param positive_prompt: 正面提示词
        :param negative_prompt: 负面提示词
        :param workflow_file: 本地工作流JSON文件路径（优先使用）
        :param kwargs: 其他可覆盖的参数，如 steps, width, height
        """
        # 1. 加载工作流定义
        if workflow_file and os.path.exists(workflow_file):
            print(f"从文件加载工作流: {workflow_file}")
            workflow = self._load_workflow_from_file(workflow_file)
        else:
            print("使用内置工作流模板。")
            workflow = self.get_workflow_template()
            # 注意：内置模板是示例，节点ID可能与你的实际工作流不匹配。
            # 直接修改模板字典来更新参数可能很复杂，更推荐使用文件方式。
        
        # 2. 动态修改工作流参数（高级操作，需要对工作流结构非常熟悉）
        # 这里演示一个简单思路：如果你知道对应节点的ID，可以直接修改。
        # 例如，假设正面提示词节点ID是"5"：
        # if "5" in workflow and "inputs" in workflow["5"]:
        #     workflow["5"]["inputs"]["text"] = positive_prompt
        # 更稳健的方式是在Web界面配置好模板，然后通过文件加载。
        
        print(f"提示词: {positive_prompt}")
        
        # 3. 提交任务
        response = self.queue_prompt(workflow)
        prompt_id = response.get('prompt_id')
        if not prompt_id:
            print("提交任务失败，未获取到任务ID。")
            print(f"服务器响应: {response}")
            return
        
        # 4. 等待任务完成
        history_info = self.wait_for_completion(prompt_id)
        
        # 5. 获取并保存图片
        self.get_generated_images(history_info)

# 使用示例
if __name__ == "__main__":
    # 初始化客户端，如果ComfyUI运行在其他机器或端口，请修改
    client = ComfyUIAPIClient(server_address="127.0.0.1", port=8188)
    
    # 你的创意提示词
    my_prompt = "A beautiful landscape with mountains and lakes, ultra HD, realistic, 8K, sunset, cinematic lighting"
    my_negative_prompt = "blurry, ugly, deformed, low quality"
    
    # 方法1：使用本地保存的工作流文件（推荐！）
    # 首先，在ComfyUI Web界面配置好Nunchaku FLUX.1-dev的所有节点并连接，然后保存为 `my_flux_workflow.json`
    workflow_json_path = "./my_flux_workflow.json" # 替换为你的实际文件路径
    
    if os.path.exists(workflow_json_path):
        client.generate_image(
            positive_prompt=my_prompt,
            negative_prompt=my_negative_prompt,
            workflow_file=workflow_json_path
            # 还可以传递其他参数，如 steps=25, width=1024, height=768
        )
    else:
        print(f"工作流文件 {workflow_json_path} 不存在。")
        print("请先在ComfyUI界面配置并保存工作流，或使用内置模板（可能需要调整）。")
        # 方法2：使用内置模板（需要你根据实际情况调整模板中的节点ID和类型）
        # client.generate_image(positive_prompt=my_prompt, negative_prompt=my_negative_prompt)

3.3 脚本使用步骤详解

这个脚本看起来有点长，但逻辑很清晰。我们分解一下如何使用：

1. 准备工作流文件：这是最关键的一步。在ComfyUI的Web界面中，手动拖拽节点，配置好一个完整的Nunchaku FLUX.1-dev文生图流程（包括加载模型、编码提示词、采样、保存图片等）。然后点击菜单的“保存”（Save）按钮，将工作流保存为JSON文件（例如 my_flux_workflow.json）。这个文件包含了所有节点的精确ID和连接关系。

2. 修改脚本配置：将上述脚本中的 workflow_json_path = "./my_flux_workflow.json" 改为你实际保存的文件路径。

3. 修改提示词：在 __main__ 部分，将 my_prompt 和 my_negative_prompt 替换成你想要生成图片的描述。

4. 运行脚本：确保ComfyUI服务正在运行（python main.py --listen），然后在终端执行：

   python generate_with_api.py
   

5. 查看结果：脚本会自动提交任务、等待生成，并将最终图片保存到 ./output 目录（默认）下。

4. 进阶技巧与批量处理

掌握了单次生成后，我们可以玩点更高级的，比如批量生成、参数随机化，让脚本真正发挥威力。

4.1 实现批量图片生成

我们可以轻松地用一个循环，来生成一系列不同主题的图片。

def batch_generate_images(client, prompt_list, output_base_dir="./batch_output"):
    """
    批量生成图片
    :param client: ComfyUIAPIClient实例
    :param prompt_list: 提示词列表，每个元素可以是一个字符串或字典（包含正负面提示词）
    :param output_base_dir: 输出根目录
    """
    import os
    if not os.path.exists(output_base_dir):
        os.makedirs(output_base_dir)
    
    for i, prompt_item in enumerate(prompt_list):
        if isinstance(prompt_item, dict):
            pos_prompt = prompt_item.get('positive', '')
            neg_prompt = prompt_item.get('negative', '')
            sub_dir = prompt_item.get('subfolder', f'batch_{i}')
        else:
            pos_prompt = prompt_item
            neg_prompt = ""
            sub_dir = f'batch_{i}'
        
        current_output_dir = os.path.join(output_base_dir, sub_dir)
        print(f"\n--- 开始生成第 {i+1}/{len(prompt_list)} 张: {pos_prompt[:50]}... ---")
        
        # 这里需要修改client的get_generated_images方法，使其能接受自定义输出目录
        # 为了简化，我们这里演示逻辑。实际中可以修改类方法或传递参数。
        # 假设我们有一个新方法 `generate_image_to_folder`
        # client.generate_image_to_folder(...)
        print(f"提示词: {pos_prompt}")
        # ... 调用生成逻辑，将图片保存到 current_output_dir ...

# 使用示例
if __name__ == "__main__":
    client = ComfyUIAPIClient()
    
    my_prompts = [
        {"positive": "A cyberpunk cityscape at night, neon lights, rain, futuristic vehicles", "negative": "daytime, sunny", "subfolder": "cyberpunk"},
        {"positive": "A serene Japanese garden with koi pond, cherry blossoms, traditional wooden bridge", "negative": "modern, people, buildings", "subfolder": "garden"},
        "A majestic dragon soaring above ancient mountains, fantasy art, detailed scales", # 简写，只提供正面词
    ]
    
    # 调用批量生成函数（需要你根据上面的提示实现细节）
    # batch_generate_images(client, my_prompts)

4.2 动态修改工作流参数

有时我们不想为每个小参数（如尺寸、步数）都保存一个单独的JSON文件。我们可以加载基础工作流后，用代码动态修改特定节点的参数。

def modify_workflow_params(workflow_dict, modifications):
    """
    动态修改工作流字典中的参数
    :param workflow_dict: 从JSON加载的工作流字典
    :param modifications: 修改规则列表，每条规则格式为：
                          {"node_id": "节点ID", "input_key": "参数名", "value": 新值}
    :return: 修改后的工作流字典
    """
    for mod in modifications:
        node_id = mod.get("node_id")
        input_key = mod.get("input_key")
        value = mod.get("value")
        
        if node_id in workflow_dict and "inputs" in workflow_dict[node_id]:
            workflow_dict[node_id]["inputs"][input_key] = value
            print(f"已修改节点 {node_id} 的参数 `{input_key}` 为: {value}")
        else:
            print(f"警告：未找到节点 {node_id} 或其inputs字段。")
    return workflow_dict

# 使用示例：修改采样步数和图片尺寸
if __name__ == "__main__":
    client = ComfyUIAPIClient()
    workflow = client._load_workflow_from_file("./my_flux_workflow.json")
    
    # 假设我们知道“EmptyLatentImage”节点ID是“10”，用于控制尺寸
    # 假设我们知道“NunchakuSampler”节点ID是“11”，用于控制步数
    # 这些ID需要你从自己的工作流JSON文件中确认
    modifications = [
        {"node_id": "10", "input_key": "width", "value": 768},
        {"node_id": "10", "input_key": "height", "value": 512},
        {"node_id": "11", "input_key": "steps", "value": 30},
        {"node_id": "5", "input_key": "text", "value": "A new prompt here"}, # 修改正面提示词
    ]
    
    updated_workflow = modify_workflow_params(workflow, modifications)
    # 然后使用 updated_workflow 提交任务
    # response = client.queue_prompt(updated_workflow)

关键点：要使用这个功能，你必须知道你工作流中关键节点的ID。这些ID在你保存的JSON文件中可以找到。通常，你可以通过节点的“标题”在JSON中搜索对应的 class_type 来定位。

5. 常见问题与调试技巧

自动化过程中难免会遇到问题，这里总结几个常见坑点和解决方法。

5.1 API连接失败

• 症状：脚本报错 requests.exceptions.ConnectionError。
• 检查：
  1. ComfyUI服务启动了吗？运行 python main.py --listen 了吗？
  2. 服务器地址和端口对吗？默认是 http://127.0.0.1:8188。
  3. 防火墙是否阻止了连接？

5.2 工作流执行失败

• 症状：任务提交后，在ComfyUI Web界面的“队列”中看到错误，或者脚本一直等待不到完成。
• 检查：
  1. 工作流JSON文件是否正确：最可能的原因。确保你的JSON文件是从当前运行的ComfyUI版本中保存的，并且所有用到的自定义节点（如 NunchakuLoader）都已正确安装。
  2. 模型路径是否正确：在 NunchakuLoader 节点中，ckpt_name 等参数必须与 ComfyUI/models/ 目录下的实际文件名完全一致。
  3. 节点ID是否对应：如果你用代码动态修改参数，确保你引用的节点ID（如 "10", "11"）在JSON文件中真实存在。
• 调试方法：
  1. 先在Web界面手动加载并运行这个JSON工作流，确保它能成功生成图片。
  2. 使用简单的提示词和参数。
  3. 查看ComfyUI服务启动终端的输出日志，里面通常有详细的错误信息。

5.3 生成的图片找不到

• 症状：脚本显示任务完成，但 get_generated_images 函数没有找到图片。
• 检查：
  1. 工作流中是否包含了 SaveImage 节点？这是将图片数据持久化到 output 文件夹的关键。
  2. SaveImage 节点的输出是否连接正确？它应该连接到 VAEDecode 节点的输出。
  3. 可以手动访问 http://127.0.0.1:8188/history 查看最近任务的历史记录，确认 outputs 字段里是否有图片信息。

5.4 提升脚本健壮性

给你的脚本加上更完善的错误处理和日志记录。

import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RobustComfyUIClient(ComfyUIAPIClient):
    def queue_prompt(self, prompt_workflow):
        api_url = f"{self.base_url}/prompt"
        payload = {"prompt": prompt_workflow, "client_id": self.client_id}
        
        try:
            logger.info(f"正在向 {api_url} 提交工作流...")
            response = requests.post(api_url, json=payload, timeout=30) # 增加超时
            response.raise_for_status()
            result = response.json()
            logger.info(f"任务提交成功，ID: {result.get('prompt_id')}")
            return result
        except requests.exceptions.Timeout:
            logger.error("请求超时，请检查ComfyUI服务状态。")
            raise
        except requests.exceptions.RequestException as e:
            logger.error(f"网络请求异常: {e}")
            raise
        except json.JSONDecodeError as e:
            logger.error(f"解析服务器响应失败: {e}")
            logger.error(f"原始响应: {response.text}")
            raise

6. 总结

通过本文的步骤，你已经掌握了使用Python调用ComfyUI API驱动Nunchaku FLUX.1-dev模型进行自动化文生图的核心方法。我们来回顾一下关键点：

1. 核心原理：自动化就是让Python脚本模拟你在Web界面的操作，通过向ComfyUI的 /prompt API发送一个描述完整工作流的JSON数据来实现。
2. 成功关键：正确的工作流JSON文件。务必在Web界面仔细配置并保存一个能成功运行的工作流作为模板。
3. 进阶之路：基于模板，你可以用代码动态修改提示词、尺寸、步数等参数，甚至实现循环批量生成，将AI绘画彻底集成到你的自动化流水线中。
4. 调试心态：遇到问题别慌，从检查服务状态、工作流文件、模型路径和终端日志这几个方面入手，大部分问题都能解决。

从此，你可以将创意写作、参数探索、批量出图等重复性劳动交给脚本，自己则专注于构思更精彩的提示词和筛选更优秀的作品。自动化不是为了取代创造力，而是为了让创造力更高效地绽放。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。