프로그램 방식 이미지 생성: 확장 가능한 배치 파이프라인 구축 방법

대부분의 이미지 생성 튜토리얼은 이미지 한 장에서 멈춥니다. curl 명령어 하나와 예쁜 결과물을 보여주고 끝이죠. "고양이 사진 하나 만들어줘"에는 충분할지 모릅니다. 하지만 실제 프로젝트에서 500장의 이미지가 필요할 때는 전혀 쓸모가 없습니다.

프로그램 방식 이미지 생성 — 코드로 사람의 개입 없이 대규모로 이미지를 생성하는 것 — 은 전혀 다른 기술입니다. 이 가이드는 전체 파이프라인을 다룹니다: 대규모 프롬프트 엔지니어링, 배치 처리, 오류 처리, 비동기 처리, 출력 관리, 그리고 프로덕션 시스템 통합까지.

프로덕션 이미지 파이프라인의 3계층

모든 프로덕션 이미지 파이프라인에는 세 개의 계층이 있습니다:

대부분의 개발자는 계층 1만 생각합니다. 하지만 계층 2와 3이 파이프라인의 성패를 좌우합니다.

계층 1: 대규모 프롬프트 엔지니어링

이미지 한 장을 만들 때는 정성껏 완벽한 프롬프트를 다듬을 수 있습니다. 500장을 만들어야 할 때는 프롬프트 시스템이 필요합니다.

템플릿 접근법

# prompts.py — 중앙 집중식 프롬프트 템플릿
from dataclasses import dataclass
from typing import Optional

@dataclass
class ImageJob:
    template: str
    params: dict
    output_path: str
    model: str = "nano-banana-2"
    
PROMPT_TEMPLATES = {
    "product_hero": "이커머스 제품 사진: {product_name}, {color}, 스튜디오 조명, 흰색 배경, 1024x1024, 상업용 사진",
    "blog_hero": "블로그 헤더 일러스트: {topic}, {style} 스타일, {mood} 분위기, 1200x630, 에디토리얼",
    "social_post": "소셜 미디어 비주얼: {subject}, {platform} 형식, {vibe} 무드, {dimensions}",
}

def build_prompt(template_key: str, **params) -> str:
    return PROMPT_TEMPLATES[template_key].format(**params)

스케일업 패턴

# CSV에서 100개의 제품 사진 생성
import csv, subprocess, json
from concurrent.futures import ThreadPoolExecutor, as_completed

def generate_single(job: ImageJob) -> dict:
    prompt = build_prompt(job.template, **job.params)
    
    result = subprocess.run([
        "anycap", "image", "generate",
        "--prompt", prompt,
        "--model", job.model,
        "--output-format", "json",
        "-o", job.output_path
    ], capture_output=True, text=True)
    
    return {
        "output_path": job.output_path,
        "success": result.returncode == 0,
        "data": json.loads(result.stdout) if result.returncode == 0 else None,
        "error": result.stderr if result.returncode != 0 else None
    }

# 데이터에서 작업 목록 생성
jobs = []
with open("products.csv") as f:
    for row in csv.DictReader(f):
        jobs.append(ImageJob(
            template="product_hero",
            params={"product_name": row["name"], "color": row["color"]},
            output_path=f"output/{row['sku']}.png"
        ))

# 동시성 제어로 실행
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(generate_single, job): job for job in jobs}
    for future in as_completed(futures):
        result = future.result()
        status = "✅" if result["success"] else "❌"
        print(f"{status} {result['output_path']}")

계층 2: 오케스트레이션 — 모두가 잊는 부분

생성은 쉽습니다. 대규모에서 안정적으로 만드는 것이 진짜 엔지니어링입니다.

패턴 1: 비동기 배치 처리

대규모 배치(100장 이상)에서는 블로킹을 피하기 위해 비동기 모드를 사용하세요:

# 배치 작업 제출
anycap image generate \
  --prompt "$(python build-prompts.py --csv products.csv)" \
  --model nano-banana-2 \
  --async \
  --batch-size 20 \
  --webhook "https://your-app.com/webhooks/images" \
  -o output/products/

웹훅이 완료되는 대로 결과를 받습니다. 폴링도, 타임아웃 문제도 없습니다.

패턴 2: 지수 백오프 재시도

import time, random

