Building Production-Ready AI Applications (Part 3): Building RAG Systems with ChromaDB

Welcome back! In Part 1, we set up our local LLM environment, and in Part 2, we mastered prompt engineering. Now comes the exciting part: building a RAG (Retrieval-Augmented Generation) system — the technology that powers AI applications like ChatGPT’s document analysis, Notion AI, and countless other tools.

By the end of this post, your DocuChat application will be able to answer questions about PDF documents with accurate, cited responses.

What You’ll Learn

• What RAG is and why it’s crucial for AI applications
• How to process and chunk PDF documents effectively
• Creating embeddings and semantic search
• Setting up ChromaDB for vector storage
• Combining retrieval with generation for accurate answers
• Building the complete DocuChat backend

Quick Recap: The Journey So Far

Part 1: Set up Ollama, created a basic FastAPI endpoint Part 2: Mastered prompting techniques, built streaming responses Part 3 (Today): Add document processing and RAG capabilities

After today, you’ll have a working AI assistant that can read PDFs and answer questions about them!

What is RAG and Why Do We Need It?

The Problem with Basic LLMs

Imagine asking your local LLM: “What were the key findings in my company’s Q3 report?”

The model has no idea. It was trained on general internet data, not your specific documents. It will either:

• Make up an answer (hallucinate)
• Say it doesn’t know
• Give generic information about Q3 reports

The RAG Solution

RAG = Retrieval-Augmented Generation

It’s a simple but powerful pattern:

1. Retrieval: Find relevant information from your documents
2. Augmentation: Add that information to the prompt
3. Generation: Let the LLM answer based on the retrieved context

Example:

User: "What are the main points in this research paper?"

Without RAG:
LLM: "I don't have access to that paper."

With RAG:
1. Find relevant sections from the paper
2. Add them to the prompt: "Based on these sections: [excerpts]..."
3. LLM: "The main points are: 1) ... 2) ... 3) ..."

Why RAG is Better Than Fine-Tuning

You might wonder: “Why not just fine-tune the model on my documents?”

Fine-tuning:

• ❌ Expensive (requires GPU, time, expertise)
• ❌ Static (need to retrain for new documents)
• ❌ Can cause model to forget general knowledge
• ❌ Overkill for most use cases

RAG:

• ✅ Works with any model, no training needed
• ✅ Dynamic (add/remove documents anytime)
• ✅ Model keeps its general knowledge
• ✅ Fast and cheap to implement

The RAG Architecture

Here’s how our DocuChat system will work:

User Question
    ↓
1. Convert question to embedding (vector)
    ↓
2. Search vector database for similar chunks
    ↓
3. Retrieve top matching document chunks
    ↓
4. Combine chunks + question into prompt
    ↓
5. Send to LLM
    ↓
6. Get answer with citations
    ↓
Return to user

Understanding Vector Embeddings

Before we code, let’s understand the magic behind RAG: embeddings.

What Are Embeddings?

Embeddings convert text into numbers (vectors) that capture semantic meaning.

Example:

"The cat sat on the mat" → [0.2, -0.5, 0.8, ..., 0.3]  # 384 numbers
"A feline rested on the rug" → [0.3, -0.4, 0.7, ..., 0.4]  # Very similar!

Similar sentences have similar vectors, even with different words!

Why This Matters

Traditional keyword search:

Query: "How do I reset my password?"
Document: "Click forgot credentials to recover account access"
Match: ❌ (no word "password" or "reset")

Embedding-based search:

Query embedding: [0.1, 0.5, ...]
Document embedding: [0.2, 0.4, ...]
Similarity: ✅ 87% match!

It understands meaning, not just keywords!

Step 1: Install Required Packages

First, let’s add the new dependencies:

pip install chromadb pypdf sentence-transformers

Update your requirements.txt:

fastapi
uvicorn
ollama
requests
chromadb
pypdf
sentence-transformers

What each package does:

• chromadb: Vector database for storing embeddings
• pypdf: Extract text from PDF files
• sentence-transformers: Create embeddings from text

Step 2: Document Processing and Chunking

Create a new file document_processor.py:

from pypdf import PdfReader
from typing import List, Dict
import re

