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.

SandBase Team By SandBase Team
#image-generation #developer-tools #tutorial #agent-workflows #python #batch-processing

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.

You May Also Like

image-to-video-api-product-demo-generator
Tutorials

Image to Video API: Build a Product Demo Generator

Build a Python script that turns product images into polished video demos using image-to-video APIs. Comparison of Kling, MiniMax, Luma with costs and code.

Cover image for Cloudsway vs Exa: Which Search API for Your Agent
Developer Tools

Cloudsway vs Exa: Which Search API for Your Agent

Detailed comparison of Cloudsway Search and Exa Search for AI agent workflows — ranked results vs semantic matching, when to use each, and how to combine them for maximum coverage.

Cover image for Cloudsway Search: Ranked Results + Summaries
Developer Tools

Cloudsway Search: Ranked Results + Summaries

Deep dive into Cloudsway Search on SandBase — ranked web search results with dynamic summaries for AI agents. Architecture, use cases, API usage, and comparison with semantic search approaches.