Módulo 2: Vision + LLMs

8. Proyecto: Image Analyzer Multi-Provider

Descripción

Este proyecto cierra el Módulo 2 integrando todo lo que aprendiste: APIs de visión de OpenAI (cápsula 02), Anthropic (cápsula 03) y Google (cápsula 04), la comparación entre proveedores (cápsula 05), los patrones de análisis — descripción, OCR, clasificación, extracción, Q&A — (cápsula 06) y el manejo de múltiples imágenes (cápsula 07).

Vas a construir un Image Analyzer — un sistema que recibe una imagen, elige el patrón de análisis, enruta al proveedor óptimo con fallback automático, calcula costos reales y devuelve un resultado estructurado con toda la metadata.

Por qué importa: En producción, ningún proveedor tiene 100% de uptime. Si tu sistema depende de un solo proveedor y ese proveedor falla a las 3am, tu pipeline se detiene. Un analyzer con fallback automático, tracking de costos y logging es la diferencia entre un prototipo y un sistema operable.

Conexión con la guía: En el Módulo 3 (Comprensión de Documentos), este analyzer se convierte en el motor de visión del Document Analyzer. En el Módulo 8 (Proyecto Final), la lógica de routing y fallback se reutiliza para manejar documentos multi-página con diferentes proveedores por página.

Especificaciones Técnicas

Input

Parámetro	Tipo	Descripción
`image_path`	`str`	Ruta local a imagen (PNG, JPEG, WebP, GIF)
`pattern`	`str`	Patrón de análisis: `description`, `ocr`, `classification`, `extraction`, `qa`
`provider`	`str \| None`	Proveedor preferido. `None` = auto-selección
`priority`	`str`	`"quality"`, `"cost"` o `"balanced"`
`question`	`str \| None`	Pregunta para patrón `qa`

Output

@dataclass
class AnalysisResult:
    response: str
    pattern: str
    provider: str
    model: str
    latency_seconds: float
    input_tokens: int
    output_tokens: int
    estimated_cost_usd: float
    warnings: list[str]
    fallback_log: list[str]

Requisitos

Soportar los 3 proveedores con código real (no stubs)
5 patrones de análisis con prompts dedicados
Fallback automático con cadena configurable
Cálculo de costo real basado en tokens consumidos
Logging de cada intento (éxito y fallo)
Validación de imagen antes de enviar

Paso 1: Configuración y Constantes

import os
import time
import math
import base64
import logging
from enum import Enum
from pathlib import Path
from dataclasses import dataclass, field

logger = logging.getLogger("image_analyzer")
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")


class AnalysisPattern(Enum):
    DESCRIPTION = "description"
    OCR = "ocr"
    CLASSIFICATION = "classification"
    EXTRACTION = "extraction"
    QA = "qa"


@dataclass
class AnalysisResult:
    response: str = ""
    pattern: str = ""
    provider: str = ""
    model: str = ""
    latency_seconds: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0
    estimated_cost_usd: float = 0.0
    warnings: list[str] = field(default_factory=list)
    fallback_log: list[str] = field(default_factory=list)


