NanoBanana: Building Agentic Workflows

The Problem: Why Your Image Pipeline Might Be Burning Money

Imagine you run a blog that needs 50 new thumbnail images every day. You write a simple script that calls an image API, waits for the result, and saves the file. It works fine for the first 10 images. But when you scale to 200, 500, or 2,000 images per day, things fall apart.

Here’s what happens: each image takes about 47 seconds to generate. One in four requests fails. You don’t know why. You retry blindly, often hitting the same broken endpoint. Each failed attempt costs you money before you even know it failed.

At 2,000 images per day with a 23% failure rate, that’s 460 failed generations. That’s $175 in pure waste every day — plus the time you spend manually re-running them.

The real problem wasn’t the AI model. It was how we called it. We were treating an async-first API (one designed for background tasks) like a regular database call (where you wait for an instant answer).

Why this matters: If you’re building any pipeline that calls an external API — image generation, video processing, data enrichment — you’ll hit these same walls. The fix isn’t a better model. It’s a better architecture.

The Investigation: Measuring the Pain

We tracked every step of the pipeline with monitoring tools. Here’s what the numbers told us.

What each metric means:

• P50 (median): Half of all requests finish faster than this time. Half take longer.
• P95 (95th percentile): Only 5% of requests are slower than this. This shows your worst-case performance.
• Error rate: The percentage of requests that fail. Lower is better.
• Cost per success: The total money spent (including failed attempts) divided by successful images.

Latency by Step (P50 / P95)

Step	P50	P95	Notes
Auth handshake	1.2s	4.7s	JWT refresh on cold start
Image generation	18.4s	43.2s	Heavy model, queue wait
Post-processing	3.1s	8.9s	Resize, crop, format convert
Upload to CDN	0.8s	2.1s	S3 multipart
Total synchronous	23.5s	58.9s	Blocking the whole pipeline

Error Rates by Step

Step	Error Rate	Primary Cause
Auth	1.2%	Expired tokens
Generation	18.7%	Timeout (default 30s)
Post-processing	2.1%	Memory limit on large images
Upload	1.0%	Network blips
Overall	23.0%	Cumulative

Cost Per Successful Run

Component	Cost
API call (standard tier)	$0.12
Retry overhead (1.3 avg retries)	$0.16
Wasted compute on failures	$0.10
Total per success	$0.38

The numbers told a clear story: our synchronous approach (waiting for each request to finish before starting the next) was the bottleneck. We needed an async-first architecture (one that starts many tasks at once and checks on them later).

The Solution: Async-First Architecture with NanoBanana

NanoBanana’s API is built for async workflows. Instead of waiting for a response, it gives you a task_id right away. You check back later for the result. This is perfect for production pipelines — you can queue hundreds of tasks, check their status in batches, and handle each result as it finishes.

Here’s what each piece does:

• Content Scheduler: Decides what images to generate and when
• Pipeline Orchestrator: Manages the flow — submits tasks, checks status, handles failures
• NanoBanana API: The image generation service that runs in the background
• Checkpoint Store (SQL): Saves progress so you can restart if something crashes
• Result Processor: Handles the finished images (resize, crop, format)
• CDN + Cache: Stores and serves the final images quickly

Architecture Overview

Upload Request
  → Submit to NanoBanana (gets task_id instantly)
  → Poll for completion (check every few seconds)
  → Download result
  → Post-process (resize, crop, format)
  → Upload to CDN
  → Save checkpoint

Before: Synchronous (The Old Way)

import requests
from time import sleep

def generate_image(prompt: str, style: str) -> str:
    """Synchronous generation — blocks until complete or timeout.
    
    This function sends a request and waits. 
    Your program can't do anything else until it finishes.
    """
    url = "https://api.nanobanana.com/v1/generate"
    payload = {
        "prompt": prompt,          # What to generate
        "style": style,            # Art style (photorealistic, digital-art, etc.)
        "width": 1024,
        "height": 1024,
    }
    headers = {"Authorization": f"Bearer {API_KEY}"}

    try:
        # This line blocks — your program pauses here for 20-50 seconds
        resp = requests.post(url, json=payload, headers=headers, timeout=30)
        resp.raise_for_status()    # Raise error if status code isn't 200
        data = resp.json()
        return data["image_url"]
    except requests.Timeout:
        logger.error(f"Timeout generating image for prompt: {prompt[:50]}")
        raise
    except requests.HTTPError as e:
        logger.error(f"HTTP {e.response.status_code} for prompt: {prompt[:50]}")
        raise

# Usage — blocks the entire pipeline
# Each image waits for the previous one to finish
for prompt in prompts:
    try:
        url = generate_image(prompt, "photorealistic")
        upload_to_cdn(url)
    except Exception:
        retry_queue.put(prompt)

After: Async (The New Way)

import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import Optional
import uuid

@dataclass
class GenerationTask:
    """A single image generation job.
    
    Tracks everything: the prompt, the task ID from NanoBanana,
    current status, and any errors.
    """
    prompt: str
    style: str
    task_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    status: str = "pending"       # pending → running → completed/failed
    result_url: Optional[str] = None
    error: Optional[str] = None
    attempts: int = 0

