RAG — 3 Documents

Retrieval-Augmented Generation (RAG) is an AI paradigm that combines a retrieval system with a generative language model. Instead of relying solely on the parametric knowledge baked into an LLM during training, RAG first retrieves relevant document chunks from an external knowledge base (corpus, database, or search index) and then feeds them as context to the LLM to produce a grounded, factually grounded response.

The term was formally introduced by Lewis et al. in the 2020 paper "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks", where they proposed a hybrid architecture that marries a dense retriever (DPR — Dense Passage Retriever) with a seq2seq generator (BART).

Core Concept

User Query
    |
    v
[Retriever] ---> External Knowledge Base (Vector DB / Index)
    |
    v
Retrieved Context (top-k chunks)
    |
    v
[Generator / LLM] ---> Grounded Response

RAG is not a single technique but a family of approaches that differ in when retrieval happens, what is retrieved, and how it is fused with generation.

Large Language Models (LLMs) are powerful but suffer from fundamental limitations that RAG directly addresses:

2.1 Stale Knowledge

LLMs have a knowledge cutoff — they know nothing about events, documents, or data created after their training date. RAG enables access to real-time or recent information without retraining.

2.2 Hallucination

LLMs are prone to generating plausible-sounding but factually incorrect content ("hallucination"). RAG grounds generation in retrieved evidence, dramatically reducing fabricated claims.

2.3 Lack of Attributability

A pure LLM cannot cite sources for its claims. RAG naturally supports attribution — the response can reference specific chunks from the knowledge base, enabling verifiability.

2.4 Domain Specificity

General-purpose LLMs lack deep knowledge of proprietary or niche domains (legal, medical, enterprise internal docs). RAG allows injecting domain-specific knowledge without fine-tuning.

2.5 Data Freshness and Updates

Updating a vector database is orders of magnitude cheaper and faster than retraining a model. RAG systems can be updated by simply re-indexing new documents.

2.6 Cost Efficiency

Fine-tuning an LLM for every new knowledge domain is expensive and risks catastrophic forgetting. RAG keeps the LLM frozen and updates only the retrieval index.

2.7 Regulatory Compliance

Industries like healthcare (HIPAA), finance (SOX), and law (GDPR) often require that AI outputs be traceable to source documents. RAG provides this traceability natively.

A canonical RAG system comprises three main stages:

3.1 Indexing Pipeline

The indexing pipeline prepares raw documents for efficient retrieval.

Raw Documents
    |
[Document Loader] ---> Parsed Text / Markdown / PDF
    |
[Text Splitter / Chunker] ---> Overlapping Chunks (tokens / sentences / paragraphs)
    |
[Embedding Model] ---> Dense Vector Representations
    |
[Vector Store] ---> Indexed Embeddings + Metadata

Key components:
- Document Loader: Ingests files from various formats (PDF, HTML, DOCX, Markdown, plain text, databases).
- Text Splitter / Chunker: Divides documents into manageable segments. Strategy choices (fixed-size, semantic, recursive, agentic) have a direct impact on retrieval quality.
- Embedding Model: Converts text chunks into fixed-dimensional dense vectors that capture semantic meaning.
- Vector Store: Persists embeddings and supports efficient similarity search (ANN — Approximate Nearest Neighbor).

3.2 Retrieval Stage

At query time, the user's question is embedded with the same embedding model, and the vector store returns the top-k most semantically similar chunks.

User Query
    |
[Embedding Model] ---> Query Vector
    |
[Vector Store (ANN Search)] ---> Top-k Chunks
    |
[Optional: Re-ranker] ---> Re-scored / Filtered Context

Common retrieval strategies:
- Dense Retrieval: Query and documents embedded into the same dense vector space; cosine / dot-product / L2 similarity used to rank results.
- Sparse Retrieval (BM25): Keyword-based lexical matching; complementary to dense retrieval.
- Hybrid Retrieval: Combination of dense + sparse, often with reciprocal rank fusion (RRF) or learned weighting.
- Re-ranking: A cross-encoder model (e.g., Cohere Rerank, BGE-Reranker) re-scores the top-k retrieved chunks for improved precision.

3.3 Generation Stage

The retrieved chunks are injected into the LLM's prompt as context, and the LLM generates a response conditioned on both the query and the evidence.

System Prompt + Retrieved Chunks + User Query
    |
[LLM] ---> Grounded Response

Prompt template (simplified):

You are a helpful assistant. Use the following context to answer the question.
If the context does not contain enough information, say so.

Context:
{retrieved_chunks}

Question: {user_query}

Answer:

Variations:
- Concatenation: All top-k chunks are concatenated into a single context window.
- Sliding Window: Chunks are fed iteratively for long-document scenarios.
- Conditional Generation: Only retrieve when the LLM's internal uncertainty is high (adaptive RAG).

Below is a step-by-step walkthrough of a RAG query lifecycle.

Step 1: Query Input

The user submits a natural language query (e.g., "What were the Q4 2024 revenue numbers for Acme Corp?").

Step 2: Query Pre-processing

• Query Rewriting: LLM reframes the query for better retrieval (e.g., "Acme Corp Q4 2024 revenue").
• Query Decomposition: Complex queries are split into sub-queries for multi-hop retrieval.
• HyDE (Hypothetical Document Embedding): An LLM generates a hypothetical answer document, then that document's embedding is used for retrieval instead of the raw query.

Step 3: Query Embedding

The (possibly rewritten) query is passed through the embedding model to produce a query vector.

Step 4: Vector Search

The query vector searches the vector store using an ANN algorithm (HNSW, IVF, PQ, etc.) to retrieve the top-k nearest neighbors.

Parameters:
- k — number of chunks to retrieve (typically 3–10).
- score_threshold — minimum similarity score for inclusion.
- filter — metadata filters (date range, document source, author, etc.).

Step 5: Post-retrieval Processing

• Re-ranking: Apply a cross-encoder to reorder results by query-chunk relevance.
• Deduplication: Remove near-identical chunks.
• Compression: Summarize or extract the most relevant sentences from long chunks.
• Context Window Management: Truncate or prioritize chunks to fit within the LLM's context window (e.g., 4K, 8K, 128K tokens).

Step 6: Prompt Construction

Assemble the prompt with system instructions, the processed context, and the original user query.

Step 7: Generation

The LLM generates the final response, often with instructions to cite chunk IDs or source documents.

Step 8: Post-generation Processing

• Citation Formatting: Convert internal chunk references to human-readable citations.
• Factuality Check: Optional second LLM call to verify claims against retrieved context.
• Audit Logging: Store query, retrieved chunks, generated response, and metrics for monitoring.

RAG systems can be classified along a maturity spectrum:

5.1 Naive RAG (or Simple RAG)

The original formulation: index → retrieve → generate, in a single pass.

Characteristics:
- Single retrieval step before generation.
- Fixed chunk size (typically 256–1024 tokens).
- No query rewriting or post-processing.
- No re-ranking.

Pros: Simple to implement, low latency.
Cons: Poor handling of complex / multi-hop queries, no mechanism for retrieval failure, context window limitations.

When to use: Prototyping, simple Q&A over small corpora, low-latency requirements.

5.2 Advanced RAG

Introduces pre-retrieval and post-retrieval optimizations.

Pre-retrieval improvements:
- Query expansion: Generate multiple related queries to improve recall.
- Query transformation: LLM-based rewriting, Step-back prompting (ask a broader question first).
- HyDE: Use LLM-generated hypothetical document for retrieval.

Post-retrieval improvements:
- Re-ranking: Cross-encoder or Cohere Rerank on top-k results.
- Context compression: Filter irrelevant sentences within retrieved chunks.
- Sliding window / iterative retrieval: For multi-hop questions.

