Como Executar um Fluxo de Trabalho Flux ComfyUI como uma API

Flux é um modelo de texto-para-imagem de última geração desenvolvido pela Black Forest Labs, apresentando capacidades avançadas de geração de imagens. ComfyUI, por outro lado, é uma interface poderosa e modular de modelo de difusão com um sistema de fluxo de trabalho baseado em nós. Quando combinados, eles criam uma solução robusta para a geração de imagens de alta qualidade que pode ser acessada programaticamente.

Combinar o poder dos modelos Flux com a versatilidade do ComfyUI pode aprimorar significativamente o seu fluxo de trabalho de geração de imagens com IA. Este tutorial irá guiá-lo pelo processo de execução do Flux no ComfyUI como uma API, abrindo possibilidades para automação e integração com outros sistemas.

💡

Antes de mergulharmos na configuração da nossa API Flux ComfyUI, vamos falar sobre ferramentas de teste. Ao trabalhar com APIs, ter uma plataforma de teste confiável é essencial, e o Apidog se destaca como a principal alternativa ao Postman neste espaço.

O Apidog oferece um ecossistema abrangente de desenvolvimento de APIs com recursos como testes automatizados, servidores simulados e geração de documentação detalhada, tudo em uma plataforma unificada.

botão

Diferente do Postman, o Apidog fornece gerenciamento contínuo de ciclo de vida da API com ferramentas de depuração integradas, especificamente otimizadas para fluxos de trabalho complexos como os do ComfyUI. Sua interface intuitiva torna o teste dos seus endpoints do Flux ComfyUI significativamente mais eficiente, com recursos colaborativos que permitem que os membros da equipe compartilhem e aprimorem as configurações da API juntos.

Se você está sério sobre desenvolvimento e teste de APIs, a solução all-in-one do Apidog irá agilizar consideravelmente seu fluxo de trabalho.

botão

Configurando Seu Ambiente Flux ComfyUI

Antes de mergulhar na implementação da API, precisamos configurar o ambiente necessário. Usaremos o Modal, uma plataforma em nuvem que simplifica a execução de aplicativos serverless com acesso a GPU.

Pré-requisitos para Flux ComfyUI

Uma conta no Modal (crie uma em modal.com)
Python 3.11 ou posterior
Git instalado na sua máquina local
Compreensão básica de Python e conceitos de API

Criando a Imagem do Container Flux ComfyUI

Nossa primeira etapa é criar uma imagem de container que inclua todos os componentes necessários para executar o Flux no ComfyUI. Vamos dividir o processo:

import json
import subprocess
import uuid
from pathlib import Path
from typing import Dict

import modal

# Crie uma imagem base com dependências necessárias
image = (
    modal.Image.debian_slim(python_version="3.11")
    .apt_install("git")  # Instalar git para clonar o ComfyUI
    .pip_install("fastapi[standard]==0.115.4")  # Dependências da web
    .pip_install("comfy-cli==1.3.5")  # Ferramenta CLI do ComfyUI
    .run_commands(
        "comfy --skip-prompt install --nvidia --version 0.3.10"  # Instalar ComfyUI
    )
)

Este código cria uma imagem Modal baseada no Debian Slim com Python 3.11, em seguida, instala Git, FastAPI e comfy-cli. A ferramenta comfy-cli é então utilizada para instalar o próprio ComfyUI.

Aprimorando Seu Flux ComfyUI com Nodes Personalizados

Para desbloquear funcionalidades adicionais, iremos adicionar nodes personalizados à nossa instalação do ComfyUI:

image = (
    image.run_commands(
        "comfy node install was-node-suite-comfyui@1.0.2"  # Instalar WAS Node Suite
    )
    # Adicione mais nodes personalizados conforme necessário
)

O WAS Node Suite oferece funcionalidades adicionais para geração e manipulação de imagens, que pode ser particularmente útil ao trabalhar com modelos Flux.

Configurando o Download do Modelo Flux ComfyUI

Um gerenciamento eficiente de modelos é crucial para uma experiência suave. Vamos criar uma função para baixar o modelo Flux e armazená-lo em um volume persistente:

def hf_download():
    from huggingface_hub import hf_hub_download

    flux_model = hf_hub_download(
        repo_id="Comfy-Org/flux1-schnell",
        filename="flux1-schnell-fp8.safetensors",
        cache_dir="/cache",
    )

    # Criar um symlink do modelo para o diretório do ComfyUI
    subprocess.run(
        f"ln -s {flux_model} /root/comfy/ComfyUI/models/checkpoints/flux1-schnell-fp8.safetensors",
        shell=True,
        check=True,
    )

# Criar um volume persistente para armazenamento de modelos
vol = modal.Volume.from_name("hf-hub-cache", create_if_missing=True)

# Adicionar a função de download do modelo à nossa imagem
image = (
    image.pip_install("huggingface_hub[hf_transfer]==0.26.2")
    .env({"HF_HUB_ENABLE_HF_TRANSFER": "1"})
    .run_function(
        hf_download,
        volumes={"/cache": vol},
    )
)

# Adicionar nosso JSON de fluxo de trabalho ao container
image = image.add_local_file(
    Path(__file__).parent / "workflow_api.json", "/root/workflow_api.json"
)

Este código define uma função para baixar o modelo Flux Schnell do Hugging Face e criar um symlink para o diretório de modelos do ComfyUI. Também criamos um Volume Modal para persistir os modelos baixados entre as execuções.

Desenvolvimento Interativo do Flux ComfyUI

Antes de implantar como uma API, é útil testar interativamente seu fluxo de trabalho do Flux ComfyUI:

app = modal.App(name="example-comfyui", image=image)

@app.function(
    allow_concurrent_inputs=10,  # Necessário para startup da UI
    max_containers=1,  # Limitar a um container para uso interativo
    gpu="L40S",  # GPU poderosa para inferência do modelo Flux
    volumes={"/cache": vol},  # Montar nossos modelos em cache
)
@modal.web_server(8000, startup_timeout=60)
def ui():
    subprocess.Popen(
        "comfy launch -- --listen 0.0.0.0 --port 8000",
        shell=True
    )

Você pode rodar isso com modal serve seu_arquivo.py e acessar a UI em seu navegador para projetar e testar seus fluxos de trabalho antes de implementar a API.

Implementando a API Flux ComfyUI

Agora, o principal evento - transformando o ComfyUI em um endpoint de API para inferência de modelos Flux:

@app.cls(
    allow_concurrent_inputs=10,  # Permitir múltiplas chamadas de API simultâneas
    scaledown_window=300,  # Manter containers vivos por 5 minutos após o uso
    gpu="L40S",
    volumes={"/cache": vol},
)
class ComfyUI:
    @modal.enter()
    def launch_comfy_background(self):
        # Iniciar o servidor ComfyUI em segundo plano
        cmd = "comfy launch --background"
        subprocess.run(cmd, shell=True, check=True)

    @modal.method()
    def infer(self, workflow_path: str = "/root/workflow_api.json"):
        # Verificar a saúde do servidor ComfyUI
        self.poll_server_health()

        # Executar fluxo de trabalho com comfy-cli
        cmd = f"comfy run --workflow {workflow_path} --wait --timeout 1200 --verbose"
        subprocess.run(cmd, shell=True, check=True)

        # Encontrar imagem de saída
        output_dir = "/root/comfy/ComfyUI/output"
        workflow = json.loads(Path(workflow_path).read_text())
        file_prefix = [
            node.get("inputs")
            for node in workflow.values()
            if node.get("class_type") == "SaveImage"
        ][0]["filename_prefix"]

        # Retornar imagem como bytes
        for f in Path(output_dir).iterdir():
            if f.name.startswith(file_prefix):
                return f.read_bytes()

    @modal.fastapi_endpoint(method="POST")
    def api(self, item: Dict):
        from fastapi import Response

        # Carregar o template do fluxo de trabalho
        workflow_data = json.loads(
            (Path(__file__).parent / "workflow_api.json").read_text()
        )

        # Inserir o prompt
        workflow_data["6"]["inputs"]["text"] = item["prompt"]

        # Gerar ID exclusivo para este pedido
        client_id = uuid.uuid4().hex
        workflow_data["9"]["inputs"]["filename_prefix"] = client_id

        # Salvar fluxo de trabalho modificado 
        new_workflow_file = f"{client_id}.json"
        json.dump(workflow_data, Path(new_workflow_file).open("w"))

        # Executar inferência
        img_bytes = self.infer.local(new_workflow_file)

        return Response(img_bytes, media_type="image/jpeg")

    def poll_server_health(self) -> Dict:
        import socket
        import urllib

        try:
            # Verificar se o servidor está saudável
            req = urllib.request.Request("<http://127.0.0.1:8188/system_stats>")
            urllib.request.urlopen(req, timeout=5)
            print("O servidor ComfyUI está saudável")
        except (socket.timeout, urllib.error.URLError) as e:
            # Parar o container se o servidor estiver doente
            print(f"A verificação de saúde do servidor falhou: {str(e)}")
            modal.experimental.stop_fetching_inputs()
            raise Exception("O servidor ComfyUI não está saudável, parando o container")