class NanoBananaAsyncClient:
    """Async client for NanoBanana's task-based API.
    
    Instead of waiting for each request, it submits tasks
    and checks on them later. This lets you run many at once.
    """

    BASE_URL = "https://api.nanobanana.com/v1"

    def __init__(self, api_key: str, max_retries: int = 3, poll_interval: float = 1.0):
        self.api_key = api_key
        self.max_retries = max_retries    # How many times to retry on failure
        self.poll_interval = poll_interval  # How often to check task status
        self.session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        # Set up the HTTP session (like opening a connection pool)
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=aiohttp.ClientTimeout(total=120),
        )
        return self

    async def __aexit__(self, *args):
        # Clean up the session when done
        await self.session.close()

    async def submit_generation(self, prompt: str, style: str, **kwargs) -> str:
        """Submit a generation task and return the task_id.
        
        This returns instantly — the actual generation happens in the background.
        """
        url = f"{self.BASE_URL}/generate"
        payload = {
            "prompt": prompt,
            "style": style,
            "width": kwargs.get("width", 1024),
            "height": kwargs.get("height", 1024),
            "async": True,  # Key: returns immediately with task_id
        }

        async with self.session.post(url, json=payload) as resp:
            resp.raise_for_status()
            data = await resp.json()
            return data["task_id"]

    async def poll_task(self, task_id: str, timeout: float = 120.0) -> dict:
        """Poll for task completion with exponential backoff.
        
        Checks the task status, waiting longer between each check.
        Like checking your laundry — wait a bit, check, wait longer, check again.
        """
        start = asyncio.get_event_loop().time()
        backoff = self.poll_interval

        while True:
            elapsed = asyncio.get_event_loop().time() - start
            if elapsed > timeout:
                raise asyncio.TimeoutError(f"Task {task_id} timed out after {timeout}s")

            url = f"{self.BASE_URL}/tasks/{task_id}"
            async with self.session.get(url) as resp:
                resp.raise_for_status()
                data = await resp.json()

            if data["status"] == "completed":
                return data
            elif data["status"] == "failed":
                raise RuntimeError(f"Task {task_id} failed: {data.get('error', 'unknown')}")

            # Wait longer each time (1s, 1.5s, 2.25s, ...) — capped at 5s
            await asyncio.sleep(backoff)
            backoff = min(backoff * 1.5, 5.0)

    async def generate_with_retry(self, prompt: str, style: str, **kwargs) -> str:
        """Submit, poll, and retry on failure.
        
        One method that does it all: submit, wait, and retry if needed.
        """
        last_error = None
        for attempt in range(self.max_retries):
            try:
                task_id = await self.submit_generation(prompt, style, **kwargs)
                result = await self.poll_task(task_id)
                return result["image_url"]
            except (aiohttp.ClientError, asyncio.TimeoutError, RuntimeError) as e:
                last_error = e
                logger.warning(f"Attempt {attempt + 1} failed for prompt '{prompt[:50]}': {e}")
                # Wait longer between retries: 1s, 2s, 4s
                await asyncio.sleep(2 ** attempt)

        raise RuntimeError(f"All {self.max_retries} attempts failed. Last error: {last_error}")

# Usage — fully async, non-blocking
# All images are generated in parallel, not one at a time
async def process_batch(prompts: list[str], style: str):
    async with NanoBananaAsyncClient(API_KEY) as client:
        # Create a task for each prompt — all submitted at once
        tasks = [client.generate_with_retry(p, style) for p in prompts]
        # Wait for ALL of them to finish
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for prompt, result in zip(prompts, results):
            if isinstance(result, Exception):
                logger.error(f"Failed to generate for '{prompt[:50]}': {result}")
            else:
                await upload_to_cdn(result)

How to Use NanoBanana Effectively

Here’s a step-by-step guide to build a production-ready pipeline.

Step 1: Set up your project

# Create a new Python project
mkdir image-pipeline
cd image-pipeline
python -m venv venv
source venv/bin/activate

# Install dependencies
pip install aiohttp aiofiles

Step 2: Get your API key

Sign up at nanobanana.com, go to your dashboard, and create an API key. Save it as an environment variable:

export NANOBANANA_API_KEY=nb_key_abc123

Step 3: Build the orchestrator

Here’s a production-grade orchestrator with checkpointing (saves progress), rate limiting (controls how many tasks run at once), and monitoring (tracks what’s happening):

import asyncio
import json
import sqlite3
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Optional
import aiohttp
import aiofiles

class PipelinePhase(Enum):
    """Tracks where each task is in the pipeline."""
    SUBMITTING = "submitting"   # About to send to NanoBanana
    POLLING = "polling"         # Waiting for generation to finish
    PROCESSING = "processing"   # Resizing, cropping, formatting
    UPLOADING = "uploading"     # Sending to CDN
    COMPLETE = "complete"       # Done!
    FAILED = "failed"           # Something went wrong

