RAG Pipeline Optimization Strategy: Chunking, Reranking, and Hybrid Search

[Indexing Phase]
Document Collection -> Preprocessing -> Chunking -> Embedding -> Vector DB Storage + Inverted Index Storage

[Query Phase]
Question -> Query Transformation -> Hybrid Search (Dense + Sparse)
         -> Reranking -> Top K Selection -> Prompt Construction -> LLM Generation

from langchain_text_splitters import CharacterTextSplitter

# Fixed-size chunking - the most basic approach
fixed_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=512,
    chunk_overlap=50,       # 10% overlap recommended
    length_function=len,
    is_separator_regex=False,
)

chunks = fixed_splitter.split_text(document_text)
print(f"Total {len(chunks)} chunks created")

from langchain_text_splitters import RecursiveCharacterTextSplitter

# Recursive chunking - recommended production default
recursive_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ". ", " ", ""],
    chunk_size=512,
    chunk_overlap=64,
    length_function=len,
    add_start_index=True,   # For source position tracking
)

chunks = recursive_splitter.split_documents(documents)

# Each chunk automatically includes metadata
for chunk in chunks[:3]:
    print(f"Size: {len(chunk.page_content)}, "
          f"Start position: {chunk.metadata.get('start_index')}")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Semantic chunking - splitting by meaning units
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",  # percentile, standard_deviation, interquartile
    breakpoint_threshold_amount=75,          # Split at 75th percentile difference
)

