◈ API / SDK/2026-06-22Advanced

Structured Product Image Analysis with the Gemini API — A Production Pipeline Built on Thousands of Photos

Turn a one-off image analysis script into a production pipeline that auto-generates tags, descriptions, and categories at scale — covering structured output, resumable batches, measured cost, and model routing learned from real indie-developer operation.

Gemini API²⁵⁰ Vision⁴ structured-output²⁸ python¹⁰⁸batchcost-optimization³⁰ indie-dev⁴⁸

✦ Premium Article

As an indie developer working on apps and store assets, I keep rediscovering how quietly the boring work — tidying up image metadata — eats my time. For a stretch I was tagging and writing descriptions for assets by hand as they grew by the hundreds. Even at a few dozen seconds each, that adds up to half a day gone.

So I switched to multimodal analysis with the Gemini API. But the first script I wrote — "send one image, get one result" — worked as a prototype yet showed its cracks the moment I pushed thousands of images through it. A single mid-run failure meant starting over, my cost estimates were too optimistic, and the category coming back would occasionally drift. After rebuilding it a few times for Dolice Labs, it settled into something that keeps running in production.

This article focuses on that delta — from a one-off prototype to a pipeline that survives failure. We'll start with the basics, then layer in measured cost, a resumable batch, and the operational details you won't find in the docs.

Prerequisites and Setup

What You'll Need

Python 3.10 or later
A Google AI Studio API key (get one at Google AI Studio)
The google-genai package

Setting Up Your Environment

# Create and activate a virtual environment
python -m venv gemini-image-env
source gemini-image-env/bin/activate  # Windows: gemini-image-env\Scripts\activate
 
# Install dependencies
pip install google-genai Pillow

Set your API key as an environment variable:

export GEMINI_API_KEY="your-api-key-here"

If you're new to the Gemini API, the Gemini API Quickstart Guide is a great place to start.

Basic Image Analysis — Extracting Information from a Single Product Photo

Let's start with the simplest case: sending a single product image to the Gemini API and getting a natural language description back.

# basic_image_analysis.py
import os
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def analyze_product_image(image_path: str) -> str:
    """Analyze a product image and return a description."""
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "Analyze this product image. Describe the product name, "
                        "category, color, material, and key features in detail."
                    ),
                ],
            )
        ],
    )
    return response.text
 
# Usage
result = analyze_product_image("product_sample.jpg")
print(result)
 
# Expected output:
# This is a white crew-neck T-shirt made from soft cotton fabric.
# It features a minimalist design with a small embroidered logo
# on the chest. The material appears to be 100% cotton, making it
# ideal for casual everyday wear.

This works well enough for a quick analysis, but the free-form text output is hard to process programmatically. Let's fix that with Structured Output.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Measured cost, latency, and accuracy across thousands of images, plus the break-even point for routing between Flash and Pro

✦A checkpoint-based batch you can stop and resume without losing work — complete, copy-paste-ready code

✦The undocumented gotchas: enum drift, schema strictness, and pre-upload downscaling, with fixes that survive real runs

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Structured Output — Getting Clean JSON Data

Gemini API's Structured Output feature forces the response to conform to a JSON schema you define. This means the data you get back is instantly ready for database storage, CSV export, or any other automated pipeline.

# structured_image_analysis.py
import os
import json
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Define the output schema for product analysis
product_schema = types.Schema(
    type="object",
    properties={
        "product_name": types.Schema(
            type="string",
            description="Name of the product",
        ),
        "category": types.Schema(
            type="string",
            enum=["Clothing", "Shoes", "Bags", "Accessories", "Electronics", "Food", "Other"],
            description="Product category",
        ),
        "tags": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="Tags describing the product (5-10 items)",
        ),
        "color": types.Schema(
            type="string",
            description="Primary color of the product",
        ),
        "material": types.Schema(
            type="string",
            description="Estimated material",
        ),
        "description_short": types.Schema(
            type="string",
            description="Brief product description (under 100 characters)",
        ),
        "description_long": types.Schema(
            type="string",
            description="Detailed product description (~200 words)",
        ),
        "target_audience": types.Schema(
            type="string",
            description="Target demographic",
        ),
        "estimated_season": types.Schema(
            type="string",
            enum=["Spring", "Summer", "Fall", "Winter", "All Seasons"],
            description="Recommended season for the product",
        ),
    },
    required=[
        "product_name", "category", "tags", "color",
        "description_short", "description_long",
    ],
)
 
