Skip to content

BACKEND_DOCUMENTATION.md

Latest commit

 

History

History
2004 lines (1622 loc) · 53.9 KB

BACKEND_DOCUMENTATION.md

File metadata and controls

2004 lines (1622 loc) · 53.9 KB
Raw

DocuMentor Backend Documentation

Table of Contents

  1. Overview
  2. Architecture
  3. Project Structure
  4. API Endpoints
  5. Models & AI Components
  6. Services
  7. Data Flow & Workflows
  8. Configuration
  9. Key Features Implementation
  10. Code Deep Dive

Overview

What is DocuMentor Backend?

The backend is a FastAPI-based REST API that powers an offline, privacy-first AI tutor. It processes PDF documents, generates embeddings, and provides intelligent features like summarization, question-answering (using RAG), and quiz generation - all running 100% locally without any cloud services or API keys.

Technology Stack

  • Framework: FastAPI 0.109 + Uvicorn (ASGI server)
  • ML/AI: PyTorch 2.1+, Transformers 4.45+, Sentence Transformers
  • Vector Database: FAISS (CPU-based semantic search)
  • PDF Processing: PyMuPDF, pdfplumber
  • Model Optimization: bitsandbytes (4-bit quantization), PEFT
  • Language: Python 3.8+

Core Capabilities

  1. PDF Upload & Processing - Extract, chunk, and index documents
  2. Semantic Search - FAISS-powered vector similarity search
  3. Summarization - Academic-focused text summarization (3 styles)
  4. Question Answering - RAG-based Q&A with context retrieval
  5. Quiz Generation - Automated MCQ creation from documents

Architecture

High-Level Architecture

┌─────────────────────────────────────────────────────────────┐
│                        FastAPI Server                        │
│                     (main.py + routes.py)                    │
└─────────────────┬───────────────────────────────────────────┘
                  │
        ┌─────────┼─────────┐
        │         │         │
   ┌────▼───┐ ┌──▼────┐ ┌─▼─────┐
   │ Models │ │Services│ │ Utils │
   └────┬───┘ └──┬────┘ └─┬─────┘
        │        │         │
   ┌────▼────────▼─────────▼────┐
   │   Embeddings   RAG   Config │
   │   Phi3Sum     Vector  Chunk │
   │   T5Quiz      PDFProc Clean │
   └────────────┬─────────────────┘
                │
        ┌───────▼────────┐
        │  Data Storage  │
        │ /data/uploads  │
        │ /data/vectors  │
        │ /data/processed│
        └────────────────┘

Request-Response Flow

Client Request
    ↓