Indexing improvements:
- Hierarchical indices: Summary-level index → chunk-level index.
- Metadata filtering: Pre-filter on date, author, source type.
- Multi-representation indexing: Store both a summary and full text for each chunk.

Pros: Significantly higher accuracy than Naive RAG.
Cons: Increased latency and complexity, more moving parts to tune.

5.3 Modular RAG

A flexible, composable architecture where retrieval and generation components can be rearranged, replaced, or augmented with additional modules.

Core modules:
- Search Module: Dense, sparse, hybrid, or external API (Google Search, Bing, etc.).
- Routing Module: Direct query to the appropriate retriever or data source based on query type.
- Memory Module: Maintain conversation history for multi-turn retrieval.
- Fusion Module: Combine results from multiple retrievers.
- Adaptive Module: Decide whether to retrieve based on query difficulty.
- Verification Module: Confirm generated claims against retrieved evidence.

Common Modular RAG patterns:

Pros: Maximum flexibility and accuracy; state-of-the-art results.
Cons: Highest complexity; engineering overhead; latency management is non-trivial.

Chunking is arguably the single most impactful design decision in a RAG system. The goal is to produce segments that are semantically self-contained, contextually coherent, and optimally sized for both retrieval and generation.

6.1 Fixed-Size Chunking

The simplest approach: split text every N characters / tokens, often with overlap.

def fixed_size_chunk(text, chunk_size=512, overlap=64):
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        # optionally break at nearest sentence boundary
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks

Pros: Trivial to implement, deterministic, fast.
Cons: No semantic boundaries — chunks may split sentences or paragraphs in the middle, losing context.

6.2 Semantic Chunking

Split text at natural semantic boundaries — sentences, paragraphs, or topic shifts.

Approaches:
- Paragraph-based: Split on double newlines.
- Document structure: Use Markdown / HTML headings as boundaries.
- Embedding-based: Chunk boundaries are determined by abrupt changes in embedding similarity (smaller inter-sentence similarity → boundary).
- Model-based: Use a sentence-boundary detection model (e.g., spaCy sentence tokenizer, nltk.sent_tokenize).

Pros: Chunks are contextually coherent; higher retrieval quality.
Cons: Chunks can be highly variable in length; risk of very long or very short chunks.

6.3 Recursive Chunking

A strategy popularized by LangChain: apply multiple splitters in sequence, falling back from most aggressive to least aggressive.

LangChain RecursiveCharacterTextSplitter behavior:
1. Attempt to split on paragraph boundaries (\n\n).
2. If any chunk exceeds the max size, fall back to sentence boundaries (.).
3. If still too large, split on word boundaries.
4. Final fallback: character-by-character split.

recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)

Pros: Graceful handling of long documents; produces chunks of approximately uniform size while respecting structure.
Cons: Still operates at the lexical level, not fully semantic.

6.4 Agentic / LLM-based Chunking

Use an LLM to determine chunk boundaries by understanding the document's narrative structure and topic flow.

Example approach:
- Feed a sliding window of N sentences to an LLM.
- Ask the LLM: "Does this sentence start a new topic? If yes, propose a chunk boundary here."
- Merge decisions across the document.

LLM-assisted summarization chunking:
- Generate a short summary for each natural paragraph.
- Store both summary (for retrieval) and full paragraph (for generation).
- Retrieve by summary embedding; use full content for generation.

Pros: Highest quality chunking; respects document-level semantics.
Cons: Expensive (LLM calls), slow, non-deterministic, difficult to scale to large corpora.

6.5 Comparison Table

Embedding models map text to dense vector representations. The quality of embeddings directly determines retrieval accuracy.

7.1 text-embedding-ada-002 (OpenAI)

• Dimensions: 1536
• Context window: 8192 tokens
• Pricing: Pay-per-token (approximately $0.0001 / 1K tokens)
• Strength: Excellent general-purpose embeddings; strong zero-shot performance across domains.
• Limitations: Closed-source; vendor lock-in; data sent to OpenAI; no fine-tuning on custom data.

7.2 Sentence-Transformers (SBERT)

An open-source library providing hundreds of pre-trained models.

Strengths: Open-source; can be fine-tuned on domain data; runs locally; huge model zoo.
Limitations: Requires local GPU for high throughput; performance varies by model size.

7.3 BGE (BAAI General Embedding)

Developed by the Beijing Academy of Artificial Intelligence (BAAI). Current state-of-the-art among open-source embedding models.

Key feature: Trained with a dedicated retrieval fine-tuning stage; supports instruction= parameter for query-side embedding.

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-base-en-v1.5")
query_emb = model.encode("Query: What is RAG?")
doc_emb = model.encode("RAG is a retrieval-augmented generation paradigm...")

7.4 E5 / E5-Mistral (Microsoft)

E5 (EmbEddings from bidirEctional Encoder rEpresentations) and its successor E5-Mistral achieve top-tier results on MTEB (Massive Text Embedding Benchmark).

Key feature: Uses LLM (Mistral-7B) as backbone for E5-Mistral, enabling instruction-aware embeddings.

7.5 Embedding Model Selection Criteria

7.6 MTEB Leaderboard (Top Models as of 2025)

Note: Scores change frequently; always check the latest MTEB results.

Vector databases are purpose-built for storing embeddings and performing fast Approximate Nearest Neighbor (ANN) search.

8.1 Pinecone

Type: Fully managed cloud service (serverless / pod-based).

Key Features:
- Auto-scaling; no infrastructure management.
- Built-in hybrid search (dense + sparse via pinecone-text).
- Single-stage metadata filtering.
- Namespace isolation for multi-tenancy.
- Supports up to 200 dimensions free tier (pod-based).

Algorithms: HNSW, with optimized SIMD instructions.

Pros: No ops overhead; excellent documentation; fast onboarding.
Cons: Vendor lock-in; only cloud-hosted (no self-hosted option); expensive at scale.

8.2 Weaviate

Type: Open-source / cloud (Weaviate Cloud Services).

Key Features:
- Native vector + scalar storage.
- Built-in modules for OpenAI, Cohere, HuggingFace, and custom models.
- GraphQL and REST APIs.
- Hybrid search (BM25 + vector) with weighted fusion.
- Multi-tenancy, sharding, replication.

Algorithms: HNSW by default; also supports PQ compression.

Pros: Open-source; self-hostable; rich module ecosystem; hybrid search.
Cons: Steeper learning curve; self-hosting requires operational expertise.

8.3 Qdrant

Type: Open-source / cloud (Qdrant Cloud).

Key Features:
- Written in Rust — very fast.
- Rich filtering with payload indexing.
- Scalar and Product Quantization for memory reduction.
- Binary quantization (1-bit) for extreme compression.
- Multi-vector support.

Algorithms: HNSW only; highly optimized Rust implementation.

Pros: Fastest ANN search speeds; memory-efficient; Docker one-liner to self-host.
Cons: Fewer built-in integrations than Weaviate; smaller ecosystem.

8.4 Milvus

Type: Open-source / cloud (Zilliz Cloud).

Key Features:
- Cloud-native design (separate storage, compute, and indexing layers).
- Multiple index types: IVF, HNSW, DiskANN, and GPU-accelerated indices.
- Hybrid search (dense + sparse) with ensemble reranking.
- Streaming and batch ingestion.

Algorithms: IVF_FLAT, IVF_SQ8, HNSW, GPU_IVF, DiskANN.

Pros: Production-proven at billion-scale; GPU indexing for massive throughput; Apache 2.0 license.
Cons: Heavy deployment (~6+ microservices for full cluster); steep learning curve; more complex than alternatives.

8.5 Chroma

Type: Open-source, lightweight embedded database.

Key Features:
- Python-native; pip-installable.
- In-memory and persistent (SQLite-backed) modes.
- Simple API: collection.add, collection.query.
- Metadata storage and filtering.
- Built-in embedding function wrappers (OpenAI, SentenceTransformers).

