RAG Systems Deep Dive: Retrieval-Augmented Generation Explained

Large language models are brilliant at generating fluent text, but they have a fundamental flaw: they only know what was in their training data. Ask GPT-4 about yesterday's news or your company's internal API documentation, and it will either guess or admit it does not know. That is where Retrieval-Augmented Generation (RAG) comes in.

RAG is not a model. It is an architecture pattern that plugs live, queryable knowledge into the generation process. Instead of asking the LLM to recall information from its parameters, you retrieve the relevant information from a database and feed it into the prompt. The LLM then generates an answer grounded in that retrieved context.

I have built RAG systems for a global ratings agency, xThreads.app, and PostQueue.app. Across these projects, the core pattern is the same: chunk → embed → index → retrieve → generate. This post explains each step in detail.

What you will learn: Why RAG exists and what problem it solves · The five-stage RAG pipeline · Chunking strategies and tradeoffs · Embedding models and vector databases · Hybrid search (semantic + keyword) · Query transformation techniques · Advanced RAG: iterative, multi-modal, self-RAG · Production considerations: latency, freshness, cost

1. The Problem RAG Solves

LLMs are trained on a static snapshot of the internet. They have three limitations:

1. Stale knowledge: They do not know anything after their training cutoff date.
2. No private knowledge: They have never seen your company's internal documents.
3. Hallucination: When they do not know the answer, they make something up.

RAG solves all three. Instead of relying on the model's parametric memory, you retrieve the relevant documents and include them in the prompt. The LLM becomes a reader and synthesizer of information you provide, rather than a generator of information it memorized.

2. The Five-Stage RAG Pipeline

Every RAG system follows the same five stages. These stages are the same whether you are building a customer support bot or a financial research tool:

Stage 1: Chunking

Raw documents are too large to fit in an LLM context window, and they contain irrelevant information. Chunking splits documents into semantically coherent pieces:

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # tokens per chunk
    chunk_overlap=50,     # overlap to preserve context across boundaries
    separators=["\n\n", "\n", ".", " ", ""],
)

document = open("annual_report_2025.pdf").read()
chunks = text_splitter.split_text(document)

print(f"Document: {len(document)} chars → {len(chunks)} chunks")
# Output: Document: 45200 chars → 142 chunks

The chunk size is a critical tradeoff. Smaller chunks (256 tokens) give more precise retrieval but lose surrounding context. Larger chunks (1024 tokens) preserve context but may include irrelevant information that dilutes relevance scores.

Stage 2: Embedding

Each chunk is converted into a vector embedding - a list of floating point numbers that represents the semantic meaning of the text:

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")  # 384-dim embeddings

chunks = ["Revenue grew 15% in Q3", "The CEO announced a new strategy", "..."]
embeddings = model.encode(chunks)

print(f"Shape: {embeddings.shape}")
# Output: Shape: (142, 384) - 142 chunks, 384 dimensions each

# Similar chunks have similar vectors
vec_a = model.encode("Revenue growth in Q3")
vec_b = model.encode("Third quarter financial performance")
similarity = cosine_similarity([vec_a], [vec_b])[0][0]
print(f"Semantic similarity: {similarity:.2f}")
# Output: Semantic similarity: 0.87

Stage 3: Indexing

Embeddings are stored in a vector database that supports efficient similarity search:

import chromadb

client = chromadb.Client()
collection = client.create_collection("company_docs")

# Index all chunks with their embeddings
collection.add(
    documents=chunks,
    embeddings=embeddings.tolist(),
    metadatas=[{"source": "annual_report_2025.pdf", "page": i} for i in range(len(chunks))],
    ids=[f"chunk_{i}" for i in range(len(chunks))],
)

3. Retrieval: Finding the Right Information

When a user asks a question, we embed their query and search the vector database for the most similar chunks. This is semantic search - it finds information based on meaning, not just keywords.

Semantic Search

