效率提升技巧

本文档介绍ComfyUI使用过程中的效率提升技巧,包括工作流优化、快捷操作、自动化脚本等,帮助用户提高工作效率。

简介

效率提升分类

graph TD
    A[效率提升技巧] --> B[工作流优化]
    A --> C[快捷操作]
    A --> D[自动化脚本]
    A --> E[批量处理]
    A --> F[工具集成]

    B --> B1[节点优化]
    B --> B2[连接优化]
    B --> B3[参数优化]

    C --> C1[快捷键]
    C --> C2[模板使用]
    C --> C3[预设管理]

    D --> D1[脚本自动化]
    D --> D2[API调用]
    D --> D3[工作流编排]

    E --> E1[批量生成]
    E --> E2[批量处理]
    E --> E3[批量导出]

    F --> F1[外部工具]
    F --> F2[插件扩展]
    F --> F3[自定义节点]

    style A fill:#e1f5ff
    style B fill:#e1ffe1
    style C fill:#e1ffe1
    style D fill:#e1ffe1
    style E fill:#e1ffe1
    style F fill:#e1ffe1

工作流优化

1. 节点优化

graph TD
    A[节点复用] --> B[共享模型加载]
    A --> C[共享条件编码]
    A --> D[共享VAE解码]

    B --> B1[一次加载多次使用]
    C --> C2[一次编码多次使用]
    D --> D3[一次解码多次输出]

    style A fill:#e1f5ff
    style B fill:#e1ffe1
    style C fill:#e1ffe1
    style D fill:#e1ffe1
{
  "before_optimization": {
    "description": "重复加载模型",
    "nodes": [
      {"id": 1, "class": "CheckpointLoaderSimple", "model": "model.safetensors"},
      {"id": 2, "class": "CheckpointLoaderSimple", "model": "model.safetensors"},
      {"id": 3, "class": "CheckpointLoaderSimple", "model": "model.safetensors"}
    ]
  },
  "after_optimization": {
    "description": "共享模型加载",
    "nodes": [
      {"id": 1, "class": "CheckpointLoaderSimple", "model": "model.safetensors"},
      {"id": 2, "class": "KSampler", "model": [1, 0]},
      {"id": 3, "class": "KSampler", "model": [1, 0]}
    ]
  }
}
def merge_nodes(node1, node2):
    """合并功能相似的节点"""
    if node1['class_type'] == node2['class_type']:
        # 合并参数
        merged_params = {**node1['inputs'], **node2['inputs']}
        return {
            'class_type': node1['class_type'],
            'inputs': merged_params
        }
    return None

2. 连接优化

graph TD
    A[连接优化] --> B[减少交叉连接]
    A --> C[优化数据流向]
    A --> D[使用中间节点]

    B --> B1[重新布局节点]
    C --> C2[简化数据流]
    D --> D3[引入中间节点]

    style A fill:#e1f5ff
    style B fill:#e1ffe1
    style C fill:#e1ffe1
    style D fill:#e1ffe1
{
  "connection_optimization": {
    "strategies": [
      "使用分组节点减少连接",
      "引入中间节点简化连接",
      "优化节点布局减少交叉",
      "使用条件节点控制数据流"
    ]
  }
}

3. 参数优化

class ParameterPresets:
    def __init__(self):
        self.presets = {
            'high_quality': {
                'steps': 50,
                'cfg': 7.5,
                'denoise': 0.9
            },
            'fast': {
                'steps': 20,
                'cfg': 6.0,
                'denoise': 0.8
            },
            'balanced': {
                'steps': 30,
                'cfg': 7.0,
                'denoise': 0.85
            }
        }

    def get_preset(self, name):
        return self.presets.get(name, {})
def optimize_parameters(workflow, target_quality=0.85):
    """自动优化参数"""
    # 实现参数优化算法
    pass

快捷操作

4. 快捷键

{
  "keyboard_shortcuts": {
    "workflow": {
      "Ctrl+N": "新建工作流",
      "Ctrl+O": "打开工作流",
      "Ctrl+S": "保存工作流",
      "Ctrl+Z": "撤销",
      "Ctrl+Y": "重做"
    },
    "nodes": {
      "Ctrl+A": "全选节点",
      "Ctrl+C": "复制节点",
      "Ctrl+V": "粘贴节点",
      "Delete": "删除节点",
      "Ctrl+D": "复制节点"
    },
    "execution": {
      "F5": "执行工作流",
      "Shift+F5": "停止执行",
      "Ctrl+R": "重置工作流"
    }
  }
}
def setup_custom_shortcuts():
    """设置自定义快捷键"""
    shortcuts = {
        'Ctrl+Shift+Q': 'quick_export',
        'Ctrl+Shift+P': 'quick_preview',
        'Ctrl+Shift+S': 'quick_save'
    }
    return shortcuts