PROVIDER_CONFIG = {
    "openai": {
        "model_quality": "gpt-4o",
        "model_cost": "gpt-4o-mini",
        "pricing": {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        },
        "max_image_mb": 20,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
    "anthropic": {
        "model_quality": "claude-3-5-sonnet-latest",
        "model_cost": "claude-3-haiku-20240307",
        "pricing": {
            "claude-3-5-sonnet-latest": {"input": 3.00, "output": 15.00},
            "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
        },
        "max_image_mb": 5,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
    "google": {
        "model_quality": "gemini-1.5-pro",
        "model_cost": "gemini-2.0-flash",
        "pricing": {
            "gemini-1.5-pro": {"input": 1.25, "output": 5.00},
            "gemini-2.0-flash": {"input": 0.10, "output": 0.40},
        },
        "max_image_mb": 20,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
}

MIME_TYPES = {
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
    ".gif": "image/gif", ".webp": "image/webp",
}

Paso 2: Preparar Imagen por Proveedor

Cada proveedor espera la imagen en un formato distinto. OpenAI acepta data URIs (Base64 con prefijo MIME), Anthropic necesita el Base64 crudo con media_type separado, y Gemini recibe bytes crudos.

def validate_image(image_path: str) -> tuple[Path, list[str]]:
    """Valida existencia, formato y tamaño de la imagen."""
    warnings = []
    path = Path(image_path)

    if not path.exists():
        raise FileNotFoundError(f"Imagen no encontrada: {image_path}")

    ext = path.suffix.lower()
    if ext not in MIME_TYPES:
        raise ValueError(f"Formato no soportado: {ext}")

    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > 20:
        raise ValueError(f"Imagen demasiado grande: {size_mb:.1f} MB (máximo 20 MB)")
    if size_mb > 4:
        warnings.append(f"Imagen grande ({size_mb:.1f} MB) — puede ser lenta con Anthropic (límite 5 MB)")

    return path, warnings


def prepare_image_openai(path: Path) -> dict:
    """Data URI para OpenAI."""
    mime = MIME_TYPES[path.suffix.lower()]
    encoded = base64.b64encode(path.read_bytes()).decode()
    return {
        "type": "image_url",
        "image_url": {"url": f"data:{mime};base64,{encoded}", "detail": "high"},
    }


def prepare_image_anthropic(path: Path) -> dict:
    """Base64 con media_type separado para Anthropic."""
    mime = MIME_TYPES[path.suffix.lower()]
    encoded = base64.b64encode(path.read_bytes()).decode()
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": mime, "data": encoded},
    }


def prepare_image_gemini(path: Path) -> dict:
    """Bytes crudos con MIME para Gemini."""
    mime = MIME_TYPES[path.suffix.lower()]
    return {"mime_type": mime, "data": path.read_bytes()}

Paso 3: Generar Prompt por Patrón

Cada patrón de análisis produce un prompt especializado. Estos vienen directamente de la cápsula 06.

PATTERN_PROMPTS = {
    AnalysisPattern.DESCRIPTION: (
        "Describe esta imagen de forma objetiva.\n"
        "Incluye: escena principal, objetos visibles, colores dominantes, texto visible (si hay).\n"
        "Máximo 100 palabras."
    ),
    AnalysisPattern.OCR: (
        "Extrae todo el texto visible en esta imagen.\n"
        "Mantén la estructura: párrafos, listas, tablas.\n"
        "Si hay tablas, usa formato markdown.\n"
        "No inventes texto que no esté visible."
    ),
    AnalysisPattern.CLASSIFICATION: (
        "Clasifica esta imagen en exactamente una de las siguientes categorías:\n"
        "documento, fotografía, diagrama, captura_de_pantalla, arte, meme, otro.\n"
        "Responde ÚNICAMENTE con el nombre de la categoría."
    ),
    AnalysisPattern.EXTRACTION: (
        "Extrae toda la información estructurada visible en esta imagen.\n"
        "Responde ÚNICAMENTE con un JSON válido.\n"
        "Incluye todos los campos que puedas identificar: fechas, números, nombres, montos.\n"
        "Usa null para campos ambiguos."
    ),
    AnalysisPattern.QA: (
        "Responde la siguiente pregunta basándote ÚNICAMENTE en lo visible en la imagen.\n"
        "Si la información no está en la imagen, responde 'No visible en la imagen'.\n\n"
        "Pregunta: {question}"
    ),
}


def get_prompt(pattern: AnalysisPattern, question: str | None = None) -> str:
    """Retorna el prompt para el patrón dado."""
    prompt = PATTERN_PROMPTS[pattern]
    if pattern == AnalysisPattern.QA:
        if not question:
            raise ValueError("El patrón 'qa' requiere una pregunta")
        prompt = prompt.format(question=question)
    return prompt

Paso 4: Llamar a Cada Proveedor

Cada función envía la imagen con el prompt al proveedor correspondiente y retorna un AnalysisResult parcial con la respuesta, tokens y latencia.

OpenAI

def _call_openai(path: Path, prompt: str, model: str) -> AnalysisResult:
    from openai import OpenAI

    client = OpenAI()
    image_block = prepare_image_openai(path)

    start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                image_block,
            ],
        }],
        max_tokens=1024,
    )
    latency = time.time() - start

    usage = response.usage
    return AnalysisResult(
        response=response.choices[0].message.content,
        provider="openai",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
    )

