How RAG Works Step by Step: From Document to Answer in 7 Stages

TL;DR — RAG works in 7 stages across two phases. Pinecone: "Indexing: collect → chunk → embed → store. Retrieval+generation: query → retrieve → rerank → generate." Lushbinary: "In 2026, retrieval is the critical bottleneck, not generation. RAG has moved beyond basic vector search into hybrid, agentic, and graph-augmented architectures." Techment: "Hybrid retrieval is the default in 2026 — combining dense vector search with sparse BM25 using Reciprocal Rank Fusion." Inventiple: "512-token chunks + Cohere reranker beats 1024-token without reranking. RAGAS faithfulness below 0.8 means hallucination risk." AI with Aish: "The fundamental flow is Index → Query → Retrieve → Augment → Generate." StackAI: "Use metadata filtering and lexical retrieval to guarantee coverage." Learn more with what is RAG, build RAG from scratch, chunking strategies, and reranking.

Lushbinary sets the 2026 context: "In 2026, the retrieval step is the critical bottleneck, not generation. That reality has pushed RAG beyond basic vector search into hybrid, agentic, and graph-augmented architectures. This guide covers what actually works in production: chunking strategies, embedding model selection, hybrid search, reranking, and evaluation."

AI with Aish summarizes the flow: "The fundamental flow is Index → Query → Retrieve → Augment → Generate. The fundamental problem is that bi-encoder vector embeddings are lossy by design. They compress a complex paragraph into a single point in a 1536-dimensional space. This is great for finding concepts but terrible for finding exact keywords."

RAG Step-by-Step Architecture

7 RAG Stages Detail

Implementation

import hashlib
import json
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Optional

