Blog/Developer Tools/

Batch Image Generation Pipeline for Agents

Step-by-step tutorial for building a batch image generation pipeline — agent takes product descriptions, generates variants, scores them, picks the best. Python code with parallel generation, error handling, and cost tracking.

Cover image for Batch Image Generation Pipeline for Agents

TL;DR — This tutorial builds a complete batch image generation pipeline: an agent takes 100 product descriptions, generates 5 image variants each (500 images), scores them with a vision model, and picks the best. Full Python code using the OpenAI SDK on SandBase, with parallel execution, error handling, retry logic, and cost tracking. Estimated cost: $7.50–$20.00 depending on model choice.

What we’re building

A production-grade image generation pipeline for e-commerce that:

  1. Takes product descriptions as input (name, category, key features)
  2. Generates 5 image variants per product using different prompts/angles
  3. Scores each variant using a vision model for quality and relevance
  4. Selects the best variant per product
  5. Tracks costs, handles failures, and logs everything

This pattern applies to any batch generation workflow — marketing campaigns, social media content calendars, catalog refreshes, A/B test asset creation.

For model selection guidance, see our best AI image generation APIs guide. For a similar pattern applied to video, see our video generation agent tutorial.

Architecture overview

┌─────────────────────────────────────────────────────────┐
│                    Batch Pipeline                         │
├─────────────────────────────────────────────────────────┤
│  Input: 100 product descriptions                         │
│    ↓                                                     │
│  Prompt Generator (5 prompts per product = 500 prompts) │
│    ↓                                                     │
│  Parallel Image Generation (batch of 10 concurrent)      │
│    ↓                                                     │
│  Quality Scorer (vision model evaluation)                │
│    ↓                                                     │
│  Selector (pick best per product)                        │
│    ↓                                                     │
│  Output: 100 best images + metadata                      │
└─────────────────────────────────────────────────────────┘

Prerequisites

pip install openai aiohttp asyncio aiofiles tenacity pydantic

Step 1: Define the data model

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class Product(BaseModel):
    id: str
    name: str
    category: str
    features: list[str]
    style_preference: Optional[str] = None

class ImageVariant(BaseModel):
    product_id: str
    variant_index: int
    prompt: str
    image_url: Optional[str] = None
    image_b64: Optional[str] = None
    score: Optional[float] = None
    error: Optional[str] = None
    latency_ms: int = 0
    cost: float = 0.0

class PipelineResult(BaseModel):
    product_id: str
    best_variant: Optional[ImageVariant] = None
    all_variants: list[ImageVariant] = []
    total_cost: float = 0.0
    total_time_ms: int = 0

Step 2: Prompt generation

Each product gets 5 different prompts to maximize variety:

class PromptGenerator:
    """Generate diverse prompts for each product."""
    
    ANGLES = [
        ("hero", "clean white background, studio lighting, centered product, "
                 "e-commerce hero shot, ultra-sharp detail"),
        ("lifestyle", "product in use, natural environment, lifestyle photography, "
                      "warm lighting, realistic setting"),
        ("detail", "macro close-up, showing texture and material quality, "
                   "shallow depth of field, studio lighting"),
        ("context", "product on a styled surface with complementary props, "
                    "flat lay or shelf arrangement, editorial style"),
        ("dramatic", "dramatic lighting, dark background with rim light, "
                     "premium feel, high contrast, luxury presentation"),
    ]
    
    def generate_prompts(self, product: Product) -> list[str]:
        """Generate 5 diverse prompts for one product."""
        prompts = []
        
        features_text = ", ".join(product.features[:3])
        
        for angle_name, angle_style in self.ANGLES:
            prompt = (
                f"{product.name}, {features_text}, "
                f"{angle_style}, "
                f"professional product photography, 8K quality"
            )
            
            if product.style_preference:
                prompt += f", {product.style_preference}"
            
            prompts.append(prompt)
        
        return prompts

Step 3: Parallel image generation with error handling

import asyncio
import time
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

