FluxワークフローComfyUIをAPIとして実行する方法

Fluxは、ブラックフォレストラボによって開発された最新鋭のテキストから画像へのモデルで、高度な画像生成機能を備えています。一方で、ComfyUIは、ノードベースのワークフローシステムを持つ強力でモジュラーな拡散モデルインターフェースです。これらを組み合わせることで、高品質な画像生成のための堅牢なソリューションをプログラム的にアクセスできるようになります。

Fluxモデルの力とComfyUIの柔軟性を組み合わせることで、AI画像生成ワークフローを大幅に向上させることができます。このチュートリアルでは、FluxをAPIとしてComfyUI上で実行するプロセスをガイドし、自動化や他のシステムとの統合の可能性を開きます。

Apidogは、自動テスト、モックサーバー、詳細なドキュメント生成などの機能を持つ包括的なAPI開発エコシステムを一つの統合されたプラットフォームで提供しています。

Postmanとは異なり、Apidogは、特にComfyUIのような複雑なワークフローに最適化された組み込みのデバッグツールを使用したシームレスなAPIライフサイクル管理を提供します。その直感的なインターフェースにより、Flux ComfyUIエンドポイントのテストを大幅に効率化し、チームメンバーがAPI設定を共有・洗練するための共同作業機能を備えています。

API開発とテストに本気で取り組んでいるなら、Apidogのオールインワンのソリューションがワークフローを大幅に効率化します。

Flux ComfyUI環境の設定

API実装に入る前に、必要な環境を設定する必要があります。私たちは、GPUアクセスを持つサーバーレスアプリケーションの実行を容易にするクラウドプラットフォームであるModalを使用します。

Flux ComfyUIの前提条件

• Modalアカウント（modal.comで作成できます）
• Python 3.11以上
• ローカルマシンにインストールされたGit
• PythonとAPIの基本的な理解

Flux ComfyUIコンテナイメージの構築

最初のステップは、FluxをComfyUI上で実行するために必要なすべてのコンポーネントを含むコンテナイメージを作成することです。プロセスを分解してみましょう：

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

import modal

# 必要な依存関係を持つ基本イメージを作成
image = (
    modal.Image.debian_slim(python_version="3.11")
    .apt_install("git")  # ComfyUIのクローン用にgitをインストール
    .pip_install("fastapi[standard]==0.115.4")  # Web依存関係
    .pip_install("comfy-cli==1.3.5")  # ComfyUI CLIツール
    .run_commands(
        "comfy --skip-prompt install --nvidia --version 0.3.10"  # ComfyUIをインストール
    )
)

このコードは、Python 3.11を使用したDebian Slimに基づいたModalイメージを作成し、次にGit、FastAPI、comfy-cliをインストールします。その後、comfy-cliツールを使用してComfyUI自体をインストールします。

カスタムノードでFlux ComfyUIを強化する

追加機能を解除するために、ComfyUIインストールにカスタムノードを追加します：

image = (
    image.run_commands(
        "comfy node install was-node-suite-comfyui@1.0.2"  # WAS Node Suiteをインストール
    )
    # 必要に応じてさらにカスタムノードを追加
)

WAS Node Suiteは、画像生成と操作のための追加機能を提供し、Fluxモデルを使用する際に特に役立ちます。

Flux ComfyUIモデルのダウンロード設定

効率的なモデル管理はスムーズな体験のために重要です。Fluxモデルをダウンロードし、永続的なボリュームに保存するための機能を作成します：

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

    # モデルをComfyUIディレクトリにシンリンク
    subprocess.run(
        f"ln -s {flux_model} /root/comfy/ComfyUI/models/checkpoints/flux1-schnell-fp8.safetensors",
        shell=True,
        check=True,
    )

# モデルストレージのための永続的なボリュームを作成
vol = modal.Volume.from_name("hf-hub-cache", create_if_missing=True)

# モデルダウンロード機能をイメージに追加
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},
    )
)