5. 模板使用

graph TD
    A[模板库] --> B[基础模板]
    A --> C[进阶模板]
    A --> D[高级模板]

    B --> B1[文生图模板]
    B --> B2[图生图模板]
    B --> B3[图像修复模板]

    C --> C1[LoRA模板]
    C --> C2[ControlNet模板]
    C --> C3[批量处理模板]

    D --> D1[多模型融合]
    D --> D2[智能提示词]
    D --> D3[自适应优化]

    style A fill:#e1f5ff
    style B fill:#e1ffe1
    style C fill:#e1ffe1
    style D fill:#e1ffe1
class TemplateManager:
    def __init__(self):
        self.templates = {}

    def load_template(self, template_path):
        """加载模板"""
        with open(template_path, 'r') as f:
            template = json.load(f)
        self.templates[template['name']] = template

    def save_template(self, name, workflow):
        """保存模板"""
        template = {
            'name': name,
            'workflow': workflow,
            'created_at': datetime.now().isoformat()
        }
        self.templates[name] = template

    def apply_template(self, name):
        """应用模板"""
        return self.templates.get(name, {})

6. 预设管理

{
  "preset_system": {
    "categories": {
      "quality": ["high_quality", "medium_quality", "fast_preview"],
      "style": ["realistic", "artistic", "anime"],
      "resolution": ["512x512", "768x768", "1024x1024"]
    },
    "presets": {
      "portrait_high_quality": {
        "quality": "high_quality",
        "style": "realistic",
        "resolution": "1024x1024"
      }
    }
  }
}
def apply_preset(workflow, preset_name):
    """应用预设到工作流"""
    preset = load_preset(preset_name)

    for node_id, node_data in workflow.items():
        if node_data['class_type'] in preset:
            node_data['inputs'].update(preset[node_data['class_type']])

    return workflow

自动化脚本

7. 脚本自动化

import requests
import json

class ComfyUIAutomation:
    def __init__(self, server_url="http://127.0.0.1:8188"):
        self.server_url = server_url

    def execute_workflow(self, workflow):
        """执行工作流"""
        response = requests.post(
            f"{self.server_url}/prompt",
            json={"prompt": workflow}
        )
        return response.json()

    def get_history(self, prompt_id):
        """获取执行历史"""
        response = requests.get(
            f"{self.server_url}/history/{prompt_id}"
        )
        return response.json()

    def download_image(self, filename, save_path):
        """下载生成的图像"""
        response = requests.get(
            f"{self.server_url}/view?filename={filename}"
        )
        with open(save_path, 'wb') as f:
            f.write(response.content)

# 使用示例
automation = ComfyUIAutomation()
result = automation.execute_workflow(workflow)
prompt_id = result['prompt_id']
def batch_generate(prompts, output_dir):
    """批量生成图像"""
    automation = ComfyUIAutomation()

    for i, prompt in enumerate(prompts):
        # 修改工作流中的提示词
        workflow['3']['inputs']['text'] = prompt

        # 执行工作流
        result = automation.execute_workflow(workflow)
        prompt_id = result['prompt_id']

        # 等待完成
        wait_for_completion(prompt_id)

        # 下载图像
        filename = f"output_{i}.png"
        automation.download_image(filename, f"{output_dir}/{filename}")

8. API调用

import requests

class ComfyUIAPI:
    def __init__(self, base_url="http://127.0.0.1:8188"):
        self.base_url = base_url

    def queue_prompt(self, workflow):
        """提交工作流到队列"""
        endpoint = f"{self.base_url}/prompt"
        payload = {"prompt": workflow}
        response = requests.post(endpoint, json=payload)
        return response.json()

    def get_queue_info(self):
        """获取队列信息"""
        endpoint = f"{self.base_url}/queue"
        response = requests.get(endpoint)
        return response.json()

    def get_history(self, prompt_id=None):
        """获取历史记录"""
        endpoint = f"{self.base_url}/history"
        if prompt_id:
            endpoint += f"/{prompt_id}"
        response = requests.get(endpoint)
        return response.json()

    def get_images(self, filename, subfolder="", folder_type="output"):
        """获取图像"""
        endpoint = f"{self.base_url}/view"
        params = {
            "filename": filename,
            "subfolder": subfolder,
            "type": folder_type
        }
        response = requests.get(endpoint, params=params)
        return response.content

9. 工作流编排

class WorkflowOrchestrator:
    def __init__(self):
        self.workflows = {}
        self.dependencies = {}

    def add_workflow(self, name, workflow, dependencies=None):
        """添加工作流"""
        self.workflows[name] = workflow
        self.dependencies[name] = dependencies or []

    def execute_all(self):
        """执行所有工作流"""
        executed = set()

        for workflow_name in self.workflows:
            self._execute_workflow(workflow_name, executed)

    def _execute_workflow(self, name, executed):
        """执行单个工作流"""
        if name in executed:
            return

        # 先执行依赖
        for dep in self.dependencies[name]:
            self._execute_workflow(dep, executed)

        # 执行当前工作流
        workflow = self.workflows[name]
        result = execute_workflow(workflow)
        executed.add(name)

        return result