def semantic_search(query: str, collection, n: int = 5) -> list[str]:
    """Find the n most semantically similar chunks to the query."""
    query_embedding = model.encode([query])
    results = collection.query(
        query_embeddings=query_embedding.tolist(),
        n_results=n,
    )
    return results["documents"][0]

Hybrid Search

Semantic search is powerful, but it can miss exact matches. A query for "Q3 2025 revenue" might not find a chunk that says "revenue in the third quarter of fiscal year 2025" if the embedding model did not capture that equivalence perfectly. Hybrid search combines semantic and keyword search:

def hybrid_search(query: str, collection, n: int = 5) -> list[str]:
    """Combine semantic search with keyword (BM25) search."""
    # Semantic results
    query_embedding = model.encode([query])
    semantic_results = collection.query(
        query_embeddings=query_embedding.tolist(),
        n_results=n * 2,  # get more candidates
    )

    # BM25 keyword results (using a simple TF-IDF approach)
    keyword_results = bm25_search(query, all_chunks, n=n * 2)

    # Reciprocal Rank Fusion - merge both result sets
    merged = reciprocal_rank_fusion(
        semantic_results["documents"][0],
        keyword_results,
    )
    return merged[:n]  # return top n

def reciprocal_rank_fusion(*ranked_lists: list[str], k: int = 60) -> list[str]:
    """Merge multiple ranked lists using RRF scoring."""
    scores = {}
    for rank_list in ranked_lists:
        for rank, doc in enumerate(rank_list, 1):
            scores[doc] = scores.get(doc, 0) + 1 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)

4. Generation: Grounded Answers

Once we have retrieved the relevant chunks, we insert them into the LLM prompt ascontext. The LLM generates an answer using that context:

def generate_answer(query: str, context_chunks: list[str], llm) -> str:
    """Generate a grounded answer from retrieved context."""
    context = "\n\n---\n\n".join(context_chunks)

    prompt = f"""You are a helpful assistant. Answer the question using ONLY the context below.
If the context does not contain the answer, say "I don't know".

CONTEXT:
{context}

QUESTION: {query}

ANSWER:"""

    response = llm.generate(prompt)
    return response

The key instruction is "using ONLY the context below." Without this, the LLM may fall back to its parametric knowledge and hallucinate. With it, the LLM is constrained to the information you provided.

5. Query Transformation: Better Retrieval Through Rethinking

Sometimes the best way to improve retrieval is to improve the query, not the index. Several techniques transform the user's original question into something the retrieval system can answer more effectively:

Query Decomposition

Break complex questions into simpler sub-questions, retrieve for each, then combine:

def decompose_query(complex_query: str) -> list[str]:
    """Break a complex query into simpler sub-queries."""
    prompt = f"""Break this question into simpler sub-questions:
    Question: {complex_query}
    Sub-questions:"""
    response = llm.generate(prompt)
    return response.strip().split("\n")

# Example
decompose_query("What was Apple's revenue growth and R&D spend in 2025?")
# Returns: ["What was Apple's revenue in 2025?",
#           "What was Apple's revenue growth rate in 2025?",
#           "What was Apple's R&D spending in 2025?"]

Hypothetical Document Embeddings (HyDE)

Generate a hypothetical answer first, then use that to search. The hypothetical answer often has better semantic alignment with relevant documents than the original query:

def hyde_search(query: str, collection, llm, n: int = 5) -> list[str]:
    """Generate a hypothetical document, then search with it."""
    prompt = f"Write a paragraph that answers this question: {query}"
    hypothetical_doc = llm.generate(prompt)

    # Search using the hypothetical document instead of the query
    embedding = model.encode([hypothetical_doc])
    results = collection.query(
        query_embeddings=embedding.tolist(),
        n_results=n,
    )
    return results["documents"][0]

6. Advanced RAG Patterns

Once the basic pipeline is working, several advanced patterns can improve quality:

Iterative / Multi-Hop RAG

Instead of one round of retrieval, the system retrieves, reads, identifies gaps, and retrieves again:

def multi_hop_rag(query: str, max_hops: int = 3) -> str:
    """Iterative retrieval - each hop fills gaps from the previous."""
    context = []
    current_query = query

    for hop in range(max_hops):
        # Retrieve
        chunks = semantic_search(current_query, collection)
        context.extend(chunks)

        # Generate partial answer
        answer = generate_answer(query, context, llm)

        # Check if answer is complete
        if is_answer_complete(answer):
            return answer

        # Generate follow-up query for missing information
        current_query = generate_follow_up(query, answer)
        # Example: "What was Apple's revenue in 2025?" →
        # "Now I need: What was Apple's R&D spend in 2025?"

    return generate_answer(query, context, llm)

Re-Ranking

After retrieving candidates, use a cross-encoder model to re-rank by actual relevance (more accurate than the embedding model's cosine similarity):

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def retrieve_and_rerank(query: str, n_candidates: int = 20, n_final: int = 5):
    """Retrieve many candidates, then re-rank for precision."""
    candidates = semantic_search(query, collection, n=n_candidates)

    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)

    # Sort by score, return top n_final
    ranked = [doc for _, doc in sorted(zip(scores, candidates), reverse=True)]
    return ranked[:n_final]

Self-RAG

The most advanced pattern: the LLM decides when to retrieve, what to retrieve, and whether the retrieved information is sufficient:

def self_rag(query: str, llm, max_iterations: int = 5) -> str:
    """Self-RAG: the LLM controls the retrieval process."""
    state = {
        "query": query,
        "context": [],
        "iterations": 0,
    }

    while state["iterations"] < max_iterations:
        # LLM decides: retrieve, generate, or stop
        decision = llm.generate(f"""State: {state}
Decide: (R)etrieve more, (G)enerate answer, or (S)top?""")

        if decision == "R":
            # LLM generates the search query
            search_query = llm.generate(f"What should I search for? Context: {state}")
            chunks = semantic_search(search_query, collection)
            state["context"].extend(chunks)

        elif decision == "G":
            return generate_answer(query, state["context"], llm)

        elif decision == "S":
            return "I don't have enough information to answer this."

        state["iterations"] += 1

    return generate_answer(query, state["context"], llm)

7. Production RAG: Latency, Freshness, and Cost

A working RAG pipeline is a prototype. A production RAG system needs to handle:

Latency

Every RAG step adds latency. For user-facing applications, keep total response time under 2 seconds:

import asyncio
from time import perf_counter

class LatencyTracker:
    def __init__(self):
        self.timings = {}

    async def __aenter__(self):
        self.start = perf_counter()
        return self

    async def __aexit__(self, *args):
        elapsed = perf_counter() - self.start
        print(f"Total RAG latency: {elapsed*1000:.0f}ms")

# Typical latency budget:
# Embed query:     50ms
# Vector search:  100ms
# Rerank:         200ms
# LLM generation: 800ms
# Total:         ~1150ms ✓

Freshness

Knowledge goes stale. Production systems need automated re-indexing:

class IndexManager:
    def __init__(self, collection, source_path: str):
        self.collection = collection
        self.source = source_path
        self.last_indexed = None

    def check_and_update(self):
        """Re-index if source files have changed."""
        latest_mtime = max(
            os.path.getmtime(f) for f in glob.glob(f"{self.source}/**/*")
        )
        if self.last_indexed is None or latest_mtime > self.last_indexed:
            print("Source files changed. Re-indexing...")
            self._full_reindex()
            self.last_indexed = latest_mtime

    def _full_reindex(self):
        documents = load_all_documents(self.source)
        chunks = chunk_all(documents)
        embeddings = embed_all(chunks)
        self.collection.delete_all()
        self.collection.add(
            documents=chunks,
            embeddings=embeddings,
            ids=[f"chunk_{i}" for i in range(len(chunks))],
        )
        print(f"Indexed {len(chunks)} chunks")

8. Summary: What to Remember