4. Combinaciones Multimodales

Descripción

Hasta ahora viste visión (imagen → texto) y audio (audio ↔ texto) como modalidades independientes. Pero el verdadero poder de la IA multimodal aparece cuando las combinas en pipelines: imagen + texto → análisis, audio → texto → LLM → imagen, documento → extracción → resumen en audio. En esta cápsula aprenderás los patrones de combinación más usados y cómo diseñar pipelines multimodales efectivos.

Por qué importa: En producción, rara vez usas una sola modalidad. Un sistema de análisis documental combina visión (leer el documento), texto (procesarlo con LLM) y audio (generar resumen hablado). Un asistente de soporte combina audio (escuchar pregunta), texto (procesarla) y visión (mostrar capturas). Dominar combinaciones es lo que diferencia un "script con API" de un "sistema multimodal".

Conexión con el módulo: Esta cápsula conecta las cápsulas 02 (visión) y 03 (audio) en una visión integrada. Los pipelines que verás aquí son la base del Clasificador multimodal (cápsula 08) y del Document Analyzer final (módulo 8).

Tipos de Combinaciones

Combinación por input

Envías múltiples modalidades como input al mismo modelo o pipeline.

Texto + Imagen → LLM con visión → Texto
"¿Qué muestra este gráfico?" + [grafico.png] → "El gráfico muestra ventas Q4..."

Combinación por pipeline

Encadenas modelos que procesan diferentes modalidades en secuencia.

Audio → Whisper → Texto → GPT-4o → Texto → TTS → Audio
[reunion.mp3] → "Resumen: ventas subieron 15%" → [resumen.mp3]

Combinación por output

Un sistema genera outputs en múltiples modalidades.

Documento PDF → Vision + LLM →
  ├── Texto: datos extraídos (JSON)
  ├── Imagen: gráfico generado (DALL-E)
  └── Audio: resumen hablado (TTS)

Los 6 Patrones de Pipelines Multimodales

Patrón 1: Vision + LLM (el más básico)

Envías imagen + prompt a un modelo con visión. Es el patrón más simple y más usado.

from openai import OpenAI
import base64

client = OpenAI()

def vision_plus_llm(image_path: str, question: str) -> str:
    """Patrón básico: imagen + pregunta → respuesta."""
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/jpeg;base64,{image_data}"
                }}
            ]
        }],
        max_tokens=500
    )
    return response.choices[0].message.content

# Uso
answer = vision_plus_llm(
    "diagrama_arquitectura.png",
    "¿Cuántos microservicios tiene este sistema y cuáles son?"
)
print(answer)

Casos de uso: Q&A sobre imágenes, análisis de documentos, descripción de productos.

Patrón 2: Audio → Texto → LLM

Transcribes audio y luego lo procesas con un LLM. Viste esto en la cápsula 03, aquí lo formalizamos.

def audio_to_analysis(audio_path: str, analysis_prompt: str) -> str:
    """Patrón: audio → transcripción → análisis con LLM."""
    # Paso 1: Transcribir
    with open(audio_path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language="es"
        ).text

    # Paso 2: Analizar
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": analysis_prompt},
            {"role": "user", "content": f"Transcripción:\n\n{transcript}"}
        ],
        max_tokens=500
    )
    return response.choices[0].message.content

# Uso: extraer action items de una reunión
result = audio_to_analysis(
    "standup.mp3",
    "Extrae los action items de esta reunión. Lista cada uno con responsable y deadline."
)

Casos de uso: Resumen de reuniones, análisis de llamadas de soporte, extracción de datos de entrevistas.

Patrón 3: Texto → LLM → Imagen

Generas texto con un LLM y luego creas una imagen basada en ese texto. Útil para automatizar contenido visual.

def text_to_visual(topic: str) -> dict:
    """Patrón: tema → descripción → imagen generada."""
    # Paso 1: LLM genera prompt para imagen
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Genera un prompt en inglés para DALL-E que cree una "
                f"ilustración profesional sobre: {topic}\n\n"
                f"El prompt debe ser descriptivo (estilo, colores, composición). "
                f"Máximo 100 palabras."
            )
        }],
        max_tokens=150
    )
    image_prompt = response.choices[0].message.content

    # Paso 2: Generar imagen
    image_response = client.images.generate(
        model="dall-e-3",
        prompt=image_prompt,
        size="1024x1024",
        quality="standard",
        n=1
    )
    image_url = image_response.data[0].url

    return {
        "topic": topic,
        "image_prompt": image_prompt,
        "image_url": image_url
    }

