openclaw+Nunchaku FLUX.1-dev：开源大模型文生图API服务封装实践

想用最新的FLUX.1-dev模型生成惊艳图片，但觉得ComfyUI操作太复杂？想把它变成简单的API服务，让其他程序也能轻松调用？今天就来分享一个实战方案：用openclaw把Nunchaku FLUX.1-dev模型封装成RESTful API服务。

这个方案的核心思路很简单：在ComfyUI的基础上，加一层Web服务封装。这样你就能通过HTTP请求来生成图片，不用再手动操作ComfyUI界面。无论是集成到自己的应用里，还是搭建一个图片生成服务，都变得特别方便。

1. 项目背景与方案选择

1.1 为什么需要API封装？

你可能已经体验过FLUX.1-dev模型的强大能力——它能生成细节丰富、质量极高的图片。但在实际应用中，我们经常遇到这些问题：

• 操作繁琐：每次生成图片都要打开ComfyUI界面，手动设置参数
• 难以集成：其他程序无法直接调用ComfyUI的功能
• 批量处理困难：需要一张张手动生成，效率低下
• 资源管理复杂：多用户同时使用时，显存、计算资源难以调度

API封装就是为了解决这些问题。把复杂的ComfyUI工作流变成简单的HTTP接口，就像给强大的引擎装上了方向盘和油门踏板，谁都能轻松驾驶。

1.2 openclaw是什么？

openclaw是一个开源的ComfyUI API封装工具，它有几个关键特点：

• 轻量级：基于Python开发，依赖简单，部署方便
• 功能完整：支持ComfyUI的大部分核心功能
• 易于扩展：你可以根据自己的需求定制API接口
• 社区活跃：有持续的更新和维护

选择openclaw而不是自己从头开发，能节省大量时间和精力。它已经解决了ComfyUI API调用的很多底层问题，我们只需要关注业务逻辑就行。

2. 环境准备与基础部署

在开始API封装之前，我们需要先搭建好基础环境。这部分和标准的ComfyUI部署类似，但有一些额外的要求。

2.1 硬件与软件要求

硬件配置建议：

• GPU：NVIDIA显卡，显存至少16GB（推荐24GB+）
• 内存：32GB以上
• 存储：至少50GB可用空间（用于存放模型文件）

软件环境：

• 操作系统：Ubuntu 20.04/22.04或Windows 10/11
• Python：3.10或3.11版本
• CUDA：11.8或12.1（根据PyTorch版本选择）
• Git：用于代码克隆

2.2 安装ComfyUI与Nunchaku插件

首先安装ComfyUI和Nunchaku插件，这是整个服务的基础：

# 1. 克隆ComfyUI仓库
git clone https://github.com/comfyanonymous/ComfyUI.git
cd ComfyUI

# 2. 创建虚拟环境（可选但推荐）
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或 venv\Scripts\activate  # Windows

# 3. 安装依赖
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install -r requirements.txt

# 4. 安装Nunchaku插件
cd custom_nodes
git clone https://github.com/mit-han-lab/ComfyUI-nunchaku nunchaku_nodes
cd ..

# 5. 安装Nunchaku后端
# 进入nunchaku_nodes目录，运行安装脚本
cd custom_nodes/nunchaku_nodes
python install_wheel.py
cd ../..

2.3 下载FLUX.1-dev模型文件

模型文件需要正确放置到指定目录：

# 创建必要的目录结构
mkdir -p models/{unet,loras,text_encoders,vae}

# 下载文本编码器模型
hf download comfyanonymous/flux_text_encoders clip_l.safetensors --local-dir models/text_encoders
hf download comfyanonymous/flux_text_encoders t5xxl_fp16.safetensors --local-dir models/text_encoders

# 下载VAE模型
hf download black-forest-labs/FLUX.1-schnell ae.safetensors --local-dir models/vae