class DocumentProcessor:
    """Process PDF documents and split into chunks"""

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        """
        Initialize document processor

        Args:
            chunk_size: Target size of each chunk in characters
            chunk_overlap: Number of characters to overlap between chunks
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def extract_text_from_pdf(self, pdf_path: str) -> str:
        """Extract all text from a PDF file"""
        try:
            reader = PdfReader(pdf_path)
            text = ""

            for page_num, page in enumerate(reader.pages):
                page_text = page.extract_text()
                # Add page marker for citation purposes
                text += f"\n\n--- Page {page_num + 1} ---\n\n"
                text += page_text

            return text

        except Exception as e:
            raise Exception(f"Error reading PDF: {str(e)}")

    def clean_text(self, text: str) -> str:
        """Clean and normalize text"""
        # Remove excessive whitespace
        text = re.sub(r'\s+', ' ', text)

        # Remove special characters but keep punctuation
        text = re.sub(r'[^\w\s\.\,\!\?\-\:\;\(\)]', '', text)

        # Remove page markers temporarily for cleaning
        text = re.sub(r'--- Page \d+ ---', '', text)

        return text.strip()

    def chunk_text(self, text: str) -> List[Dict[str, any]]:
        """
        Split text into overlapping chunks

        Returns:
            List of chunks with metadata
        """
        # Clean the text first
        cleaned_text = self.clean_text(text)

        chunks = []
        start = 0
        chunk_id = 0

        while start < len(cleaned_text):
            # Get chunk
            end = start + self.chunk_size
            chunk = cleaned_text[start:end]

            # Try to break at sentence boundary
            if end < len(cleaned_text):
                # Look for sentence ending
                last_period = chunk.rfind('.')
                last_question = chunk.rfind('?')
                last_exclamation = chunk.rfind('!')

                break_point = max(last_period, last_question, last_exclamation)

                if break_point > self.chunk_size * 0.5:  # At least 50% through chunk
                    chunk = chunk[:break_point + 1]
                    end = start + break_point + 1

            # Store chunk with metadata
            chunks.append({
                'id': f'chunk_{chunk_id}',
                'text': chunk.strip(),
                'start_char': start,
                'end_char': end,
                'chunk_index': chunk_id
            })

            chunk_id += 1
            start = end - self.chunk_overlap  # Overlap for context

        return chunks

    def process_pdf(self, pdf_path: str) -> List[Dict[str, any]]:
        """
        Complete pipeline: PDF → chunks

        Args:
            pdf_path: Path to PDF file

        Returns:
            List of text chunks with metadata
        """
        print(f"Processing PDF: {pdf_path}")

        # Extract text
        raw_text = self.extract_text_from_pdf(pdf_path)
        print(f"Extracted {len(raw_text)} characters")

        # Chunk text
        chunks = self.chunk_text(raw_text)
        print(f"Created {len(chunks)} chunks")

        return chunks

# Test the processor
if __name__ == "__main__":
    processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)

    # Test with a sample PDF (you'll need to provide one)
    # chunks = processor.process_pdf("sample.pdf")
    # for i, chunk in enumerate(chunks[:3]):  # Show first 3 chunks
    #     print(f"\nChunk {i}:")
    #     print(chunk['text'][:200] + "...")

Understanding Chunking Strategy

Why chunk?

• LLMs have limited context windows
• Smaller chunks = more precise retrieval
• Better citations (can point to specific sections)

Our strategy:

• 500 characters per chunk — Roughly 1–2 paragraphs
• 50 character overlap — Ensures we don’t cut sentences awkwardly
• Break at sentence boundaries — More coherent chunks

Example:

Original: "AI is amazing. It can do many things. Machine learning is a subset."

Chunk 1: "AI is amazing. It can do many things."
Chunk 2: "many things. Machine learning is a subset."
         ↑ Overlap ensures context isn't lost

Step 3: Setting Up ChromaDB

Create vector_store.py:

import chromadb
from chromadb.config import Settings
from typing import List, Dict
from sentence_transformers import SentenceTransformer

class VectorStore:
    """Manage vector database for document chunks"""

    def __init__(self, collection_name: str = "documents", persist_directory: str = "./chroma_db"):
        """
        Initialize ChromaDB

        Args:
            collection_name: Name of the collection to store documents
            persist_directory: Where to save the database
        """
        # Initialize ChromaDB client
        self.client = chromadb.PersistentClient(path=persist_directory)

        # Initialize embedding model (free, runs locally)
        print("Loading embedding model...")
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        print("Embedding model loaded!")

        # Get or create collection
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"description": "Document chunks for RAG"}
        )

    def create_embeddings(self, texts: List[str]) -> List[List[float]]:
        """
        Convert texts to embeddings

        Args:
            texts: List of text strings

        Returns:
            List of embedding vectors
        """
        embeddings = self.embedding_model.encode(texts, show_progress_bar=True)
        return embeddings.tolist()

    def add_documents(self, chunks: List[Dict[str, any]], document_name: str):
        """
        Add document chunks to vector database

        Args:
            chunks: List of chunks from DocumentProcessor
            document_name: Name of the source document
        """
        print(f"Adding {len(chunks)} chunks to vector store...")

        # Prepare data
        texts = [chunk['text'] for chunk in chunks]
        ids = [f"{document_name}_{chunk['id']}" for chunk in chunks]
        metadatas = [
            {
                'document': document_name,
                'chunk_index': chunk['chunk_index'],
                'start_char': chunk['start_char'],
                'end_char': chunk['end_char']
            }
            for chunk in chunks
        ]

        # Create embeddings
        embeddings = self.create_embeddings(texts)

        # Add to database
        self.collection.add(
            embeddings=embeddings,
            documents=texts,
            ids=ids,
            metadatas=metadatas
        )

        print(f"✓ Added {len(chunks)} chunks from {document_name}")

    def search(self, query: str, n_results: int = 3) -> Dict:
        """
        Search for relevant chunks

        Args:
            query: User's question
            n_results: Number of results to return

        Returns:
            Dictionary with results and metadata
        """
        # Create embedding for query
        query_embedding = self.create_embeddings([query])[0]

        # Search
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            include=['documents', 'metadatas', 'distances']
        )

        return results

    def delete_document(self, document_name: str):
        """Delete all chunks from a specific document"""
        # Get all IDs for this document
        results = self.collection.get(
            where={"document": document_name}
        )

        if results['ids']:
            self.collection.delete(ids=results['ids'])
            print(f"✓ Deleted {len(results['ids'])} chunks from {document_name}")
        else:
            print(f"No chunks found for {document_name}")

    def list_documents(self) -> List[str]:
        """List all unique documents in the database"""
        all_data = self.collection.get()

        if not all_data['metadatas']:
            return []

        documents = set(meta['document'] for meta in all_data['metadatas'])
        return list(documents)

    def get_stats(self) -> Dict:
        """Get database statistics"""
        count = self.collection.count()
        documents = self.list_documents()

        return {
            'total_chunks': count,
            'documents': len(documents),
            'document_names': documents
        }

# Test the vector store
if __name__ == "__main__":
    store = VectorStore()

    # Example: Add some test data
    test_chunks = [
        {
            'id': 'chunk_0',
            'text': 'FastAPI is a modern web framework for Python.',
            'chunk_index': 0,
            'start_char': 0,
            'end_char': 50
        },
        {
            'id': 'chunk_1',
            'text': 'It is very fast and easy to use for building APIs.',
            'chunk_index': 1,
            'start_char': 50,
            'end_char': 100
        }
    ]

    store.add_documents(test_chunks, "test_doc")

    # Test search
    results = store.search("What is FastAPI?", n_results=2)
    print("\nSearch Results:")
    for i, doc in enumerate(results['documents'][0]):
        print(f"\n{i+1}. {doc}")

    # Get stats
    stats = store.get_stats()
    print(f"\nDatabase Stats: {stats}")

Understanding the Vector Store

Key components:

1. Embedding Model: all-MiniLM-L6-v2

• Free, runs locally
• Converts text → 384-dimensional vectors
• Fast and accurate for most use cases

2. ChromaDB:

• Stores embeddings persistently
• Handles similarity search
• Lightweight (perfect for free tier)

3. Similarity Search:

• Uses cosine similarity
• Returns closest matching chunks
• Includes distance scores (0 = identical, 2 = opposite)

Step 4: Building the RAG Pipeline

Create rag_engine.py:

from document_processor import DocumentProcessor
from vector_store import VectorStore
import ollama
from typing import List, Dict

class RAGEngine:
    """Complete RAG pipeline: process → store → retrieve → generate"""

    def __init__(self, model_name: str = "llama3.2"):
        """
        Initialize RAG engine

        Args:
            model_name: Ollama model to use for generation
        """
        self.processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)
        self.vector_store = VectorStore()
        self.model_name = model_name

        print(f"RAG Engine initialized with model: {model_name}")

    def ingest_document(self, pdf_path: str, document_name: str):
        """
        Process and store a document

        Args:
            pdf_path: Path to PDF file
            document_name: Unique name for this document
        """
        print(f"\n{'='*60}")
        print(f"Ingesting document: {document_name}")
        print(f"{'='*60}")

        # Process PDF into chunks
        chunks = self.processor.process_pdf(pdf_path)

        # Store in vector database
        self.vector_store.add_documents(chunks, document_name)

        print(f"✓ Document '{document_name}' successfully ingested!")
        print(f"{'='*60}\n")

    def retrieve_context(self, query: str, n_results: int = 3) -> tuple[List[str], List[Dict]]:
        """
        Retrieve relevant chunks for a query

        Args:
            query: User's question
            n_results: Number of chunks to retrieve

        Returns:
            Tuple of (context_texts, metadata)
        """
        results = self.vector_store.search(query, n_results=n_results)

        contexts = results['documents'][0] if results['documents'] else []
        metadatas = results['metadatas'][0] if results['metadatas'] else []
        distances = results['distances'][0] if results['distances'] else []

        # Add distance scores to metadata
        for i, meta in enumerate(metadatas):
            meta['relevance_score'] = round(1 - (distances[i] / 2), 3)  # Convert to 0-1 scale

        return contexts, metadatas

    def generate_answer(self, query: str, contexts: List[str]) -> str:
        """
        Generate answer using LLM with retrieved context

        Args:
            query: User's question
            contexts: Retrieved document chunks

        Returns:
            Generated answer
        """
        # Build prompt with context
        context_text = "\n\n".join([
            f"[Context {i+1}]\n{ctx}"
            for i, ctx in enumerate(contexts)
        ])

        prompt = f"""You are a helpful assistant answering questions about documents.

