Sync backend Docker context from GitHub main

data/data_loader.py

@@ -0,0 +1,91 @@
+import re
+import pandas as pd
+from typing import List, Dict, Any
+
+
+def load_cbt_book(file_path: str = "EntireBookCleaned.txt") -> pd.DataFrame:
+    """
+    Loads the CBT book from a text file and parses it into documents.
+    Each page is treated as a separate document.
+    
+    Args:
+        file_path: Path to the cleaned book text file
+        
+    Returns:
+        DataFrame with columns: id, title, url, full_text
+    """
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            content = f.read()
+    except FileNotFoundError:
+        raise FileNotFoundError(f"Book file not found: {file_path}")
+    
+    # Split content by page markers
+    # Pattern matches "--- Page X ---" or "--- Page X of Y ---"
+    page_pattern = r'---\s*Page\s+(\d+)(?:\s+of\s+\d+)?\s*---'
+    
+    # Split the content into pages
+    pages = re.split(page_pattern, content)
+    
+    # pages[0] is content before first page marker (usually empty)
+    # Then alternating: page_number, page_content, page_number, page_content...
+    
+    documents = []
+    i = 1  # Start from first page number
+    
+    while i < len(pages) - 1:
+        page_num = pages[i].strip()
+        page_content = pages[i + 1].strip() if i + 1 < len(pages) else ""
+        
+        # Clean up the content - remove excessive whitespace
+        page_content = re.sub(r'\n{3,}', '\n\n', page_content)
+        page_content = page_content.strip()
+        
+        if page_content:  # Only add non-empty pages
+            # Extract a title from the first line if possible
+            lines = page_content.split('\n')
+            title_line = lines[0].strip() if lines else f"Page {page_num}"
+            
+            # Use first meaningful line as title, or default to page number
+            if len(title_line) > 10 and len(title_line) < 200:
+                title = title_line
+            else:
+                title = f"CBT Book - Page {page_num}"
+            
+            documents.append({
+                "id": f"cbt-page-{page_num}",
+                "title": title,
+                "url": f"https://res.cloudinary.com/dajb4c1g5/image/upload/v1774864993/topic_pdfs/93/merged_pdf_1774864989649.pdf.pdf#page={page_num}",
+                "full_text": page_content
+            })
+        
+        i += 2  # Move to next page number
+    
+    if not documents:
+        raise ValueError("No documents were parsed from the book file")
+    
+    df = pd.DataFrame(documents)
+    print(f"Loaded {len(df)} pages from CBT book")
+    return df
+
+
+def get_book_stats(df: pd.DataFrame) -> Dict[str, Any]:
+    """
+    Get statistics about the loaded book.
+    
+    Args:
+        df: DataFrame containing book pages
+        
+    Returns:
+        Dictionary with statistics
+    """
+    total_chars = df['full_text'].str.len().sum()
+    avg_chars = df['full_text'].str.len().mean()
+    
+    return {
+        "total_pages": len(df),
+        "total_characters": total_chars,
+        "average_chars_per_page": round(avg_chars, 2),
+        "min_chars": df['full_text'].str.len().min(),
+        "max_chars": df['full_text'].str.len().max()
+    }

data/ingest.py