semantic_chunks = semantic_splitter.split_text(document_text)
print(f"Semantic chunk count: {len(semantic_chunks)}")
print(f"Average chunk length: {sum(len(c) for c in semantic_chunks) / len(semantic_chunks):.0f}")

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text: str, max_chunks: int = 20) -> list[dict]:
    """LLM-based agentic chunking"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Split the given text into logical units. "
                    "Each chunk should cover one complete topic. "
                    "Return as JSON array: "
                    '[{"title": "chunk title", "content": "chunk content", "summary": "one-line summary"}]'
                ),
            },
            {"role": "user", "content": text[:8000]},  # Watch token limits
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return result.get("chunks", [])

# Usage example
chunks = agentic_chunk(long_document_text)
for chunk in chunks:
    print(f"[{chunk['title']}] {chunk['summary']}")

from langchain_openai import OpenAIEmbeddings

# OpenAI embeddings - leveraging dimension reduction
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1024,  # Reduce 3072 -> 1024 for cost/storage savings
)

# BGE-M3: Simultaneous Dense + Sparse generation (optimal for hybrid search)
from FlagEmbedding import BGEM3FlagModel

bge_model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

# Generate Dense and Sparse vectors simultaneously
output = bge_model.encode(
    ["RAG pipeline optimization methodology"],
    return_dense=True,
    return_sparse=True,
)

dense_vector = output["dense_vecs"][0]    # (1024,) float vector
sparse_vector = output["lexical_weights"][0]  # Sparse vector (per-word weights)

print(f"Dense dimensions: {len(dense_vector)}")
print(f"Sparse active token count: {len(sparse_vector)}")

Query: "What is the CPU threshold in Kubernetes HPA settings?"

Dense Search (vector similarity):
  -> "How to configure autoscaling in container orchestration" (semantically similar)

Sparse Search (BM25 keyword matching):
  -> "Set HPA targetCPUUtilizationPercentage to 80" (exact keyword match)

Hybrid Combination (RRF or weighted sum):
  -> Merge both results for optimal document retrieval

import weaviate, { WeaviateClient } from 'weaviate-client'

// Connect Weaviate client
const client: WeaviateClient = await weaviate.connectToLocal({
  host: 'localhost',
  port: 8080,
})

// Execute hybrid search
const collection = client.collections.get('Documents')

const result = await collection.query.hybrid('Kubernetes HPA CPU threshold', {
  alpha: 0.7, // 0 = pure BM25, 1 = pure vector, 0.7 = 70% vector
  limit: 20, // Candidates before reranking
  fusionType: 'RelativeScore', // RelativeScore or Ranked
  returnMetadata: ['score', 'explainScore'],
  returnProperties: ['title', 'content', 'source'],
})

for (const item of result.objects) {
  console.log(`[${item.metadata?.score?.toFixed(3)}] ${item.properties.title}`)
}

from rank_bm25 import BM25Okapi
import numpy as np
from typing import List, Tuple

def hybrid_search(
    query: str,
    documents: list[dict],
    dense_scores: np.ndarray,
    k: int = 10,
    alpha: float = 0.7,
    rrf_k: int = 60,
) -> list[dict]:
    """
    RRF-based hybrid search
    alpha: Dense search weight (1-alpha is Sparse weight)
    rrf_k: RRF constant (default 60, paper recommended)
    """
    # Sparse search (BM25)
    tokenized_docs = [doc["content"].split() for doc in documents]
    bm25 = BM25Okapi(tokenized_docs)
    sparse_scores = bm25.get_scores(query.split())

    # Dense rank calculation
    dense_ranks = np.argsort(-dense_scores) + 1  # 1-indexed rank
    sparse_ranks = np.argsort(-sparse_scores) + 1

    # RRF score calculation
    rrf_scores = []
    for i in range(len(documents)):
        dense_rrf = alpha / (rrf_k + dense_ranks[i])
        sparse_rrf = (1 - alpha) / (rrf_k + sparse_ranks[i])
        rrf_scores.append(dense_rrf + sparse_rrf)

    # Return top K
    top_indices = np.argsort(rrf_scores)[::-1][:k]
    return [
        {**documents[i], "hybrid_score": rrf_scores[i]}
        for i in top_indices
    ]

# Usage example
results = hybrid_search(
    query="Kubernetes HPA CPU threshold",
    documents=all_documents,
    dense_scores=embedding_similarity_scores,
    k=20,       # Generous before reranking
    alpha=0.7,  # Dense 70%, Sparse 30%
)

import cohere

co = cohere.ClientV2(api_key="your-cohere-api-key")

def rerank_with_cohere(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """Rerank documents with Cohere Rerank 4"""
    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
        return_documents=True,
    )

    results = []
    for item in response.results:
        results.append({
            "index": item.index,
            "score": item.relevance_score,
            "text": item.document.text if item.document else documents[item.index],
        })
    return results

# Usage: Rerank 20 hybrid search results to select top 5
hybrid_results = hybrid_search(query, documents, dense_scores, k=20)
reranked = rerank_with_cohere(
    query="How to set Kubernetes HPA CPU threshold",
    documents=[r["content"] for r in hybrid_results],
    top_n=5,
)

for r in reranked:
    print(f"[{r['score']:.4f}] {r['text'][:80]}...")

from FlagEmbedding import FlagReranker

# BGE Reranker v2 M3 - lightweight multilingual reranker
reranker = FlagReranker(
    "BAAI/bge-reranker-v2-m3",
    use_fp16=True,  # Save GPU memory
)

def rerank_with_bge(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """Rerank documents with BGE Reranker"""
    # Create query-document pairs
    pairs = [[query, doc] for doc in documents]

    # Calculate relevance scores
    scores = reranker.compute_score(pairs, normalize=True)

    # Sort by score
    scored_docs = [
        {"index": i, "score": s, "text": d}
        for i, (s, d) in enumerate(zip(scores, documents))
    ]
    scored_docs.sort(key=lambda x: x["score"], reverse=True)

    return scored_docs[:top_n]

# Usage example
results = rerank_with_bge(
    query="What criteria determine chunking size in RAG pipelines?",
    documents=candidate_documents,
    top_n=5,
)

for r in results:
    print(f"[{r['score']:.4f}] {r['text'][:100]}...")

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class OptimizedRAGPipeline:
    """Advanced RAG Pipeline"""

    def __init__(self):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512, chunk_overlap=64
        )
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-large", dimensions=1024
        )
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)

    def query(self, question: str, top_k: int = 5) -> str:
        # 1. Hybrid search: 20 candidates
        candidates = self._hybrid_search(question, k=20)

        # 2. Reranking: select top 5
        reranked = self._rerank(question, candidates, top_n=top_k)

        # 3. Prompt construction and LLM generation
        context = "\n\n---\n\n".join([doc["text"] for doc in reranked])
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer the question based on the following context.\n\n{context}"),
            ("human", "{question}"),
        ])
        chain = prompt | self.llm
        response = chain.invoke({"context": context, "question": question})
        return response.content

    def _hybrid_search(self, query: str, k: int = 20) -> list[dict]:
        # Dense + Sparse hybrid search (see code above)
        ...

    def _rerank(self, query: str, docs: list[dict], top_n: int) -> list[dict]:
        pairs = [[query, doc["text"]] for doc in docs]
        scores = self.reranker.compute_score(pairs, normalize=True)
        for doc, score in zip(docs, scores):
            doc["rerank_score"] = score
        docs.sort(key=lambda x: x["rerank_score"], reverse=True)
        return docs[:top_n]

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Evaluation dataset construction
eval_data = {
    "question": [
        "What is the default CPU threshold for Kubernetes HPA?",
        "What criteria determine chunking size in RAG?",
    ],
    "answer": [
        "The default CPU threshold for HPA is 80%.",
        "It is determined in the 256-1024 token range based on query type.",
    ],
    "contexts": [
        ["HPA uses targetCPUUtilizationPercentage default value of 80."],
        ["Factoid queries recommend 256-512, analytical queries 1024+ tokens."],
    ],
    "ground_truth": [
        "The default is 80%.",
        "It is determined by query type and document characteristics.",
    ],
}

dataset = Dataset.from_dict(eval_data)

# Run evaluation
results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.92, answer_relevancy: 0.88,
# context_precision: 0.85, context_recall: 0.90

from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    FaithfulnessMetric,
    ContextualRelevancyMetric,
    AnswerRelevancyMetric,
)

def test_rag_faithfulness():
    """Test that RAG responses are faithful to context"""
    test_case = LLMTestCase(
        input="What is the Kubernetes HPA CPU threshold?",
        actual_output="The default CPU threshold for HPA is 80%.",
        retrieval_context=[
            "HPA uses targetCPUUtilizationPercentage default value of 80."
        ],
    )

    faithfulness = FaithfulnessMetric(threshold=0.8)
    relevancy = ContextualRelevancyMetric(threshold=0.7)
    answer_rel = AnswerRelevancyMetric(threshold=0.7)

    assert_test(test_case, [faithfulness, relevancy, answer_rel])

# Run with pytest: pytest test_rag.py -v

# Blue-Green index switching example
import time

def reindex_with_blue_green(
    old_collection: str,
    new_collection: str,
    new_embedding_model: str,
):
    """Zero-downtime re-indexing"""
    # 1. Index into new collection (existing service uses old_collection)
    print(f"Starting indexing for new collection '{new_collection}'...")
    create_and_populate_collection(new_collection, new_embedding_model)

    # 2. Validation: run test queries against new collection
    test_results = run_evaluation_suite(new_collection)
    if test_results["context_precision"] < 0.8:
        raise ValueError(
            f"New index quality below threshold: {test_results['context_precision']:.2f}"
        )

    # 3. Traffic switch: alias or config change
    update_active_collection(new_collection)
    print(f"Traffic switched: {old_collection} -> {new_collection}")

    # 4. Keep old collection for rollback for a period
    schedule_cleanup(old_collection, delay_days=7)

import tiktoken

enc = tiktoken.encoding_for_model("text-embedding-3-small")

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=lambda text: len(enc.encode(text)),  # Token-count based
)

import asyncio

async def rerank_with_fallback(
    query: str,
    documents: list[str],
    top_n: int = 5,
    timeout: float = 2.0,
) -> list[dict]:
    """Reranking with fallback"""
    try:
        # Primary: Cohere Rerank (2s timeout)
        result = await asyncio.wait_for(
            cohere_rerank_async(query, documents, top_n),
            timeout=timeout,
        )
        return result
    except (asyncio.TimeoutError, Exception) as e:
        print(f"Cohere reranking failed, BGE fallback: {e}")
        try:
            # Fallback: Self-hosted BGE Reranker
            return rerank_with_bge(query, documents, top_n)
        except Exception as e2:
            print(f"BGE reranking also failed, returning original ranking: {e2}")
            # Final fallback: return hybrid search results as-is
            return [
                {"index": i, "score": 1.0 - i * 0.05, "text": d}
                for i, d in enumerate(documents[:top_n])
            ]