8. Proyecto: Clasificador Multimodal

Descripción

Este proyecto cierra el Módulo 1 integrando todo lo que aprendiste: modalidades (visión, audio), formatos (Base64, URLs), APIs, costos, rate limits y troubleshooting. Vas a construir un Clasificador Multimodal — un sistema que recibe cualquier input (texto, ruta a imagen, ruta a audio, URL), detecta la modalidad, valida el formato, recomienda el modelo y proveedor óptimo, estima el costo, y prepara el input en el formato correcto para la API.

No es un toy example. Es el patrón que usan sistemas reales de procesamiento de contenido: un router inteligente que decide qué pipeline aplicar antes de hacer la llamada a la API.

Por qué importa: En producción, los inputs llegan en cualquier formato. Un endpoint recibe una imagen JPEG, un audio MP3, o un texto plano, y debe decidir qué hacer con cada uno. Sin un clasificador previo, terminas con if/elif chains frágiles que fallan con el primer formato inesperado.

Conexión con el módulo: Cada componente del clasificador viene de una cápsula anterior:

• Detección de modalidad → Cápsulas 02 (visión), 03 (audio), 04 (combinaciones)
• Selección de modelo → Cápsula 05 (landscape de modelos)
• Validación de formato y encoding → Cápsula 06 (formatos y APIs)
• Manejo de errores → Cápsula 07 (troubleshooting)

Conexión con la guía: En módulos posteriores, este clasificador evoluciona. En el Módulo 2, el pipeline de visión usa lógica similar para decidir entre GPT-4 Vision, Claude 3 y Gemini. En el Módulo 8 (Document Analyzer), el sistema detecta tipo de documento y enruta al pipeline correcto.

Especificaciones Técnicas

Input

El clasificador acepta un único argumento input_value: str que puede ser:

Output

@dataclass
class ClassificationResult:
    modality: str              # "text" | "image" | "audio"
    source_type: str           # "raw" | "local_file" | "url"
    recommended_model: str     # "gpt-4o" | "gpt-4o-mini" | "whisper-1" | etc.
    recommended_provider: str  # "openai" | "anthropic" | "google"
    estimated_cost_usd: float  # Costo estimado de procesar este input
    format_valid: bool         # Si el formato pasó validación
    metadata: dict             # Tamaño, formato, dimensiones, duración, etc.
    warnings: list[str]        # Advertencias (archivo grande, formato subóptimo)
    api_payload: dict | None   # Payload listo para enviar a la API

Requisitos funcionales

1. Detectar modalidad por extensión, URL pattern, o contenido
2. Validar formato antes de clasificar (formato real, no solo extensión)
3. Recomendar modelo y proveedor según modalidad, tamaño y balance costo/calidad
4. Estimar costo basado en tokens de imagen o minutos de audio
5. Preparar payload en el formato correcto del proveedor (Base64 para Anthropic, URL o Base64 para OpenAI)
6. Manejar errores con mensajes claros (archivo no existe, formato inválido, excede límite de tamaño)

Paso 1: Constantes y Configuración

from dataclasses import dataclass, field
from pathlib import Path
import base64, re, math

IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".mp4", ".mpeg", ".webm", ".ogg", ".flac"}
MIME_TYPES = {
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
    ".gif": "image/gif", ".webp": "image/webp",
    ".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4",
    ".ogg": "audio/ogg", ".flac": "audio/flac",
}

SIZE_LIMITS_MB = {
    "openai": {"image": 20, "audio": 25},
    "anthropic": {"image": 5, "audio": 0},
    "google": {"image": 20, "audio": 0},
}

MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "whisper-1": {"per_minute": 0.006},
    "claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
    "gemini-1.5-flash": {"input": 0.075, "output": 0.30},
}


@dataclass
class ClassificationResult:
    modality: str = "unknown"
    source_type: str = "unknown"
    recommended_model: str = ""
    recommended_provider: str = ""
    estimated_cost_usd: float = 0.0
    format_valid: bool = False
    metadata: dict = field(default_factory=dict)
    warnings: list[str] = field(default_factory=list)
    api_payload: dict | None = None
    error: str | None = None

La separación entre constantes y lógica permite cambiar precios o formatos soportados sin tocar la lógica de clasificación.

Paso 2: Detectar Modalidad

La detección sigue un orden de prioridad: primero verifica si es URL, luego si es archivo local, y como fallback asume texto.

def detect_modality(input_value: str) -> tuple[str, str]:
    """Detecta modalidad y tipo de source.

    Returns:
        (modality, source_type) donde modality es "text"|"image"|"audio"
        y source_type es "raw"|"local_file"|"url"
    """
    if re.match(r"https?://", input_value):
        ext = _extract_extension_from_url(input_value)
        if ext in IMAGE_EXTENSIONS:
            return "image", "url"
        if ext in AUDIO_EXTENSIONS:
            return "audio", "url"
        return "text", "url"

    path = Path(input_value)
    if path.exists() and path.is_file():
        ext = path.suffix.lower()
        if ext in IMAGE_EXTENSIONS:
            return "image", "local_file"
        if ext in AUDIO_EXTENSIONS:
            return "audio", "local_file"

    return "text", "raw"


def _extract_extension_from_url(url: str) -> str:
    """Extrae extensión de una URL, ignorando query params."""
    clean = url.split("?")[0].split("#")[0]
    path = Path(clean)
    return path.suffix.lower()

Decisión de diseño: URLs sin extensión reconocida se clasifican como texto. Esto es intencional — si no puedes determinar la modalidad sin descargar el contenido, es más seguro no asumir. El caller puede forzar la modalidad si tiene información adicional.

Paso 3: Recolectar Metadata

La metadata varía según la modalidad y el tipo de source:

def collect_metadata(input_value: str, modality: str, source_type: str) -> dict:
    """Recolecta metadata relevante según la modalidad."""
    if modality == "text":
        return {
            "length_chars": len(input_value),
            "length_words": len(input_value.split()),
            "estimated_tokens": len(input_value) // 4,
        }

    if source_type == "url":
        return {"url": input_value, "format": _extract_extension_from_url(input_value)[1:]}

    path = Path(input_value)
    size_bytes = path.stat().st_size
    meta = {
        "file_path": str(path),
        "file_size_kb": round(size_bytes / 1024, 1),
        "file_size_mb": round(size_bytes / (1024 * 1024), 2),
        "format": path.suffix[1:].lower(),
    }

    if modality == "image":
        meta.update(_get_image_metadata(path))
    elif modality == "audio":
        meta.update(_get_audio_metadata(path, size_bytes))

    return meta


def _get_image_metadata(path: Path) -> dict:
    """Obtiene dimensiones de la imagen si PIL está disponible."""
    try:
        from PIL import Image
        with Image.open(path) as img:
            return {"width": img.width, "height": img.height, "mode": img.mode}
    except ImportError:
        return {"width": None, "height": None, "note": "PIL no instalado"}
    except Exception:
        return {"width": None, "height": None, "note": "No se pudo leer dimensiones"}


def _get_audio_metadata(path: Path, size_bytes: int) -> dict:
    """Estima duración del audio por tamaño (heurística sin pydub)."""
    format_bitrates_kbps = {"mp3": 128, "wav": 1411, "m4a": 128, "ogg": 112, "flac": 800}
    ext = path.suffix[1:].lower()
    bitrate = format_bitrates_kbps.get(ext, 128)
    estimated_seconds = (size_bytes * 8) / (bitrate * 1000)
    return {
        "estimated_duration_seconds": round(estimated_seconds, 1),
        "estimated_duration_minutes": round(estimated_seconds / 60, 2),
    }

Para imágenes intentamos obtener dimensiones con PIL (opcional). Para audio estimamos duración por tamaño — una heurística que funciona razonablemente bien sin depender de pydub o ffprobe.

Paso 4: Validar Formato

La validación verifica que el archivo sea realmente lo que dice la extensión y que no exceda los límites del proveedor:

def validate_format(
    input_value: str, modality: str, source_type: str, metadata: dict, provider: str
) -> tuple[bool, list[str]]:
    """Valida formato y retorna (is_valid, warnings)."""
    warnings = []

    if source_type == "url":
        warnings.append("URL no validada localmente — errores se detectarán al llamar la API")
        return True, warnings

    if source_type == "raw":
        return True, warnings

    path = Path(input_value)
    if not path.exists():
        return False, [f"Archivo no encontrado: {input_value}"]

    size_mb = metadata.get("file_size_mb", 0)
    limit = SIZE_LIMITS_MB.get(provider, {}).get(modality, 20)

    if limit == 0 and modality == "audio":
        return False, [f"{provider} no soporta audio directamente"]

    if size_mb > limit:
        return False, [f"Archivo excede límite de {provider}: {size_mb:.1f} MB > {limit} MB"]

    if size_mb > limit * 0.8:
        warnings.append(f"Archivo cerca del límite ({size_mb:.1f}/{limit} MB) — considera redimensionar")

    if modality == "image":
        _validate_image_format(path, warnings)

    return True, warnings


def _validate_image_format(path: Path, warnings: list[str]):
    """Verifica que el contenido de la imagen coincida con la extensión."""
    try:
        from PIL import Image
        with Image.open(path) as img:
            img.verify()
    except ImportError:
        warnings.append("PIL no instalado — no se verificó integridad de imagen")
    except Exception as e:
        warnings.append(f"Imagen posiblemente corrupta: {e}")

Punto clave: La validación es conservadora con URLs (no las descarga solo para validar) y estricta con archivos locales (verifica integridad con PIL cuando está disponible).

Paso 5: Recomendar Modelo y Proveedor

La recomendación considera modalidad, tamaño, y balance costo/calidad:

def recommend_model(
    modality: str, metadata: dict, priority: str = "balanced"
) -> tuple[str, str, str]:
    """Recomienda modelo y proveedor.

    Args:
        priority: "quality" | "cost" | "balanced"

    Returns:
        (model, provider, reason)
    """
    if modality == "text":
        if priority == "quality":
            return "gpt-4o", "openai", "Texto complejo: modelo de alta capacidad"
        return "gpt-4o-mini", "openai", "Texto: modelo económico con buena calidad"

    if modality == "audio":
        return "whisper-1", "openai", "Audio: Whisper es el estándar para STT"

    if modality == "image":
        return _recommend_vision_model(metadata, priority)

    return "gpt-4o-mini", "openai", "Fallback: modalidad no reconocida"


def _recommend_vision_model(metadata: dict, priority: str) -> tuple[str, str, str]:
    """Selecciona modelo de visión según tamaño y prioridad."""
    size_mb = metadata.get("file_size_mb", 0)

    if priority == "cost":
        if size_mb > 5:
            return "gemini-1.5-flash", "google", "Imagen grande + prioridad costo: Gemini Flash"
        return "gpt-4o-mini", "openai", "Imagen: gpt-4o-mini es 17x más barato que gpt-4o"

    if priority == "quality":
        return "gpt-4o", "openai", "Imagen + prioridad calidad: GPT-4 Vision"

    if size_mb > 10:
        return "gpt-4o-mini", "openai", "Imagen grande (>10MB): modelo económico para reducir costo"

    return "gpt-4o-mini", "openai", "Imagen estándar: buena calidad a bajo costo"

Decisión de diseño: El parámetro priority permite al caller ajustar la recomendación. Por defecto es "balanced" (gpt-4o-mini para la mayoría de casos), pero el caller puede pedir "quality" para tareas críticas o "cost" para batches grandes.

Paso 6: Estimar Costo

def estimate_cost(modality: str, model: str, metadata: dict) -> float:
    """Estima costo de procesar el input con el modelo recomendado."""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return 0.0

    if modality == "audio":
        minutes = metadata.get("estimated_duration_minutes", 1.0)
        return round(minutes * pricing["per_minute"], 6)

    if modality == "image":
        image_tokens = _estimate_image_tokens(metadata)
        prompt_tokens = 50
        output_tokens = 150
        input_cost = ((image_tokens + prompt_tokens) / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    if modality == "text":
        input_tokens = metadata.get("estimated_tokens", 100)
        output_tokens = min(input_tokens, 500)
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    return 0.0


def _estimate_image_tokens(metadata: dict) -> int:
    """Estima tokens de imagen usando el sistema de tiles de OpenAI."""
    width = metadata.get("width")
    height = metadata.get("height")

    if not width or not height:
        return 765  # default: 1 tile (cápsula 06)

    if max(width, height) <= 512:
        return 85  # low detail

    scale = min(2048 / max(width, height), 1.0)
    w, h = int(width * scale), int(height * scale)

    scale_short = 768 / min(w, h)
    if scale_short < 1:
        w, h = int(w * scale_short), int(h * scale_short)

    tiles_w = math.ceil(w / 512)
    tiles_h = math.ceil(h / 512)
    return 85 + (170 * tiles_w * tiles_h)

Este cálculo de tokens replica exactamente la lógica de la cápsula 06: escalar a 2048px, luego a 768px del lado corto, y contar tiles de 512×512.

Paso 7: Preparar API Payload

def prepare_api_payload(
    input_value: str, modality: str, source_type: str, provider: str
) -> dict | None:
    """Prepara el payload en el formato del proveedor."""
    if modality == "text":
        return {"type": "text", "text": input_value}

    if modality == "audio":
        if source_type == "local_file":
            return {"type": "audio", "file_path": input_value, "model": "whisper-1"}
        return None

    if modality == "image":
        return _prepare_image_payload(input_value, source_type, provider)

    return None


def _prepare_image_payload(input_value: str, source_type: str, provider: str) -> dict:
    """Prepara payload de imagen según proveedor."""
    if source_type == "url":
        if provider == "anthropic":
            return {
                "type": "image",
                "note": "Anthropic requiere Base64 — descarga la URL primero",
                "url": input_value,
            }
        return {"type": "image_url", "image_url": {"url": input_value}}

    path = Path(input_value)
    with open(path, "rb") as f:
        encoded = base64.b64encode(f.read()).decode("utf-8")
    mime = MIME_TYPES.get(path.suffix.lower(), "image/jpeg")

    if provider == "anthropic":
        return {
            "type": "image",
            "source": {"type": "base64", "media_type": mime, "data": encoded},
        }

    data_uri = f"data:{mime};base64,{encoded}"
    return {"type": "image_url", "image_url": {"url": data_uri}}

El payload respeta las diferencias entre proveedores: Anthropic necesita media_type + data separados; OpenAI acepta data URIs.

Paso 8: Función Principal — Integración

def classify_multimodal(
    input_value: str,
    priority: str = "balanced",
    provider: str | None = None,
    prepare_payload: bool = True,
) -> ClassificationResult:
    """Clasifica un input multimodal y produce resultado completo.

    Args:
        input_value: Texto, ruta a archivo, o URL
        priority: "quality" | "cost" | "balanced"
        provider: Forzar proveedor específico. Si None, se auto-selecciona.
        prepare_payload: Si True, genera el payload listo para la API.
    """
    result = ClassificationResult()

    modality, source_type = detect_modality(input_value)
    result.modality = modality
    result.source_type = source_type

    result.metadata = collect_metadata(input_value, modality, source_type)

    model, rec_provider, reason = recommend_model(modality, result.metadata, priority)
    result.recommended_model = model
    result.recommended_provider = provider or rec_provider

    is_valid, warnings = validate_format(
        input_value, modality, source_type, result.metadata, result.recommended_provider
    )
    result.format_valid = is_valid
    result.warnings = warnings

    if not is_valid:
        result.error = warnings[0] if warnings else "Formato inválido"
        return result

    result.estimated_cost_usd = estimate_cost(modality, model, result.metadata)

    if prepare_payload and source_type != "url":
        try:
            result.api_payload = prepare_api_payload(
                input_value, modality, source_type, result.recommended_provider
            )
        except Exception as e:
            result.warnings.append(f"No se pudo preparar payload: {e}")

    return result

Código Completo

El código completo integra todos los pasos anteriores en un único archivo ejecutable. Lo que sigue es la interfaz de uso y un demo:

def print_classification(result: ClassificationResult):
    """Imprime resultado de clasificación de forma legible."""
    status = "VÁLIDO" if result.format_valid else "INVÁLIDO"
    print(f"\n{'='*60}")
    print(f"  Modalidad:   {result.modality}")
    print(f"  Source:      {result.source_type}")
    print(f"  Formato:     {status}")
    print(f"  Modelo:      {result.recommended_model} ({result.recommended_provider})")
    print(f"  Costo est.:  ${result.estimated_cost_usd:.6f}")

    if result.metadata:
        print(f"  Metadata:")
        for k, v in result.metadata.items():
            print(f"    {k}: {v}")

    if result.warnings:
        print(f"  Warnings:")
        for w in result.warnings:
            print(f"    ⚠ {w}")

    if result.error:
        print(f"  ERROR: {result.error}")

    if result.api_payload:
        payload_type = result.api_payload.get("type", "?")
        print(f"  Payload:     {payload_type} (listo para API)")

    print(f"{'='*60}")


# Demo
inputs = [
    "Analiza las tendencias de ventas del Q4",
    "./fotos/producto_laptop.jpg",
    "./grabaciones/llamada_cliente.mp3",
    "https://upload.wikimedia.org/wikipedia/commons/3/3a/Cat03.jpg",
    "./archivo_inexistente.png",
]

for inp in inputs:
    result = classify_multimodal(inp, priority="balanced")
    print_classification(result)

Salida esperada (esquemática):

============================================================
  Modalidad:   text
  Source:      raw
  Formato:     VÁLIDO
  Modelo:      gpt-4o-mini (openai)
  Costo est.:  $0.000005
  Metadata:
    length_chars: 46
    length_words: 8
    estimated_tokens: 11
  Payload:     text (listo para API)
============================================================

============================================================
  Modalidad:   image
  Source:      local_file
  Formato:     VÁLIDO
  Modelo:      gpt-4o-mini (openai)
  Costo est.:  $0.000124
  Metadata:
    file_path: ./fotos/producto_laptop.jpg
    file_size_kb: 245.3
    file_size_mb: 0.24
    format: jpg
    width: 1024
    height: 768
  Payload:     image_url (listo para API)
============================================================

Extensión 1: Clasificación en Batch

Para procesar múltiples inputs, agrega tracking de costos acumulados:

@dataclass
class BatchReport:
    total: int = 0
    by_modality: dict = field(default_factory=lambda: {"text": 0, "image": 0, "audio": 0})
    valid: int = 0
    invalid: int = 0
    total_cost_usd: float = 0.0
    results: list[ClassificationResult] = field(default_factory=list)


def classify_batch(
    inputs: list[str], priority: str = "balanced", budget: float | None = None
) -> BatchReport:
    """Clasifica múltiples inputs con tracking de costos."""
    report = BatchReport()

    for inp in inputs:
        result = classify_multimodal(inp, priority=priority)
        report.total += 1
        report.by_modality[result.modality] = report.by_modality.get(result.modality, 0) + 1

        if result.format_valid:
            report.valid += 1
        else:
            report.invalid += 1

        if budget and (report.total_cost_usd + result.estimated_cost_usd) > budget:
            result.warnings.append(f"Presupuesto excedido: ${report.total_cost_usd:.4f}/{budget}")
            result.api_payload = None

        report.total_cost_usd += result.estimated_cost_usd
        report.results.append(result)

    return report


def print_batch_report(report: BatchReport):
    print(f"\n{'='*60}")
    print(f"  REPORTE DE BATCH")
    print(f"{'='*60}")
    print(f"  Total inputs:  {report.total}")
    print(f"  Válidos:       {report.valid}")
    print(f"  Inválidos:     {report.invalid}")
    print(f"  Por modalidad: {report.by_modality}")
    print(f"  Costo total:   ${report.total_cost_usd:.6f}")
    print(f"{'='*60}")

Extensión 2: Modo Comparativo Multi-Proveedor

Para decidir entre proveedores, clasifica el mismo input con cada uno y compara:

def compare_providers(
    input_value: str, providers: list[str] = None
) -> list[dict]:
    """Compara clasificación entre proveedores."""
    providers = providers or ["openai", "anthropic", "google"]
    comparisons = []

    for provider in providers:
        result = classify_multimodal(input_value, provider=provider)
        comparisons.append({
            "provider": provider,
            "model": result.recommended_model,
            "cost": result.estimated_cost_usd,
            "valid": result.format_valid,
            "warnings": len(result.warnings),
        })

    comparisons.sort(key=lambda x: x["cost"])
    return comparisons

Uso:

for c in compare_providers("./fotos/producto.jpg"):
    status = "✓" if c["valid"] else "✗"
    print(f"  {status} {c['provider']:<12} {c['model']:<20} ${c['cost']:.6f}")

Troubleshooting del Proyecto

Problema 1: Archivo de imagen con extensión incorrecta

Síntoma: Una imagen .txt que en realidad es JPEG se clasifica como texto.

Solución: Agrega detección por magic bytes como segundo nivel:

MAGIC_BYTES = {
    b"\xff\xd8\xff": "image",      # JPEG
    b"\x89PNG": "image",            # PNG
    b"GIF8": "image",              # GIF
    b"RIFF": "audio",             # WAV
    b"ID3": "audio",              # MP3
    b"\xff\xfb": "audio",         # MP3 (sin ID3)
}

def detect_by_magic_bytes(path: Path) -> str | None:
    """Detecta modalidad por los primeros bytes del archivo."""
    try:
        with open(path, "rb") as f:
            header = f.read(8)
        for magic, modality in MAGIC_BYTES.items():
            if header.startswith(magic):
                return modality
    except Exception:
        pass
    return None

Integra esto en detect_modality como fallback cuando la extensión no es reconocida pero el archivo existe.

Problema 2: URL con query params se clasifica mal

Síntoma: https://images.com/foto.jpg?w=800&h=600 funciona, pero https://api.example.com/image?id=123 no tiene extensión y se clasifica como texto.

Solución: El diseño actual es correcto — sin extensión reconocible, no asumimos modalidad. El caller puede usar el parámetro provider o pasar la URL descargada localmente.

Problema 3: Estimación de costo incorrecta para imágenes sin PIL

Síntoma: Sin PIL instalado, todas las imágenes se estiman con 765 tokens (default de 1 tile).

Solución: Instala pillow (pip install Pillow). Sin PIL, la estimación es conservadora pero imprecisa. Agrega un warning:

if not width or not height:
    warnings.append("Sin PIL: costo estimado con 765 tokens (default). Instala Pillow para precisión.")

Problema 4: Audio largo estima mal la duración

Síntoma: Un podcast de 2 horas en MP3 variable bitrate se estima con duración incorrecta.

Solución: La heurística por tamaño asume bitrate constante. Para precisión real, usa pydub:

from pydub import AudioSegment
audio = AudioSegment.from_file(path)
duration_minutes = len(audio) / (1000 * 60)

Pero pydub requiere ffmpeg. La heurística es suficiente para estimaciones de costo (error ≤20%).

Problema 5: Base64 payload es demasiado grande para Anthropic

Síntoma: Imagen de 4 MB genera payload Base64 de ~5.3 MB, excediendo el límite de Anthropic (5 MB).

Solución: La función validate_format ya detecta esto al verificar contra SIZE_LIMITS_MB. Si necesitas enviar la imagen a Anthropic, redimensiona antes:

from PIL import Image
import io

def resize_for_anthropic(image_path: str, max_mb: float = 3.5) -> bytes:
    """Redimensiona para que el Base64 quepa en 5 MB."""
    with Image.open(image_path) as img:
        img = img.convert("RGB")
        quality = 90
        while True:
            buffer = io.BytesIO()
            img.save(buffer, "JPEG", quality=quality, optimize=True)
            encoded_size = len(buffer.getvalue()) * 4 / 3
            if encoded_size <= max_mb * 1024 * 1024:
                return buffer.getvalue()
            quality -= 10
            if quality < 30:
                img = img.resize((img.width // 2, img.height // 2))
                quality = 85

Checklist de Completitud

Antes de considerar el proyecto terminado, verifica cada punto:

Funcionalidad core:

• Detecta texto, imagen local, audio local, URL de imagen, URL de audio
• Retorna ClassificationResult con todos los campos poblados
• Metadata incluye tamaño, formato, y dimensiones/duración cuando es posible
• Recomendación de modelo es coherente con la modalidad y prioridad
• Estimación de costo usa el sistema de tiles para imágenes y por-minuto para audio

Validación:

• Archivo no existente → format_valid=False con mensaje claro
• Archivo que excede límite → format_valid=False con límite del proveedor
• Archivo cerca del límite → warning
• Imagen corrupta → warning (si PIL disponible)

Payload:

• Imagen local → Base64 con data URI (OpenAI) o media_type+data (Anthropic)
• Imagen por URL → URL directo (OpenAI/Google), nota de descarga (Anthropic)
• Audio → file_path para Whisper
• Texto → dict con type "text"

Extensiones (opcionales):

• Batch classification con tracking de costos
• Comparación multi-proveedor
• Detección por magic bytes

Ejercicios

Ejercicio 1: Tests del clasificador (Fácil)

Crea una función test_classifier() que pruebe los 5 tipos de input (texto, imagen local, audio local, URL imagen, URL audio) con asserts. No necesitas archivos reales — para archivos locales, crea temporales con tempfile.

import tempfile, os

def test_classifier():
    r = classify_multimodal("Hola mundo")
    assert r.modality == "text"
    assert r.source_type == "raw"
    assert r.format_valid is True

    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
        f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
        img_path = f.name
    r = classify_multimodal(img_path)
    assert r.modality == "image"
    assert r.source_type == "local_file"
    os.unlink(img_path)

    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
        f.write(b"ID3" + b"\x00" * 100)
        audio_path = f.name
    r = classify_multimodal(audio_path)
    assert r.modality == "audio"
    assert r.source_type == "local_file"
    os.unlink(audio_path)

    r = classify_multimodal("https://example.com/photo.png")
    assert r.modality == "image"
    assert r.source_type == "url"

    r = classify_multimodal("https://example.com/audio.mp3")
    assert r.modality == "audio"
    assert r.source_type == "url"

    r = classify_multimodal("./no_existe.jpg")
    assert r.modality == "text"  # archivo no existe → fallback a texto

    print("Todos los tests pasaron")

test_classifier()

Ejercicio 2: Agregar soporte para video (Medio)

Extiende el clasificador para soportar archivos de video (.mp4, .mov, .avi, .mkv). El modelo recomendado para video debe ser gemini-1.5-flash (Google es el único que soporta video directamente). Agrega una estimación de costo basada en duración estimada del video.

VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
VIDEO_BITRATES_KBPS = {"mp4": 2500, "mov": 3000, "avi": 2000, "mkv": 2500, "webm": 1500}

def detect_modality_v2(input_value: str) -> tuple[str, str]:
    if re.match(r"https?://", input_value):
        ext = _extract_extension_from_url(input_value)
        if ext in VIDEO_EXTENSIONS:
            return "video", "url"
        # ... (resto igual)

    path = Path(input_value)
    if path.exists() and path.is_file():
        ext = path.suffix.lower()
        if ext in VIDEO_EXTENSIONS:
            return "video", "local_file"
        # ... (resto igual)

    return "text", "raw"

def recommend_model_v2(modality, metadata, priority="balanced"):
    if modality == "video":
        return "gemini-1.5-flash", "google", "Video: Gemini soporta video nativo"
    return recommend_model(modality, metadata, priority)

MODEL_PRICING["gemini-1.5-flash-video"] = {"per_minute": 0.015}

Ejercicio 3: Dashboard de batch con costos acumulados (Medio)

Usando classify_batch, procesa una lista de 20 inputs mixtos y genera un reporte que muestre: distribución por modalidad (gráfico de texto), top 3 inputs más caros, costo total, y porcentaje del presupuesto usado.

def batch_dashboard(inputs: list[str], budget: float = 1.0):
    report = classify_batch(inputs, budget=budget)

    print("\n📊 DASHBOARD DE CLASIFICACIÓN")
    print(f"{'='*50}")

    total = report.total
    for mod, count in report.by_modality.items():
        bar = "█" * int(count / total * 30) if total > 0 else ""
        pct = count / total * 100 if total > 0 else 0
        print(f"  {mod:<8} {bar:<30} {count:>3} ({pct:.0f}%)")

    print(f"\n  Válidos:   {report.valid}/{total}")
    print(f"  Inválidos: {report.invalid}/{total}")

    top_costly = sorted(report.results, key=lambda r: r.estimated_cost_usd, reverse=True)[:3]
    print(f"\n  Top 3 más caros:")
    for i, r in enumerate(top_costly, 1):
        path = r.metadata.get("file_path", r.metadata.get("url", "texto"))
        print(f"    {i}. {path}: ${r.estimated_cost_usd:.6f} ({r.recommended_model})")

    pct_budget = (report.total_cost_usd / budget * 100) if budget > 0 else 0
    print(f"\n  Costo total: ${report.total_cost_usd:.6f} ({pct_budget:.1f}% del presupuesto)")
    print(f"{'='*50}")

Resumen

En este proyecto construiste un Clasificador Multimodal completo que:

• Detecta modalidad por extensión de archivo, patrón de URL, o contenido (texto como fallback)
• Valida formato verificando existencia, tamaño contra límites del proveedor, e integridad de imagen
• Recomienda modelo y proveedor según modalidad, tamaño y prioridad (calidad vs costo)
• Estima costo usando el sistema de tiles de OpenAI para imágenes y costo por minuto para audio
• Prepara payloads en el formato correcto de cada proveedor (data URI para OpenAI, media_type+data para Anthropic)
• Maneja errores con mensajes claros y warnings para situaciones ambiguas

Este clasificador es la base para los routers de pipelines que construirás en módulos posteriores. En el Módulo 2 (Vision), la lógica de selección de modelo se expande para incluir GPT-4 Vision, Claude 3 y Gemini con sus capacidades específicas. En el Módulo 8 (Document Analyzer), la detección de modalidad se integra con procesamiento de PDFs, facturas y contratos.

Próximo módulo: Módulo 2 — Vision con LLMs. Ya sabes qué es multimodal; ahora vas a dominar el análisis de imágenes con los principales proveedores.

Recursos Adicionales

1. OpenAI Vision Guide — Formatos de input y límites
2. OpenAI Pricing — Precios actualizados para estimar costos
3. Anthropic Vision Docs — Formato Base64 requerido por Claude
4. Google Gemini Vision — Soporte de video nativo
5. Python Pathlib — Manejo de rutas y extensiones
6. Pillow Documentation — Validación y manipulación de imágenes
7. Python dataclasses — Estructuras de datos para resultados