Anthropic

def _call_anthropic(path: Path, prompt: str, model: str) -> AnalysisResult:
    import anthropic

    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > PROVIDER_CONFIG["anthropic"]["max_image_mb"]:
        raise ValueError(f"Imagen excede límite de Anthropic: {size_mb:.1f} MB > 5 MB")

    client = anthropic.Anthropic()
    image_block = prepare_image_anthropic(path)

    start = time.time()
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [image_block, {"type": "text", "text": prompt}],
        }],
    )
    latency = time.time() - start

    return AnalysisResult(
        response=response.content[0].text,
        provider="anthropic",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
    )

Google Gemini

def _call_gemini(path: Path, prompt: str, model: str) -> AnalysisResult:
    import google.generativeai as genai

    genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
    gemini_model = genai.GenerativeModel(model)
    image_part = prepare_image_gemini(path)

    start = time.time()
    response = gemini_model.generate_content([prompt, image_part])
    latency = time.time() - start

    input_tokens = 0
    output_tokens = 0
    if hasattr(response, "usage_metadata") and response.usage_metadata:
        input_tokens = getattr(response.usage_metadata, "prompt_token_count", 0)
        output_tokens = getattr(response.usage_metadata, "candidates_token_count", 0)

    return AnalysisResult(
        response=response.text,
        provider="google",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=input_tokens,
        output_tokens=output_tokens,
    )

Paso 5: Router con Fallback

El router intenta proveedores en orden. Si el primero falla, pasa al siguiente y registra el fallo.

CALLER_MAP = {
    "openai": _call_openai,
    "anthropic": _call_anthropic,
    "google": _call_gemini,
}

DEFAULT_CHAINS = {
    "quality": ["openai", "anthropic", "google"],
    "cost": ["google", "openai", "anthropic"],
    "balanced": ["openai", "google", "anthropic"],
}


def route_with_fallback(
    path: Path,
    prompt: str,
    priority: str = "balanced",
    preferred_provider: str | None = None,
) -> AnalysisResult:
    """Intenta proveedores en orden con fallback automático."""
    chain = list(DEFAULT_CHAINS.get(priority, DEFAULT_CHAINS["balanced"]))

    if preferred_provider and preferred_provider in CALLER_MAP:
        chain.remove(preferred_provider) if preferred_provider in chain else None
        chain.insert(0, preferred_provider)

    fallback_log = []

    for provider_name in chain:
        config = PROVIDER_CONFIG[provider_name]
        model = config["model_quality"] if priority == "quality" else config["model_cost"]
        caller = CALLER_MAP[provider_name]

        try:
            logger.info(f"Intentando {provider_name} ({model})...")
            result = caller(path, prompt, model)
            result.fallback_log = fallback_log
            if fallback_log:
                result.warnings.append(f"Proveedor primario falló, se usó fallback: {provider_name}")
            return result
        except Exception as e:
            msg = f"{provider_name} ({model}) falló: {type(e).__name__}: {e}"
            logger.warning(msg)
            fallback_log.append(msg)
            continue

    error_result = AnalysisResult(
        response="",
        warnings=["Todos los proveedores fallaron"],
        fallback_log=fallback_log,
    )
    raise RuntimeError(f"Todos los proveedores fallaron. Log: {fallback_log}")

Paso 6: Cálculo de Costos