@dataclass
class PipelineTask:
    """A single task moving through the pipeline."""
    id: str
    prompt: str
    style: str
    params: dict
    phase: PipelinePhase = PipelinePhase.SUBMITTING
    nano_task_id: Optional[str] = None
    result_url: Optional[str] = None
    error: Optional[str] = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    completed_at: Optional[datetime] = None
    attempts: int = 0

class CheckpointStore:
    """Saves progress to a local database.
    
    If your script crashes, you can restart from where you left off
    instead of starting over. Like saving a video game.
    """

    def __init__(self, db_path: str = "pipeline_checkpoints.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        # Create the database table if it doesn't exist
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS checkpoints (
                    task_id TEXT PRIMARY KEY,
                    prompt TEXT NOT NULL,
                    style TEXT NOT NULL,
                    params TEXT NOT NULL,
                    phase TEXT NOT NULL,
                    nano_task_id TEXT,
                    result_url TEXT,
                    error TEXT,
                    attempts INTEGER DEFAULT 0,
                    created_at TEXT NOT NULL,
                    completed_at TEXT
                )
            """)
            conn.commit()

    def save(self, task: PipelineTask):
        """Save or update a task's progress."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """INSERT OR REPLACE INTO checkpoints
                   (task_id, prompt, style, params, phase, nano_task_id,
                    result_url, error, attempts, created_at, completed_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    task.id, task.prompt, task.style, json.dumps(task.params),
                    task.phase.value, task.nano_task_id, task.result_url,
                    task.error, task.attempts, task.created_at.isoformat(),
                    task.completed_at.isoformat() if task.completed_at else None,
                ),
            )
            conn.commit()

    def load_pending(self) -> list[PipelineTask]:
        """Load tasks that haven't finished yet (for restarting)."""
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT * FROM checkpoints WHERE phase != 'complete' AND phase != 'failed'"
            ).fetchall()
            tasks = []
            for row in rows:
                task = PipelineTask(
                    id=row[0], prompt=row[1], style=row[2],
                    params=json.loads(row[3]), phase=PipelinePhase(row[4]),
                    nano_task_id=row[5], result_url=row[6], error=row[7],
                    attempts=row[8],
                    created_at=datetime.fromisoformat(row[9]),
                    completed_at=datetime.fromisoformat(row[10]) if row[10] else None,
                )
                tasks.append(task)
            return tasks

class RateLimiter:
    """Controls how many tasks run at the same time.
    
    Like a bouncer at a club — only lets N people in at once.
    Prevents your computer from running out of memory.
    """

    def __init__(self, max_concurrent: int = 5):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.active = 0

    async def acquire(self):
        """Wait until there's room to run another task."""
        await self.semaphore.acquire()
        self.active += 1

    def release(self):
        """Signal that a task finished — room for another."""
        self.semaphore.release()
        self.active -= 1

    @property
    def utilization(self) -> float:
        """What fraction of capacity is being used (0.0 to 1.0)."""
        return self.active / self.semaphore._value if self.semaphore._value > 0 else 0

