Module 6: Multimodal RAG

5. Documents with Images

Description

Real-world documents aren't pure text. A technical manual has architecture diagrams. A financial report has trend charts. An academic paper has figures with experimental results. A product catalog has photos. Multimodal RAG only works if you can process these documents while preserving the relationship between text and images.

In this capsule you'll learn strategies for chunking documents that contain figures, how to associate each image with the surrounding text, and how to build a complete pipeline that takes a PDF with mixed content and prepares it for indexing.

Why it matters: If you extract text and images separately without preserving their relationship, you lose critical context. A text chunk that says "as Figure 3 shows" is useless if you don't know what Figure 3 is. And an image description that says "bar chart showing data" is vague if you don't include the title and the text that explains it.

Connection with the module: This capsule feeds the indexing pipeline from capsule 03. The chunks you generate here are searched with the retrieval from capsule 04. And the project in capsule 08 processes real documents with this logic.


The Problem: Mixed Documents

What a typical document contains

Technical manual (PDF, 50 pages):
├── Body text: explanations, descriptions, procedures
├── Diagrams: architecture, flow, entity-relationship
├── Tables: configurations, comparisons, specifications
├── Screenshots: interface, logs, configuration
├── Referenced figures: "see Figure 3", "as Diagram 2 shows"
└── Headers/footers: section titles, page numbers

What happens if you ignore the images

Question: "How does the payment service connect to the database?"

Text only:
  → Retrieves: "The payment service connects to PostgreSQL via a connection pool"
  → Answer: partial, without the diagram showing the full flow

Text + images:
  → Retrieves: paragraph + "Diagram showing the payment service connected
     to PostgreSQL via PgBouncer with a pool of 20 connections"
  → Answer: complete, includes diagram details

What happens if you don't preserve the relationship

Text chunk: "As Figure 3 shows, the authentication flow
  goes through three stages."

Image chunk (unrelated): "Flow diagram with three boxes
  connected by arrows."

→ Retrieval may find the text OR the image, but doesn't know
  they go together. The answer loses coherence.

Chunking Strategies for Mixed Documents

Strategy 1: Chunk per page

Each page of the PDF is a chunk. The chunk includes all the page's text plus references to that page's images.

import fitz
from pathlib import Path


def chunk_by_page(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    chunks = []

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

        if not text and not images:
            continue

        chunks.append({
            "text": text,
            "page": page_num + 1,
            "chunk_index": page_num,
            "has_images": len(images) > 0,
            "image_count": len(images),
            "strategy": "page",
        })

    doc.close()
    return chunks

Advantages: Simple. A page's images are naturally associated with that page's text.

Disadvantages: Long pages produce large chunks that dilute the signal. Pages with just one image produce chunks empty of text.

Strategy 2: Chunk per section with associated images

Split by sections (headers) and associate those pages' images with the corresponding section.

import re


def chunk_by_section(pdf_path: str, max_chunk_words: int = 500) -> list[dict]:
    doc = fitz.open(pdf_path)
    full_text = ""
    page_boundaries = {}
    current_pos = 0

    for page_num in range(len(doc)):
        page = doc[page_num]
        page_text = page.get_text()
        page_boundaries[page_num + 1] = (current_pos, current_pos + len(page_text))
        full_text += page_text
        current_pos += len(page_text)

    doc.close()

    sections = re.split(r'\n(?=[A-Z][^\n]{3,50}\n)', full_text)

    chunks = []
    for i, section in enumerate(sections):
        section = section.strip()
        if not section:
            continue

        section_page = 1
        section_start = full_text.find(section)
        for page, (start, end) in page_boundaries.items():
            if start <= section_start < end:
                section_page = page
                break

        words = section.split()
        if len(words) > max_chunk_words:
            sub_chunks = split_text_with_overlap(section, max_chunk_words, overlap=50)
            for j, sub in enumerate(sub_chunks):
                chunks.append({
                    "text": sub,
                    "page": section_page,
                    "chunk_index": len(chunks),
                    "strategy": "section",
                    "section_part": j,
                })
        else:
            chunks.append({
                "text": section,
                "page": section_page,
                "chunk_index": len(chunks),
                "strategy": "section",
            })

    return chunks


def split_text_with_overlap(
    text: str,
    max_words: int,
    overlap: int = 50
) -> list[str]:
    words = text.split()
    chunks = []
    start = 0

    while start < len(words):
        end = start + max_words
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start += max_words - overlap

    return chunks

Strategy 3: Chunk enriched with image context

The text chunk includes the image descriptions as part of the content. That way the embedding captures both the text and the visual content.

from openai import OpenAI
import base64

client = OpenAI()


def describe_image_bytes(image_bytes: bytes, ext: str = "png") -> str:
    b64 = base64.b64encode(image_bytes).decode("utf-8")
    mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Describe this image in 1-2 sentences. Include the content type and main elements."
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:{mime};base64,{b64}"}
                }
            ]
        }],
        max_tokens=100
    )
    return response.choices[0].message.content