class BatchImageGenerator:
    """Generate images in parallel with rate limiting and error handling."""
    
    def __init__(
        self,
        api_key: str,
        model: str = "bytedance/seedream/5.0/pro/fast",
        max_concurrent: int = 10,
        cost_per_image: float = 0.015
    ):
        self.api_key = api_key
        self.base_url = "https://api.sandbase.ai/v1"
        self.model = model
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.cost_per_image = cost_per_image
        self.total_cost = 0.0
        self.success_count = 0
        self.failure_count = 0
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30)
    )
    async def _generate_single(self, prompt: str) -> dict:
        """Generate a single image with retry logic (submit + poll)."""
        async with self.semaphore:
            start = time.time()
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            }
            async with aiohttp.ClientSession() as session:
                # Submit generation task
                async with session.post(
                    f"{self.base_url}/run",
                    headers=headers,
                    json={"model": self.model, "prompt": prompt},
                ) as resp:
                    submit = await resp.json()
                task_id = submit["id"]
                # Poll for completion
                poll_headers = {"Authorization": f"Bearer {self.api_key}"}
                while True:
                    async with session.get(
                        f"{self.base_url}/generations/{task_id}",
                        headers=poll_headers,
                    ) as resp:
                        result = await resp.json()
                    if result["status"] in ("completed", "failed", "timeout"):
                        break
                    await asyncio.sleep(2)
            latency = int((time.time() - start) * 1000)
            return {
                "url": result["outputs"][0]["url"],
                "latency_ms": latency
            }
    
    async def generate_variant(
        self, product_id: str, variant_index: int, prompt: str
    ) -> ImageVariant:
        """Generate one image variant, handling errors gracefully."""
        variant = ImageVariant(
            product_id=product_id,
            variant_index=variant_index,
            prompt=prompt
        )
        
        try:
            result = await self._generate_single(prompt)
            variant.image_url = result["url"]
            variant.latency_ms = result["latency_ms"]
            variant.cost = self.cost_per_image
            self.total_cost += self.cost_per_image
            self.success_count += 1
        except Exception as e:
            variant.error = str(e)
            self.failure_count += 1
        
        return variant
    
    async def generate_batch(
        self, products: list[Product], prompts_per_product: dict[str, list[str]]
    ) -> list[PipelineResult]:
        """Generate all variants for all products in parallel."""
        all_tasks = []
        
        for product in products:
            prompts = prompts_per_product[product.id]
            for i, prompt in enumerate(prompts):
                task = self.generate_variant(product.id, i, prompt)
                all_tasks.append((product.id, task))
        
        # Execute all tasks with concurrency control
        results_by_product: dict[str, list[ImageVariant]] = {}
        
        tasks = [task for _, task in all_tasks]
        product_ids = [pid for pid, _ in all_tasks]
        
        variants = await asyncio.gather(*tasks)
        
        for pid, variant in zip(product_ids, variants):
            if pid not in results_by_product:
                results_by_product[pid] = []
            results_by_product[pid].append(variant)
        
        # Build results
        pipeline_results = []
        for product in products:
            variants = results_by_product.get(product.id, [])
            result = PipelineResult(
                product_id=product.id,
                all_variants=variants,
                total_cost=sum(v.cost for v in variants),
                total_time_ms=max((v.latency_ms for v in variants), default=0)
            )
            pipeline_results.append(result)
        
        return pipeline_results
    
    def report(self):
        """Print generation statistics."""
        total = self.success_count + self.failure_count
        print(f"Generated: {self.success_count}/{total} images")
        print(f"Failed: {self.failure_count}/{total}")
        print(f"Total cost: ${self.total_cost:.2f}")
        print(f"Success rate: {self.success_count/max(total,1)*100:.1f}%")

Step 4: Quality scoring with vision model

from openai import AsyncOpenAI