class ImagePipelineOrchestrator:
    """The main controller — runs the whole pipeline.
    
    This is the brain of the operation. It submits tasks,
    tracks progress, handles failures, and reports metrics.
    """

    def __init__(
        self,
        api_key: str,
        max_concurrent: int = 5,
        checkpoint_path: str = "pipeline_checkpoints.db",
        max_retries: int = 3,
    ):
        self.client = NanoBananaAsyncClient(api_key, max_retries=max_retries)
        self.rate_limiter = RateLimiter(max_concurrent)
        self.store = CheckpointStore(checkpoint_path)
        self.metrics = {
            "submitted": 0,
            "completed": 0,
            "failed": 0,
            "retries": 0,
        }

    async def process_task(self, task: PipelineTask) -> PipelineTask:
        """Process a single task through the pipeline with checkpointing.
        
        Each step saves progress. If the script crashes,
        you can pick up where you left off.
        """
        async with self.client as client:
            try:
                # Phase 1: Submit to NanoBanana
                task.phase = PipelinePhase.SUBMITTING
                task.attempts += 1
                self.store.save(task)

                task.nano_task_id = await client.submit_generation(
                    task.prompt, task.style, **task.params
                )
                self.metrics["submitted"] += 1

                # Phase 2: Wait for generation to finish
                task.phase = PipelinePhase.POLLING
                self.store.save(task)

                result = await client.poll_task(task.nano_task_id)
                task.result_url = result["image_url"]

                # Phase 3: Post-process (resize, format, etc.)
                task.phase = PipelinePhase.PROCESSING
                self.store.save(task)

                processed_url = await self._post_process(task.result_url, task.params)

                # Phase 4: Upload to CDN
                task.phase = PipelinePhase.UPLOADING
                self.store.save(task)

                final_url = await self._upload_to_cdn(processed_url)

                # Done!
                task.phase = PipelinePhase.COMPLETE
                task.completed_at = datetime.now(timezone.utc)
                self.store.save(task)
                self.metrics["completed"] += 1

                return task

            except Exception as e:
                task.error = str(e)
                task.phase = PipelinePhase.FAILED
                self.store.save(task)
                self.metrics["failed"] += 1

                # Retry if we haven't exhausted our attempts
                if task.attempts < self.client.max_retries:
                    logger.info(f"Retrying task {task.id} (attempt {task.attempts})")
                    self.metrics["retries"] += 1
                    return await self.process_task(task)

                return task

    async def _post_process(self, image_url: str, params: dict) -> str:
        """Post-process: resize, crop, format conversion."""
        # Download image from NanoBanana
        async with aiohttp.ClientSession() as session:
            async with session.get(image_url) as resp:
                image_data = await resp.read()

        # Apply transformations (simplified — in production use Pillow or libvips)
        width = params.get("width", 1024)
        height = params.get("height", 1024)

        # Save processed version to temp file
        output_path = Path(f"/tmp/processed_{uuid.uuid4().hex}.png")
        async with aiofiles.open(output_path, "wb") as f:
            await f.write(image_data)

        return str(output_path)

    async def _upload_to_cdn(self, file_path: str) -> str:
        """Upload processed image to CDN."""
        # Simplified — in production, use boto3 for S3 or similar
        cdn_url = f"https://cdn.nivantlabs.com/images/{Path(file_path).name}"
        return cdn_url

    async def run_batch(self, prompts: list[dict]) -> list[PipelineTask]:
        """Run a batch of generation tasks with rate limiting."""
        tasks = []
        for item in prompts:
            task = PipelineTask(
                id=str(uuid.uuid4()),
                prompt=item["prompt"],
                style=item.get("style", "photorealistic"),
                params={k: v for k, v in item.items() if k not in ("prompt", "style")},
            )
            self.store.save(task)
            tasks.append(task)

        async def _rate_limited_process(task: PipelineTask) -> PipelineTask:
            # Wait until there's room in the rate limiter
            await self.rate_limiter.acquire()
            try:
                return await self.process_task(task)
            finally:
                self.rate_limiter.release()

        # Run all tasks, respecting the rate limit
        results = await asyncio.gather(
            *[_rate_limited_process(t) for t in tasks],
            return_exceptions=True,
        )

        return [r if not isinstance(r, Exception) else tasks[i] for i, r in enumerate(results)]

    def get_metrics_report(self) -> dict:
        """Return current metrics for monitoring."""
        return {
            **self.metrics,
            "rate_limiter_utilization": self.rate_limiter.utilization,
            "success_rate": (
                self.metrics["completed"] / max(self.metrics["submitted"], 1) * 100
            ),
        }

# Usage
async def main():
    orchestrator = ImagePipelineOrchestrator(
        api_key=API_KEY,
        max_concurrent=5,    # Run 5 tasks at the same time
        max_retries=3,       # Retry up to 3 times on failure
    )

    batch = [
        {"prompt": "Futuristic cityscape at sunset, cyberpunk style", "style": "digital-art", "width": 1024, "height": 768},
        {"prompt": "Minimalist logo for a tech startup, blue and white", "style": "vector-art", "width": 512, "height": 512},
        {"prompt": "Product mockup of a smartphone on a wooden desk", "style": "photorealistic", "width": 1200, "height": 800},
    ]

    results = await orchestrator.run_batch(batch)

    for task in results:
        if task.phase == PipelinePhase.COMPLETE:
            print(f"✓ {task.prompt[:50]}... → {task.result_url}")
        else:
            print(f"✗ {task.prompt[:50]}... → {task.error}")

    print(f"\nMetrics: {json.dumps(orchestrator.get_metrics_report(), indent=2)}")

if __name__ == "__main__":
    asyncio.run(main())

Use Cases

1. Blog Thumbnail Pipeline

When you’d use this: You publish 50+ blog posts per month and need a unique thumbnail for each one. You want consistent branding — same style, same colors — without hiring a designer.

Why NanoBanana fits: The async API lets you generate all thumbnails in parallel. A batch of 50 takes about 3 minutes instead of 40 minutes with synchronous code.

async def generate_blog_thumbnails(articles: list[dict]) -> list[str]:
    """Generate branded thumbnails for a batch of articles."""
    prompts = []
    for article in articles:
        prompt = (
            f"Abstract background representing {article['topic']}, "
            f"soft gradients, {article['brand_color']} color scheme, "
            f"minimalist, no text, 16:9 aspect ratio"
        )
        prompts.append({
            "prompt": prompt,
            "style": "digital-art",
            "width": 1280,
            "height": 720,
        })

    orchestrator = ImagePipelineOrchestrator(API_KEY, max_concurrent=3)
    results = await orchestrator.run_batch(prompts)
    return [r.result_url for r in results if r.phase == PipelinePhase.COMPLETE]

2. E-Commerce Product Photography

When you’d use this: You run an online store and want to show each product in multiple styles — photorealistic, minimalist, lifestyle, 3D render. You want to A/B test which style gets more sales.

Why NanoBanana fits: Submit all style variants at once. The async pipeline handles them in parallel, and you get results in minutes instead of hours.

STYLES = ["photorealistic", "minimalist", "lifestyle", "3d-render"]