Esta classe define a funcionalidade central da API:

launch_comfy_background inicia o servidor ComfyUI quando o container é iniciado
infer executa um fluxo de trabalho e retorna a imagem gerada
api é o endpoint FastAPI que aceita um prompt, modifica o fluxo de trabalho e retorna a imagem
poll_server_health garante que o servidor ComfyUI esteja responsivo

Testando Sua API Flux ComfyUI

Para testar sua API, você pode criar um script cliente simples:

import requests
import base64
from PIL import Image
import io
import argparse

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", default="Uma paisagem surreal com ilhas flutuantes e luz etérea")
    parser.add_argument("--endpoint", default="<https://seu-endpoint-modal.modal.run>")
    args = parser.parse_args()

    # Enviar pedido para a API
    response = requests.post(
        f"{args.endpoint}/api",
        json={"prompt": args.prompt}
    )

    # Lidar com a resposta
    if response.status_code == 200:
        # Salvar e exibir a imagem
        image = Image.open(io.BytesIO(response.content))
        image.save("flux_output.jpg")
        print(f"Imagem gerada e salva como flux_output.jpg")

        # Opcional: exibir a imagem se estiver em um ambiente adequado
        try:
            image.show()
        except:
            pass
    else:
        print(f"Erro: {response.status_code}, {response.text}")

if __name__ == "__main__":
    main()

Otimizando o Desempenho do Flux ComfyUI

Para garantir que sua API continue responsiva e econômica, considere implementar as seguintes otimizações:

1. Snapshots de Memória para Inicializações Mais Rápidas do Flux ComfyUI

Inicializações frias podem ser um gargalo para aplicações serverless. O Modal oferece snapshots de memória que podem reduzir significativamente os tempos de inicialização:

@app.cls(
    allow_concurrent_inputs=10,
    gpu="L40S",
    volumes={"/cache": vol},
    memory_snapshot=modal.MemorySnapshot(
        snapshot_path="/root/comfy-snapshot",
        boot_command="comfy launch --background",
    ),
)
class ComfyUI:
    # Resto da implementação da classe

2. Lote de Pedidos para Flux ComfyUI

Para cenários de alta vazão, implementar o lote de pedidos pode melhorar a eficiência:

@modal.method()
def batch_inference(self, prompts: list[str]):
    results = []
    for prompt in prompts:
        # Criar fluxo de trabalho para cada prompt
        client_id = uuid.uuid4().hex
        workflow_data = json.loads(
            (Path(__file__).parent / "workflow_api.json").read_text()
        )
        workflow_data["6"]["inputs"]["text"] = prompt
        workflow_data["9"]["inputs"]["filename_prefix"] = client_id

        # Salvar e processar fluxo de trabalho
        new_workflow_file = f"{client_id}.json"
        json.dump(workflow_data, Path(new_workflow_file).open("w"))
        img_bytes = self.infer.local(new_workflow_file)
        results.append(img_bytes)

    return results