El costo real se calcula a partir de los tokens reportados por cada proveedor, usando la tabla de precios por millón de tokens.

def calculate_cost(result: AnalysisResult) -> float:
    """Calcula costo real basado en tokens consumidos."""
    config = PROVIDER_CONFIG.get(result.provider)
    if not config:
        return 0.0

    pricing = config["pricing"].get(result.model)
    if not pricing:
        return 0.0

    input_cost = (result.input_tokens / 1_000_000) * pricing["input"]
    output_cost = (result.output_tokens / 1_000_000) * pricing["output"]
    return round(input_cost + output_cost, 6)


def estimate_cost_before_call(
    image_path: str, priority: str, provider: str | None = None
) -> dict:
    """Estima costo antes de llamar a la API (sin hacer la llamada)."""
    path = Path(image_path)
    size_mb = path.stat().st_size / (1024 * 1024)
    image_tokens = max(85, int(size_mb * 1000))
    output_tokens_est = 200

    estimates = {}
    for name, config in PROVIDER_CONFIG.items():
        model = config["model_quality"] if priority == "quality" else config["model_cost"]
        pricing = config["pricing"][model]
        cost = (image_tokens / 1_000_000) * pricing["input"] + (output_tokens_est / 1_000_000) * pricing["output"]
        estimates[name] = {"model": model, "estimated_cost": round(cost, 6)}

    return estimates

Paso 7: Función Principal — Integración

def analyze_image(
    image_path: str,
    pattern: str = "description",
    provider: str | None = None,
    priority: str = "balanced",
    question: str | None = None,
) -> AnalysisResult:
    """Analiza una imagen con el patrón y proveedor indicados.

    Args:
        image_path: Ruta local a la imagen.
        pattern: Patrón de análisis (description, ocr, classification, extraction, qa).
        provider: Proveedor preferido. None = auto-selección según priority.
        priority: "quality", "cost" o "balanced".
        question: Pregunta para el patrón qa.

    Returns:
        AnalysisResult con respuesta, metadata, costos y log de fallback.
    """
    path, warnings = validate_image(image_path)

    analysis_pattern = AnalysisPattern(pattern)
    prompt = get_prompt(analysis_pattern, question)

    result = route_with_fallback(path, prompt, priority, provider)

    result.pattern = analysis_pattern.value
    result.estimated_cost_usd = calculate_cost(result)
    result.warnings.extend(warnings)

    logger.info(
        f"Análisis completado: pattern={result.pattern} provider={result.provider} "
        f"model={result.model} tokens={result.input_tokens}+{result.output_tokens} "
        f"cost=${result.estimated_cost_usd:.6f} latency={result.latency_seconds}s"
    )

    return result

Código de Demo

def print_result(result: AnalysisResult):
    print(f"\n{'='*60}")
    print(f"  Patrón:     {result.pattern}")
    print(f"  Proveedor:  {result.provider} ({result.model})")
    print(f"  Latencia:   {result.latency_seconds}s")
    print(f"  Tokens:     {result.input_tokens} in / {result.output_tokens} out")
    print(f"  Costo:      ${result.estimated_cost_usd:.6f}")
    if result.warnings:
        for w in result.warnings:
            print(f"  ⚠ {w}")
    if result.fallback_log:
        print(f"  Fallback:   {len(result.fallback_log)} intentos fallidos")
    print(f"  Respuesta:  {result.response[:200]}...")
    print(f"{'='*60}")


# --- Demo ---
if __name__ == "__main__":
    # Descripción con auto-selección de proveedor
    r1 = analyze_image("./sample.jpg", pattern="description", priority="balanced")
    print_result(r1)

    # OCR priorizando calidad
    r2 = analyze_image("./factura.png", pattern="ocr", priority="quality")
    print_result(r2)

    # Clasificación priorizando costo
    r3 = analyze_image("./screenshot.png", pattern="classification", priority="cost")
    print_result(r3)

    # Q&A con proveedor específico
    r4 = analyze_image(
        "./diagrama.png",
        pattern="qa",
        provider="anthropic",
        question="¿Cuántos nodos tiene el diagrama?",
    )
    print_result(r4)

    # Extracción estructurada
    r5 = analyze_image("./recibo.jpg", pattern="extraction", priority="quality")
    print_result(r5)