async def generate_product_variants(product_description: str) -> dict[str, str]:
    """Generate product images in multiple styles."""
    prompts = [
        {"prompt": f"{product_description}, {style} product photography, white background", "style": style}
        for style in STYLES
    ]

    orchestrator = ImagePipelineOrchestrator(API_KEY, max_concurrent=4)
    results = await orchestrator.run_batch(prompts)

    return {
        style: result.result_url
        for style, result in zip(STYLES, results)
        if result.phase == PipelinePhase.COMPLETE
    }

3. App Store Optimization (ASO) Screenshots

When you’d use this: You’re launching an app in 10 countries. Each country needs screenshots in the local language, showing different features, in the right device size.

Why NanoBanana fits: Generate all combinations (locales x features x devices) in one batch. The pipeline handles the complexity while you focus on the prompts.

async def generate_aso_screenshots(
    app_name: str,
    features: list[str],
    locales: list[str],
) -> dict[str, dict[str, str]]:
    """Generate ASO screenshots for multiple locales and features."""
    tasks = []
    for locale in locales:
        for feature in features:
            prompt = (
                f"Mobile app screenshot showing {feature}, "
                f"{app_name} interface, {locale} language text, "
                f"iOS style, clean UI, 6.7 inch display"
            )
            tasks.append({
                "prompt": prompt,
                "style": "screenshot",
                "width": 1290,
                "height": 2796,
            })

    orchestrator = ImagePipelineOrchestrator(API_KEY, max_concurrent=5)
    results = await orchestrator.run_batch(tasks)

    # Organize by locale
    output = {}
    idx = 0
    for locale in locales:
        output[locale] = {}
        for feature in features:
            if idx < len(results) and results[idx].phase == PipelinePhase.COMPLETE:
                output[locale][feature] = results[idx].result_url
            idx += 1

    return output

4. Video Storyboard Generation

When you’d use this: You’re planning a video and need to visualize each scene from multiple camera angles before filming.

Why NanoBanana fits: Generate all scenes and angles in one batch. The async pipeline means you don’t wait for scene 1 to finish before starting scene 2.

async def generate_storyboard(script_scenes: list[dict]) -> list[dict]:
    """Generate storyboard frames from script scenes."""
    tasks = []
    for scene in script_scenes:
        for angle in scene.get("angles", ["wide", "close-up", "over-the-shoulder"]):
            prompt = (
                f"Storyboard frame: {scene['description']}, "
                f"{angle} shot, cinematic lighting, "
                f"concept art style, rough sketch"
            )
            tasks.append({
                "prompt": prompt,
                "style": "concept-art",
                "width": 1920,
                "height": 1080,
            })

    orchestrator = ImagePipelineOrchestrator(API_KEY, max_concurrent=8)
    results = await orchestrator.run_batch(tasks)

    storyboard = []
    scene_idx = 0
    for scene in script_scenes:
        frames = []
        for angle in scene.get("angles", ["wide", "close-up", "over-the-shoulder"]):
            if scene_idx < len(results) and results[scene_idx].phase == PipelinePhase.COMPLETE:
                frames.append({
                    "angle": angle,
                    "url": results[scene_idx].result_url,
                })
            scene_idx += 1
        storyboard.append({"scene": scene["name"], "frames": frames})

    return storyboard

5. Multi-Model Benchmarking Pipeline

When you’d use this: You want to compare how different models and styles perform on the same prompt. Which one is fastest? Which has the best success rate?

Why NanoBanana fits: Run the same prompt across multiple model configurations. The pipeline collects timing and success metrics automatically.

MODELS = [
    {"style": "photorealistic", "model": "v2"},
    {"style": "digital-art", "model": "v2"},
    {"style": "anime", "model": "v1"},
    {"style": "vector-art", "model": "v1"},
]

async def benchmark_prompts(prompts: list[str]) -> dict[str, dict]:
    """Run the same prompts across multiple models and collect metrics."""
    results = {}

    for config in MODELS:
        orchestrator = ImagePipelineOrchestrator(API_KEY, max_concurrent=3)
        batch = [
            {"prompt": p, "style": config["style"], "model": config["model"]}
            for p in prompts
        ]

        start = asyncio.get_event_loop().time()
        task_results = await orchestrator.run_batch(batch)
        elapsed = asyncio.get_event_loop().time() - start

        key = f"{config['style']} ({config['model']})"
        results[key] = {
            "total_time": round(elapsed, 2),
            "avg_time_per_image": round(elapsed / len(prompts), 2),
            "success_rate": (
                sum(1 for r in task_results if r.phase == PipelinePhase.COMPLETE)
                / len(task_results) * 100
            ),
            "metrics": orchestrator.get_metrics_report(),
        }

    return results

Cheat Sheet

NanoBanana API Reference

