Module 7: Use Cases

2. Document Q&A

Next →

Description

Document Q&A is the most in-demand pattern in multimodal AI: a user uploads a document (PDF, document image, Word) and asks questions about its content. The system extracts text and images, indexes the content, retrieves the relevant fragments, and generates an answer citing the sources.

Why it matters: This pattern is the foundation of legal assistants, technical support systems, educational platforms, and compliance tools. It's the use case that most justifies investments in multimodal AI because it solves a real problem: "I have 200 pages of documentation and I need an answer in 5 seconds".

Connection with the module: Document Q&A combines Module 3 (document extraction), Module 6 (RAG with embeddings), and Module 1-2 (generation with LLM). Capsule 08 (Use Case Selector) will use this pipeline as one of its routing destinations.

Visual Pipeline

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Document    │────▶│   Extract    │────▶│   Chunking   │
│  (PDF/img)   │     │  text+imgs   │     │              │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Answer     │◀────│     LLM      │◀────│  Retrieval   │
│  + sources   │     │  (generate)  │     │  (search)    │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  ▲
                                                  │
                                           ┌──────────────┐
                                           │  User's      │
                                           │  question    │
                                           └──────────────┘

Pipeline stages:

  1. Load document — Open PDF, scanned image, or text
  2. Extract content — Text with PyMuPDF, embedded images, tables
  3. Chunking — Split into manageable-sized fragments
  4. Index — Create embeddings and store in a vector store
  5. Query — Receive the user's question
  6. Retrieval — Search for the most relevant chunks by similarity
  7. Generate — LLM produces an answer based on the retrieved chunks
  8. Cite sources — Include references to the chunks used

Step 1: Load and Extract Content

Text extraction

import fitz

def extract_text_from_pdf(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    pages = []
    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text()
        if text.strip():
            pages.append({
                "page": page_num + 1,
                "text": text.strip()
            })
    doc.close()
    return pages

Image extraction

import base64
from pathlib import Path

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

    for page_num in range(len(doc)):
        page = doc[page_num]
        image_list = page.get_images(full=True)

        for img_idx, img_info in enumerate(image_list):
            xref = img_info[0]
            base_image = doc.extract_image(xref)
            image_bytes = base_image["image"]
            ext = base_image["ext"]

            img_path = f"{output_dir}/page{page_num + 1}_img{img_idx + 1}.{ext}"
            with open(img_path, "wb") as f:
                f.write(image_bytes)

            images.append({
                "page": page_num + 1,
                "path": img_path,
                "size_bytes": len(image_bytes)
            })

    doc.close()
    return images

Combined extraction

def extract_document(pdf_path: str) -> dict:
    pages = extract_text_from_pdf(pdf_path)
    images = extract_images_from_pdf(pdf_path)
    return {
        "pages": pages,
        "images": images,
        "total_pages": len(pages),
        "total_images": len(images)
    }

Step 2: Smart Chunking

Chunking is critical. Chunks that are too big dilute relevance; chunks that are too small lose context.

Chunking by size with overlap

def chunk_text(text: str, chunk_size: int = 1500, overlap: int = 200) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = end - overlap
    return chunks

Chunking by pages with metadata

def chunk_pages(pages: list[dict], chunk_size: int = 1500, overlap: int = 200) -> list[dict]:
    chunks = []
    for page in pages:
        page_chunks = chunk_text(page["text"], chunk_size, overlap)
        for i, chunk in enumerate(page_chunks):
            chunks.append({
                "text": chunk,
                "page": page["page"],
                "chunk_index": i,
                "source": f"Page {page['page']}, fragment {i + 1}"
            })
    return chunks

Semantic chunking (by paragraphs)

def chunk_by_paragraphs(text: str, max_chunk_size: int = 1500) -> list[str]:
    paragraphs = text.split("\n\n")
    chunks = []
    current_chunk = ""

    for para in paragraphs:
        if len(current_chunk) + len(para) + 2 <= max_chunk_size:
            current_chunk += para + "\n\n"
        else:
            if current_chunk.strip():
                chunks.append(current_chunk.strip())
            current_chunk = para + "\n\n"

    if current_chunk.strip():
        chunks.append(current_chunk.strip())

    return chunks

Step 3: Index with Embeddings

With ChromaDB

import chromadb
from openai import OpenAI

client = OpenAI()

def create_document_index(chunks: list[dict], collection_name: str = "document_qa") -> chromadb.Collection:
    chroma_client = chromadb.Client()

    try:
        chroma_client.delete_collection(collection_name)
    except Exception:
        pass

    collection = chroma_client.create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"}
    )

    texts = [c["text"] for c in chunks]
    embeddings = get_embeddings_batch(texts)

    collection.add(
        documents=texts,
        embeddings=embeddings,
        metadatas=[{"page": c["page"], "source": c["source"]} for c in chunks],
        ids=[f"chunk_{i}" for i in range(len(chunks))]
    )

    return collection


