So führen Sie einen Flux-Workflow in ComfyUI als API aus

Flux ist ein hochmodernes Text-zu-Bild-Modell, das von Black Forest Labs entwickelt wurde und fortschrittliche Bilderzeugungsfähigkeiten bietet. ComfyUI hingegen ist eine leistungsstarke und modulare Diffusionsmodell-Oberfläche mit einem knotenbasierten Workflow-System. In Kombination bilden sie eine robuste Lösung für die Erzeugung hochwertiger Bilder, auf die programmatisch zugegriffen werden kann.

Die Kombination der Leistung von Flux-Modellen mit der Vielseitigkeit von ComfyUI kann Ihren KI-Bilderzeugungs-Workflow erheblich verbessern. Dieses Tutorial führt Sie durch den Prozess der Ausführung von Flux auf ComfyUI als API und eröffnet Möglichkeiten für Automatisierung und Integration mit anderen Systemen.

💡

Bevor wir uns mit der Einrichtung unserer Flux ComfyUI API befassen, sprechen wir über Testtools. Bei der Arbeit mit APIs ist eine zuverlässige Testplattform unerlässlich, und Apidog sticht in diesem Bereich als die führende Postman-Alternative hervor.

Apidog bietet ein umfassendes API-Entwicklungsumfeld mit Funktionen wie automatisiertem Testen, Mock-Servern und detaillierter Dokumentationserstellung, alles in einer einheitlichen Plattform.

button

Im Gegensatz zu Postman bietet Apidog ein nahtloses API-Lifecycle-Management mit integrierten Debugging-Tools, die speziell für komplexe Workflows wie die in ComfyUI optimiert sind. Seine intuitive Benutzeroberfläche macht das Testen Ihrer Flux ComfyUI-Endpunkte deutlich effizienter, mit kollaborativen Funktionen, die es Teammitgliedern ermöglichen, API-Konfigurationen gemeinsam zu nutzen und zu verfeinern.

Wenn Sie es mit der API-Entwicklung und -Tests ernst meinen, wird die All-in-One-Lösung von Apidog Ihren Workflow erheblich optimieren.

button

Einrichten Ihrer Flux ComfyUI-Umgebung

Bevor wir uns mit der API-Implementierung befassen, müssen wir die erforderliche Umgebung einrichten. Wir verwenden Modal, eine Cloud-Plattform, die die Ausführung serverloser Anwendungen mit GPU-Zugriff vereinfacht.

Voraussetzungen für Flux ComfyUI

Ein Modal-Konto (erstellen Sie eines unter modal.com)
Python 3.11 oder höher
Git auf Ihrem lokalen Rechner installiert
Grundlegendes Verständnis von Python und API-Konzepten

Erstellen des Flux ComfyUI-Container-Images

Unser erster Schritt ist die Erstellung eines Container-Images, das alle notwendigen Komponenten für die Ausführung von Flux auf ComfyUI enthält. Lassen Sie uns den Prozess aufschlüsseln:

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

Dieser Code erstellt ein Modal-Image basierend auf Debian Slim mit Python 3.11 und installiert dann Git, FastAPI und comfy-cli. Das comfy-cli-Tool wird dann verwendet, um ComfyUI selbst zu installieren.

Erweitern Sie Ihr Flux ComfyUI mit benutzerdefinierten Knoten

Um zusätzliche Funktionen freizuschalten, fügen wir unserer ComfyUI-Installation benutzerdefinierte Knoten hinzu:

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

Die WAS Node Suite bietet zusätzliche Funktionen für die Bilderzeugung und -manipulation, was besonders nützlich sein kann, wenn Sie mit Flux-Modellen arbeiten.

Einrichten des Flux ComfyUI-Modelldownloads

Ein effizientes Modellmanagement ist entscheidend für ein reibungsloses Erlebnis. Wir erstellen eine Funktion, um das Flux-Modell herunterzuladen und in einem persistenten Volume zu speichern:

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