@dataclass
class RAGStepByStep:
    """Step-by-step RAG pipeline implementation."""

    def __init__(self, embedding_client, vector_db, llm_client,
                 reranker=None):
        self.embedder = embedding_client
        self.vdb = vector_db
        self.llm = llm_client
        self.reranker = reranker

    # ─── Stage 1: Collect Sources ───
    async def stage_1_collect(self, sources: list) -> list:
        """Collect documents from various sources."""
        documents = []
        for source in sources:
            if source["type"] == "file":
                content = self._read_file(source["path"])
            elif source["type"] == "web":
                content = await self._fetch_web(source["url"])
            elif source["type"] == "database":
                content = await self._query_db(source["query"])
            else:
                continue

            documents.append({
                "id": source["id"],
                "content": content,
                "source": source.get("url", source.get("path", "unknown")),
                "title": source.get("title", ""),
                "type": source["type"],
            })
        return documents

    # ─── Stage 2: Chunk ───
    async def stage_2_chunk(self, documents: list,
                            chunk_size: int = 512,
                            overlap: int = 64) -> list:
        """Chunk documents into overlapping segments."""
        all_chunks = []
        for doc in documents:
            # Normalize: clean whitespace, remove boilerplate
            text = " ".join(doc["content"].split())
            words = text.split()

            start = 0
            chunk_idx = 0
            while start < len(words):
                end = min(start + chunk_size, len(words))
                chunk_text = " ".join(words[start:end])

                all_chunks.append({
                    "chunk_id": hashlib.sha256(
                        f"{doc['id']}:{chunk_idx}".encode()
                    ).hexdigest()[:16],
                    "doc_id": doc["id"],
                    "content": chunk_text,
                    "chunk_index": chunk_idx,
                    "source": doc["source"],
                    "title": doc["title"],
                    "token_count": end - start,
                })

                if end == len(words):
                    break
                start = end - overlap
                chunk_idx += 1

        return all_chunks

    # ─── Stage 3: Embed ───
    async def stage_3_embed(self, chunks: list) -> list:
        """Convert chunks to vector embeddings."""
        texts = [c["content"] for c in chunks]
        embeddings = await self.embedder.embed_batch(texts)

        for chunk, embedding in zip(chunks, embeddings):
            chunk["embedding"] = embedding

        return chunks

    # ─── Stage 4: Store ───
    async def stage_4_store(self, chunks: list) -> dict:
        """Store embeddings in vector database with metadata."""
        points = []
        for chunk in chunks:
            points.append({
                "id": chunk["chunk_id"],
                "vector": chunk["embedding"],
                "payload": {
                    "content": chunk["content"],
                    "doc_id": chunk["doc_id"],
                    "chunk_index": chunk["chunk_index"],
                    "source": chunk["source"],
                    "title": chunk["title"],
                    "token_count": chunk["token_count"],
                    "indexed_at": datetime.now(timezone.utc).isoformat(),
                },
            })

        await self.vdb.upsert(
            collection_name="knowledge_base",
            points=points,
        )

        return {
            "chunks_stored": len(points),
            "collection": "knowledge_base",
        }

    # ─── Stage 5: Retrieve ───
    async def stage_5_retrieve(self, query: str,
                               top_k: int = 5,
                               use_hybrid: bool = True) -> list:
        """Embed query and search vector database."""
        # Embed query with same model as indexing
        query_embedding = await self.embedder.embed(query)

        if use_hybrid:
            # Hybrid: dense vector + sparse BM25 + RRF fusion
            dense_results = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding,
                limit=top_k * 3,
            )
            sparse_results = await self.vdb.search(
                collection_name="knowledge_base",
                query=query,
                limit=top_k * 3,
            )
            results = self._rrf_fusion(dense_results, sparse_results)
        else:
            results = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding,
                limit=top_k * 3,
            )

        return results

    # ─── Stage 6: Rerank ───
    async def stage_6_rerank(self, query: str,
                             candidates: list,
                             top_k: int = 5) -> list:
        """Rerank candidates with cross-encoder."""
        if not self.reranker:
            return candidates[:top_k]

        # Cross-encoder processes query + document together
        pairs = [(query, c["payload"]["content"]) for c in candidates]
        scores = await self.reranker.predict(pairs)

        for candidate, score in zip(candidates, scores):
            candidate["rerank_score"] = float(score)

        # Sort by rerank score and return top_k
        ranked = sorted(
            candidates,
            key=lambda x: x["rerank_score"],
            reverse=True,
        )
        return ranked[:top_k]

    # ─── Stage 7: Generate ───
    async def stage_7_generate(self, query: str,
                               retrieved: list) -> dict:
        """Generate answer with grounded prompt."""
        # Assemble context with citations
        context_parts = []
        sources = []
        for i, result in enumerate(retrieved):
            content = result["payload"]["content"]
            source = result["payload"].get("source", "unknown")
            context_parts.append(f"[{i+1}] ({source})\n{content}")
            sources.append({
                "citation": i + 1,
                "source": source,
                "doc_id": result["payload"]["doc_id"],
            })

        context = "\n\n".join(context_parts)

        # Build grounded prompt with refusal behavior
        prompt = (
            "You are a helpful assistant. Answer the question "
            "based ONLY on the provided context. If the context "
            "does not contain enough information, say 'I don't "
            "have enough information to answer this question.'\n\n"
            f"Context:\n{context}\n\n"
            f"Question: {query}\n\n"
            "Answer (cite sources using [1], [2], etc.):"
        )

        # Generate
        answer = await self.llm.generate(prompt)

        return {
            "query": query,
            "answer": answer,
            "sources": sources,
            "context_chunks": len(retrieved),
        }

    # ─── Full Pipeline ───
    async def run_full_pipeline(self, query: str,
                                top_k: int = 5) -> dict:
        """Run complete RAG pipeline: retrieve → rerank → generate."""
        import time
        start = time.time()

        # Stage 5: Retrieve
        retrieve_start = time.time()
        candidates = await self.stage_5_retrieve(query, top_k=top_k)
        retrieve_time = time.time() - retrieve_start

        # Stage 6: Rerank
        rerank_start = time.time()
        ranked = await self.stage_6_rerank(query, candidates, top_k)
        rerank_time = time.time() - rerank_start

        # Stage 7: Generate
        generate_start = time.time()
        result = await self.stage_7_generate(query, ranked)
        generate_time = time.time() - generate_start

        total_time = time.time() - start

        return {
            **result,
            "timing": {
                "retrieve_ms": round(retrieve_time * 1000, 1),
                "rerank_ms": round(rerank_time * 1000, 1),
                "generate_ms": round(generate_time * 1000, 1),
                "total_ms": round(total_time * 1000, 1),
            },
            "candidates_retrieved": len(candidates),
            "chunks_after_rerank": len(ranked),
        }

    def _rrf_fusion(self, dense: list, sparse: list,
                    bm25_weight: float = 0.3, k: int = 60) -> list:
        """Reciprocal Rank Fusion of dense and sparse results."""
        scores = {}
        results_map = {}

        for rank, r in enumerate(dense):
            scores[r["id"]] = scores.get(r["id"], 0) + \
                (1 - bm25_weight) / (k + rank + 1)
            results_map[r["id"]] = r

        for rank, r in enumerate(sparse):
            scores[r["id"]] = scores.get(r["id"], 0) + \
                bm25_weight / (k + rank + 1)
            results_map[r["id"]] = r

        sorted_ids = sorted(scores, key=scores.get, reverse=True)
        return [results_map[i] for i in sorted_ids]

    def _read_file(self, path: str) -> str:
        with open(path, "r") as f:
            return f.read()

    async def _fetch_web(self, url: str) -> str:
        # In production: use httpx to fetch web content
        return f"Web content from {url}"

    async def _query_db(self, query: str) -> str:
        # In production: execute database query
        return f"DB result for {query}"