Use the following context to answer the question. If the answer cannot be found in the context, say so clearly.

CONTEXT:
{context_text}

QUESTION: {query}

ANSWER (be concise and cite which context sections you used):"""

        # Generate response
        response = ollama.chat(
            model=self.model_name,
            messages=[
                {
                    'role': 'system',
                    'content': 'You are a helpful assistant that answers questions based on provided context. Always cite your sources.'
                },
                {
                    'role': 'user',
                    'content': prompt
                }
            ],
            options={
                'temperature': 0.3  # Lower temperature for more factual responses
            }
        )

        return response['message']['content']

    def query(self, question: str, n_results: int = 3, verbose: bool = True) -> Dict:
        """
        Complete RAG query: retrieve + generate

        Args:
            question: User's question
            n_results: Number of contexts to retrieve
            verbose: Print detailed information

        Returns:
            Dictionary with answer, contexts, and metadata
        """
        if verbose:
            print(f"\n{'='*60}")
            print(f"Question: {question}")
            print(f"{'='*60}\n")

        # Step 1: Retrieve relevant contexts
        if verbose:
            print("🔍 Searching for relevant information...")

        contexts, metadatas = self.retrieve_context(question, n_results)

        if not contexts:
            return {
                'answer': "I don't have any relevant information to answer this question.",
                'contexts': [],
                'metadatas': [],
                'sources': []
            }

        if verbose:
            print(f"✓ Found {len(contexts)} relevant chunks\n")
            for i, (ctx, meta) in enumerate(zip(contexts, metadatas)):
                print(f"Context {i+1} (relevance: {meta['relevance_score']}):")
                print(f"  Document: {meta['document']}")
                print(f"  Preview: {ctx[:100]}...")
                print()

        # Step 2: Generate answer
        if verbose:
            print("🤖 Generating answer...\n")

        answer = self.generate_answer(question, contexts)

        if verbose:
            print(f"{'='*60}")
            print(f"Answer:\n{answer}")
            print(f"{'='*60}\n")

        # Prepare sources for citation
        sources = [
            {
                'document': meta['document'],
                'chunk_index': meta['chunk_index'],
                'relevance': meta['relevance_score'],
                'text': ctx[:200] + "..." if len(ctx) > 200 else ctx
            }
            for ctx, meta in zip(contexts, metadatas)
        ]

        return {
            'answer': answer,
            'contexts': contexts,
            'metadatas': metadatas,
            'sources': sources
        }

    def list_documents(self) -> List[str]:
        """List all ingested documents"""
        return self.vector_store.list_documents()

    def delete_document(self, document_name: str):
        """Remove a document from the system"""
        self.vector_store.delete_document(document_name)

    def get_stats(self) -> Dict:
        """Get system statistics"""
        return self.vector_store.get_stats()

# Interactive test
if __name__ == "__main__":
    print("Initializing RAG Engine...")
    rag = RAGEngine()

    # Example: Ingest a document
    # rag.ingest_document("path/to/your.pdf", "my_document")

    # Example: Query
    # result = rag.query("What are the main points?")

    # Show stats
    stats = rag.get_stats()
    print(f"\nSystem Stats: {stats}")

Understanding the RAG Flow

Ingestion (happens once per document):

PDF → Extract text → Clean → Chunk → Embed → Store

Querying (happens every time user asks):

Question → Embed → Search → Retrieve contexts → Build prompt → LLM → Answer

The magic prompt:

You are a helpful assistant...

CONTEXT:
[Retrieved chunk 1]
[Retrieved chunk 2]
[Retrieved chunk 3]

QUESTION: What is FastAPI?

ANSWER:

The LLM now has specific information to answer from!

Step 5: Testing the RAG System

Create test_rag.py:

from rag_engine import RAGEngine

def main():
    # Initialize
    print("="*60)
    print("RAG System Test")
    print("="*60)

    rag = RAGEngine()

    # Check if we have any documents
    stats = rag.get_stats()
    print(f"\nCurrent database stats:")
    print(f"  Total chunks: {stats['total_chunks']}")
    print(f"  Documents: {stats['documents']}")

    if stats['documents'] > 0:
        print(f"  Document names: {stats['document_names']}")

    # If you have a PDF to test, uncomment and modify:
    # print("\nIngesting document...")
    # rag.ingest_document("path/to/your.pdf", "test_document")

    # Test queries
    if stats['total_chunks'] > 0:
        print("\n" + "="*60)
        print("Testing Queries")
        print("="*60)

        test_questions = [
            "What is this document about?",
            "What are the main points discussed?",
            "Can you summarize the key findings?"
        ]

        for question in test_questions:
            result = rag.query(question, n_results=3, verbose=True)
            input("\nPress Enter to continue to next question...")
    else:
        print("\n⚠️  No documents in database yet!")
        print("Add a document using:")
        print('  rag.ingest_document("your_file.pdf", "doc_name")')

if __name__ == "__main__":
    main()

Step 6: Integrating RAG with FastAPI

Now let’s update our API to support document upload and RAG queries. Create rag_api.py:

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import os
import shutil
from rag_engine import RAGEngine

app = FastAPI(title="DocuChat API", version="1.0.0")

# Add CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize RAG engine
rag = RAGEngine()
# Create uploads directory
UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
# Request/Response models
class QueryRequest(BaseModel):
    question: str
    n_results: Optional[int] = 3
class QueryResponse(BaseModel):
    answer: str
    sources: List[dict]
class DocumentInfo(BaseModel):
    name: str
    chunks: int
@app.get("/")
def read_root():
    return {
        "message": "DocuChat API - RAG-powered document Q&A",
        "version": "1.0.0",
        "endpoints": {
            "upload": "/upload",
            "query": "/query",
            "documents": "/documents",
            "stats": "/stats"
        }
    }
@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """
    Upload and process a PDF document
    """
    # Validate file type
    if not file.filename.endswith('.pdf'):
        raise HTTPException(status_code=400, detail="Only PDF files are supported")

    try:
        # Save uploaded file
        file_path = os.path.join(UPLOAD_DIR, file.filename)

        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)

        # Process and ingest
        document_name = file.filename.replace('.pdf', '')
        rag.ingest_document(file_path, document_name)

        # Get stats
        stats = rag.get_stats()

        return {
            "message": f"Document '{file.filename}' uploaded and processed successfully",
            "document_name": document_name,
            "total_documents": stats['documents'],
            "total_chunks": stats['total_chunks']
        }

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error processing document: {str(e)}")
@app.post("/query", response_model=QueryResponse)
def query_documents(request: QueryRequest):
    """
    Query the document database
    """
    try:
        result = rag.query(request.question, n_results=request.n_results, verbose=False)

        return QueryResponse(
            answer=result['answer'],
            sources=result['sources']
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error processing query: {str(e)}")
@app.get("/documents")
def list_documents():
    """
    List all uploaded documents
    """
    documents = rag.list_documents()
    stats = rag.get_stats()

    return {
        "documents": documents,
        "count": len(documents),
        "total_chunks": stats['total_chunks']
    }
@app.delete("/documents/{document_name}")
def delete_document(document_name: str):
    """
    Delete a document from the database
    """
    try:
        rag.delete_document(document_name)
        return {"message": f"Document '{document_name}' deleted successfully"}

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error deleting document: {str(e)}")
@app.get("/stats")
def get_stats():
    """
    Get system statistics
    """
    return rag.get_stats()
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Fast API docs page showing the new endpoints

Step 7: Testing the Complete System

Terminal Setup

You’ll need two terminals:

Terminal 1: Ollama Server

ollama serve

Terminal 2: FastAPI Application

cd docuchat
source venv/bin/activate
python rag_api.py

Test with curl

1. Check API is running:

curl http://localhost:8000/

Checking if API is running

2. Upload a PDF:

curl -X POST "http://localhost:8000/upload" \
  -H "Content-Type: multipart/form-data" \
  -F "file=@/path/to/your/document.pdf"

You could also upload the PDF using the FastAPI docs page. If Using FastAPI Docs:

1. Go to [http://localhost:8000/docs](http://localhost:8000/docs)
2. Click on /upload endpoint
3. Click “Try it out”
4. Click “Choose File” button
5. Select your PDF
6. Click “Execute”

Successful PDF upload

3. Query the document:

curl -X POST "http://localhost:8000/query" \
  -H "Content-Type: application/json" \
  -d '{"question": "What is this document about?"}'

Showing response for the above query

4. List documents:

curl http://localhost:8000/documents

5. Get stats:

curl http://localhost:8000/stats

Check stats for the document I uploaded

Test with Python

Create test_api_client.py:

import requests

BASE_URL = "http://localhost:8000"
def test_upload():
    """Test document upload"""
    print("Testing document upload...")

    # Replace with your PDF path
    pdf_path = "path/to/your/document.pdf"

    with open(pdf_path, 'rb') as f:
        files = {'file': f}
        response = requests.post(f"{BASE_URL}/upload", files=files)

    print(f"Status: {response.status_code}")
    print(f"Response: {response.json()}")
    return response.json()
def test_query(question: str):
    """Test querying"""
    print(f"\nTesting query: {question}")

    response = requests.post(
        f"{BASE_URL}/query",
        json={"question": question, "n_results": 3}
    )

    result = response.json()
    print(f"\nAnswer: {result['answer']}")
    print(f"\nSources:")
    for i, source in enumerate(result['sources']):
        print(f"{i+1}. {source['document']} (relevance: {source['relevance']})")
        print(f"   {source['text'][:100]}...")
def test_list_documents():
    """Test listing documents"""
    print("\nListing documents...")

    response = requests.get(f"{BASE_URL}/documents")
    print(f"Response: {response.json()}")
if __name__ == "__main__":
    # Test upload (uncomment and provide PDF path)
    # test_upload()

    # Test queries
    test_query("What is this document about?")
    test_query("What are the main points?")

    # List documents
    test_list_documents()

Showing output for test_api_client.py

Understanding RAG Performance

Chunking Strategy Trade-offs

Smaller chunks (200–300 chars):

• ✅ More precise retrieval
• ✅ Better citations
• ❌ Less context per chunk
• ❌ More chunks = slower search

Larger chunks (800–1000 chars):

• ✅ More context
• ✅ Fewer chunks = faster search
• ❌ Less precise retrieval
• ❌ Harder to cite specific info

Our choice (500 chars): Good balance for most use cases

Number of Retrieved Chunks

Fewer chunks (1–2):

• ✅ Faster generation
• ✅ More focused answers
• ❌ Might miss relevant info

More chunks (5–10):

• ✅ More comprehensive
• ✅ Less likely to miss info
• ❌ Slower generation
• ❌ Can confuse the model

Our choice (3): Sweet spot for accuracy vs speed

Temperature Settings

For RAG, we use low temperature (0.3):

• More factual, less creative
• Sticks closer to source material
• Reduces hallucinations

Common Issues and Solutions

Issue 1: Irrelevant Results

Problem: Search returns chunks that don’t match the question

Solutions:

1. Improve chunking (better sentence boundaries)
2. Use better embedding model
3. Increase number of results and filter by relevance score
4. Add query preprocessing (expand acronyms, fix typos)

Issue 2: Answer Not in Context

Problem: User asks about something not in the documents

Solution: The prompt tells LLM to say so clearly:

"If the answer cannot be found in the context, say so clearly."

Issue 3: Hallucination

Problem: LLM makes up information

Solutions:

1. Lower temperature (we use 0.3)
2. Emphasize in prompt to only use context
3. Show retrieved chunks to user for verification

Issue 4: Slow Performance

Problem: Queries take too long

Solutions:

1. Use smaller embedding model
2. Reduce chunk count
3. Cache frequent queries
4. Use smaller LLM (llama3.2:1b)

What We’ve Accomplished

Incredible progress! You now have:

✅ PDF document processing with intelligent chunking ✅ Vector embeddings and semantic search ✅ ChromaDB for persistent storage ✅ Complete RAG pipeline (retrieve + generate) ✅ FastAPI endpoints for upload and query ✅ Citation and source tracking ✅ Production-ready error handling

Your DocuChat application can now:

• Upload PDF documents
• Answer questions about them
• Cite sources
• Handle multiple documents
• Run 100% locally and free

What’s Next?

In Part 4, we’ll build a beautiful frontend for DocuChat:

• File upload interface with drag-and-drop
• Chat interface with message history
• Source citations display
• Document management
• Responsive design

Then in later parts:

• Part 5: Deployment (free hosting options)
• Part 6: Advanced features (conversation memory, multi-turn chat)
• Part 7: Performance optimization and caching
• Part 8: Testing and quality assurance

Exercises

Before Part 4, try these:

1. Test with different documents: Upload various PDFs and see how it performs
2. Experiment with chunk sizes: Try 300, 500, and 800 — which works best for your documents?
3. Compare retrieval counts: Test n_results of 1, 3, 5, 10 — how does it affect answers?
4. Add metadata: Modify the chunking to include page numbers, section headers
5. Improve prompts: Experiment with different system prompts for better answers

Complete Code Repository

All code is available on GitHub:

• Branch: [part-3-rag-system](https://github.com/meetnandu05/docuchat/tree/part-3)
• New files: document_processor.py, vector_store.py, rag_engine.py, rag_api.py

GitHub Repository: [https://github.com/meetnandu05/docuchat](https://github.com/meetnandu05/docuchat)

Additional Resources

Learn more about RAG:

• LangChain RAG Guide
• ChromaDB Documentation
• Sentence Transformers

Advanced RAG techniques:

• Hybrid search (combining keyword + semantic)
• Re-ranking retrieved results
• Query expansion
• Recursive retrieval

Next in the series: Part 4 — Building the Frontend Interface

Have questions? Drop them in the comments!

Enjoying the series? Give it a clap and follow for Part 4!

Happy coding! 🚀