Dieser Code definiert eine Funktion zum Herunterladen des Flux Schnell-Modells von Hugging Face und zum Erstellen eines Symlinks zum ComfyUI-Modellverzeichnis. Wir erstellen auch ein Modal Volume, um die heruntergeladenen Modelle zwischen den Läufen beizubehalten.

Interaktive Flux ComfyUI-Entwicklung

Vor der Bereitstellung als API ist es nützlich, Ihren Flux ComfyUI-Workflow interaktiv zu testen:

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
    )

Sie können dies mit modal serve your_file.py ausführen und über Ihren Browser auf die Benutzeroberfläche zugreifen, um Ihre Workflows zu entwerfen und zu testen, bevor Sie die API implementieren.

Implementieren der Flux ComfyUI-API

Nun zum Hauptereignis – ComfyUI in einen API-Endpunkt für die Flux-Modell-Inferenz verwandeln:

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

Diese Klasse definiert die Kernfunktionalität der API:

launch_comfy_background startet den ComfyUI-Server, wenn der Container gestartet wird
infer führt einen Workflow aus und gibt das generierte Bild zurück
api ist der FastAPI-Endpunkt, der eine Eingabeaufforderung akzeptiert, den Workflow ändert und das Bild zurückgibt
poll_server_health stellt sicher, dass der ComfyUI-Server reagiert

Testen Ihrer Flux ComfyUI-API

Um Ihre API zu testen, können Sie ein einfaches Client-Skript erstellen:

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

Optimieren der Flux ComfyUI-Leistung

Um sicherzustellen, dass Ihre API reaktionsfähig und kostengünstig bleibt, sollten Sie die folgenden Optimierungen in Betracht ziehen:

1. Memory Snapshotting für schnellere Flux ComfyUI Cold Starts

Cold Starts können ein Engpass für serverlose Anwendungen sein. Modal bietet Memory Snapshotting, das die Startzeiten erheblich verkürzen kann:

@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. Request Batching für Flux ComfyUI

Für Szenarien mit hohem Durchsatz kann die Implementierung von Request Batching die Effizienz verbessern:

@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

Erweiterte Flux ComfyUI-API-Funktionen

Anpassen der Workflow-Parameter

Erweitern Sie Ihre API, um zusätzliche Parameter über die Eingabeaufforderung hinaus zuzulassen:

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

Umgang mit verschiedenen Flux ComfyUI-Modellen

Erlauben Sie das Umschalten zwischen verschiedenen Flux-Modellen:

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

Überwachen Ihrer Flux ComfyUI-API

Die Implementierung einer ordnungsgemäßen Überwachung ist für Produktionseinsätze unerlässlich:

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

Fazit: Die Leistungsfähigkeit von Flux ComfyUI als API freisetzen

Indem Sie diesem Tutorial folgen, haben Sie gelernt, wie Sie ComfyUI mit Flux-Modellen in einen skalierbaren, produktionsreifen API-Dienst verwandeln. Dieser Ansatz bietet mehrere Vorteile:

Skalierbarkeit: Verarbeiten Sie mehrere Anfragen gleichzeitig mit automatischer Containerskalierung
Flexibilität: Passen Sie Workflows basierend auf Benutzeranforderungen an
Integration: Verbinden Sie Ihre Bilderzeugungsfunktionen mit anderen Anwendungen
Kosteneffizienz: Bezahlen Sie nur für die Rechenleistung, die Sie nutzen

Die Kombination aus der serverlosen Infrastruktur von Modal, dem leistungsstarken Workflow-System von ComfyUI und der hochmodernen Bilderzeugung von Flux schafft eine robuste Lösung für eine Vielzahl von kreativen und geschäftlichen Anwendungen.

Egal, ob Sie ein Kreativ-Tool, ein E-Commerce-Visualisierungssystem oder eine Content-Erstellungsplattform erstellen, die Ausführung von Flux auf ComfyUI als API bietet die Grundlage für innovative, KI-gestützte visuelle Erlebnisse.

button