7. Limitaciones y Optimización

Descripción

El RAG multimodal que construiste en las cápsulas anteriores funciona. Pero en producción, "funciona" no es suficiente. Necesitas que funcione rápido (latencia), barato (costo), bien (calidad), y a escala (miles de documentos). Esta cápsula cubre las limitaciones reales del RAG multimodal y las optimizaciones que las mitigan.

No es una cápsula teórica. Cada limitación viene con números concretos, y cada optimización viene con código que puedes implementar. Al terminar, sabrás exactamente cuánto cuesta tu pipeline, dónde están los cuellos de botella, y cómo reducir costos en un 60-80% sin sacrificar calidad significativa.

Por qué importa: Un prototipo que procesa 5 documentos en un notebook no tiene problemas de costo ni latencia. Pero un sistema que procesa 1,000 documentos con 5,000 imágenes en producción puede costar cientos de dólares y tardar horas. Las optimizaciones de esta cápsula son la diferencia entre un proyecto que se queda en demo y uno que llega a producción.

Conexión con el módulo: Estas optimizaciones se aplican directamente al proyecto de la cápsula 08. Saber cuándo usar CLIP local vs Vision API, cuándo cachear, y cuándo hacer batch processing te permite diseñar un pipeline que sea viable económicamente.

Limitaciones del RAG Multimodal

Mapa de limitaciones

Anatomía de costos

def estimate_pipeline_cost(
    num_documents: int,
    avg_pages_per_doc: int,
    avg_images_per_doc: int,
    avg_text_chunks_per_doc: int,
    avg_queries_per_day: int,
    days: int = 30
) -> dict:
    vision_cost_per_image = 0.015
    embedding_cost_per_1k_tokens = 0.00002
    avg_tokens_per_chunk = 200
    llm_cost_per_query = 0.03

    total_images = num_documents * avg_images_per_doc
    total_chunks = num_documents * avg_text_chunks_per_doc
    total_queries = avg_queries_per_day * days

    indexing_costs = {
        "vision_descriptions": total_images * vision_cost_per_image,
        "text_embeddings": (total_chunks * avg_tokens_per_chunk / 1000) * embedding_cost_per_1k_tokens,
        "image_embeddings": (total_images * 50 / 1000) * embedding_cost_per_1k_tokens,
    }

    query_costs = {
        "query_embeddings": (total_queries * 50 / 1000) * embedding_cost_per_1k_tokens,
        "llm_responses": total_queries * llm_cost_per_query,
    }

    total_indexing = sum(indexing_costs.values())
    total_queries_cost = sum(query_costs.values())
    total = total_indexing + total_queries_cost

    return {
        "indexing": {**indexing_costs, "total": round(total_indexing, 2)},
        "queries_monthly": {**query_costs, "total": round(total_queries_cost, 2)},
        "grand_total": round(total, 2),
        "breakdown": {
            "documents": num_documents,
            "images": total_images,
            "chunks": total_chunks,
            "queries_monthly": total_queries,
        }
    }


costs = estimate_pipeline_cost(
    num_documents=100,
    avg_pages_per_doc=20,
    avg_images_per_doc=10,
    avg_text_chunks_per_doc=40,
    avg_queries_per_day=50,
    days=30
)
print(f"Costo de indexación: ${costs['indexing']['total']}")
print(f"Costo mensual de queries: ${costs['queries_monthly']['total']}")
print(f"Total estimado: ${costs['grand_total']}")

Optimización 1: Cachear Descripciones de Imágenes

El problema

Cada vez que indexas un documento, describes sus imágenes con Vision API. Si re-indexas, pagas de nuevo. Si indexas el mismo documento en otro entorno, pagas de nuevo.

Solución: cache persistente

import json
import hashlib
from pathlib import Path
from functools import lru_cache


CACHE_DIR = Path("./description_cache")
CACHE_DIR.mkdir(exist_ok=True)


def image_hash(image_data: bytes) -> str:
    return hashlib.sha256(image_data).hexdigest()


def get_cached_description(img_hash: str) -> str | None:
    cache_file = CACHE_DIR / f"{img_hash}.json"
    if cache_file.exists():
        data = json.loads(cache_file.read_text())
        return data.get("description")
    return None


def save_description_to_cache(
    img_hash: str,
    description: str,
    model: str,
    image_path: str = ""
) -> None:
    cache_file = CACHE_DIR / f"{img_hash}.json"
    data = {
        "description": description,
        "model": model,
        "image_path": image_path,
        "hash": img_hash,
    }
    cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=2))