class ImageScorer:
    """Score images using a vision model for quality and relevance."""
    
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(
            base_url="https://api.sandbase.ai/v1",
            api_key=api_key
        )
    
    async def score_variant(
        self, variant: ImageVariant, product: Product
    ) -> float:
        """Score an image variant 0-1 for quality and product relevance."""
        if not variant.image_url:
            return 0.0
        
        scoring_prompt = f"""Rate this product image on a scale of 0 to 100.

Product: {product.name}
Category: {product.category}
Key features that should be visible: {', '.join(product.features)}

Scoring criteria:
- Image quality and sharpness (25 points)
- Product visibility and focus (25 points)
- Professional composition (25 points)  
- Relevance to product description (25 points)

Return ONLY a number between 0 and 100."""

        try:
            response = await self.client.chat.completions.create(
                model="openai/gpt-4o-mini",  # Cost-effective for scoring
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": scoring_prompt},
                            {"type": "image_url", "image_url": {"url": variant.image_url}}
                        ]
                    }
                ],
                max_tokens=10
            )
            
            score_text = response.choices[0].message.content.strip()
            score = float(score_text) / 100.0
            return min(max(score, 0.0), 1.0)
        except Exception:
            return 0.5  # Default score on error
    
    async def score_all(
        self, results: list[PipelineResult], products: dict[str, Product]
    ) -> list[PipelineResult]:
        """Score all variants and select the best per product."""
        scoring_tasks = []
        
        for result in results:
            product = products[result.product_id]
            for variant in result.all_variants:
                if variant.image_url:
                    scoring_tasks.append(
                        self._score_and_assign(variant, product)
                    )
        
        await asyncio.gather(*scoring_tasks)
        
        # Select best variant per product
        for result in results:
            scored_variants = [v for v in result.all_variants if v.score is not None]
            if scored_variants:
                result.best_variant = max(scored_variants, key=lambda v: v.score)
        
        return results
    
    async def _score_and_assign(self, variant: ImageVariant, product: Product):
        variant.score = await self.score_variant(variant, product)

Step 5: Putting it all together

import json

async def run_batch_pipeline(
    products: list[Product],
    api_key: str,
    model: str = "bytedance/seedream/5.0/pro/fast",
    max_concurrent: int = 10,
):
    """Run the complete batch image generation pipeline."""
    
    print(f"Starting batch pipeline: {len(products)} products × 5 variants = "
          f"{len(products) * 5} images")
    print(f"Model: {model}")
    print(f"Concurrency: {max_concurrent}")
    print("=" * 60)
    
    # Phase 1: Generate prompts
    prompt_gen = PromptGenerator()
    prompts_per_product = {}
    for product in products:
        prompts_per_product[product.id] = prompt_gen.generate_prompts(product)
    
    print(f"Phase 1: Generated {sum(len(p) for p in prompts_per_product.values())} prompts")
    
    # Phase 2: Generate images
    generator = BatchImageGenerator(
        api_key=api_key,
        model=model,
        max_concurrent=max_concurrent
    )
    
    results = await generator.generate_batch(products, prompts_per_product)
    print(f"Phase 2: Image generation complete")
    generator.report()
    
    # Phase 3: Score and select
    scorer = ImageScorer(api_key=api_key)
    products_dict = {p.id: p for p in products}
    results = await scorer.score_all(results, products_dict)
    
    # Phase 4: Report
    selected_count = sum(1 for r in results if r.best_variant)
    total_cost = sum(r.total_cost for r in results)
    
    print(f"\nPhase 3: Scoring complete")
    print(f"Products with selected images: {selected_count}/{len(products)}")
    print(f"Total pipeline cost: ${total_cost:.2f}")
    
    # Save results
    output = []
    for result in results:
        output.append({
            "product_id": result.product_id,
            "best_image": result.best_variant.image_url if result.best_variant else None,
            "best_score": result.best_variant.score if result.best_variant else None,
            "variants_generated": len(result.all_variants),
            "cost": result.total_cost
        })
    
    with open("pipeline_results.json", "w") as f:
        json.dump(output, f, indent=2)
    
    return results


# Example usage
if __name__ == "__main__":
    # Sample products
    products = [
        Product(
            id=f"prod_{i:03d}",
            name=f"Premium Wireless Earbuds Model {i}",
            category="electronics",
            features=["noise cancellation", "30h battery", "IPX5 waterproof"],
            style_preference="modern minimalist"
        )
        for i in range(100)
    ]
    
    results = asyncio.run(run_batch_pipeline(
        products=products,
        api_key="your-sandbase-api-key",
        model="bytedance/seedream/5.0/pro/fast",
        max_concurrent=10
    ))

Cost estimation

By model choice (100 products × 5 variants = 500 images)

ModelPer image500 images+ Scoring (500 calls)Total
Nano Banana Lite$0.008$4.00~$1.50$5.50
Nano Banana 2 Lite$0.01$5.00~$1.50$6.50
Seedream Fast$0.015$7.50~$1.50$9.00
Qwen-Image-3$0.03$15.00~$1.50$16.50
Seedream Pro$0.04$20.00~$1.50$21.50