# Uso
result = text_to_visual("cloud computing y microservicios")
print(f"Prompt generado: {result['image_prompt']}")
print(f"Imagen: {result['image_url']}")

Casos de uso: Generación automática de thumbnails, ilustraciones para artículos, assets de marketing.

Patrón 4: Documento → Extracción → JSON

Combinas visión (para leer el documento) con LLM (para estructurar datos). Es el pipeline core del módulo 3.

import json

def document_to_structured(image_path: str, fields: dict) -> dict:
    """Patrón: imagen de documento → datos estructurados."""
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()

    fields_str = "\n".join(f"- {k}: {v}" for k, v in fields.items())

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Extrae estos campos del documento:\n{fields_str}\n\n"
                        f"Responde SOLO con JSON válido. "
                        f"Usa null si un campo no es visible."
                    )
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{image_data}",
                        "detail": "high"
                    }
                }
            ]
        }],
        max_tokens=500,
        temperature=0
    )
    return json.loads(response.choices[0].message.content)

# Uso: extraer datos de factura
data = document_to_structured("factura.jpg", {
    "numero_factura": "número de factura",
    "fecha": "fecha en formato YYYY-MM-DD",
    "total": "monto total con decimales",
    "empresa": "nombre del emisor"
})

Casos de uso: Procesamiento de facturas, formularios, recibos, contratos.

Patrón 5: Pipeline completo (audio + visión + texto)

El pipeline más complejo: combina todas las modalidades disponibles.

def full_multimodal_pipeline(
    audio_path: str = None,
    image_path: str = None,
    text_input: str = None
) -> dict:
    """Pipeline que acepta cualquier combinación de inputs."""
    context_parts = []

    # Procesar audio si existe
    if audio_path:
        with open(audio_path, "rb") as f:
            transcript = client.audio.transcriptions.create(
                model="whisper-1", file=f
            ).text
        context_parts.append(f"[Audio transcrito]: {transcript}")

    # Procesar imagen si existe
    if image_path:
        with open(image_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()

        img_desc = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": "Describe esta imagen en detalle."},
                    {"type": "image_url", "image_url": {
                        "url": f"data:image/jpeg;base64,{img_b64}"
                    }}
                ]
            }],
            max_tokens=300
        ).choices[0].message.content
        context_parts.append(f"[Imagen analizada]: {img_desc}")

    # Agregar texto si existe
    if text_input:
        context_parts.append(f"[Texto del usuario]: {text_input}")

    # Combinar y procesar
    combined = "\n\n".join(context_parts)
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Analiza toda la información proporcionada "
                    "(puede incluir audio transcrito, descripción de imagen, "
                    "y texto). Da un análisis integrado."
                )
            },
            {"role": "user", "content": combined}
        ],
        max_tokens=500
    )

    return {
        "inputs": {
            "audio": audio_path is not None,
            "image": image_path is not None,
            "text": text_input is not None
        },
        "context": combined,
        "analysis": response.choices[0].message.content
    }

Casos de uso: Asistentes multimodales, sistemas de soporte, análisis de contenido mixto.

Patrón 6: Generación → Análisis (ciclo cerrado)

Generas contenido y luego lo analizas para validar calidad.

def generate_and_validate(prompt: str) -> dict:
    """Genera imagen y la valida automáticamente."""
    # Paso 1: Generar imagen
    image_response = client.images.generate(
        model="dall-e-3",
        prompt=prompt,
        size="1024x1024",
        n=1
    )
    image_url = image_response.data[0].url

    # Paso 2: Analizar la imagen generada
    validation = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Se pidió generar: '{prompt}'\n\n"
                        f"¿La imagen cumple con lo solicitado?\n"
                        f"Responde con JSON: "
                        f'{{"cumple": true/false, "razon": "...", '
                        f'"calidad": 1-10}}'
                    )
                },
                {"type": "image_url", "image_url": {"url": image_url}}
            ]
        }],
        max_tokens=200,
        temperature=0
    ).choices[0].message.content

    return {
        "prompt": prompt,
        "image_url": image_url,
        "validation": json.loads(validation)
    }

Casos de uso: QA automático de contenido generado, pipelines de generación con validación.

Diseñando Pipelines: Principios

1. Cada paso tiene un propósito claro

❌ MALO: Audio → Texto → LLM → LLM → LLM → Texto
   (¿por qué 3 pasos de LLM?)

✅ BUENO: Audio → Whisper(transcribir) → GPT-4o-mini(resumir) → TTS(audio)
   (cada paso hace algo distinto)

2. Usa el modelo más barato que funcione

# Para tareas simples: gpt-4o-mini ($0.15/1M tokens)
# Para vision compleja: gpt-4o ($2.50/1M tokens)
# Para transcripción: whisper-1 ($0.006/min)
# Para TTS estándar: tts-1 ($15/1M chars)

3. Maneja errores en cada paso

def safe_pipeline(audio_path: str) -> dict:
    result = {"status": "ok", "errors": []}

    # Paso 1: Transcribir
    try:
        transcript = transcribe_audio(audio_path)
    except Exception as e:
        result["errors"].append(f"Transcripción falló: {e}")
        result["status"] = "partial"
        transcript = None

    # Paso 2: Resumir (solo si paso 1 tuvo éxito)
    summary = None
    if transcript:
        try:
            summary = summarize_text(transcript)
        except Exception as e:
            result["errors"].append(f"Resumen falló: {e}")
            result["status"] = "partial"

    result["transcript"] = transcript
    result["summary"] = summary
    return result

4. Mide latencia y costo

import time

def timed_pipeline(func, *args, **kwargs):
    start = time.time()
    result = func(*args, **kwargs)
    elapsed = time.time() - start
    return {
        "result": result,
        "latency_seconds": round(elapsed, 2)
    }

Troubleshooting

Problema 1: Pipeline lento (>30 segundos)

Causa: Múltiples llamadas API secuenciales.

Solución: Paralelizar pasos independientes con asyncio o concurrent.futures.

from concurrent.futures import ThreadPoolExecutor

def parallel_analysis(image_path: str, audio_path: str):
    with ThreadPoolExecutor(max_workers=2) as executor:
        image_future = executor.submit(describe_image, image_path)
        audio_future = executor.submit(transcribe_audio, audio_path)

        image_desc = image_future.result()
        transcript = audio_future.result()

    return {"image": image_desc, "audio": transcript}

Problema 2: Costo alto en pipelines complejos

Causa: Usar gpt-4o para todo.

Solución: Usar modelos económicos para pasos intermedios:

• Clasificación/routing → gpt-4o-mini
• Análisis profundo → gpt-4o
• Transcripción → whisper-1

Problema 3: Resultado inconsistente entre ejecuciones

Causa: temperature > 0 en pasos de extracción.

Solución: Usar temperature=0 en pasos que requieren determinismo (extracción JSON, clasificación).

Ejercicios

Ejercicio 1: Pipeline imagen → descripción → audio (Fácil)

Crea un pipeline que: recibe imagen → la describe con vision → genera audio de la descripción.

def image_to_audio_description(image_path: str) -> dict:
    # 1. Describir imagen
    description = vision_plus_llm(
        image_path,
        "Describe esta imagen en 3 oraciones claras y concisas."
    )

    # 2. Generar audio
    response = client.audio.speech.create(
        model="tts-1",
        voice="nova",
        input=description
    )
    audio_path = "image_description.mp3"
    response.stream_to_file(audio_path)

    return {
        "description": description,
        "audio": audio_path
    }

Ejercicio 2: Router de modalidad (Medio)

Crea una función que detecte si un archivo es imagen, audio o texto, y aplique el procesamiento correcto.

from pathlib import Path

IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".mp4", ".webm"}
TEXT_EXTENSIONS = {".txt", ".md", ".csv"}

def route_and_process(file_path: str, prompt: str = "Analiza este contenido.") -> dict:
    ext = Path(file_path).suffix.lower()

    if ext in IMAGE_EXTENSIONS:
        result = vision_plus_llm(file_path, prompt)
        return {"modality": "image", "result": result}

    elif ext in AUDIO_EXTENSIONS:
        result = audio_to_analysis(file_path, prompt)
        return {"modality": "audio", "result": result}

    elif ext in TEXT_EXTENSIONS:
        with open(file_path, "r") as f:
            text = f.read()
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": prompt},
                {"role": "user", "content": text}
            ],
            max_tokens=500
        )
        return {"modality": "text", "result": response.choices[0].message.content}

    else:
        return {"modality": "unknown", "result": f"Formato no soportado: {ext}"}

Ejercicio 3: Pipeline con fallback (Medio)

Crea un pipeline de análisis de imagen que intente GPT-4o primero, y si falla (error, timeout), use gpt-4o-mini como fallback.

import time

def analyze_with_fallback(image_path: str, prompt: str) -> dict:
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()

    models = [("gpt-4o", "primary"), ("gpt-4o-mini", "fallback")]

    for model, tier in models:
        try:
            start = time.time()
            response = client.chat.completions.create(
                model=model,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {"type": "image_url", "image_url": {
                            "url": f"data:image/jpeg;base64,{image_data}"
                        }}
                    ]
                }],
                max_tokens=500,
                timeout=30
            )
            return {
                "model_used": model,
                "tier": tier,
                "result": response.choices[0].message.content,
                "latency": round(time.time() - start, 2)
            }
        except Exception as e:
            if tier == "fallback":
                raise RuntimeError(f"Todos los modelos fallaron: {e}")
            continue

Ejercicio 4: Pipeline multi-input (Difícil)

Crea una función que acepte una lista de archivos (mezclando imágenes, audios y textos) y devuelva un análisis integrado de todos.

def analyze_multiple_inputs(file_paths: list[str]) -> dict:
    contexts = []

    for path in file_paths:
        ext = Path(path).suffix.lower()

        if ext in IMAGE_EXTENSIONS:
            desc = vision_plus_llm(path, "Describe esta imagen en 2 frases.")
            contexts.append(f"[Imagen: {Path(path).name}] {desc}")

        elif ext in AUDIO_EXTENSIONS:
            with open(path, "rb") as f:
                text = client.audio.transcriptions.create(
                    model="whisper-1", file=f
                ).text
            contexts.append(f"[Audio: {Path(path).name}] {text}")

        elif ext in TEXT_EXTENSIONS:
            with open(path, "r") as f:
                text = f.read()[:2000]  # Limitar largo
            contexts.append(f"[Texto: {Path(path).name}] {text}")

    combined = "\n\n".join(contexts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Analiza toda esta información y genera:\n"
                f"1. Resumen integrado\n"
                f"2. Conexiones entre los diferentes inputs\n"
                f"3. Insights principales\n\n{combined}"
            )
        }],
        max_tokens=500
    )

    return {
        "num_inputs": len(file_paths),
        "modalities": [Path(p).suffix for p in file_paths],
        "analysis": response.choices[0].message.content
    }

Resumen

En esta cápsula aprendiste:

• Las combinaciones multimodales pueden ser por input (múltiples modalidades al mismo modelo), por pipeline (encadenar modelos), o por output (generar en múltiples modalidades)
• Los 6 patrones principales: Vision+LLM, Audio→Texto→LLM, Texto→LLM→Imagen, Documento→Extracción, Pipeline completo, Generación→Análisis
• Los principios de diseño: cada paso con propósito claro, usar modelo más barato posible, manejar errores por paso, medir latencia
• Paralelizar pasos independientes reduce latencia significativamente
• El router de modalidad (detectar tipo → aplicar pipeline) es el core del Clasificador multimodal

Próxima cápsula: Landscape de modelos — comparación detallada de GPT-4V vs Claude 3 vs Gemini.

Recursos Adicionales

1. OpenAI API Reference — Referencia completa de todas las APIs
2. LangChain Multimodal — Pipelines multimodales con LangChain
3. OpenAI Cookbook — Ejemplos prácticos de combinaciones
4. asyncio Documentation — Para paralelizar pipelines
5. concurrent.futures — Threading para I/O-bound tasks
6. Building Multimodal AI Applications — Curso de DeepLearning.AI