def get_embeddings_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(
        model=model,
        input=texts
    )
    return [item.embedding for item in response.data]

Index image descriptions (multimodal RAG)

def index_document_images(images: list[dict], collection: chromadb.Collection, start_id: int = 10000) -> None:
    for i, img in enumerate(images):
        description = describe_image_for_indexing(img["path"])

        embedding = get_embeddings_batch([description])[0]

        collection.add(
            documents=[description],
            embeddings=[embedding],
            metadatas=[{
                "page": img["page"],
                "source": f"Image on page {img['page']}",
                "type": "image"
            }],
            ids=[f"img_{start_id + i}"]
        )


def describe_image_for_indexing(image_path: str) -> str:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this image in detail for indexing. Include all visible data, text, and structure."},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        max_tokens=300
    )
    return response.choices[0].message.content

Step 4: Retrieval

def retrieve_relevant_chunks(
    collection: chromadb.Collection,
    question: str,
    n_results: int = 5
) -> list[dict]:
    query_embedding = get_embeddings_batch([question])[0]

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
        include=["documents", "metadatas", "distances"]
    )

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

    return chunks

Step 5: Generate an Answer with Sources

def generate_answer(question: str, chunks: list[dict]) -> dict:
    context_parts = []
    for i, chunk in enumerate(chunks):
        source = chunk["metadata"].get("source", f"Fragment {i + 1}")
        context_parts.append(f"[{source}]\n{chunk['text']}")

    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "You are a document Q&A assistant. Answer ONLY based on the provided context. If you can't find the answer, say so explicitly. Cite sources in brackets [Page X, fragment Y]."
            },
            {
                "role": "user",
                "content": f"Context:\n\n{context}\n\n---\n\nQuestion: {question}"
            }
        ],
        max_tokens=500,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [c["metadata"] for c in chunks],
        "model": "gpt-4o-mini",
        "chunks_used": len(chunks)
    }

Complete Pipeline

def document_qa_pipeline(pdf_path: str, question: str) -> dict:
    doc_content = extract_document(pdf_path)

    chunks = chunk_pages(doc_content["pages"])

    collection = create_document_index(chunks)

    if doc_content["images"]:
        index_document_images(doc_content["images"], collection)

    relevant = retrieve_relevant_chunks(collection, question)

    answer = generate_answer(question, relevant)
    answer["document"] = pdf_path
    answer["total_pages"] = doc_content["total_pages"]

    return answer

Usage:

result = document_qa_pipeline(
    "service_contract.pdf",
    "What is the penalty clause for breach of contract?"
)
print(result["answer"])
print(f"Sources: {result['sources']}")

Multi-Turn Conversation

A useful Q&A system keeps context between questions. The user asks "What is the penalty clause?" and then "And how much is the amount?", and the system understands that "the amount" refers to the penalty.

class DocumentQASession:
    def __init__(self, pdf_path: str):
        self.pdf_path = pdf_path
        self.history: list[dict] = []
        self.collection = None
        self._setup()

    def _setup(self):
        doc_content = extract_document(self.pdf_path)
        chunks = chunk_pages(doc_content["pages"])
        self.collection = create_document_index(chunks)
        if doc_content["images"]:
            index_document_images(doc_content["images"], self.collection)

    def ask(self, question: str) -> dict:
        contextualized_question = self._contextualize(question)

        relevant = retrieve_relevant_chunks(self.collection, contextualized_question)

        context_parts = []
        for i, chunk in enumerate(relevant):
            source = chunk["metadata"].get("source", f"Fragment {i + 1}")
            context_parts.append(f"[{source}]\n{chunk['text']}")
        context = "\n\n---\n\n".join(context_parts)

        messages = [
            {
                "role": "system",
                "content": "You are a document Q&A assistant. Answer ONLY based on the provided context. Cite sources in brackets. Stay consistent with the previous conversation."
            }
        ]

        for h in self.history[-6:]:
            messages.append({"role": "user", "content": h["question"]})
            messages.append({"role": "assistant", "content": h["answer"]})

        messages.append({
            "role": "user",
            "content": f"Document context:\n\n{context}\n\n---\n\nQuestion: {question}"
        })

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            max_tokens=500,
            temperature=0.1
        )

        answer = response.choices[0].message.content

        self.history.append({
            "question": question,
            "answer": answer,
            "sources": [c["metadata"] for c in relevant]
        })

        return {
            "answer": answer,
            "sources": [c["metadata"] for c in relevant],
            "turn": len(self.history)
        }

    def _contextualize(self, question: str) -> str:
        if not self.history:
            return question

        recent = self.history[-3:]
        history_text = "\n".join(
            f"Q: {h['question']}\nA: {h['answer'][:200]}"
            for h in recent
        )

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"History:\n{history_text}\n\nNew question: {question}\n\nRewrite the question so it's self-contained (no ambiguous pronouns). If it's already clear, repeat it as is."
            }],
            max_tokens=100,
            temperature=0
        )
        return response.choices[0].message.content