# 下载FLUX.1-dev主模型（INT4量化版，显存占用较低）
hf download nunchaku-tech/nunchaku-flux.1-dev svdq-int4_r32-flux.1-dev.safetensors --local-dir models/unet/

# 下载可选LoRA模型（增强效果）
# FLUX.1-Turbo-Alpha LoRA
hf download nunchaku-tech/nunchaku-flux.1-dev-lora-turbo-alpha diffusion_pytorch_model.safetensors --local-dir models/loras/

如果你的网络环境访问Hugging Face较慢，也可以先下载到本地，然后创建软链接：

# 假设模型文件下载到了 ~/ai-models 目录
ln -s ~/ai-models/comfyanonymous/unet/svdq-int4_r32-flux.1-dev.safetensors models/unet/
ln -s ~/ai-models/comfyanonymous/text_encoders/clip_l.safetensors models/text_encoders/
ln -s ~/ai-models/comfyanonymous/text_encoders/t5xxl_fp16.safetensors models/text_encoders/
ln -s ~/ai-models/comfyanonymous/vae/ae.safetensors models/vae/
ln -s ~/ai-models/comfyanonymous/loras/diffusion_pytorch_model.safetensors models/loras/

3. openclaw安装与配置

有了ComfyUI基础环境，现在我们来安装和配置openclaw。

3.1 安装openclaw

openclaw可以通过pip直接安装：

# 在ComfyUI的虚拟环境中
pip install openclaw

# 或者从源码安装（获取最新版本）
git clone https://github.com/openclaw-ai/openclaw.git
cd openclaw
pip install -e .

3.2 基础配置

创建openclaw的配置文件：

# 在ComfyUI目录下创建配置目录
mkdir -p config

# 创建基础配置文件
cat > config/openclaw_config.yaml << EOF
# openclaw基础配置
server:
  host: "0.0.0.0"
  port: 8188
  workers: 1
  
comfyui:
  path: "."  # ComfyUI根目录路径
  auto_launch: true
  check_interval: 5
  
models:
  default_workflow: "nunchaku-flux.1-dev.json"
  workflows_dir: "user/default/example_workflows"
  
api:
  enable_cors: true
  rate_limit: 10  # 每分钟请求限制
  max_image_size: 2048  # 最大图片尺寸
  
logging:
  level: "INFO"
  file: "logs/openclaw.log"
EOF

3.3 准备ComfyUI工作流

openclaw需要ComfyUI的工作流文件来定义生成逻辑。我们先准备好Nunchaku FLUX.1-dev的工作流：

# 确保工作流目录存在
mkdir -p user/default/example_workflows

