7. Limitations and Optimization

Description

The multimodal RAG you built in the previous capsules works. But in production, "it works" isn't enough. You need it to work fast (latency), cheap (cost), well (quality), and at scale (thousands of documents). This capsule covers the real limitations of multimodal RAG and the optimizations that mitigate them.

This isn't a theoretical capsule. Each limitation comes with concrete numbers, and each optimization comes with code you can implement. By the end, you'll know exactly how much your pipeline costs, where the bottlenecks are, and how to reduce costs by 60-80% without sacrificing significant quality.

Why it matters: A prototype that processes 5 documents in a notebook has no cost or latency problems. But a system that processes 1,000 documents with 5,000 images in production can cost hundreds of dollars and take hours. The optimizations in this capsule are the difference between a project that stays a demo and one that reaches production.

Connection with the module: These optimizations apply directly to the capsule 08 project. Knowing when to use local CLIP vs the Vision API, when to cache, and when to batch process lets you design a pipeline that's economically viable.

Limitations of Multimodal RAG

Map of limitations

Anatomy of costs

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"Indexing cost: ${costs['indexing']['total']}")
print(f"Monthly query cost: ${costs['queries_monthly']['total']}")
print(f"Estimated total: ${costs['grand_total']}")

Optimization 1: Cache Image Descriptions

The problem

Every time you index a document, you describe its images with the Vision API. If you re-index, you pay again. If you index the same document in another environment, you pay again.

Solution: persistent cache

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 this image in 1-2 sentences for indexing in semantic search."},
                {"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

Impact

Without cache: 500 images × $0.015 = $7.50 every time you index
With cache: $7.50 the first time, $0 thereafter
Savings: 100% on re-indexing

LRU cache in memory

For frequent queries during a session:

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

Optimization 2: Async Batch Processing

The problem

Describing images sequentially: 100 images × 3 sec = 5 minutes. Unacceptable.

Solution: async processing with controlled concurrency

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 this image in 1-2 sentences for indexing."},
                {"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"Described: {len(successful)}, Failed: {len(failed)}")

Impact

Sequential: 100 images × 3 sec = 300 sec (5 min)
Async (10 concurrent): 100 images / 10 × 3 sec = 30 sec
Speedup: ~10x

Batch embeddings

OpenAI lets you send up to 2048 texts in a single embeddings call.

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

Optimization 3: Local CLIP Instead of the Vision API

The problem

The Vision API costs money and has network latency. For each image, you pay and wait.

Solution: use local CLIP for direct embeddings

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

Cost comparison

Hybrid strategy: CLIP for volume, Vision for quality

def smart_embed_image(
    image_path: str,
    use_vision_threshold: int = 50
) -> dict:
    """
    For small corpora (< threshold), use the Vision API (better quality).
    For large corpora (>= threshold), use CLIP (lower cost).
    """
    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 this image in 1-2 sentences."},
                        {"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
        ]

Optimization 4: Reduce Image Resolution

The problem

High-resolution images consume more tokens in the Vision API. A 4000×3000 px image can cost 4x more than a 1000×750 px one.

Solution: resize before sending

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)

Impact on 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)

Resize to 1024 max: 60-95% savings on large images

Optimization 5: Selective Indexing

The problem

Not all images in a document are useful for RAG. Logos, decorative icons, separators, and repeated headers are noise.

Solution: filter before describing

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

Impact

Typical document with 20 images:
  - 5 repeated logos/headers   → filtered out (duplicates)
  - 3 small icons              → filtered out (too_small)
  - 2 decorative separators    → filtered out (too_few_bytes)
  - 10 relevant figures        → indexed

Without filter: 20 × $0.015 = $0.30
With filter: 10 × $0.015 = $0.15
Savings: 50% + better search quality (less noise)

Optimization 6: Larger Chunking

The problem

Small chunks (100-200 words) generate many embeddings. More chunks = more indexing cost + more results to filter.

Trade-off

Small chunks (100 words):
  + More precise — the result points exactly to the relevant paragraph
  - More embeddings, more cost
  - May lose context (a paragraph without its section makes no sense)

Large chunks (500-1000 words):
  + Fewer embeddings, lower cost
  + More context per chunk
  - Less precise — the chunk may hold both relevant and irrelevant info

Solution: adaptive chunk_size

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"Text of {text_length} chars → suggested chunk_size: {suggested_size}")

Pipeline Monitoring

Basic instrumentation

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

Usage example

pipeline = MonitoredPipeline()

# ... run pipeline ...

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

Optimizations Table: Summary

Recommended order of implementation

1. Description cache           ← Implement FIRST (maximum savings, minimum effort)
2. Selective indexing          ← Filter out noise from the start
3. Resolution reduction        ← One line of code, big impact
4. Async batch                 ← When latency matters
5. Local CLIP                  ← When budget is the limiting factor
6. Adaptive chunks             ← Fine-tuning
7. Monitoring                  ← For production

Troubleshooting

The cache doesn't work

Verify that the cache directory exists and has write permissions.

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

Vision API rate limit (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. Waiting {wait}s...")
                await asyncio.sleep(wait)
            else:
                raise

CLIP produces low-quality embeddings for technical diagrams

CLIP was trained on natural photos and generic descriptions. Technical diagrams aren't well represented.

Solution: For technical diagrams, use Vision + Embedding (strategy 1).
Reserve CLIP for photos, screenshots, and general visual content.

The pipeline takes too long on a large document

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)",
    }

Costs out of 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")

Exercises

Exercise 1: Implement a cache with TTL

Extend the description cache so it has a TTL (time-to-live). If the description is more than X days old, re-describe.

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

Exercise 2: Cost dashboard

Implement a function that, given a directory of documents, estimates the total indexing cost with and without optimizations.

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}%)",
    }

Exercise 3: Auto strategy selector

Implement a function that, given the corpus size and the budget, automatically selects the best combination of optimizations.

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

Summary

• Multimodal RAG has three axes of limitation: cost (Vision API), latency (sequential descriptions), and quality (incomplete descriptions).
• Description cache is the most impactful optimization: $0 on re-indexing, minimal effort.
• Async batch reduces latency 5-10x by processing images in parallel with controlled concurrency.
• Local CLIP eliminates the Vision API cost by generating embeddings directly, in exchange for lower semantic quality.
• Reducing resolution lowers Vision token consumption by 60-95% on large images.
• Selective indexing filters out logos, icons and duplicates before spending on descriptions.
• Monitoring gives you visibility into real costs, latency and cache hit rate.
• The right combination of optimizations depends on your corpus, budget and latency requirements.

Additional Resources

1. OpenAI Pricing — Up-to-date Vision and Embeddings costs
2. OpenAI Vision Token Calculation — How image tokens are calculated
3. CLIP (Hugging Face) — CLIP model for local use
4. asyncio Documentation — Async Python reference
5. Redis Caching Patterns — Caching patterns for production