def describe_image_with_cache(
    image_data: bytes,
    ext: str = "png",
    model: str = "gpt-4o-mini"
) -> str:
    import base64
    from openai import OpenAI

    img_hash = image_hash(image_data)

    cached = get_cached_description(img_hash)
    if cached:
        return cached

    client = OpenAI()
    b64 = base64.b64encode(image_data).decode("utf-8")
    mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"

    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe esta imagen en 1-2 oraciones para indexación en búsqueda semántica."},
                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
            ]
        }],
        max_tokens=100
    )
    description = response.choices[0].message.content

    save_description_to_cache(img_hash, description, model)

    return description

Impacto

Sin cache: 500 imágenes × $0.015 = $7.50 cada vez que indexas
Con cache: $7.50 la primera vez, $0 las siguientes
Ahorro: 100% en re-indexaciones

Cache con LRU en memoria

Para queries frecuentes durante una sesión:

@lru_cache(maxsize=2000)
def describe_image_memory_cache(image_path: str) -> str:
    with open(image_path, "rb") as f:
        data = f.read()
    ext = Path(image_path).suffix.lstrip(".")
    return describe_image_with_cache(data, ext)

Optimización 2: Batch Processing Async

El problema

Describir imágenes secuencialmente: 100 imágenes × 3 seg = 5 minutos. Inaceptable.

Solución: procesamiento async con concurrencia controlada

import asyncio
from openai import AsyncOpenAI
import base64


async def describe_image_async(
    async_client: AsyncOpenAI,
    image_data: bytes,
    ext: str = "png"
) -> str:
    img_hash = image_hash(image_data)
    cached = get_cached_description(img_hash)
    if cached:
        return cached

    b64 = base64.b64encode(image_data).decode("utf-8")
    mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"

    response = await async_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe esta imagen en 1-2 oraciones para indexación."},
                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
            ]
        }],
        max_tokens=100
    )
    description = response.choices[0].message.content
    save_description_to_cache(img_hash, description, "gpt-4o-mini")
    return description


async def batch_describe_images(
    images: list[dict],
    max_concurrent: int = 10
) -> list[dict]:
    async_client = AsyncOpenAI()
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []

    async def process_one(img: dict, index: int) -> dict:
        async with semaphore:
            try:
                desc = await describe_image_async(
                    async_client,
                    img["data"],
                    img.get("ext", "png")
                )
                return {"index": index, "description": desc, "success": True}
            except Exception as e:
                return {"index": index, "description": "", "success": False, "error": str(e)}

    tasks = [process_one(img, i) for i, img in enumerate(images)]
    results = await asyncio.gather(*tasks)

    return sorted(results, key=lambda x: x["index"])


# results = asyncio.run(batch_describe_images(images, max_concurrent=10))
# successful = [r for r in results if r["success"]]
# failed = [r for r in results if not r["success"]]
# print(f"Descritas: {len(successful)}, Fallidas: {len(failed)}")

Impacto

Secuencial: 100 imágenes × 3 seg = 300 seg (5 min)
Async (10 concurrentes): 100 imágenes / 10 × 3 seg = 30 seg
Speedup: ~10x

Batch de embeddings

OpenAI permite enviar hasta 2048 textos en una sola llamada de embeddings.

from openai import OpenAI

client = OpenAI()


def batch_embeddings(
    texts: list[str],
    batch_size: int = 500
) -> list[list[float]]:
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=batch
        )
        batch_embs = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embs)

    return all_embeddings

Optimización 3: CLIP Local en Lugar de Vision API

El problema

Vision API cuesta dinero y tiene latencia de red. Para cada imagen, pagas y esperas.

Solución: usar CLIP local para embeddings directos

from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import torch

clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
clip_model.eval()

device = "cuda" if torch.cuda.is_available() else "cpu"
clip_model = clip_model.to(device)


def clip_embed_image(image_path: str) -> list[float]:
    image = Image.open(image_path).convert("RGB")
    inputs = clip_processor(images=image, return_tensors="pt")
    inputs = {k: v.to(device) for k, v in inputs.items()}

    with torch.no_grad():
        features = clip_model.get_image_features(**inputs)
    normalized = features / features.norm(dim=-1, keepdim=True)
    return normalized[0].cpu().numpy().tolist()