Endpoint	Method	Purpose	Parameters	Returns
/v1/generate	POST	Submit generation task	prompt, style, width, height, async	{ task_id }
/v1/tasks/{id}	GET	Poll task status	—	{ status, image_url?, error? }
/v1/styles	GET	List available styles	—	{ styles: [...] }
/v1/models	GET	List available models	—	{ models: [...] }
/v1/account	GET	Account info & limits	—	{ tier, rate_limit, usage }
/v1/batch	POST	Submit batch of tasks	{ tasks: [...] }	{ batch_id, task_ids: [...] }
/v1/batches/{id}	GET	Poll batch status	—	{ status, tasks: [...] }
/v1/credits	GET	Check remaining credits	—	{ credits, expires_at }
/v1/queue	GET	Queue depth & wait time	—	{ depth, estimated_wait_s }
/v1/cancel/{id}	POST	Cancel a pending task	—	{ status: "cancelled" }
/v1/upscale	POST	Upscale an existing image	image_url, scale	{ task_id }
/v1/variations	POST	Generate variations	image_url, count	{ task_id }
/v1/inpaint	POST	Inpaint region	image_url, mask, prompt	{ task_id }
/v1/outpaint	POST	Outpaint beyond borders	image_url, direction, pixels	{ task_id }
/v1/controlnet	POST	ControlNet-guided gen	image_url, condition, prompt	{ task_id }
/v1/loras	GET	List available LoRAs	—	{ loras: [...] }
/v1/loras/apply	POST	Apply LoRA to generation	lora_id, prompt, weight	{ task_id }
/v1/webhooks	POST	Register webhook URL	url, events	{ webhook_id }
/v1/webhooks/{id}	DELETE	Remove webhook	—	{ status: "deleted" }
/v1/rate-limits	GET	Current rate limit status	—	{ remaining, reset_at, limit }
/v1/health	GET	API health check	—	{ status, uptime, version }

Pricing Tiers

Tier	Price	Rate Limit	Max Resolution	Concurrent Tasks	Webhooks	Priority Queue
Hobby	$0.00	10/min	512x512	1	No	No
Free Tier	Limited free tier (varies)	Great for prototyping and learning
Starter	$29/mo	100/min	1024x1024	5	Yes	No
Pro	$99/mo	500/min	2048x2048	20	Yes	Yes
Enterprise	Custom	Custom	Custom	Custom	Yes	Yes

Rate Limit Headers

Every response includes these headers for tracking your usage:

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 87
X-RateLimit-Reset: 1687532400
X-Request-Id: nb_req_abc123def456

Common Gotchas

Issue	Symptom	Fix
Cold start latency	First request takes 5-10s	Send a warm-up request every 60s
Token expiration	401 after 1 hour	Implement token refresh with 5min buffer
Queue saturation	429 Too Many Requests	Implement exponential backoff with jitter
Large payloads	413 Payload Too Large	Compress prompts > 1000 chars
Webhook timeout	No callback received	Set up polling fallback with 30s interval
Memory limits	OOM on 2048x2048+ images	Process in tiles, then stitch
Model version drift	Output quality changes	Pin model version in request
Credit exhaustion	402 Payment Required	Monitor credits via /v1/credits every 5 min
Concurrent limit	Tasks queued indefinitely	Use /v1/queue to check depth before submitting
Webhook delivery	Missed callbacks	Implement idempotency keys on webhook events

Debugging Tips

1. Always log the X-Request-Id header — support will ask for it.
2. Use /v1/health before submitting — if it returns non-200, wait 30s and retry.
3. Set up webhook monitoring — if you don’t receive a callback within 2x the expected generation time, poll manually.
4. Track credit usage — set up an alert at 20% remaining credits.
5. Use the batch endpoint for >10 tasks — it’s more efficient than individual submissions.
6. Enable compression — prompts over 500 chars should be compressed or truncated.
7. Test with the Hobby tier first — validate your pipeline before committing to a paid plan.

Vibe Coding Projects

Project 1: Automated Social Media Content Factory

Difficulty: Intermediate | Time: 4-6 hours

Build a system that generates branded social media images for multiple platforms (Twitter/X, LinkedIn, Instagram) from a content calendar.

What you’ll build:

• A scheduler that reads from a Google Sheet or Airtable
• Template system for each platform’s dimensions (Twitter: 1200x675, LinkedIn: 1200x627, Instagram: 1080x1080)
• Batch generation with NanoBanana’s async API
• Auto-posting via each platform’s API

Key skills practiced:

• Async pipeline orchestration
• Template-based prompt engineering
• Multi-platform output formatting
• Rate limit management

Starter code:

PLATFORM_CONFIGS = {
    "twitter": {"width": 1200, "height": 675, "style": "social-media"},
    "linkedin": {"width": 1200, "height": 627, "style": "professional"},
    "instagram": {"width": 1080, "height": 1080, "style": "artistic"},
}

async def social_media_factory(content_calendar: list[dict]) -> dict:
    """Generate and post social media images from a content calendar."""
    # Your implementation here
    pass

Project 2: Real-Time Image Generation Dashboard

Difficulty: Advanced | Time: 3-5 hours

Build a real-time dashboard that shows generation progress, metrics, and allows manual intervention (retry, cancel, re-prompt).

What you’ll build:

• WebSocket server that streams task status updates
• React dashboard with real-time metrics (success rate, latency, queue depth)
• Manual intervention controls (retry failed tasks, cancel stuck tasks, edit prompts)
• Historical metrics with time-series charts

Key skills practiced:

• WebSocket integration with async pipelines
• Real-time data visualization
• Human-in-the-loop workflow design
• Event sourcing for pipeline state

Tech stack: FastAPI + WebSocket + React + Chart.js

Project 3: Multi-Modal Content Generation Pipeline

Difficulty: Advanced | Time: 8-12 hours

Build a pipeline that generates images, captions, and alt text from a single content brief, then publishes to a CMS.

What you’ll build:

• Content brief parser that extracts key themes, colors, and styles
• Image generation via NanoBanana
• Caption generation via LLM (Claude API)
• Alt text generation with SEO optimization
• CMS integration (WordPress, Contentful, or Strapi)
• A/B testing framework for image variants

Key skills practiced:

• Multi-API orchestration
• Content generation pipelines
• SEO-aware output formatting
• CMS API integration

Problems Solved Efficiently

Problem Type	Why NanoBanana Fits	When to Look Elsewhere
10-50 assets per day (small scale)	Hobby or Starter tier is enough. Batch all tasks, poll in bulk. Handles 50 images in ~3 minutes.	For fewer than 10 images per day, a simple synchronous script is easier.
Multi-step image edits	Task-based API lets you chain operations (generate, upscale, apply LoRA) without blocking.	For single-step generation, the overhead of async isn’t worth it.
Cross-platform content	Generate images for web, mobile, and print from one prompt. Different sizes and styles in one batch.	If all your images are the same size and style, you don’t need this complexity.
Budget-constrained pipelines	Credit-based system means you never overspend. Set a daily budget and pause when exhausted.	For unlimited usage, look at providers with flat-rate pricing.

The Results

After switching to the async architecture, the numbers improved dramatically.

Metric	Before	After	Improvement
P50 latency	23.5s	8.9s	2.6x faster
P95 latency	58.9s	22.1s	2.7x faster
Error rate	23.0%	4.7%	4.9x reduction
Cost per success	$0.38	$0.14	2.7x cheaper
Throughput (images/hr)	85	420	4.9x more
Developer time (manual retries/hr)	15 min	2 min	7.5x less

What this means for you: Async architecture isn’t just about speed. It’s about reliability and cost. At 2,000 images per day, you save 8 hours of pipeline runtime, $480 per month in wasted API calls, and 22 hours of manual retry work. The async pattern pays for itself the first week.

What to Watch Out For

Beginner-Friendly Advice

1. Simplicity vs. resilience. The async pipeline is more complex than the synchronous version. You add checkpointing, rate limiting, and retry logic — each piece is something that could break. When to keep it simple: If you’re generating fewer than 50 images per day, just use the synchronous approach. It’s easier to write and debug.

2. Memory vs. throughput. Running 20 images at once means holding 20 images in memory. We hit out-of-memory errors twice before adding the rate limiter. Fix: Start with max_concurrent=3 and increase slowly. Watch your memory usage.

3. Latency vs. reliability. The polling loop adds about 500ms overhead per task. Webhooks (where the API calls you back) are faster but harder to set up. Fix: Start with polling. Switch to webhooks only if you need sub-second response times.

What Failed (And How We Fixed It)

1. Blind retries. We retried immediately on failure. This hammered the API during outages and made things worse. Fix: Wait longer between each retry (1s, then 2s, then 4s). Add random jitter so not everyone retries at the same time.

2. No checkpointing. When the pipeline crashed, we lost all in-flight tasks. We had to manually figure out which images were done and which weren’t. Fix: Save progress to a database after every step. If it crashes, restart from the last saved point.

3. Ignoring rate limits. We hit 429 errors constantly because we didn’t check how many requests we had left. Fix: Check your remaining rate limit before submitting. Slow down when you’re running low.

Advice for Beginners

1. Start synchronous, go async when it hurts. Don’t over-engineer from day one. A simple synchronous pipeline works fine for the first 500 images. Switch to async when you see the pain.

2. Measure everything. Without tracking latency and error rates, you won’t know where the bottlenecks are. Measure before you optimize.

3. Test with real loads. Testing with 10 images won’t reveal problems that show up at 200. Test at your actual scale.

4. Assume everything will fail. Every API call will fail eventually. The process will crash. The network will drop. Design your pipeline to survive all of these.

5. Use the batch endpoint. For more than 10 tasks, /v1/batch is significantly more efficient. We saw 15% lower latency and 30% fewer rate limit hits.

Course-Style Deep Dive

How NanoBanana Works Under the Hood (Simplified)

Think of NanoBanana’s API like a restaurant kitchen.

When you call the synchronous way, you’re standing at the counter waiting for your food. The chef can’t start the next order until yours is done. If the kitchen is busy, you wait. And wait.

When you call the async way, you place your order and get a ticket number. You sit down. The chef adds your order to a queue and works through orders in priority order. You check back periodically: “Is my order ready?” When it is, you pick it up.

The architecture has three main parts:

1. The Task Queue — Like the order board on a kitchen wall. When you submit a request, it goes on the board. Workers (the chefs) pick tasks off the board as they become available.

2. The Priority System — Higher-paying tiers get priority, like express lane at a deli. Your task jumps ahead of lower-tier tasks in the queue.