Recursos Avançados da API Flux ComfyUI

Customizando os Parâmetros do Fluxo de Trabalho

Amplie sua API para permitir parâmetros adicionais além do prompt apenas:

@modal.fastapi_endpoint(method="POST")
def advanced_api(self, item: Dict):
    from fastapi import Response

    workflow_data = json.loads(
        (Path(__file__).parent / "workflow_api.json").read_text()
    )

    # Inserir o prompt
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # Opcional: Definir parâmetros adicionais se fornecidos
    if "negative_prompt" in item:
        workflow_data["7"]["inputs"]["text"] = item["negative_prompt"]

    if "cfg_scale" in item:
        workflow_data["3"]["inputs"]["cfg"] = item["cfg_scale"]

    if "steps" in item:
        workflow_data["3"]["inputs"]["steps"] = item["steps"]

    # Gerar ID exclusivo
    client_id = uuid.uuid4().hex
    workflow_data["9"]["inputs"]["filename_prefix"] = client_id

    # Salvar e processar fluxo de trabalho
    new_workflow_file = f"{client_id}.json"
    json.dump(workflow_data, Path(new_workflow_file).open("w"))
    img_bytes = self.infer.local(new_workflow_file)

    return Response(img_bytes, media_type="image/jpeg")

Manipulando Diferentes Modelos Flux ComfyUI

Permita a troca entre diferentes modelos Flux:

@modal.fastapi_endpoint(method="POST")
def model_selection_api(self, item: Dict):
    from fastapi import Response

    workflow_data = json.loads(
        (Path(__file__).parent / "workflow_api.json").read_text()
    )

    # Inserir o prompt
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # Selecionar modelo se especificado
    if "model" in item:
        if item["model"] == "flux-schnell":
            workflow_data["2"]["inputs"]["ckpt_name"] = "flux1-schnell-fp8.safetensors"
        elif item["model"] == "flux-turbo":
            workflow_data["2"]["inputs"]["ckpt_name"] = "flux1-turbo-fp8.safetensors"

    # Gerar ID exclusivo
    client_id = uuid.uuid4().hex
    workflow_data["9"]["inputs"]["filename_prefix"] = client_id

    # Salvar e processar fluxo de trabalho
    new_workflow_file = f"{client_id}.json"
    json.dump(workflow_data, Path(new_workflow_file).open("w"))
    img_bytes = self.infer.local(new_workflow_file)

    return Response(img_bytes, media_type="image/jpeg")

Monitorando Sua API Flux ComfyUI

Implementar monitoramento adequado é essencial para implantações em produção:

@app.cls(
    # Outros parâmetros
    monitor_agent=modal.MonitorAgent(),
)
class ComfyUI:
    # Implementação existente

    def log_request(self, prompt, model, processing_time):
        # Implementação de logs personalizada
        print(f"Imagem gerada para o prompt: '{prompt}' usando o modelo {model} em {processing_time:.2f}s")

Conclusão: Liberando o Poder do Flux ComfyUI como API

Seguindo este tutorial, você aprendeu como transformar o ComfyUI com modelos Flux em um serviço de API escalável e pronto para produção. Essa abordagem oferece várias vantagens:

Escalabilidade: Lidar com múltiplos pedidos simultaneamente com escalonamento automático de containers
Flexibilidade: Customizar fluxos de trabalho com base nas necessidades dos usuários
Integração: Conectar suas capacidades de geração de imagens com outras aplicações
Custo-efetividade: Pague apenas pela computação que você usa

A combinação da infraestrutura serverless do Modal, o poderoso sistema de fluxo de trabalho do ComfyUI e a geração de imagens de última geração do Flux cria uma solução robusta para uma ampla gama de aplicações criativas e empresariais.

Seja construindo uma ferramenta criativa, um sistema de visualização de e-commerce ou uma plataforma de geração de conteúdo, executar o Flux no ComfyUI como uma API fornece a base para experiências visuais inovadoras alimentadas por IA.

botão