def enrich_chunks_with_image_context(
    chunks: list[dict],
    images_by_page: dict[int, list[dict]]
) -> list[dict]:
    enriched = []

    for chunk in chunks:
        page = chunk.get("page", 0)
        page_images = images_by_page.get(page, [])

        if page_images:
            descriptions = []
            for img in page_images:
                desc = describe_image_bytes(img["data"], img.get("ext", "png"))
                descriptions.append(desc)

            image_context = "\n\n".join(
                [f"[Figure on page {page}: {d}]" for d in descriptions]
            )
            enriched_text = f"{chunk['text']}\n\n{image_context}"

            enriched.append({
                **chunk,
                "text": enriched_text,
                "has_images": True,
                "image_descriptions": descriptions,
            })
        else:
            enriched.append({
                **chunk,
                "has_images": False,
                "image_descriptions": [],
            })

    return enriched

Comparison of strategies

StrategyBest forTypical chunksRelationship quality
Per pageShort PDFs, well-laid-out documents~1 per pageMedium — the whole page goes together
Per sectionDocuments with clear headers (markdown, HTML)VariableHigh — thematic sections
EnrichedAny documentSame as base + contextHigh — text + image description

Extract Images from a PDF

Extraction with PyMuPDF

def extract_images_by_page(pdf_path: str) -> dict[int, list[dict]]:
    doc = fitz.open(pdf_path)
    images_by_page = {}

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

        if not img_list:
            continue

        page_images = []
        for img_ref in img_list:
            xref = img_ref[0]
            try:
                base_image = doc.extract_image(xref)
                if base_image and base_image.get("image"):
                    page_images.append({
                        "data": base_image["image"],
                        "ext": base_image.get("ext", "png"),
                        "width": base_image.get("width", 0),
                        "height": base_image.get("height", 0),
                        "xref": xref,
                    })
            except Exception:
                continue

        if page_images:
            images_by_page[page_num + 1] = page_images

    doc.close()
    return images_by_page

Save extracted images

def save_extracted_images(
    images_by_page: dict[int, list[dict]],
    output_dir: str,
    pdf_name: str
) -> dict[int, list[dict]]:
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    saved = {}
    for page, images in images_by_page.items():
        saved[page] = []
        for i, img in enumerate(images):
            filename = f"{pdf_name}_p{page}_img{i}.{img['ext']}"
            filepath = output / filename

            with open(filepath, "wb") as f:
                f.write(img["data"])

            saved[page].append({
                **img,
                "path": str(filepath),
                "filename": filename,
            })

    return saved

Filter out irrelevant images

Not all images in a PDF are useful. Logos, small icons, separators and backgrounds are noise.

MIN_IMAGE_WIDTH = 100
MIN_IMAGE_HEIGHT = 100
MIN_IMAGE_BYTES = 5000


def filter_relevant_images(
    images_by_page: dict[int, list[dict]]
) -> dict[int, list[dict]]:
    filtered = {}

    for page, images in images_by_page.items():
        relevant = []
        for img in images:
            width = img.get("width", 0)
            height = img.get("height", 0)
            size = len(img.get("data", b""))

            if width < MIN_IMAGE_WIDTH or height < MIN_IMAGE_HEIGHT:
                continue
            if size < MIN_IMAGE_BYTES:
                continue

            relevant.append(img)

        if relevant:
            filtered[page] = relevant

    return filtered