def analyze_product_structured(image_path: str) -> dict:
    """Analyze a product image and return structured data."""
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "Analyze this product image and generate e-commerce listing data."
                    ),
                ],
            )
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=product_schema,
        ),
    )
    return json.loads(response.text)
 
# Usage
product_data = analyze_product_structured("product_sample.jpg")
print(json.dumps(product_data, indent=2))
 
# Expected output:
# {
#   "product_name": "Organic Cotton T-Shirt",
#   "category": "Clothing",
#   "tags": ["t-shirt", "cotton", "white", "casual", "plain", "unisex"],
#   "color": "White",
#   "material": "Cotton",
#   "description_short": "Soft organic cotton tee with a clean, minimal design",
#   "description_long": "A premium quality T-shirt crafted from 100% organic cotton...",
#   "target_audience": "Adults aged 20-40",
#   "estimated_season": "All Seasons"
# }

For keeping paired bilingual structured output consistent, see stabilizing terminology with paired structured output.

Batch Processing Multiple Images

In production, you'll need to process large numbers of product images at once. The following script handles batch processing across an entire directory and exports results to a CSV file.

# batch_product_analysis.py
import os
import csv
import json
import time
from pathlib import Path
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def batch_analyze_products(
    image_dir: str,
    output_csv: str,
    max_concurrent: int = 5,
    delay_sec: float = 1.0,
) -> list[dict]:
    """Batch-analyze all product images in a directory."""
    image_dir = Path(image_dir)
    supported_extensions = {".jpg", ".jpeg", ".png", ".webp"}
    image_files = sorted(
        f for f in image_dir.iterdir()
        if f.suffix.lower() in supported_extensions
    )
 
    print(f"Found {len(image_files)} images to process")
    results = []
    errors = []
 
    for i, image_file in enumerate(image_files):
        print(f"[{i + 1}/{len(image_files)}] Analyzing {image_file.name}...")
 
        try:
            product_data = analyze_product_structured(str(image_file))
            product_data["file_name"] = image_file.name
            results.append(product_data)
        except Exception as e:
            print(f"  Error: {e}")
            errors.append({"file_name": image_file.name, "error": str(e)})
 
        # Pause to avoid rate limits
        if (i + 1) % max_concurrent == 0:
            time.sleep(delay_sec)
 
    # Export to CSV
    if results:
        fieldnames = [
            "file_name", "product_name", "category", "tags",
            "color", "material", "description_short",
            "description_long", "target_audience", "estimated_season",
        ]
        with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for row in results:
                row["tags"] = ", ".join(row.get("tags", []))
                writer.writerow({k: row.get(k, "") for k in fieldnames})
 
        print(f"\nDone: {len(results)} succeeded, {len(errors)} failed")
        print(f"CSV saved to: {output_csv}")
 
    return results
 
# Usage
if __name__ == "__main__":
    results = batch_analyze_products(
        image_dir="./product_images",
        output_csv="./product_analysis_results.csv",
    )

This script is fine for a small directory. But push thousands of images through it and a single mid-run stop — a network blip or a rate limit — wipes everything, because results live only in memory. We'll replace this weakness with a resumable implementation below.

Error Handling and Retry Logic

For production reliability, you need proper error handling and retry logic. API rate limits and transient errors are inevitable, so here's a decorator that handles exponential backoff automatically.

# retry_utils.py
import time
import functools
from google.api_core import exceptions as google_exceptions
 