Pros: Zero-config; ideal for prototyping and small projects; tight Python integration.
Cons: No distributed / production-scale capabilities; limited index types; not suitable for >1M vectors.

8.6 Comparison Table

8.7 Choosing a Vector Database

Evaluating RAG systems is more nuanced than evaluating either retrieval or generation in isolation. Both components must be assessed together.

9.1 Evaluation Dimensions

9.2 RAGAS Framework

RAGAS (Retrieval Augmented Generation Assessment) is the most widely adopted evaluation framework for RAG systems, introduced by Shahul et al. in 2024. It defines several reference-free metrics:

Faithfulness

Measures whether the generated answer is factually consistent with the retrieved context.

How it works:
1. Decompose the generated answer into individual claims (atomic statements).
2. For each claim, check if it can be inferred from the retrieved context.
3. Faithfulness = (Number of supported claims) / (Total claims).

Range: 0 to 1. Higher is better.

Answer Relevance

Measures how well the generated answer addresses the user's question.

How it works:
1. Given the question and answer, ask an LLM to generate n reverse questions based on the answer.
2. Compute the cosine similarity between the original question and the generated questions (embedded).
3. Answer Relevance = average cosine similarity.

Range: 0 to 1. Higher is better.

Context Precision

Measures the signal-to-noise ratio in the retrieved context — are the relevant chunks ranked higher?

How it works:
1. For each ranked chunk, check relevance to the question.
2. Weight chunks by rank (higher rank = more weight if relevant).
3. Context Precision = sum over relevant chunks of (precision at that rank) / total relevant chunks.

Range: 0 to 1. Higher is better.

Context Recall

Measures whether all the information needed to answer the question is present in the retrieved context.

How it works:
1. Decompose the ground-truth answer (or the question itself) into claims.
2. For each claim, check if it can be inferred from the retrieved context.
3. Context Recall = (Number of supported claims) / (Total claims).

Range: 0 to 1. Higher is better.

Aspect Critique

Evaluates generation on specific qualitative aspects:

9.3 Other Evaluation Approaches

LLM-as-a-Judge

Use a strong LLM (GPT-4, Claude, Gemini) to score RAG outputs on custom rubrics.

Score 1-5 for each criterion:
- Groundedness: Does the answer reference retrieved context?
- Completeness: Does the answer fully address the question?
- Helpfulness: How useful is the answer to the user?

Human Evaluation

Gold standard but expensive. Useful for:
- Building and validating automated metrics.
- End-user satisfaction studies.
- A/B testing in production.

Embedding-based Metrics

• Answer Semantic Similarity: Cosine similarity between embedding of generated answer and ground-truth answer.
• BERTScore: Token-level precision and recall using BERT embeddings.

9.4 Evaluation Pipeline Best Practices

1. Build a diverse evaluation dataset: 100–500+ query-answer pairs covering typical, edge-case, and adversarial queries.
2. Test retrieval in isolation: Measure recall@k before evaluating the full pipeline.
3. Track cost and latency alongside quality metrics.
4. Segment metrics by query type: Factual lookup vs. synthesis vs. multi-hop.
5. Monitor drift over time: Re-run the same eval set after every index update.

10.1 Indexing

• Chunk with paragraphs, not characters. Use recursive or semantic chunking; avoid fixed-size splitting across paragraphs.
• Add chunk overlap. 10–20% overlap between consecutive chunks prevents information loss at boundaries.
• Store rich metadata. Include document title, date, source URL, section heading, and chunk index for filtering and citation.
• Create parent-child relationships. Retrieve summaries (child) but generate from full documents (parent) — improves retrieval quality while preserving context.
• Multi-representation indexing. Store a concise summary and the full content; retrieve by summary for better semantic matching.

10.2 Embeddings

• Normalize embeddings (L2 normalize) to enable cosine similarity via dot product.
• Use separate instructions for query vs. document embedding (BGE, E5-instruct).
• Batch encode during indexing for GPU efficiency.
• Fine-tune on domain data if your corpus is specialized (legal, medical, code).
• Monitor embedding drift when updating models.

10.3 Retrieval

• Always use hybrid search (dense + sparse) for best recall.
• Retrieve more than you need and re-rank. Fetch k=20-50, then re-rank to top 5-10.
• Apply metadata filters aggressively. Date range, source type, author filters dramatically improve precision.
• Implement query rewriting. An LLM can transform vague queries into specific, search-optimized queries.
• Use multi-stage retrieval for complex queries: broad recall → narrow precision.

10.4 Generation

• Instruct the LLM to cite sources. "For each claim, cite the source chunk ID."
• Handle "no relevant context" gracefully. If retrieval returns low-confidence chunks, have the LLM say "I could not find sufficient information."
• Enable streaming for better user experience (LLM generates while context is visible).
• Limit context window. Don't fill the entire LLM context window — leave room for reasoning.
• Use system prompts to reduce hallucination. "If the context does not contain the answer, say 'I don't know'. Do not make up information."

10.5 Production Considerations

• Cache frequent queries at the retrieval layer to reduce latency and cost.
• Measure and log retrieval latency, generation latency, token counts, and per-query cost.
• Set up evaluation as CI/CD pipeline. Run your eval suite on every index update and model change.
• Use A/B testing. Compare retrieval strategies, chunk sizes, and embedding models on live traffic.
• Implement guardrails. Input filtering (prompt injection, PII detection) and output filtering (toxicity, hallucination detection).

11.1 Retrieval Failure Modes

• Irrelevant context retrieved: The retriever returns chunks that are semantically similar but factually irrelevant to the user's question.
• Missing context: The required information exists in the corpus but was not retrieved (low recall).
• Out-of-date context: Retrieved chunks contain stale information that contradicts the current state.

11.2 Context Window Constraints

Even with long-context LLMs (128K–200K tokens), feeding many chunks can:
- Introduce noise and distract the LLM.
- Cause the LLM to "lost in the middle" — performance degrades on information in the middle of long prompts.
- Increase cost (tokens are expensive).

11.3 Chunking Trade-offs

• Small chunks → higher retrieval precision, but may lack full context for answering.
• Large chunks → more context per retrieval, but lower embedding specificity and noisier retrieval.

11.4 Embedding Quality

• Domain shift: Off-the-shelf embedding models perform poorly on specialized domains (legal, medical, scientific).
• Fine-tuning cost: Domain-adapted embeddings require labeled data and compute.
• Multi-lingual challenges: Many embedding models perform poorly on non-English text.

11.5 Latency and Cost

End-to-end latency for a production RAG system is typically 1–5 seconds, depending on context size and model.

11.6 Evaluation Challenges

• No single ground truth — many queries have multiple valid answers.
• Reference-free metrics are noisy — LLM-as-a-judge has its own biases and inconsistencies.
• Evaluation is expensive — running full RAGAS or GPT-4 evaluation on every change is cost-prohibitive.

11.7 Security and Privacy

• Prompt injection: Malicious queries can extract or manipulate retrieved content.
• Data leakage: Retrieved sensitive documents may appear in generated responses.
• Indirect prompt injection: Adversarial text in retrieved chunks can hijack the LLM.

11.8 Maintenance Burden

• Index updates: Re-chunking and re-embedding the entire corpus when strategies change.
• Model versioning: Coordinating embedding model versions with retrieval index compatibility.
• Monitoring drift: Embedding distributions drift over time as new documents are added.

Chunking is the process of dividing documents into smaller, retrievable pieces. The quality of chunking directly impacts retrieval accuracy, context relevance, and overall RAG system performance.

Fixed-Size Chunking

The simplest approach: documents are split into chunks of a predetermined size (character count, token count, or word count).

Implementation:

def fixed_size_chunk(text, chunk_size=512, overlap=0):
    """Fixed-size chunking by character count."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

def fixed_token_chunk(text, chunk_tokens=256, overlap_tokens=25, tokenizer=None):
    """Fixed-size chunking by token count."""
    tokens = tokenizer.encode(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = start + chunk_tokens
        chunk_tokens_list = tokens[start:end]
        chunk_text = tokenizer.decode(chunk_tokens_list)
        chunks.append(chunk_text)
        start += chunk_tokens - overlap_tokens
    return chunks

Parameters:
- chunk_size: Number of characters/tokens per chunk (typical: 256-1024 tokens).
- overlap: Number of tokens shared between consecutive chunks (typical: 10-25%).

Advantages:
- Simple and fast
- Deterministic and reproducible
- Easy to benchmark and optimize

Disadvantages:
- No semantic boundaries
- May split sentences or paragraphs
- Context can be lost at chunk boundaries
- Not content-aware

When to Use:
- Initial prototyping and baselines
- Homogeneous content (e.g., logs, structured text)
- When speed is primary concern

Recursive Splitter

Recursively splits text using a hierarchy of separators (paragraphs, sentences, words) to maintain semantic coherence.

Algorithm:
1. Try splitting by the first separator (e.g., "\n\n" for paragraphs).
2. If any chunk exceeds max size, recursively split it using the next separator.
3. Continue until all chunks are under the size limit.

Implementation (LangChain-style):

from typing import List

class RecursiveCharacterSplitter:
    def __init__(self, chunk_size=1000, chunk_overlap=200, separators=None):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators or ["\n\n", "\n", ".", " ", ""]

    def split_text(self, text: str) -> List[str]:
        return self._recursive_split(text, self.separators)

    def _recursive_split(self, text: str, separators: List[str]) -> List[str]:
        if len(text) <= self.chunk_size:
            return [text]

        if not separators:
            # No more separators, split by characters
            return self._split_by_chars(text)

        separator = separators[0]
        if separator:
            splits = text.split(separator)
        else:
            splits = list(text)

        chunks = []
        current_chunk = []
        current_size = 0

        for split in splits:
            split_size = len(split)
            if current_size + split_size > self.chunk_size:
                if current_chunk:
                    chunks.append(separator.join(current_chunk))
                    current_chunk = current_chunk[-self._overlap_count:]
                    current_size = sum(len(s) for s in current_chunk) + len(current_chunk) - 1
                current_chunk.append(split)
                current_size = len(split)
            else:
                current_chunk.append(split)
                current_size += split_size + (1 if current_chunk else 0)  # +1 for separator

        if current_chunk:
            chunks.append(separator.join(current_chunk))

        return chunks

Common Separator Hierarchy:

["\n\n", "\n", ".", "!", "?", ",", " ", ""]

Advantages:
- Respects natural text boundaries
- Produces more coherent chunks
- Better than fixed-size for most documents
- Configurable separator hierarchy

Disadvantages:
- More complex than fixed-size
- May still produce imperfect chunks
- Separator choice impacts quality

Semantic Chunking

Chunks are created based on semantic similarity between adjacent sentences or paragraphs, ensuring each chunk covers a coherent topic.

Embedding Similarity Approach

Uses embedding similarity to detect topic boundaries:

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticChunker:
    def __init__(self, embedding_model, threshold=0.5, min_chunk_tokens=50, max_chunk_tokens=500):
        self.embedding_model = embedding_model
        self.threshold = threshold
        self.min_chunk_tokens = min_chunk_tokens
        self.max_chunk_tokens = max_chunk_tokens

    def chunk(self, sentences):
        """Group sentences into semantic chunks based on embedding similarity."""
        if len(sentences) <= 1:
            return sentences

        # Get embeddings for all sentences
        embeddings = self.embedding_model.embed_documents(sentences)

        chunks = []
        current_chunk = [sentences[0]]

        for i in range(1, len(sentences)):
            # Measure similarity between consecutive sentences
            similarity = cosine_similarity(
                [embeddings[i-1]], [embeddings[i]]
            )[0][0]

            current_tokens = len(' '.join(current_chunk).split())

            if similarity < self.threshold and current_tokens >= self.min_chunk_tokens:
                # Topic boundary detected
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentences[i]]
            el...

Chunk Optimization

class ChunkEvaluator:
    def __init__(self, retriever, qa_pairs):
        self.retriever = retriever
        self.qa_pairs = qa_pairs  # List of (question, relevant_chunk_id, answer)

    def evaluate(self):
        """Evaluate chunking quality."""
        results = {
            "hit_rate": 0,
            "mrr": 0,  # Mean Reciprocal Rank
            "precision": 0,
            "recall": 0
        }

        for question, relevant_id, _ in self.qa_pairs:
            retrieved_chunks = self.retriever.retrieve(question, k=10)
            retrieved_ids = [c.id for c in retrieved_chunks]

            # Hit rate: was relevant chunk in top-k?
            if relevant_id in retrieved_ids:
                results["hit_rate"] += 1

            # MRR: rank position of first relevant
            rank = retrieved_ids.index(relevant_id) if relevant_id in retrieved_ids else -1
            if rank >= 0:
                results["mrr"] += 1.0 / (rank + 1)

            # Precision at k
            results["precision"] += self._precision_at_k(retrieved_ids, relevant_id)

        n = len(self.qa_pairs)
        return {k: v/n for k, v in results.items()}

import matplotlib.pyplot as plt

class ChunkSizeOptimizer:
    def __init__(self, documents, retriever_factory, evaluator):
        self.documents = documents
        self.retriever_factory = retriever_factory
        self.evaluator = evaluator

    def optimize(self, chunk_sizes=[256, 512, 768, 1024, 1536, 2048], overlaps=[0, 50, 100, 200]):
        """Test different chunk sizes and overlaps."""
        results = []

        for size in chunk_sizes:
            for overlap in overlaps:
                if overlap >= size:
                    continue

                # Create chunks with these parameters
                chunks = self._create_chunks(size, overlap)

                # Build retriever
                retriever = self.retriever_factory(chunks)

                # Evaluate
                metrics = self.evaluator(retriever).evaluate()
                metrics["chunk_size"] = size
                metrics["overlap"] = overlap
                results.append(metrics)

        return results

    def visualize(self, results):
        """Plot chunk size vs performance."""
        sizes = sorted(set(r["chunk_size"] for r in results))

        for metric in ["hit_rate", "mrr"]:
            scores = []
            for size in sizes:
                size_results = [r for r in results if r["chunk_size"] == size]
                avg_score = sum(r[metric] for r in size_results) / len(size_results)
                scores.append(avg_score)

            plt.plot(sizes, scores, label=metric)

        plt.xlabel("Chunk Size (tokens)")
        plt.ylabel("Score")
        plt.legend()
        plt.show()

def chunk_with_overlap(text, chunk_size, overlap_tokens):
    """Fixed overlap between consecutive chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunk = text[start:end]
        chunks.append(chunk)
        start += chunk_size - overlap_tokens
    return chunks

def dynamic_overlap(text, chunk_size, min_overlap=0.1, max_overlap=0.3):
    """Adjust overlap based on content similarity."""
    sentences = split_sentences(text)
    chunks = []
    i = 0

    while i < len(sentences):
        current = []
        token_count = 0

        while i < len(sentences) and token_count < chunk_size:
            current.append(sentences[i])
            token_count += len(sentences[i].split())
            i += 1

        # Check semantic similarity at boundary
        if i < len(sentences) and token_count >= chunk_size:
            similarity = sentence_similarity(sentences[i-1], sentences[i])
            overlap_ratio = min_overlap + (max_overlap - min_overlap) * (1 - similarity)
            overlap_tokens = int(chunk_size * overlap_ratio)

            # Roll back to include overlap
            while overlap_tokens > 0 and i > 0:
                prev_tokens = len(sentences[i-1

class DenseXIndexer:
    def __init__(self, embedding_model, num_representations=3):
        self.embedding_model = embedding_model
        self.num_representations = num_representations

    def index_chunk(self, chunk_text):
        """Create multiple representations for a single chunk."""
        representations = []

        # 1. Direct embedding
        direct_emb = self.embedding_model.embed_query(chunk_text)
        representations.append(direct_emb)

        # 2. Summary embedding
        summary = self._summarize(chunk_text)
        summary_emb = self.embedding_model.embed_query(summary)
        representations.append(summary_emb)

        # 3. Keyword-focused embedding
        keywords = self._extract_keywords(chunk_text)
        keyword_text = " ".join(keywords)
        keyword_emb = self.embedding_model.embed_query(keyword_text)
        representations.append(keyword_emb)

        # 4. Question-focused embedding (what questions does this chunk answer?)
        questions = self._generate_questions(chunk_text)
        for q in questions[:self.num_representations - 3]:
            q_emb = self.embedding_model.embed_query(q)
            representations.append(q_emb)

        return representations

class SummaryRetrievalIndex:
    def __init__(self, vector_store, summary_model):
        self.vector_store = vector_store
        self.summary_model = summary_model

    def index_document(self, doc_id, chunks, metadata=None):
        """Index both summaries and chunks."""
        # Create document summary
        full_text = "\n".join(chunks)
        summary = self.summary_model.summarize(full_text)

        # Store summary with document-level metadata
        self.vector_store.add_documents(
            [summary],
            metadatas=[{
                "type": "summary",
                "doc_id": doc_id,
                "chunk_ids": list(range(len(chunks))),
                **(metadata or {})
            }],
            ids=[f"summary_{doc_id}"]
        )

        # Store individual chunks
        for i, chunk in enumerate(chunks):
            self.vector_store.add_documents(
                [chunk],
                metadatas=[{
                    "type": "chunk",
                    "doc_id": doc_id,
                    "chunk_index": i,
                    **(metadata or {})
                }],
                ids=[f"chunk_{doc_id}_{i}"]
            )

    def retrieve(self, query, k=5):
        """Two-stage retrieval: summaries first, then chunks."""
        # Stage 1: Retrieve summaries
        summary_results = self.vector_store.similarity_search(
            query, k=k, filter={"type": "summary"}
        )

        # Stage 2: Retrieve chunks from top documents
        doc_ids = [r.metadata["doc_id"] for r in summary_results]
        chunk_results = self.vector_store.similarity_search(
            query, k=k, filter={"type": "chunk", "doc_id": {"$in": doc_ids}}
        )

        return chunk_results

class ContextualRetrieval:
    """
    Enriches each chunk with context from the document before indexing.
    Based on Anthropic's contextual retrieval approach.
    """

    def __init__(self, llm, embedding_model):
        self.llm = llm
        self.embedding_model = embedding_model

    def prepare_chunk(self, chunk, document_context):
        """Add explanatory context to each chunk."""
        context_prompt = f"""
        Here is a chunk from a larger document:
        <chunk>{chunk}</chunk>

        Here is the document context surrounding this chunk:
        <context>{document_context}</context>

        Provide a concise context for this chunk that would help a reader
        understand its place in the document. Include:
        - What is this chunk about
        - How it relates to surrounding content
        - Key entities or topics referenced

        Context:
        """

        context = self.llm.invoke(context_prompt)

        # Store the enriched chunk
        enriched_chunk = f"[Context: {context}]\n\n{chunk}"
        return enriched_chunk

    def index_document(self, chunks):
        """Index chunks with contextual enrichment."""
        enriched_chunks = []

        for i, chunk in enumerate(chunks):
            # Get surrounding context
            surrounding = self._get_surrounding_context(chunks, i)
            enriched = self.prepare_chunk(chunk, surrounding)
            enriched_chunks.append(enriched)

        # Index enriched chunks
        embeddings = self.embedding_model.embed_documents(enriched_chunks)
        return enriched_chunks, embeddings

    def _get_surrounding_context(self, chunks, idx, window=2):
        """Get text surrounding a chunk."""
        start = max(0, idx - window)
        end = min(len(chunks), idx + window + 1)
        return "\n...

class ParentChildRetriever:
    """
    Parent-child chunking: retrieve child chunks but return parent context.
    """
    def __init__(self, vector_store, parent_size=1000, child_size=200):
        self.vector_store = vector_store
        self.parent_size = parent_size
        self.child_size = child_size

    def index_document(self, doc_id, text):
        """Create parent and child chunks."""
        # Create parent chunks
        parent_chunks = self._split(text, self.parent_size, overlap=100)

        for parent_idx, parent in enumerate(parent_chunks):
            parent_id = f"parent_{doc_id}_{parent_idx}"

            # Store parent
            self.vector_store.add_documents(
                [parent],
                metadatas=[{
                    "type": "parent",
                    "doc_id": doc_id,
                    "parent_id": parent_id
                }],
                ids=[parent_id]
            )

            # Create and store child chunks
            child_chunks = self._split(parent, self.child_size, overlap=20)
            for child_idx, child in enumerate(child_chunks):
                child_id = f"child_{doc_id}_{parent_idx}_{child_idx}"
                self.vector_store.add_documents(
                    [child],
                    metadatas=[{
                        "type": "child",
                        "doc_id": doc_id,
                        "parent_id": parent_id,
                        "child_id": child_id
                    }],
                    ids=[child_id]
                )

    def retrieve(self, query, k=5):
        """Retrieve child chunks and return parent context."""
        # Retrieve child chunks
        child_results = self.vector_store.similarity_search(
            query, k=k, filter={"type": "child"}
        )

        # Get unique parent IDs
        parent_ids = list(set(r.metadata["parent_id"] for r in child_results))

        # Retrieve parent chunks
        parent_results = self.vector_store.similarity_search(
            query, k=len(parent_ids), filter={"type": "parent", "parent_id": {"$in": parent_ids}}
        )

        return parent_results, child_results

class MultiLevelIndex:
    def __init__(self, vector_store):
        self.vector_store = vector_store

    def index_structured_document(self, doc_id, sections):
        """Index document with section and chunk levels.

        sections: list of {heading, content, sub_sections}
        """
        for i, section in enumerate(sections):
            section_id = f"sec_{doc_id}_{i}"

            # Index section summary
            section_summary = f"{section['heading']}: {self._summarize(section['content'])}"
            self.vector_store.add_documents(
                [section_summary],
                metadatas=[{"level": "section", "doc_id": doc_id, "section_id": section_id}],
                ids=[section_id]
            )

            # Index chunks within section
            chunks = self._chunk_content(section['content'])
            for j, chunk in enumerate(chunks):
                chunk_id = f"ch_{doc_id}_{i}_{j}"
                self.vector_store.add_documents(
                    [chunk],
                    metadatas=[{
                        "level": "chunk",
                        "doc_id": doc_id,
                        "section_id": section_id,
                        "heading": section['heading']
                    }],
                    ids=[chunk_id]
                )

    def iterative_retrieve(self, query, k_final=5):
        """Iterative retrieval: sections first, then chunks."""
        # Level 1: Retrieve relevant sections
        section_results = self.vector_store.similarity_search(
            query, k=3, filter={"level": "section"}
        )

        section_ids = [r.metadata["section_id"] for r in section_results]

        # Level 2: Retrieve chunks from relevant sections
        chunk_results = self.vector_store.similarity_search(
            query, k=k_final,
            filter={"level": "chunk", "section_id": {"$in": section_ids}}
        )

        return chunk_results

class IterativeRetriever:
    def __init__(self, vector_store, llm, max_iterations=3):
        self.vector_store = vector_store
        self.llm = llm
        self.max_iterations = max_iterations

    def retrieve(self, query, k=5):
        """Iteratively refine retrieval based on intermediate results."""
        all_results = []
        current_query = query

        for iteration in r...

class HyDE:
    """
    Hypothetical Document Embeddings:
    1. Generate a hypothetical document that would answer the query
    2. Embed the hypothetical document
    3. Retrieve chunks similar to the hypothetical document
    """

    def __init__(self, llm, embedding_model, vector_store):
        self.llm = llm
        self.embedding_model = embedding_model
        self.vector_store = vector_store

    def generate_hypothetical_document(self, query):
        """Generate a hypothetical document that answers the query."""
        prompt = f"""
        Given the following question, write a paragraph that would serve
        as a perfect answer. Include specific details and factual information.

        Question: {query}

        Hypothetical Answer:
        """

        response = self.llm.invoke(prompt)
        return response.content

    def retrieve(self, query, k=5):
        """Retrieve using HyDE."""
        # Step 1: Generate hypothetical document
        hypothetical_doc = self.generate_hypothetical_document(query)

        # Step 2: Embed the hypothetical document
        hyde_embedding = self.embedding_model.embed_query(hypothetical_doc)

        # Step 3: Retrieve using hypothetical document embedding
        results = self.vector_store.similarity_search_by_vector(
            hyde_embedding, k=k
        )

        return results, hypothetical_doc

class InverseHyDE:
    """
    Alternative approach: retrieve first, then generate hypothetical 
    document for re-ranking.
    """

    def __init__(self, llm, embedding_model, vector_store):
        self.llm = llm
        self.embedding_model = embedding_model
        self.vector_store = vector_store

    def retrieve(self, query, k_initial=20, k_final=5):
        """Retrieve, generate HyDE, re-retrieve."""
        # Step 1: Initial retrieval with query
        initial_results = self.vector_store.similarity_search(query, k=k_initial)

        # Step 2: Generate hypothetical document from initial results
        context = "\n\n".join([r.page_content[:500] for r in initial_results[:3]])
        hypo_doc = self.llm.invoke(f"""
        Question: {query}

        Based on these sources, generate a comprehensive answer:
        {context}

        Answer:
        """)

        # Step 3: Re-retrieve using hypothetical document
        hyde_emb = self.embedding_model.embed_query(hypo_doc.content)
        final_results = self.vector_store.similarity_search_by_vector(
            hyde_emb, k=k_final
        )

        return final_results

class MultiHyDE:
    def generate_diverse_hypotheticals(self, query, n=3):
        """Generate multiple hypothetical documents from different angles."""
        prompts = [
            f"Write a detailed technical answer to: {query}",
            f"Write a simple, accessible explanation for: {query}",
            f"Write a critical analysis addressing: {query}"
        ]

        hypotheticals = []
        for prompt in prompts:
            response = self.llm.invoke(prompt)
            hypotheticals.append(response.content)

        return hypotheticals

    def retrieve(self, query, k=5):
        """Retrieve using multiple hypothetical documents."""
        hypotheticals = self.generate_diverse_hypotheticals(query)

        all_results = []
        for hypo in hypotheticals:
            embedding = self.embedding_model.embed_query(hypo)
            results = self.vector_store.similarity_search_by_vector(embedding, k=k)
            all_results.extend(results)

        # Deduplicate and rank
        return self.deduplicate_and_rank(all_results, k)

class SelfCritiqueHyDE:
    def retrieve(self, query, k=5):
        """HyDE with self-critique of hypothetical document quality."""
        # Generate initial hypothetical document
        hypo = self.generate_hypothetical_document(query)

        # Critique and refine
        critique = self.llm.invoke(f"""
        Evaluate this hypothetical answer for:
        1. Relevance to the question
        2. Factual accuracy
        3. Completeness

        Question: {query}
        Answer: {hypo}

        Suggested improvements:
        """)

        # Generate improved hypothetical document
        improved_hypo = self.llm.invoke(f"""
        Original question: {query}
        Previous attempt: {hypo}
        Critique: {critique.content}

        Write an improved answer:
        """)

        # Retrieve with improved hypothetical document
        embedding = self.embedding_model.embed_query(improved_hypo.content)
        return self.vector_store.similarity_search_by_vector(embedding, k=k)

from sentence_transformers import CrossEncoder

class CrossEncoderReranker:
    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)

    def rerank(self, query, documents, k=10):
        """Score and re-rank documents."""
        pairs = [[query, doc] for doc in documents]
        scores = self.model.predict(pairs)

        # Sort by score descending
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)

        return [doc for doc, score in scored_docs[:k]], [score for doc, score in scored_docs[:k]]

import cohere

class CohereReranker:
    def __init__(self, api_key, model="rerank-english-v3.0"):
        self.client = cohere.Client(api_key)
        self.model = model

    def rerank(self, query, documents, k=10):
        """Use Cohere's rerank API."""
        response = self.client.rerank(
            model=self.model,
            query=query,
            documents=documents,
            top_n=k
        )

        reranked = []
        for result in response.results:
            reranked.append({
                "document": documents[result.index],
                "relevance_score": result.relevance_score
            })

        return reranked

from FlagEmbedding import FlagReranker

class BGEReranker:
    def __init__(self, model_name="BAAI/bge-reranker-v2-m3"):
        self.model = FlagReranker(model_name, use_fp16=True)

    def rerank(self, query, documents, k=10):
        """BGE re-ranker from BAAI."""
        pairs = [[query, doc] for doc in documents]
        scores = self.model.compute_score(pairs)

        scored = list(zip(documents, scores))
        scored.sort(key=lambda x: x[1], reverse=True)

        return scored[:k]

class MonoBERTReranker:
    def __init__(self, model_name="monobert-base-msmarco"):
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

    def rerank(self, query, documents, k=10):
        """monoBERT re-ranker."""
        features = self.tokenizer(
            [query] * len(documents),
            documents,
            padding=True,
            truncation=True,
            return_tensors="pt",
            max_length=512
        )

        with torch.no_grad():
            scores = self.model(**features).logits.squeeze(-1).numpy()

        scored = list(zip(documents, scores))
        scored.sort(key=lambda x: x[1], reverse=True)

        return scored[:k]

class ColBERTv2Reranker:
    """
    ColBERTv2 late interaction: token-level matching instead of single score.
    """
    def __init__(self, checkpoint="colbert-ir/colbertv2.0"):
        from colbert import Searcher
        self.searcher = Searcher(index="index_path", checkpoint=checkpoint)

    def search(self, query, k=10):
        """ColBERTv2 search with late interaction scoring."""
        results = self.searcher.search(query, k=k)

        # results: (rank, passage_id, score) tuples
        return results

    def rerank(self, query, documents, k=10):
        """Late interaction re-ranking."""
        # Encode query and documents
        query_emb = self.encode_query(query)
        doc_embs = [self.encode_doc(doc) for doc in documents]

        # Late interaction: max similarity over all token pairs
        scores = []
        for doc_emb in doc_embs:
            # (num_query_tokens, num_doc_tokens)
            sim_matrix = torch.matmul(query_emb, doc_emb.T)
            max_sim = sim_matrix.max(dim=-1).values.sum()
            scores.append(max_sim.item())

        scored = l...

class DenseRetriever:
    def __init__(self, embedding_model, vector_store):
        self.embedding_model = embedding_model
        self.vector_store = vector_store

    def search(self, query, k=10):
        query_emb = self.embedding_model.embed_query(query)
        results = self.vector_store.similarity_search_by_vector(query_emb, k=k)
        return results

from transformers import AutoModelForMaskedLM, AutoTokenizer

class SPLADERetriever:
    """
    SPLADE: Sparse Lexical and Expansion Model.
    Learns to expand queries with relevant terms and produce sparse vectors.
    """
    def __init__(self, model_name="naver/splade-cocondenser-selfdistil"):
        self.model = AutoModelForMaskedLM.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

    def encode(self, text):
        """Generate sparse vector representation."""
        tokens = self.tokenizer(text, return_tensors="pt")
        with torch.no_grad():
            output = self.model(**tokens)

        # SPLADE uses ReLU + log1p activation
        sparse_emb = torch.log1p(torch.relu(output.logits))
        sparse_emb = sparse_emb.max(dim=1).values

        return sparse_emb.squeeze()

    def search(self, query, index, k=10):
        """SPLADE search over pre-computed sparse index."""
        query_vec = self.encode(query)
        scores = {}

        for doc_id, doc_vec in index.items():
            score = (query_vec * doc_vec).sum().item()
            scores[doc_id] = score

        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_docs[:k]

from rank_bm25 import BM25Okapi

class BM25Retriever:
    """
    BM25: Probabilistic retrieval model based on term frequency and document length.
    """
    def __init__(self):
        self.bm25 = None
        self.documents = []

    def index(self, documents):
        """Index documents with BM25."""
        tokenized_corpus = [self._tokenize(doc) for doc in documents]
        self.bm25 = BM25Okapi(tokenized_corpus)
        self.documents = documents

    def search(self, query, k=10):
        """Search using BM25."""
        tokenized_query = self._tokenize(query)
        scores = self.bm25.get_scores(tokenized_query)

        # Get top-k indices
        top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]

        return [(self.documents[i], scores[i]) for i in top_indices]

    def _tokenize(self, text):
        """Simple tokenization."""
        return text.lower().split()

class UniCOILRetriever:
    """
    uniCOIL: Expanded term weighting with COIL architecture.
    """
    def __init__(self, model_name="castorini/unicoil-d2v-msmarco-passage"):
        self.model = T5EncoderModel.from_pretrained(model_name)
        self.tokenizer = T5Tokenizer.from_pretrained(model_name)

    def encode(self, text):
        """Generate term-weighted sparse representation."""
        inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
        with torch.no_grad():
            outputs = self.model(**inputs)

        # uniCOIL generates term importance weights
        weights = outputs.last_hidden_state.mean(dim=1).squeeze()
        return weights

from elasticsearch import Elasticsearch

class ElasticsearchKeywordRetriever:
    def __init__(self, host="localhost", port=9200, index="documents"):
        self.es = Elasticsearch([f"http://{host}:{port}"])
        self.index = index

    def create_index(self):
        """Create index with keyword search configuration."""
        mapping = {
            "mappings": {
                "properties": {
                    "text": {
                        "type": "text",
                        "analyzer": "standard",
                        "similarity": "BM25"
                    }
                }
            }
        }
        self.es.indices.create(index=self

class RRF:
    """
    Reciprocal Rank Fusion: Combines multiple ranked lists.
    RRF score = sum(1 / (k + rank(doc, list)))
    """
    def __init__(self, k=60):
        self.k = k

    def fuse(self, ranked_lists):
        """
        ranked_lists: list of lists, each is [(doc_id, score), ...]
        Returns: fused ranking [(doc_id, rrf_score), ...]
        """
        rrf_scores = {}

        for ranked_list in ranked_lists:
            for rank, (doc_id, _) in enumerate(ranked_list):
                if doc_id not in rrf_scores:
                    rrf_scores[doc_id] = 0
                rrf_scores[doc_id] += 1.0 / (self.k + rank + 1)

        # Sort by RRF score descending
        fused = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
        return fused

class WeightedFusion:
    """
    Weighted combination of scores from different retrieval methods.
    """
    def __init__(self, weights=None):
        self.weights = weights or {"dense": 0.5, "sparse": 0.3, "keyword": 0.2}

    def fuse(self, results_dict):
        """
        results_dict: {"method_name": {doc_id: score, ...}, ...}
        """
        fused_scores = {}

        for method, docs in results_dict.items():
            weight = self.weights.get(method, 1.0)

            # Normalize scores within each method
            max_score = max(docs.values()) if docs else 1.0
            for doc_id, score in docs.items():
                normalized = score / max_score if max_score > 0 else 0
                if doc_id not in fused_scores:
                    fused_scores[doc_id] = 0
                fused_scores[doc_id] += weight * normalized

        return sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)

class LinearCombinationFusion:
    """
    Simple linear combination of normalized scores.
    """
    def combine(self, scores_list, weights=None):
        """
        scores_list: list of dicts {doc_id: normalized_score}
        weights: list of weights for each score dict
        """
        if weights is None:
            weights = [1.0/len(scores_list)] * len(scores_list)

        combined = {}
        for scores, weight in zip(scores_list, weights):
            if scores:
                max_score = max(scores.values())
                for doc_id, score in scores.items():
                    normalized = score / max_score
                    combined[doc_id] = combined.get(doc_id, 0) + weight * normalized

        return sorted(combined.items(), key=lambda x: x[1], reverse=True)

from sklearn.ensemble import GradientBoostingRegressor

class LearningToRank:
    """
    Learned fusion using gradient boosted trees.
    """
    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=100,
            max_depth=3,
            learning_rate=0.1
        )

    def extract_features(self, doc, query):
        """Extract ranking features."""
        features = {
            "bm25_score": compute_bm25(doc, query),
            "dense_similarity": cosine_similarity(doc_emb, query_emb),
            "cross_encoder_score": cross_encoder_score(query, doc),
            "doc_length": len(doc),
            "query_term_overlap": term_overlap(query, doc),
            "title_match": title_match(query, doc),
        }
        return list(features.values())

    def train(self, training_pairs):
        """
        training_pairs: [(query, doc, relevance_label), ...]
        """
        X, y = [], []
        for query, doc, label in training_pairs:
            features = self.extract_features(doc, query)
            X.append(features)
            y.append(label)

        self.model.fit(X, y)

    def predict(self, query, documents):
        """Score documents using learned model."""
        scores = []
        for doc in documents:
            features = self.extract_features(doc, query)
            score = self.model.predict([features])[0]
            scores.append(score)

        return scores

class QueryRewriter:
    def __init__(self, llm):
        self.llm = llm

    def rewrite_with_history(self, current_query, history):
        """Rewrite query incorporating conversation history."""
        prompt = f"""
        Conversation history:
        {history}

        Current query: {current_query}

        Rewrite the current query to include necessary context from the 
        conversation history. Output only the rewritten query.
        """

        response = self.llm.invoke(prompt)
        return response.content.strip()

    def rewrite_for_retrieval(self, query):
        """Rewrite query to be more effective for retrieval."""
        prompt = f"""
        Original query: {query}

        Rewrite this query to improve search retrieval effectiveness:
        - Use keywords that would appear in relevant documents
        - Remove conversational filler
        - Make it specific and searchable

        Rewritten query:
        """

        response = self.llm.invoke(prompt)
        return response.content.strip()

class QueryExpander:
    def __init__(self, llm):
        self.llm = llm

    def expand_with_terms(self, query, num_expansions=5):
        """Expand query with related terms."""
        prompt = f"""
        Query: {query}

        Generate {num_expansions} alternative phrasings or related terms 
        for this query that would help find relevant documents.
        Output one per line.
        """

        response = self.llm.invoke(prompt)
        expansions = [q.strip() for q in response.content.strip().split('\n') if q.strip()]

        return [query] + expansions[:num_expansions]

    def expand_with_generated_docs(self, query, num_docs=3):
        """Generate hypothetical relevant documents and extract terms."""
        prompt = f"""
        For the query: "{query}"

        Generate {num_docs} short relevant document snippets that would 
        ideally answer this query.
        """

        response = self.llm.invoke(prompt)
        generated_docs = response.content.strip()

        # Extract key terms from generated docs
        extraction_prompt = f"""
        From these documents relevant to "{query}":
        {generated_docs}

        Extract 5 key search terms or phrases that would be most useful for retrieval.
        """

        terms_response = self.llm.invoke(extraction_prompt)
        terms = [t.strip() for t in terms_response.content.strip().split('\n') if t.strip()]

        return [query] + terms

class StepBackPrompting:
    """
    Step-back prompting: generate a more abstract query to retrieve
    high-level context, then answer the specific query.
    """
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever

    def step_back(self, query):
        """Generate a step-back (more abstract) question."""
        prompt = f"""
        Original question: {query}

        What broader topic or concept does this question relate to?
        Generate a more general question that would help answer the original.

        Step-back question:
        """

        response = self.llm.invoke(prompt)
        return response.content.strip()

    def retrieve_with_step_back(self, query, k=5):
        """Retrieve using both specific and step-back queries."""
        # Get step-back query
        stepback_query = self.step_back(query)

        # Retrieve with both queries
        specific_results = self.retriever.retrieve(query, k=k)
        general_results = self.retriever.retrieve(stepback_query, k=k)

        # Combine results
        combined = self._combine_results(specific_results, general_results)
        return combined

class QueryDecomposer:
    def __init__(self, llm):
        self.llm = llm

    def decompose(self, complex_query):
        """Decompose multi-hop query into sub-queries."""
        prompt = f"""
        Complex query: {complex_query}

        Break this down into simpler sub-queries that can be answered 
        independently and then combined.
        Output one sub-query per line.
        """

        response = self.llm.invoke(prompt)
        sub_queries = [q.strip() for q in response.content.strip().split('\n') if q.strip()]
        return sub_queries

    def parallel_retrieve(self, complex_query, retriever):
        """Retrieve for all sub-queries in parallel."""
        sub_queries = self.decompose(complex_query)

        all_results = []
        for sub_query in sub_queries:
            results = retriever.retrieve(sub_query)
            all_results.extend(results)

        return all_results

class FixedRouter:
    """Route based on fixed rules."""

    def route(self, query):
        """Simple rule-based routing."""
        if "news" in query.lower() or "latest" in query.lower():
            return "web_search"
        elif "code" in query.lower() or "function" in query.lower():
            return "code_index"
        elif "who" in query.lower() or "what" in query.lower():
            return "knowledge_base"
        else:
            return "general_retriever"

class ContentBasedRouter:
    """Route based on query content analysis."""

    def __init__(self):
        self.classifier = load_classifier()  # e.g., zero-shot classifier

    def route(self, query):
        """Classify query domain and route accordingly."""
        domains = ["technical", "medical", "legal", "general", "creative"]
        classification = self.classifier(query, domains)

        domain = classification[0]["label"]
        return self._domain_to_index(domain)

    def _domain_to_index(self, domain):
        router = {
            "technical": "tech_docs_index",
            "medical": "medical_knowledge_base",
            "legal": "legal_corpus",
            "creative": "general_index",
            "general": "general_index"
        }
        return router.get(domain, "general_index")

class LLMBasedRouter:
    """Use LLM to determine best routing strategy."""

    def __init__(self, llm):
        self.llm = llm

    def route(self, query, available_strategies):
        """LLM decides which retrieval strategy to use."""
        prompt = f"""
        Query: {query}

        Available strategies:
        {available_strategies}

        Which strategy is best suited for this query?
        Consider: query complexity, required knowledge domain, specificity.

        Strategy:
        Reason:
        """

        response = self.llm.invoke(prompt)
        # Parse strategy from response
        return self._parse_strategy(response.content)

    def route_with_parameters(self, query):
        """Get routing decision with retrieval parameters."""
        prompt = f"""
        Query: {query}

        Decide:
        1. Retrieval method (dense, sparse, hybrid)
        2. Number of documents to retrieve
        3. Re-ranker needed (yes/no)
        4. Query expansion needed (yes/no)

        Output JSON:
        {{"method": "...", "k": 5, "rerank": true, "expand": false}}
        """

        response = self.llm.invoke(prompt)
        return json.loads(response.content)

class MetadataRouter:
    """Route based on query metadata or context."""

    def route(self, query, user_context=None):
        """Route based on user and query metadata."""
        route_decision = {
            "index": "general",
            "filters": {}
        }

        if user_context:
            # User-specific routing
            if user_context.get("department") == "engineering":
                route_decision["index"] = "technical_docs"
            elif user_context.get("department") == "legal":
                route_decision["index"] = "legal_docs"

            # Date filter
            if user_context.get("timeframe"):
                route_decision["filters"]["date"] = {
                    "$gte": user_context["timeframe"]["start"],
                    "$lte": user_context["timeframe"]["end"]
                }

        # Content-based metadata routing
        if "product" in query.lower():
            route_decision["filters"]["category"] = "product"

        return route_decision

class IterativeMultiHopRAG:
    """
    Iterative retrieval: retrieve → generate → extract queries → retrieve again
    """
    def __init__(self, llm, retriever, max_hops=3):
        self.llm = llm
        self.retriever = retriever
        self.max_hops = max_hops

    def answer(self, question):
        """Multi-hop RAG with iterative retrieval."""
        context = []

        for hop in range(self.max_hops):
            # Retrieve
            results = self.retriever.retrieve(question, k=5)
            new_context = [r.page_content for r in results]
            context.extend(new_context)

            if hop < self.max_hops - 1:
                # Determine if more information is needed
                analysis = self.llm.invoke(f"""
                Question: {question}

                Known information:
                {chr(10).join(context)}

                Can this question be answered with current information?
                If no, what additional information is needed?
                Generate a follow-up search query.

                Answerable (yes/no):
                Follow-up query (if not answerable):
                """)

                if "yes" in analysis.content.lower().split('\n')[0]:
                    break

                # Extract follow-up query
                lines = analysis.content.strip().split('\n')
                for line in lines:
                    if "follow-up" in line.lower() or "query" in line.lower():
                        question = line.split(":")[-1].strip()
                        break

        # Generate final answer
        final_prompt = f"""
        Question: {question}

        Retrieved information:
        {chr(10).join(context)}

        Provide a comprehensive answer based on the retrieved information:
        """

        return self.llm.invoke(final_prompt).content

class DecompositionMultiHop:
    """
    Decompose multi-hop question into sub-questions and answer sequentially.
    """
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever

    def decompose_and_answer(self, question):
        """Decompose into sub-questions and answer sequentially."""
        # Step 1: Decompose
        sub_questions = self._decompose(question)

        # Step 2: Answer sub-questions sequentially
        intermediate_answers = {}
        for i, sub_q in enumerate(sub_questions):
            # Retrieve with context from previous answers
            context = list(intermediate_answers.values())
            results = self.retriever.retrieve(
                sub_q + " " + " ".join(context), k=5
            )

            # Answer sub-question
            answer = self.llm.invoke(f"""
            Sub-question: {sub_q}

            Retrieved:
            {chr(10).join([r.page_content for r in results])}

            Previous context: {chr(10).join(context)}

            Answer:
            """)

            intermediate_answers[f"step_{i}"] = answer.content

        # Step 3: Synthesize final answer
        final = self.llm.invoke(f"""
        Original question: {question}

        Intermediate answers:
        {chr(10).join([f"{k}: {v}" for k, v in intermediate_answers.items()])}

        Final answer:
        """)

        return final.content

    def _decompose(self, question):
        """Use LLM to break down multi-hop question."""
        prompt = f"""
        Break this multi-hop question into sequential sub-questions.
        Each sub-question should be answerable from retrieved documents.

        Question: {question}

        Sub-questions (one per line):
        """

        response = self.llm.invoke(prompt)
        return [q.strip() for q in response.content.strip().split('\n') if q.strip()]