How RAG Works Step by Step Checklist

• [ ] Understand the 7 RAG stages across 2 phases (indexing + retrieval/generation)
• [ ] Phase 1 runs offline; Phase 2 runs at runtime (inference time)
• [ ] Stage 1 — Collect: identify all source types (docs, PDFs, code, web, databases)
• [ ] Stage 1: use connectors to gather content from each source
• [ ] Stage 1: normalize — extract text, clean boilerplate, remove duplicates
• [ ] Stage 2 — Chunk: choose chunk size (512 tokens production default)
• [ ] Stage 2: set overlap (64-128 tokens to preserve context across boundaries)
• [ ] Stage 2: choose chunking strategy (fixed, sentence, semantic, recursive)
• [ ] Stage 2: attach metadata to each chunk (doc_id, source, title, timestamp)
• [ ] Stage 3 — Embed: select embedding model (text-embedding-3-small, MiniLM, BGE)
• [ ] Stage 3: use the same embedding model for indexing and retrieval
• [ ] Stage 3: track embedding model version for reindexing
• [ ] Stage 3: batch embed chunks for efficiency
• [ ] Stage 4 — Store: choose vector database (Qdrant production default)
• [ ] Stage 4: configure index type (HNSW for speed, IVF for scale)
• [ ] Stage 4: store metadata alongside vectors for filtering
• [ ] Stage 4: implement reindexing strategy for embedding model changes
• [ ] Stage 5 — Retrieve: embed user query with same model as indexing
• [ ] Stage 5: use hybrid search (dense vector + BM25 + RRF fusion) for production
• [ ] Stage 5: apply metadata filtering (source, date, category, permissions)
• [ ] Stage 5: over-fetch candidates (3x top_k) for reranking
• [ ] Stage 5: track recall@k and precision@k metrics
• [ ] Stage 6 — Rerank: use cross-encoder reranker (Cohere, ms-marco-MiniLM)
• [ ] Stage 6: cross-encoder processes query + document together (more accurate)
• [ ] Stage 6: 256-token chunks + reranker beats 1024-token without reranker
• [ ] Stage 6: select top_k final results after reranking
• [ ] Stage 7 — Generate: assemble context with deduplication and citation numbers
• [ ] Stage 7: build grounded prompt with context, query, and refusal behavior
• [ ] Stage 7: include citation instruction ([1], [2], etc.)
• [ ] Stage 7: if context is insufficient, LLM should say so (refusal behavior)
• [ ] Stage 7: log retrieval set, reranked set, final context, latency, cost
• [ ] Evaluation: create test set of 20+ questions with ground truth
• [ ] Evaluation: compute RAGAS faithfulness (target: 0.8+)
• [ ] Evaluation: compute answer relevancy, context precision, context recall
• [ ] Evaluation: evaluate before and after every significant change
• [ ] Evaluation: track metrics in CI/CD pipeline
• [ ] Add semantic caching for repeated queries
• [ ] Add query rewriting for complex multi-part questions
• [ ] Consider GraphRAG for relationship-heavy knowledge
• [ ] Consider agentic RAG for multi-step reasoning
• [ ] Implement streaming responses (SSE) for better UX
• [ ] Add access controls — filter chunks by user permissions
• [ ] Use MMR (Maximum Marginal Relevance) for result diversity
• [ ] In 2026, retrieval is the bottleneck, not generation — optimize retrieval
• [ ] Read what is RAG for architecture overview
• [ ] Build RAG from scratch for hands-on
• [ ] Optimize chunking strategies
• [ ] Understand embeddings
• [ ] Add hybrid search for production
• [ ] Add reranking for precision
• [ ] Prevent hallucinations with grounding
• [ ] Apply zero-trust to RAG access
• [ ] Log all RAG operations in audit logs
• [ ] Test: each stage produces expected output
• [ ] Test: hybrid search returns both semantic and keyword matches
• [ ] Test: reranking improves top-k precision
• [ ] Test: grounded prompt refuses when context is insufficient
• [ ] Test: RAGAS faithfulness meets 0.8+ threshold
• [ ] Document each stage's configuration and expected behavior