批量处理

10. 批量生成

graph TD
    A[批量生成] --> B[提示词批量]
    A --> C[参数批量]
    A --> D[模型批量]

    B --> B1[提示词列表]
    B --> B2[提示词模板]
    B --> B3[提示词变体]

    C --> C1[参数网格搜索]
    C --> C2[参数随机采样]
    C --> C3[参数优化]

    D --> D1[模型列表]
    D --> D2[模型组合]
    D --> D3[模型切换]

    style A fill:#e1f5ff
    style B fill:#e1ffe1
    style C fill:#e1ffe1
    style D fill:#e1ffe1
class BatchGenerator:
    def __init__(self, base_workflow):
        self.base_workflow = base_workflow

    def generate_from_prompts(self, prompts):
        """从提示词列表批量生成"""
        results = []

        for prompt in prompts:
            # 修改提示词
            workflow = self._modify_prompt(prompt)

            # 执行生成
            result = execute_workflow(workflow)
            results.append(result)

        return results

    def generate_from_parameters(self, parameter_grid):
        """从参数网格批量生成"""
        results = []

        for params in parameter_grid:
            # 修改参数
            workflow = self._modify_parameters(params)

            # 执行生成
            result = execute_workflow(workflow)
            results.append(result)

        return results

    def _modify_prompt(self, prompt):
        """修改工作流中的提示词"""
        workflow = json.loads(json.dumps(self.base_workflow))
        workflow['3']['inputs']['text'] = prompt
        return workflow

11. 批量处理

class BatchImageProcessor:
    def __init__(self, workflow_template):
        self.workflow_template = workflow_template

    def process_directory(self, input_dir, output_dir):
        """批量处理目录中的图像"""
        # 获取所有图像文件
        image_files = self._get_image_files(input_dir)

        results = []
        for image_file in image_files:
            # 加载图像
            image = load_image(image_file)

            # 创建工作流
            workflow = self._create_workflow(image)

            # 执行处理
            result = execute_workflow(workflow)

            # 保存结果
            output_path = f"{output_dir}/{os.path.basename(image_file)}"
            save_result(result, output_path)

            results.append(output_path)

        return results

    def _get_image_files(self, directory):
        """获取目录中的图像文件"""
        extensions = ['.png', '.jpg', '.jpeg', '.webp']
        files = []

        for file in os.listdir(directory):
            if any(file.lower().endswith(ext) for ext in extensions):
                files.append(os.path.join(directory, file))

        return files

12. 批量导出

class BatchExporter:
    def __init__(self):
        self.export_formats = ['png', 'jpg', 'webp']

    def export_results(self, results, output_dir, format='png'):
        """批量导出结果"""
        if format not in self.export_formats:
            raise ValueError(f"Unsupported format: {format}")

        exported_files = []

        for i, result in enumerate(results):
            filename = f"output_{i:04d}.{format}"
            output_path = os.path.join(output_dir, filename)

            # 保存图像
            result.save(output_path, format=format.upper())
            exported_files.append(output_path)

        return exported_files

    def export_metadata(self, results, metadata_file):
        """导出元数据"""
        metadata = []

        for i, result in enumerate(results):
            metadata.append({
                'index': i,
                'filename': result['filename'],
                'parameters': result['parameters'],
                'timestamp': result['timestamp']
            })

        with open(metadata_file, 'w') as f:
            json.dump(metadata, f, indent=2)

工具集成

13. 外部工具

from PIL import Image, ImageEnhance

class ImageEditorIntegration:
    def __init__(self):
        self.editors = {
            'enhance': self._enhance_image,
            'resize': self._resize_image,
            'crop': self._crop_image
        }

    def apply_edit(self, image, edit_type, **kwargs):
        """应用编辑操作"""
        editor = self.editors.get(edit_type)
        if editor:
            return editor(image, **kwargs)
        return image

    def _enhance_image(self, image, brightness=1.0, contrast=1.0, saturation=1.0):
        """增强图像"""
        enhancer = ImageEnhance.Brightness(image)
        image = enhancer.enhance(brightness)

        enhancer = ImageEnhance.Contrast(image)
        image = enhancer.enhance(contrast)

        enhancer = ImageEnhance.Color(image)
        image = enhancer.enhance(saturation)

        return image

    def _resize_image(self, image, size):
        """调整图像大小"""
        return image.resize(size, Image.LANCZOS)

    def _crop_image(self, image, box):
        """裁剪图像"""
        return image.crop(box)
