Upload folder using huggingface_hub

src/rag_engine/__init__.py

@@ -0,0 +1,41 @@
+"""RAG Engine: A clear, educational Retrieval-Augmented Generation system.
+
+Designed to illustrate how a production-grade RAG pipeline works:
+- Ingestion & Text Extraction (PDF / Text)
+- Semantic Text Chunking
+- Dense Vector Embeddings (with offline Hash fallback)
+- Sparse BM25 Keyword Search
+- Hybrid Retrieval with Reciprocal Rank Fusion (RRF)
+- Augmented Generation with Source Citations
+- Faithfulness and Relevancy Evaluation
+"""
+
+from rag_engine.documents import Document, load_text_file, load_pdf_file
+from rag_engine.chunking import TextChunk, chunk_document, chunk_documents
+from rag_engine.embeddings import HashEmbeddings, get_embedding_model
+from rag_engine.vector_store import VectorStore
+from rag_engine.bm25_search import BM25SearchIndex
+from rag_engine.hybrid_search import HybridRetriever, reciprocal_rank_fusion
+from rag_engine.generator import generate_answer
+from rag_engine.eval import evaluate_faithfulness, evaluate_relevancy, evaluate_response
+from rag_engine.pipeline import RAGPipeline
+
+__all__ = [
+    "Document",
+    "load_text_file",
+    "load_pdf_file",
+    "TextChunk",
+    "chunk_document",
+    "chunk_documents",
+    "HashEmbeddings",
+    "get_embedding_model",
+    "VectorStore",
+    "BM25SearchIndex",
+    "HybridRetriever",
+    "reciprocal_rank_fusion",
+    "generate_answer",
+    "evaluate_faithfulness",
+    "evaluate_relevancy",
+    "evaluate_response",
+    "RAGPipeline",
+]

src/rag_engine/bm25_search.py

@@ -0,0 +1,133 @@
+"""Sparse keyword search using BM25 (Best Matching 25).
+
+Implemented cleanly in pure Python with optional rank_bm25 fallback:
+Zero external dependencies required!
+
+Why BM25?
+While vector embeddings capture semantic meaning (synonyms, conceptual similarity),
+BM25 captures exact lexical matches (rare keywords, acronyms, specific numbers).
+"""
+
+import math
+import re
+from collections import Counter
+
+TOKEN_RE = re.compile(r"[A-Za-z0-9_\-]+")
+
+
+def tokenize(text):
+    """Clean and tokenize text into lowercase words/terms."""
+    if not text:
+        return []
+    return [t.lower() for t in TOKEN_RE.findall(text)]
+
+
+class SimpleBM25:
+    """Pure Python implementation of Okapi BM25.
+    
+    Formula:
+      IDF(q) = ln((N - n(q) + 0.5) / (n(q) + 0.5) + 1)
+      Score(D, Q) = sum( IDF(q) * (f(q, D) * (k1 + 1)) / (f(q, D) + k1 * (1 - b + b * (|D| / avgdl))) )
+    """
+
+    def __init__(self, corpus, k1=1.5, b=0.75):
+        self.k1 = k1
+        self.b = b
+        self.corpus_size = len(corpus)
+        self.doc_lengths = [len(doc) for doc in corpus]
+        self.avg_doc_len = sum(self.doc_lengths) / max(1, self.corpus_size)
+
+        # Document frequencies for each term
+        self.doc_freqs = Counter()
+        self.doc_term_counts = []
+
+        for doc in corpus:
+            counts = Counter(doc)
+            self.doc_term_counts.append(counts)
+            for term in counts.keys():
+                self.doc_freqs[term] += 1
+
+    def get_scores(self, query_tokens):
+        """Calculate BM25 relevance score for each document in the corpus."""
+        scores = [0.0] * self.corpus_size
+        for token in query_tokens:
+            if token not in self.doc_freqs:
+                continue
+
+            # Inverse Document Frequency (IDF)
+            df = self.doc_freqs[token]
+            idf = math.log((self.corpus_size - df + 0.5) / (df + 0.5) + 1.0)
+
+            # Term frequency & length normalization for each doc
+            for doc_idx, term_counts in enumerate(self.doc_term_counts):
+                tf = term_counts.get(token, 0)
+                if tf == 0:
+                    continue
+                doc_len = self.doc_lengths[doc_idx]
+                numerator = tf * (self.k1 + 1.0)
+                denominator = tf + self.k1 * (1.0 - self.b + self.b * (doc_len / max(1.0, self.avg_doc_len)))
+                scores[doc_idx] += idf * (numerator / denominator)
+
+        return scores
+
+
+class BM25SearchIndex:
+    """Keyword search index over chunks."""
+
+    def __init__(self):
+        self.chunks = []
+        self.corpus_tokens = []
+        self.bm25 = None
+
+    def build(self, chunks):
+        """Index a list of TextChunk objects."""
+        if not chunks:
+            self.chunks = []
+            self.corpus_tokens = []
+            self.bm25 = None
+            return
+
+        self.chunks = chunks
+        self.corpus_tokens = []
+        for chunk in chunks:
+            # Include text and document title for matching
+            combined_text = f"{chunk.title} {chunk.text}"
+            self.corpus_tokens.append(tokenize(combined_text))
+
+        try:
+            from rank_bm25 import BM25Okapi
+            self.bm25 = BM25Okapi(self.corpus_tokens)
+        except ImportError:
+            # Use pure Python BM25 fallback
+            self.bm25 = SimpleBM25(self.corpus_tokens)
+
+    def search(self, query, top_k=5):
+        """Find the top_k best matching chunks by BM25 keyword relevance.
+        
+        Returns:
+            List of dicts: [{"chunk": TextChunk, "score": float, "source": "bm25"}]
+        """
+        if not self.bm25 or not self.chunks:
+            return []
+
+        query_tokens = tokenize(query)
+        if not query_tokens:
+            return []
+
+        scores = self.bm25.get_scores(query_tokens)
+        scored = list(enumerate(scores))
+        scored.sort(key=lambda x: x[1], reverse=True)
+
+        results = []
+        for idx, score in scored[:top_k]:
+            if score <= 0:
+                continue
+            results.append({
+                "chunk": self.chunks[idx],
+                "score": round(float(score), 4),
+                "source": "bm25",
+            })
+        return results
+
+    def __len__(self):
+        return len(self.chunks)