# ワークフローのJSONをコンテナに追加
image = image.add_local_file(
    Path(__file__).parent / "workflow_api.json", "/root/workflow_api.json"
)

このコードは、Hugging FaceからFlux Schnellモデルをダウンロードし、ComfyUIモデルディレクトリへのシンリンクを作成する機能を定義しています。また、実行間のダウンロードしたモデルを永続化するためにModalボリュームを作成します。

インタラクティブなFlux ComfyUI開発

APIとして展開する前に、Flux ComfyUIワークフローをインタラクティブにテストすることが役立ちます：

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

@app.function(
    allow_concurrent_inputs=10,  # UIスタートアップのために必要
    max_containers=1,  # インタラクティブ使用のためにコンテナを1つに制限
    gpu="L40S",  # Fluxモデル推論のための強力なGPU
    volumes={"/cache": vol},  # キャッシュされたモデルをマウント
)
@modal.web_server(8000, startup_timeout=60)
def ui():
    subprocess.Popen(
        "comfy launch -- --listen 0.0.0.0 --port 8000",
        shell=True
    )

これをmodal serve your_file.pyで実行し、ブラウザでUIにアクセスしてワークフローを設計・テストしてからAPIを実装できます。

Flux ComfyUI APIの実装

さて、メインイベントです - ComfyUIをFluxモデル推論のためのAPIエンドポイントに変換します：

@app.cls(
    allow_concurrent_inputs=10,  # 複数の同時API呼び出しを許可
    scaledown_window=300,  # 使用後5分間コンテナを生かす
    gpu="L40S",
    volumes={"/cache": vol},
)
class ComfyUI:
    @modal.enter()
    def launch_comfy_background(self):
        # バックグラウンドでComfyUIサーバーを起動
        cmd = "comfy launch --background"
        subprocess.run(cmd, shell=True, check=True)

    @modal.method()
    def infer(self, workflow_path: str = "/root/workflow_api.json"):
        # ComfyUIサーバーのヘルスチェック
        self.poll_server_health()

        # comfy-cliでワークフローを実行
        cmd = f"comfy run --workflow {workflow_path} --wait --timeout 1200 --verbose"
        subprocess.run(cmd, shell=True, check=True)

        # 出力画像を見つける
        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"]

        # 画像をバイトとして返す
        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

        # ワークフローテンプレートをロード
        workflow_data = json.loads(
            (Path(__file__).parent / "workflow_api.json").read_text()
        )

        # プロンプトを挿入
        workflow_data["6"]["inputs"]["text"] = item["prompt"]

        # このリクエストのためのユニークIDを生成
        client_id = uuid.uuid4().hex
        workflow_data["9"]["inputs"]["filename_prefix"] = client_id

        # 修正されたワークフローを保存
        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")

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

        try:
            # サーバーが健康かチェック
            req = urllib.request.Request("http://127.0.0.1:8188/system_stats")
            urllib.request.urlopen(req, timeout=5)
            print("ComfyUIサーバーは健康です")
        except (socket.timeout, urllib.error.URLError) as e:
            # サーバーが健康でない場合はコンテナを停止
            print(f"サーバーのヘルスチェックが失敗しました: {str(e)}")
            modal.experimental.stop_fetching_inputs()
            raise Exception("ComfyUIサーバーが健康でないため、コンテナを停止します")

このクラスは、コアAPI機能を定義します：

1. launch_comfy_backgroundは、コンテナ起動時にComfyUIサーバーを起動します
2. inferはワークフローを実行し、生成された画像を返します
3. apiはプロンプトを受け入れ、ワークフローを修正して画像を返すFastAPIエンドポイントです
4. poll_server_healthはComfyUIサーバーが応答していることを確認します

Flux ComfyUI APIのテスト

