Flux는 Black Forest Labs에서 개발한 최첨단 텍스트-이미지 모델로, 고급 이미지 생성 기능을 제공합니다. 반면 ComfyUI는 노드 기반 워크플로우 시스템을 갖춘 강력하고 모듈식의 확산 모델 인터페이스입니다. 두 가지를 결합하면 프로그래밍적으로 접근할 수 있는 고품질 이미지 생성을 위한 견고한 솔루션이 만들어집니다.
Flux 모델의 힘과 ComfyUI의 다재다능함을 결합하면 AI 이미지 생성 워크플로우를 크게 향상시킬 수 있습니다. 이 튜토리얼은 API로서 ComfyUI에서 Flux를 실행하는 과정을 안내할 것이며, 자동화 및 다른 시스템과의 통합 가능성을 열어줍니다.
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 컨테이너 이미지 만들기
첫 번째 단계는 ComfyUI에서 Flux를 실행하는 데 필요한 모든 구성 요소를 포함하는 컨테이너 이미지를 만드는 것입니다. 프로세스를 살펴보겠습니다:
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") # 웹 종속성
.pip_install("comfy-cli==1.3.5") # ComfyUI CLI 도구
.run_commands(
"comfy --skip-prompt install --nvidia --version 0.3.10" # ComfyUI 설치
)
)
이 코드는 Python 3.11을 기반으로 하는 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 노드 스위트 설치
)
# 필요에 따라 추가 맞춤형 노드 추가
)
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, # 인터랙티브 사용을 위해 하나의 컨테이너로 제한
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 기능을 정의합니다:
launch_comfy_background는 컨테이너가 시작될 때 ComfyUI 서버를 시작합니다.infer는 워크플로를 실행하고 생성된 이미지를 반환합니다.api는 프롬프트를 수신하고, 워크플로를 수정하며 이미지를 반환하는 FastAPI 엔드포인트입니다.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. 빠른 Flux 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의 힘을 활용하기
이 튜토리얼을 통해 ComfyUI와 Flux 모델을 스케일 가능하고 프로덕션 준비된 API 서비스로 변환하는 방법을 배웠습니다. 이 접근 방식은 여러 가지 이점을 제공합니다:
- 확장성: 자동 컨테이너 스케일링으로 여러 요청을 동시 처리
- 유연성: 사용자 요구에 따라 워크플로우 맞춤화
- 통합: 이미지 생성 기능을 다른 응용 프로그램과 연결
- 비용 효율성: 사용한 컴퓨트에 대해서만 지불
Modal의 서버리스 인프라, ComfyUI의 강력한 워크플로 시스템, Flux의 최첨단 이미지 생성이 결합되어 다양한 창의적 및 비즈니스 애플리케이션을 위한 견고한 솔루션을 만들어냅니다.
창의적인 도구, 전자상거래 시각화 시스템, 콘텐츠 생성 플랫폼을 구축하든, API로서 ComfyUI에서 Flux를 실행하는 것은 혁신적인 AI 기반 시각적 경험을 위한 기반이 됩니다.