# 复制Nunchaku示例工作流
cp custom_nodes/nunchaku_nodes/example_workflows/* user/default/example_workflows/

# 查看可用的工作流
ls user/default/example_workflows/
# 应该能看到 nunchaku-flux.1-dev.json 等文件

4. API服务开发实战

现在进入核心部分：开发图片生成API服务。我们将创建一个完整的Flask应用，封装ComfyUI的功能。

4.1 创建基础API服务

首先创建一个简单的API服务文件：

# app.py - 主应用文件
import os
import json
import uuid
from flask import Flask, request, jsonify, send_file
from flask_cors import CORS
import openclaw
from openclaw.comfyui_client import ComfyUIClient

app = Flask(__name__)
CORS(app)  # 允许跨域请求

# 初始化openclaw客户端
comfyui_client = None

def init_comfyui():
    """初始化ComfyUI客户端"""
    global comfyui_client
    
    # ComfyUI配置
    comfyui_config = {
        "server_address": "http://127.0.0.1:8188",
        "client_id": str(uuid.uuid4()),
    }
    
    # 工作流配置
    workflow_config = {
        "workflow_path": "user/default/example_workflows/nunchaku-flux.1-dev.json",
        "output_dir": "output",
    }
    
    # 创建客户端
    comfyui_client = ComfyUIClient(
        server_address=comfyui_config["server_address"],
        client_id=comfyui_config["client_id"]
    )
    
    # 加载工作流
    with open(workflow_config["workflow_path"], "r") as f:
        workflow = json.load(f)
    
    comfyui_client.load_workflow(workflow)
    
    print("ComfyUI客户端初始化完成")
    return True

@app.route('/api/health', methods=['GET'])
def health_check():
    """健康检查接口"""
    return jsonify({
        "status": "healthy",
        "service": "FLUX.1-dev API",
        "version": "1.0.0"
    })

@app.route('/api/generate', methods=['POST'])
def generate_image():
    """图片生成接口"""
    try:
        # 获取请求参数
        data = request.json
        prompt = data.get('prompt', '')
        negative_prompt = data.get('negative_prompt', '')
        width = data.get('width', 1024)
        height = data.get('height', 1024)
        steps = data.get('steps', 20)
        cfg_scale = data.get('cfg_scale', 7.0)
        seed = data.get('seed', -1)  # -1表示随机种子
        
        if not prompt:
            return jsonify({"error": "提示词不能为空"}), 400
        
        # 设置工作流参数
        workflow_params = {
            "positive_prompt": prompt,
            "negative_prompt": negative_prompt,
            "width": width,
            "height": height,
            "steps": steps,
            "cfg_scale": cfg_scale,
            "seed": seed,
        }
        
        # 执行图片生成
        print(f"开始生成图片: {prompt[:50]}...")
        result = comfyui_client.generate_image(workflow_params)
        
        if result and result.get('images'):
            # 保存图片到临时文件
            image_data = result['images'][0]
            filename = f"generated_{uuid.uuid4().hex[:8]}.png"
            filepath = os.path.join('temp', filename)
            
            os.makedirs('temp', exist_ok=True)
            with open(filepath, 'wb') as f:
                f.write(image_data)
            
            return jsonify({
                "success": True,
                "image_url": f"/api/image/{filename}",
                "metadata": {
                    "prompt": prompt,
                    "size": f"{width}x{height}",
                    "steps": steps,
                    "seed": result.get('seed', seed)
                }
            })
        else:
            return jsonify({"error": "图片生成失败"}), 500
            
    except Exception as e:
        print(f"生成图片时出错: {str(e)}")
        return jsonify({"error": str(e)}), 500

@app.route('/api/image/<filename>', methods=['GET'])
def get_image(filename):
    """获取生成的图片"""
    filepath = os.path.join('temp', filename)
    if os.path.exists(filepath):
        return send_file(filepath, mimetype='image/png')
    else:
        return jsonify({"error": "图片不存在"}), 404

if __name__ == '__main__':
    # 初始化
    if init_comfyui():
        # 确保临时目录存在
        os.makedirs('temp', exist_ok=True)
        os.makedirs('logs', exist_ok=True)
        
        # 启动服务
        print("启动API服务...")
        app.run(host='0.0.0.0', port=5000, debug=False)
    else:
        print("初始化失败，服务无法启动")

4.2 创建启动脚本

为了方便管理，我们创建一个启动脚本：

# start_service.sh
#!/bin/bash

# 启动ComfyUI服务
echo "启动ComfyUI服务..."
cd /path/to/ComfyUI
python main.py --listen 0.0.0.0 --port 8188 &

# 等待ComfyUI启动
echo "等待ComfyUI启动..."
sleep 10

# 启动API服务
echo "启动API服务..."
cd /path/to/api_project
python app.py

# start_service.py - Python版本启动脚本
import subprocess
import time
import sys
import os

def start_comfyui():
    """启动ComfyUI服务"""
    print("启动ComfyUI服务...")
    
    # ComfyUI启动命令
    comfyui_cmd = [
        sys.executable, "main.py",
        "--listen", "0.0.0.0",
        "--port", "8188",
        "--disable-auto-launch"
    ]
    
    # 在后台启动ComfyUI
    comfyui_process = subprocess.Popen(
        comfyui_cmd,
        cwd=".",  # ComfyUI目录
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    print(f"ComfyUI进程ID: {comfyui_process.pid}")
    return comfyui_process

def check_comfyui_ready(max_retries=30, interval=2):
    """检查ComfyUI是否就绪"""
    import requests
    
    url = "http://127.0.0.1:8188"
    
    for i in range(max_retries):
        try:
            response = requests.get(f"{url}/history", timeout=5)
            if response.status_code == 200:
                print("ComfyUI服务已就绪")
                return True
        except:
            pass
        
        print(f"等待ComfyUI启动... ({i+1}/{max_retries})")
        time.sleep(interval)
    
    print("ComfyUI启动超时")
    return False

def start_api_service():
    """启动API服务"""
    print("启动API服务...")
    
    api_cmd = [sys.executable, "app.py"]
    
    # 前台运行API服务
    subprocess.run(api_cmd)

if __name__ == "__main__":
    # 切换到ComfyUI目录
    os.chdir("/path/to/ComfyUI")
    
    # 启动ComfyUI
    comfyui_process = start_comfyui()
    
    # 等待ComfyUI就绪
    if check_comfyui_ready():
        # 切换到API项目目录
        os.chdir("/path/to/api_project")
        
        # 启动API服务
        start_api_service()
    else:
        print("启动失败")
        comfyui_process.terminate()
        sys.exit(1)

4.3 创建Docker部署配置

为了便于部署，我们可以创建Docker配置：

# Dockerfile
FROM pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    git \
    wget \
    curl \
    libgl1-mesa-glx \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*

# 复制项目文件
COPY requirements.txt .
COPY . .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 下载模型文件（可以在构建时下载，或运行时下载）
# 这里假设模型文件已经准备好，通过volume挂载

# 暴露端口
EXPOSE 5000  # API服务端口
EXPOSE 8188  # ComfyUI端口

# 启动脚本
COPY start_service.sh .
RUN chmod +x start_service.sh

# 启动服务
CMD ["./start_service.sh"]

# docker-compose.yml
version: '3.8'

services:
  flux-api:
    build: .
    ports:
      - "5000:5000"  # API服务
      - "8188:8188"  # ComfyUI服务
    volumes:
      - ./models:/app/ComfyUI/models  # 挂载模型目录
      - ./output:/app/ComfyUI/output  # 挂载输出目录
      - ./temp:/app/api_project/temp  # 挂载临时文件
    environment:
      - PYTHONUNBUFFERED=1
      - CUDA_VISIBLE_DEVICES=0
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    restart: unless-stopped

5. 高级功能与优化

基础API服务搭建好后，我们可以添加更多实用功能。

5.1 批量处理接口

实际应用中，经常需要批量生成图片：

@app.route('/api/batch_generate', methods=['POST'])
def batch_generate():
    """批量图片生成接口"""
    try:
        data = request.json
        prompts = data.get('prompts', [])
        batch_size = data.get('batch_size', 1)
        
        if not prompts:
            return jsonify({"error": "提示词列表不能为空"}), 400
        
        results = []
        for i, prompt in enumerate(prompts):
            print(f"处理第 {i+1}/{len(prompts)} 个提示词: {prompt[:50]}...")
            
            # 设置参数
            workflow_params = {
                "positive_prompt": prompt,
                "negative_prompt": data.get('negative_prompt', ''),
                "width": data.get('width', 1024),
                "height": data.get('height', 1024),
                "steps": data.get('steps', 20),
                "cfg_scale": data.get('cfg_scale', 7.0),
                "seed": data.get('seed', -1),
            }
            
            # 生成图片
            result = comfyui_client.generate_image(workflow_params)
            
            if result and result.get('images'):
                # 保存图片
                image_data = result['images'][0]
                filename = f"batch_{uuid.uuid4().hex[:8]}.png"
                filepath = os.path.join('temp', filename)
                
                with open(filepath, 'wb') as f:
                    f.write(image_data)
                
                results.append({
                    "prompt": prompt,
                    "image_url": f"/api/image/{filename}",
                    "success": True
                })
            else:
                results.append({
                    "prompt": prompt,
                    "error": "生成失败",
                    "success": False
                })
        
        return jsonify({
            "success": True,
            "total": len(prompts),
            "success_count": len([r for r in results if r['success']]),
            "results": results
        })
        
    except Exception as e:
        return jsonify({"error": str(e)}), 500

5.2 图片编辑功能

除了文生图，还可以扩展图生图功能：

@app.route('/api/img2img', methods=['POST'])
def img2img():
    """图生图接口"""
    try:
        # 获取上传的图片
        if 'image' not in request.files:
            return jsonify({"error": "没有上传图片"}), 400
        
        image_file = request.files['image']
        prompt = request.form.get('prompt', '')
        
        # 保存上传的图片
        upload_dir = 'uploads'
        os.makedirs(upload_dir, exist_ok=True)
        
        upload_path = os.path.join(upload_dir, f"upload_{uuid.uuid4().hex[:8]}.png")
        image_file.save(upload_path)
        
        # 读取图片并转换为base64
        import base64
        with open(upload_path, 'rb') as f:
            image_data = f.read()
        
        image_b64 = base64.b64encode(image_data).decode('utf-8')
        
        # 设置图生图参数
        workflow_params = {
            "positive_prompt": prompt,
            "image_data": image_b64,
            "strength": float(request.form.get('strength', 0.7)),
            "width": int(request.form.get('width', 1024)),
            "height": int(request.form.get('height', 1024)),
            "steps": int(request.form.get('steps', 20)),
        }
        
        # 调用ComfyUI的图生图功能
        # 这里需要根据实际工作流调整参数设置
        result = comfyui_client.generate_image(workflow_params)
        
        if result and result.get('images'):
            # 保存生成的图片
            generated_data = result['images'][0]
            filename = f"img2img_{uuid.uuid4().hex[:8]}.png"
            filepath = os.path.join('temp', filename)
            
            with open(filepath, 'wb') as f:
                f.write(generated_data)
            
            return jsonify({
                "success": True,
                "image_url": f"/api/image/{filename}",
                "original_image": f"/api/upload/{os.path.basename(upload_path)}"
            })
        else:
            return jsonify({"error": "图片生成失败"}), 500
            
    except Exception as e:
        return jsonify({"error": str(e)}), 500

5.3 异步处理与队列

对于耗时的生成任务，可以使用队列系统：

# task_queue.py - 简单的任务队列实现
import queue
import threading
import time
from datetime import datetime

class TaskQueue:
    def __init__(self, max_workers=2):
        self.task_queue = queue.Queue()
        self.results = {}
        self.max_workers = max_workers
        self.workers = []
        self.running = True
        
        # 启动工作线程
        for i in range(max_workers):
            worker = threading.Thread(target=self._worker, args=(i,))
            worker.daemon = True
            worker.start()
            self.workers.append(worker)
    
    def _worker(self, worker_id):
        """工作线程函数"""
        while self.running:
            try:
                task_id, task_func, args, kwargs = self.task_queue.get(timeout=1)
                
                print(f"Worker {worker_id} 开始处理任务 {task_id}")
                
                try:
                    result = task_func(*args, **kwargs)
                    self.results[task_id] = {
                        "status": "completed",
                        "result": result,
                        "completed_at": datetime.now().isoformat()
                    }
                except Exception as e:
                    self.results[task_id] = {
                        "status": "failed",
                        "error": str(e),
                        "failed_at": datetime.now().isoformat()
                    }
                
                self.task_queue.task_done()
                print(f"Worker {worker_id} 完成任务 {task_id}")
                
            except queue.Empty:
                continue
    
    def submit(self, task_func, *args, **kwargs):
        """提交任务"""
        task_id = str(uuid.uuid4())
        
        self.results[task_id] = {
            "status": "pending",
            "submitted_at": datetime.now().isoformat()
        }
        
        self.task_queue.put((task_id, task_func, args, kwargs))
        return task_id
    
    def get_result(self, task_id, timeout=None):
        """获取任务结果"""
        start_time = time.time()
        
        while True:
            if task_id in self.results:
                result = self.results[task_id]
                if result["status"] in ["completed", "failed"]:
                    return result
            
            if timeout and (time.time() - start_time) > timeout:
                return {"status": "timeout", "error": "等待结果超时"}
            
            time.sleep(0.1)
    
    def stop(self):
        """停止队列"""
        self.running = False
        for worker in self.workers:
            worker.join()

# 在API服务中使用任务队列
task_queue = TaskQueue(max_workers=2)

@app.route('/api/async/generate', methods=['POST'])
def async_generate():
    """异步图片生成接口"""
    try:
        data = request.json
        prompt = data.get('prompt', '')
        
        if not prompt:
            return jsonify({"error": "提示词不能为空"}), 400
        
        # 定义生成任务函数
        def generate_task(prompt, params):
            workflow_params = {
                "positive_prompt": prompt,
                "negative_prompt": params.get('negative_prompt', ''),
                "width": params.get('width', 1024),
                "height": params.get('height', 1024),
                "steps": params.get('steps', 20),
                "cfg_scale": params.get('cfg_scale', 7.0),
            }
            
            result = comfyui_client.generate_image(workflow_params)
            
            if result and result.get('images'):
                # 保存图片
                image_data = result['images'][0]
                filename = f"async_{uuid.uuid4().hex[:8]}.png"
                filepath = os.path.join('temp', filename)
                
                with open(filepath, 'wb') as f:
                    f.write(image_data)
                
                return {
                    "image_url": f"/api/image/{filename}",
                    "prompt": prompt
                }
            else:
                raise Exception("图片生成失败")
        
        # 提交任务到队列
        task_id = task_queue.submit(generate_task, prompt, data)
        
        return jsonify({
            "success": True,
            "task_id": task_id,
            "status_url": f"/api/task/{task_id}"
        })
        
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/task/<task_id>', methods=['GET'])
def get_task_status(task_id):
    """获取任务状态"""
    result = task_queue.get_result(task_id, timeout=0.5)
    
    if result["status"] == "completed":
        return jsonify({
            "status": "completed",
            "result": result["result"]
        })
    elif result["status"] == "failed":
        return jsonify({
            "status": "failed",
            "error": result["error"]
        }), 500
    else:
        return jsonify({
            "status": result["status"]
        })

6. 性能优化与监控

6.1 缓存优化

对于频繁使用的提示词，可以添加缓存机制：

import hashlib
from functools import lru_cache

class ImageCache:
    def __init__(self, cache_dir="cache", max_size=100):
        self.cache_dir = cache_dir
        self.max_size = max_size
        os.makedirs(cache_dir, exist_ok=True)
        
        # 使用LRU缓存
        self.memory_cache = {}
    
    def get_cache_key(self, params):
        """生成缓存键"""
        # 基于参数生成唯一键
        param_str = json.dumps(params, sort_keys=True)
        return hashlib.md5(param_str.encode()).hexdigest()
    
    def get(self, params):
        """获取缓存"""
        cache_key = self.get_cache_key(params)
        
        # 先检查内存缓存
        if cache_key in self.memory_cache:
            return self.memory_cache[cache_key]
        
        # 检查文件缓存
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.png")
        if os.path.exists(cache_file):
            with open(cache_file, 'rb') as f:
                image_data = f.read()
            
            # 存入内存缓存
            self.memory_cache[cache_key] = image_data
            
            # 维护缓存大小
            if len(self.memory_cache) > self.max_size:
                # 移除最旧的缓存
                oldest_key = next(iter(self.memory_cache))
                del self.memory_cache[oldest_key]
            
            return image_data
        
        return None
    
    def set(self, params, image_data):
        """设置缓存"""
        cache_key = self.get_cache_key(params)
        
        # 保存到内存
        self.memory_cache[cache_key] = image_data
        
        # 保存到文件
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.png")
        with open(cache_file, 'wb') as f:
            f.write(image_data)
        
        # 维护缓存大小
        if len(self.memory_cache) > self.max_size:
            oldest_key = next(iter(self.memory_cache))
            del self.memory_cache[oldest_key]

# 在API服务中使用缓存
image_cache = ImageCache()

@app.route('/api/generate_cached', methods=['POST'])
def generate_cached():
    """带缓存的图片生成"""
    try:
        data = request.json
        prompt = data.get('prompt', '')
        
        # 检查缓存
        cached_image = image_cache.get(data)
        if cached_image:
            # 返回缓存的图片
            filename = f"cached_{uuid.uuid4().hex[:8]}.png"
            filepath = os.path.join('temp', filename)
            
            with open(filepath, 'wb') as f:
                f.write(cached_image)
            
            return jsonify({
                "success": True,
                "cached": True,
                "image_url": f"/api/image/{filename}"
            })
        
        # 没有缓存，生成新图片
        workflow_params = {
            "positive_prompt": prompt,
            "negative_prompt": data.get('negative_prompt', ''),
            "width": data.get('width', 1024),
            "height": data.get('height', 1024),
            "steps": data.get('steps', 20),
        }
        
        result = comfyui_client.generate_image(workflow_params)
        
        if result and result.get('images'):
            image_data = result['images'][0]
            
            # 保存到缓存
            image_cache.set(data, image_data)
            
            # 保存到临时文件
            filename = f"generated_{uuid.uuid4().hex[:8]}.png"
            filepath = os.path.join('temp', filename)
            
            with open(filepath, 'wb') as f:
                f.write(image_data)
            
            return jsonify({
                "success": True,
                "cached": False,
                "image_url": f"/api/image/{filename}"
            })
        else:
            return jsonify({"error": "图片生成失败"}), 500
            
    except Exception as e:
        return jsonify({"error": str(e)}), 500

6.2 监控与日志

添加监控和日志功能：

import logging
from logging.handlers import RotatingFileHandler
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from flask import Response

# 设置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        RotatingFileHandler('logs/api.log', maxBytes=10485760, backupCount=5),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Prometheus指标
REQUEST_COUNT = Counter('api_requests_total', 'Total API requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('api_request_latency_seconds', 'API request latency', ['endpoint'])
GENERATION_COUNT = Counter('image_generation_total', 'Total image generations', ['status'])
GENERATION_LATENCY = Histogram('image_generation_latency_seconds', 'Image generation latency')

@app.before_request
def before_request():
    """请求前处理"""
    request.start_time = time.time()

@app.after_request
def after_request(response):
    """请求后处理"""
    # 记录请求指标
    if hasattr(request, 'start_time'):
        latency = time.time() - request.start_time
        REQUEST_LATENCY.labels(request.endpoint).observe(latency)
    
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.endpoint,
        status=response.status_code
    ).inc()
    
    # 记录访问日志
    logger.info(f"{request.method} {request.path} - {response.status_code}")
    
    return response

@app.route('/api/generate_with_metrics', methods=['POST'])
def generate_with_metrics():
    """带监控的图片生成"""
    start_time = time.time()
    
    try:
        data = request.json
        prompt = data.get('prompt', '')
        
        # 记录生成开始
        logger.info(f"开始生成图片: {prompt[:50]}...")
        
        # 生成图片
        workflow_params = {
            "positive_prompt": prompt,
            "width": data.get('width', 1024),
            "height": data.get('height', 1024),
            "steps": data.get('steps', 20),
        }
        
        with GENERATION_LATENCY.time():
            result = comfyui_client.generate_image(workflow_params)
        
        if result and result.get('images'):
            # 记录成功
            GENERATION_COUNT.labels(status='success').inc()
            
            # 保存图片
            image_data = result['images'][0]
            filename = f"monitored_{uuid.uuid4().hex[:8]}.png"
            filepath = os.path.join('temp', filename)
            
            with open(filepath, 'wb') as f:
                f.write(image_data)
            
            # 记录生成时间
            generation_time = time.time() - start_time
            logger.info(f"图片生成成功，耗时: {generation_time:.2f}秒")
            
            return jsonify({
                "success": True,
                "image_url": f"/api/image/{filename}",
                "generation_time": generation_time
            })
        else:
            # 记录失败
            GENERATION_COUNT.labels(status='failed').inc()
            logger.error("图片生成失败")
            return jsonify({"error": "图片生成失败"}), 500
            
    except Exception as e:
        GENERATION_COUNT.labels(status='error').inc()
        logger.error(f"生成图片时出错: {str(e)}")
        return jsonify({"error": str(e)}), 500

@app.route('/metrics', methods=['GET'])
def metrics():
    """Prometheus指标端点"""
    return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)

@app.route('/api/stats', methods=['GET'])
def get_stats():
    """获取服务统计信息"""
    import psutil
    import GPUtil
    
    stats = {
        "system": {
            "cpu_percent": psutil.cpu_percent(),
            "memory_percent": psutil.virtual_memory().percent,
            "disk_usage": psutil.disk_usage('/').percent,
        },
        "gpu": [],
        "api": {
            "total_requests": REQUEST_COUNT._value.get(),
            "generation_success": GENERATION_COUNT.labels(status='success')._value.get(),
            "generation_failed": GENERATION_COUNT.labels(status='failed')._value.get(),
        }
    }
    
    # 获取GPU信息
    try:
        gpus = GPUtil.getGPUs()
        for gpu in gpus:
            stats["gpu"].append({
                "name": gpu.name,
                "load": gpu.load * 100,
                "memory_used": gpu.memoryUsed,
                "memory_total": gpu.memoryTotal,
                "temperature": gpu.temperature,
            })
    except:
        stats["gpu"] = "GPU信息不可用"
    
    return jsonify(stats)

7. 总结

通过openclaw封装Nunchaku FLUX.1-dev模型，我们成功将复杂的ComfyUI工作流转换成了简单易用的API服务。这个方案有几个明显的优势：

7.1 主要收获

技术层面：

• 简化了使用流程：从复杂的界面操作变成了简单的API调用
• 提高了集成性：其他程序可以通过HTTP请求轻松调用图片生成功能
• 支持批量处理：可以同时处理多个生成任务，提高效率
• 便于扩展：可以根据需求添加更多功能，如图片编辑、风格转换等

工程层面：

• 部署方便：支持Docker容器化部署，一键启动
• 监控完善：内置了性能监控和日志系统
• 缓存优化：对重复请求进行缓存，减少计算资源消耗
• 异步处理：支持长时间任务的异步执行，不阻塞主线程

7.2 实际应用建议

在实际部署和使用时，有几个建议：

1. 资源管理：根据显存大小调整并发数，避免显存溢出
2. 缓存策略：对常用提示词启用缓存，提高响应速度
3. 监控告警：设置GPU使用率、内存使用率等监控指标
4. 版本管理：定期更新模型和插件，获取更好的生成效果
5. 安全考虑：在生产环境添加API密钥验证、请求限流等安全措施

7.3 后续优化方向

如果你需要进一步优化这个方案，可以考虑：

• 模型量化：使用更低精度的模型减少显存占用
• 分布式部署：多GPU或多机器部署，提高并发处理能力
• 模型预热：提前加载模型到显存，减少首次生成延迟
• 智能调度：根据任务优先级和资源情况动态调度生成任务
• 结果后处理：自动添加水印、压缩图片、格式转换等

这个方案最大的价值在于，它把先进的AI图片生成能力变成了标准化的服务。无论是个人项目还是企业应用，都可以快速集成和使用。希望这个实践分享对你有所帮助！

---

获取更多AI镜像

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