Time estimation (sequential vs parallel)

ModelSequential (500 images)Parallel (10 concurrent)
Nano Banana Lite~17 min~2 min
Seedream Fast~25 min~3 min
Qwen-Image-3~58 min~6 min
Seedream Pro~83 min~9 min

With 10 concurrent requests, the entire 500-image pipeline completes in 2–9 minutes depending on model choice.

Advanced: hybrid model strategy

Use different models for exploration vs. final selection:

async def hybrid_pipeline(products: list[Product], api_key: str):
    """Two-phase pipeline: Fast exploration, Pro refinement."""
    
    # Phase A: Generate all 500 with Fast ($7.50)
    fast_results = await run_batch_pipeline(
        products=products,
        api_key=api_key,
        model="bytedance/seedream/5.0/pro/fast",
        max_concurrent=10
    )
    
    # Phase B: Regenerate top variant prompts with Pro ($4.00 for 100 images)
    pro_generator = BatchImageGenerator(
        api_key=api_key,
        model="bytedance/seedream/5.0/pro",
        max_concurrent=5
    )
    
    for result in fast_results:
        if result.best_variant:
            # Regenerate the winning prompt with Pro quality
            pro_variant = await pro_generator.generate_variant(
                result.product_id,
                variant_index=99,  # Mark as Pro version
                prompt=result.best_variant.prompt
            )
            if pro_variant.image_url:
                result.best_variant = pro_variant
    
    # Total cost: $7.50 (Fast exploration) + $4.00 (Pro finals) = $11.50
    # vs. all-Pro: $21.50 — saving 46% while getting Pro quality on finals
    return fast_results

Error handling patterns

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APITimeoutError, APIConnectionError

# Retry configuration for production
RETRY_CONFIG = {
    "stop": stop_after_attempt(3),
    "wait": wait_exponential(multiplier=1, min=2, max=60),
    "retry": retry_if_exception_type((RateLimitError, APITimeoutError, APIConnectionError)),
}

# Dead letter queue for persistent failures
class DeadLetterQueue:
    def __init__(self):
        self.failed_items = []
    
    def add(self, product_id: str, prompt: str, error: str):
        self.failed_items.append({
            "product_id": product_id,
            "prompt": prompt,
            "error": error,
            "timestamp": datetime.now().isoformat()
        })
    
    def retry_all(self):
        """Retry all failed items in next pipeline run."""
        items = self.failed_items.copy()
        self.failed_items.clear()
        return items

Monitoring and observability

class PipelineMetrics:
    """Track pipeline performance metrics."""
    
    def __init__(self):
        self.start_time = time.time()
        self.images_generated = 0
        self.images_failed = 0
        self.total_cost = 0.0
        self.latencies: list[int] = []
    
    def record_success(self, latency_ms: int, cost: float):
        self.images_generated += 1
        self.total_cost += cost
        self.latencies.append(latency_ms)
    
    def record_failure(self):
        self.images_failed += 1
    
    def summary(self) -> dict:
        elapsed = time.time() - self.start_time
        return {
            "duration_seconds": round(elapsed, 1),
            "images_generated": self.images_generated,
            "images_failed": self.images_failed,
            "success_rate": f"{self.images_generated / max(self.images_generated + self.images_failed, 1) * 100:.1f}%",
            "total_cost": f"${self.total_cost:.2f}",
            "avg_latency_ms": round(sum(self.latencies) / max(len(self.latencies), 1)),
            "p95_latency_ms": sorted(self.latencies)[int(len(self.latencies) * 0.95)] if self.latencies else 0,
            "images_per_second": round(self.images_generated / max(elapsed, 1), 2),
        }

Conclusion

Batch image generation for agents follows a consistent pattern: generate many, score automatically, keep the best. The key engineering decisions are:

  1. Model choice — Seedream Fast offers the best quality-per-dollar for batch operations
  2. Concurrency — 10 parallel requests is a good default; adjust based on rate limits
  3. Error handling — Retry with exponential backoff, dead letter queue for persistent failures
  4. Cost control — Track spend per product, set budgets, use hybrid strategies

The complete pipeline handles 100 products (500 images) in under 10 minutes for $9–$21 depending on quality requirements. Scale linearly to thousands of products by increasing concurrency or running multiple batches.