src/rag_engine/chunking.py

@@ -0,0 +1,117 @@
+"""Text chunking strategies for the RAG engine.
+
+Splitting text into smaller chunks is essential in RAG because:
+1. Embedding models have context length limits (typically 256-512 tokens).
+2. Small chunks provide high-precision retrieval matching.
+3. Overlap between chunks ensures ideas aren't cut in half at boundaries.
+"""
+
+import re
+
+
+class TextChunk:
+    """Represents a chunk of text derived from a source document."""
+
+    def __init__(self, chunk_id, doc_id, text, title=None, chunk_index=0, parent_text=None):
+        self.chunk_id = chunk_id
+        self.doc_id = doc_id
+        self.text = text.strip()
+        self.title = title or ""
+        self.chunk_index = chunk_index
+        # Parent text is the broader surrounding context (optional parent-child retrieval)
+        self.parent_text = parent_text or self.text
+
+    def to_dict(self):
+        return {
+            "chunk_id": self.chunk_id,
+            "doc_id": self.doc_id,
+            "text": self.text,
+            "title": self.title,
+            "chunk_index": self.chunk_index,
+            "parent_text": self.parent_text,
+        }
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            chunk_id=data.get("chunk_id"),
+            doc_id=data.get("doc_id"),
+            text=data.get("text", ""),
+            title=data.get("title", ""),
+            chunk_index=data.get("chunk_index", 0),
+            parent_text=data.get("parent_text"),
+        )
+
+    def __repr__(self):
+        return f"<TextChunk id={self.chunk_id} doc={self.doc_id} len={len(self.text)}>"
+
+
+def split_text_into_chunks(text, chunk_size=400, chunk_overlap=50):
+    """Split a string of text into overlapping word chunks.
+    
+    Args:
+        text: The string content to split.
+        chunk_size: Target number of words per chunk.
+        chunk_overlap: Number of overlapping words between consecutive chunks.
+        
+    Returns:
+        List of string chunks.
+    """
+    text = (text or "").strip()
+    if not text:
+        return []
+
+    words = text.split()
+    if len(words) <= chunk_size:
+        return [text]
+
+    chunks = []
+    step = max(1, chunk_size - chunk_overlap)
+    for start in range(0, len(words), step):
+        end = start + chunk_size
+        chunk_words = words[start:end]
+        chunks.append(" ".join(chunk_words))
+        if end >= len(words):
+            break
+
+    return chunks
+
+
+def chunk_document(document, chunk_size=400, chunk_overlap=50):
+    """Chunk a Document into TextChunk objects.
+    
+    Args:
+        document: A Document instance.
+        chunk_size: Words per chunk.
+        chunk_overlap: Overlap between consecutive chunks.
+        
+    Returns:
+        List of TextChunk instances.
+    """
+    text_chunks = split_text_into_chunks(
+        document.text,
+        chunk_size=chunk_size,
+        chunk_overlap=chunk_overlap,
+    )
+    
+    chunks = []
+    for idx, piece in enumerate(text_chunks):
+        cid = f"{document.doc_id}_chunk_{idx}"
+        chunk = TextChunk(
+            chunk_id=cid,
+            doc_id=document.doc_id,
+            text=piece,
+            title=document.title,
+            chunk_index=idx,
+            parent_text=document.text,  # Keep full document text as parent context
+        )
+        chunks.append(chunk)
+    return chunks
+
+
+def chunk_documents(documents, chunk_size=400, chunk_overlap=50):
+    """Chunk multiple Document objects."""
+    all_chunks = []
+    for doc in documents:
+        all_chunks.extend(chunk_document(doc, chunk_size, chunk_overlap))
+    return all_chunks

