Capstone Project: Building a Production Technical PDF Chatbot

In this hands-on project, you will build an end-to-end Technical Documentation RAG System that ingests complex technical PDFs (such as PySpark architectural whitepapers or Snowflake user manuals), indexes them into a vector database, and answers user questions with strict factual grounding and page-level source citations.

1. Project Architecture

flowchart TD
    subgraph IngestionPipeline["1. Ingestion Pipeline"]
        PDF["Technical PDF Document<br/>(e.g., 'Spark_Tuning_Guide.pdf')"] --> Reader["PDF Text & Metadata Extractor<br/>(pypdf / pymupdf)"]
        Reader --> Chunker["Recursive Chunk Splitter<br/>(Size: 600 chars, Overlap: 100 chars)"]
        Chunker --> MetaEnrich["Metadata Enrichment<br/>(page_number, document_title, chunk_index)"]
        MetaEnrich --> Embedder["Embedding Generator<br/>(ChromaDB Default / MiniLM)"]
        Embedder --> VectorDB[("ChromaDB Vector Store")]
    end

    subgraph QueryPipeline["2. Query & Answer Generation Pipeline"]
        UserQ["User Query:<br/>'What is the recommended partition size in Spark?'"] --> Search["Similarity Search (Top 3 Chunks)"]
        VectorDB --> Search
        Search --> Context["Formatted Grounded Context<br/>with exact page numbers"]
        Context --> LLM["LLM (Temperature = 0.0)"]
        LLM --> Out["Grounded Answer with Page Citations:<br/>'According to Page 14, target 100MB-200MB...'"]
    end

    style IngestionPipeline fill:#1e293b,color:#fff
    style QueryPipeline fill:#064e3b,color:#fff
    style VectorDB fill:#d97706,color:#fff
    style Out fill:#059669,color:#fff

2. Step-by-Step Implementation

Step 1: Ingesting and Chunking PDF Documents with Page Tracking

When chunking PDFs, you must retain the source page number for every single chunk to generate verifiable citations:

import os
import chromadb
from typing import List, Dict

# Simulated multi-page PDF document parser
MOCK_PDF_PAGES = [
    {
        "page_number": 1,
        "title": "Spark Core Memory Architecture",
        "text": """
        Apache Spark organizes executor memory into three distinct pools: Execution Memory, 
        Storage Memory, and User Memory. Execution memory is used for shuffles, joins, sorts, 
        and aggregations. Storage memory is used for caching DataFrames and broadcasting variables.
        Spark dynamically borrows memory between execution and storage pools.
        """
    },
    {
        "page_number": 2,
        "title": "Shuffle Partition Configuration",
        "text": """
        The default value for spark.sql.shuffle.partitions is 200. For large production workloads 
        processing hundreds of gigabytes, 200 is often too small, leading to data skew and disk spilling.
        Target 100MB to 200MB per shuffle partition on disk. If your shuffle stage writes 40GB, 
        configure approximately 200 to 400 shuffle partitions.
        """
    },
    {
        "page_number": 3,
        "title": "Dynamic Partition Pruning (DPP)",
        "text": """
        Dynamic Partition Pruning (DPP) was introduced in Spark 3.0. DPP optimizes star-schema queries 
        by using the results of dimension table filtering to prune partition scanning on large fact tables 
        at runtime. It is enabled by default via spark.sql.optimizer.dynamicPartitionPruning.enabled.
        """
    }
]

def chunk_pdf_document(pages: List[Dict], chunk_size: int = 250, chunk_overlap: int = 40) -> List[Dict]:
    """
    Splits multi-page text into chunks while preserving page metadata.
    """
    chunks = []
    chunk_counter = 0

    for page in pages:
        p_num = page["page_number"]
        p_title = page["title"]
        text = page["text"].strip()

        stride = chunk_size - chunk_overlap
        for i in range(0, len(text), stride):
            c_text = text[i : i + chunk_size].strip()
            if len(c_text) > 30: # Ignore tiny trailing fragments
                chunk_counter += 1
                chunks.append({
                    "id": f"chunk_p{p_num}_{chunk_counter}",
                    "text": c_text,
                    "metadata": {
                        "page_number": p_num,
                        "section_title": p_title,
                        "char_length": len(c_text)
                    }
                })
            if i + chunk_size >= len(text):
                break

    return chunks

Step 2: Indexing Chunks into ChromaDB

class PDFKnowledgeIndex:
    def __init__(self, collection_name: str = "spark_tuning_manual"):
        self.client = chromadb.Client() # In-memory client for demo
        self.collection = self.client.get_or_create_collection(name=collection_name)

    def index_document(self, chunks: List[Dict]):
        self.collection.add(
            ids=[c["id"] for c in chunks],
            documents=[c["text"] for c in chunks],
            metadatas=[c["metadata"] for c in chunks]
        )
        print(f"Successfully indexed {len(chunks)} chunks into ChromaDB.")

    def search(self, query: str, top_k: int = 2):
        return self.collection.query(
            query_texts=[query],
            n_results=top_k
        )

Step 3: Grounded RAG Prompt Assembly with Page Citations

def generate_grounded_rag_response(index: PDFKnowledgeIndex, user_question: str) -> str:
    # 1. Retrieve top matching chunks
    search_results = index.search(user_question, top_k=2)

    docs = search_results["documents"][0]
    metas = search_results["metadatas"][0]

    # 2. Format context with verifiable citation anchors
    context_blocks = []
    for doc, meta in zip(docs, metas):
        context_blocks.append(
            f"--- [SOURCE: Page {meta['page_number']} | Section: '{meta['section_title']}'] ---
{doc}"
        )
    formatted_context = "

".join(context_blocks)

    # 3. Assemble strict grounded system prompt
    prompt = f"""
You are an expert Technical Documentation AI Assistant.
Answer the user's question strictly using the verified reference context below.

CITATION RULES:

1. Every factual statement in your answer MUST cite its source using the format: [Page X, 'Section Title'].
2. If the answer cannot be found in the context, reply: 'I cannot answer this based on the provided document.'
3. Do not extrapolate or introduce ungrounded knowledge.

### VERIFIED REFERENCE CONTEXT ###
{formatted_context}

### USER QUESTION ###
{user_question}

### GROUNDED ANSWER WITH CITATIONS ###
""".strip()
    return prompt

Step 4: Verification & Execution

if __name__ == "__main__":
    # 1. Chunk document
    doc_chunks = chunk_pdf_document(MOCK_PDF_PAGES)

    # 2. Build index
    kb_index = PDFKnowledgeIndex()
    kb_index.index_document(doc_chunks)

    # 3. Test queries
    queries = [
        "What is the recommended size for a shuffle partition?",
        "How does Dynamic Partition Pruning optimize star schema queries?"
    ]

    for q in queries:
        print(f"
==================================================")
        print(f"QUERY: {q}")
        print(f"==================================================")
        prompt_payload = generate_grounded_rag_response(kb_index, q)
        print(prompt_payload)

3. Production Deployment Considerations

When moving this pipeline to AWS Lambda or production containers:

1. Replace pypdf with PyMuPDF (fitz): PyMuPDF is 10x faster and preserves multi-column layouts cleanly.
2. Handle Scanned PDFs with OCR: For scanned documents, integrate AWS Textract or Tesseract OCR before chunking.
3. Persist ChromaDB to Amazon S3 / EFS: Mount an AWS EFS volume to your Lambda or container to persist vector indices across cold starts.