Usage:

session = DocumentQASession("technical_manual.pdf")

r1 = session.ask("What hardware requirements are mentioned?")
print(r1["answer"])

r2 = session.ask("And the software ones?")
print(r2["answer"])

r3 = session.ask("Are they compatible with each other?")
print(r3["answer"])

Pipeline Variations

With reranking

After retrieval, reorder by relevance using a second model:

def rerank_chunks(question: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
    prompt_parts = []
    for i, chunk in enumerate(chunks):
        prompt_parts.append(f"[{i}] {chunk['text'][:300]}")

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Question: {question}\n\nFragments:\n" + "\n\n".join(prompt_parts) +
                       f"\n\nOrder the indices from most to least relevant. Only the numbers separated by commas."
        }],
        max_tokens=50,
        temperature=0
    )

    try:
        indices = [int(x.strip()) for x in response.choices[0].message.content.split(",")]
        return [chunks[i] for i in indices[:top_k] if i < len(chunks)]
    except (ValueError, IndexError):
        return chunks[:top_k]

With structured answer

def generate_structured_answer(question: str, chunks: list[dict]) -> dict:
    context = "\n\n".join(c["text"] for c in chunks)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Answer in JSON format with: answer (string), confidence (low/medium/high), key_points (array of strings), sources_used (array of strings)."
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}"
            }
        ],
        max_tokens=500,
        temperature=0,
        response_format={"type": "json_object"}
    )

    import json
    return json.loads(response.choices[0].message.content)

Troubleshooting

Problem 1: Made-up answers (hallucinations)

Symptom: The model answers with information that isn't in the document.

Cause: Insufficient context or a prompt that doesn't restrict the model.

Solution:

system_prompt = (
    "Answer EXCLUSIVELY with information from the provided context. "
    "If the answer is not in the context, reply: "
    "'I didn't find information about this in the document.' "
    "NEVER make up data."
)

Problem 2: Poor retrieval (irrelevant chunks)

Symptom: The retrieved chunks don't contain the answer even though the document does have it.

Cause: Chunks that are too big, insufficient overlap, or embeddings that don't capture the semantics.

Solution:

  • Reduce chunk_size to 800-1000 characters
  • Increase overlap to 200-300
  • Use text-embedding-3-large instead of small
  • Implement reranking (see previous variation)

Problem 3: Scanned PDFs with no text

Symptom: extract_text_from_pdf returns empty pages.

Cause: The PDF is a scanned image, it has no embedded text.

Solution:

def extract_with_ocr_fallback(pdf_path: str) -> list[dict]:
    pages = extract_text_from_pdf(pdf_path)

    empty_pages = [p for p in pages if len(p["text"].strip()) < 50]
    if len(empty_pages) > len(pages) * 0.5:
        return extract_via_vision(pdf_path)

    return pages


def extract_via_vision(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    pages = []
    for page_num in range(len(doc)):
        page = doc[page_num]
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")
        b64 = base64.b64encode(img_bytes).decode()

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": "Extract all the visible text on this page. Keep the structure."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
                ]
            }],
            max_tokens=2000
        )
        pages.append({
            "page": page_num + 1,
            "text": response.choices[0].message.content
        })
    doc.close()
    return pages

Problem 4: Very long documents (100+ pages)

Symptom: Indexing takes a long time or exceeds the embeddings API limits.

Solution:

def index_large_document(chunks: list[dict], batch_size: int = 100) -> chromadb.Collection:
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("large_doc")

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        texts = [c["text"] for c in batch]
        embeddings = get_embeddings_batch(texts)

        collection.add(
            documents=texts,
            embeddings=embeddings,
            metadatas=[{"page": c["page"], "source": c["source"]} for c in batch],
            ids=[f"chunk_{i + j}" for j in range(len(batch))]
        )

    return collection

Problem 5: Context exceeds the LLM's token limit

Symptom: Error maximum context length exceeded.

Solution:

def trim_context_to_limit(chunks: list[dict], max_chars: int = 12000) -> list[dict]:
    trimmed = []
    total = 0
    for chunk in chunks:
        if total + len(chunk["text"]) > max_chars:
            break
        trimmed.append(chunk)
        total += len(chunk["text"])
    return trimmed

Exercises

Exercise 1: Document Q&A with confidence score

Modify generate_answer so it includes a confidence score (low/medium/high) based on the distance of the retrieved chunks.

Hint: If the average distance is < 0.3, high confidence; < 0.5, medium; > 0.5, low.

See solution 
def generate_answer_with_confidence(question: str, chunks: list[dict]) -> dict:
    avg_distance = sum(c["distance"] for c in chunks) / len(chunks) if chunks else 1.0

    if avg_distance < 0.3:
        confidence = "high"
    elif avg_distance < 0.5:
        confidence = "medium"
    else:
        confidence = "low"

    context_parts = []
    for i, chunk in enumerate(chunks):
        source = chunk["metadata"].get("source", f"Fragment {i + 1}")
        context_parts.append(f"[{source}]\n{chunk['text']}")
    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Answer ONLY based on the context. Cite sources. If the information is insufficient, say so."
            },
            {
                "role": "user",
                "content": f"Context:\n\n{context}\n\n---\n\nQuestion: {question}"
            }
        ],
        max_tokens=500,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "confidence": confidence,
        "avg_distance": round(avg_distance, 3),
        "sources": [c["metadata"] for c in chunks]
    }

Exercise 2: Retrieval with page filter

Implement a function that lets the user restrict the search to specific pages of the document.

Hint: Use ChromaDB's where parameter to filter by metadata.

See solution 
def retrieve_from_pages(
    collection: chromadb.Collection,
    question: str,
    pages: list[int] = None,
    n_results: int = 5
) -> list[dict]:
    query_embedding = get_embeddings_batch([question])[0]

    query_params = {
        "query_embeddings": [query_embedding],
        "n_results": n_results,
        "include": ["documents", "metadatas", "distances"]
    }

    if pages:
        query_params["where"] = {"page": {"$in": pages}}

    results = collection.query(**query_params)

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

    return chunks

Exercise 3: Compare two documents

Create a pipeline that takes two PDFs and a question, indexes both, and generates a comparative answer.

Hint: Use prefixes in the IDs and metadata to distinguish which document each chunk comes from.

See solution 
def compare_documents_qa(pdf_path_a: str, pdf_path_b: str, question: str) -> dict:
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("compare_docs")

    for label, pdf_path in [("Document A", pdf_path_a), ("Document B", pdf_path_b)]:
        pages = extract_text_from_pdf(pdf_path)
        chunks = chunk_pages(pages)

        texts = [c["text"] for c in chunks]
        embeddings = get_embeddings_batch(texts)

        collection.add(
            documents=texts,
            embeddings=embeddings,
            metadatas=[{
                "page": c["page"],
                "source": f"{label}, {c['source']}",
                "document": label
            } for c in chunks],
            ids=[f"{label.lower().replace(' ', '_')}_{i}" for i in range(len(chunks))]
        )

    relevant = retrieve_relevant_chunks(collection, question, n_results=8)

    context_parts = []
    for chunk in relevant:
        source = chunk["metadata"]["source"]
        context_parts.append(f"[{source}]\n{chunk['text']}")
    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Compare the information from Document A and Document B to answer. Cite which document says what."
            },
            {
                "role": "user",
                "content": f"Context:\n\n{context}\n\n---\n\nComparative question: {question}"
            }
        ],
        max_tokens=600,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [c["metadata"] for c in relevant]
    }

Additional Resources

  1. RAG Guide — LangChain — Reference implementation
  2. ChromaDB Documentation — Vector store used in the examples
  3. OpenAI Embeddings — Embedding models
  4. PyMuPDF Documentation — PDF extraction
← Previous1. Introduction: Use CasesNext →3. Automated Image Analysis