Dev Note

RAG in 2026: Advanced Retrieval Strategies Beyond Naive Vector Search
Retrieval-Augmented Generation (RAG) is now a standard building block in AI-powered applications. But the gap between a weekend RAG prototype and a production RAG system that users actually trust has never been more apparent. In 2026, basic vector similarity search — the approach that dominates tutorials — fails to meet the quality bar for real applications. This post covers the advanced patterns engineering teams are using to build RAG systems that actually work.
Photo by ZHENYU LUO on Unsplash
Why Naive RAG Fails
The basic RAG pipeline looks like this:
1. Chunk documents into fixed-size pieces (512 tokens)
2. Embed each chunk with a sentence transformer
3. At query time, embed the query, find top-K similar chunks
4. Stuff chunks into LLM context and generate answer
This fails in practice because:
Chunking breaks semantic units: a 512-token chunk might cut a table, code block, or argument in half
Single-vector search misses multi-hop questions: “What are the pros and cons of approach X mentioned in the architecture section?” requires multiple retrievals
No query understanding: the raw user query is often not the best search query
Recall vs. precision tradeoff: increasing K retrieves more relevant chunks but adds noise
Temporal confusion: no mechanism to prefer recent documents over stale ones
Pattern 1: Hierarchical Chunking
Instead of fixed-size chunks, structure documents hierarchically:
from dataclasses import dataclass
from typing import Optional
@dataclass
class DocumentNode:
    content: str
    node_type: str  # "document", "section", "paragraph", "sentence"
    parent_id: Optional[str]
    children_ids: list[str]
    metadata: dict
def hierarchical_chunk(document: str) -> list[DocumentNode]:
    nodes = []
    
    # Level 1: Full document summary (for high-level questions)
    doc_node = DocumentNode(
        content=summarize(document),
        node_type="document",
        parent_id=None,
        children_ids=[],
        metadata={"type": "summary"}
    )
    nodes.append(doc_node)
    
    # Level 2: Section-level chunks
    sections = split_by_headers(document)
    for section in sections:
        section_node = DocumentNode(
            content=section.text,
            node_type="section",
            parent_id=doc_node.id,
            children_ids=[],
            metadata={"header": section.header}
        )
        
        # Level 3: Paragraph chunks (what gets retrieved)
        paragraphs = split_paragraphs(section.text)
        for para in paragraphs:
            para_node = DocumentNode(
                content=para,
                node_type="paragraph",
                parent_id=section_node.id,
                children_ids=[],
                metadata={"section": section.header}
            )
            section_node.children_ids.append(para_node.id)
            nodes.append(para_node)
        
        doc_node.children_ids.append(section_node.id)
        nodes.append(section_node)
    
    return nodes
When a paragraph is retrieved, expand context by fetching its parent section. This gives the LLM the retrieved chunk with surrounding context, significantly reducing cases where an answer is cut off.
Pattern 2: Hybrid Search (BM25 + Dense Retrieval)
Pure vector search fails on exact keyword queries. A user asking for “error code E-4021” gets poor results from semantic search because the meaning of an error code is mostly in its precise syntax, not its semantics.
The solution: combine BM25 sparse retrieval (great for keywords, exact matches) with dense vector retrieval (great for semantic similarity):
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import numpy as np
class HybridRetriever:
    def __init__(self, documents: list[str], alpha: float = 0.5):
        self.alpha = alpha  # Weight for dense vs. sparse
        self.documents = documents
        
        # BM25 for sparse retrieval
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        
        # Dense embeddings
        self.model = SentenceTransformer("BAAI/bge-large-en-v1.5")
        self.embeddings = self.model.encode(documents, normalize_embeddings=True)
    
    def retrieve(self, query: str, top_k: int = 10) -> list[tuple[int, float]]:
        # Sparse scores
        bm25_scores = self.bm25.get_scores(query.split())
        bm25_normalized = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() + 1e-8)
        
        # Dense scores
        query_embedding = self.model.encode([query], normalize_embeddings=True)
        dense_scores = np.dot(self.embeddings, query_embedding.T).flatten()
        
        # Reciprocal Rank Fusion (RRF) — often better than linear combination
        combined = self.alpha * dense_scores + (1 - self.alpha) * bm25_normalized
        
        top_indices = np.argsort(combined)[::-1][:top_k]
        return [(int(idx), float(combined[idx])) for idx in top_indices]
Reciprocal Rank Fusion (RRF) is often more robust than linear combination because it’s less sensitive to score scale differences:
def reciprocal_rank_fusion(rankings: list[list[int]], k: int = 60) -> list[int]:
    scores = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
    return sorted(scores, key=scores.get, reverse=True)
Pattern 3: Query Transformation
The user’s query is often not the optimal retrieval query. Query transformation rewrites the query before retrieval:
async def transform_query(user_query: str) -> list[str]:
    """Generate multiple query variants for better retrieval coverage."""
    
    prompt = f"""Given this user question, generate 3-4 different search queries 
    that would help retrieve relevant documents to answer it.
    
    Consider:
    - Synonyms and alternative phrasings
    - Sub-questions that need to be answered
    - Both specific and general versions of the question
    
    Question: {user_query}
    
    Return as JSON array of strings."""
    
    response = await llm.ainvoke(prompt)
    queries = json.loads(response.content)
    
    # Always include the original query
    return [user_query] + queries[:3]