Salida esperada (esquemática):

============================================================
  Patrón:     description
  Proveedor:  openai (gpt-4o-mini)
  Latencia:   1.34s
  Tokens:     1150 in / 87 out
  Costo:      $0.000225
  Respuesta:  La imagen muestra un paisaje urbano al atardecer...
============================================================

Extensión 1: Batch Analysis

Procesar múltiples imágenes con tracking de costos acumulados y reporte de progreso.

@dataclass
class BatchResult:
    total: int = 0
    successful: int = 0
    failed: int = 0
    total_cost_usd: float = 0.0
    total_latency_seconds: float = 0.0
    by_provider: dict = field(default_factory=dict)
    results: list[AnalysisResult] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


def analyze_batch(
    image_paths: list[str],
    pattern: str = "description",
    priority: str = "cost",
    budget: float | None = None,
) -> BatchResult:
    """Analiza múltiples imágenes con tracking de costo y progreso."""
    batch = BatchResult()

    for i, img_path in enumerate(image_paths):
        batch.total += 1
        logger.info(f"[{i+1}/{len(image_paths)}] Procesando: {img_path}")

        if budget and batch.total_cost_usd >= budget:
            batch.errors.append(f"Presupuesto agotado (${budget:.4f}) en imagen {i+1}")
            break

        try:
            result = analyze_image(img_path, pattern=pattern, priority=priority)
            batch.successful += 1
            batch.total_cost_usd += result.estimated_cost_usd
            batch.total_latency_seconds += result.latency_seconds
            batch.by_provider[result.provider] = batch.by_provider.get(result.provider, 0) + 1
            batch.results.append(result)
        except Exception as e:
            batch.failed += 1
            batch.errors.append(f"{img_path}: {e}")

    return batch


Uso:

```python
images = ["./img1.jpg", "./img2.png", "./img3.webp"]
batch = analyze_batch(images, pattern="classification", priority="cost", budget=0.05)
print(f"Exitosos: {batch.successful}/{batch.total}, Costo: ${batch.total_cost_usd:.6f}")

Extensión 2: Consensus Voting

Envía la misma imagen a los 3 proveedores y combina los resultados. Útil para tareas de clasificación donde quieres alta confianza.

@dataclass
class ConsensusResult:
    individual_results: list[AnalysisResult] = field(default_factory=list)
    consensus_response: str = ""
    agreement_score: float = 0.0
    total_cost_usd: float = 0.0
    total_latency_seconds: float = 0.0


def analyze_with_consensus(
    image_path: str,
    pattern: str = "classification",
    question: str | None = None,
) -> ConsensusResult:
    """Envía la imagen a los 3 proveedores y vota por consenso."""
    path, warnings = validate_image(image_path)
    prompt = get_prompt(AnalysisPattern(pattern), question)
    consensus = ConsensusResult()

    providers = [
        ("openai", PROVIDER_CONFIG["openai"]["model_cost"]),
        ("anthropic", PROVIDER_CONFIG["anthropic"]["model_cost"]),
        ("google", PROVIDER_CONFIG["google"]["model_cost"]),
    ]

    for provider_name, model in providers:
        caller = CALLER_MAP[provider_name]
        try:
            result = caller(path, prompt, model)
            result.pattern = pattern
            result.estimated_cost_usd = calculate_cost(result)
            consensus.individual_results.append(result)
            consensus.total_cost_usd += result.estimated_cost_usd
            consensus.total_latency_seconds += result.latency_seconds
        except Exception as e:
            logger.warning(f"Consensus: {provider_name} falló: {e}")

    if not consensus.individual_results:
        raise RuntimeError("Ningún proveedor respondió para consensus")

    responses = [r.response.strip().lower() for r in consensus.individual_results]
    from collections import Counter
    vote_counts = Counter(responses)
    winner, count = vote_counts.most_common(1)[0]
    consensus.consensus_response = winner
    consensus.agreement_score = count / len(responses)

    return consensus

Troubleshooting del Proyecto

Problema 1: Anthropic rechaza imágenes grandes

Síntoma: anthropic.BadRequestError con imágenes mayores a ~4 MB.

Causa: Anthropic tiene un límite de 5 MB por imagen, pero el encoding Base64 aumenta el tamaño ~33%. Una imagen de 4 MB genera ~5.3 MB en Base64.

Solución: Verificar el tamaño antes de enviar y redimensionar si es necesario:

def resize_if_needed(path: Path, max_raw_mb: float = 3.5) -> bytes:
    from PIL import Image
    import io

    raw_bytes = path.read_bytes()
    if len(raw_bytes) / (1024 * 1024) <= max_raw_mb:
        return raw_bytes

    with Image.open(path) as img:
        img = img.convert("RGB")
        quality = 85
        while quality >= 30:
            buffer = io.BytesIO()
            img.save(buffer, "JPEG", quality=quality, optimize=True)
            if len(buffer.getvalue()) / (1024 * 1024) <= max_raw_mb:
                return buffer.getvalue()
            quality -= 10
        img = img.resize((img.width // 2, img.height // 2))
        buffer = io.BytesIO()
        img.save(buffer, "JPEG", quality=80, optimize=True)
        return buffer.getvalue()

Problema 2: Gemini no reporta tokens

Síntoma: input_tokens y output_tokens son 0 después de llamar a Gemini.

Causa: No todos los modelos de Gemini incluyen usage_metadata en la respuesta.

Solución: El código ya maneja esto con getattr y defaults a 0. Para estimar tokens cuando no se reportan:

if result.input_tokens == 0 and result.provider == "google":
    size_kb = path.stat().st_size / 1024
    result.input_tokens = max(258, int(size_kb * 1.5))
    result.output_tokens = len(result.response.split()) * 2
    result.warnings.append("Tokens estimados (Gemini no reportó usage)")

Problema 3: Rate limit con todos los proveedores simultáneamente

Síntoma: En consensus voting, los 3 proveedores fallan por rate limit si envías muchas imágenes rápido.

Solución: Agregar un delay entre llamadas en el loop de consensus:

import time

for provider_name, model in providers:
    try:
        result = caller(path, prompt, model)
        consensus.individual_results.append(result)
    except Exception as e:
        logger.warning(f"{provider_name} falló: {e}")
    time.sleep(0.5)

Problema 4: `FileNotFoundError` con rutas relativas

Síntoma: La ruta "./images/foto.jpg" falla aunque el archivo existe.

Causa: El working directory del script no es el que esperas.

Solución: Usa rutas absolutas o resuelve relativas:

path = Path(image_path).resolve()

Problema 5: Respuestas inconsistentes entre proveedores para OCR

Síntoma: OpenAI extrae texto correctamente, pero Gemini omite secciones o inventa texto.

Causa: Los modelos tienen diferente rendimiento en OCR. GPT-4o y Claude 3.5 Sonnet son superiores en OCR preciso.

Solución: Para OCR en producción, prioriza OpenAI o Anthropic. Usa analyze_image(path, pattern="ocr", priority="quality", provider="openai").

Checklist de Completitud

Funcionalidad core:

analyze_image() acepta los 5 parámetros (image_path, pattern, provider, priority, question)
Retorna AnalysisResult con todos los campos poblados
Soporta los 5 patrones: description, ocr, classification, extraction, qa
Patrón qa valida que question no sea None

Proveedores:

_call_openai() envía imagen como data URI y parsea usage
_call_anthropic() envía como Base64 con media_type y valida tamaño
_call_gemini() envía como bytes crudos y maneja ausencia de usage_metadata
Los 3 proveedores retornan AnalysisResult con tokens y latencia

Fallback:

Router intenta proveedores en orden según priority
Si el preferido falla, pasa al siguiente
Cada fallo se registra en fallback_log
Si todos fallan, lanza RuntimeError con el log completo

Cálculo de costos:

calculate_cost() usa tokens reales y precios por proveedor/modelo
Costo se incluye en el resultado final
estimate_cost_before_call() funciona sin hacer llamada API

Validación y errores:

Imagen inexistente → FileNotFoundError
Formato no soportado → ValueError
Imagen > 20 MB → ValueError
Imagen grande genera warning (no error)

Extensiones:

Batch analysis con budget y reporte
Consensus voting con agreement_score

Ejercicios

Ejercicio 1: Cache de análisis (Medio)

Implementa una capa de cache que evite re-analizar la misma combinación de imagen + patrón + proveedor. Usa un hash del contenido de la imagen (no la ruta) como clave. El cache debe persistir en disco como JSON.

Requisitos:

Hash SHA-256 del contenido de la imagen
Clave compuesta: {image_hash}_{pattern}_{provider}
Almacenamiento en ./cache/analysis_cache.json
Método get() y set() en una clase AnalysisCache
TTL configurable (default 24 horas)

Ver solución

import json
import hashlib
from datetime import datetime, timedelta


class AnalysisCache:
    def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
        self.cache_path = Path(cache_dir) / "analysis_cache.json"
        self.cache_path.parent.mkdir(parents=True, exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
        self._cache = self._load()

    def _load(self) -> dict:
        if self.cache_path.exists():
            return json.loads(self.cache_path.read_text())
        return {}

    def _save(self):
        self.cache_path.write_text(json.dumps(self._cache, indent=2, default=str))

    def _image_hash(self, image_path: str) -> str:
        return hashlib.sha256(Path(image_path).read_bytes()).hexdigest()[:16]

    def _key(self, image_path: str, pattern: str, provider: str) -> str:
        return f"{self._image_hash(image_path)}_{pattern}_{provider}"

    def get(self, image_path: str, pattern: str, provider: str) -> AnalysisResult | None:
        key = self._key(image_path, pattern, provider)
        entry = self._cache.get(key)
        if not entry:
            return None
        cached_at = datetime.fromisoformat(entry["cached_at"])
        if datetime.now() - cached_at > self.ttl:
            del self._cache[key]
            self._save()
            return None
        data = entry["result"]
        result = AnalysisResult(**{k: v for k, v in data.items() if k in AnalysisResult.__dataclass_fields__})
        result.warnings.append("Resultado desde cache")
        return result

    def set(self, image_path: str, result: AnalysisResult):
        key = self._key(image_path, result.pattern, result.provider)
        self._cache[key] = {
            "cached_at": datetime.now().isoformat(),
            "result": {
                "response": result.response,
                "pattern": result.pattern,
                "provider": result.provider,
                "model": result.model,
                "latency_seconds": result.latency_seconds,
                "input_tokens": result.input_tokens,
                "output_tokens": result.output_tokens,
                "estimated_cost_usd": result.estimated_cost_usd,
            },
        }
        self._save()


cache = AnalysisCache()

def analyze_image_cached(image_path: str, pattern: str = "description", **kwargs) -> AnalysisResult:
    provider = kwargs.get("provider", "auto")
    cached = cache.get(image_path, pattern, provider)
    if cached:
        logger.info(f"Cache hit: {pattern}/{provider}")
        return cached
    result = analyze_image(image_path, pattern=pattern, **kwargs)
    cache.set(image_path, result)
    return result

Ejercicio 2: Evaluación automática de calidad (Difícil)

Implementa un sistema que envíe la misma imagen a los 3 proveedores y luego use un LLM para evaluar la calidad de cada respuesta con un score de 1 a 10 en: precisión, completitud y claridad.