def generate_with_retry(job: ImageJob, max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        result = generate_single(job)
        if result["success"]:
            return result
        
        if attempt < max_retries - 1:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"재시도 {attempt + 1}/{max_retries} ({job.output_path}) {wait:.1f}초 후")
            time.sleep(wait)
    
    return result  # 마지막 실패 반환

패턴 3: 큐 기반 아키텍처

프로덕션 시스템에는 제대로 된 큐를 사용하세요:

# Redis 기반 간단한 작업 큐
import redis, json

r = redis.Redis()

def enqueue_job(job: ImageJob):
    r.lpush("image_jobs", json.dumps({
        "template": job.template,
        "params": job.params,
        "output_path": job.output_path,
        "model": job.model,
    }))

def worker_loop():
    while True:
        _, job_data = r.brpop("image_jobs")
        job = json.loads(job_data)
        result = generate_single(ImageJob(**job))
        
        if result["success"]:
            r.lpush("image_results", json.dumps(result))
        else:
            r.lpush("image_failures", json.dumps(result))

계층 3: 통합 — 이미지를 필요한 곳으로 전달

S3에 업로드

import boto3

s3 = boto3.client("s3")

def upload_to_s3(local_path: str, bucket: str, key: str) -> str:
    s3.upload_file(local_path, bucket, key, ExtraArgs={
        "ContentType": "image/png",
        "CacheControl": "public, max-age=31536000",
    })
    return f"https://{bucket}.s3.amazonaws.com/{key}"

CMS에 게시

import requests

def update_cms_product_image(sku: str, image_url: str):
    requests.patch(
        f"https://cms.example.com/api/products/{sku}",
        headers={"Authorization": "Bearer $CMS_TOKEN"},
        json={"image_url": image_url}
    )

팀에 알림

def notify_slack(message: str):
    requests.post(
        "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
        json={"text": message}
    )

전체 파이프라인 스크립트

#!/usr/bin/env python3
"""production-pipeline.py — 전체 이미지 생성 파이프라인"""

import csv, subprocess, json, time, random, sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
import boto3, requests

# --- 설정 ---
S3_BUCKET = "my-assets"
SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK"
MAX_WORKERS = 4
MAX_RETRIES = 3

PROMPTS = {
    "product": "이커머스 사진: {name}, {color}, 스튜디오, 흰색 배경, 1024x1024",
    "lifestyle": "라이프스타일 사진: {name}, {color}, {scene}, 자연광, 1024x1024",
}

@dataclass
class Job:
    template: str
    params: dict
    output: str
    model: str = "nano-banana-2"

def generate(job: Job) -> dict:
    prompt = PROMPTS[job.template].format(**job.params)
    for attempt in range(MAX_RETRIES):
        result = subprocess.run([
            "anycap", "image", "generate",
            "--prompt", prompt, "--model", job.model,
            "--output-format", "json", "-o", job.output
        ], capture_output=True, text=True)
        if result.returncode == 0:
            data = json.loads(result.stdout)
            return {"path": job.output, "url": data.get("image_url"), "success": True}
        if attempt < MAX_RETRIES - 1:
            time.sleep((2 ** attempt) + random.uniform(0, 1))
    return {"path": job.output, "success": False, "error": result.stderr}

def upload(path: str) -> str:
    key = path.replace("output/", "")
    s3 = boto3.client("s3")
    s3.upload_file(path, S3_BUCKET, key, ExtraArgs={"ContentType": "image/png"})
    return f"https://{S3_BUCKET}.s3.amazonaws.com/{key}"

def notify(text: str):
    requests.post(SLACK_WEBHOOK, json={"text": text})

def run_pipeline(csv_path: str):
    jobs = []
    with open(csv_path) as f:
        for row in csv.DictReader(f):
            jobs.append(Job("product", {"name": row["name"], "color": row["color"]}, f"output/{row['sku']}.png"))
    
    notify(f"🚀 파이프라인 시작: {len(jobs)}장 이미지")
    
    results = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(generate, job): job for job in jobs}
        for future in as_completed(futures):
            result = future.result()
            if result["success"]:
                result["s3_url"] = upload(result["path"])
                results.append(result)
    
    success = len(results)
    failed = len(jobs) - success
    notify(f"{'✅' if failed == 0 else '⚠️'} 파이프라인 완료: {success}/{len(jobs)}장 성공. {failed}장 실패.")
    return results

if __name__ == "__main__":
    run_pipeline(sys.argv[1])

파이프라인에 적합한 모델 선택

규모별 비용

최종 업데이트: 2026년 5월