Comment exécuter un flux de travail Flux ComfyUI en tant qu'API

Flux est un modèle de conversion texte-en-image de pointe développé par Black Forest Labs, doté de capacités avancées de génération d'images. ComfyUI, quant à lui, est une interface de modèle de diffusion puissante et modulaire avec un système de flux de travail basé sur des nœuds. Lorsqu'ils sont combinés, ils créent une solution robuste pour la génération d'images de haute qualité accessible par programmation.

Combiner la puissance des modèles Flux avec la polyvalence de ComfyUI peut considérablement améliorer votre flux de travail de génération d'images par IA. Ce tutoriel vous guidera tout au long du processus d'exécution de Flux sur ComfyUI en tant qu'API, ouvrant des possibilités d'automatisation et d'intégration avec d'autres systèmes.

💡

Avant de plonger dans la configuration de notre API Flux ComfyUI, parlons des outils de test. Lorsque vous travaillez avec des API, il est essentiel de disposer d'une plateforme de test fiable, et Apidog se distingue comme la principale alternative à Postman dans ce domaine.

Apidog offre un écosystème complet de développement d'API avec des fonctionnalités telles que les tests automatisés, les serveurs simulés et la génération de documentation détaillée, le tout sur une plateforme unifiée.

button

Contrairement à Postman, Apidog offre une gestion transparente du cycle de vie des API avec des outils de débogage intégrés spécialement optimisés pour les flux de travail complexes comme ceux de ComfyUI. Son interface intuitive rend les tests de vos points de terminaison Flux ComfyUI beaucoup plus efficaces, avec des fonctionnalités collaboratives qui permettent aux membres de l'équipe de partager et d'affiner les configurations d'API ensemble.

Si vous êtes sérieux au sujet du développement et des tests d'API, la solution tout-en-un d'Apidog rationalisera considérablement votre flux de travail.

button

Configuration de votre environnement Flux ComfyUI

Avant de plonger dans l'implémentation de l'API, nous devons configurer l'environnement nécessaire. Nous utiliserons Modal, une plateforme cloud qui facilite l'exécution d'applications sans serveur avec accès GPU.

Conditions préalables pour Flux ComfyUI

Un compte Modal (créez-en un sur modal.com)
Python 3.11 ou version ultérieure
Git installé sur votre machine locale
Compréhension de base de Python et des concepts d'API

Construction de l'image conteneur Flux ComfyUI

Notre première étape consiste à créer une image conteneur qui comprend tous les composants nécessaires à l'exécution de Flux sur ComfyUI. Décomposons le processus :

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

import modal

# Create a base image with necessary dependencies
image = (
    modal.Image.debian_slim(python_version="3.11")
    .apt_install("git")  # Install git for cloning ComfyUI
    .pip_install("fastapi[standard]==0.115.4")  # Web dependencies
    .pip_install("comfy-cli==1.3.5")  # ComfyUI CLI tool
    .run_commands(
        "comfy --skip-prompt install --nvidia --version 0.3.10"  # Install ComfyUI
    )
)

Ce code crée une image Modal basée sur Debian Slim avec Python 3.11, puis installe Git, FastAPI et comfy-cli. L'outil comfy-cli est ensuite utilisé pour installer ComfyUI lui-même.

Améliorer votre Flux ComfyUI avec des nœuds personnalisés

Pour débloquer des fonctionnalités supplémentaires, nous ajouterons des nœuds personnalisés à notre installation ComfyUI :

image = (
    image.run_commands(
        "comfy node install was-node-suite-comfyui@1.0.2"  # Install WAS Node Suite
    )
    # Add more custom nodes as needed
)

La suite de nœuds WAS fournit des fonctionnalités supplémentaires pour la génération et la manipulation d'images, ce qui peut être particulièrement utile lorsque vous travaillez avec des modèles Flux.

Configuration du téléchargement du modèle Flux ComfyUI