def retry_on_api_error(max_retries: int = 3, base_delay: float = 2.0):
    """Decorator that retries on API errors with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except google_exceptions.ResourceExhausted:
                    # Rate limit hit — wait longer
                    wait = base_delay * (2 ** attempt) * 2
                    print(f"  Rate limited. Waiting {wait:.1f}s ({attempt + 1}/{max_retries})")
                    time.sleep(wait)
                except google_exceptions.ServiceUnavailable:
                    # Server error — wait shorter
                    wait = base_delay * (2 ** attempt)
                    print(f"  Server error. Waiting {wait:.1f}s ({attempt + 1}/{max_retries})")
                    time.sleep(wait)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    wait = base_delay * (2 ** attempt)
                    print(f"  Unexpected error: {e}. Waiting {wait:.1f}s")
                    time.sleep(wait)
            raise RuntimeError(f"Failed after {max_retries} retries")
        return wrapper
    return decorator
 
# Usage: Apply the decorator to any analysis function
@retry_on_api_error(max_retries=3, base_delay=2.0)
def analyze_with_retry(image_path: str) -> dict:
    return analyze_product_structured(image_path)

What Thousands of Images Revealed About Cost and Throughput

At the prototype stage it's tempting to settle for "probably cheap." But measuring once before you go to production keeps later decisions from wobbling. Here's the rough breakdown from when I ran my own catalog images through it. These are condition-dependent reference numbers, but they're enough to anchor the order of magnitude.

Test conditions: JPEGs downscaled to a 1024px long edge, a fixed structured-output schema, one request per image.

Metric	gemini-3.5-flash	gemini-3.1-pro
Perceived latency per image	~1.2-2.0s	~2.5-4.0s
Approx. cost per 1,000 images	tens of yen	roughly 3-5x that
Category match rate (vs. manual ground truth)	~88%	~94%
Description edit rate	~20%	under ~10%
Retry rate from schema violations	under 1%	under 1%

The practical takeaway is that you don't need to run everything through the higher tier. Unambiguous products — a solid-color T-shirt, a plain mug, common SKUs — come back at fine quality on Flash. Where judgment splits is on photos with multiple elements or products where the material is hard to read.

So I went with a two-stage setup: a first pass on Flash, then re-analyzing only the low-confidence items on Pro. Overall cost dropped to roughly half or less, while accuracy approached Pro-only. I'll cover how to build the threshold in the operational notes below.

Note that gemini-3.5-flash is now generally available (GA), and its balance of speed and cost makes it a natural choice for this kind of first pass. The broader thinking on routing and caching to cut cost is laid out in a record of cutting cost with caching and model routing.

What the Docs Don't Tell You About Running This in Production

This is where prototype and production part ways. Every one of these is something I actually tripped over while pushing thousands of images through.

1. Absorb enum drift with normalization

Even with an enum in the schema, the model occasionally returns a near-miss value (a typo'd category, a mix of English and the target language). Rather than rejecting it as a schema violation, normalize it on the receiving side — your yield goes up.

# normalize.py
CATEGORY_CANON = {
    "Clothing": "Clothing", "Apparel": "Clothing", "Clothes": "Clothing",
    "Bags": "Bags", "Bag": "Bags", "Handbag": "Bags",
    "Shoes": "Shoes", "Footwear": "Shoes",
    "Accessories": "Accessories", "Accessory": "Accessories",
}
 
def canon_category(value: str, fallback: str = "Other") -> str:
    """Snap a model-returned category onto a canonical value."""
    v = (value or "").strip()
    return CATEGORY_CANON.get(v, fallback)

2. Downscale before upload to cut cost and latency at once

Product photos are often several MB, and sending them at full size inflates both input cost and upload time for no benefit. Classification and description quality barely drop when you shrink to around a 1024px long edge. Adding this single preprocessing step makes a clearly noticeable difference in perceived speed.

# downscale.py
import io
from PIL import Image
 
def downscale_jpeg(path: str, max_side: int = 1024, quality: int = 85) -> bytes:
    """Fit the long edge within max_side and return JPEG bytes."""
    img = Image.open(path).convert("RGB")
    w, h = img.size
    scale = min(1.0, max_side / max(w, h))
    if scale < 1.0:
        img = img.resize((round(w * scale), round(h * scale)), Image.LANCZOS)
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=quality, optimize=True)
    return buf.getvalue()

3. Make the batch resumable so a crash doesn't cost you everything

This was the single biggest win. Append each result to a JSON Lines file one image at a time, and on startup read the already-processed filenames and skip them. If it stops, just start it again and it picks up where it left off.

# resumable_batch.py
import json
from pathlib import Path
 
def load_done(jsonl_path: str) -> set[str]:
    """Load the set of already-processed file names."""
    done = set()
    p = Path(jsonl_path)
    if p.exists():
        for line in p.read_text(encoding="utf-8").splitlines():
            try:
                done.add(json.loads(line)["file_name"])
            except (json.JSONDecodeError, KeyError):
                continue  # ignore corrupt lines and keep going
    return done
 
def resumable_batch(image_dir: str, jsonl_path: str) -> None:
    """Analyze all images with a checkpoint (resumable)."""
    done = load_done(jsonl_path)
    targets = sorted(
        f for f in Path(image_dir).iterdir()
        if f.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}
        and f.name not in done
    )
    print(f"{len(targets)} remaining (skipping {len(done)} already done)")
 
    with open(jsonl_path, "a", encoding="utf-8") as out:
        for i, f in enumerate(targets, 1):
            try:
                data = analyze_with_retry(str(f))  # decorated in retry_utils
                data["file_name"] = f.name
                out.write(json.dumps(data, ensure_ascii=False) + "\n")
                out.flush()  # commit per item so a stop loses nothing
                print(f"[{i}/{len(targets)}] {f.name} ✓")
            except Exception as e:
                print(f"[{i}/{len(targets)}] {f.name} ✗ {e}")

Calling flush() per item is the key. Leave results buffered and a forced termination wipes the last few dozen records. It trades a little speed for certainty.

4. Cache when you want the same product to return the same result

If the wording shifts slightly on every regeneration, diff management becomes a chore. Key a cache on a hash of the image content, and identical images return identical results.

# cache_key.py
import hashlib
 
def image_cache_key(image_bytes: bytes, schema_version: str = "v1") -> str:
    """A cache key combining image content and schema version."""
    h = hashlib.sha256(image_bytes).hexdigest()[:16]
    return f"{schema_version}:{h}"

Including the schema version in the key matters because if you change the schema, the result should change too. Forget this and you'll keep getting stale results after a schema update — confusing to debug.

Going Further — Multilingual Product Descriptions

If you're targeting international markets, Gemini can generate product descriptions in multiple languages simultaneously. Simply extend the schema to include fields for each language.

# multilingual_product_analysis.py
multilingual_schema = types.Schema(
    type="object",
    properties={
        "product_name_en": types.Schema(type="string", description="Product name (English)"),
        "product_name_ja": types.Schema(type="string", description="Product name (Japanese)"),
        "description_en": types.Schema(type="string", description="Product description (English, ~100 words)"),
        "description_ja": types.Schema(type="string", description="Product description (Japanese, ~200 chars)"),
        "seo_keywords_en": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="SEO keywords (English, 5 items)",
        ),
        "seo_keywords_ja": types.Schema(
            type="array",
            items=types.Schema(type="string"),
            description="SEO keywords (Japanese, 5 items)",
        ),
    },
    required=[
        "product_name_en", "product_name_ja",
        "description_en", "description_ja",
    ],
)
 
def analyze_multilingual(image_path: str) -> dict:
    """Generate multilingual product information from a single image."""
    with open(image_path, "rb") as f:
        image_data = f.read()
 
    response = client.models.generate_content(
        model="gemini-3.1-pro",
        contents=[
            types.Content(
                role="user",
                parts=[
                    types.Part.from_bytes(data=image_data, mime_type="image/jpeg"),
                    types.Part.from_text(
                        "Analyze this product image and generate product information "
                        "in both English and Japanese for an e-commerce listing."
                    ),
                ],
            )
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=multilingual_schema,
        ),
    )
    return json.loads(response.text)
 
# Expected output:
# {
#   "product_name_en": "Leather Mini Crossbody Bag",
#   "product_name_ja": "レザーミニショルダーバッグ",
#   "description_en": "A compact crossbody bag crafted from premium genuine leather...",
#   "description_ja": "上質な本革を使用したコンパクトなショルダーバッグ...",
#   "seo_keywords_en": ["leather bag", "crossbody bag", "genuine leather", "mini bag", "women"],
#   "seo_keywords_ja": ["レザーバッグ", "ショルダーバッグ", "本革", "ミニバッグ", "レディース"]
# }

For more on composing multimodal work that spans images and documents, see an implementation report on handling documents multimodally.

Wrapping Up — Measure 100 First, Then Scale

What actually moved the needle in growing a one-off script into a production pipeline wasn't anything flashy — it was three quiet things: a resumable batch, downscaling in preprocessing, and a two-stage model route.

As your next step, run this article's checkpoint approach over 100 of your own images and measure the Flash-vs-Pro match rate and cost on your own data. Those numbers become the basis for how you split the full set. Ever since I started adding this "measure small first" step, my estimates have missed by far less. Thank you for reading.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →