Módulo 7: Casos de Uso

3. Image Analysis Automatizado

Siguiente →

Descripción

Analizar una imagen es útil. Analizar 500 imágenes con clasificación, extracción de datos, y control de calidad es un sistema de producción. En esta cápsula construyes pipelines de análisis de imágenes automatizado: procesamiento en batch, clasificación por categorías, extracción estructurada de atributos, y control de calidad visual.

Por qué importa: Los casos de uso más rentables de vision en producción son automatizaciones: un e-commerce que clasifica 1000 productos al día, un sistema de quality control que revisa fotos de manufactura, una plataforma inmobiliaria que extrae características de fotos de propiedades. Todo requiere procesamiento en batch, manejo de errores, y resultados estructurados.

Conexión con el módulo: Este patrón es uno de los destinos del Use Case Selector (cápsula 08). Cuando el router detecta que el input es una imagen, aplica el pipeline de análisis que construyes aquí. Los patrones de batch y retry que implementas se reutilizan en la cápsula 06 (Patrones de Producción).

Pipeline Visual

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Lista de    │────▶│  Validar y   │────▶│  Procesar    │
│  imágenes    │     │  preparar    │     │  en batch    │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Resultados  │◀────│   Agregar    │◀────│  Vision API  │
│  (JSON/CSV)  │     │  resultados  │     │  por imagen  │
└──────────────┘     └──────────────┘     └──────────────┘

Etapas:

  1. Recibir lista de imágenes — Paths locales o URLs
  2. Validar — Verificar que existen, formato soportado, tamaño razonable
  3. Preparar — Redimensionar si necesario, convertir a base64
  4. Procesar en batch — Enviar cada imagen a Vision API (con paralelismo controlado)
  5. Agregar resultados — Combinar en estructura unificada
  6. Exportar — JSON, CSV, o base de datos

Paso 1: Validación y Preparación

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": "Archivo no encontrado"}

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

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

    try:
        img = Image.open(path)
        width, height = img.size
        img.close()
    except Exception as e:
        return {"valid": False, "error": f"No se puede abrir: {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)
    }

Preparar imagen para 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()

Paso 2: Clasificación en Batch

Clasificador simple

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"Clasifica esta imagen en UNA de estas categorías: {', '.join(categories)}.\nResponde SOLO con el nombre de la categoría, nada más."
                },
                {
                    "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

Clasificador con confianza

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"Clasifica esta imagen en una de: {', '.join(categories)}.\n"
                        "Responde en 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)

Paso 3: Extracción Estructurada

Extractor de atributos de producto

def extract_product_attributes(image_b64: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Analiza este producto y extrae atributos en JSON:\n"
                        "{\n"
                        "  \"product_type\": \"tipo de producto\",\n"
                        "  \"color_primary\": \"color principal\",\n"
                        "  \"color_secondary\": \"color secundario o null\",\n"
                        "  \"material\": \"material visible\",\n"
                        "  \"condition\": \"nuevo/usado/dañado\",\n"
                        "  \"brand_visible\": \"marca visible o null\",\n"
                        "  \"text_visible\": \"texto legible o null\",\n"
                        "  \"description\": \"descripción en 1 oración\"\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

Extractor genérico con schema configurable

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"Extrae datos de esta imagen siguiendo este schema JSON:\n{schema_str}\n\nResponde SOLO con el 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)

Paso 4: Pipeline de Catálogo de Productos

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"Imágenes inválidas: {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 hay resultados para exportar."

        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
        }

Uso:

analyzer = ProductCatalogAnalyzer(
    categories=["electrónica", "ropa", "hogar", "deportes", "alimentos"]
)

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

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

Paso 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"Resolución insuficiente: {w}x{h}, mínimo {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 (¿está borrosa?)")
        if self.criteria.get("check_lighting"):
            checks.append("iluminación (¿demasiado oscura/brillante?)")
        if self.criteria.get("check_composition"):
            checks.append("composición (¿el sujeto está centrado y completo?)")

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            "Evalúa la calidad de esta imagen para un catálogo de productos.\n"
                            f"Verifica: {', '.join(checks)}.\n"
                            "Responde en JSON:\n"
                            "{\n"
                            "  \"score\": 1-10,\n"
                            "  \"issues\": [\"lista de problemas encontrados\"],\n"
                            "  \"blur\": \"ok/issue\",\n"
                            "  \"lighting\": \"ok/issue\",\n"
                            "  \"composition\": \"ok/issue\",\n"
                            "  \"recommendation\": \"aprobada/retomar\"\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
        }

Procesamiento Paralelo

Para procesar cientos de imágenes, usar paralelismo controlado:

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"]))

Con retry automático

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

Problema 1: Rate limit con muchas imágenes

Síntoma: Error 429 después de procesar 20-30 imágenes.

Causa: Vision API tiene rate limits por minuto (RPM) y por tokens (TPM).

Solución:

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

Problema 2: Clasificación inconsistente

Síntoma: La misma imagen se clasifica diferente en llamadas sucesivas.

Solución: Usar temperature=0 y prompt más restrictivo:

prompt = (
    f"Clasifica esta imagen en EXACTAMENTE UNA categoría de esta lista: {categories_str}.\n"
    "Reglas:\n"
    "- Responde SOLO con el nombre exacto de la categoría\n"
    "- NO agregues explicación ni puntuación\n"
    "- Si no encaja en ninguna, elige la más cercana"
)

Problema 3: Imágenes muy pesadas

Síntoma: Llamadas lentas o timeout en imágenes de 10+ MB.

Solución: Redimensionar antes de enviar (ya incluido en prepare_image). Para reducir aún más:

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()

Problema 4: JSON malformado en extracción

Síntoma: json.loads falla al parsear la respuesta.

Solución: Usar response_format={"type": "json_object"} (ya incluido en los ejemplos). Como 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}

Ejercicios

Ejercicio 1: Comparador de imágenes

Crea una función que reciba dos imágenes del mismo producto y determine si son del mismo artículo, comparando atributos extraídos.

Ver solución 
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": (
                        "Compara estas dos imágenes de producto.\n"
                        "Responde en 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

Ejercicio 2: Pipeline con progreso y logging

Modifica ProductCatalogAnalyzer.analyze para que imprima progreso (X/N procesadas) y registre timing por imagen.

Ver solución 
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

Ejercicio 3: Multi-proveedor con fallback

Implementa clasificación que intente con GPT-4o-mini primero, y si falla, use Claude 3 como fallback.

Ver solución 
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"Clasifica esta imagen en UNA de: {', '.join(categories)}. Solo el nombre."
                    },
                    {
                        "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)
            }
        }

Recursos Adicionales

  1. OpenAI Vision Guide — Guía oficial de Vision
  2. OpenAI Batch API — Procesamiento en batch
  3. Pillow Documentation — Procesamiento de imágenes en Python
  4. Anthropic Vision — Vision con Claude
← Anterior2. Document Q&ASiguiente →4. Video: Frames + LLM