APIをテストするには、シンプルなクライアントスクリプトを作成できます：

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

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", default="浮遊する島と神秘的な光を持つ超現実的な風景")
    parser.add_argument("--endpoint", default="https://your-modal-endpoint.modal.run")
    args = parser.parse_args()

    # APIにリクエストを送信
    response = requests.post(
        f"{args.endpoint}/api",
        json={"prompt": args.prompt}
    )

    # 応答を処理
    if response.status_code == 200:
        # 画像を保存して表示
        image = Image.open(io.BytesIO(response.content))
        image.save("flux_output.jpg")
        print(f"画像が生成され、flux_output.jpgとして保存されました")

        # 必要に応じて表示
        try:
            image.show()
        except:
            pass
    else:
        print(f"エラー: {response.status_code}, {response.text}")

if __name__ == "__main__":
    main()

Flux ComfyUIパフォーマンスの最適化

APIが応答性を維持し、コスト効果があるようにするため、以下の最適化を実施することを検討してください：

1. フラックス ComfyUI コールド スタートを早くするためのメモリ スナップショット

コールドスタートはサーバーレスアプリケーションのボトルネックになる可能性があります。Modalは、起動時間を大幅に短縮できるメモリ スナップショットを提供します：

@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:
    # クラスの残りの実装

2. リクエストバッチングによるFlux ComfyUIの効率向上

高スループットのシナリオでは、リクエストのバッチ処理を実装することで効率を改善できます：

@modal.method()
def batch_inference(self, prompts: list[str]):
    results = []
    for prompt in prompts:
        # 各プロンプト用にワークフローを作成
        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

        # ワークフローを保存して処理
        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

Flux ComfyUI APIの高度な機能

ワークフローパラメーターのカスタマイズ

プロンプトだけにとどまらず、追加のパラメーターを許可するようにAPIを拡張します：

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

    # プロンプトを挿入
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # 提供された場合は追加のパラメーターを設定
    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"]

    # ユニークIDを生成
    client_id = uuid.uuid4().hex
    workflow_data["9"]["inputs"]["filename_prefix"] = client_id

    # ワークフローを保存して処理
    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")

異なるFlux ComfyUIモデルの取り扱い

異なる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()
    )

    # プロンプトを挿入
    workflow_data["6"]["inputs"]["text"] = item["prompt"]

    # 指定された場合はモデルを選択
    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"

    # ユニークIDを生成
    client_id = uuid.uuid4().hex
    workflow_data["9"]["inputs"]["filename_prefix"] = client_id

    # ワークフローを保存して処理
    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")

Flux ComfyUI APIの監視

適切な監視を実施することは、プロダクションデプロイメントにとって不可欠です：

@app.cls(
    # 他のパラメーター
    monitor_agent=modal.MonitorAgent(),
)
class ComfyUI:
    # 既存の実装

    def log_request(self, prompt, model, processing_time):
        # カスタムロギング実装
        print(f"プロンプト: '{prompt}' に対して {model} モデルを使用して画像を生成しました。処理時間: {processing_time:.2f}s")

結論: Flux ComfyUIのAPIとしての力の解放

このチュートリアルを通じて、FluxモデルでComfyUIをスケーラブルでプロダクション準備が整ったAPIサービスに変換する方法を学びました。このアプローチは、いくつかの利点を提供します：

1. スケーラビリティ: 自動コンテナスケーリングによって、複数のリクエストを同時に処理します
2. 柔軟性: ユーザーのニーズに応じてワークフローをカスタマイズします
3. 統合: あなたの画像生成機能を他のアプリケーションと接続します
4. コスト効率: 使用したコンピューティングにだけ支払います

Modalのサーバーレスインフラストラクチャ、ComfyUIの強力なワークフローシステム、Fluxの最先端の画像生成を組み合わせることで、幅広い創造的およびビジネスアプリケーションに適した堅牢なソリューションが構築されます。

創造的なツール、eコマースビジュアライゼーションシステム、コンテンツ生成プラットフォームを構築しているかどうかにかかわらず、FluxをAPIとしてComfyUI上で実行することで、革新的なAI駆動のビジュアル体験の基盤が提供されます。