FLUX 1.1 Pro API Python 튜토리얼: 5분 안에 이미지 생성 완료

핵심 수치 3가지:

• 생성 속도: FLUX 1.0 Pro 대비 6배 빠름 (Black Forest Labs 공식 발표)
• API 비용: Replicate 기준 약 $0.04/이미지 (1024×1024 기준)
• ELO 벤치마크: 출시 당시 Ideogram 1.0, DALL-E 3 대비 상위 랭크 기록

이 튜토리얼은 FLUX 1.1 Pro API를 Python으로 연동하는 실제 코드를 다룬다. Replicate API와 BFL(Black Forest Labs) 직접 API 두 가지 경로를 모두 커버한다. 코드 블록은 복사 후 즉시 실행 가능하다.

---

목차

1. Prerequisites: 계정, 라이브러리, 환경 설정
2. Authentication & 초기 설정
3. 기본 구현: 첫 번째 이미지 생성
4. 프로덕션용 구현: 에러 처리 포함
5. API 파라미터 레퍼런스
6. 에러 코드 및 해결 방법
7. 비용 및 성능 비교 테이블
8. 한계점 및 주의사항

---

1. Prerequisites: 계정, 라이브러리, 환경 설정 {#prerequisites}

필요한 계정

경로 A — Replicate (권장: 초기 테스트용)

• replicate.com 계정 생성
• 신용카드 등록 후 API token 발급 (r8_ prefix로 시작)
• 신규 가입 시 $5 무료 크레딧 제공 (약 125장 생성 가능)

경로 B — BFL Direct API (권장: 프로덕션)

• api.bfl.ml 계정 생성
• API key 발급
• Pay-as-you-go 요금제 (최소 충전 필요)

Python 버전 및 라이브러리 설치

# Python 3.8 이상 필요 (3.10+ 권장)
python --version

# 가상환경 생성 (선택이 아니라 필수 — 의존성 충돌 방지)
python -m venv flux_env
source flux_env/bin/activate  # Windows: flux_env\Scripts\activate

# 필수 라이브러리 설치
pip install replicate requests pillow python-dotenv

# 버전 고정 (재현 가능한 환경을 위해)
pip freeze > requirements.txt

환경 변수 설정

.env 파일을 프로젝트 루트에 생성:

# .env 파일 내용
REPLICATE_API_TOKEN=r8_your_token_here
BFL_API_KEY=your_bfl_key_here

# .env 파일은 절대 git에 커밋하지 말 것
echo ".env" >> .gitignore

---

2. Authentication & 초기 설정 {#authentication}

두 API 모두 Bearer token 방식을 사용한다. Replicate는 공식 Python 클라이언트가 있어 설정이 간단하다. BFL Direct API는 requests 라이브러리로 직접 HTTP 호출한다.

# config.py — 프로젝트 전체에서 재사용하는 설정 모듈
import os
from dotenv import load_dotenv

# .env 파일 로드 (프로젝트 루트 기준)
load_dotenv()

# 환경 변수 검증 — 키가 없으면 조용히 실패하는 대신 즉시 에러를 낸다
REPLICATE_API_TOKEN = os.getenv("REPLICATE_API_TOKEN")
BFL_API_KEY = os.getenv("BFL_API_KEY")

if not REPLICATE_API_TOKEN and not BFL_API_KEY:
    raise EnvironmentError(
        "API key가 설정되지 않았습니다. "
        ".env 파일에 REPLICATE_API_TOKEN 또는 BFL_API_KEY를 설정하세요."
    )

# Replicate 클라이언트는 환경 변수를 자동으로 읽는다
# REPLICATE_API_TOKEN이 환경에 있으면 별도 초기화 불필요
import replicate  # 이 시점에서 토큰이 자동 로드됨

print("설정 완료. API 클라이언트 준비됨.")

설정 검증:

python config.py
# 출력: 설정 완료. API 클라이언트 준비됨.

---

3. 기본 구현: 첫 번째 이미지 생성 {#basic-implementation}

경로 A: Replicate API

# generate_basic.py — Replicate를 통한 FLUX 1.1 Pro 기본 이미지 생성
import replicate
import requests
import os
from pathlib import Path

def generate_image_replicate(
    prompt: str,
    output_path: str = "output.png",
    width: int = 1024,
    height: int = 1024,
    num_inference_steps: int = 25,
    guidance_scale: float = 3.5,
) -> str:
    """
    Replicate API를 통해 FLUX 1.1 Pro로 이미지를 생성한다.
    
    반환값: 저장된 파일 경로
    """
    
    # FLUX 1.1 Pro의 Replicate 모델 ID
    # "latest" 태그 대신 특정 버전을 고정하면 재현성이 보장된다
    model_id = "black-forest-labs/flux-1.1-pro"
    
    print(f"이미지 생성 시작: {prompt[:50]}...")
    
    # API 호출 — run()은 동기 방식으로 완료될 때까지 대기한다
    output = replicate.run(
        model_id,
        input={
            "prompt": prompt,
            "width": width,           # 픽셀 단위, 256-1440 범위
            "height": height,         # 픽셀 단위, 256-1440 범위
            "num_inference_steps": num_inference_steps,  # 높을수록 품질↑ 속도↓
            "guidance_scale": guidance_scale,  # 프롬프트 충실도 조절
            "output_format": "png",   # "png" 또는 "jpg"
            "output_quality": 100,    # jpg 선택 시 1-100
        }
    )
    
    # output은 FileOutput 객체 — URL이나 파일 스트림을 담고 있다
    # 리스트로 반환되는 경우도 있으므로 타입 체크
    if isinstance(output, list):
        image_url = output[0]
    else:
        image_url = output
    
    # 이미지 다운로드 및 로컬 저장
    response = requests.get(str(image_url), timeout=30)
    response.raise_for_status()  # HTTP 에러 시 즉시 예외 발생
    
    # 출력 디렉토리가 없으면 생성
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    
    with open(output_path, "wb") as f:
        f.write(response.content)
    
    file_size_kb = os.path.getsize(output_path) / 1024
    print(f"저장 완료: {output_path} ({file_size_kb:.1f} KB)")
    
    return output_path


# 실행 예시
if __name__ == "__main__":
    result = generate_image_replicate(
        prompt="A photorealistic portrait of an astronaut on Mars, golden hour lighting, 8k detail",
        output_path="outputs/astronaut_mars.png",
        width=1024,
        height=1024,
    )
    print(f"완료: {result}")

경로 B: BFL Direct API (비동기 폴링 방식)

BFL Direct API는 Replicate와 달리 비동기 작업 방식을 사용한다. 요청 → 작업 ID 수신 → 폴링(polling)으로 결과 확인하는 흐름이다.

# generate_bfl.py — BFL Direct API를 통한 FLUX 1.1 Pro 이미지 생성
import requests
import time
import os
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()
BFL_API_KEY = os.getenv("BFL_API_KEY")

BFL_BASE_URL = "https://api.bfl.ml"
FLUX_11_PRO_ENDPOINT = f"{BFL_BASE_URL}/v1/flux-pro-1.1"
RESULT_ENDPOINT = f"{BFL_BASE_URL}/v1/get_result"


def generate_image_bfl(
    prompt: str,
    output_path: str = "output.png",
    width: int = 1024,
    height: int = 1024,
    steps: int = 25,
    guidance: float = 3.0,
    safety_tolerance: int = 2,
    poll_interval: float = 1.5,  # 초 단위 — 너무 짧으면 rate limit에 걸린다
    max_wait: int = 120,          # 최대 대기 시간(초)
) -> str:
    """
    BFL Direct API로 FLUX 1.1 Pro 이미지를 생성한다.
    비동기 방식이므로 작업 ID를 받고 완료될 때까지 폴링한다.
    """
    
    headers = {
        "x-key": BFL_API_KEY,          # BFL은 x-key 헤더 사용
        "Content-Type": "application/json",
    }
    
    payload = {
        "prompt": prompt,
        "width": width,
        "height": height,
        "steps": steps,
        "guidance": guidance,
        "safety_tolerance": safety_tolerance,  # 0(엄격)-6(관대)
        "output_format": "png",
    }
    
    # Step 1: 생성 작업 제출
    print(f"작업 제출 중...")
    response = requests.post(FLUX_11_PRO_ENDPOINT, json=payload, headers=headers)
    response.raise_for_status()
    
    task_data = response.json()
    task_id = task_data.get("id")
    
    if not task_id:
        raise ValueError(f"task_id를 받지 못했습니다. 응답: {task_data}")
    
    print(f"작업 ID: {task_id}")
    
    # Step 2: 결과 폴링
    elapsed = 0.0
    while elapsed < max_wait:
        time.sleep(poll_interval)
        elapsed += poll_interval
        
        result_response = requests.get(
            RESULT_ENDPOINT,
            params={"id": task_id},
            headers=headers,
        )
        result_response.raise_for_status()
        result = result_response.json()
        
        status = result.get("status")
        print(f"  상태: {status} ({elapsed:.0f}초 경과)")
        
        if status == "Ready":
            # 이미지 URL 추출 — BFL API의 응답 구조
            image_url = result["result"]["sample"]
            break
        elif status in ("Error", "Content Moderated", "Request Moderated"):
            raise RuntimeError(f"생성 실패. 상태: {status}, 응답: {result}")
        # "Pending" 또는 "Processing" 상태면 계속 대기
    else:
        raise TimeoutError(f"{max_wait}초 이내에 생성이 완료되지 않았습니다.")
    
    # Step 3: 이미지 다운로드
    img_response = requests.get(image_url, timeout=30)
    img_response.raise_for_status()
    
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "wb") as f:
        f.write(img_response.content)
    
    print(f"저장 완료: {output_path}")
    return output_path


if __name__ == "__main__":
    generate_image_bfl(
        prompt="A serene Japanese garden at dawn, cherry blossoms, mist, photorealistic",
        output_path="outputs/japanese_garden.png",
    )

---

4. 프로덕션용 구현: 에러 처리 + 배치 생성 {#production-implementation}

실제 서비스에서는 단건 생성보다 배치 처리, 재시도 로직, 결과 로깅이 필요하다.

# flux_client.py — 프로덕션 환경에서 사용할 FLUX 1.1 Pro 클라이언트
import replicate
import requests
import time
import logging
import json
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

# 로깅 설정 — print 대신 logging을 사용해야 레벨 제어와 파일 출력이 가능하다
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("flux_generation.log"),  # 로그 파일도 동시에 기록
    ],
)
logger = logging.getLogger(__name__)


@dataclass
class GenerationRequest:
    """이미지 생성 요청 파라미터를 타입 안전하게 관리"""
    prompt: str
    width: int = 1024
    height: int = 1024
    num_inference_steps: int = 25
    guidance_scale: float = 3.5
    output_format: str = "png"
    output_quality: int = 100
    seed: Optional[int] = None  # None이면 랜덤, 정수 지정 시 재현 가능


@dataclass
class GenerationResult:
    """생성 결과와 메타데이터를 함께 보관"""
    success: bool
    file_path: Optional[str] = None
    error_message: Optional[str] = None
    generation_time_seconds: float = 0.0
    request_params: dict = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


class FluxProClient:
    """
    FLUX 1.1 Pro API 클라이언트.
    재시도 로직, 에러 처리, 결과 로깅을 포함한다.
    """
    
    MODEL_ID = "black-forest-labs/flux-1.1-pro"
    
    def __init__(
        self,
        output_dir: str = "outputs",
        max_retries: int = 3,
        retry_delay: float = 2.0,  # 지수 백오프의 기본 딜레이(초)
    ):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    
    def generate(
        self,
        request: GenerationRequest,
        filename: Optional[str] = None,
    ) -> GenerationResult:
        """
        단건 이미지 생성. 실패 시 max_retries 횟수만큼 재시도한다.
        """
        if filename is None:
            # 타임스탬프 기반 파일명 — 덮어쓰기 방지
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"flux_{ts}.{request.output_format}"
        
        output_path = self.output_dir / filename
        
        # 요청 파라미터 구성
        input_params = {
            "prompt": request.prompt,
            "width": request.width,
            "height": request.height,
            "num_inference_steps": request.num_inference_steps,
            "guidance_scale": request.guidance_scale,
            "output_format": request.output_format,
            "output_quality": request.output_quality,
        }
        
        # seed가 있으면 추가 — 없으면 API 기본값(랜덤) 사용
        if request.seed is not None:
            input_params["seed"] = request.seed
        
        start_time = time.time()
        last_error = None
        
        for attempt in range(1, self.max_retries + 1):
            try:
                logger.info(f"생성 시도 {attempt}/{self.max_retries}: {request.prompt[:60]}...")
                
                output = replicate.run(self.MODEL_ID, input=input_params)
                
                # 결과 URL 추출
                image_url = output[0] if isinstance(output, list) else output
                
                # 이미지 다운로드
                response = requests.get(str(image_url), timeout=30)
                response.raise_for_status()
                
                with open(output_path, "wb") as f:
                    f.write(response.content)
                
                elapsed = time.time() - start_time
                logger.info(f"생성 완료: {output_path} ({elapsed:.2f}초)")
                
                # 성공 결과 반환
                result = GenerationResult(
                    success=True,
                    file_path=str(output_path),
                    generation_time_seconds=elapsed,
                    request_params=input_params,
                )
                
                # 메타데이터 JSON으로 저장 (감사 추적용)
                self._save_metadata(result, str(output_path))
                
                return result
            
            except replicate.exceptions.ReplicateError as e:
                last_error = f"Replicate API 에러 (시도 {attempt}): {e}"
                logger.warning(last_error)
                
                # 401/403은 재시도해도 의미 없음 — 즉시 포기
                if "401" in str(e) or "403" in str(e):
                    break
                
            except requests.exceptions.RequestException as e:
                last_error = f"네트워크 에러 (시도 {attempt}): {e}"
                logger.warning(last_error)
            
            # 지수 백오프: 2초, 4초, 8초...
            if attempt < self.max_retries:
                wait = self.retry_delay * (2 ** (attempt - 1))
                logger.info(f"{wait:.0f}초 후 재시도...")
                time.sleep(wait)
        
        elapsed = time.time() - start_time
        return GenerationResult(
            success=False,
            error_message=last_error,
            generation_time_seconds=elapsed,
            request_params=input_params,
        )
    
    def batch_generate(
        self,
        requests_list: list[GenerationRequest],
        delay_between: float = 0.5,  # 요청 사이 딜레이 — rate limit 방지
    ) -> list[GenerationResult]:
        """여러 이미지를 순차적으로 생성. 병렬 처리는 rate limit 문제로 권장하지 않는다."""
        results = []
        total = len(requests_list)
        
        for i, req in enumerate(requests_list, 1):
            logger.info(f"배치 진행: {i}/{total}")
            result = self.generate(req)
            results.append(result)
            
            if i < total:
                time.sleep(delay_between)  # 마지막 항목 이후에는 딜레이 불필요
        
        # 배치 요약 출력
        success_count = sum(1 for r in results if r.success)
        logger.info(f"배치 완료: {success_count}/{total} 성공")
        
        return results
    
    def _save_metadata(self, result: GenerationResult, image_path: str) -> None:
        """이미지와 함께 JSON 메타데이터 파일을 저장한다. 나중에 프롬프트 추적 용이."""
        meta_path = image_path.replace(f".{result.request_params.get('output_format', 'png')}", "_meta.json")
        with open(meta_path, "w", encoding="utf-8") as f:
            json.dump({
                "timestamp": result.timestamp,
                "generation_time_seconds": result.generation_time_seconds,
                "params": result.request_params,
            }, f, ensure_ascii=False, indent=2)


# 사용 예시
if __name__ == "__main__":
    client = FluxProClient(output_dir="outputs", max_retries=3)
    
    # 단건 생성
    req = GenerationRequest(
        prompt="Macro photography of a dewdrop on a spider web, sunrise backlight, f/2.8",
        width=1024,
        height=1024,
        seed=42,  # 재현 가능한 결과
    )
    result = client.generate(req)
    
    if result.success:
        print(f"성공: {result.file_path} ({result.generation_time_seconds:.2f}초)")
    else:
        print(f"실패: {result.error_message}")
    
    # 배치 생성
    batch_requests = [
        GenerationRequest(prompt="A neon-lit cyberpunk street in Tokyo, rain reflections"),
        GenerationRequest(prompt="Minimalist product photo of a white ceramic mug, studio light"),
        GenerationRequest(prompt="Oil painting of a medieval castle at sunset, dramatic clouds"),
    ]
    
    batch_results = client.batch_generate(batch_requests)

---

5. API 파라미터 레퍼런스 {#api-parameters}

파라미터	타입	기본값	유효 범위	영향
prompt	string	없음 (필수)	최대 ~2000자	생성 이미지 내용 전체
width	integer	1024	256–1440 (64 배수)	출력 이미지 가로 픽셀
height	integer	1024	256–1440 (64 배수)	출력 이미지 세로 픽셀
num_inference_steps	integer	25	1–50	높을수록 품질↑, 속도↓
guidance_scale	float	3.5	1.0–20.0	높을수록 프롬프트 충실도↑, 다양성↓
seed	integer	None (랜덤)	0–2147483647	동일 seed → 동일 결과 (재현성)
output_format	string	"png"	"png", "jpg", "webp"	파일 형식 및 압축 방식
output_quality	integer	100	1–100	JPG/WEBP 압축 품질 (PNG는 무시)
safety_tolerance	integer	2	1–6	컨텐츠 필터 강도 (1=엄격, 6=관대)
prompt_upsampling	boolean	false	true/false	true 시 API가 프롬프트를 자동 확장

주의사항:

• width와 height는 반드시 64의 배수여야 한다. 그렇지 않으면 422 에러 반환.
• guidance_scale 7.0 초과 시 이미지가 과포화(oversaturation)되는 경향이 있다.
• safety_tolerance 값은 계정 등급에 따라 허용 범위가 다를 수 있다.

---

6. 에러 코드 및 해결 방법 {#error-handling}

HTTP 코드	에러 메시지	원인	해결 방법
401	Invalid token	API 토큰 오류 또는 만료	토큰 재발급, .env 파일 확인
402	Insufficient balance	크레딧 잔액 부족	대시보드에서 충전
422	Validation error	잘못된 파라미터 값	width/height 64 배수 확인, 범위 벗어난 값 수정
429	Rate limit exceeded	초당 요청 초과	요청 사이 time.sleep() 추가, exponential backoff 적용
500	Internal server error	서버 측 문제	3회까지 재시도, 지속 시 상태 페이지 확인
503	Service unavailable	서버 과부하 또는 점검	status.replicate.com 확인 후 대기

Content Moderated (BFL API): safety_tolerance를 높이거나 프롬프트에서 민감한 단어 제거.

output 값이 None 반환: 모델 로드 실패 또는 타임아웃. Replicate의 경우 replicate.run() 대신 replicate.models.predictions.create()로 비동기 처리 전환 고려.

---

7. 비용 및 성능 비교 테이블 {#cost-performance}

API 경로별 비용 비교

경로	이미지당 비용	평균 생성 시간	최소 충전액	무료 크레딧
Replicate	~$0.04	8–15초	없음 (카드 등록)	$5 (신규)
BFL Direct API	~$0.04	5–12초	별도 확인 필요	없음
Together AI (FLUX 1.1 Pro)	~$0.04	10–20초	없음	$5 (신규)

가격은 2024년 기준이며 변동될 수 있음

해상도별 비용 추정 (Replicate 기준)

해상도	픽셀 수	예상 비용	생성 시간
512×512	262,144	~$0.03	5–8초
1024×1024	1,048,576	~$0.04	8–15초
1024×1440	1,474,560	~$0.05	12–20초
1440×1440	2,073,600	~$0.06	15–25초

월간 사용량 시뮬레이션

사용 시나리오	일일 생성 수	월간 예상 비용
개인 프로젝트	10장	~$12
소규모 서비스	100장	~$120
중간 규모 서비스	1,000장	~$1,200
대규모 서비스	10,000장	~$12,000

대규모 사용의 경우 BFL Direct API 또는 Together AI를 통한 볼륨 할인 협의를 권장한다.

---

8. 한계점 및 주의사항 {#limitations}

FLUX 1.1 Pro를 사용하지 말아야 할 경우:

• 실시간 응답이 필요한 경우: 최소 5초 이상 생성 시간이 필요하므로 즉각적인 응답이 요구되는 UI에는 부적합하다.
• 텍스트 렌더링이 중요한 경우: 이미지 내 정확한 텍스트 생성은 FLUX 계열 모델 전반의 약점이다. 텍스트 포함 이미지는 후처리(PIL, ImageMagick 등)로 별도 합성을 권장한다.
• 초당 수십 건 이상 처리 요구: 단일 Replicate 계정의 rate limit이 있으므로 대용량 배치는 여러 API 키를 분산하거나 Replicate Deployments 기능 사용을 검토해야 한다.
• 엄격한 저작권 준수가 필요한 경우: 생성 이미지의 저작권 관련 법적 사항은 국가별로 다르며, Black Forest Labs의 이용약관을 직접 확인해야 한다.

기술적 제약:

• 최대 해상도 1440×1440 (2MP)으로 인쇄용 고해상도(300 DPI 이상)에는 upscaling 후처리 필요
• seed 고정 시에도 모델 버전 업데이트가 발생하면 동일한 출력이 보장되지 않는다

---

결론

이 튜토리얼의 코드 블록은 Replicate API와 BFL Direct API 두 경로 모두 프로덕션 환경에서 즉시 사용 가능하다. 테스트 단계에서는 Replicate의 무료 크레딧을 활용하고, 서비스 규모가 커지면 BFL Direct API로 전환하는 것이 비용 효율적이다. 에러 처리와 재시도 로직은 선택이 아니라 API 연동의 기본값으로 처음부터 코드에 포함시켜야 한다.

참고: 여러 AI 모델을 하나의 파이프라인에서 사용한다면, AtlasCloud는 Kling, Flux, Seedance, Claude, GPT 등 300개 이상의 모델에 단일 API로 접근할 수 있습니다. API 키 하나로 모든 모델 사용 가능. 신규 사용자는 첫 충전 시 25% 보너스(최대 $100).