@@ -0,0 +1,310 @@
+"""
+Script to ingest CBT book data into Pinecone vector database.
+Ingests the book 6 times with different chunking formats for ablation study.
+All chunks are stored in a SINGLE index with metadata to differentiate.
+Run this once before starting the API server.
+"""
+import os
+import time
+from dotenv import load_dotenv
+from config_loader import cfg
+from data.data_loader import load_cbt_book, get_book_stats
+from data.vector_db import get_pinecone_index, refresh_pinecone_index
+from retriever.processor import ChunkProcessor
+
+
+# 6 different chunking techniques for ablation study
+CHUNKING_TECHNIQUES = [
+    {
+        "name": "fixed",
+        "description": "Fixed-size chunking - splits every N characters (may cut sentences mid-way)",
+        "chunk_size": 1000,
+        "chunk_overlap": 100,
+        "kwargs": {"separator": ""},  # No separator for fixed splitting
+    },
+    {
+        "name": "sentence",
+        "description": "Sentence-level chunking - respects sentence boundaries (NLTK)",
+        "chunk_size": 1000,
+        "chunk_overlap": 100,
+        "kwargs": {},
+    },
+    {
+        "name": "paragraph",
+        "description": "Paragraph-level chunking - uses natural paragraph breaks",
+        "chunk_size": 2500,
+        "chunk_overlap": 100,
+        "kwargs": {"separator": "\n\n"},  # Split on paragraph breaks
+    },
+    # {
+    #     "name": "semantic",
+    #     "description": "Semantic chunking - splits where topic/meaning shifts (embedding similarity)",
+    #     "chunk_size": 2000,
+    #     "chunk_overlap": 100,
+    #     "kwargs": {"breakpoint_threshold_type": "percentile", "breakpoint_threshold_amount": 70},
+    # },
+    {
+        "name": "recursive",
+        "description": "Recursive chunking - hierarchical splitting (paragraphs → sentences → words → chars)",
+        "chunk_size": 2000,
+        "chunk_overlap": 100,
+        "kwargs": {"separators": ["\n\n", "\n", ". ", "! ", "? ", "; ", ", ", " ", ""], "keep_separator": True},
+    },
+    {
+        "name": "page",
+        "description": "Page-level chunking - uses entire book pages as-is",
+        "chunk_size": 10000,  # Very large to keep full pages
+        "chunk_overlap": 0,   # No overlap between pages
+        "kwargs": {"separator": "--- Page"},  # Split on page markers
+    },
+    {
+        "name": "markdown",
+        "description": "Markdown header chunking - splits by headers (#, ##, ###, ####) with 4k char limit",
+        "chunk_size": 4000,  # Max 4k chars per chunk
+        "chunk_overlap": 0,  # No overlap for markdown
+        "kwargs": {},  # Custom implementation
+    },
+]
+
+
+def ingest_single_technique(
+    raw_data,
+    proc,
+    technique_config,
+    technique_index,
+    total_techniques,
+):
+    """Chunk the book using a single technique and return chunks with metadata."""
+    technique_name = technique_config["name"]
+    chunk_size = technique_config["chunk_size"]
+    chunk_overlap = technique_config["chunk_overlap"]
+    kwargs = technique_config.get("kwargs", {})
+
+    print(f"\n[{technique_index}/{total_techniques}] Processing '{technique_name}'...")
+    print(f"  Description: {technique_config['description']}")
+    print(f"  Chunk size: {chunk_size}, Overlap: {chunk_overlap}")
+
+    # Chunk and embed
+    final_chunks = proc.process(
+        raw_data,
+        technique=technique_name,
+        chunk_size=chunk_size,
+        chunk_overlap=chunk_overlap,
+        max_docs=cfg.project.get("doc_limit"),
+        verbose=False,
+        **kwargs,
+    )
+
+    # Add technique metadata to each chunk for differentiation
+    # Prefix ID with technique name to ensure uniqueness across techniques
+    for chunk in final_chunks:
+        chunk["metadata"]["chunking_technique"] = technique_name
+        chunk["id"] = f"{technique_name}-{chunk['id']}"
+
+    print(f"  Created {len(final_chunks)} chunks")
+
+    return final_chunks
+
+
+def ingest_data():
+    """Load CBT book, chunk it 6 ways, and upload ALL to a SINGLE Pinecone index.
+    
+    Returns:
+        Tuple of (all_chunks, configured_technique_chunks, processor) for reuse in retrieval pipeline.
+    """
+    load_dotenv()
+
+    pinecone_key = os.getenv("PINECONE_API_KEY")
+    if not pinecone_key:
+        raise RuntimeError("PINECONE_API_KEY not found in environment variables")
+
+    print("=" * 80)
+    print("CBT BOOK INGESTION PIPELINE - 6 TECHNIQUES (SINGLE INDEX)")
+    print("=" * 80)
+    print(f"\nTechniques to process: {len(CHUNKING_TECHNIQUES)}")
+    for i, tech in enumerate(CHUNKING_TECHNIQUES, 1):
+        print(f"  {i}. {tech['name']}: {tech['description']}")
+    print(f"\nAll chunks will be stored in a SINGLE index: {cfg.db['base_index_name']}-{cfg.processing['technique']}")
+    print("Chunks are differentiated by 'chunking_technique' metadata field.")
+
+    # 1. Load the CBT book (once, reused for all techniques)
+    print(f"\n{'='*80}")
+    print("STEP 1: LOADING CBT BOOK")
+    print(f"{'='*80}")
+    print("\nLoading CBT book from EntireBookCleaned.txt...")
+    raw_data = load_cbt_book("data/EntireBookCleaned.txt")
+    stats = get_book_stats(raw_data)
+    print(f"  Loaded {stats['total_pages']} pages")
+    print(f"  Total characters: {stats['total_characters']:,}")
+    print(f"  Average chars per page: {stats['average_chars_per_page']:.0f}")
+
+    # 2. Initialize processor (once, reused for all techniques)
+    print(f"\nInitializing embedding model: {cfg.processing['embedding_model']}")
+    proc = ChunkProcessor(model_name=cfg.processing['embedding_model'], verbose=False)
+
+    # 3. Process each technique sequentially and collect all chunks
+    print(f"\n{'='*80}")
+    print("STEP 2: CHUNKING WITH 6 TECHNIQUES")
+    print(f"{'='*80}")
+
+    all_chunks = []
+    configured_technique_chunks = []
+    results = {}
+
+    for i, technique in enumerate(CHUNKING_TECHNIQUES, 1):
+        try:
+            chunks = ingest_single_technique(
+                raw_data=raw_data,
+                proc=proc,
+                technique_config=technique,
+                technique_index=i,
+                total_techniques=len(CHUNKING_TECHNIQUES),
+            )
+            all_chunks.extend(chunks)
+            
+            # Save chunks for the configured technique (for retrieval pipeline)
+            if technique["name"] == cfg.processing['technique']:
+                configured_technique_chunks = chunks
+            
+            results[technique["name"]] = {
+                "status": "success",
+                "chunks": len(chunks),
+            }
+
+            # Wait between techniques to avoid rate limits (for embedding API)
+            if i < len(CHUNKING_TECHNIQUES):
+                print(f"  Waiting 5 seconds before next technique (rate limit protection)...")
+                import time
+                time.sleep(5)
+
+        except Exception as e:
+            print(f"  ERROR with technique '{technique['name']}': {e}")
+            results[technique["name"]] = {
+                "status": "failed",
+                "error": str(e),
+            }
+
+    # 4. Upload ALL chunks to a SINGLE Pinecone index
+    print(f"\n{'='*80}")
+    print("STEP 3: UPLOADING TO SINGLE PINECONE INDEX")
+    print(f"{'='*80}")
+
+    index_name = f"{cfg.db['base_index_name']}-{cfg.processing['technique']}"
+    print(f"\nIndex name: {index_name}")
+    print(f"Dimension: {cfg.db['dimension']}")
+    print(f"Metric: {cfg.db['metric']}")
+    print(f"Total chunks to upload: {len(all_chunks)}")
+
+    index = get_pinecone_index(
+        pinecone_key,
+        cfg.db['base_index_name'],
+        technique=cfg.processing['technique'],
+        dimension=cfg.db['dimension'],
+        metric=cfg.db['metric'],
+    )
+
+    print("Uploading " + str(len(all_chunks)) + " vectors to Pinecone...")
+    refresh_pinecone_index(index, all_chunks, batch_size=cfg.db['batch_size'])
+
+    # Upload sparse vectors to a separate index
+
+    print("Preparing to upload sparse vectors for BM25...")
+    try:
+        from pinecone import Pinecone, ServerlessSpec
+        try:
+            from pinecone_text.sparse import BM25Encoder
+        except ImportError:
+            print("Skipping BM25 indexing - run pip install pinecone-text")
+            return all_chunks, configured_technique_chunks, proc, index
+        pc = Pinecone(api_key=pinecone_key)
+        
+        sparse_index_name = "cbt-book-sparse"
+        existing_indexes = [idx.name for idx in pc.list_indexes()]
+        if sparse_index_name not in existing_indexes:
+            print(f"Creating sparse index: {sparse_index_name}")
+            pc.create_index(
+                name=sparse_index_name,
+                dimension=512,  # required space-filler dimension
+                metric="dotproduct",
+                spec=ServerlessSpec(cloud="aws", region="us-east-1")
+            )
+            # wait for index
+            import time
+            while not pc.describe_index(sparse_index_name).status["ready"]:
+                time.sleep(1)
+            
+        sparse_index = pc.Index(sparse_index_name)
+        
+        # Encode sparse vectors
+        print("Encoding sparse vectors...")
+        bm25 = BM25Encoder().default()
+        sparse_chunks = []
+        
+        # Learn BM25
+        corpus = [chunk["metadata"]["text"] for chunk in all_chunks]
+        bm25.fit(corpus)
+        
+        for chunk in all_chunks:
+            sparse_values = bm25.encode_documents(chunk["metadata"]["text"])
+            
+            # Skip empty sparse vectors to prevent Pinecone errors
+            if not sparse_values.get("indices") or len(sparse_values.get("indices", [])) == 0:
+                continue
+                
+            new_chunk = {
+                "id": chunk["id"],
+                
+                "sparse_values": sparse_values,
+                "metadata": chunk["metadata"]
+            }
+            sparse_chunks.append(new_chunk)
+            
+        print(f"Upserting {len(sparse_chunks)} valid sparse vectors to {sparse_index_name}...")
+        
+        # Upsert sparse vectors
+        if sparse_chunks:
+            batch_size = cfg.db.get("batch_size", 100)
+            for i in range(0, len(sparse_chunks), batch_size):
+                batch = sparse_chunks[i:i+batch_size]
+                sparse_index.upsert(vectors=batch)
+            print("Sparse vector upsert complete.")
+        else:
+            print("No valid sparse vectors to upsert.")
+            
+    except Exception as e:
+        print(f"Error during sparse vector upload: {e}")
+
+    # 5. Summary
+    print(f"\n{'='*80}")
+    print("INGESTION COMPLETE - SUMMARY")
+    print(f"{'='*80}")
+    print(f"\n{'Technique':<15} {'Status':<12} {'Chunks':<10}")
+    print("-" * 40)
+    total_chunks = 0
+    for tech in CHUNKING_TECHNIQUES:
+        name = tech["name"]
+        result = results.get(name, {})
+        status = result.get("status", "unknown")
+        chunks = result.get("chunks", 0)
+        if status == "success":
+            total_chunks += chunks
+        print(f"{name:<15} {status:<12} {chunks:<10}")
+    print("-" * 40)
+    print(f"{'TOTAL':<15} {'':<12} {total_chunks:<10}")
+
+    print(f"\nSingle index: {index_name}")
+    print(f"Total vectors: {len(all_chunks)}")
+    print("\nChunks can be filtered by 'chunking_technique' metadata field:")
+    for tech in CHUNKING_TECHNIQUES:
+        if results.get(tech["name"], {}).get("status") == "success":
+            print(f"  - chunking_technique: '{tech['name']}'")
+
+    print("\nYou can now start the API server with:")
+    print("  python -m uvicorn api:app --host 0.0.0.0 --port 8000")
+    
+    # Return chunks and processor for reuse in retrieval pipeline
+    return all_chunks, configured_technique_chunks, proc, index
+
+
+if __name__ == "__main__":
+    ingest_data()