FastAPI Endpoint (/api/v1/*)
    ↓
Schema Validation (Pydantic)
    ↓
Service Layer (pdf_processor, vector_store, rag_pipeline)
    ↓
Model Layer (embeddings, phi3_summarizer, t5_quiz_generator)
    ↓
Utilities (chunker, config, quiz_cleaner)
    ↓
Response JSON
    ↓
Client

Project Structure

backend/
├── main.py                      # FastAPI app initialization
│
├── api/
│   ├── routes.py                # All 7 API endpoints
│   └── schemas.py               # Pydantic request/response models
│
├── models/
│   ├── embeddings.py            # BAAI/bge-small wrapper
│   ├── phi3_summarizer.py       # Academic summarization model
│   └── t5_quiz_generator.py     # MCQ generation model
│
├── services/
│   ├── pdf_processor.py         # PDF text extraction & chunking
│   ├── vector_store.py          # FAISS vector database manager
│   └── rag_pipeline.py          # Retrieval-Augmented Generation
│
├── utils/
│   ├── config.py                # Global configuration
│   ├── chunker.py               # Smart text chunking
│   └── quiz_cleaner.py          # Post-process quiz output
│
└── testing_scripts/
    ├── pegasustest.py           # Summarization evaluation
    └── evaluation_script.py     # General testing utilities

API Endpoints

Base URL

http://localhost:8000/api/v1

Endpoint Reference

Endpoint Method Purpose Request Response
/health GET Check backend & model status None HealthResponse
/upload POST Upload PDF document multipart/form-data UploadResponse
/summarize POST Generate document summary SummarizeRequest SummarizeResponse
/ask POST Answer question using RAG ChatRequest ChatResponse
/generate-quiz POST Generate MCQ quiz QuizRequest QuizResponse
/documents GET List all uploaded documents None List[DocumentInfo]
/documents/{doc_id} DELETE Delete document & data doc_id path param DeleteResponse

1. Health Check - /health

Purpose: Check if backend is running and models are loaded

Request:

GET /api/v1/health

Response:

{
  "status": "healthy",
  "models_loaded": {
    "embeddings": true,
    "summarizer": true,
    "quiz_generator": false
  },
  "device": "cuda",
  "memory_usage": "5.2 GB"
}

Implementation: backend/api/routes.py:26

@router.get("/health", response_model=HealthResponse)
async def health_check():
    """Check backend health and model loading status"""
    models_loaded = {
        "embeddings": embedding_model is not None,
        "summarizer": phi3_summarizer is not None,
        "quiz_generator": quiz_gen is not None
    }

    return HealthResponse(
        status="healthy",
        models_loaded=models_loaded,
        device=str(settings.device)
    )

Why it matters: Frontend uses this to show model loading status in the sidebar

2. Upload PDF - /upload

Purpose: Upload PDF, extract text, chunk it, generate embeddings, create vector index

Request:

POST /api/v1/upload
Content-Type: multipart/form-data

file=@document.pdf

Response:

{
  "message": "Document uploaded successfully",
  "doc_id": "document_1732898765",
  "file_name": "document.pdf",
  "pages": 42,
  "chunks": 156,
  "metadata": {
    "author": "John Doe",
    "title": "Machine Learning Fundamentals"
  }
}

Implementation: backend/api/routes.py:45

Step-by-step workflow:

  1. Validate File
if not file.filename.endswith('.pdf'):
    raise HTTPException(status_code=400, detail="Only PDF files allowed")
  1. Generate Document ID
doc_id = f"{file.filename.rsplit('.', 1)[0]}_{int(time.time())}"
  1. Save to Disk
file_path = settings.uploads_dir / f"{doc_id}.pdf"
with open(file_path, "wb") as f:
    f.write(await file.read())
  1. Process PDF
processor = PDFProcessor()
result = processor.process_pdf(str(file_path), doc_id)
# Returns: full_text, chunks, metadata, structure
  1. Generate Embeddings
chunk_texts = [chunk["text"] for chunk in result["chunks"]]
embeddings = embedding_model.encode_documents(chunk_texts)
  1. Create Vector Store
vector_store = VectorStore(embedding_dim=384)
metadata = [
    {
        "chunk_id": chunk["chunk_id"],
        "text": chunk["text"],
        "word_count": chunk["word_count"]
    }
    for chunk in result["chunks"]
]
vector_store.add_vectors(embeddings, metadata)
vector_store.save(settings.vectors_dir / doc_id)
  1. Save Processed Data
processed_path = settings.processed_dir / f"{doc_id}.json"
with open(processed_path, "w") as f:
    json.dump(result, f, indent=2)

Key Files:

  • PDF Processing: backend/services/pdf_processor.py:45-120
  • Vector Store: backend/services/vector_store.py:15-150
  • Embeddings: backend/models/embeddings.py:25-65

3. Summarize Document - /summarize

Purpose: Generate summary of document in bullet, paragraph, or exam style

Request:

{
  "doc_id": "document_1732898765",
  "summary_style": "bullet",
  "chunk_wise": false
}

Response:

{
  "summary": "• Machine learning is a subset of AI...\n• Key algorithms include...\n• Applications range from...",
  "doc_id": "document_1732898765",
  "style": "bullet",
  "chunks_summarized": 156
}

Implementation: backend/api/routes.py:120

Summary Styles:

  • bullet - Concise bullet points
  • paragraph - Narrative summary
  • exam - Exam-focused key points

Workflow:

  1. Load Processed Document
processed_path = settings.processed_dir / f"{doc_id}.json"
with open(processed_path, "r") as f:
    doc_data = json.load(f)
  1. Choose Summarization Method
if chunk_wise:
    # Summarize each chunk, then combine
    chunk_summaries = phi3_summarizer.summarize_chunks(
        chunks=doc_data["chunks"],
        style=summary_style
    )
    final_summary = "\n\n".join(chunk_summaries)
else:
    # Summarize full document
    final_summary = phi3_summarizer.summarize(
        text=doc_data["full_text"],
        style=summary_style,
        max_length=500
    )
  1. Return Summary
return SummarizeResponse(
    summary=final_summary,
    doc_id=doc_id,
    style=summary_style,
    chunks_summarized=len(doc_data["chunks"])
)

Key Files:

  • Summarization Model: backend/models/phi3_summarizer.py:45-180
  • Routes: backend/api/routes.py:120-165

4. Ask Question (RAG) - /ask

Purpose: Answer questions about the document using Retrieval-Augmented Generation

Request:

{
  "doc_id": "document_1732898765",
  "query": "What are the main types of machine learning?",
  "include_sources": true
}

Response:

{
  "answer": "The main types of machine learning are supervised learning, unsupervised learning, and reinforcement learning. Supervised learning uses labeled data...",
  "sources": [
    {
      "chunk_id": 12,
      "text": "Machine learning can be categorized into...",
      "relevance_score": 0.89
    }
  ],
  "doc_id": "document_1732898765"
}

Implementation: backend/api/routes.py:180

RAG Pipeline Steps:

  1. Load Vector Store
vector_store_path = settings.vectors_dir / doc_id
doc_vector_store = VectorStore(embedding_dim=384)
doc_vector_store.load(vector_store_path)
  1. Initialize RAG Pipeline
rag = RAGPipeline(
    vector_store=doc_vector_store,
    embedding_model=embedding_model,
    llm=phi3_summarizer  # Used for generation
)
  1. Retrieve Relevant Chunks
# Inside RAGPipeline.answer()
query_embedding = self.embedding_model.encode_query(query)
similar_chunks = self.vector_store.search(
    query_vector=query_embedding,
    k=5  # Top 5 most relevant chunks
)
  1. Build Context
context = "\n\n".join([
    f"[Chunk {meta['chunk_id']}]: {meta['text']}"
    for vec, meta in similar_chunks
])
  1. Generate Answer
prompt = f"""Based on the following context, answer the question.

Context:
{context}

Question: {query}

Answer:"""

answer = self.llm.generate(prompt, max_length=300)
  1. Return with Sources
return ChatResponse(
    answer=answer,
    sources=[
        {
            "chunk_id": meta["chunk_id"],
            "text": meta["text"],
            "relevance_score": float(distance)
        }
        for vec, meta in similar_chunks
    ] if include_sources else None,
    doc_id=doc_id
)

Key Files:

  • RAG Pipeline: backend/services/rag_pipeline.py:30-180
  • Vector Search: backend/services/vector_store.py:75-120

5. Generate Quiz - /generate-quiz

Purpose: Generate multiple-choice questions from the document

Request:

{
  "doc_id": "document_1732898765",
  "num_questions": 10,
  "questions_per_chunk": 2
}

Response:

{
  "questions": [
    {
      "question": "What is the primary goal of supervised learning?",
      "options": [
        "To classify labeled data",
        "To find patterns in unlabeled data",
        "To maximize rewards through actions",
        "To compress data dimensions"
      ],
      "correct_answer": 0,
      "explanation": "Supervised learning uses labeled data to train models..."
    }
  ],
  "doc_id": "document_1732898765",
  "total_questions": 10
}

Implementation: backend/api/routes.py:235

Quiz Generation Workflow:

  1. Load Document Chunks
processed_path = settings.processed_dir / f"{doc_id}.json"
with open(processed_path, "r") as f:
    doc_data = json.load(f)
chunks = doc_data["chunks"]
  1. Smart Chunk Sampling
# Don't process all chunks - sample intelligently
num_chunks_to_use = min(
    len(chunks),
    (num_questions // questions_per_chunk) + 2
)
sampled_chunks = random.sample(chunks, num_chunks_to_use)
  1. Generate Questions per Chunk
all_questions = []
for chunk in sampled_chunks:
    questions = quiz_gen.generate_quiz(
        text=chunk["text"],
        num_questions=questions_per_chunk
    )
    all_questions.extend(questions)
  1. Clean and Validate
cleaned_questions = []
for q in all_questions:
    cleaned = QuizCleaner.clean_question(q)
    if QuizCleaner.validate_question(cleaned):
        cleaned_questions.append(cleaned)
  1. Limit to Requested Number
final_questions = cleaned_questions[:num_questions]

Key Files:

  • Quiz Generator: backend/models/t5_quiz_generator.py:45-250
  • Quiz Cleaner: backend/utils/quiz_cleaner.py:15-100

6. List Documents - /documents

Purpose: Get list of all uploaded documents with metadata

Request:

GET /api/v1/documents

Response:

[
  {
    "doc_id": "document_1732898765",
    "file_name": "ml_fundamentals.pdf",
    "pages": 42,
    "chunks": 156,
    "upload_date": "2024-11-29T10:32:45"
  },
  {
    "doc_id": "another_doc_1732900000",
    "file_name": "deep_learning.pdf",
    "pages": 58,
    "chunks": 203,
    "upload_date": "2024-11-29T11:00:00"
  }
]

Implementation: backend/api/routes.py:295

@router.get("/documents", response_model=List[DocumentInfo])
async def list_documents():
    """List all uploaded documents"""
    documents = []

    # Scan processed directory
    for file_path in settings.processed_dir.glob("*.json"):
        with open(file_path, "r") as f:
            doc_data = json.load(f)

        documents.append(DocumentInfo(
            doc_id=doc_data["doc_id"],
            file_name=doc_data["metadata"]["file_name"],
            pages=doc_data["metadata"]["total_pages"],
            chunks=doc_data["total_chunks"],
            upload_date=file_path.stat().st_mtime
        ))

    return documents

7. Delete Document - /documents/{doc_id}

Purpose: Remove document and all associated data

Request:

DELETE /api/v1/documents/document_1732898765

Response:

{
  "message": "Document deleted successfully",
  "doc_id": "document_1732898765"
}

Implementation: backend/api/routes.py:330

Deletion Process:

  1. Remove Original PDF
pdf_path = settings.uploads_dir / f"{doc_id}.pdf"
if pdf_path.exists():
    pdf_path.unlink()
  1. Remove Vector Index
vector_index = settings.vectors_dir / f"{doc_id}.index"
vector_meta = settings.vectors_dir / f"{doc_id}.meta"
if vector_index.exists():
    vector_index.unlink()
if vector_meta.exists():
    vector_meta.unlink()
  1. Remove Processed Data
processed_path = settings.processed_dir / f"{doc_id}.json"
if processed_path.exists():
    processed_path.unlink()
  1. Clear from Memory
if doc_id in loaded_vector_stores:
    del loaded_vector_stores[doc_id]

Models & AI Components

1. Embedding Model - models/embeddings.py

Model: BAAI/bge-small-en-v1.5

  • Parameters: 33M
  • Embedding Dimension: 384
  • Purpose: Convert text to semantic vectors for similarity search

Key Methods:

class EmbeddingModel:
    def __init__(self, model_name: str, device: str):
        self.model = SentenceTransformer(model_name, device=device)
        self.embedding_dim = 384

    def encode(self, texts: Union[str, List[str]]) -> np.ndarray:
        """General encoding - handles both single and batch"""
        return self.model.encode(
            texts,
            normalize_embeddings=True,  # L2 normalization
            show_progress_bar=True
        )

    def encode_query(self, query: str) -> np.ndarray:
        """Optimized for search queries"""
        return self.encode(query)

    def encode_documents(self, documents: List[str]) -> np.ndarray:
        """Batch encoding for document chunks"""
        return self.encode(documents)

Why this model?

  • Small and fast (33M params)
  • Optimized for English semantic search
  • High quality embeddings in 384 dimensions
  • Runs efficiently on CPU or GPU

Usage in Pipeline:

# During upload
chunk_texts = ["chunk 1 text...", "chunk 2 text..."]
embeddings = embedding_model.encode_documents(chunk_texts)
# Shape: (num_chunks, 384)

# During search
query = "What is machine learning?"
query_emb = embedding_model.encode_query(query)
# Shape: (384,)

2. Summarization Model - models/phi3_summarizer.py

Model: RandipR/pegasus-560m-academic-sum

  • Parameters: 560M
  • Base: Google PEGASUS
  • Fine-tuned: Academic papers and textbooks
  • Architecture: Seq2Seq (encoder-decoder)

Key Features:

class Phi3Summarizer:
    def __init__(self, model_name: str, device: str):
        # Load model and tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSeq2SeqLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16  # Memory optimization
        ).to(device)

        self.max_input_length = 1024
        self.max_output_length = 256

Method 1: Full Document Summary

def summarize(
    self,
    text: str,
    style: str = "paragraph",
    max_length: int = 150
) -> str:
    """Summarize entire document at once"""

    # Clean text
    text = self._clean_text(text)

    # Add style instruction
    if style == "bullet":
        text = "Summarize in bullet points:\n" + text
    elif style == "exam":
        text = "Summarize key exam points:\n" + text

    # Tokenize
    inputs = self.tokenizer(
        text,
        max_length=self.max_input_length,
        truncation=True,
        return_tensors="pt"
    ).to(self.device)

    # Generate
    outputs = self.model.generate(
        **inputs,
        max_length=max_length,
        num_beams=4,
        early_stopping=True,
        no_repeat_ngram_size=3
    )

    # Decode
    summary = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
    return self._clean_artifacts(summary)

Method 2: Chunk-wise Summary

def summarize_chunks(
    self,
    chunks: List[Dict],
    style: str = "bullet",
    combine: bool = True
) -> Union[str, List[str]]:
    """Summarize each chunk separately, optionally combine"""

    chunk_summaries = []

    for i, chunk in enumerate(chunks):
        # Summarize individual chunk
        summary = self.summarize(
            text=chunk["text"],
            style=style,
            max_length=100
        )
        chunk_summaries.append(summary)

        # Clear GPU memory periodically
        if i % 10 == 0:
            torch.cuda.empty_cache()

    if combine:
        # Combine all chunk summaries
        combined = "\n\n".join(chunk_summaries)

        # Optionally: summarize the summaries
        if len(combined) > 2000:
            combined = self.summarize(
                text=combined,
                style=style,
                max_length=300
            )

        return combined
    else:
        return chunk_summaries

Text Cleaning:

def _clean_text(self, text: str) -> str:
    """Remove noise and normalize"""
    # Remove extra whitespace
    text = re.sub(r'\s+', ' ', text)
    # Remove special characters
    text = re.sub(r'[^\w\s\.,!?-]', '', text)
    # Remove URLs
    text = re.sub(r'http\S+|www.\S+', '', text)
    return text.strip()

def _clean_artifacts(self, text: str) -> str:
    """Remove model artifacts from output"""
    # Remove incomplete sentences at end
    text = re.sub(r'\s+[A-Z][a-z]*$', '', text)
    # Fix spacing
    text = re.sub(r'\s+([.,!?])', r'\1', text)
    return text.strip()

Summary Styles Implementation:

Style Prefix Max Length Output Format
bullet "Summarize in bullet points:" 200 • Point 1\n• Point 2
paragraph None (default) 150 Narrative paragraph
exam "Summarize key exam points:" 180 Key concepts for studying

Why Pegasus?

  • Specifically designed for abstractive summarization
  • Pre-training uses gap-sentence generation
  • Academic fine-tuning improves technical text handling
  • Produces fluent, coherent summaries

3. Quiz Generation Model - models/t5_quiz_generator.py

Model: RandipR/Qwen2.5-0.5B-Instruct-MCQ-Generation

  • Parameters: 500M
  • Base: Qwen 2.5 (Alibaba)
  • Fine-tuned: MCQ generation task
  • Architecture: Decoder-only (Causal LM)

Model Loading with PEFT:

class T5QuizGenerator:
    def __init__(self, model_name: str, device: str, use_4bit: bool = True):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

        if use_4bit:
            # 4-bit quantization for memory efficiency
            bnb_config = BitsAndBytesConfig(
                load_in_4bit=True,
                bnb_4bit_quant_type="nf4",
                bnb_4bit_compute_dtype=torch.float16
            )

            self.model = AutoModelForCausalLM.from_pretrained(
                model_name,
                quantization_config=bnb_config,
                device_map="auto"
            )
        else:
            self.model = AutoModelForCausalLM.from_pretrained(
                model_name,
                torch_dtype=torch.float16
            ).to(device)

        # Check for PEFT adapter
        if hasattr(self.model, 'peft_config'):
            print("PEFT adapter detected")

Quiz Generation Process:

def generate_quiz(
    self,
    text: str,
    num_questions: int = 5,
    difficulty: str = "medium"
) -> List[MCQuestion]:
    """Generate MCQs from text chunk"""

    # Build prompt
    prompt = self._build_prompt(text, num_questions, difficulty)

    # Tokenize
    inputs = self.tokenizer(
        prompt,
        max_length=1024,
        truncation=True,
        return_tensors="pt"
    ).to(self.device)

    # Generate
    outputs = self.model.generate(
        **inputs,
        max_new_tokens=512,
        temperature=0.7,
        top_p=0.9,
        do_sample=True,
        num_return_sequences=1
    )

    # Decode
    generated = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

    # Parse MCQs from output
    questions = self._parse_mcqs(generated)

    return questions[:num_questions]

Prompt Engineering:

def _build_prompt(self, text: str, num_questions: int, difficulty: str) -> str:
    """Craft effective prompt for MCQ generation"""

    prompt = f"""You are an expert quiz creator. Generate {num_questions} multiple-choice questions from the following text.

Requirements:
- Difficulty: {difficulty}
- 4 options per question (A, B, C, D)
- Only one correct answer
- Distractors should be plausible but incorrect
- Questions should test understanding, not memorization

Text:
{text}

Generate questions in this format:

Question 1: [Question text]
A) [Option A]
B) [Option B]
C) [Option C]
D) [Option D]
Correct Answer: [A/B/C/D]
Explanation: [Why this is correct]

Begin:
"""
    return prompt

MCQ Parsing:

def _parse_mcqs(self, generated_text: str) -> List[MCQuestion]:
    """Extract structured MCQs from model output"""

    questions = []

    # Regex pattern for MCQ format
    pattern = r"Question \d+:\s*(.+?)\n[A-D]\)\s*(.+?)\n[A-D]\)\s*(.+?)\n[A-D]\)\s*(.+?)\n[A-D]\)\s*(.+?)\nCorrect Answer:\s*([A-D])"

    matches = re.finditer(pattern, generated_text, re.DOTALL)

    for match in matches:
        question_text = match.group(1).strip()
        options = [
            match.group(2).strip(),
            match.group(3).strip(),
            match.group(4).strip(),
            match.group(5).strip()
        ]
        correct = ord(match.group(6)) - ord('A')  # Convert A-D to 0-3

        questions.append(MCQuestion(
            question=question_text,
            options=options,
            correct_answer=correct,
            explanation=match.group(7).strip() if len(match.groups()) > 6 else ""
        ))

    return questions

Smart Chunk Sampling:

def generate_quiz_from_document(
    self,
    chunks: List[Dict],
    total_questions: int = 10,
    questions_per_chunk: int = 2
) -> List[MCQuestion]:
    """Generate quiz by sampling chunks intelligently"""

    # Don't process all chunks - sample based on need
    num_chunks_needed = (total_questions // questions_per_chunk) + 2
    num_chunks_needed = min(num_chunks_needed, len(chunks))

    # Sample diverse chunks (not just first N)
    sampled_chunks = random.sample(chunks, num_chunks_needed)

    all_questions = []

    for chunk in sampled_chunks:
        # Generate questions from chunk
        chunk_questions = self.generate_quiz(
            text=chunk["text"],
            num_questions=questions_per_chunk
        )
        all_questions.extend(chunk_questions)

        # Stop if we have enough
        if len(all_questions) >= total_questions:
            break

    return all_questions[:total_questions]

Why Qwen 2.5?

  • Excellent instruction following
  • Good at structured output generation
  • Efficient at 0.5B parameters
  • Fine-tuned specifically for MCQ creation

Services

1. PDF Processor - services/pdf_processor.py

Purpose: Extract text from PDFs, chunk it, detect structure

Key Methods:

Method 1: Text Extraction

class PDFProcessor:
    def extract_text_pymupdf(self, pdf_path: str) -> Tuple[str, Dict]:
        """Fast extraction using PyMuPDF"""
        doc = fitz.open(pdf_path)
        full_text = ""
        metadata = {
            "total_pages": len(doc),
            "author": doc.metadata.get("author", "Unknown"),
            "title": doc.metadata.get("title", "Untitled")
        }

        for page_num in range(len(doc)):
            page = doc[page_num]
            text = page.get_text()
            full_text += text + "\n"

        doc.close()
        return full_text, metadata

    def extract_text_pdfplumber(self, pdf_path: str) -> str:
        """Fallback extraction using pdfplumber"""
        full_text = ""

        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                text = page.extract_text()
                if text:
                    full_text += text + "\n"

        return full_text

Method 2: Text Chunking

def chunk_text(
    self,
    text: str,
    chunk_size: int = 250,  # words
    overlap: int = 50       # words
) -> List[Dict]:
    """Split text into overlapping chunks"""

    # Clean text first
    text = self._clean_text(text)

    # Split into words
    words = text.split()

    chunks = []
    chunk_id = 0
    start = 0

    while start < len(words):
        # Get chunk
        end = min(start + chunk_size, len(words))
        chunk_words = words[start:end]

        # Create chunk dict
        chunk = {
            "chunk_id": chunk_id,
            "text": " ".join(chunk_words),
            "word_count": len(chunk_words),
            "start_word": start,
            "end_word": end
        }

        chunks.append(chunk)

        # Move to next chunk with overlap
        start += (chunk_size - overlap)
        chunk_id += 1

    return chunks

Method 3: Structure Detection

def detect_structure(self, text: str) -> Dict:
    """Detect headers and sections"""

    lines = text.split('\n')
    headers = []
    sections = []

    current_section = None
    section_text = []

    for line in lines:
        line = line.strip()

        # Detect headers (all caps, or numbered)
        if self._is_header(line):
            if current_section:
                sections.append({
                    "header": current_section,
                    "text": "\n".join(section_text)
                })
                section_text = []

            current_section = line
            headers.append(line)
        else:
            section_text.append(line)

    # Add final section
    if current_section:
        sections.append({
            "header": current_section,
            "text": "\n".join(section_text)
        })

    return {
        "headers": headers,
        "sections": sections
    }

def _is_header(self, line: str) -> bool:
    """Heuristic to detect if line is a header"""
    if not line:
        return False

    # All caps (at least 50% uppercase)
    if sum(c.isupper() for c in line) / len(line) > 0.5:
        return True

    # Numbered (e.g., "1. Introduction")
    if re.match(r'^\d+\.?\s+[A-Z]', line):
        return True

    # Short and title case
    if len(line.split()) <= 5 and line[0].isupper():
        return True

    return False

Full Processing Pipeline:

def process_pdf(self, pdf_path: str, doc_id: str) -> Dict:
    """Complete PDF processing pipeline"""

    try:
        # 1. Extract text
        full_text, metadata = self.extract_text_pymupdf(pdf_path)
    except Exception as e:
        print(f"PyMuPDF failed: {e}, trying pdfplumber")
        full_text = self.extract_text_pdfplumber(pdf_path)
        metadata = {"total_pages": 0, "author": "Unknown", "title": "Untitled"}

    # 2. Clean text
    full_text = self._clean_text(full_text)

    # 3. Detect structure
    structure = self.detect_structure(full_text)

    # 4. Chunk text
    chunks = self.chunk_text(full_text)

    # 5. Compile result
    result = {
        "doc_id": doc_id,
        "metadata": metadata,
        "full_text": full_text,
        "chunks": chunks,
        "total_chunks": len(chunks),
        "structure": structure
    }

    return result

Configuration:

  • Default chunk size: 250 words (~500 tokens)
  • Default overlap: 50 words (20%)
  • Min chunk size: 150 words
  • Max chunk size: 300 words

2. Vector Store - services/vector_store.py

Purpose: FAISS-based semantic search with metadata

Classes:

Class 1: Single Document Vector Store

class VectorStore:
    def __init__(self, embedding_dim: int = 384):
        self.embedding_dim = embedding_dim
        self.index = None
        self.metadata = []

    def add_vectors(
        self,
        vectors: np.ndarray,
        metadata: List[Dict]
    ):
        """Add vectors to index"""

        # Create index if doesn't exist
        if self.index is None:
            self.index = faiss.IndexFlatL2(self.embedding_dim)

        # Ensure correct shape
        if len(vectors.shape) == 1:
            vectors = vectors.reshape(1, -1)

        # Add to FAISS
        self.index.add(vectors.astype('float32'))

        # Store metadata
        self.metadata.extend(metadata)

    def search(
        self,
        query_vector: np.ndarray,
        k: int = 5
    ) -> List[Tuple[np.ndarray, Dict]]:
        """Search for top-k similar vectors"""

        if self.index is None:
            return []

        # Ensure query shape
        if len(query_vector.shape) == 1:
            query_vector = query_vector.reshape(1, -1)

        # Search
        distances, indices = self.index.search(
            query_vector.astype('float32'),
            k
        )

        # Combine with metadata
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx < len(self.metadata):
                results.append((
                    dist,  # L2 distance (lower = more similar)
                    self.metadata[idx]
                ))

        return results

    def save(self, path: Path):
        """Save index and metadata to disk"""

        # Save FAISS index
        faiss.write_index(self.index, str(path) + ".index")

        # Save metadata as pickle
        with open(str(path) + ".meta", "wb") as f:
            pickle.dump(self.metadata, f)

    def load(self, path: Path):
        """Load index and metadata from disk"""

        # Load FAISS index
        self.index = faiss.read_index(str(path) + ".index")

        # Load metadata
        with open(str(path) + ".meta", "rb") as f:
            self.metadata = pickle.load(f)

Class 2: Multi-Document Manager

class DocumentVectorStore:
    def __init__(self):
        self.stores: Dict[str, VectorStore] = {}

    def add_document(
        self,
        doc_id: str,
        vectors: np.ndarray,
        metadata: List[Dict]
    ):
        """Add new document"""
        store = VectorStore()
        store.add_vectors(vectors, metadata)
        self.stores[doc_id] = store

    def search_document(
        self,
        doc_id: str,
        query_vector: np.ndarray,
        k: int = 5
    ) -> List[Tuple[np.ndarray, Dict]]:
        """Search within specific document"""
        if doc_id not in self.stores:
            raise ValueError(f"Document {doc_id} not found")

        return self.stores[doc_id].search(query_vector, k)

    def save_all(self, base_path: Path):
        """Save all document stores"""
        for doc_id, store in self.stores.items():
            store.save(base_path / doc_id)

    def load_all(self, base_path: Path):
        """Load all document stores from directory"""
        for index_file in base_path.glob("*.index"):
            doc_id = index_file.stem
            store = VectorStore()
            store.load(base_path / doc_id)
            self.stores[doc_id] = store

Why FAISS?

  • Fast similarity search (millions of vectors)
  • CPU-based (no GPU needed for search)
  • L2 distance metric (Euclidean)
  • Compact index files

3. RAG Pipeline - services/rag_pipeline.py

Purpose: Retrieval-Augmented Generation for Q&A

Complete Implementation:

class RAGPipeline:
    def __init__(
        self,
        vector_store: VectorStore,
        embedding_model: EmbeddingModel,
        llm: Phi3Summarizer  # Reuse summarizer for generation
    ):
        self.vector_store = vector_store
        self.embedding_model = embedding_model
        self.llm = llm

    def answer(
        self,
        query: str,
        k: int = 5,
        include_sources: bool = False
    ) -> Dict:
        """Answer question using RAG"""

        # 1. Encode query
        query_embedding = self.embedding_model.encode_query(query)

        # 2. Retrieve similar chunks
        results = self.vector_store.search(query_embedding, k=k)

        if not results:
            return {
                "answer": "No relevant information found.",
                "sources": []
            }

        # 3. Build context from retrieved chunks
        context_parts = []
        sources = []

        for dist, meta in results:
            context_parts.append(f"[Chunk {meta['chunk_id']}]: {meta['text']}")

            if include_sources:
                sources.append({
                    "chunk_id": meta["chunk_id"],
                    "text": meta["text"],
                    "relevance_score": float(dist)
                })

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

        # 4. Build prompt
        prompt = self._build_qa_prompt(query, context)

        # 5. Generate answer
        answer = self.llm.generate(
            prompt,
            max_length=300,
            temperature=0.7
        )

        return {
            "answer": answer,
            "sources": sources if include_sources else None
        }

    def _build_qa_prompt(self, query: str, context: str) -> str:
        """Construct RAG prompt"""

        prompt = f"""You are a helpful AI tutor. Answer the question based on the provided context. If the context doesn't contain enough information, say so.

Context:
{context}

Question: {query}

Answer (be concise and accurate):"""

        return prompt

RAG Flow Diagram:

User Query: "What is supervised learning?"
    ↓
Encode Query → [0.23, -0.45, 0.67, ...]  (384-dim vector)
    ↓
Search FAISS Index
    ↓
Top-5 Similar Chunks:
    1. Chunk 12: "Supervised learning uses labeled data..." (distance: 0.15)
    2. Chunk 15: "Examples include classification..." (distance: 0.22)
    3. Chunk 8: "Training requires input-output pairs..." (distance: 0.28)
    4. Chunk 20: "Common algorithms are SVM, Random Forest..." (distance: 0.31)
    5. Chunk 3: "Machine learning has three types..." (distance: 0.35)
    ↓
Build Context:
    "[Chunk 12]: Supervised learning uses labeled data...
     [Chunk 15]: Examples include classification...
     ..."
    ↓
Build Prompt:
    "Context: [...]
     Question: What is supervised learning?
     Answer:"
    ↓
Generate with Phi-3:
    "Supervised learning is a type of machine learning that uses labeled
     data to train models. It involves learning from input-output pairs..."
    ↓
Return Answer + Sources

Why RAG?

  • Grounds answers in document content
  • Reduces hallucination
  • Provides source attribution
  • Works with any document without fine-tuning

Data Flow & Workflows

Complete Upload Workflow

┌─────────────────────────────────────────────────────────────┐
│                    1. Client Upload                          │
│ POST /upload with PDF file (multipart/form-data)            │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                  2. Validate & Save PDF                      │
│ - Check .pdf extension                                       │
│ - Generate doc_id (filename_timestamp)                       │
│ - Save to /data/uploads/{doc_id}.pdf                        │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│               3. Extract Text (PDFProcessor)                 │
│ - Try PyMuPDF first (fast)                                  │
│ - Fallback to pdfplumber if needed                          │
│ - Extract metadata (pages, author, title)                   │
│ Result: full_text string                                    │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                  4. Clean & Structure                        │
│ - Remove extra whitespace                                    │
│ - Normalize special characters                               │
│ - Detect headers and sections                                │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                    5. Chunk Text                             │
│ - Split into 250-word chunks                                 │
│ - 50-word overlap between chunks                             │
│ - Preserve sentence boundaries                               │
│ Result: List of {chunk_id, text, word_count}                │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│             6. Generate Embeddings                           │
│ - Extract chunk texts                                        │
│ - Batch encode with BAAI/bge-small                          │
│ - Normalize L2 distance                                      │
│ Result: numpy array (num_chunks, 384)                       │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│              7. Create Vector Index                          │
│ - Initialize FAISS IndexFlatL2                               │
│ - Add embeddings to index                                    │
│ - Store metadata (chunk_id, text, word_count)               │
│ - Save to /data/vectors/{doc_id}.index                      │
│ - Save metadata to {doc_id}.meta (pickle)                   │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│             8. Save Processed Document                       │
│ - Compile: doc_id, metadata, full_text, chunks, structure   │
│ - Save to /data/processed/{doc_id}.json                     │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                  9. Return Response                          │
│ {                                                             │
│   "message": "Document uploaded successfully",               │
│   "doc_id": "document_1732898765",                          │
│   "file_name": "ml_fundamentals.pdf",                       │
│   "pages": 42,                                               │
│   "chunks": 156                                              │
│ }                                                             │
└─────────────────────────────────────────────────────────────┘

Time Estimate: ~10-30 seconds for typical document

Configuration

Settings - utils/config.py

from pydantic_settings import BaseSettings
from pathlib import Path
import torch

class Settings(BaseSettings):
    # Paths
    base_dir: Path = Path(__file__).parent.parent.parent
    data_dir: Path = base_dir / "data"
    uploads_dir: Path = data_dir / "uploads"
    vectors_dir: Path = data_dir / "vectors"
    processed_dir: Path = data_dir / "processed"
    models_dir: Path = base_dir / "models"

    # Model names
    embedding_model: str = "BAAI/bge-small-en-v1.5"
    summarizer_model: str = "RandipR/pegasus-560m-academic-sum"
    quiz_model: str = "RandipR/Qwen2.5-0.5B-Instruct-MCQ-Generation"

    # Device
    device: str = "cuda" if torch.cuda.is_available() else "cpu"

    # Chunking
    chunk_size: int = 250  # words
    chunk_overlap: int = 50  # words
    min_chunk_size: int = 150
    max_chunk_size: int = 300

    # Vector search
    default_k: int = 5  # Top-k retrieval

    # Memory optimization
    use_4bit: bool = True
    clear_cache_freq: int = 10  # Every N operations

    # API
    cors_origins: list = ["*"]
    max_upload_size: int = 50 * 1024 * 1024  # 50MB

    class Config:
        env_file = ".env"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Create directories
        self.uploads_dir.mkdir(parents=True, exist_ok=True)
        self.vectors_dir.mkdir(parents=True, exist_ok=True)
        self.processed_dir.mkdir(parents=True, exist_ok=True)

settings = Settings()

Key Features Implementation

Feature 1: Offline Operation

How it works:

  1. Model Storage: All models downloaded to ~/models/ during first use
  2. HuggingFace Cache: Transformers library caches models locally
  3. No API Calls: Zero external dependencies during runtime
  4. Local Processing: All computation on user's machine

Implementation:

# Models check local cache first
model = AutoModelForSeq2SeqLM.from_pretrained(
    model_name,
    cache_dir=settings.models_dir,  # Local cache
    local_files_only=False  # Download if not cached
)

# After first download, works offline
model = AutoModelForSeq2SeqLM.from_pretrained(
    model_name,
    cache_dir=settings.models_dir,
    local_files_only=True  # Offline mode
)

Feature 2: Memory Optimization

Techniques Used:

  1. 4-bit Quantization
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",  # NormalFloat4
    bnb_4bit_compute_dtype=torch.float16
)
# Reduces memory by 75% (16-bit → 4-bit)
  1. FP16 Precision
model = AutoModelForSeq2SeqLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16  # Half precision
)
# Reduces memory by 50% vs FP32
  1. Cache Clearing
if batch_idx % 10 == 0:
    torch.cuda.empty_cache()
  1. Lazy Loading
# Models only load on first use
@lru_cache(maxsize=1)
def get_embedding_model():
    return EmbeddingModel(settings.embedding_model, settings.device)

Memory Usage:

  • Embeddings (BAAI): ~130 MB
  • Summarizer (Pegasus): ~1.2 GB (4-bit) or ~2.4 GB (FP16)
  • Quiz Generator (Qwen): ~500 MB (4-bit) or ~1 GB (FP16)
  • FAISS Index: ~1-5 MB per document
  • Total: ~5-6 GB GPU / ~8-10 GB system RAM

Feature 3: Robust Error Handling

Patterns Used:

  1. Try-Except with Fallbacks
try:
    text, metadata = self.extract_text_pymupdf(pdf_path)
except Exception as e:
    logger.warning(f"PyMuPDF failed: {e}")
    text = self.extract_text_pdfplumber(pdf_path)
    metadata = self._default_metadata()
  1. HTTPException for API Errors
if not file.filename.endswith('.pdf'):
    raise HTTPException(
        status_code=400,
        detail="Only PDF files are allowed"
    )
  1. Validation with Pydantic
class SummarizeRequest(BaseModel):
    doc_id: str
    summary_style: Literal["bullet", "paragraph", "exam"] = "paragraph"
    chunk_wise: bool = False

    @validator('doc_id')
    def doc_id_not_empty(cls, v):
        if not v.strip():
            raise ValueError("doc_id cannot be empty")
        return v

Code Deep Dive

Example 1: How RAG Q&A Works (End-to-End)

User asks: "What are the applications of machine learning?"

Step 1: Encode query

# In backend/models/embeddings.py
query = "What are the applications of machine learning?"
query_embedding = embedding_model.encode_query(query)
# Shape: (384,)
# Values: [0.234, -0.456, 0.789, ...]

Step 2: Search vector store

# In backend/services/vector_store.py
results = vector_store.search(query_embedding, k=5)
# Returns: [(distance, metadata), ...]
# [
#   (0.15, {"chunk_id": 42, "text": "ML applications include..."}),
#   (0.22, {"chunk_id": 38, "text": "Common uses are..."}),
#   ...
# ]

Step 3: Build context

# In backend/services/rag_pipeline.py
context_parts = []
for dist, meta in results:
    context_parts.append(f"[Chunk {meta['chunk_id']}]: {meta['text']}")

context = "\n\n".join(context_parts)
# Context is now:
# "[Chunk 42]: ML applications include healthcare, finance...
#  [Chunk 38]: Common uses are image recognition...
#  ..."

Step 4: Create prompt

prompt = f"""You are a helpful AI tutor. Answer the question based on the provided context.

Context:
{context}

Question: {query}

Answer (be concise and accurate):"""

Step 5: Generate answer

# In backend/models/phi3_summarizer.py
inputs = tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True)
outputs = model.generate(
    **inputs,
    max_length=300,
    num_beams=4,
    temperature=0.7
)
answer = tokenizer.decode(outputs[0], skip_special_tokens=True)
# Answer: "Machine learning has numerous applications including healthcare (disease diagnosis),
#          finance (fraud detection), image recognition, natural language processing..."

Step 6: Return response

return ChatResponse(
    answer=answer,
    sources=[
        {"chunk_id": 42, "text": "ML applications include...", "relevance_score": 0.15},
        {"chunk_id": 38, "text": "Common uses are...", "relevance_score": 0.22}
    ],
    doc_id=doc_id
)

Example 2: How Summarization Works

User requests: "Summarize this document in bullet points"

Chunk-wise Approach:

Step 1: Load document

with open(f"/data/processed/{doc_id}.json", "r") as f:
    doc_data = json.load(f)

chunks = doc_data["chunks"]  # 156 chunks

Step 2: Summarize each chunk

chunk_summaries = []

for chunk in chunks:
    # Add style instruction
    text = f"Summarize in bullet points:\n{chunk['text']}"

    # Tokenize
    inputs = tokenizer(text, max_length=1024, truncation=True, return_tensors="pt")

    # Generate summary
    outputs = model.generate(
        **inputs,
        max_length=100,
        num_beams=4,
        early_stopping=True
    )

    summary = tokenizer.decode(outputs[0], skip_special_tokens=True)
    chunk_summaries.append(summary)

Step 3: Combine summaries

combined = "\n\n".join(chunk_summaries)
# Too long? Summarize the summaries
if len(combined.split()) > 500:
    final_summary = phi3_summarizer.summarize(
        text=combined,
        style="bullet",
        max_length=300
    )
else:
    final_summary = combined

Step 4: Format as bullets

# Model output might be:
# "Machine learning is a subset of AI. It has many applications. ..."

# Post-process to bullets:
lines = final_summary.split('. ')
bullets = [f"• {line.strip()}" for line in lines if line.strip()]
final_summary = "\n".join(bullets)

# Final output:
# • Machine learning is a subset of AI
# • It has many applications in healthcare and finance
# • Common algorithms include decision trees and neural networks

Example 3: Quiz Generation Process

Step 1: Sample chunks

# Don't process all 156 chunks - sample 10
num_chunks = (10 questions // 2 per chunk) + 2 = 7
sampled = random.sample(chunks, 7)

Step 2: Generate questions per chunk

all_questions = []

for chunk in sampled:
    # Build prompt
    prompt = f"""Generate 2 multiple-choice questions from this text:

{chunk['text']}

Format:
Question 1: [text]
A) [option]
B) [option]
C) [option]
D) [option]
Correct Answer: A

Begin:"""

    # Generate
    inputs = tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True)
    outputs = model.generate(
        **inputs,
        max_new_tokens=512,
        temperature=0.7,
        do_sample=True
    )

    generated = tokenizer.decode(outputs[0], skip_special_tokens=True)

    # Parse MCQs
    questions = parse_mcqs(generated)  # Extracts structured questions
    all_questions.extend(questions)

Step 3: Clean and validate

cleaned = []

for q in all_questions:
    # Remove artifacts
    q_clean = QuizCleaner.clean_question(q)

    # Validate
    if (
        len(q_clean['options']) == 4 and
        0 <= q_clean['correct_answer'] < 4 and
        len(q_clean['question']) > 10
    ):
        cleaned.append(q_clean)

# Limit to requested number
final = cleaned[:10]

Summary

The DocuMentor backend is a sophisticated, privacy-focused AI system that:

  1. Processes PDFs - Extracts, chunks, and indexes documents
  2. Generates Embeddings - Creates semantic vectors for search
  3. Provides RAG Q&A - Answers questions grounded in document content
  4. Summarizes Documents - Creates bullet, paragraph, or exam-style summaries
  5. Generates Quizzes - Automatically creates MCQs from content

All running 100% offline with efficient memory usage and robust error handling.

Key Technologies:

  • FastAPI for REST API
  • PyTorch + Transformers for ML
  • FAISS for vector search
  • Specialized fine-tuned models for academic text

Performance:

  • Upload: 10-30 seconds
  • Summarize: 5-15 seconds
  • Q&A: 2-5 seconds
  • Quiz: 15-30 seconds

Memory:

  • 5-6 GB GPU recommended
  • Can run on CPU (slower)
  • 4-bit quantization reduces memory by 75%