import shutil
import os

class FileManagerIntegration:
    def __init__(self):
        self.organize_rules = {
            'by_date': self._organize_by_date,
            'by_type': self._organize_by_type,
            'by_size': self._organize_by_size
        }

    def organize_files(self, source_dir, target_dir, rule='by_date'):
        """组织文件"""
        organizer = self.organize_rules.get(rule)
        if organizer:
            return organizer(source_dir, target_dir)

    def _organize_by_date(self, source_dir, target_dir):
        """按日期组织文件"""
        for file in os.listdir(source_dir):
            file_path = os.path.join(source_dir, file)

            if os.path.isfile(file_path):
                # 获取文件修改日期
                mtime = os.path.getmtime(file_path)
                date_str = datetime.fromtimestamp(mtime).strftime('%Y-%m-%d')

                # 创建目标目录
                date_dir = os.path.join(target_dir, date_str)
                os.makedirs(date_dir, exist_ok=True)

                # 移动文件
                shutil.move(file_path, os.path.join(date_dir, file))

14. 插件扩展

class CustomPlugin:
    def __init__(self):
        self.name = "CustomPlugin"
        self.version = "1.0.0"
        self.nodes = {}

    def register_node(self, node_class):
        """注册自定义节点"""
        self.nodes[node_class.__name__] = node_class

    def get_node(self, node_name):
        """获取节点类"""
        return self.nodes.get(node_name)

# 自定义节点示例
class CustomNode:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "input_data": ("DATA",),
                "parameter": ("FLOAT", {"default": 1.0})
            }
        }

    RETURN_TYPES = ("DATA",)
    FUNCTION = "process"

    def process(self, input_data, parameter):
        # 处理逻辑
        result = input_data * parameter
        return (result,)

15. 自定义节点

class CustomNodeTemplate:
    """自定义节点模板"""

    @classmethod
    def INPUT_TYPES(cls):
        """定义输入类型"""
        return {
            "required": {
                "input1": ("TYPE1",),
                "input2": ("TYPE2", {"default": default_value})
            },
            "optional": {
                "optional_input": ("TYPE3",)
            }
        }

    RETURN_TYPES = ("OUTPUT_TYPE",)
    RETURN_NAMES = ("output_name",)
    FUNCTION = "execute"

    CATEGORY = "Custom"

    def execute(self, input1, input2, optional_input=None):
        """执行节点逻辑"""
        # 处理逻辑
        result = self._process(input1, input2, optional_input)
        return (result,)

    def _process(self, input1, input2, optional_input):
        """内部处理方法"""
        pass

效率监控

16. 性能监控

import time
from collections import defaultdict

class EfficiencyMonitor:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.start_times = {}

    def start_timer(self, name):
        """开始计时"""
        self.start_times[name] = time.time()

    def stop_timer(self, name):
        """停止计时"""
        if name in self.start_times:
            elapsed = time.time() - self.start_times[name]
            self.metrics[name].append(elapsed)
            del self.start_times[name]

    def get_stats(self):
        """获取统计信息"""
        stats = {}
        for name, times in self.metrics.items():
            stats[name] = {
                'count': len(times),
                'total': sum(times),
                'average': sum(times) / len(times),
                'min': min(times),
                'max': max(times)
            }
        return stats

    def print_report(self):
        """打印性能报告"""
        stats = self.get_stats()
        print("\nEfficiency Report:")
        print("-" * 60)
        for name, data in sorted(stats.items(), key=lambda x: x[1]['total'], reverse=True):
            print(f"{name}:")
            print(f"  Count: {data['count']}")
            print(f"  Total: {data['total']:.3f}s")
            print(f"  Average: {data['average']:.3f}s")
            print()

最佳实践

效率提升清单

## 工作流优化
- [ ] 复用节点减少冗余
- [ ] 优化连接复杂度
- [ ] 使用参数预设

## 快捷操作
- [ ] 学习快捷键
- [ ] 使用工作流模板
- [ ] 管理预设配置

## 自动化
- [ ] 编写自动化脚本
- [ ] 使用API调用
- [ ] 编排工作流

## 批量处理
- [ ] 批量生成图像
- [ ] 批量处理文件
- [ ] 批量导出结果

## 工具集成
- [ ] 集成外部工具
- [ ] 开发自定义插件
- [ ] 创建自定义节点

总结

效率提升是ComfyUI使用中的重要方面。关键要点:

  1. 工作流优化: 优化节点、连接和参数
  2. 快捷操作: 使用快捷键、模板和预设
  3. 自动化: 编写脚本和使用API
  4. 批量处理: 实现批量生成和处理
  5. 工具集成: 集成外部工具和扩展功能

通过应用这些技巧,可以显著提高ComfyUI的使用效率。