src/rag_engine/documents.py

@@ -0,0 +1,86 @@
+"""Document representations and file loaders for the RAG engine.
+
+Zero external complexity: clean, readable objects representing documents.
+"""
+
+import json
+import hashlib
+from pathlib import Path
+
+
+class Document:
+    """Represents a source document in the RAG system."""
+
+    def __init__(self, text, doc_id=None, title=None, source="upload", metadata=None):
+        self.text = text.strip() if text else ""
+        self.doc_id = doc_id or hashlib.md5(self.text[:100].encode("utf-8")).hexdigest()[:12]
+        self.title = title or f"Document-{self.doc_id[:6]}"
+        self.source = source
+        self.metadata = metadata or {}
+
+    def to_dict(self):
+        return {
+            "doc_id": self.doc_id,
+            "title": self.title,
+            "text": self.text,
+            "source": self.source,
+            "metadata": self.metadata,
+        }
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            text=data.get("text", ""),
+            doc_id=data.get("doc_id"),
+            title=data.get("title"),
+            source=data.get("source", "upload"),
+            metadata=data.get("metadata", {}),
+        )
+
+    def __repr__(self):
+        return f"<Document id={self.doc_id} title='{self.title}'>"
+
+
+def load_text_file(filepath):
+    """Load a plain text or markdown file into a Document."""
+    path = Path(filepath)
+    text = path.read_text(encoding="utf-8").strip()
+    title = path.stem.replace("_", " ").replace("-", " ").title()
+    return Document(text=text, doc_id=path.stem, title=title, source=path.name)
+
+
+def load_pdf_file(filepath):
+    """Extract text from a PDF file into a Document."""
+    path = Path(filepath)
+    try:
+        from pypdf import PdfReader
+        reader = PdfReader(str(path))
+        pages = [page.extract_text() or "" for page in reader.pages]
+        text = "\n\n".join(pages).strip()
+    except Exception:
+        text = f"Failed to extract text from {path.name}"
+    title = path.stem.replace("_", " ").replace("-", " ").title()
+    return Document(text=text, doc_id=path.stem, title=title, source=path.name)
+
+
+def save_documents_jsonl(documents, filepath):
+    """Save a list of Document objects to a JSONL file."""
+    path = Path(filepath)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        for doc in documents:
+            f.write(json.dumps(doc.to_dict()) + "\n")
+
+
+def load_documents_jsonl(filepath):
+    """Load Document objects from a JSONL file."""
+    path = Path(filepath)
+    if not path.exists():
+        return []
+    docs = []
+    with open(path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                docs.append(Document.from_dict(json.loads(line)))
+    return docs

src/rag_engine/embeddings.py

@@ -0,0 +1,81 @@
+"""Embedding functions for converting text to semantic vector representations.
+
+Embeddings map human text into high-dimensional vectors such that texts with
+similar meanings have high cosine similarity (close to 1.0).
+"""
+
+import hashlib
+import math
+
+
+class HashEmbeddings:
+    """A deterministic, zero-dependency embedding generator.
+    
+    Uses SHA-256 token hashing into 384 buckets.
+    Great for instant offline demos, testing, and machines without GPUs or PyTorch.
+    """
+
+    def __init__(self, dim=384):
+        self.dim = dim
+
+    def embed_text(self, text):
+        """Convert a single string into a normalized float vector."""
+        vector = [0.0] * self.dim
+        if not text:
+            return vector
+
+        tokens = text.lower().split()
+        for idx, token in enumerate(tokens):
+            digest = hashlib.sha256(token.encode("utf-8")).digest()
+            bucket = int.from_bytes(digest[:4], "little") % self.dim
+            # Weigh by position and presence
+            vector[bucket] += 1.0 + (idx % 5) * 0.05
+
+        # L2 normalize
+        squared_sum = sum(v * v for v in vector)
+        norm = math.sqrt(squared_sum) or 1.0
+        return [v / norm for v in vector]
+
+    def embed_documents(self, texts):
+        """Batch embed a list of texts."""
+        return [self.embed_text(t) for t in texts]
+
+    def embed_query(self, query):
+        """Embed a search query."""
+        return self.embed_text(query)
+
+
+def cosine_similarity(vec_a, vec_b):
+    """Compute cosine similarity between two unit-normalized vectors."""
+    if not vec_a or not vec_b or len(vec_a) != len(vec_b):
+        return 0.0
+    dot_product = sum(a * b for a, b in zip(vec_a, vec_b))
+    return float(dot_product)
+
+
+def get_embedding_model(settings=None):
+    """Factory to get the configured embedding model."""
+    if settings is None or getattr(settings, "use_hash_embeddings", True):
+        return HashEmbeddings()
+
+    # If requested and available, use SentenceTransformers
+    try:
+        from sentence_transformers import SentenceTransformer
+
+        class STEmbeddings:
+            def __init__(self, model_name):
+                self.model = SentenceTransformer(model_name)
+
+            def embed_text(self, text):
+                return self.model.encode(text, normalize_embeddings=True).tolist()
+
+            def embed_documents(self, texts):
+                return self.model.encode(texts, normalize_embeddings=True).tolist()
+
+            def embed_query(self, query):
+                return self.embed_text(query)
+
+        return STEmbeddings(settings.embedding_model)
+    except Exception:
+        # Fall back to HashEmbeddings if sentence-transformers is missing
+        return HashEmbeddings()

src/rag_engine/eval.py

@@ -0,0 +1,65 @@
+"""Lightweight RAG evaluation metrics: Faithfulness and Answer Relevancy.
+
+Evaluation answers the core question: 'Is my RAG system working well?'
+- Faithfulness: Does the answer stick to the facts present in the retrieved context?
+- Answer Relevancy: Does the answer actually address what the user asked?
+"""
+
+import re
+
+WORD_RE = re.compile(r"[a-z0-9]{3,}")
+
+
+def _extract_keywords(text):
+    """Extract lowercase keywords of 3+ letters."""
+    if not text:
+        return set()
+    return set(WORD_RE.findall(text.lower()))
+
+
+def evaluate_faithfulness(answer, retrieved_items):
+    """Compute lexical faithfulness of an answer against retrieved context.
+    
+    Returns a score between 0.0 and 1.0 representing what percentage of
+    key terms in the answer are supported by the retrieved context.
+    """
+    if not answer or not retrieved_items:
+        return 0.0
+
+    context_text = " ".join(item["chunk"].text for item in retrieved_items)
+    answer_words = _extract_keywords(answer)
+    context_words = _extract_keywords(context_text)
+
+    if not answer_words:
+        return 0.0
+
+    overlap = answer_words.intersection(context_words)
+    score = len(overlap) / len(answer_words)
+    return round(min(1.0, score), 3)
+
+
+def evaluate_relevancy(question, answer):
+    """Compute lexical relevancy between question and answer."""
+    if not question or not answer:
+        return 0.0
+
+    q_words = _extract_keywords(question)
+    a_words = _extract_keywords(answer)
+
+    if not q_words or not a_words:
+        return 0.0
+
+    overlap = q_words.intersection(a_words)
+    score = len(overlap) / len(q_words)
+    return round(min(1.0, score), 3)
+
+
+def evaluate_response(question, answer, retrieved_items):
+    """Evaluate a complete RAG query result."""
+    faithfulness = evaluate_faithfulness(answer, retrieved_items)
+    relevancy = evaluate_relevancy(question, answer)
+    return {
+        "faithfulness": faithfulness,
+        "answer_relevancy": relevancy,
+        "overall_score": round((faithfulness + relevancy) / 2.0, 3),
+    }

src/rag_engine/generator.py

@@ -0,0 +1,114 @@
+"""Answer generator for RAG: Combines retrieved context with the user question.
+
+Augments the prompt sent to the LLM with relevant retrieved chunks.
+Provides an extractive fallback mode if no API key is set so the entire pipeline
+runs 100% offline out-of-the-box.
+"""
+
+from config.prompts import INSUFFICIENT_EVIDENCE, SYSTEM_PROMPT
+
+
+def format_context(retrieved_items):
+    """Format retrieved chunks into a clean, numbered context block for the prompt."""
+    if not retrieved_items:
+        return "No relevant context found."
+
+    blocks = []
+    for idx, item in enumerate(retrieved_items, start=1):
+        chunk = item["chunk"]
+        title = chunk.title or f"Document {chunk.doc_id}"
+        blocks.append(f"[Source {idx}: {title}]\n{chunk.text}")
+    return "\n\n".join(blocks)
+
+
+def extractive_fallback(question, retrieved_items):
+    """Offline extractive summary when no LLM API key is configured."""
+    if not retrieved_items:
+        return INSUFFICIENT_EVIDENCE
+
+    lines = [
+        f"Answer synthesized from indexed documents for: '{question}'",
+        "",
+        "Key relevant excerpts from your knowledge base:",
+    ]
+    for idx, item in enumerate(retrieved_items[:3], start=1):
+        chunk = item["chunk"]
+        snippet = chunk.text.strip().replace("\n", " ")
+        if len(snippet) > 280:
+            snippet = snippet[:280] + "..."
+        lines.append(f"{idx}. [Source: {chunk.title}] \"{snippet}\"")
+
+    lines.append("")
+    lines.append(
+        "*Note: Set USER_LLM_API_KEY in your .env file to enable full generative AI answers with OpenAI/Claude.*"
+    )
+    return "\n".join(lines)
+
+
+def generate_answer(question, retrieved_items, settings):
+    """Generate an answer using OpenAI LLM or graceful offline fallback."""
+    if not retrieved_items:
+        return {
+            "answer": INSUFFICIENT_EVIDENCE,
+            "sources": [],
+            "used_llm": False,
+        }
+
+    # Extract unique source titles
+    sources = []
+    seen_titles = set()
+    for item in retrieved_items:
+        title = item["chunk"].title
+        if title and title not in seen_titles:
+            seen_titles.add(title)
+            sources.append(title)
+
+    # Check if LLM API key is present
+    if not settings.has_llm_key():
+        return {
+            "answer": extractive_fallback(question, retrieved_items),
+            "sources": sources,
+            "used_llm": False,
+        }
+
+    # Generate with LLM (OpenAI via LangChain)
+    try:
+        from langchain_openai import ChatOpenAI
+        from langchain_core.prompts import ChatPromptTemplate
+        from langchain_core.output_parsers import StrOutputParser
+
+        context = format_context(retrieved_items)
+        prompt = ChatPromptTemplate.from_messages([
+            ("system", SYSTEM_PROMPT),
+            ("human", "{question}"),
+        ])
+
+        llm_kwargs = {
+            "api_key": settings.user_llm_api_key,
+            "model": settings.user_llm_model,
+            "temperature": settings.llm_temperature,
+            "max_tokens": settings.llm_max_tokens,
+        }
+        if settings.user_llm_base_url:
+            llm_kwargs["base_url"] = settings.user_llm_base_url
+
+        llm = ChatOpenAI(**llm_kwargs)
+        chain = prompt | llm | StrOutputParser()
+
+        raw_answer = chain.invoke({
+            "context": context,
+            "question": question,
+        })
+
+        return {
+            "answer": raw_answer.strip(),
+            "sources": sources,
+            "used_llm": True,
+        }
+    except Exception as exc:
+        # If LLM API call fails (e.g. invalid key or network error), fallback cleanly
+        return {
+            "answer": f"{extractive_fallback(question, retrieved_items)}\n\n*(LLM generation failed: {str(exc)})*",
+            "sources": sources,
+            "used_llm": False,
+        }

src/rag_engine/hybrid_search.py

@@ -0,0 +1,100 @@
+"""Hybrid search combining Dense Vector Search and BM25 Keyword Search.
+
+Uses Reciprocal Rank Fusion (RRF) to merge the ranked result lists from both
+retrieval methods into a single, high-quality candidate list.
+
+Why Hybrid Search?
+- Dense vectors understand semantic similarity (e.g. 'space telescope' -> 'astronomy').
+- BM25 understands exact identifiers and keywords (e.g. 'NIRCam', 'L2', '15 millikelvin').
+- Together, they significantly outperform either method on its own.
+"""
+
+from collections import defaultdict
+
+
+def reciprocal_rank_fusion(dense_results, bm25_results, dense_weight=0.6, bm25_weight=0.4, k=60):
+    """Merge two ranked lists using Reciprocal Rank Fusion (RRF).
+    
+    Formula for each document:
+      Score = sum( weight / (k + rank) )
+      
+    Args:
+        dense_results: List of dicts from vector search [{"chunk": ..., "score": ...}].
+        bm25_results: List of dicts from BM25 search.
+        dense_weight: Relative importance of vector search (default: 0.6).
+        bm25_weight: Relative importance of keyword search (default: 0.4).
+        k: Smoothing constant to dampen extreme high ranks (standard is 60).
+        
+    Returns:
+        List of fused dicts sorted by combined RRF score.
+    """
+    rrf_scores = defaultdict(float)
+    chunks_by_id = {}
+    dense_ranks = {}
+    bm25_ranks = {}
+
+    # Accumulate dense ranks
+    for rank, item in enumerate(dense_results, start=1):
+        chunk = item["chunk"]
+        cid = chunk.chunk_id
+        chunks_by_id[cid] = chunk
+        dense_ranks[cid] = rank
+        rrf_scores[cid] += dense_weight * (1.0 / (k + rank))
+
+    # Accumulate BM25 ranks
+    for rank, item in enumerate(bm25_results, start=1):
+        chunk = item["chunk"]
+        cid = chunk.chunk_id
+        chunks_by_id[cid] = chunk
+        bm25_ranks[cid] = rank
+        rrf_scores[cid] += bm25_weight * (1.0 / (k + rank))
+
+    # Sort descending by fused score
+    sorted_items = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
+
+    fused_results = []
+    for cid, combined_score in sorted_items:
+        fused_results.append({
+            "chunk": chunks_by_id[cid],
+            "score": round(combined_score, 5),
+            "dense_rank": dense_ranks.get(cid),
+            "bm25_rank": bm25_ranks.get(cid),
+            "source": "hybrid",
+        })
+
+    return fused_results
+
+
+class HybridRetriever:
+    """Orchestrates dense vector search and BM25 search."""
+
+    def __init__(self, vector_store, bm25_index, dense_weight=0.6, bm25_weight=0.4):
+        self.vector_store = vector_store
+        self.bm25_index = bm25_index
+        self.dense_weight = dense_weight
+        self.bm25_weight = bm25_weight
+
+    def retrieve(self, query, top_k=4, dense_k=10, bm25_k=10):
+        """Retrieve the best matching chunks using hybrid search.
+        
+        Returns:
+            Dict containing:
+              - 'results': top_k fused results
+              - 'dense_results': raw vector results
+              - 'bm25_results': raw BM25 results
+        """
+        dense_hits = self.vector_store.search(query, top_k=dense_k)
+        bm25_hits = self.bm25_index.search(query, top_k=bm25_k)
+
+        fused = reciprocal_rank_fusion(
+            dense_results=dense_hits,
+            bm25_results=bm25_hits,
+            dense_weight=self.dense_weight,
+            bm25_weight=self.bm25_weight,
+        )
+
+        return {
+            "results": fused[:top_k],
+            "dense_results": dense_hits,
+            "bm25_results": bm25_hits,
+        }

src/rag_engine/pipeline.py

@@ -0,0 +1,154 @@
+"""Complete end-to-end RAG Pipeline orchestrator.
+
+Coordinates all 5 core stages of Retrieval-Augmented Generation:
+1. Ingestion: Load raw text, PDFs, or demo articles into Document objects.
+2. Chunking: Split documents into smaller, overlapping TextChunk passages.
+3. Indexing: Index chunks into both Dense Vector Store and BM25 Keyword Search.
+4. Hybrid Retrieval: Retrieve and rank chunks using Reciprocal Rank Fusion (RRF).
+5. Generation: Augment prompt with retrieved chunks and generate answer with LLM.
+"""
+
+import time
+from pathlib import Path
+
+from config.settings import get_settings
+from rag_engine.documents import Document, load_text_file, load_pdf_file, save_documents_jsonl, load_documents_jsonl
+from rag_engine.chunking import chunk_documents
+from rag_engine.vector_store import VectorStore
+from rag_engine.bm25_search import BM25SearchIndex
+from rag_engine.hybrid_search import HybridRetriever
+from rag_engine.generator import generate_answer
+from rag_engine.eval import evaluate_response
+
+
+class RAGPipeline:
+    """The central RAG engine coordinating ingestion, search, and generation."""
+
+    def __init__(self, settings=None):
+        self.settings = settings or get_settings()
+        self.documents = []
+        self.chunks = []
+        self.vector_store = VectorStore()
+        self.bm25_index = BM25SearchIndex()
+        self.retriever = HybridRetriever(
+            vector_store=self.vector_store,
+            bm25_index=self.bm25_index,
+            dense_weight=self.settings.dense_weight,
+            bm25_weight=self.settings.bm25_weight,
+        )
+
+    def add_document(self, text, title=None, doc_id=None, source="upload"):
+        """Add a single raw text document and index it."""
+        doc = Document(text=text, title=title, doc_id=doc_id, source=source)
+        self.documents.append(doc)
+        new_chunks = chunk_documents(
+            [doc],
+            chunk_size=self.settings.chunk_size,
+            chunk_overlap=self.settings.chunk_overlap,
+        )
+        self.chunks.extend(new_chunks)
+        self.vector_store.add_chunks(new_chunks)
+        self.bm25_index.build(self.chunks)
+        save_documents_jsonl(self.documents, self.settings.documents_file)
+        return doc
+
+    def add_pdf(self, filepath):
+        """Extract text from a PDF file and index it."""
+        doc = load_pdf_file(filepath)
+        return self.add_document(text=doc.text, title=doc.title, doc_id=doc.doc_id, source=doc.source)
+
+    def load_demo_corpus(self):
+        """Load bundled educational demo articles."""
+        demo_dir = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "demo_corpus"
+        if not demo_dir.exists():
+            return []
+
+        loaded = []
+        for txt_file in sorted(demo_dir.glob("*.txt")):
+            doc = load_text_file(txt_file)
+            self.documents.append(doc)
+            loaded.append(doc)
+
+        if loaded:
+            self.chunks = chunk_documents(
+                self.documents,
+                chunk_size=self.settings.chunk_size,
+                chunk_overlap=self.settings.chunk_overlap,
+            )
+            self.vector_store.clear()
+            self.vector_store.add_chunks(self.chunks)
+            self.bm25_index.build(self.chunks)
+            save_documents_jsonl(self.documents, self.settings.documents_file)
+
+        return loaded
+
+    def query(self, question, top_k=None):
+        """Execute the end-to-end RAG workflow for a user question.
+        
+        Steps:
+        1. Query dense vector store and BM25 keyword index.
+        2. Merge results using Reciprocal Rank Fusion.
+        3. Format retrieved context into LLM prompt.
+        4. Synthesize final answer with source citations.
+        5. Evaluate faithfulness and relevancy.
+        
+        Returns:
+            Dict containing:
+              - 'question': input question
+              - 'answer': generated answer
+              - 'sources': cited document titles
+              - 'retrieved_chunks': top ranked chunks
+              - 'dense_results': vector search results
+              - 'bm25_results': keyword search results
+              - 'evaluation': faithfulness & relevancy metrics
+              - 'latency_ms': elapsed execution time
+        """
+        start_time = time.perf_counter()
+        top_k = top_k or self.settings.final_top_k
+
+        # Step 1: Hybrid Retrieval
+        retrieval_output = self.retriever.retrieve(
+            query=question,
+            top_k=top_k,
+            dense_k=self.settings.dense_top_k,
+            bm25_k=self.settings.bm25_top_k,
+        )
+
+        retrieved_chunks = retrieval_output["results"]
+
+        # Step 2: Generation
+        gen_output = generate_answer(
+            question=question,
+            retrieved_items=retrieved_chunks,
+            settings=self.settings,
+        )
+
+        # Step 3: Evaluation
+        evaluation = evaluate_response(
+            question=question,
+            answer=gen_output["answer"],
+            retrieved_items=retrieved_chunks,
+        )
+
+        elapsed_ms = round((time.perf_counter() - start_time) * 1000, 1)
+
+        return {
+            "question": question,
+            "answer": gen_output["answer"],
+            "sources": gen_output["sources"],
+            "used_llm": gen_output["used_llm"],
+            "retrieved_chunks": retrieved_chunks,
+            "dense_results": retrieval_output["dense_results"],
+            "bm25_results": retrieval_output["bm25_results"],
+            "evaluation": evaluation,
+            "latency_ms": elapsed_ms,
+        }
+
+    def get_stats(self):
+        """Return summary statistics of the indexed knowledge base."""
+        return {
+            "document_count": len(self.documents),
+            "chunk_count": len(self.chunks),
+            "vector_store_size": len(self.vector_store),
+            "bm25_index_size": len(self.bm25_index),
+        }

src/rag_engine/vector_store.py

@@ -0,0 +1,57 @@
+"""Vector Store implementation for indexing and semantic similarity search.
+
+A vector database stores chunks alongside their embedding vectors and finds
+the nearest neighbors (most relevant chunks) using cosine similarity.
+"""
+
+from rag_engine.embeddings import cosine_similarity, get_embedding_model
+
+
+class VectorStore:
+    """Simple in-memory vector store with cosine similarity ranking."""
+
+    def __init__(self, embedding_model=None):
+        self.embedding_model = embedding_model or get_embedding_model()
+        self.chunks = []
+        self.vectors = []
+
+    def add_chunks(self, chunks):
+        """Add TextChunk objects and calculate their embeddings."""
+        if not chunks:
+            return
+        new_texts = [c.text for c in chunks]
+        new_vectors = self.embedding_model.embed_documents(new_texts)
+        self.chunks.extend(chunks)
+        self.vectors.extend(new_vectors)
+
+    def search(self, query, top_k=5):
+        """Find the top_k most semantically similar chunks for a query.
+        
+        Returns:
+            List of dicts: [{"chunk": TextChunk, "score": float, "source": "dense"}]
+        """
+        if not self.chunks:
+            return []
+
+        query_vector = self.embedding_model.embed_query(query)
+        scored_results = []
+
+        for idx, (chunk, doc_vector) in enumerate(zip(self.chunks, self.vectors)):
+            similarity = cosine_similarity(query_vector, doc_vector)
+            scored_results.append({
+                "chunk": chunk,
+                "score": round(similarity, 4),
+                "source": "dense",
+            })
+
+        # Sort descending by similarity score
+        scored_results.sort(key=lambda x: x["score"], reverse=True)
+        return scored_results[:top_k]
+
+    def clear(self):
+        """Empty the store."""
+        self.chunks = []
+        self.vectors = []
+
+    def __len__(self):
+        return len(self.chunks)