Une gestion efficace des modèles est cruciale pour une expérience fluide. Nous allons créer une fonction pour télécharger le modèle Flux et le stocker dans un volume persistant :

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",
    )

    # Symlink the model to the ComfyUI directory
    subprocess.run(
        f"ln -s {flux_model} /root/comfy/ComfyUI/models/checkpoints/flux1-schnell-fp8.safetensors",
        shell=True,
        check=True,
    )

# Create a persistent volume for model storage
vol = modal.Volume.from_name("hf-hub-cache", create_if_missing=True)

# Add the model download function to our image
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},
    )
)

# Add our workflow JSON to the container
image = image.add_local_file(
    Path(__file__).parent / "workflow_api.json", "/root/workflow_api.json"
)

Ce code définit une fonction pour télécharger le modèle Flux Schnell depuis Hugging Face et créer un lien symbolique vers le répertoire des modèles ComfyUI. Nous créons également un volume Modal pour conserver les modèles téléchargés entre les exécutions.

Développement interactif de Flux ComfyUI

Avant de déployer en tant qu'API, il est utile de tester de manière interactive votre flux de travail Flux ComfyUI :

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

@app.function(
    allow_concurrent_inputs=10,  # Needed for UI startup
    max_containers=1,  # Limit to one container for interactive use
    gpu="L40S",  # Powerful GPU for Flux model inference
    volumes={"/cache": vol},  # Mount our cached models
)
@modal.web_server(8000, startup_timeout=60)
def ui():
    subprocess.Popen(
        "comfy launch -- --listen 0.0.0.0 --port 8000",
        shell=True
    )

Vous pouvez l'exécuter avec modal serve your_file.py et accéder à l'interface utilisateur dans votre navigateur pour concevoir et tester vos flux de travail avant d'implémenter l'API.

Implémentation de l'API Flux ComfyUI

Maintenant, l'événement principal : transformer ComfyUI en un point de terminaison d'API pour l'inférence de modèle Flux :

@app.cls(
    allow_concurrent_inputs=10,  # Allow multiple concurrent API calls
    scaledown_window=300,  # Keep containers alive for 5 minutes after use
    gpu="L40S",
    volumes={"/cache": vol},
)
class ComfyUI:
    @modal.enter()
    def launch_comfy_background(self):
        # Start ComfyUI server in the background
        cmd = "comfy launch --background"
        subprocess.run(cmd, shell=True, check=True)

    @modal.method()
    def infer(self, workflow_path: str = "/root/workflow_api.json"):
        # Check ComfyUI server health
        self.poll_server_health()

        # Run workflow with comfy-cli
        cmd = f"comfy run --workflow {workflow_path} --wait --timeout 1200 --verbose"
        subprocess.run(cmd, shell=True, check=True)

        # Find output image
        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]["inputs"]["filename_prefix"]

        # Return image as 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

        # Load workflow template
        workflow_data = json.loads(
            (Path(__file__).parent / "workflow_api.json").read_text()
        )

        # Insert the prompt
        workflow_data["6"]["inputs"]["text"] = item["prompt"]

        # Generate unique ID for this request
        client_id = uuid.uuid4().hex
        workflow_data["9"]["inputs"]["filename_prefix"] = client_id

        # Save modified workflow
        new_workflow_file = f"{client_id}.json"
        json.dump(workflow_data, Path(new_workflow_file).open("w"))

        # Run inference
        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:
            # Check if server is healthy
            req = urllib.request.Request("<http://127.0.0.1:8188/system_stats>")
            urllib.request.urlopen(req, timeout=5)
            print("ComfyUI server is healthy")
        except (socket.timeout, urllib.error.URLError) as e:
            # Stop container if server is unhealthy
            print(f"Server health check failed: {str(e)}")
            modal.experimental.stop_fetching_inputs()
            raise Exception("ComfyUI server is not healthy, stopping container")

Cette classe définit la fonctionnalité principale de l'API :

