Module 6: Multimodal RAG
8. Project: Basic Multimodal RAG
Description
This project integrates everything you learned in the module: multimodal embeddings, indexing in vector stores, hybrid retrieval, processing documents with images, and answer generation with an LLM. You'll build a complete multimodal RAG system that takes PDFs with text and images, processes them, indexes them in ChromaDB, retrieves relevant fragments for a question, and generates answers citing the sources.
The system has four components: a document processor that extracts text and images from PDFs, an indexer that generates embeddings and stores them in ChromaDB, a hybrid retriever that searches by semantic similarity with metadata filters, and an answer generator that uses an LLM with the retrieved context.
What you're going to build: An end-to-end pipeline that receives PDFs, processes them preserving the text-image relationship, indexes the content, and answers questions based on what's indexed.
Connection with the module: It directly uses the embeddings from capsule 02, the indexing from capsule 03, the retrieval from capsule 04, the document processing from capsule 05, and optionally LangChain from capsule 06. The optimizations from capsule 07 apply as extensions.
Specifications
Input
- One or more PDF documents with text and images
- A user question in natural language
Output
- An answer generated by the LLM based exclusively on the retrieved content
- A list of sources used (document, page, content type)
- Pipeline metrics (chunks processed, images described, time)
Requirements
Functional:
1. Process PDFs, extracting text and images per page
2. Filter out irrelevant images (logos, icons, duplicates)
3. Describe relevant images with the Vision API
4. Generate chunks with associated image context
5. Index chunks in ChromaDB with metadata
6. Retrieve relevant chunks for a query
7. Generate an answer with the LLM using the chunks as context
8. Return sources and metrics along with the answer
Non-functional:
- Image description cache
- Error handling without stopping the whole pipeline
- Cost and time metrics per operation
- Support for multiple documents in the same collection
Setup
pip install openai chromadb pymupdf pillow python-dotenvimport os
from dotenv import load_dotenv
load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY missing in .env"Step 1: Description Cache
import json
import hashlib
from pathlib import Path
from datetime import datetime
class DescriptionCache:
def __init__(self, cache_dir: str = "./description_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.hits = 0
self.misses = 0
def _hash(self, image_data: bytes) -> str:
return hashlib.sha256(image_data).hexdigest()
def get(self, image_data: bytes) -> str | None:
cache_file = self.cache_dir / f"{self._hash(image_data)}.json"
if cache_file.exists():
self.hits += 1
return json.loads(cache_file.read_text()).get("description")
self.misses += 1
return None
def set(self, image_data: bytes, description: str, model: str = "") -> None:
img_hash = self._hash(image_data)
data = {
"description": description,
"model": model,
"hash": img_hash,
"created_at": datetime.now().isoformat(),
}
(self.cache_dir / f"{img_hash}.json").write_text(
json.dumps(data, ensure_ascii=False, indent=2)
)
def stats(self) -> dict:
total = self.hits + self.misses
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{self.hits / max(total, 1):.1%}",
"cached_files": len(list(self.cache_dir.glob("*.json"))),
}Step 2: Document Processor
The processor takes a PDF and produces a list of chunks. Each chunk can be plain text or an image description.
import fitz
import base64
from openai import OpenAI
oai_client = OpenAI()
MIN_IMAGE_WIDTH = 100
MIN_IMAGE_HEIGHT = 100
MIN_IMAGE_BYTES = 5000
class DocumentProcessor:
def __init__(
self,
cache: DescriptionCache,
output_dir: str = "./extracted_images",
max_chunk_words: int = 500,
vision_model: str = "gpt-4o-mini",
):
self.cache = cache
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.max_chunk_words = max_chunk_words
self.vision_model = vision_model
self.stats = {
"pages": 0, "images_extracted": 0,
"images_filtered": 0, "images_described": 0,
"text_chunks": 0, "image_chunks": 0,
}
def process(self, pdf_path: str) -> list[dict]:
doc = fitz.open(pdf_path)
pdf_name = Path(pdf_path).stem
all_chunks = []
for page_num in range(len(doc)):
page = doc[page_num]
self.stats["pages"] += 1
text = page.get_text().strip()
if text:
all_chunks.extend(self._split_text(text, page_num + 1, pdf_path))
all_chunks.extend(
self._process_images(doc, page, page_num, pdf_name, pdf_path)
)
doc.close()
for i, chunk in enumerate(all_chunks):
chunk["chunk_index"] = i
return all_chunks
def _split_text(self, text: str, page: int, source: str) -> list[dict]:
words = text.split()
chunks = []
start = 0
overlap = 50
while start < len(words):
end = start + self.max_chunk_words
chunk_text = " ".join(words[start:end])
chunks.append({
"text": chunk_text, "page": page,
"source": source, "type": "text",
"word_count": len(chunk_text.split()),
})
self.stats["text_chunks"] += 1
start += self.max_chunk_words - overlap
return chunks
def _process_images(
self, doc, page, page_num: int, pdf_name: str, source: str
) -> list[dict]:
chunks = []
for img_idx, img_ref in enumerate(page.get_images()):
xref = img_ref[0]
try:
base_image = doc.extract_image(xref)
if not base_image or not base_image.get("image"):
continue
image_data = base_image["image"]
width = base_image.get("width", 0)
height = base_image.get("height", 0)
if (width < MIN_IMAGE_WIDTH or height < MIN_IMAGE_HEIGHT
or len(image_data) < MIN_IMAGE_BYTES):
self.stats["images_filtered"] += 1
continue
self.stats["images_extracted"] += 1
ext = base_image.get("ext", "png")
filename = f"{pdf_name}_p{page_num + 1}_img{img_idx}.{ext}"
filepath = self.output_dir / filename
filepath.write_bytes(image_data)
description = self._describe_image(image_data, ext)
self.stats["images_described"] += 1
chunks.append({
"text": description, "page": page_num + 1,
"source": source, "type": "image",
"image_path": str(filepath),
})
except Exception:
continue
return chunks
def _describe_image(self, image_data: bytes, ext: str) -> str:
cached = self.cache.get(image_data)
if cached:
return cached
b64 = base64.b64encode(image_data).decode("utf-8")
mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
response = oai_client.chat.completions.create(
model=self.vision_model,
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": (
"Describe this image in 2-3 sentences for indexing "
"in semantic search. Include the content type, "
"main elements, and specific visible data."
),
},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
],
}],
max_tokens=150,
)
description = response.choices[0].message.content
self.cache.set(image_data, description, self.vision_model)
return descriptionStep 3: Indexer with ChromaDB
import chromadb
from chromadb.utils import embedding_functions
class MultimodalIndexer:
def __init__(
self,
persist_directory: str = "./chroma_db",
collection_name: str = "multimodal_rag",
embedding_model: str = "text-embedding-3-small",
):
self.collection_name = collection_name
self.ef = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv("OPENAI_API_KEY"),
model_name=embedding_model,
)
self.chroma_client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name, embedding_function=self.ef,
)
self.stats = {"indexed": 0, "errors": 0}
def index_chunks(self, chunks: list[dict], batch_size: int = 100) -> dict:
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
documents, ids, metadatas = [], [], []
for chunk in batch:
chunk_id = (
f"{Path(chunk.get('source', 'unknown')).stem}"
f"_{chunk['type']}_{chunk.get('chunk_index', 0)}"
)
documents.append(chunk["text"])
ids.append(chunk_id)
metadatas.append({
"type": chunk["type"],
"source": chunk.get("source", ""),
"page": chunk.get("page", 0),
"image_path": chunk.get("image_path", ""),
})
try:
self.collection.upsert(
documents=documents, ids=ids, metadatas=metadatas,
)
self.stats["indexed"] += len(batch)
except Exception as e:
print(f"Error indexing batch {i}: {e}")
self.stats["errors"] += 1
return self.stats
def get_info(self) -> dict:
count = self.collection.count()
if count == 0:
return {"total": 0, "by_type": {}}
sample = self.collection.get(limit=min(count, 1000), include=["metadatas"])
type_counts = {}
for meta in sample["metadatas"]:
t = meta.get("type", "unknown")
type_counts[t] = type_counts.get(t, 0) + 1
return {"total": count, "by_type": type_counts}
def clear(self) -> None:
self.chroma_client.delete_collection(self.collection_name)
self.collection = self.chroma_client.get_or_create_collection(
name=self.collection_name, embedding_function=self.ef,
)Step 4: Hybrid Retriever
Searches relevant chunks combining semantic similarity with metadata filters.
class HybridRetriever:
def __init__(self, indexer: MultimodalIndexer, default_k: int = 5):
self.collection = indexer.collection
self.default_k = default_k
def retrieve(
self, query: str, k: int | None = None,
content_type: str | None = None,
) -> list[dict]:
k = k or self.default_k
params = {
"query_texts": [query], "n_results": k,
"include": ["documents", "metadatas", "distances"],
}
if content_type:
params["where"] = {"type": {"$eq": content_type}}
results = self.collection.query(**params)
retrieved = []
for i in range(len(results["ids"][0])):
similarity = round(1 - results["distances"][0][i], 4)
retrieved.append({
"id": results["ids"][0][i],
"text": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"similarity": similarity,
})
return retrieved
def retrieve_hybrid(
self, query: str, text_k: int = 4, image_k: int = 2,
) -> list[dict]:
text_results = self.retrieve(query, k=text_k, content_type="text")
image_results = self.retrieve(query, k=image_k, content_type="image")
combined = text_results + image_results
combined.sort(key=lambda x: x["similarity"], reverse=True)
seen = set()
return [r for r in combined if r["id"] not in seen and not seen.add(r["id"])]
def retrieve_smart(self, query: str, k: int | None = None) -> list[dict]:
visual_keywords = {
"diagram", "image", "figure", "chart", "table",
"screenshot", "photo", "shows", "visualize",
}
if set(query.lower().split()) & visual_keywords:
return self.retrieve_hybrid(query, text_k=2, image_k=4)
return self.retrieve_hybrid(query, text_k=k or self.default_k, image_k=1)Step 5: Answer Generator
class AnswerGenerator:
def __init__(self, model: str = "gpt-4o", max_tokens: int = 800):
self.client = OpenAI()
self.model = model
self.max_tokens = max_tokens
self.stats = {"queries": 0, "total_tokens": 0}
def generate(self, query: str, chunks: list[dict]) -> dict:
if not chunks:
return {
"answer": "I couldn't find relevant information to answer.",
"sources": [], "tokens": {},
}
context = self._format_context(chunks)
sources = self._extract_sources(chunks)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": (
"Answer based EXCLUSIVELY on the provided context.\n"
"The context includes text and image descriptions from documents.\n"
"If an image contains relevant information, mention it.\n"
"If there isn't enough information, say so clearly.\n"
"At the end, list the sources: [Source: document, page X, type]."
),
},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
],
max_tokens=self.max_tokens,
temperature=0,
)
usage = response.usage
self.stats["queries"] += 1
self.stats["total_tokens"] += usage.total_tokens if usage else 0
return {
"answer": response.choices[0].message.content,
"sources": sources,
"tokens": {
"prompt": usage.prompt_tokens if usage else 0,
"completion": usage.completion_tokens if usage else 0,
"total": usage.total_tokens if usage else 0,
},
}
def _format_context(self, chunks: list[dict]) -> str:
parts = []
for i, chunk in enumerate(chunks):
meta = chunk.get("metadata", {})
header = (
f"[CHUNK {i+1} | {meta.get('type','text').upper()} | "
f"{Path(meta.get('source','')).name} p.{meta.get('page','?')} | "
f"sim: {chunk.get('similarity', 0)}]"
)
parts.append(f"{header}\n{chunk['text']}")
return "\n\n---\n\n".join(parts)
def _extract_sources(self, chunks: list[dict]) -> list[dict]:
return [{
"source": Path(c.get("metadata", {}).get("source", "")).name,
"page": c.get("metadata", {}).get("page", 0),
"type": c.get("metadata", {}).get("type", "text"),
"similarity": c.get("similarity", 0),
} for c in chunks]Step 6: Complete Pipeline
import time
class MultimodalRAGPipeline:
def __init__(
self,
persist_dir: str = "./chroma_db",
collection_name: str = "multimodal_rag",
image_dir: str = "./extracted_images",
cache_dir: str = "./description_cache",
chunk_words: int = 500,
llm_model: str = "gpt-4o",
vision_model: str = "gpt-4o-mini",
):
self.cache = DescriptionCache(cache_dir)
self.processor = DocumentProcessor(
cache=self.cache, output_dir=image_dir,
max_chunk_words=chunk_words, vision_model=vision_model,
)
self.indexer = MultimodalIndexer(
persist_directory=persist_dir, collection_name=collection_name,
)
self.retriever = HybridRetriever(self.indexer)
self.generator = AnswerGenerator(model=llm_model)
self.metrics = {
"docs_processed": 0, "total_chunks": 0,
"indexing_time": 0.0, "queries": 0, "avg_query_time": 0.0,
}
def index_document(self, pdf_path: str) -> dict:
start = time.time()
chunks = self.processor.process(pdf_path)
index_stats = self.indexer.index_chunks(chunks)
elapsed = time.time() - start
self.metrics["docs_processed"] += 1
self.metrics["total_chunks"] += len(chunks)
self.metrics["indexing_time"] += elapsed
return {
"document": pdf_path, "chunks": len(chunks),
"stats": index_stats, "time_seconds": round(elapsed, 2),
}
def index_directory(self, dir_path: str, glob: str = "*.pdf") -> dict:
results = []
for pdf_path in Path(dir_path).glob(glob):
result = self.index_document(str(pdf_path))
results.append(result)
print(f" Indexed: {pdf_path.name} ({result['chunks']} chunks, {result['time_seconds']}s)")
return {
"documents": len(results),
"total_chunks": sum(r["chunks"] for r in results),
"details": results,
}
def query(self, question: str, k: int = 5, smart: bool = True) -> dict:
start = time.time()
retrieved = (
self.retriever.retrieve_smart(question, k=k)
if smart
else self.retriever.retrieve(question, k=k)
)
result = self.generator.generate(question, retrieved)
elapsed = time.time() - start
self.metrics["queries"] += 1
total_time = self.metrics.get("_total_q_time", 0) + elapsed
self.metrics["_total_q_time"] = total_time
self.metrics["avg_query_time"] = round(total_time / self.metrics["queries"], 2)
return {
"question": question,
"answer": result["answer"],
"sources": result["sources"],
"retrieved_chunks": len(retrieved),
"tokens": result["tokens"],
"time_seconds": round(elapsed, 2),
}
def interactive(self) -> None:
print("=== Multimodal RAG — Interactive Session ===")
print(f"Collection: {self.indexer.get_info()}")
print("Type 'exit' to finish.\n")
while True:
question = input("Question: ").strip()
if question.lower() in ("exit", "quit"):
break
if not question:
continue
result = self.query(question)
print(f"\nAnswer:\n{result['answer']}\n")
for src in result["sources"]:
print(f" - {src['source']} p.{src['page']} ({src['type']})")
print(f"Time: {result['time_seconds']}s | Tokens: {result['tokens'].get('total', 0)}\n")
def get_metrics(self) -> dict:
return {
**{k: v for k, v in self.metrics.items() if not k.startswith("_")},
"collection": self.indexer.get_info(),
"cache": self.cache.stats(),
}Complete usage
pipeline = MultimodalRAGPipeline(chunk_words=500, llm_model="gpt-4o")
# --- Index ---
# result = pipeline.index_document("docs/technical_manual.pdf")
# print(f"Indexed: {result['chunks']} chunks in {result['time_seconds']}s")
# --- Query ---
# answer = pipeline.query("How do the microservices connect?")
# print(answer["answer"])
# --- Interactive session ---
# pipeline.interactive()
# --- Metrics ---
# print(pipeline.get_metrics())Extension 1: Result Reranking
The cosine-similarity retriever returns reasonable results, but a reranker improves precision by reordering the candidates with a second model.
class Reranker:
def __init__(self, model: str = "gpt-4o-mini"):
self.client = OpenAI()
self.model = model
def rerank(self, query: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
if len(chunks) <= top_k:
return chunks
candidates = "\n".join(
f"[{i}] {c['text'][:200].replace(chr(10), ' ')}"
for i, c in enumerate(chunks)
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": (
"Given a question and document chunks, return the "
"indices of the MOST relevant chunks from highest to lowest. "
"Respond ONLY with indices separated by commas. Example: 2,0,4"
),
},
{
"role": "user",
"content": (
f"Question: {query}\n\nChunks:\n{candidates}\n\n"
f"Top {top_k} (indices only):"
),
},
],
max_tokens=50, temperature=0,
)
try:
indices = [int(x.strip()) for x in response.choices[0].message.content.strip().split(",")]
indices = [i for i in indices if 0 <= i < len(chunks)][:top_k]
except ValueError:
return chunks[:top_k]
return [chunks[i] for i in indices]
def query_with_reranking(pipeline, question: str, initial_k: int = 10, final_k: int = 3) -> dict:
retrieved = pipeline.retriever.retrieve_smart(question, k=initial_k)
reranked = Reranker().rerank(question, retrieved, top_k=final_k)
result = pipeline.generator.generate(question, reranked)
return {
"question": question, "answer": result["answer"],
"initial_candidates": len(retrieved), "after_reranking": len(reranked),
"sources": result["sources"],
}Extension 2: Direct Image Q&A
When the retriever finds a relevant image and the original file is available, you can send it directly to the LLM for deeper analysis.
class ImageQA:
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def answer_with_image(self, question: str, image_path: str, text_context: str = "") -> str:
image_data = Path(image_path).read_bytes()
ext = Path(image_path).suffix.lstrip(".")
mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
b64 = base64.b64encode(image_data).decode("utf-8")
user_text = (
f"Context:\n{text_context}\n\nQuestion: {question}"
if text_context
else f"Question about this image: {question}"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "Answer based on the image and the context. Describe relevant elements.",
},
{
"role": "user",
"content": [
{"type": "text", "text": user_text},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
],
},
],
max_tokens=500,
)
return response.choices[0].message.content
def query_with_image_qa(pipeline, question: str, k: int = 5) -> dict:
retrieved = pipeline.retriever.retrieve_smart(question, k=k)
image_chunks = [
r for r in retrieved
if r["metadata"].get("type") == "image"
and r["metadata"].get("image_path")
and Path(r["metadata"]["image_path"]).exists()
]
text_chunks = [r for r in retrieved if r["metadata"].get("type") == "text"]
text_context = "\n\n".join(c["text"] for c in text_chunks[:3])
base_result = pipeline.generator.generate(question, retrieved)
image_answers = []
qa = ImageQA()
for img_chunk in image_chunks[:2]:
answer = qa.answer_with_image(question, img_chunk["metadata"]["image_path"], text_context)
image_answers.append({
"image": img_chunk["metadata"]["image_path"],
"page": img_chunk["metadata"]["page"],
"answer": answer,
})
return {
"question": question,
"text_answer": base_result["answer"],
"image_answers": image_answers,
"sources": base_result["sources"],
}Troubleshooting
ChromaDB "Collection already exists" when re-running
def reset_pipeline(pipeline: MultimodalRAGPipeline) -> None:
pipeline.indexer.clear()
print("Collection cleared. You can re-index.")Images not extracted from certain PDFs
Some PDFs have images as SVG vectors or as part of the rendering. PyMuPDF only extracts raster. As an alternative, render the whole page as an image.
def extract_page_as_image_fallback(pdf_path: str, page_num: int, dpi: int = 150) -> bytes:
doc = fitz.open(pdf_path)
page = doc[page_num]
pix = page.get_pixmap(matrix=fitz.Matrix(dpi / 72, dpi / 72))
image_bytes = pix.tobytes("png")
doc.close()
return image_bytesThe LLM hallucinates despite the context
Verify that the retrieved chunks are relevant (a retrieval problem, not a generation one). Lower the temperature to 0 and use a stricter prompt. If it persists, increase k to provide more context.
Slow queries (>10 seconds)
1. Use gpt-4o-mini instead of gpt-4o (3-5x faster)
2. Reduce the generator's max_tokens
3. Reduce k (fewer chunks in context)
4. Pre-compute answers for frequent queries
Generic image descriptions
Use gpt-4o instead of gpt-4o-mini for the processor's vision_model. It's more expensive but produces more detailed descriptions.
Memory error with many documents
ChromaDB PersistentClient keeps indexes in memory. For >100K chunks, use a dedicated vector store (Pinecone, Weaviate). For the scope of this project, ChromaDB is enough for hundreds of documents.
Checklist
-
DescriptionCacheimplemented and persisting to disk -
DocumentProcessorextracts text and images from PDFs - Irrelevant images are filtered out before describing
- Relevant images are described with the Vision API (with cache)
-
MultimodalIndexerstores chunks in ChromaDB with metadata -
HybridRetrieversearches by text, image, or both - Smart retrieval detects visual queries
-
AnswerGeneratorproduces answers with cited sources -
MultimodalRAGPipelineorchestrates indexing and querying - Pipeline supports multiple documents
- Cost, time and cache metrics available
- Reranking extension functional
- Image Q&A extension functional
Exercises
Exercise 1: Pipeline with a quality report
Implement a function that, given a set of test questions with expected answers, runs the pipeline and generates a report measuring whether the answers contain the expected key points.
See solution
def evaluate_pipeline(pipeline: MultimodalRAGPipeline, test_cases: list[dict]) -> dict:
results = []
for case in test_cases:
question = case["question"]
expected = case.get("expected_keywords", [])
response = pipeline.query(question)
answer_lower = response["answer"].lower()
found = [kw for kw in expected if kw.lower() in answer_lower]
missed = [kw for kw in expected if kw.lower() not in answer_lower]
coverage = len(found) / max(len(expected), 1)
results.append({
"question": question,
"coverage": round(coverage, 2),
"found": found, "missed": missed,
"sources": len(response["sources"]),
"time": response["time_seconds"],
})
avg_coverage = sum(r["coverage"] for r in results) / max(len(results), 1)
return {
"cases": len(test_cases),
"avg_coverage": round(avg_coverage, 2),
"details": results,
}
test_cases = [
{"question": "What protocol does the API Gateway use?", "expected_keywords": ["http", "rest", "gateway"]},
{"question": "What database does the payment service use?", "expected_keywords": ["postgresql", "payments"]},
]
# report = evaluate_pipeline(pipeline, test_cases)
# print(f"Average coverage: {report['avg_coverage']}")Exercise 2: Multi-document RAG with comparison
Implement a function that indexes multiple documents and answers comparative questions across them, citing which document each piece of data comes from.
See solution
def comparative_query(
pipeline: MultimodalRAGPipeline,
question: str,
doc_names: list[str],
k_per_doc: int = 3,
) -> dict:
all_chunks = []
for doc_name in doc_names:
results = pipeline.retriever.retrieve(question, k=k_per_doc, content_type=None)
matching = [r for r in results if doc_name in r.get("metadata", {}).get("source", "")]
for r in matching:
r["from_doc"] = doc_name
all_chunks.extend(matching)
all_chunks.sort(key=lambda x: x["similarity"], reverse=True)
top = all_chunks[:k_per_doc * 2]
context = "\n\n---\n\n".join(
f"[{c.get('from_doc', '?')} | p.{c['metadata']['page']}] {c['text'][:300]}"
for c in top
)
response = OpenAI().chat.completions.create(
model=pipeline.generator.model,
messages=[
{
"role": "system",
"content": (
"Answer by comparing information from DIFFERENT documents. "
"For each point, indicate which document it comes from."
),
},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
],
max_tokens=800, temperature=0,
)
return {
"question": question,
"answer": response.choices[0].message.content,
"documents_compared": list(set(c.get("from_doc", "?") for c in top)),
}
# result = comparative_query(pipeline, "How does each system handle authentication?", ["manual_v1", "manual_v2"])
# print(result["answer"])Summary
- The project has four components: document processor, indexer, hybrid retriever, and answer generator.
- The processor extracts text and images from PDFs, filters out irrelevant images, and describes the relevant ones with the Vision API using a persistent cache.
- The indexer stores chunks with embeddings in ChromaDB, supporting upsert and batch processing.
- The hybrid retriever combines text and image search, with automatic detection of visual queries.
- The generator builds prompts with formatted context and returns answers with sources.
- The pipeline orchestrates everything: document indexing, queries with smart retrieval, and interactive sessions.
- Reranking improves precision by reordering candidates with a second model.
- Image Q&A allows analyzing original images directly when they're available.
Next module: Module 7 — Use Cases. You already have vision, documents, generation, audio and RAG; now you'll apply everything in real design patterns: Document Q&A, automated image analysis, video analysis, and a Use Case Selector that routes inputs to the right pipeline.
Additional Resources
- ChromaDB Documentation — API and configuration
- OpenAI Vision API — Using images with GPT-4o
- PyMuPDF (fitz) — PDF extraction
- RAG Best Practices — RAG patterns in production
- Embedding Models Comparison — Model benchmark