Pembuatan Gambar Terprogram: Cara Membangun Pipeline Batch yang Skalabel

Sebagian besar tutorial pembuatan gambar berhenti di satu gambar. Mereka menunjukkan perintah curl, hasil yang cantik, dan menganggapnya selesai. Itu cukup untuk "buat gambar kucing." Tapi tidak berguna saat Anda butuh 500 gambar untuk proyek nyata.

Pembuatan gambar terprogram — menghasilkan gambar dalam skala besar, dari kode, tanpa campur tangan manusia — adalah keterampilan yang berbeda. Panduan ini mencakup pipeline lengkap: prompt engineering dalam skala besar, pemrosesan batch, penanganan error, pemrosesan async, manajemen output, dan integrasi ke sistem produksi.

Tiga Lapisan Pipeline Gambar Produksi

Setiap pipeline gambar produksi memiliki tiga lapisan:

Sebagian besar developer hanya memikirkan Lapisan 1. Tapi Lapisan 2 dan 3 adalah tempat pipeline berhasil atau gagal.

Lapisan 1: Prompt Engineering dalam Skala Besar

Saat Anda membuat satu gambar, Anda bisa merangkai prompt sempurna dengan hati-hati. Saat Anda membuat 500, Anda butuh sistem prompt.

Pendekatan Template

# prompts.py — Template prompt terpusat
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": "Foto produk e-commerce: {product_name}, {color}, pencahayaan studio, latar putih, 1024x1024, fotografi komersial",
    "blog_hero": "Ilustrasi header blog: {topic}, gaya {style}, suasana {mood}, 1200x630, editorial",
    "social_post": "Visual media sosial: {subject}, format {platform}, estetika {vibe}, {dimensions}",
}

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

Pola Scale-Up

# Generate 100 foto produk dari CSV
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
    }

# Bangun daftar job dari data
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"
        ))

# Eksekusi dengan kontrol konkurensi
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']}")

Lapisan 2: Orchestration — Bagian yang Sering Dilupakan

Pembuatan itu mudah. Membuatnya andal dalam skala besar adalah rekayasa yang sesungguhnya.

Pola 1: Async Batch Processing

Untuk batch besar (100+ gambar), gunakan mode async untuk menghindari pemblokiran:

# Kirim job batch
anycap image generate \
  --prompt "$(python build-prompts.py --csv products.csv)" \
  --model nano-banana-2 \
  --async \
  --batch-size 20 \
  --webhook "https://aplikasi-anda.com/webhooks/images" \
  -o output/products/

Webhook Anda menerima hasil begitu selesai. Tidak perlu polling. Tidak ada masalah timeout.

Pola 2: Retry dengan Exponential Backoff

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"Retry {attempt + 1}/{max_retries} untuk {job.output_path} dalam {wait:.1f}dtk")
            time.sleep(wait)
    
    return result  # Kembalikan kegagalan terakhir

Pola 3: Arsitektur Berbasis Antrean

Untuk sistem produksi, gunakan antrean yang proper:

# Job queue sederhana berbasis 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))

Lapisan 3: Integration — Mengantarkan Gambar ke Tujuan

Upload ke 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}"

Post ke 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}
    )

Notifikasi ke Tim

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

Skrip Pipeline Lengkap

#!/usr/bin/env python3
"""production-pipeline.py — Pipeline pembuatan gambar lengkap"""

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

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

PROMPTS = {
    "product": "Foto e-commerce: {name}, {color}, studio, latar putih, 1024x1024",
    "lifestyle": "Foto gaya hidup: {name}, {color}, {scene}, cahaya alami, 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"🚀 Pipeline dimulai: {len(jobs)} gambar")
    
    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 '⚠️'} Pipeline selesai: {success}/{len(jobs)} gambar. {failed} gagal.")
    return results

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

Memilih Model yang Tepat untuk Pipeline Anda

Biaya dalam Skala Besar

Terakhir diperbarui: Mei 2026.