def clip_embed_text(text: str) -> list[float]:
    inputs = clip_processor(text=[text], return_tensors="pt", padding=True)
    inputs = {k: v.to(device) for k, v in inputs.items()}

    with torch.no_grad():
        features = clip_model.get_text_features(**inputs)
    normalized = features / features.norm(dim=-1, keepdim=True)
    return normalized[0].cpu().numpy().tolist()

Comparación de costos

Estrategia híbrida: CLIP para volumen, Vision para calidad

def smart_embed_image(
    image_path: str,
    use_vision_threshold: int = 50
) -> dict:
    """
    Para corpus pequeños (< threshold), usa Vision API (mejor calidad).
    Para corpus grandes (>= threshold), usa CLIP (menor costo).
    """
    return {
        "clip_embedding": clip_embed_image(image_path),
        "strategy": "clip"
    }


def embed_image_batch_smart(
    image_paths: list[str],
    vision_threshold: int = 50
) -> list[dict]:
    if len(image_paths) < vision_threshold:
        results = []
        for path in image_paths:
            from openai import OpenAI
            import base64
            oai = OpenAI()
            with open(path, "rb") as f:
                b64 = base64.b64encode(f.read()).decode("utf-8")
            ext = Path(path).suffix.lstrip(".")
            mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"

            response = oai.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "Describe esta imagen en 1-2 oraciones."},
                        {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
                    ]
                }],
                max_tokens=100
            )
            desc = response.choices[0].message.content
            emb = oai.embeddings.create(model="text-embedding-3-small", input=desc).data[0].embedding
            results.append({"embedding": emb, "description": desc, "strategy": "vision"})
        return results
    else:
        return [
            {"embedding": clip_embed_image(p), "description": "", "strategy": "clip"}
            for p in image_paths
        ]

Optimización 4: Reducir Resolución de Imágenes

El problema

Imágenes de alta resolución consumen más tokens en Vision API. Una imagen de 4000×3000 px puede costar 4x más que una de 1000×750 px.

Solución: redimensionar antes de enviar

from PIL import Image
from io import BytesIO


def resize_image_for_api(
    image_data: bytes,
    max_dimension: int = 1024
) -> bytes:
    img = Image.open(BytesIO(image_data))
    width, height = img.size

    if width <= max_dimension and height <= max_dimension:
        return image_data

    if width > height:
        new_width = max_dimension
        new_height = int(height * (max_dimension / width))
    else:
        new_height = max_dimension
        new_width = int(width * (max_dimension / height))

    img_resized = img.resize((new_width, new_height), Image.LANCZOS)

    buffer = BytesIO()
    img_format = img.format or "PNG"
    img_resized.save(buffer, format=img_format)
    return buffer.getvalue()


def describe_image_optimized(
    image_data: bytes,
    ext: str = "png",
    max_dimension: int = 1024
) -> str:
    resized = resize_image_for_api(image_data, max_dimension)
    return describe_image_with_cache(resized, ext)

Impacto en tokens

OpenAI Vision token calculation:
  512×512:   ~170 tokens (~$0.003)
  1024×1024: ~680 tokens (~$0.010)
  2048×2048: ~2720 tokens (~$0.040)
  4096×4096: ~10880 tokens (~$0.160)

Redimensionar a 1024 max: ahorro del 60-95% en imágenes grandes

Optimización 5: Indexación Selectiva

El problema

No todas las imágenes en un documento son útiles para RAG. Logos, iconos decorativos, separadores, y headers repetidos son ruido.

Solución: filtrar antes de describir

import hashlib
from collections import Counter

MIN_WIDTH = 150
MIN_HEIGHT = 150
MIN_SIZE_BYTES = 10000
MAX_ASPECT_RATIO = 8.0


def should_index_image(
    image_data: bytes,
    width: int,
    height: int,
    seen_hashes: set
) -> tuple[bool, str]:
    if width < MIN_WIDTH or height < MIN_HEIGHT:
        return False, "too_small"

    if len(image_data) < MIN_SIZE_BYTES:
        return False, "too_few_bytes"

    aspect = max(width, height) / max(min(width, height), 1)
    if aspect > MAX_ASPECT_RATIO:
        return False, "extreme_aspect_ratio"

    img_hash = hashlib.md5(image_data).hexdigest()
    if img_hash in seen_hashes:
        return False, "duplicate"
    seen_hashes.add(img_hash)

    return True, "accepted"


def filter_images_for_indexing(
    images: list[dict]
) -> tuple[list[dict], dict]:
    seen_hashes = set()
    filtered = []
    rejection_counts = Counter()

    for img in images:
        should_index, reason = should_index_image(
            img.get("data", b""),
            img.get("width", 0),
            img.get("height", 0),
            seen_hashes
        )
        if should_index:
            filtered.append(img)
        else:
            rejection_counts[reason] += 1

    stats = {
        "total": len(images),
        "accepted": len(filtered),
        "rejected": len(images) - len(filtered),
        "rejection_reasons": dict(rejection_counts),
    }

    return filtered, stats

Impacto

Documento típico con 20 imágenes:
  - 5 logos/headers repetidos  → filtrados (duplicados)
  - 3 iconos pequeños          → filtrados (too_small)
  - 2 separadores decorativos  → filtrados (too_few_bytes)
  - 10 figuras relevantes      → indexadas

Sin filtro: 20 × $0.015 = $0.30
Con filtro: 10 × $0.015 = $0.15
Ahorro: 50% + mejor calidad de búsqueda (menos ruido)

Optimización 6: Chunking Más Grande

El problema

Chunks pequeños (100-200 palabras) generan muchos embeddings. Más chunks = más costo de indexación + más resultados a filtrar.

Trade-off

Chunks pequeños (100 palabras):
  + Más precisos — el resultado apunta exactamente al párrafo relevante
  - Más embeddings, más costo
  - Puede perder contexto (el párrafo sin su sección no tiene sentido)

Chunks grandes (500-1000 palabras):
  + Menos embeddings, menor costo
  + Más contexto por chunk
  - Menos precisos — el chunk puede tener info relevante e irrelevante

Solución: chunk_size adaptativo

def adaptive_chunk_size(
    total_text_length: int,
    budget_embeddings: float = 1.0
) -> int:
    cost_per_embedding = 0.00002 * 0.2
    max_chunks = budget_embeddings / cost_per_embedding

    avg_chunk_size = total_text_length / max(max_chunks, 1)
    chunk_size = max(200, min(2000, int(avg_chunk_size)))

    return chunk_size


text_length = 50000
suggested_size = adaptive_chunk_size(text_length, budget_embeddings=0.50)
print(f"Texto de {text_length} chars → chunk_size sugerido: {suggested_size}")

Monitoreo del Pipeline

Instrumentación básica

import time
from dataclasses import dataclass, field


@dataclass
class PipelineMetrics:
    indexing_time: float = 0.0
    query_time: float = 0.0
    images_described: int = 0
    images_from_cache: int = 0
    embeddings_generated: int = 0
    llm_calls: int = 0
    total_cost_estimate: float = 0.0
    errors: list = field(default_factory=list)

    def summary(self) -> dict:
        total_images = self.images_described + self.images_from_cache
        cache_rate = self.images_from_cache / max(total_images, 1)
        return {
            "indexing_time_sec": round(self.indexing_time, 2),
            "query_time_sec": round(self.query_time, 2),
            "images_described": self.images_described,
            "cache_hit_rate": f"{cache_rate:.1%}",
            "embeddings_generated": self.embeddings_generated,
            "llm_calls": self.llm_calls,
            "estimated_cost": f"${self.total_cost_estimate:.4f}",
            "errors": len(self.errors),
        }


class MonitoredPipeline:
    def __init__(self):
        self.metrics = PipelineMetrics()

    def describe_image(self, image_data: bytes, ext: str = "png") -> str:
        img_hash = image_hash(image_data)
        cached = get_cached_description(img_hash)

        if cached:
            self.metrics.images_from_cache += 1
            return cached

        start = time.time()
        description = describe_image_with_cache(image_data, ext)
        elapsed = time.time() - start

        self.metrics.images_described += 1
        self.metrics.total_cost_estimate += 0.015
        self.metrics.indexing_time += elapsed

        return description

    def generate_embeddings(self, texts: list[str]) -> list[list[float]]:
        start = time.time()
        embeddings = batch_embeddings(texts)
        elapsed = time.time() - start

        self.metrics.embeddings_generated += len(texts)
        self.metrics.total_cost_estimate += len(texts) * 0.2 * 0.00002
        self.metrics.indexing_time += elapsed

        return embeddings

    def query_llm(self, messages: list[dict], model: str = "gpt-4o") -> str:
        start = time.time()
        response = client.chat.completions.create(
            model=model, messages=messages, max_tokens=500, temperature=0
        )
        elapsed = time.time() - start

        self.metrics.llm_calls += 1
        self.metrics.query_time += elapsed
        self.metrics.total_cost_estimate += 0.03

        return response.choices[0].message.content

    def report(self) -> dict:
        return self.metrics.summary()

Ejemplo de uso

pipeline = MonitoredPipeline()

# ... ejecutar pipeline ...

report = pipeline.report()
for key, value in report.items():
    print(f"  {key}: {value}")

Tabla de Optimizaciones: Resumen

Orden recomendado de implementación

1. Cache de descripciones      ← Implementa PRIMERO (máximo ahorro, mínimo esfuerzo)
2. Indexación selectiva        ← Filtra ruido desde el inicio
3. Reducir resolución          ← Una línea de código, gran impacto
4. Batch async                 ← Cuando la latencia importa
5. CLIP local                  ← Cuando el presupuesto es el factor limitante
6. Chunks adaptativos          ← Ajuste fino
7. Monitoreo                   ← Para producción

Troubleshooting

El cache no funciona

Verifica que el directorio de cache existe y tiene permisos de escritura.

cache_dir = Path("./description_cache")
cache_dir.mkdir(exist_ok=True)
print(f"Cache dir: {cache_dir.absolute()}")
print(f"Archivos en cache: {len(list(cache_dir.glob('*.json')))}")

Rate limit de Vision API (429 Too Many Requests)

import asyncio


async def describe_with_retry(
    async_client,
    image_data: bytes,
    ext: str = "png",
    max_retries: int = 3
) -> str:
    for attempt in range(max_retries):
        try:
            return await describe_image_async(async_client, image_data, ext)
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"Rate limited. Esperando {wait}s...")
                await asyncio.sleep(wait)
            else:
                raise

CLIP produce embeddings de baja calidad para diagramas técnicos

CLIP fue entrenado con fotos naturales y descripciones genéricas. Los diagramas técnicos no están bien representados.

Solución: Para diagramas técnicos, usa Vision + Embedding (estrategia 1).
Reserva CLIP para fotos, capturas de pantalla, y contenido visual general.

El pipeline tarda demasiado en un documento grande

def estimate_processing_time(
    num_images: int,
    num_text_chunks: int,
    use_async: bool = True,
    max_concurrent: int = 10,
    seconds_per_image: float = 3.0,
    seconds_per_embedding_batch: float = 0.5,
    embedding_batch_size: int = 100
) -> dict:
    if use_async:
        image_time = (num_images / max_concurrent) * seconds_per_image
    else:
        image_time = num_images * seconds_per_image

    embedding_batches = (num_text_chunks + num_images) / embedding_batch_size
    embedding_time = embedding_batches * seconds_per_embedding_batch

    total = image_time + embedding_time

    return {
        "image_description_time": f"{image_time:.0f}s",
        "embedding_time": f"{embedding_time:.1f}s",
        "total_estimated": f"{total:.0f}s ({total/60:.1f} min)",
    }

Costos fuera de control

def set_budget_guard(
    max_budget: float = 5.0,
    current_spend: float = 0.0
) -> callable:
    remaining = max_budget - current_spend

    def check_and_deduct(cost: float) -> bool:
        nonlocal remaining
        if cost > remaining:
            print(f"BUDGET EXCEEDED: need ${cost:.4f}, have ${remaining:.4f}")
            return False
        remaining -= cost
        return True

    return check_and_deduct


guard = set_budget_guard(max_budget=2.0)
if guard(0.015):
    pass  # describe_image(...)
else:
    print("Switching to CLIP local to save budget")

Ejercicios

Ejercicio 1: Implementar cache con TTL

Extiende el cache de descripciones para que tenga un TTL (time-to-live). Si la descripción tiene más de X días, re-describir.

from datetime import datetime, timedelta


def get_cached_description_with_ttl(
    img_hash: str,
    ttl_days: int = 30
) -> str | None:
    cache_file = CACHE_DIR / f"{img_hash}.json"
    if not cache_file.exists():
        return None

    data = json.loads(cache_file.read_text())
    created = data.get("created_at")

    if created:
        created_dt = datetime.fromisoformat(created)
        if datetime.now() - created_dt > timedelta(days=ttl_days):
            return None

    return data.get("description")


def save_description_with_ttl(
    img_hash: str,
    description: str,
    model: str
) -> None:
    cache_file = CACHE_DIR / f"{img_hash}.json"
    data = {
        "description": description,
        "model": model,
        "hash": img_hash,
        "created_at": datetime.now().isoformat(),
    }
    cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=2))

Ejercicio 2: Dashboard de costos

Implementa una función que, dado un directorio de documentos, estime el costo total de indexación con y sin optimizaciones.

def cost_dashboard(
    pdf_paths: list[str]
) -> dict:
    import fitz

    total_images = 0
    total_text_chars = 0
    cached_images = 0

    for pdf_path in pdf_paths:
        doc = fitz.open(pdf_path)
        for page_num in range(len(doc)):
            page = doc[page_num]
            total_text_chars += len(page.get_text())

            for img_ref in page.get_images():
                total_images += 1
                xref = img_ref[0]
                try:
                    base_image = doc.extract_image(xref)
                    if base_image and base_image.get("image"):
                        img_h = image_hash(base_image["image"])
                        if get_cached_description(img_h):
                            cached_images += 1
                except Exception:
                    pass
        doc.close()

    estimated_chunks = total_text_chars / 2000

    no_opt = {
        "vision_cost": total_images * 0.015,
        "embedding_cost": (estimated_chunks + total_images) * 0.2 * 0.00002,
    }
    no_opt["total"] = sum(no_opt.values())

    with_opt = {
        "vision_cost": (total_images - cached_images) * 0.015,
        "embedding_cost": estimated_chunks * 0.2 * 0.00002,
    }
    with_opt["total"] = sum(with_opt.values())

    savings = no_opt["total"] - with_opt["total"]

    return {
        "documents": len(pdf_paths),
        "total_images": total_images,
        "cached_images": cached_images,
        "estimated_chunks": int(estimated_chunks),
        "cost_without_optimization": f"${no_opt['total']:.2f}",
        "cost_with_optimization": f"${with_opt['total']:.2f}",
        "savings": f"${savings:.2f} ({savings/max(no_opt['total'],0.01)*100:.0f}%)",
    }

Ejercicio 3: Auto-selector de estrategia

Implementa una función que, dado el tamaño del corpus y el presupuesto, seleccione automáticamente la mejor combinación de optimizaciones.

def auto_select_strategy(
    num_images: int,
    budget: float,
    latency_requirement: str = "normal"
) -> dict:
    vision_cost = num_images * 0.015

    if vision_cost <= budget * 0.5:
        image_strategy = "vision_api"
        estimated_cost = vision_cost
    else:
        image_strategy = "clip_local"
        estimated_cost = 0.0

    use_async = num_images > 20 or latency_requirement == "low"
    use_cache = True
    use_resize = num_images > 10
    use_selective = num_images > 30

    max_concurrent = 5 if latency_requirement == "normal" else 15

    return {
        "image_strategy": image_strategy,
        "use_async": use_async,
        "use_cache": use_cache,
        "use_resize": use_resize,
        "use_selective_indexing": use_selective,
        "max_concurrent": max_concurrent,
        "estimated_image_cost": f"${estimated_cost:.2f}",
        "within_budget": estimated_cost <= budget,
    }


strategy = auto_select_strategy(num_images=500, budget=5.0, latency_requirement="low")
for key, value in strategy.items():
    print(f"  {key}: {value}")

Resumen

• El RAG multimodal tiene tres ejes de limitación: costo (Vision API), latencia (descripciones secuenciales), y calidad (descripciones incompletas).
• Cache de descripciones es la optimización más impactante: $0 en re-indexaciones, mínimo esfuerzo.
• Batch async reduce latencia 5-10x procesando imágenes en paralelo con concurrencia controlada.
• CLIP local elimina el costo de Vision API al generar embeddings directamente, a cambio de menor calidad semántica.
• Reducir resolución baja el consumo de tokens de Vision un 60-95% en imágenes grandes.
• Indexación selectiva filtra logos, iconos y duplicados antes de gastar en descripciones.
• Monitoreo te da visibilidad sobre costos reales, latencia y cache hit rate.
• La combinación correcta de optimizaciones depende de tu corpus, presupuesto y requerimientos de latencia.

Recursos Adicionales

1. OpenAI Pricing — Costos actualizados de Vision y Embeddings
2. OpenAI Vision Token Calculation — Cómo se calculan tokens de imagen
3. CLIP (Hugging Face) — Modelo CLIP para uso local
4. asyncio Documentation — Referencia de async Python
5. Redis Caching Patterns — Patrones de caching para producción