Complete Pipeline: PDF → Chunks with Image Context

The integrated pipeline

def process_document_with_images(
    pdf_path: str,
    strategy: str = "enriched",
    chunk_size: int = 500,
    output_dir: str = "./extracted_images",
    describe_images: bool = True
) -> list[dict]:
    pdf_name = Path(pdf_path).stem

    if strategy == "page":
        chunks = chunk_by_page(pdf_path)
    elif strategy == "section":
        chunks = chunk_by_section(pdf_path, max_chunk_words=chunk_size)
    else:
        chunks = chunk_by_section(pdf_path, max_chunk_words=chunk_size)

    images_by_page = extract_images_by_page(pdf_path)
    images_by_page = filter_relevant_images(images_by_page)

    saved_images = save_extracted_images(images_by_page, output_dir, pdf_name)

    if strategy == "enriched" and describe_images:
        chunks = enrich_chunks_with_image_context(chunks, images_by_page)

    image_chunks = []
    for page, images in saved_images.items():
        for i, img in enumerate(images):
            if describe_images:
                desc = describe_image_bytes(img["data"], img.get("ext", "png"))
            else:
                desc = f"Image on page {page}"

            image_chunks.append({
                "text": desc,
                "page": page,
                "chunk_index": len(chunks) + len(image_chunks),
                "type": "image",
                "image_path": img.get("path", ""),
                "has_images": True,
            })

    for chunk in chunks:
        if "type" not in chunk:
            chunk["type"] = "text"

    all_chunks = chunks + image_chunks

    return all_chunks


# chunks = process_document_with_images("technical_manual.pdf")
# print(f"Total chunks: {len(chunks)}")
# text_chunks = [c for c in chunks if c["type"] == "text"]
# image_chunks = [c for c in chunks if c["type"] == "image"]
# print(f"  Text: {len(text_chunks)}, Images: {len(image_chunks)}")

Index the result

import chromadb
from chromadb.utils import embedding_functions
import os

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


def index_processed_document(
    chunks: list[dict],
    collection,
    source: str
) -> dict:
    stats = {"text": 0, "image": 0, "errors": 0}

    documents = []
    ids = []
    metadatas = []

    for chunk in chunks:
        chunk_type = chunk.get("type", "text")
        chunk_id = f"{source}_{chunk_type}_{chunk['chunk_index']}"

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

        stats[chunk_type] = stats.get(chunk_type, 0) + 1

    if documents:
        try:
            collection.add(
                documents=documents,
                ids=ids,
                metadatas=metadatas
            )
        except Exception as e:
            print(f"Error indexing: {e}")
            stats["errors"] += 1

    return stats

Handle Figure References

Detect references in the text

Many technical documents use "see Figure X", "as Diagram Y shows", etc. Detecting these references lets you associate text chunks with the correct images.

import re


def find_figure_references(text: str) -> list[dict]:
    patterns = [
        r'(?:see)\s+(?:the\s+)?(?:Figure|Fig\.?)\s*(\d+)',
        r'(?:Figure|Fig\.?)\s*(\d+)',
        r'(?:Diagram)\s*(\d+)',
        r'(?:Table)\s*(\d+)',
        r'(?:Graph|Chart)\s*(\d+)',
    ]

    references = []
    for pattern in patterns:
        for match in re.finditer(pattern, text, re.IGNORECASE):
            references.append({
                "type": "figure",
                "number": int(match.group(1)),
                "span": match.span(),
                "context": text[max(0, match.start()-50):match.end()+50],
            })

    return references


sample = "As Figure 3 shows, the payment service connects to the gateway via Diagram 1."
refs = find_figure_references(sample)
for r in refs:
    print(f"  Reference: {r['type']} #{r['number']} → ...{r['context']}...")

Resolve references: associate text with images

def resolve_figure_references(
    text_chunks: list[dict],
    image_chunks: list[dict],
    images_by_page: dict
) -> list[dict]:
    figure_number_to_image = {}
    sorted_images = []
    for page in sorted(images_by_page.keys()):
        for img in images_by_page[page]:
            sorted_images.append({"page": page, **img})

    for i, img in enumerate(sorted_images):
        figure_number_to_image[i + 1] = img

    resolved_chunks = []
    for chunk in text_chunks:
        refs = find_figure_references(chunk["text"])

        associated_images = []
        for ref in refs:
            fig_num = ref["number"]
            if fig_num in figure_number_to_image:
                associated_images.append(figure_number_to_image[fig_num])

        resolved_chunks.append({
            **chunk,
            "figure_references": refs,
            "associated_images": associated_images,
        })

    return resolved_chunks

Enrich chunks with resolved figures

def enrich_with_resolved_figures(
    resolved_chunks: list[dict]
) -> list[dict]:
    enriched = []

    for chunk in resolved_chunks:
        if chunk["associated_images"]:
            image_descriptions = []
            for img in chunk["associated_images"]:
                desc = describe_image_bytes(img["data"], img.get("ext", "png"))
                image_descriptions.append(desc)

            fig_context = "\n".join(
                [f"[Referenced figure: {d}]" for d in image_descriptions]
            )
            enriched_text = f"{chunk['text']}\n\n{fig_context}"

            enriched.append({
                **chunk,
                "text": enriched_text,
                "image_descriptions": image_descriptions,
            })
        else:
            enriched.append(chunk)

    return enriched

Handle Tables in Documents

Extract tables with PyMuPDF

def extract_tables_from_page(page) -> list[dict]:
    tables = page.find_tables()
    extracted = []

    for table in tables:
        rows = table.extract()
        if not rows:
            continue

        headers = rows[0] if rows else []
        data_rows = rows[1:] if len(rows) > 1 else []

        markdown = format_table_as_markdown(headers, data_rows)

        extracted.append({
            "markdown": markdown,
            "headers": headers,
            "rows": data_rows,
            "row_count": len(data_rows),
        })

    return extracted


def format_table_as_markdown(
    headers: list,
    rows: list[list]
) -> str:
    if not headers:
        return ""

    clean_headers = [str(h).strip() if h else "" for h in headers]
    header_line = "| " + " | ".join(clean_headers) + " |"
    separator = "| " + " | ".join(["---"] * len(clean_headers)) + " |"

    lines = [header_line, separator]
    for row in rows:
        clean_cells = [str(c).strip() if c else "" for c in row]
        while len(clean_cells) < len(clean_headers):
            clean_cells.append("")
        lines.append("| " + " | ".join(clean_cells[:len(clean_headers)]) + " |")

    return "\n".join(lines)

Incorporate tables into chunking

def extract_tables_by_page(pdf_path: str) -> dict[int, list[dict]]:
    doc = fitz.open(pdf_path)
    tables_by_page = {}

    for page_num in range(len(doc)):
        page = doc[page_num]
        tables = extract_tables_from_page(page)

        if tables:
            tables_by_page[page_num + 1] = tables

    doc.close()
    return tables_by_page


def enrich_chunks_with_tables(
    chunks: list[dict],
    tables_by_page: dict[int, list[dict]]
) -> list[dict]:
    enriched = []

    for chunk in chunks:
        page = chunk.get("page", 0)
        page_tables = tables_by_page.get(page, [])

        if page_tables:
            table_context = "\n\n".join(
                [f"[Table on page {page}:\n{t['markdown']}]" for t in page_tables]
            )
            enriched_text = f"{chunk['text']}\n\n{table_context}"
            enriched.append({**chunk, "text": enriched_text, "has_tables": True})
        else:
            enriched.append({**chunk, "has_tables": False})

    return enriched

Advanced Pipeline: Text + Images + Tables

def process_document_full(
    pdf_path: str,
    chunk_size: int = 500,
    output_dir: str = "./extracted_images",
    describe_images: bool = True
) -> dict:
    pdf_name = Path(pdf_path).stem

    text_chunks = chunk_by_section(pdf_path, max_chunk_words=chunk_size)

    images_by_page = extract_images_by_page(pdf_path)
    images_by_page = filter_relevant_images(images_by_page)
    saved_images = save_extracted_images(images_by_page, output_dir, pdf_name)

    tables_by_page = extract_tables_by_page(pdf_path)

    if describe_images:
        text_chunks = enrich_chunks_with_image_context(text_chunks, images_by_page)

    text_chunks = enrich_chunks_with_tables(text_chunks, tables_by_page)

    image_only_chunks = []
    for page, images in saved_images.items():
        for i, img in enumerate(images):
            desc = describe_image_bytes(img["data"], img.get("ext", "png")) if describe_images else f"Image p.{page}"
            image_only_chunks.append({
                "text": desc,
                "page": page,
                "chunk_index": len(text_chunks) + len(image_only_chunks),
                "type": "image",
                "image_path": img.get("path", ""),
            })

    for c in text_chunks:
        c["type"] = "text"

    all_chunks = text_chunks + image_only_chunks

    stats = {
        "total_chunks": len(all_chunks),
        "text_chunks": len(text_chunks),
        "image_chunks": len(image_only_chunks),
        "pages_with_images": len(images_by_page),
        "pages_with_tables": len(tables_by_page),
        "total_images": sum(len(imgs) for imgs in images_by_page.values()),
        "total_tables": sum(len(tbls) for tbls in tables_by_page.values()),
    }

    return {"chunks": all_chunks, "stats": stats}

Troubleshooting

PyMuPDF doesn't extract images from certain PDFs

Some PDFs have images embedded as vectors (SVG) or as part of the page rendering, not as raster images. PyMuPDF only extracts raster.

def extract_page_as_image(
    pdf_path: str,
    page_num: int,
    dpi: int = 150
) -> bytes:
    doc = fitz.open(pdf_path)
    page = doc[page_num]
    mat = fitz.Matrix(dpi / 72, dpi / 72)
    pix = page.get_pixmap(matrix=mat)
    image_bytes = pix.tobytes("png")
    doc.close()
    return image_bytes

Chunks too long due to large tables

If a table has 100 rows, the resulting chunk is huge.

MAX_TABLE_ROWS = 20

def truncate_table(table: dict, max_rows: int = MAX_TABLE_ROWS) -> dict:
    if table["row_count"] <= max_rows:
        return table

    truncated_rows = table["rows"][:max_rows]
    markdown = format_table_as_markdown(table["headers"], truncated_rows)
    markdown += f"\n... ({table['row_count'] - max_rows} more rows)"

    return {**table, "markdown": markdown, "rows": truncated_rows}

Duplicate image descriptions

If the same image appears on multiple pages (logos, headers), it gets described multiple times.

import hashlib


def deduplicate_images(
    images_by_page: dict[int, list[dict]]
) -> dict[int, list[dict]]:
    seen_hashes = set()
    deduped = {}

    for page, images in images_by_page.items():
        unique = []
        for img in images:
            img_hash = hashlib.md5(img["data"]).hexdigest()
            if img_hash not in seen_hashes:
                seen_hashes.add(img_hash)
                unique.append(img)
        if unique:
            deduped[page] = unique

    return deduped

References to figures that don't exist

The text says "see Figure 5" but there are only 3 images in the document.

Solution: Log the unresolved reference and continue without enriching that chunk.
Don't fail the whole pipeline over one broken reference.

Exercises

Exercise 1: Associate an image with a text chunk by proximity

Given a text chunk and a list of images by page, return the images from the same page or adjacent pages (± 1).

See solution
def get_nearby_images(
    chunk: dict,
    images_by_page: dict[int, list[dict]],
    range_pages: int = 1
) -> list[dict]:
    page = chunk.get("page", 0)
    nearby = []

    for p in range(page - range_pages, page + range_pages + 1):
        if p in images_by_page:
            for img in images_by_page[p]:
                nearby.append({**img, "from_page": p})

    return nearby


chunk = {"text": "The authentication flow...", "page": 5}
images_by_page = {
    4: [{"data": b"...", "ext": "png"}],
    5: [{"data": b"...", "ext": "png"}, {"data": b"...", "ext": "jpg"}],
    7: [{"data": b"...", "ext": "png"}],
}

nearby = get_nearby_images(chunk, images_by_page, range_pages=1)
print(f"Images near chunk p.{chunk['page']}: {len(nearby)}")
for img in nearby:
    print(f"  - Page {img['from_page']}, format {img['ext']}")

Exercise 2: Processing pipeline with statistics

Implement a pipeline that processes a PDF and returns detailed statistics: chunks by type, pages with images, average chunk size, etc.

See solution
def process_with_stats(pdf_path: str) -> dict:
    result = process_document_full(pdf_path, describe_images=False)
    chunks = result["chunks"]

    text_chunks = [c for c in chunks if c.get("type") == "text"]
    image_chunks = [c for c in chunks if c.get("type") == "image"]

    text_lengths = [len(c["text"].split()) for c in text_chunks]
    avg_text_length = sum(text_lengths) / len(text_lengths) if text_lengths else 0

    pages_with_images = set()
    for c in chunks:
        if c.get("has_images") or c.get("type") == "image":
            pages_with_images.add(c.get("page", 0))

    stats = {
        "file": pdf_path,
        "total_chunks": len(chunks),
        "text_chunks": len(text_chunks),
        "image_chunks": len(image_chunks),
        "avg_text_words": round(avg_text_length),
        "min_text_words": min(text_lengths) if text_lengths else 0,
        "max_text_words": max(text_lengths) if text_lengths else 0,
        "pages_with_images": len(pages_with_images),
        "enriched_text_chunks": sum(1 for c in text_chunks if c.get("has_images")),
    }

    return stats

Exercise 3: Compare chunking strategies

Process the same PDF with all three strategies (page, section, enriched) and compare the number of chunks, average size, and image coverage.

See solution
def compare_chunking_strategies(pdf_path: str) -> dict:
    strategies = {}

    for strategy_name in ["page", "section", "enriched"]:
        if strategy_name in ("page", "section"):
            if strategy_name == "page":
                chunks = chunk_by_page(pdf_path)
            else:
                chunks = chunk_by_section(pdf_path)
            for c in chunks:
                c["type"] = "text"
        else:
            result = process_document_full(pdf_path, describe_images=False)
            chunks = result["chunks"]

        text_chunks = [c for c in chunks if c.get("type") == "text"]
        word_counts = [len(c["text"].split()) for c in text_chunks]

        strategies[strategy_name] = {
            "total_chunks": len(chunks),
            "text_chunks": len(text_chunks),
            "avg_words": round(sum(word_counts) / len(word_counts)) if word_counts else 0,
            "with_image_context": sum(1 for c in text_chunks if c.get("has_images")),
        }

    print(f"Comparison for: {pdf_path}\n")
    print(f"{'Strategy':<15} {'Chunks':<10} {'Text':<10} {'Avg Words':<12} {'With Img':<10}")
    print("-" * 57)
    for name, stats in strategies.items():
        print(f"{name:<15} {stats['total_chunks']:<10} {stats['text_chunks']:<10} "
              f"{stats['avg_words']:<12} {stats['with_image_context']:<10}")

    return strategies

Summary

  • Real documents mix text, images, tables and cross-references.
  • Three chunking strategies: per page (simple), per section (semantic), enriched (text + image description).
  • Extract images with PyMuPDF and filter out the irrelevant ones (logos, small icons).
  • Associating images with their surrounding text preserves the context RAG needs.
  • Detect references ("see Figure 3") and resolve them against the extracted images.
  • Tables are extracted and converted to markdown to include in the chunks.
  • Deduplicate repeated images (logos on every page) reduces noise and cost.
  • The complete pipeline: PDF → extract text + images + tables → chunking → enrich → index.

Additional Resources

  1. PyMuPDF Documentation — Extracting text, images and tables
  2. Unstructured.io — Library specialized in document processing
  3. Document Chunking Strategies — Guide to chunking strategies
  4. LangChain Document Loaders — Loaders for different formats