Module 7: Use Cases

3. Automated Image Analysis

Next →

Description

Analyzing one image is useful. Analyzing 500 images with classification, data extraction, and quality control is a production system. In this capsule you build automated image analysis pipelines: batch processing, category classification, structured attribute extraction, and visual quality control.

Why it matters: The most profitable use cases of vision in production are automations: an e-commerce that classifies 1000 products a day, a quality control system that reviews manufacturing photos, a real-estate platform that extracts features from property photos. All of them require batch processing, error handling, and structured results.

Connection with the module: This pattern is one of the destinations of the Use Case Selector (capsule 08). When the router detects that the input is an image, it applies the analysis pipeline you build here. The batch and retry patterns you implement are reused in capsule 06 (Production Patterns).

Visual Pipeline

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  List of     │────▶│  Validate    │────▶│  Process     │
│  images      │     │  and prepare │     │  in batch    │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Results     │◀────│   Aggregate  │◀────│  Vision API  │
│  (JSON/CSV)  │     │  results     │     │  per image   │
└──────────────┘     └──────────────┘     └──────────────┘

Stages:

  1. Receive a list of images — Local paths or URLs
  2. Validate — Check that they exist, supported format, reasonable size
  3. Prepare — Resize if needed, convert to base64
  4. Process in batch — Send each image to the Vision API (with controlled parallelism)
  5. Aggregate results — Combine into a unified structure
  6. Export — JSON, CSV, or database

Step 1: Validation and Preparation

from pathlib import Path
from PIL import Image
import base64
import io

SUPPORTED_FORMATS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
MAX_SIZE_MB = 20
MAX_DIMENSION = 4096

def validate_image(path: str) -> dict:
    p = Path(path)

    if not p.exists():
        return {"valid": False, "error": "File not found"}

    if p.suffix.lower() not in SUPPORTED_FORMATS:
        return {"valid": False, "error": f"Unsupported format: {p.suffix}"}

    size_mb = p.stat().st_size / (1024 * 1024)
    if size_mb > MAX_SIZE_MB:
        return {"valid": False, "error": f"File too large: {size_mb:.1f}MB"}

    try:
        img = Image.open(path)
        width, height = img.size
        img.close()
    except Exception as e:
        return {"valid": False, "error": f"Can't open: {e}"}

    return {
        "valid": True,
        "path": path,
        "format": p.suffix.lower(),
        "size_mb": round(size_mb, 2),
        "dimensions": (width, height)
    }


def validate_batch(paths: list[str]) -> dict:
    valid = []
    invalid = []
    for path in paths:
        result = validate_image(path)
        if result["valid"]:
            valid.append(result)
        else:
            invalid.append({"path": path, **result})

    return {
        "valid": valid,
        "invalid": invalid,
        "total": len(paths),
        "valid_count": len(valid),
        "invalid_count": len(invalid)
    }

Prepare an image for the API

def prepare_image(path: str, max_dimension: int = 2048) -> str:
    img = Image.open(path)

    if max(img.size) > max_dimension:
        ratio = max_dimension / max(img.size)
        new_size = (int(img.size[0] * ratio), int(img.size[1] * ratio))
        img = img.resize(new_size, Image.LANCZOS)

    if img.mode == "RGBA":
        img = img.convert("RGB")

    buffer = io.BytesIO()
    img.save(buffer, format="JPEG", quality=85)
    img.close()

    return base64.b64encode(buffer.getvalue()).decode()

Step 2: Batch Classification

Simple classifier

from openai import OpenAI

client = OpenAI()

def classify_image(image_b64: str, categories: list[str]) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": f"Classify this image into ONE of these categories: {', '.join(categories)}.\nReply ONLY with the category name, nothing else."
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}
                }
            ]
        }],
        max_tokens=20,
        temperature=0
    )
    return response.choices[0].message.content.strip()


def classify_batch(
    image_paths: list[str],
    categories: list[str]
) -> list[dict]:
    results = []
    for path in image_paths:
        try:
            b64 = prepare_image(path)
            category = classify_image(b64, categories)
            results.append({
                "path": path,
                "category": category,
                "status": "success"
            })
        except Exception as e:
            results.append({
                "path": path,
                "category": None,
                "status": "error",
                "error": str(e)
            })
    return results

Classifier with confidence

import json

def classify_with_confidence(image_b64: str, categories: list[str]) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Classify this image into one of: {', '.join(categories)}.\n"
                        "Reply in JSON: {\"category\": \"...\", \"confidence\": 0.0-1.0, \"reasoning\": \"...\"}"
                    )
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}
                }
            ]
        }],
        max_tokens=100,
        temperature=0,
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

Step 3: Structured Extraction

Product attribute extractor

def extract_product_attributes(image_b64: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Analyze this product and extract attributes in JSON:\n"
                        "{\n"
                        "  \"product_type\": \"product type\",\n"
                        "  \"color_primary\": \"primary color\",\n"
                        "  \"color_secondary\": \"secondary color or null\",\n"
                        "  \"material\": \"visible material\",\n"
                        "  \"condition\": \"new/used/damaged\",\n"
                        "  \"brand_visible\": \"visible brand or null\",\n"
                        "  \"text_visible\": \"legible text or null\",\n"
                        "  \"description\": \"description in 1 sentence\"\n"
                        "}"
                    )
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}
                }
            ]
        }],
        max_tokens=300,
        temperature=0,
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)


def extract_batch_products(image_paths: list[str]) -> list[dict]:
    results = []
    for path in image_paths:
        try:
            b64 = prepare_image(path)
            attributes = extract_product_attributes(b64)
            results.append({
                "path": path,
                "attributes": attributes,
                "status": "success"
            })
        except Exception as e:
            results.append({
                "path": path,
                "attributes": None,
                "status": "error",
                "error": str(e)
            })
    return results

Generic extractor with a configurable schema

def extract_with_schema(image_b64: str, schema: dict) -> dict:
    schema_str = json.dumps(schema, indent=2, ensure_ascii=False)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": f"Extract data from this image following this JSON schema:\n{schema_str}\n\nReply ONLY with the JSON."
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}
                }
            ]
        }],
        max_tokens=500,
        temperature=0,
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

Step 4: Product Catalog Pipeline

import csv
from datetime import datetime

class ProductCatalogAnalyzer:
    def __init__(self, categories: list[str]):
        self.categories = categories
        self.results: list[dict] = []

    def analyze(self, image_paths: list[str]) -> list[dict]:
        validation = validate_batch(image_paths)

        if validation["invalid"]:
            print(f"Invalid images: {validation['invalid_count']}")
            for inv in validation["invalid"]:
                self.results.append({
                    "path": inv["path"],
                    "status": "invalid",
                    "error": inv["error"]
                })

        for img_info in validation["valid"]:
            path = img_info["path"]
            try:
                b64 = prepare_image(path)

                classification = classify_with_confidence(b64, self.categories)
                attributes = extract_product_attributes(b64)

                self.results.append({
                    "path": path,
                    "status": "success",
                    "category": classification["category"],
                    "confidence": classification["confidence"],
                    "attributes": attributes,
                    "dimensions": img_info["dimensions"],
                    "size_mb": img_info["size_mb"]
                })
            except Exception as e:
                self.results.append({
                    "path": path,
                    "status": "error",
                    "error": str(e)
                })

        return self.results

    def export_csv(self, output_path: str) -> str:
        successful = [r for r in self.results if r["status"] == "success"]

        if not successful:
            return "No results to export."

        fieldnames = ["path", "category", "confidence", "product_type",
                       "color_primary", "material", "condition", "description"]

        with open(output_path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for r in successful:
                row = {
                    "path": r["path"],
                    "category": r["category"],
                    "confidence": r["confidence"]
                }
                attrs = r.get("attributes", {})
                row.update({
                    "product_type": attrs.get("product_type", ""),
                    "color_primary": attrs.get("color_primary", ""),
                    "material": attrs.get("material", ""),
                    "condition": attrs.get("condition", ""),
                    "description": attrs.get("description", "")
                })
                writer.writerow(row)

        return output_path

    def summary(self) -> dict:
        total = len(self.results)
        success = sum(1 for r in self.results if r["status"] == "success")
        errors = sum(1 for r in self.results if r["status"] == "error")
        invalid = sum(1 for r in self.results if r["status"] == "invalid")

        category_counts = {}
        for r in self.results:
            if r["status"] == "success":
                cat = r["category"]
                category_counts[cat] = category_counts.get(cat, 0) + 1

        return {
            "total": total,
            "success": success,
            "errors": errors,
            "invalid": invalid,
            "categories": category_counts,
            "success_rate": round(success / total * 100, 1) if total > 0 else 0
        }

Usage:

analyzer = ProductCatalogAnalyzer(
    categories=["electronics", "clothing", "home", "sports", "food"]
)

results = analyzer.analyze([
    "products/img_001.jpg",
    "products/img_002.jpg",
    "products/img_003.jpg"
])

print(analyzer.summary())
analyzer.export_csv("analyzed_catalog.csv")

Step 5: Quality Control System

class QualityControlSystem:
    def __init__(self, criteria: dict = None):
        self.criteria = criteria or {
            "min_resolution": (800, 600),
            "check_blur": True,
            "check_lighting": True,
            "check_composition": True
        }

    def check_image(self, image_path: str) -> dict:
        validation = validate_image(image_path)
        if not validation["valid"]:
            return {"path": image_path, "pass": False, "issues": [validation["error"]]}

        issues = []

        w, h = validation["dimensions"]
        min_w, min_h = self.criteria["min_resolution"]
        if w < min_w or h < min_h:
            issues.append(f"Insufficient resolution: {w}x{h}, minimum {min_w}x{min_h}")

        b64 = prepare_image(image_path)
        quality_check = self._check_quality_with_vision(b64)
        issues.extend(quality_check.get("issues", []))

        return {
            "path": image_path,
            "pass": len(issues) == 0,
            "issues": issues,
            "quality_score": quality_check.get("score", 0),
            "details": quality_check
        }

    def _check_quality_with_vision(self, image_b64: str) -> dict:
        checks = []
        if self.criteria.get("check_blur"):
            checks.append("blur (is it blurry?)")
        if self.criteria.get("check_lighting"):
            checks.append("lighting (too dark/bright?)")
        if self.criteria.get("check_composition"):
            checks.append("composition (is the subject centered and complete?)")

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            "Evaluate the quality of this image for a product catalog.\n"
                            f"Check: {', '.join(checks)}.\n"
                            "Reply in JSON:\n"
                            "{\n"
                            "  \"score\": 1-10,\n"
                            "  \"issues\": [\"list of problems found\"],\n"
                            "  \"blur\": \"ok/issue\",\n"
                            "  \"lighting\": \"ok/issue\",\n"
                            "  \"composition\": \"ok/issue\",\n"
                            "  \"recommendation\": \"approved/reshoot\"\n"
                            "}"
                        )
                    },
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}
                    }
                ]
            }],
            max_tokens=200,
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

    def check_batch(self, image_paths: list[str]) -> dict:
        results = []
        for path in image_paths:
            result = self.check_image(path)
            results.append(result)

        passed = sum(1 for r in results if r["pass"])
        failed = len(results) - passed

        return {
            "results": results,
            "total": len(results),
            "passed": passed,
            "failed": failed,
            "pass_rate": round(passed / len(results) * 100, 1) if results else 0
        }

Parallel Processing

To process hundreds of images, use controlled parallelism:

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def process_batch_parallel(
    image_paths: list[str],
    process_fn,
    max_workers: int = 3,
    delay_between: float = 0.5
) -> list[dict]:
    results = []

    def process_one(path: str) -> dict:
        time.sleep(delay_between)
        try:
            return {"path": path, "result": process_fn(path), "status": "success"}
        except Exception as e:
            return {"path": path, "result": None, "status": "error", "error": str(e)}

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_one, path): path for path in image_paths}

        for future in as_completed(futures):
            results.append(future.result())

    return sorted(results, key=lambda r: image_paths.index(r["path"]))

With automatic retry

def process_with_retry(
    path: str,
    process_fn,
    max_retries: int = 3
) -> dict:
    for attempt in range(max_retries):
        try:
            result = process_fn(path)
            return {"path": path, "result": result, "status": "success", "attempts": attempt + 1}
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                time.sleep(wait)
                continue
            return {
                "path": path,
                "result": None,
                "status": "error",
                "error": str(e),
                "attempts": max_retries
            }

Troubleshooting

Problem 1: Rate limit with many images

Symptom: Error 429 after processing 20-30 images.

Cause: The Vision API has rate limits per minute (RPM) and per tokens (TPM).

Solution:

import time

def classify_batch_rate_limited(
    paths: list[str],
    categories: list[str],
    requests_per_minute: int = 30
) -> list[dict]:
    delay = 60.0 / requests_per_minute
    results = []

    for i, path in enumerate(paths):
        if i > 0:
            time.sleep(delay)
        try:
            b64 = prepare_image(path)
            category = classify_image(b64, categories)
            results.append({"path": path, "category": category, "status": "success"})
        except Exception as e:
            if "rate" in str(e).lower():
                time.sleep(10)
                try:
                    b64 = prepare_image(path)
                    category = classify_image(b64, categories)
                    results.append({"path": path, "category": category, "status": "success"})
                except Exception as e2:
                    results.append({"path": path, "category": None, "status": "error", "error": str(e2)})
            else:
                results.append({"path": path, "category": None, "status": "error", "error": str(e)})

    return results

Problem 2: Inconsistent classification

Symptom: The same image is classified differently in successive calls.

Solution: Use temperature=0 and a more restrictive prompt:

prompt = (
    f"Classify this image into EXACTLY ONE category from this list: {categories_str}.\n"
    "Rules:\n"
    "- Reply ONLY with the exact name of the category\n"
    "- Do NOT add explanation or punctuation\n"
    "- If it doesn't fit any, choose the closest one"
)

Problem 3: Very heavy images

Symptom: Slow calls or timeout on images of 10+ MB.

Solution: Resize before sending (already included in prepare_image). To reduce even further:

def prepare_image_low_cost(path: str) -> str:
    img = Image.open(path)
    img = img.resize((512, 512), Image.LANCZOS)
    if img.mode == "RGBA":
        img = img.convert("RGB")
    buffer = io.BytesIO()
    img.save(buffer, format="JPEG", quality=70)
    img.close()
    return base64.b64encode(buffer.getvalue()).decode()

Problem 4: Malformed JSON in extraction

Symptom: json.loads fails to parse the response.

Solution: Use response_format={"type": "json_object"} (already included in the examples). As a fallback:

def safe_json_parse(text: str) -> dict:
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        start = text.find("{")
        end = text.rfind("}") + 1
        if start >= 0 and end > start:
            try:
                return json.loads(text[start:end])
            except json.JSONDecodeError:
                pass
        return {"raw_response": text, "parse_error": True}

Exercises

Exercise 1: Image comparator

Create a function that takes two images of the same product and determines whether they're the same item, comparing extracted attributes.

See solution 
def compare_product_images(path_a: str, path_b: str) -> dict:
    b64_a = prepare_image(path_a)
    b64_b = prepare_image(path_b)

    attrs_a = extract_product_attributes(b64_a)
    attrs_b = extract_product_attributes(b64_b)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Compare these two product images.\n"
                        "Reply in JSON: {\"same_product\": true/false, "
                        "\"similarity_score\": 0.0-1.0, \"differences\": [\"...\"]}"
                    )
                },
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64_a}"}},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64_b}"}}
            ]
        }],
        max_tokens=200,
        temperature=0,
        response_format={"type": "json_object"}
    )

    comparison = json.loads(response.choices[0].message.content)
    comparison["attributes_a"] = attrs_a
    comparison["attributes_b"] = attrs_b
    return comparison

Exercise 2: Pipeline with progress and logging

Modify ProductCatalogAnalyzer.analyze so it prints progress (X/N processed) and logs timing per image.

See solution 
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("catalog_analyzer")

def analyze_with_progress(self, image_paths: list[str]) -> list[dict]:
    validation = validate_batch(image_paths)
    total = validation["valid_count"]
    processed = 0

    for img_info in validation["valid"]:
        path = img_info["path"]
        start = time.time()
        processed += 1

        try:
            b64 = prepare_image(path)
            classification = classify_with_confidence(b64, self.categories)
            attributes = extract_product_attributes(b64)
            elapsed = time.time() - start

            self.results.append({
                "path": path,
                "status": "success",
                "category": classification["category"],
                "confidence": classification["confidence"],
                "attributes": attributes,
                "processing_time": round(elapsed, 2)
            })

            logger.info(f"[{processed}/{total}] {path} → {classification['category']} ({elapsed:.1f}s)")

        except Exception as e:
            elapsed = time.time() - start
            self.results.append({
                "path": path,
                "status": "error",
                "error": str(e),
                "processing_time": round(elapsed, 2)
            })
            logger.error(f"[{processed}/{total}] {path} → ERROR: {e} ({elapsed:.1f}s)")

    return self.results

Exercise 3: Multi-provider with fallback

Implement classification that tries GPT-4o-mini first, and if it fails, uses Claude 3 as a fallback.

See solution 
import anthropic

def classify_with_fallback(image_path: str, categories: list[str]) -> dict:
    b64 = prepare_image(image_path)

    try:
        category = classify_image(b64, categories)
        return {"category": category, "provider": "openai", "status": "success"}
    except Exception as openai_error:
        pass

    try:
        claude = anthropic.Anthropic()
        response = claude.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=50,
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"Classify this image into ONE of: {', '.join(categories)}. Only the name."
                    },
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/jpeg",
                            "data": b64
                        }
                    }
                ]
            }]
        )
        return {
            "category": response.content[0].text.strip(),
            "provider": "anthropic",
            "status": "success"
        }
    except Exception as claude_error:
        return {
            "category": None,
            "provider": None,
            "status": "error",
            "errors": {
                "openai": str(openai_error),
                "anthropic": str(claude_error)
            }
        }

Additional Resources

  1. OpenAI Vision Guide — Official Vision guide
  2. OpenAI Batch API — Batch processing
  3. Pillow Documentation — Image processing in Python
  4. Anthropic Vision — Vision with Claude
← Previous2. Document Q&ANext →4. Video: Frames + LLM