data/vector_db.py

@@ -0,0 +1,247 @@
+import time
+import re
+import json
+from pathlib import Path
+from typing import Any, Dict, List
+from pinecone import Pinecone, ServerlessSpec
+
+
+# Added cacheing to reduce consecutive startup time
+# --@Qamar
+
+def slugify_technique(name):
+    """Converts 'Sentence Splitter' to 'sentence-splitter' for Pinecone naming."""
+    return re.sub(r'[^a-z0-9]+', '-', name.lower()).strip('-')
+
+def get_index_by_name(api_key: str, index_name: str):
+    """
+    Directly connects to a Pinecone index by its full string name.
+    Useful for the API/Production side where the name is already known.
+    """
+    pc = Pinecone(api_key=api_key)
+    
+    # Check if it exists first to avoid a 404 crash
+    existing_indexes = [idx.name for idx in pc.list_indexes()]
+    if index_name not in existing_indexes:
+        raise ValueError(f"Index '{index_name}' does not exist in your Pinecone project.")
+    
+    print(f" Connecting to Index: {index_name}")
+    return pc.Index(index_name)
+
+def get_pinecone_index(api_key, base_name, technique, dimension=384, metric="cosine"):
+    """
+    Creates/Returns an index specifically for a technique.
+    Example: 'arxiv-index-token'
+    """
+    pc = Pinecone(api_key=api_key)
+    tech_slug = slugify_technique(technique)
+    full_index_name = f"{base_name}-{tech_slug}"
+    
+    existing_indexes = [idx.name for idx in pc.list_indexes()]
+    
+    if full_index_name not in existing_indexes:
+        print(f" Creating specialized index: {full_index_name}...")
+        pc.create_index(
+            name=full_index_name,
+            dimension=dimension,
+            metric=metric,
+            spec=ServerlessSpec(cloud="aws", region="us-east-1")
+        )
+        # Wait for index to spin up
+        while not pc.describe_index(full_index_name).status['ready']:
+            time.sleep(1)
+    
+    # Use our new helper to return the index object
+    return get_index_by_name(api_key, full_index_name)
+
+def refresh_pinecone_index(index, final_chunks, batch_size=100):
+    """
+    Refreshes the specific index. Since index is now technique-specific,
+    we just check if it's already populated.
+    """
+    if not final_chunks:
+        print("No chunks provided to refresh.")
+        return False
+
+    try:
+        # Check current stats for this specific index
+        stats = index.describe_index_stats()
+        current_count = stats.get('total_vector_count', 0)
+        expected_count = len(final_chunks)
+
+        print(f" Index Stats -> Existing: {current_count} | New Chunks: {expected_count}")
+
+        if current_count == 0:
+            print(f"➕ Index is empty. Upserting {expected_count} vectors...")
+            vectors = prepare_vectors_for_upsert(final_chunks)
+            upsert_to_pinecone(index, vectors, batch_size)
+            return True
+        
+        elif current_count < expected_count:
+            # Simple check to see if we need to top up or refresh
+            print(f" Vector count mismatch ({current_count} < {expected_count}). Updating index...")
+            vectors = prepare_vectors_for_upsert(final_chunks)
+            upsert_to_pinecone(index, vectors, batch_size)
+            return True
+
+        else:
+            print(f" Index is already populated with {current_count} vectors. Ready for search.")
+            return False
+            
+    except Exception as e:
+        print(f" Error refreshing index: {e}")
+        return False
+
+# Utility functions remain the same as previous version
+def prepare_vectors_for_upsert(final_chunks):
+    vectors = []
+    for chunk in final_chunks:
+        meta = chunk.get('metadata', {})
+        vector = {
+            'id': chunk['id'],
+            'values': chunk['values'],
+            'metadata': {
+                'text': meta.get('text', ""),
+                'title': meta.get('title', ""),
+                'url': meta.get('url', ""),
+                'chunk_index': meta.get('chunk_index', 0),
+                'technique': meta.get('technique', "unknown"),
+                'chunking_technique': meta.get('chunking_technique', "unknown")
+            }
+        }
+        #if 'sparse_values' in chunk:
+            #vector['sparse_values'] = chunk['sparse_values']
+        vectors.append(vector)
+    return vectors
+
+def upsert_to_pinecone(index, chunks, batch_size=100):
+    for i in range(0, len(chunks), batch_size):
+        batch = chunks[i : i + batch_size]
+        index.upsert(vectors=batch)
+
+# Some methods for loading chunks back from Pinecone with local caching to speed up BM25 initialization
+
+def _sanitize_index_name(index_name: str) -> str:
+    return re.sub(r'[^a-zA-Z0-9._-]+', '-', index_name).strip('-') or 'default-index'
+
+
+def _chunk_cache_path(cache_dir: str, index_name: str) -> Path:
+    cache_root = Path(cache_dir)
+    cache_root.mkdir(parents=True, exist_ok=True)
+    safe_name = _sanitize_index_name(index_name)
+    return cache_root / f"bm25_chunks_{safe_name}.json"
+
+
+def _read_chunk_cache(path: Path) -> Dict[str, Any]:
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _write_chunk_cache(path: Path, payload: Dict[str, Any]) -> None:
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(payload, f)
+
+
+def load_chunks_with_local_cache(
+    index,
+    index_name: str,
+    cache_dir: str = ".cache",
+    batch_size: int = 100,
+    force_refresh: bool = False,
+) -> tuple[List[Dict[str, Any]], str]:
+
+    cache_file = _chunk_cache_path(cache_dir=cache_dir, index_name=index_name)
+    stats = index.describe_index_stats()
+    current_count = stats.get("total_vector_count", 0)
+
+    if not force_refresh and cache_file.exists():
+        try:
+            cached_payload = _read_chunk_cache(cache_file)
+            cached_meta = cached_payload.get("meta", {})
+            cached_count = cached_meta.get("vector_count", -1)
+            cached_chunks = cached_payload.get("chunks", [])
+
+            if cached_count == current_count and cached_chunks:
+                print(
+                    f" Loaded BM25 chunk cache: {cache_file} "
+                    f"(chunks={len(cached_chunks)}, vectors={cached_count})"
+                )
+                return cached_chunks, "cache"
+
+            print(
+                " BM25 cache stale or empty. "
+                f"cache_vectors={cached_count}, pinecone_vectors={current_count}. Refreshing..."
+            )
+        except Exception as e:
+            print(f" Failed to read BM25 cache ({cache_file}): {e}. Refreshing from Pinecone...")
+
+    chunks = load_chunks_from_pinecone(index=index, batch_size=batch_size)
+    payload = {
+        "meta": {
+            "index_name": index_name,
+            "vector_count": current_count,
+            "updated_at_epoch_s": int(time.time()),
+        },
+        "chunks": chunks,
+    }
+
+    try:
+        _write_chunk_cache(cache_file, payload)
+        print(f" Saved BM25 chunk cache: {cache_file} (chunks={len(chunks)})")
+    except Exception as e:
+        print(f" Failed to write BM25 cache ({cache_file}): {e}")
+
+    return chunks, "pinecone"
+
+
+def load_chunks_from_pinecone(index, batch_size: int = 100) -> list[dict[str, any]]:
+    """
+    Scans the Pinecone index to retrieve all text metadata for the BM25 corpus.
+    """
+    stats = index.describe_index_stats()
+    namespaces = list(stats.get('namespaces', {}).keys())
+    # If no namespaces are explicitly named, Pinecone uses an empty string for the default
+    if not namespaces:
+        namespaces = [""]
+
+    all_chunks: List[Dict[str, Any]] = []
+    seen_ids = set()
+
+    print(f"Loading vectors for BM25 from namespaces: {namespaces}")
+
+    for ns in namespaces:
+        # Pinecone's list() generator returns batches of IDs
+        for id_batch in index.list(namespace=ns, limit=batch_size):
+            if not id_batch:
+                continue
+
+            # Fetch the actual content (metadata) for this batch of IDs
+            fetched = index.fetch(ids=id_batch, namespace=ns)
+            vectors = getattr(fetched, "vectors", {})
+
+            for vector_id, vector_data in vectors.items():
+                if vector_id in seen_ids:
+                    continue
+                seen_ids.add(vector_id)
+
+                # Safely extract metadata
+                metadata = getattr(vector_data, "metadata", {})
+                if metadata is None:
+                    metadata = {}
+                if not isinstance(metadata, dict):
+                    metadata = dict(metadata)
+
+                text = metadata.get("text")
+                
+                if not text:
+                    continue
+
+                all_chunks.append({
+                    "id": vector_id,
+                    "metadata": metadata
+                })
+
+        print(f" Finished namespace: '{ns if ns else 'default'}'")
+
+    print(f"Total chunks loaded into memory: {len(all_chunks)}")
+    return all_chunks