3. Multimodal Indexing

Description

Having text and image embeddings is only half the problem. The other half is storing them so that search is efficient, organized and filterable. In this capsule you'll build a multimodal index in ChromaDB that combines text chunks and image representations, with metadata that lets you filter by type, source, page and domain.

Multimodal indexing differs from text-only indexing in three key aspects: you need to distinguish between content types (text vs image), you need to preserve the relationship between a text chunk and the images of its section, and you need to design the metadata so queries can filter efficiently.

Why it matters: A poorly designed index produces irrelevant search results no matter how good your embeddings are. If you don't distinguish text from images, you can't filter. If you don't store the source page, you can't cite sources. If you don't persist the index, you lose it every time you restart.

Connection with the module: The embeddings from capsule 02 are stored here. The retrieval from capsule 04 searches this index. The document processing from capsule 05 feeds this index. It's the central piece.

Structure of the Multimodal Index

Document design

Each entry in the index needs:

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

For images:

{
    "id": "doc_manual_img_03",
    "document": "Architecture diagram showing API Gateway connected to three microservices: auth, payments, notifications",
    "embedding": [0.08, -0.22, 0.11, ...],
    "metadata": {
        "type": "image",
        "source": "technical_manual.pdf",
        "page": 12,
        "image_path": "images/technical_manual_p12_img0.png",
        "original_format": "png",
        "description_model": "gpt-4o-mini",
        "domain": "technical",
        "created_at": "2024-01-15T10:30:00Z"
    }
}

Recommended metadata fields

ChromaDB: Complete Setup

Connection and 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"Existing documents: {collection.count()}")

Difference between Client and PersistentClient

# IN MEMORY — lost when the process closes
client_memory = chromadb.Client()

# PERSISTENT — saved to disk
client_persistent = chromadb.PersistentClient(path="./chroma_db")

For development and quick tests, use Client(). For anything you want to keep between runs, use PersistentClient().

Indexing Text Chunks

Basic function

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": "The API Gateway is the single entry point for all clients. It distributes requests among the microservices according to the configured routes.",
        "page": 1,
        "chunk_index": 0,
        "has_images": True
    },
    {
        "text": "The authentication service validates JWT tokens. Each request must include a valid token in the Authorization header.",
        "page": 2,
        "chunk_index": 1,
        "has_images": False
    },
    {
        "text": "The PostgreSQL database stores user and transaction data. Connection pooling with a maximum of 20 connections is recommended.",
        "page": 3,
        "chunk_index": 2,
        "has_images": False
    },
]

added = add_text_chunks(collection, sample_chunks, source="technical_manual.pdf")
print(f"Text chunks added: {added}")

Text chunking for RAG

The chunk size directly affects retrieval quality. Chunks that are too small lose context. Chunks that are too large dilute the signal.

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 = "This is a long document " * 200
chunks = chunk_text(long_text, chunk_size=50, overlap=10)
print(f"Text of {len(long_text.split())} words → {len(chunks)} chunks")
for c in chunks[:3]:
    print(f"  Chunk {c['chunk_index']}: {len(c['text'].split())} words")

Chunking with semantic separators

For structured documents (markdown, HTML), it's better to split by sections than by word count.

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

Indexing Image Descriptions

Function to index images with descriptions

from openai import OpenAI

client = OpenAI()

DESCRIPTION_PROMPT = (
    "Describe this image concisely and precisely for indexing "
    "in a search system. Include: what it shows, main elements, "
    "type of content (diagram, photo, chart, table). Maximum 2-3 sentences."
)


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)

Indexing with precomputed embeddings

If you already have the embeddings (for example, from CLIP or a previous batch), you can pass them directly.

def add_with_precomputed_embeddings(
    collection,
    items: list[dict]
) -> int:
    """
    items: [{
        "id": "...",
        "document": "text or description",
        "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)

Complete Indexing Pipeline

Process a document: extract text and images, index everything

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 indexing images: {e}")

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

    return stats

Index multiple documents

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 with {pdf_path}: {e}")

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

Querying the Index

Search with metadata filters

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, "microservices", n_results=3)
for r in results:
    print(f"  [{r['metadata'].get('type', '?')}] {r['distance']:.4f} — {r['content'][:60]}...")

text_only = query_index(collection, "authentication", content_type="text")
images_only = query_index(collection, "architecture diagram", content_type="image")

Get index statistics

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

Updating and Deleting Documents

Delete all chunks of a document

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, "technical_manual.pdf")
print(f"Deleted {deleted} chunks")

Re-index a document

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

    deleted = delete_document(collection, source)
    print(f"Deleted {deleted} previous chunks from {source}")

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

Update metadata without re-indexing

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

Hybrid Index: Text + CLIP Embeddings

If you use CLIP for images, you need a separate collection because the embedding dimension is different (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"

You're trying to add 512-dimension embeddings (CLIP) to a collection that expects 1536 (OpenAI).

Solution: Use separate collections for each embedding type.
  - "text_index" with OpenAI's embedding_function
  - "image_index" without embedding_function (you provide embeddings manually)

Error: "ID already exists"

ChromaDB rejects duplicate IDs. Make sure your IDs are unique.

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]

The index doesn't persist between runs

You're using chromadb.Client() instead of chromadb.PersistentClient().

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

Slow indexing with many images

Each image requires a Vision API call. For 100 images, that's 100 sequential calls.

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 processing {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 uses a lot of memory

With large indexes (>100K documents), ChromaDB in memory can consume GBs.

Solution:
1. Use PersistentClient so data lives on disk
2. Configure the hnsw:M parameter (default 16) — lower values use less memory
3. For production, consider Pinecone, Weaviate or Qdrant

Exercises

Exercise 1: Persist and recover an index

Create a persistent index, add documents, close the connection, and verify that the data persists on reconnect.

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=["Test document for persistence"],
        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, "The data did not persist"
    print(f"Persistence verified: {count_after} documents")

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

    client2.delete_collection("persist_test")


test_persistence()

Exercise 2: Delete and re-index a document

Implement a flow where you delete all chunks of a specific document and index it again with a different chunk_size.

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="technical_manual.pdf",
    new_chunks=[
        {"text": "Larger chunk with more context...", "page": 1, "chunk_index": 0},
        {"text": "Second chunk with additional information...", "page": 1, "chunk_index": 1},
    ]
)
print(f"Re-indexed: deleted {result['deleted']}, added {result['added']}")

Exercise 3: Advanced metadata filters

Implement a search function that lets you filter by content type, source, and page range.

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="authentication",
    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]}...")

Summary

• A multimodal index stores text and image embeddings together (or in coordinated collections).
• Metadata is key: type, source, page, image path allow efficient filters in queries.
• ChromaDB offers two modes: in memory (fast development) and persistent (durable data).
• The indexing pipeline extracts text and images from the PDF, generates embeddings/descriptions, and stores them.
• For images, you index the textual description generated by Vision as a document (strategy 1) or direct CLIP embeddings (strategy 2, separate collection).
• Updating documents requires deleting the existing chunks and re-indexing.
• For large volumes, use async indexing to describe images in parallel.

Additional Resources

1. ChromaDB Documentation — Complete ChromaDB guide
2. ChromaDB Filtering — Advanced metadata filters
3. PyMuPDF Documentation — Extracting text and images from PDFs
4. FAISS — Meta's alternative for vector search
5. Chunking Strategies — Chunking strategies for RAG