async def retrieve_with_transformation(query: str) -> list[Document]:
    transformed_queries = await transform_query(query)
    
    # Retrieve for each query variant
    all_results = []
    for q in transformed_queries:
        results = await vector_store.asimilarity_search(q, k=5)
        all_results.extend(results)
    
    # Deduplicate and re-rank
    return deduplicate_and_rerank(all_results, original_query=query)
HyDE (Hypothetical Document Embeddings) is another powerful technique: generate a hypothetical answer to the query and embed that for retrieval:
async def hyde_retrieve(query: str) -> list[Document]:
    # Generate a hypothetical answer
    hypothetical_answer = await llm.ainvoke(
        f"Write a detailed answer to: {query}\n\n"
        f"Write as if you had access to perfect documentation."
    )
    
    # Use the hypothetical answer as the retrieval query
    # (it's closer in embedding space to real documentation than the question)
    return await vector_store.asimilarity_search(
        hypothetical_answer.content, k=10
    )
Pattern 4: Contextual Compression and Re-ranking
Retrieving the right documents is only half the problem. The retrieved chunks often contain noise — irrelevant sentences, repetitive content, tangential information. Contextual compression filters the retrieved content before sending it to the LLM:
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# Cross-encoder reranking: much more accurate than bi-encoder for final ranking
cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
reranker = CrossEncoderReranker(model=cross_encoder, top_n=5)
async def retrieve_and_compress(query: str) -> list[Document]:
    # Step 1: Broad retrieval (high recall, lower precision)
    candidates = await vector_store.asimilarity_search(query, k=20)
    
    # Step 2: Re-rank with cross-encoder (better precision)
    reranked = reranker.compress_documents(candidates, query)
    
    # Step 3: Extract only relevant passages
    compressed = []
    for doc in reranked[:5]:
        relevant_passages = extract_relevant_passages(doc.page_content, query)
        compressed.append(Document(
            page_content=relevant_passages,
            metadata=doc.metadata
        ))
    
    return compressed
The pipeline is: bi-encoder for recall → cross-encoder for precision → extraction for relevance.
Pattern 5: Self-RAG and CRAG (Corrective RAG)
Self-RAG teaches the LLM to evaluate its own retrieved content and decide when retrieval is needed:
1. For each query, the model decides: "Do I need to retrieve?"
2. If yes, retrieve and assess: "Is this document relevant? (yes/no/partially)"
3. Generate answer using only relevant documents
4. Self-critique: "Is this answer supported by the retrieved text? (fully/partially/no)"
5. If not fully supported, retrieve again with a refined query
Corrective RAG (CRAG) adds an evaluator that automatically detects low-quality retrievals and falls back to web search:
async def crag_retrieve(query: str) -> list[Document]:
    # Try local knowledge base first
    docs = await vector_store.asimilarity_search(query, k=4)
    
    # Evaluate relevance
    eval_prompt = f"""Rate the relevance of these documents to the query.
    Query: {query}
    Documents: {format_docs(docs)}
    
    Return JSON: score"""
    
    evaluation = json.loads(await llm.ainvoke(eval_prompt))
    
    if evaluation["score"] < 0.5:
        # Fall back to web search
        web_results = await web_search(query)
        return web_results
    elif evaluation["score"] < 0.8:
        # Combine local + web
        web_results = await web_search(query)
        return docs + web_results
    else:
        return docs
Evaluation: How Do You Know Your RAG Is Working?
A RAG system without evaluation is a hope, not an engineering artifact:
from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy,
)
from datasets import Dataset
def evaluate_rag_pipeline(qa_pairs: list[dict]) -> dict:
    """
    qa_pairs: [{"question": "...", "answer": "...", "ground_truth": "..."}]
    """
    dataset = Dataset.from_list(qa_pairs)
    
    results = evaluate(
        dataset,
        metrics=[
            context_precision,   # Are retrieved docs relevant?
            context_recall,      # Are all relevant docs retrieved?
            faithfulness,        # Is the answer grounded in retrieved docs?
            answer_relevancy,    # Does the answer actually address the question?
        ]
    )
    return results
Build a golden dataset of question-answer pairs from your domain and run automated evaluation on every pipeline change. Treat RAG quality like a test suite — regressions should block deployment.
Conclusion
The gap between a 1-hour RAG demo and a production RAG system is significant, but the patterns are well-established:
Hierarchical chunking — preserve semantic units
Hybrid search — combine BM25 and dense retrieval
Query transformation — optimize for retrieval, not UX
Re-ranking — use cross-encoders for final ranking
Self-evaluation — let the system detect and correct bad retrievals
Continuous evaluation — treat quality as an engineering metric
None of these are particularly complex individually. The challenge is integrating them into a coherent, debuggable pipeline and measuring the right things. Teams that invest in evaluation infrastructure early will iterate much faster than those who rely on vibes.
Related posts: Agentic AI Workflows in Production, AI Inference Optimization Guide

RAG in 2026: Advanced Retrieval Strategies Beyond Naive Vector Search

댓글