# Python与ComfyUI文生图工作流结合使用方法及实现技巧 ComfyUI作为基于节点的工作流系统，提供了灵活的API接口，使其能够与Python深度集成，实现自动化图像生成和批量处理。下面详细介绍几种主要的集成方式及其实现方法。 ## 一、ComfyUI API基础调用 ### 1.1 通过HTTP API调用工作流 ComfyUI提供了完整的HTTP API接口，可以通过Python的requests库直接调用已配置的工作流： ```python import requests import json import time import uuid class ComfyUIClient: def __init__(self, server_address="127.0.0.1:8188"): self.server_address = server_address self.client_id = str(uuid.uuid4()) def queue_prompt(self, prompt, output_node_ids=None): """提交提示词到ComfyUI队列""" api_url = f"http://{self.server_address}/prompt" payload = { "prompt": prompt, "client_id": self.client_id } if output_node_ids: payload["extra_data"] = {"extra_pnginfo": {"output_nodes": output_node_ids}} response = requests.post(api_url, json=payload) return response.json() def get_image(self, prompt_id, filename): """获取生成的图像""" api_url = f"http://{self.server_address}/view" params = { "filename": filename, "subfolder": "", "type": "output", "prompt_id": prompt_id } response = requests.get(api_url, params=params) return response.content # 使用示例 client = ComfyUIClient() # 构建工作流JSON（从ComfyUI界面导出） workflow_json = { "3": { "class_type": "CLIPTextEncode", "inputs": { "text": "a beautiful landscape with mountains and rivers", "clip": ["4", 1] } }, "4": { "class_type": "LoadCheckpoint", "inputs": { "ckpt_name": "v1-5-pruned-emaonly.ckpt" } }, # ... 其他节点配置 } prompt_response = client.queue_prompt(workflow_json) print(f"任务ID: {prompt_response['prompt_id']}") ``` 这种方法适合从ComfyUI界面导出工作流后，通过Python进行参数化调用[ref_4]。 ### 1.2 工作流状态监控 ```python def wait_for_completion(self, prompt_id, check_interval=1, timeout=300): """等待任务完成""" start_time = time.time() api_url = f"http://{self.server_address}/history" while time.time() - start_time < timeout: history_response = requests.get(api_url) history_data = history_response.json() if prompt_id in history_data: job_data = history_data[prompt_id] if job_data['status'].get('completed', False): return job_data time.sleep(check_interval) raise TimeoutError(f"任务 {prompt_id} 超时") # 监控任务执行 try: job_result = client.wait_for_completion(prompt_response['prompt_id']) print("任务完成!") except TimeoutError as e: print(e) ``` ## 二、Python代码节点集成 ### 2.1 在ComfyUI工作流中使用Python节点 ComfyUI支持自定义Python节点，可以在工作流中直接嵌入Python代码： ```python # 自定义文本处理节点示例 import comfy.utils import torch class CustomTextProcessor: @classmethod def INPUT_TYPES(cls): return { "required": { "text": ("STRING", {"default": "", "multiline": True}), "processing_type": (["uppercase", "lowercase", "title_case"],) } } RETURN_TYPES = ("STRING",) FUNCTION = "process_text" CATEGORY = "custom_nodes" def process_text(self, text, processing_type): if processing_type == "uppercase": processed_text = text.upper() elif processing_type == "lowercase": processed_text = text.lower() elif processing_type == "title_case": processed_text = text.title() else: processed_text = text return (processed_text,) # 将节点注册到ComfyUI NODE_CLASS_MAPPINGS = { "CustomTextProcessor": CustomTextProcessor } NODE_DISPLAY_NAME_MAPPINGS = { "CustomTextProcessor": "Custom Text Processor" } ``` ### 2.2 动态参数生成节点 ```python class DynamicParameterGenerator: @classmethod def INPUT_TYPES(cls): return { "required": { "seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}), "variation_count": ("INT", {"default": 4, "min": 1, "max": 10}) } } RETURN_TYPES = ("INT",) # 返回多个种子 RETURN_NAMES = ("seeds",) FUNCTION = "generate_seeds" CATEGORY = "parameter_generation" OUTPUT_IS_LIST = (True,) # 表示输出是列表 def generate_seeds(self, seed, variation_count): import random random.seed(seed) seeds = [random.randint(0, 0xffffffffffffffff) for _ in range(variation_count)] return (seeds,) ``` ## 三、Dify平台集成方案 ### 3.1 使用Python代码节点调用ComfyUI API 在Dify平台中，可以通过Python代码节点直接调用ComfyUI的API： ```python import requests import base64 from io import BytesIO from PIL import Image def generate_image_with_comfyui(prompt_text, workflow_template, comfyui_server="localhost:8188"): """ 通过ComfyUI生成图像 """ # 替换工作流中的文本提示 for node_id, node_data in workflow_template.items(): if node_data.get("class_type") == "CLIPTextEncode": if "text" in node_data["inputs"]: node_data["inputs"]["text"] = prompt_text # 提交到ComfyUI queue_url = f"http://{comfyui_server}/prompt" response = requests.post(queue_url, json={"prompt": workflow_template}) if response.status_code == 200: prompt_data = response.json() prompt_id = prompt_data["prompt_id"] # 等待任务完成并获取结果 result = wait_for_completion(prompt_id, comfyui_server) if result and "images" in result: # 获取第一张图片 image_data = result["images"][0] return image_data return None def wait_for_completion(prompt_id, server_address, timeout=60): """等待ComfyUI任务完成""" import time history_url = f"http://{server_address}/history" start_time = time.time() while time.time() - start_time < timeout: history_response = requests.get(history_url) if history_response.status_code == 200: history_data = history_response.json() if prompt_id in history_data: job_info = history_data[prompt_id] if job_info.get("status", {}).get("completed"): return job_info["outputs"] time.sleep(1) return None ``` ### 3.2 工作流JSON导出与参数化 ```python def load_workflow_from_file(file_path): """从文件加载工作流模板""" import json with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) def parameterize_workflow(workflow, parameters): """ 参数化工作流配置 parameters: { "prompt": "生成的提示词", "negative_prompt": "负面提示词", "steps": 20, "cfg_scale": 7.5, "width": 512, "height": 512 } """ # 更新文本编码节点 for node_id, node_data in workflow.items(): if node_data.get("class_type") == "CLIPTextEncode": if "text" in node_data["inputs"]: # 根据节点名称判断是正面还是负面提示词 if "negative" in node_data["_meta"].get("title", "").lower(): node_data["inputs"]["text"] = parameters.get("negative_prompt", "") else: node_data["inputs"]["text"] = parameters.get("prompt", "") # 更新KSampler节点参数 elif node_data.get("class_type") == "KSampler": node_data["inputs"]["steps"] = parameters.get("steps", 20) node_data["inputs"]["cfg"] = parameters.get("cfg_scale", 7.5) # 更新EmptyLatentImage节点 elif node_data.get("class_type") == "EmptyLatentImage": node_data["inputs"]["width"] = parameters.get("width", 512) node_data["inputs"]["height"] = parameters.get("height", 512) return workflow ``` ## 四、批量处理与自动化 ### 4.1 批量图像生成 ```python import pandas as pd from concurrent.futures import ThreadPoolExecutor class BatchImageGenerator: def __init__(self, comfyui_server, workflow_template): self.client = ComfyUIClient(comfyui_server) self.workflow_template = workflow_template def generate_single_image(self, prompt_config): """生成单张图片""" try: # 参数化工作流 workflow = self.parameterize_workflow( self.workflow_template.copy(), prompt_config ) # 提交任务 response = self.client.queue_prompt(workflow) prompt_id = response['prompt_id'] # 等待完成 result = self.client.wait_for_completion(prompt_id) return { 'prompt': prompt_config['prompt'], 'prompt_id': prompt_id, 'success': True, 'image_info': result.get('images', []) } except Exception as e: return { 'prompt': prompt_config['prompt'], 'success': False, 'error': str(e) } def generate_batch(self, prompt_list, max_workers=2): """批量生成图片""" with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map(self.generate_single_image, prompt_list)) # 统计结果 successful = [r for r in results if r['success']] failed = [r for r in results if not r['success']] print(f"批量生成完成: 成功 {len(successful)} 张, 失败 {len(failed)} 张") return results # 使用示例 prompt_configs = [ {"prompt": "a serene mountain landscape at sunrise", "steps": 20, "cfg_scale": 7.5}, {"prompt": "a futuristic city with flying cars", "steps": 25, "cfg_scale": 8.0}, {"prompt": "an ancient castle in a mystical forest", "steps": 30, "cfg_scale": 7.0} ] generator = BatchImageGenerator("localhost:8188", base_workflow) results = generator.generate_batch(prompt_configs) ``` ## 五、高级集成技巧 ### 5.1 模型动态切换 ```python def switch_model_in_workflow(workflow, model_name): """动态切换模型""" for node_id, node_data in workflow.items(): if node_data.get("class_type") == "LoadCheckpoint": node_data["inputs"]["ckpt_name"] = model_name return workflow def get_available_models(server_address): """获取可用的模型列表""" api_url = f"http://{server_address}/object_info" response = requests.get(api_url) if response.status_code == 200: object_info = response.json() # 解析可用的检查点模型 checkpoint_nodes = [ node for node in object_info.values() if node.get("class_type") == "LoadCheckpoint" ] return checkpoint_nodes return [] ``` ### 5.2 工作流验证与调试 ```python def validate_workflow(workflow): """验证工作流配置""" required_nodes = {"LoadCheckpoint", "CLIPTextEncode", "KSampler", "VAEDecode"} found_nodes = set() for node_id, node_data in workflow.items(): found_nodes.add(node_data.get("class_type", "")) missing_nodes = required_nodes - found_nodes if missing_nodes: raise ValueError(f"工作流缺少必要节点: {missing_nodes}") return True def debug_workflow_execution(workflow, server_address): """调试工作流执行""" # 验证工作流 validate_workflow(workflow) # 检查服务器连接 try: response = requests.get(f"http://{server_address}/system_stats") if response.status_code != 200: raise ConnectionError("无法连接到ComfyUI服务器") except requests.exceptions.ConnectionError: raise ConnectionError("ComfyUI服务器未运行或地址错误") print("工作流验证通过，可以执行") ``` ## 六、实际应用场景 ### 6.1 电商产品图生成 ```python def generate_product_images(product_descriptions, style_template): """为电商产品生成多种风格的图片""" results = [] for description in product_descriptions: # 为每个产品生成多个变体 variations = [ f"professional product photo of {description}, studio lighting", f"lifestyle photo of {description}, natural lighting", f"creative artistic rendering of {description}" ] for variation in variations: config = { "prompt": variation, "negative_prompt": "blurry, low quality, watermark", "steps": 25, "cfg_scale": 7.5 } result = generate_single_image(config) results.append(result) return results ``` ### 6.2 社交媒体内容创作 ```python class SocialMediaContentGenerator: def __init__(self, comfyui_client, content_themes): self.client = comfyui_client self.themes = content_themes def generate_daily_content(self, date): """生成每日社交媒体内容""" theme = self.themes[date % len(self.themes)] prompts = [ f"inspiring quote about {theme}, minimalist design", f"motivational poster featuring {theme}", f"educational infographic about {theme}" ] return [self.generate_from_prompt(prompt) for prompt in prompts] ``` 通过以上方法，Python可以与ComfyUI文生图工作流深度结合，实现从简单的API调用到复杂的自动化工作流管理，大大提升了图像生成任务的效率和灵活性[ref_6]。这种集成方式特别适合需要批量处理、动态参数调整或与其他AI系统协同工作的应用场景。