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
| Strategy | Best for | Typical chunks | Relationship quality |
|---|---|---|---|
| Per page | Short PDFs, well-laid-out documents | ~1 per page | Medium — the whole page goes together |
| Per section | Documents with clear headers (markdown, HTML) | Variable | High — thematic sections |
| Enriched | Any document | Same as base + context | High — 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
- PyMuPDF Documentation — Extracting text, images and tables
- Unstructured.io — Library specialized in document processing
- Document Chunking Strategies — Guide to chunking strategies
- LangChain Document Loaders — Loaders for different formats