launch_comfy_background démarre le serveur ComfyUI lorsque le conteneur se lance
infer exécute un flux de travail et renvoie l'image générée
api est le point de terminaison FastAPI qui accepte une invite, modifie le flux de travail et renvoie l'image
poll_server_health garantit que le serveur ComfyUI est réactif

Test de votre API Flux ComfyUI

Pour tester votre API, vous pouvez créer un script client simple :

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

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", default="A surreal landscape with floating islands and ethereal light")
    parser.add_argument("--endpoint", default="<https://your-modal-endpoint.modal.run>")
    args = parser.parse_args()

    # Send request to the API
    response = requests.post(
        f"{args.endpoint}/api",
        json={"prompt": args.prompt}
    )

    # Handle the response
    if response.status_code == 200:
        # Save and display the image
        image = Image.open(io.BytesIO(response.content))
        image.save("flux_output.jpg")
        print(f"Image generated and saved as flux_output.jpg")

        # Optional: display the image if in a suitable environment
        try:
            image.show()
        except:
            pass
    else:
        print(f"Error: {response.status_code}, {response.text}")

if __name__ == "__main__":
    main()

Optimisation des performances de Flux ComfyUI

Pour vous assurer que votre API reste réactive et rentable, envisagez de mettre en œuvre les optimisations suivantes :

1. Prise d'instantanés de la mémoire pour des démarrages à froid Flux ComfyUI plus rapides

Les démarrages à froid peuvent être un goulot d'étranglement pour les applications sans serveur. Modal propose la prise d'instantanés de la mémoire qui peut réduire considérablement les temps de démarrage :

@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:
    # Rest of the class implementation

2. Traitement par lots des requêtes pour Flux ComfyUI

Pour les scénarios à haut débit, la mise en œuvre du traitement par lots des requêtes peut améliorer l'efficacité :

@modal.method()
def batch_inference(self, prompts: list[str]):
    results = []
    for prompt in prompts:
        # Create workflow for each 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

        # Save and process workflow
        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

Fonctionnalités avancées de l'API Flux ComfyUI

Personnalisation des paramètres du flux de travail

Étendez votre API pour autoriser des paramètres supplémentaires au-delà de l'invite :

@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()
    )

    # Insert the prompt
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # Optional: Set additional parameters if provided
    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"]

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

    # Save and process workflow
    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")

Gestion de différents modèles Flux ComfyUI

Autoriser le basculement entre différents modèles 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()
    )

    # Insert the prompt
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # Select model if specified
    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"

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

    # Save and process workflow
    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")

Surveillance de votre API Flux ComfyUI

La mise en œuvre d'une surveillance appropriée est essentielle pour les déploiements en production :

@app.cls(
    # Other parameters
    monitor_agent=modal.MonitorAgent(),
)
class ComfyUI:
    # Existing implementation

    def log_request(self, prompt, model, processing_time):
        # Custom logging implementation
        print(f"Generated image for prompt: '{prompt}' using {model} model in {processing_time:.2f}s")

Conclusion : Libérer la puissance de Flux ComfyUI en tant qu'API

En suivant ce tutoriel, vous avez appris à transformer ComfyUI avec les modèles Flux en un service d'API évolutif et prêt pour la production. Cette approche offre plusieurs avantages :

Évolutivité : gérez plusieurs requêtes simultanément avec la mise à l'échelle automatique des conteneurs
Flexibilité : personnalisez les flux de travail en fonction des exigences des utilisateurs
Intégration : connectez vos capacités de génération d'images à d'autres applications
Rentabilité : ne payez que pour le calcul que vous utilisez

La combinaison de l'infrastructure sans serveur de Modal, du puissant système de flux de travail de ComfyUI et de la génération d'images de pointe de Flux crée une solution robuste pour un large éventail d'applications créatives et commerciales.

Que vous construisiez un outil créatif, un système de visualisation de commerce électronique ou une plateforme de génération de contenu, l'exécution de Flux sur ComfyUI en tant qu'API fournit la base d'expériences visuelles innovantes basées sur l'IA.

button