3. The Result Store — When a task finishes, the result is saved. You can pick it up anytime by checking with your ticket number (task_id).

Conditional Branching Pattern

Sometimes you need different post-processing based on the result. For example, if the image is too dark, brighten it. Here’s the pattern:

async def generate_with_conditional_postprocessing(prompt: str) -> str:
    """Generate an image and apply conditional post-processing.
    
    Checks the result and decides what to do next.
    Like checking if your food needs salt before serving.
    """
    async with NanoBananaAsyncClient(API_KEY) as client:
        task_id = await client.submit_generation(prompt, "photorealistic")
        result = await client.poll_task(task_id)

        image_url = result["image_url"]

        # Check image properties (simplified — in production, download and analyze)
        if result.get("avg_brightness", 128) < 80:
            # Image is too dark — apply brightness adjustment
            adjustment_task = await client.submit_adjustment(
                image_url, brightness=1.3
            )
            adjusted = await client.poll_task(adjustment_task)
            return adjusted["image_url"]

        return image_url

Human-in-the-Loop Pattern

For quality-critical applications, you might want a human to review generated images before they go live:

async def generate_with_review(prompt: str, reviewer_email: str) -> str:
    """Generate an image and wait for human review before finalizing.
    
    Like having a manager approve a design before it ships.
    """
    async with NanoBananaAsyncClient(API_KEY) as client:
        task_id = await client.submit_generation(prompt, "photorealistic")
        result = await client.poll_task(task_id)

        # Send for review
        review_id = await send_for_review(result["image_url"], reviewer_email)

        # Wait for review decision (polling a review service)
        while True:
            review_status = await check_review_status(review_id)
            if review_status["status"] == "approved":
                return result["image_url"]
            elif review_status["status"] == "rejected":
                # Generate a new version with feedback
                new_prompt = f"{prompt}, {review_status['feedback']}"
                return await generate_with_review(new_prompt, reviewer_email)

            await asyncio.sleep(5)

Monitoring with Prometheus

Here’s how to track your pipeline’s health:

from prometheus_client import Counter, Histogram, Gauge

# Metrics
GENERATION_REQUESTS = Counter(
    "nanobanana_generation_requests_total",
    "Total generation requests",
    ["status", "style"],
)

GENERATION_LATENCY = Histogram(
    "nanobanana_generation_latency_seconds",
    "Generation latency in seconds",
    buckets=[1, 5, 10, 20, 30, 45, 60, 90, 120],
)

QUEUE_DEPTH = Gauge(
    "nanobanana_queue_depth",
    "Current queue depth on NanoBanana",
)

CREDITS_REMAINING = Gauge(
    "nanobanana_credits_remaining",
    "Remaining API credits",
)

async def monitored_generate(prompt: str, style: str) -> str:
    """Generate with Prometheus instrumentation."""
    GENERATION_REQUESTS.labels(status="started", style=style).inc()

    start = time.time()
    try:
        async with NanoBananaAsyncClient(API_KEY) as client:
            task_id = await client.submit_generation(prompt, style)
            result = await client.poll_task(task_id)

        GENERATION_REQUESTS.labels(status="success", style=style).inc()
        GENERATION_LATENCY.observe(time.time() - start)

        return result["image_url"]
    except Exception as e:
        GENERATION_REQUESTS.labels(status="failed", style=style).inc()
        raise

Error Handling Hierarchy

Error Type	HTTP Status	Cause	Recovery Strategy
InvalidPrompt	400	Prompt too long or contains blocked terms	Truncate or sanitize prompt, retry
Unauthorized	401	Expired or invalid API key	Refresh token, retry once
InsufficientCredits	402	Account out of credits	Pause pipeline, alert admin
RateLimited	429	Exceeded rate limit	Backoff with jitter, check rate limit headers
TaskFailed	200 (status)	Model failed to generate	Retry with different style or prompt
TaskTimeout	—	Polling exceeded timeout	Check queue depth, retry with longer timeout
ServiceUnavailable	503	API undergoing maintenance	Circuit break for 60s, then retry
InternalError	500	Unexpected server error	Retry with exponential backoff, max 3 attempts

Rate Limit Findings

After weeks of production usage, here are our empirical rate limit observations:

• Burst vs sustained: The API allows short bursts above the stated limit (up to 2x for ~10 seconds), then enforces the limit strictly.
• Credit consumption: Each generation costs 1 credit. Upscaling costs 0.5 credits. Variations cost 0.25 credits each.
• Queue depth: At Pro tier, the queue depth rarely exceeds 50. At Hobby tier, we’ve seen depths of 200+ during peak hours.
• Best submission pattern: Submit in batches of 10-20 tasks, wait for all to complete, then submit the next batch. This keeps queue depth manageable and reduces per-task overhead.
• Webhook reliability: Webhook delivery is at-least-once. We’ve seen duplicate callbacks ~0.1% of the time. Implement idempotency keys on your webhook handler.

---

This is the first post in our “AI Tools in Production” series. Next up: Building a Multi-Model Evaluation Pipeline with Claude API.