FAQ

What are the stages of a RAG pipeline?

A RAG pipeline has 7 stages across two phases. Pinecone: Indexing phase: (1) Collect sources — docs, PDFs, code, web pages. (2) Chunk — split into segments with overlap. (3) Embed — convert chunks to vectors. (4) Store — upsert into vector database with metadata. Retrieval+generation phase: (5) Retrieve — embed query, search vector DB (hybrid: vector + BM25). (6) Rerank — cross-encoder re-scores top candidates. (7) Generate — LLM produces answer with grounded prompt and citations. Lushbinary: "In 2026, the retrieval step is the critical bottleneck, not generation. That reality has pushed RAG beyond basic vector search into hybrid, agentic, and graph-augmented architectures." AI with Aish: "The fundamental flow is Index → Query → Retrieve → Augment → Generate."

How does the RAG indexing pipeline work?

The RAG indexing pipeline prepares knowledge for retrieval in 4 steps. DigitalOcean: "The indexing pipeline prepares the knowledge base so it can be searched efficiently. Documents are ingested, cleaned, split into chunks, converted into embeddings, and stored in a vector database. This process is usually executed offline or periodically when new data becomes available." Glukhov: (1) Collect sources — docs, tickets, web pages, PDFs, code. (2) Normalize — extract text, clean boilerplate, de-duplicate. (3) Chunk — choose strategy (fixed, sentence, semantic) with overlap and metadata. (4) Embed — use versioned embedding model to convert each chunk to a vector. (5) Upsert into index — store vectors with metadata fields in vector database. Inventiple: "Chunking strategy — chunk size and overlap affect retrieval precision more than almost any other variable. 512-token chunks with Cohere reranker outperforms naive 1024-token retrieval consistently." Key: use the same embedding model for indexing and retrieval.

How does RAG retrieval work?

RAG retrieval works by embedding the user query and searching the vector database for similar chunks. Pinecone: "During retrieval, we create a vector embedding from the user query to use for searching against the vectors in the database. In hybrid search, you query either a single hybrid index or both a dense and a sparse index. Then we combine and de-duplicate the results and use a reranking model to re-score them." Techment: "Hybrid retrieval is the default recommended choice in 2026. New advances in hybrid retrieval combine dense vector search with sparse BM25 keyword search using Reciprocal Rank Fusion." StackAI: "Use metadata filtering and lexical retrieval to guarantee coverage. Even with the right embedding model, your index and retrieval configuration can quietly kill recall." Steps: (1) Embed query using same model as indexing. (2) Search vector DB (dense, or hybrid with BM25). (3) Apply metadata filters. (4) Over-fetch candidates (3x top_k) for reranking. (5) Rerank with cross-encoder. (6) Select top_k final results.

How does RAG generation work?

RAG generation works by passing retrieved context and the user query to an LLM with a grounded prompt. DigitalOcean: "The retrieved chunks are combined and passed to the language model as contextual input. The model uses this context to generate a response grounded in actual documents rather than relying solely on its training data." Glukhov: "Assemble context — dedupe, order by relevance, add citations. Generate with grounded prompt — rules + refusal behavior. Log retrieval set, reranked set, final context, latency, cost." The grounded prompt includes: (1) System instruction to answer based on context only. (2) Retrieved context chunks with citation numbers. (3) The user question. (4) Refusal behavior — if context is insufficient, say so. (5) Citation instruction — reference sources using [1], [2], etc. Lushbinary: "In 2026, the retrieval step is the critical bottleneck, not generation. Generation is the easy part once you have good context."

What is the RAG evaluation process?

RAG evaluation measures retrieval quality and generation quality using RAGAS metrics. Inventiple: "Evaluate before shipping — RAGAS faithfulness below 0.8 means hallucination risk is unacceptable. Evaluate before and after every significant change." Glukhov: "Track retrieval metrics — recall@k and precision@k continuously. Mastering RAG architecture, reranking, vector databases, hybrid search, and evaluation will determine whether your AI system remains a demo or becomes production-ready." RAGAS metrics: (1) Faithfulness — is the answer grounded in retrieved context? Target: 0.8+. (2) Answer relevancy — does the answer address the question? (3) Context precision — are retrieved chunks relevant to the query? (4) Context recall — did retrieval find all necessary information? Process: create test set of 20+ questions with ground truth answers, run RAG pipeline, compute RAGAS scores, track in CI/CD, evaluate before and after every change. NerdLevelTech: "Add evaluation: create a test set of 20 questions and measure RAGAS scores. Optimize chunking, add hybrid search, add reranking, go to production."

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →