3. Indexación Multimodal

Descripción

Tener embeddings de texto e imágenes es solo la mitad del problema. La otra mitad es almacenarlos de forma que la búsqueda sea eficiente, organizada y filtrable. En esta cápsula vas a construir un índice multimodal en ChromaDB que combina chunks de texto y representaciones de imágenes, con metadata que permite filtrar por tipo, fuente, página y dominio.

La indexación multimodal es diferente a la indexación solo-texto en tres aspectos clave: necesitas distinguir entre tipos de contenido (texto vs imagen), necesitas preservar la relación entre un chunk de texto y las imágenes de su sección, y necesitas diseñar la metadata para que las queries puedan filtrar eficientemente.

Por qué importa: Un índice mal diseñado produce resultados de búsqueda irrelevantes sin importar qué tan buenos sean tus embeddings. Si no distingues texto de imágenes, no puedes filtrar. Si no guardas la página de origen, no puedes citar fuentes. Si no persistes el índice, lo pierdes cada vez que reinicias.

Conexión con el módulo: Los embeddings de la cápsula 02 se almacenan aquí. El retrieval de la cápsula 04 busca en este índice. El procesamiento de documentos de la cápsula 05 alimenta este índice. Es la pieza central.

Estructura del Índice Multimodal

Diseño del documento

Cada entrada en el índice necesita:

{
    "id": "doc_manual_chunk_05",
    "document": "El API Gateway distribuye requests entre microservicios...",
    "embedding": [0.02, -0.15, 0.33, ...],
    "metadata": {
        "type": "text",
        "source": "manual_tecnico.pdf",
        "page": 12,
        "chunk_index": 5,
        "total_chunks": 42,
        "has_related_images": True,
        "domain": "technical",
        "created_at": "2024-01-15T10:30:00Z"
    }
}

Para imágenes:

{
    "id": "doc_manual_img_03",
    "document": "Diagrama de arquitectura mostrando API Gateway conectado a tres microservicios: auth, payments, notifications",
    "embedding": [0.08, -0.22, 0.11, ...],
    "metadata": {
        "type": "image",
        "source": "manual_tecnico.pdf",
        "page": 12,
        "image_path": "images/manual_tecnico_p12_img0.png",
        "original_format": "png",
        "description_model": "gpt-4o-mini",
        "domain": "technical",
        "created_at": "2024-01-15T10:30:00Z"
    }
}

Campos de metadata recomendados

ChromaDB: Setup Completo

Conexión y collection

import chromadb
from chromadb.utils import embedding_functions
from dotenv import load_dotenv
import os

load_dotenv()

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)


def create_multimodal_index(
    name: str = "multimodal_rag",
    persist_path: str = "./chroma_db"
) -> tuple:
    client = chromadb.PersistentClient(path=persist_path)
    collection = client.get_or_create_collection(
        name=name,
        embedding_function=openai_ef,
        metadata={"hnsw:space": "cosine"}
    )
    return client, collection


chroma_client, collection = create_multimodal_index()
print(f"Collection: {collection.name}")
print(f"Documentos existentes: {collection.count()}")

Diferencia entre Client y PersistentClient

# EN MEMORIA — se pierde al cerrar el proceso
client_memory = chromadb.Client()

# PERSISTENTE — se guarda en disco
client_persistent = chromadb.PersistentClient(path="./chroma_db")

Para desarrollo y pruebas rápidas, usa Client(). Para cualquier cosa que quieras conservar entre ejecuciones, usa PersistentClient().

Indexar Chunks de Texto

Función básica

def add_text_chunks(
    collection,
    chunks: list[dict],
    source: str
) -> int:
    """
    chunks: [{"text": "...", "page": 1, "chunk_index": 0}, ...]
    """
    documents = []
    ids = []
    metadatas = []

    for chunk in chunks:
        chunk_id = f"{source}_text_{chunk['chunk_index']}"

        existing = collection.get(ids=[chunk_id])
        if existing["ids"]:
            continue

        documents.append(chunk["text"])
        ids.append(chunk_id)
        metadatas.append({
            "type": "text",
            "source": source,
            "page": chunk.get("page", 0),
            "chunk_index": chunk["chunk_index"],
            "has_related_images": chunk.get("has_images", False),
        })

    if documents:
        collection.add(
            documents=documents,
            ids=ids,
            metadatas=metadatas
        )

    return len(documents)


sample_chunks = [
    {
        "text": "El API Gateway es el punto de entrada único para todos los clientes. Distribuye las peticiones entre los microservicios según las rutas configuradas.",
        "page": 1,
        "chunk_index": 0,
        "has_images": True
    },
    {
        "text": "El servicio de autenticación valida tokens JWT. Cada request debe incluir un token válido en el header Authorization.",
        "page": 2,
        "chunk_index": 1,
        "has_images": False
    },
    {
        "text": "La base de datos PostgreSQL almacena los datos de usuarios y transacciones. Se recomienda usar connection pooling con un máximo de 20 conexiones.",
        "page": 3,
        "chunk_index": 2,
        "has_images": False
    },
]

added = add_text_chunks(collection, sample_chunks, source="manual_tecnico.pdf")
print(f"Chunks de texto añadidos: {added}")

Chunking de texto para RAG

El tamaño del chunk afecta directamente la calidad del retrieval. Chunks muy pequeños pierden contexto. Chunks muy grandes diluyen la señal.

def chunk_text(
    text: str,
    chunk_size: int = 500,
    overlap: int = 100
) -> list[dict]:
    words = text.split()
    chunks = []
    start = 0
    chunk_index = 0

    while start < len(words):
        end = start + chunk_size
        chunk_words = words[start:end]
        chunk_text = " ".join(chunk_words)

        chunks.append({
            "text": chunk_text,
            "chunk_index": chunk_index,
            "start_word": start,
            "end_word": min(end, len(words)),
        })

        start += chunk_size - overlap
        chunk_index += 1

    return chunks


long_text = "Este es un documento largo " * 200
chunks = chunk_text(long_text, chunk_size=50, overlap=10)
print(f"Texto de {len(long_text.split())} palabras → {len(chunks)} chunks")
for c in chunks[:3]:
    print(f"  Chunk {c['chunk_index']}: {len(c['text'].split())} palabras")

Chunking con separadores semánticos

Para documentos estructurados (markdown, HTML), es mejor dividir por secciones que por número de palabras.

import re


def chunk_by_sections(
    text: str,
    max_chunk_size: int = 500
) -> list[dict]:
    sections = re.split(r'\n#{1,3}\s+', text)
    chunks = []
    chunk_index = 0

    for section in sections:
        section = section.strip()
        if not section:
            continue

        words = section.split()
        if len(words) <= max_chunk_size:
            chunks.append({
                "text": section,
                "chunk_index": chunk_index,
            })
            chunk_index += 1
        else:
            sub_chunks = chunk_text(section, chunk_size=max_chunk_size, overlap=50)
            for sc in sub_chunks:
                sc["chunk_index"] = chunk_index
                chunks.append(sc)
                chunk_index += 1

    return chunks

Indexar Descripciones de Imágenes

Función para indexar imágenes con descripción

from openai import OpenAI

client = OpenAI()

DESCRIPTION_PROMPT = (
    "Describe esta imagen de forma concisa y precisa para indexación "
    "en un sistema de búsqueda. Incluye: qué muestra, elementos principales, "
    "tipo de contenido (diagrama, foto, gráfico, tabla). Máximo 2-3 oraciones."
)


def describe_image_for_indexing(image_path: str) -> str:
    import base64
    from pathlib import Path

    path = Path(image_path)
    ext = path.suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp"}
    mime = mime_map.get(ext, "image/png")

    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode("utf-8")

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": DESCRIPTION_PROMPT},
                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
            ]
        }],
        max_tokens=150
    )
    return response.choices[0].message.content


def add_image_descriptions(
    collection,
    images: list[dict],
    source: str
) -> int:
    """
    images: [{"path": "img.png", "page": 1, "image_index": 0}, ...]
    """
    documents = []
    ids = []
    metadatas = []

    for img in images:
        img_id = f"{source}_img_{img['image_index']}"

        existing = collection.get(ids=[img_id])
        if existing["ids"]:
            continue

        description = describe_image_for_indexing(img["path"])

        documents.append(description)
        ids.append(img_id)
        metadatas.append({
            "type": "image",
            "source": source,
            "page": img.get("page", 0),
            "image_path": img["path"],
            "description_model": "gpt-4o-mini",
        })

    if documents:
        collection.add(
            documents=documents,
            ids=ids,
            metadatas=metadatas
        )

    return len(documents)

Indexar con embeddings pre-computados

Si ya tienes los embeddings (por ejemplo, de CLIP o de un batch previo), puedes pasarlos directamente.

def add_with_precomputed_embeddings(
    collection,
    items: list[dict]
) -> int:
    """
    items: [{
        "id": "...",
        "document": "texto o descripción",
        "embedding": [0.1, -0.2, ...],
        "metadata": {...}
    }, ...]
    """
    if not items:
        return 0

    collection.add(
        documents=[item["document"] for item in items],
        embeddings=[item["embedding"] for item in items],
        ids=[item["id"] for item in items],
        metadatas=[item["metadata"] for item in items]
    )
    return len(items)

Pipeline de Indexación Completo

Procesar un documento: extraer texto e imágenes, indexar todo

import fitz
from pathlib import Path


def extract_text_chunks_from_pdf(
    pdf_path: str,
    chunk_size: int = 500,
    overlap: int = 100
) -> list[dict]:
    doc = fitz.open(pdf_path)
    chunks = []
    chunk_index = 0

    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text()

        if not text.strip():
            continue

        page_chunks = chunk_text(text, chunk_size=chunk_size, overlap=overlap)
        for pc in page_chunks:
            pc["page"] = page_num + 1
            pc["chunk_index"] = chunk_index
            chunk_index += 1
            chunks.append(pc)

    doc.close()
    return chunks


def extract_images_from_pdf(
    pdf_path: str,
    output_dir: str = "./extracted_images"
) -> list[dict]:
    doc = fitz.open(pdf_path)
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    images = []
    img_index = 0

    for page_num in range(len(doc)):
        page = doc[page_num]
        img_list = page.get_images()

        for img_ref in img_list:
            xref = img_ref[0]
            base_image = doc.extract_image(xref)

            if base_image["image"] is None:
                continue

            ext = base_image.get("ext", "png")
            img_filename = f"{Path(pdf_path).stem}_p{page_num+1}_img{img_index}.{ext}"
            img_path = output / img_filename

            with open(img_path, "wb") as f:
                f.write(base_image["image"])

            images.append({
                "path": str(img_path),
                "page": page_num + 1,
                "image_index": img_index,
                "format": ext,
                "size_bytes": len(base_image["image"]),
            })
            img_index += 1

    doc.close()
    return images


def index_document(
    pdf_path: str,
    collection,
    chunk_size: int = 500,
    image_output_dir: str = "./extracted_images"
) -> dict:
    source = Path(pdf_path).name
    stats = {"text_chunks": 0, "images": 0, "errors": []}

    text_chunks = extract_text_chunks_from_pdf(pdf_path, chunk_size=chunk_size)
    images = extract_images_from_pdf(pdf_path, output_dir=image_output_dir)

    image_pages = {img["page"] for img in images}
    for chunk in text_chunks:
        chunk["has_images"] = chunk.get("page", 0) in image_pages

    stats["text_chunks"] = add_text_chunks(collection, text_chunks, source)

    try:
        stats["images"] = add_image_descriptions(collection, images, source)
    except Exception as e:
        stats["errors"].append(f"Error indexando imágenes: {e}")

    print(f"Indexado {source}: {stats['text_chunks']} chunks, {stats['images']} imágenes")
    if stats["errors"]:
        for err in stats["errors"]:
            print(f"  WARNING: {err}")

    return stats

Indexar múltiples documentos

def index_multiple_documents(
    pdf_paths: list[str],
    collection,
    chunk_size: int = 500
) -> dict:
    total_stats = {"documents": 0, "text_chunks": 0, "images": 0, "errors": []}

    for pdf_path in pdf_paths:
        try:
            stats = index_document(pdf_path, collection, chunk_size=chunk_size)
            total_stats["documents"] += 1
            total_stats["text_chunks"] += stats["text_chunks"]
            total_stats["images"] += stats["images"]
            total_stats["errors"].extend(stats["errors"])
        except Exception as e:
            total_stats["errors"].append(f"Error con {pdf_path}: {e}")

    print(f"\nTotal: {total_stats['documents']} docs, "
          f"{total_stats['text_chunks']} chunks, "
          f"{total_stats['images']} imágenes")
    return total_stats

Consultar el Índice

Buscar con filtros de metadata

def query_index(
    collection,
    query: str,
    n_results: int = 5,
    content_type: str = None,
    source_filter: str = None
) -> list[dict]:
    where_filter = {}
    if content_type:
        where_filter["type"] = content_type
    if source_filter:
        where_filter["source"] = source_filter

    kwargs = {
        "query_texts": [query],
        "n_results": n_results,
    }
    if where_filter:
        if len(where_filter) == 1:
            kwargs["where"] = where_filter
        else:
            kwargs["where"] = {"$and": [{k: v} for k, v in where_filter.items()]}

    results = collection.query(**kwargs)

    matches = []
    for i in range(len(results["documents"][0])):
        matches.append({
            "id": results["ids"][0][i],
            "content": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
        })

    return matches


results = query_index(collection, "microservicios", n_results=3)
for r in results:
    print(f"  [{r['metadata'].get('type', '?')}] {r['distance']:.4f} — {r['content'][:60]}...")

text_only = query_index(collection, "autenticación", content_type="text")
images_only = query_index(collection, "diagrama de arquitectura", content_type="image")

Obtener estadísticas del índice

def index_stats(collection) -> dict:
    all_docs = collection.get(include=["metadatas"])

    stats = {
        "total": len(all_docs["ids"]),
        "text_chunks": 0,
        "image_chunks": 0,
        "sources": set(),
        "pages": set(),
    }

    for meta in all_docs["metadatas"]:
        if meta.get("type") == "text":
            stats["text_chunks"] += 1
        elif meta.get("type") == "image":
            stats["image_chunks"] += 1
        if meta.get("source"):
            stats["sources"].add(meta["source"])
        if meta.get("page"):
            stats["pages"].add(meta["page"])

    stats["sources"] = list(stats["sources"])
    stats["unique_pages"] = len(stats["pages"])
    del stats["pages"]

    return stats


print(index_stats(collection))

Actualizar y Eliminar Documentos

Eliminar todos los chunks de un documento

def delete_document(collection, source: str) -> int:
    all_docs = collection.get(
        where={"source": source},
        include=["metadatas"]
    )

    ids_to_delete = all_docs["ids"]

    if ids_to_delete:
        collection.delete(ids=ids_to_delete)

    return len(ids_to_delete)


deleted = delete_document(collection, "manual_tecnico.pdf")
print(f"Eliminados {deleted} chunks")

Re-indexar un documento

def reindex_document(
    pdf_path: str,
    collection,
    chunk_size: int = 500
) -> dict:
    source = Path(pdf_path).name

    deleted = delete_document(collection, source)
    print(f"Eliminados {deleted} chunks anteriores de {source}")

    stats = index_document(pdf_path, collection, chunk_size=chunk_size)
    return stats

Actualizar metadata sin re-indexar

def update_metadata(
    collection,
    ids: list[str],
    new_metadata: dict
) -> None:
    for doc_id in ids:
        existing = collection.get(ids=[doc_id], include=["metadatas"])
        if not existing["ids"]:
            continue

        current_meta = existing["metadatas"][0]
        current_meta.update(new_metadata)

        collection.update(
            ids=[doc_id],
            metadatas=[current_meta]
        )

Índice Híbrido: Texto + Embeddings de CLIP

Si usas CLIP para imágenes, necesitas una collection separada porque la dimensión del embedding es diferente (512 vs 1536).

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

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


def create_dual_index(persist_path: str = "./chroma_db"):
    chroma = chromadb.PersistentClient(path=persist_path)

    text_collection = chroma.get_or_create_collection(
        name="text_index",
        embedding_function=openai_ef,
        metadata={"hnsw:space": "cosine"}
    )

    image_collection = chroma.get_or_create_collection(
        name="image_index",
        metadata={"hnsw:space": "cosine"}
    )

    return chroma, text_collection, image_collection


def add_clip_images(
    image_collection,
    images: list[dict]
) -> int:
    for img in images:
        pil_image = Image.open(img["path"]).convert("RGB")
        inputs = clip_processor(images=pil_image, return_tensors="pt")

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

        image_collection.add(
            documents=[img.get("description", img["path"])],
            embeddings=[embedding],
            ids=[f"clip_img_{img['image_index']}"],
            metadatas=[{
                "type": "image",
                "source": img.get("source", ""),
                "page": img.get("page", 0),
                "image_path": img["path"],
            }]
        )

    return len(images)


def search_clip_images(
    image_collection,
    text_query: str,
    n: int = 5
) -> list[dict]:
    inputs = clip_processor(text=[text_query], return_tensors="pt", padding=True)
    with torch.no_grad():
        features = clip_model.get_text_features(**inputs)
    normalized = features / features.norm(dim=-1, keepdim=True)
    query_embedding = normalized[0].numpy().tolist()

    results = image_collection.query(
        query_embeddings=[query_embedding],
        n_results=n
    )

    matches = []
    for i in range(len(results["documents"][0])):
        matches.append({
            "description": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
        })
    return matches

Troubleshooting

Error: "Embedding dimension mismatch"

Intentas añadir embeddings de 512 dimensiones (CLIP) a una collection que espera 1536 (OpenAI).

Solución: Usa collections separadas para cada tipo de embedding.
  - "text_index" con embedding_function de OpenAI
  - "image_index" sin embedding_function (provees embeddings manualmente)

Error: "ID already exists"

ChromaDB rechaza IDs duplicados. Asegúrate de que tus IDs sean únicos.

import hashlib

def generate_unique_id(source: str, content_type: str, index: int) -> str:
    raw = f"{source}_{content_type}_{index}"
    return hashlib.md5(raw.encode()).hexdigest()[:16]

El índice no persiste entre ejecuciones

Estás usando chromadb.Client() en lugar de chromadb.PersistentClient().

client = chromadb.PersistentClient(path="./chroma_db")

Indexación lenta con muchas imágenes

Cada imagen requiere una llamada a la API de Vision. Para 100 imágenes, son 100 llamadas secuenciales.

import asyncio
from openai import AsyncOpenAI


async def index_images_async(
    images: list[dict],
    collection,
    source: str,
    max_concurrent: int = 5
) -> int:
    async_client = AsyncOpenAI()
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_image(img: dict) -> dict | None:
        async with semaphore:
            try:
                import base64
                with open(img["path"], "rb") as f:
                    b64 = base64.b64encode(f.read()).decode("utf-8")

                response = await async_client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{
                        "role": "user",
                        "content": [
                            {"type": "text", "text": DESCRIPTION_PROMPT},
                            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
                        ]
                    }],
                    max_tokens=150
                )
                return {
                    "description": response.choices[0].message.content,
                    "path": img["path"],
                    "page": img.get("page", 0),
                    "image_index": img["image_index"],
                }
            except Exception as e:
                print(f"Error procesando {img['path']}: {e}")
                return None

    tasks = [process_image(img) for img in images]
    results = await asyncio.gather(*tasks)

    valid_results = [r for r in results if r is not None]

    documents = [r["description"] for r in valid_results]
    ids = [f"{source}_img_{r['image_index']}" for r in valid_results]
    metadatas = [{
        "type": "image",
        "source": source,
        "page": r["page"],
        "image_path": r["path"],
    } for r in valid_results]

    if documents:
        collection.add(documents=documents, ids=ids, metadatas=metadatas)

    return len(valid_results)

ChromaDB consume mucha memoria

Con índices grandes (>100K documentos), ChromaDB en memoria puede consumir GBs.

Solución:
1. Usa PersistentClient para que los datos estén en disco
2. Configura el parámetro hnsw:M (default 16) — valores más bajos usan menos memoria
3. Para producción, considera Pinecone, Weaviate o Qdrant

Ejercicios

Ejercicio 1: Persistir y recuperar índice

Crea un índice persistente, añade documentos, cierra la conexión, y verifica que los datos persisten al reconectar.

def test_persistence():
    persist_path = "./test_chroma_persist"

    client1 = chromadb.PersistentClient(path=persist_path)
    coll1 = client1.get_or_create_collection("persist_test", embedding_function=openai_ef)
    coll1.add(
        documents=["Documento de prueba para persistencia"],
        ids=["persist_1"],
        metadatas=[{"type": "text"}]
    )
    count_before = coll1.count()
    del client1, coll1

    client2 = chromadb.PersistentClient(path=persist_path)
    coll2 = client2.get_collection("persist_test", embedding_function=openai_ef)
    count_after = coll2.count()

    assert count_before == count_after, "Los datos no persistieron"
    print(f"Persistencia verificada: {count_after} documentos")

    results = coll2.query(query_texts=["prueba"], n_results=1)
    print(f"Resultado: {results['documents'][0][0]}")

    client2.delete_collection("persist_test")


test_persistence()

Ejercicio 2: Eliminar y re-indexar documento

Implementa un flujo donde borras todos los chunks de un documento específico y lo vuelves a indexar con un chunk_size diferente.

def reindex_with_new_settings(
    collection,
    source: str,
    new_chunks: list[dict]
) -> dict:
    old_docs = collection.get(where={"source": source})
    old_count = len(old_docs["ids"])

    if old_docs["ids"]:
        collection.delete(ids=old_docs["ids"])

    new_count = add_text_chunks(collection, new_chunks, source)

    return {
        "source": source,
        "deleted": old_count,
        "added": new_count,
        "net_change": new_count - old_count,
    }


result = reindex_with_new_settings(
    collection,
    source="manual_tecnico.pdf",
    new_chunks=[
        {"text": "Chunk más grande con más contexto...", "page": 1, "chunk_index": 0},
        {"text": "Segundo chunk con información adicional...", "page": 1, "chunk_index": 1},
    ]
)
print(f"Re-indexado: eliminados {result['deleted']}, añadidos {result['added']}")

Ejercicio 3: Filtros avanzados de metadata

Implementa una función de búsqueda que permita filtrar por tipo de contenido, fuente, y rango de páginas.

def advanced_search(
    collection,
    query: str,
    content_type: str = None,
    source: str = None,
    page_min: int = None,
    page_max: int = None,
    n_results: int = 5
) -> list[dict]:
    conditions = []

    if content_type:
        conditions.append({"type": content_type})
    if source:
        conditions.append({"source": source})
    if page_min is not None:
        conditions.append({"page": {"$gte": page_min}})
    if page_max is not None:
        conditions.append({"page": {"$lte": page_max}})

    kwargs = {"query_texts": [query], "n_results": n_results}

    if len(conditions) == 1:
        kwargs["where"] = conditions[0]
    elif len(conditions) > 1:
        kwargs["where"] = {"$and": conditions}

    results = collection.query(**kwargs)

    matches = []
    for i in range(len(results["documents"][0])):
        matches.append({
            "content": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
        })
    return matches


results = advanced_search(
    collection,
    query="autenticación",
    content_type="text",
    page_min=1,
    page_max=10,
    n_results=3
)
for r in results:
    print(f"  p.{r['metadata'].get('page', '?')} [{r['distance']:.4f}] {r['content'][:50]}...")

Resumen

• Un índice multimodal almacena embeddings de texto e imágenes juntos (o en collections coordinadas).
• La metadata es clave: tipo, fuente, página, ruta de imagen permiten filtros eficientes en queries.
• ChromaDB ofrece dos modos: en memoria (desarrollo rápido) y persistente (datos duraderos).
• El pipeline de indexación extrae texto e imágenes del PDF, genera embeddings/descripciones, y los almacena.
• Para imágenes, indexas la descripción textual generada por Vision como documento (estrategia 1) o embeddings CLIP directos (estrategia 2, collection separada).
• Actualizar documentos requiere borrar los chunks existentes y re-indexar.
• Para volúmenes grandes, usa indexación async para describir imágenes en paralelo.

Recursos Adicionales

1. ChromaDB Documentation — Guía completa de ChromaDB
2. ChromaDB Filtering — Filtros avanzados de metadata
3. PyMuPDF Documentation — Extracción de texto e imágenes de PDFs
4. FAISS — Alternativa de Meta para búsqueda vectorial
5. Chunking Strategies — Estrategias de chunking para RAG