AI & Machine Learning

RAG Implementation Guide: Build AI Apps with Retrieval Augmented Generation in 2026

2026-08-02·15 min read
#rag#ai#llm#retrieval augmented generation#vector database

RAG Implementation Guide: Build AI Apps with Retrieval Augmented Generation in 2026

Large language models are powerful, but they have a fundamental flaw: they hallucinate. They generate text that sounds correct but has no basis in reality. For chatbots, that's a minor annoyance. For legal research, medical diagnosis, or enterprise knowledge management, it's a dealbreaker.

Retrieval Augmented Generation (RAG) solves this by grounding the model in your actual data. Instead of relying solely on what the model learned during training, RAG fetches relevant documents at query time and feeds them into the model's context window. The result: answers that are accurate, citeable, and up-to-date.

This guide walks you through building a production-grade RAG system from scratch — covering architecture decisions, embedding models, vector databases, chunking strategies, advanced retrieval techniques, evaluation frameworks, and deployment patterns. All with real Python code you can run today.

Table of Contents

  1. What Is RAG and Why It Matters
  2. RAG Architecture: The Retrieval → Augmentation → Generation Pipeline
  3. Choosing Embeddings: OpenAI, Cohere, and Open-Source Models
  4. Vector Databases Compared
  5. Chunking Strategies with Code
  6. Advanced Techniques: Re-ranking, Hybrid Search, and HyDE
  7. Building a Complete RAG Pipeline in Python
  8. Evaluation: RAGAS, Faithfulness, and Relevance
  9. Production Deployment Patterns
  10. Common Failure Modes and How to Fix Them

What Is RAG and Why It Matters

RAG — Retrieval Augmented Generation — is a technique that combines an LLM with an external knowledge source. When a user asks a question, the system first retrieves relevant documents from a knowledge base, then augments the user's prompt with that context, and finally lets the LLM generate an answer based on both the question and the retrieved documents.

Why not just fine-tune?

Fine-tuning teaches a model new patterns or styles, but it's the wrong tool for factual knowledge:

| Problem | Fine-Tuning | RAG | |---|---|---| | Knowledge updates | Requires retraining | Just update the document store | | Source attribution | Impossible | Built-in (cite the retrieved chunks) | | Cost per update | High (GPU training) | Low (re-embed changed docs) | | Hallucination risk | Still high | Dramatically reduced | | Access control | Baked into weights | Per-query filtering |

Fine-tuning is excellent for behavior — making a model respond in a certain tone, format, or domain-specific language. RAG is excellent for knowledge — making a model answer based on your proprietary data. In production, the best systems often combine both.

The grounding problem

LLMs are trained on a fixed corpus with a training cutoff date. They don't know your company's internal docs, yesterday's news, or the latest research papers. RAG bridges this gap by treating the LLM as a reasoning engine rather than a knowledge store:

  • Current data: Update your document store, and RAG reflects the change immediately.
  • Private data: Your data stays in your vector database — it's never baked into model weights.
  • Verifiable answers: Every answer can link back to the specific source chunk that was retrieved.

This is why companies building customer support bots, internal knowledge assistants, legal research tools, and code documentation assistants have converged on RAG as the standard architecture.

RAG Architecture: The Retrieval → Augmentation → Generation Pipeline

A RAG system has three core stages. Understanding each stage is essential before diving into implementation.

Stage 1: Retrieval

Given a user query, the system searches a knowledge base for the most relevant text passages. This typically involves:

  1. Embedding the query: Convert the user's question into a dense vector using the same embedding model that was used for the documents.
  2. Vector search: Find the nearest neighbors in the vector database using cosine similarity or dot product.
  3. Optional re-ranking: Re-order the top results using a more sophisticated (but slower) model.

The output is a set of text chunks with relevance scores.

Stage 2: Augmentation

The retrieved chunks are assembled into a prompt template alongside the user's question:

You are a helpful assistant. Answer the user's question based only on the
provided context. If the context doesn't contain the answer, say "I don't know."

Context:
{retrieved_chunks}

Question: {user_question}

Answer:

This prompt constrains the model to base its answer on the retrieved context rather than free-associating from training data.

Stage 3: Generation

The LLM receives the augmented prompt and generates a response. The model reads the context, reasons about the question, and produces an answer — ideally with citations to the source chunks.

The Full Pipeline

User Query
    │
    ▼
┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│   Embed     │────▶│  Vector DB   │────▶│   Re-ranker     │
│   Query     │     │  Search      │     │   (optional)    │
└─────────────┘     └──────────────┘     └────────┬────────┘
                                                  │
                                          Top-K chunks
                                                  │
                                                  ▼
                                          ┌──────────────┐
                                          │  Prompt      │
                                          │  Template    │
                                          └──────┬───────┘
                                                 │
                                                 ▼
                                         ┌──────────────┐
                                         │     LLM      │
                                         │  Generation  │
                                         └──────┬───────┘
                                                │
                                                ▼
                                        Answer + Citations

The beauty of this architecture is its modularity. You can swap the embedding model, change the vector database, adjust the chunking strategy, or switch the LLM — all without rewriting the entire system.

Choosing Embeddings

The embedding model is the heart of your RAG system. It determines how well your system understands the semantic meaning of both documents and queries. A poor embedding model means poor retrieval, and poor retrieval means wrong answers — no matter how powerful your LLM is.

What makes a good embedding model?

  • Dimensionality: Lower dimensions (384–768) are faster but less precise. Higher dimensions (1024–3072) capture more nuance but cost more in storage and compute.
  • Domain fit: General-purpose models work well for most text. Specialized models exist for code, multilingual text, medical literature, etc.
  • Max sequence length: How many tokens the model can embed in a single call. Longer is better for larger chunks.
  • Speed and cost: API-based models add latency and per-call cost. Local models are free but require infrastructure.

The leading embedding models in 2026

OpenAI text-embedding-3-large

  • Dimensions: 3072 (configurable down to 256 via truncation)
  • Max input: 8,191 tokens
  • Strengths: Excellent general-purpose performance, simple API, widely supported
  • Cost: $0.13 per million tokens

Cohere embed-english-v3.0

  • Dimensions: 1024
  • Max input: 512 tokens
  • Strengths: Strong on English-only tasks, built-in search-optimized embeddings
  • Cost: $0.10 per million tokens

Nomic nomic-embed-text-v1.5 (Open-source)

  • Dimensions: 768
  • Max input: 8,192 tokens
  • Strengths: Fully open-source, competitive with proprietary models, long context
  • Cost: Free (self-hosted)

BGE bge-large-en-v1.5 by BAAI (Open-source)

  • Dimensions: 1024
  • Max input: 512 tokens
  • Strengths: Top-tier on the MTEB benchmark, widely adopted in the open-source community
  • Cost: Free (self-hosted)

Code: Embedding a query with different providers

# OpenAI Embeddings
import openai

client = openai.Client(api_key="sk-...")

def embed_openai(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text
    )
    return response.data[0].embedding

# Nomic Embed (open-source, local via sentence-transformers)
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("nomic-ai/nomic-embed-text-v1.5")

def embed_nomic(text: str) -> list[float]:
    return model.encode(text, normalize_embeddings=True).tolist()

# Cohere Embeddings
import cohere

co = cohere.Client(api_key="...")

def embed_cohere(texts: list[str]) -> list[list[float]]:
    response = co.embed(
        texts=texts,
        model="embed-english-v3.0",
        input_type="search_document"
    )
    return response.embeddings

Practical recommendation

Start with OpenAI text-embedding-3-large for prototyping — it's the easiest to use and consistently performs well. If cost or data privacy becomes a concern, switch to nomic-embed-text-v1.5 self-hosted via sentence-transformers. The quality gap is small, and you eliminate per-call API costs entirely.

Vector Databases Compared

The vector database stores your embedded documents and performs fast similarity searches at query time. Your choice here affects scalability, latency, operational complexity, and cost.

Pinecone

Pinecone is a fully managed vector database. You don't manage infrastructure — you just create an index, upload vectors, and query.

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="...")
pc.create_index(
    name="docs",
    dimension=3072,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
index = pc.Index("docs")

# Upsert vectors
index.upsert(vectors=[
    {"id": "doc-1", "values": [0.1, 0.2, ...], "metadata": {"source": "handbook.pdf"}}
])

# Query
results = index.query(vector=[0.1, 0.2, ...], top_k=5, include_metadata=True)

Pros: Zero infrastructure, auto-scaling, serverless pricing Cons: Vendor lock-in, per-query cost at scale, limited filtering options on lower tiers Best for: Teams that want to ship fast and avoid DevOps

Weaviate

Weaviate is an open-source vector database with a powerful GraphQL API and built-in modules for automatic embedding (you can send raw text, and Weaviate handles vectorization).

import weaviate

client = weaviate.connect_to_local()

collection = client.collections.get("Document")
collection.data.insert({
    "text": "RAG combines retrieval with generation...",
    "category": "AI"
})

Pros: Hybrid search built-in (keyword + vector), self-hostable, automatic embedding modules Cons: More complex to operate than managed services, Java-based (heavier resource footprint) Best for: Teams that need hybrid search and want self-hosting

Qdrant

Qdrant is a Rust-based vector search engine with excellent performance and a rich filtering system.

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="localhost", port=6333)

client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(size=3072, distance=Distance.COSINE)
)

client.upsert(
    collection_name="docs",
    points=[
        PointStruct(id=1, vector=[0.1, 0.2, ...], payload={"text": "..."})
    ]
)

results = client.search(
    collection_name="docs",
    query_vector=[0.1, 0.2, ...],
    limit=5
)

Pros: Fast (Rust), excellent payload filtering, generous free tier, cloud or self-hosted Cons: Smaller community than Pinecone, fewer integrations Best for: Performance-sensitive applications needing complex metadata filtering

pgvector (PostgreSQL)

pgvector is a PostgreSQL extension that adds vector similarity search. If you already run Postgres, you can add vector search without a new system.

CREATE EXTENSION vector;

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(3072)
);

-- HNSW index for fast approximate search
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

-- Query
SELECT content, 1 - (embedding <=> '[0.1,0.2,...]'::vector) AS similarity
FROM documents
ORDER BY embedding <=> '[0.1,0.2,...]'::vector
LIMIT 5;

Pros: No new infrastructure, ACID transactions, SQL joins with vector search, mature ecosystem Cons: Not as fast as dedicated vector DBs at scale, index rebuild times for large datasets Best for: Teams already on Postgres with moderate-scale vector search needs

Chroma

Chroma is a lightweight, developer-friendly vector database that runs in-process or as a local server.

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection("documents")

collection.add(
    documents=["RAG combines retrieval with generation..."],
    metadatas=[{"source": "handbook.pdf"}],
    ids=["doc-1"]
)

results = collection.query(
    query_texts=["What is RAG?"],
    n_results=5
)

Pros: Zero-config setup, automatic embedding (via default sentence-transformers model), great for prototyping Cons: Not production-hardened for large scale, limited filtering Best for: Prototyping, small datasets, local development

Quick comparison table

| Database | Self-hosted | Managed | Hybrid Search | Best Feature | |---|---|---|---|---| | Pinecone | ❌ | ✅ | Partial | Zero ops | | Weaviate | ✅ | ✅ | ✅ | Auto-embedding modules | | Qdrant | ✅ | ✅ | ✅ | Payload filtering | | pgvector | ✅ | ✅ (via cloud) | Via tsvector | SQL integration | | Chroma | ✅ | ❌ | ❌ | Simplicity |

Recommendation

  • Prototyping: Start with Chroma. It's the fastest path to a working RAG demo.
  • Production (managed): Use Pinecone if budget allows, or Qdrant Cloud for better pricing at scale.
  • Production (self-hosted): Use Qdrant for pure vector search, or pgvector if you want to stay in Postgres.
  • Need hybrid search: Weaviate or Qdrant are your best bets.

Chunking Strategies

Chunking is the process of splitting your documents into smaller pieces before embedding them. This is one of the most impactful decisions in your RAG pipeline — poor chunking leads to either lost context (chunks too small) or diluted relevance (chunks too large).

Why chunk at all?

Embedding models have a maximum input length (often 512 or 8,192 tokens). Even without that constraint, embedding an entire 50-page document into a single vector produces a muddy representation that matches many queries poorly. Chunking creates focused, topical units that retrieve more precisely.

Strategy 1: Fixed-Size Chunking

Split text into chunks of N tokens with optional overlap. Simple, fast, and works reasonably well for uniform documents.

from typing import List

def fixed_size_chunk(text: str, chunk_size: int = 512, overlap: int = 50) -> List[str]:
    """Split text into fixed-size chunks with overlap."""
    tokens = text.split()
    chunks = []
    start = 0

    while start < len(tokens):
        end = start + chunk_size
        chunk = " ".join(tokens[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap  # move forward, leaving overlap

    return chunks

# Usage
text = open("document.txt").read()
chunks = fixed_size_chunk(text, chunk_size=400, overlap=50)
print(f"Created {len(chunks)} chunks")

Pros: Simple, predictable, fast Cons: Can split mid-sentence or mid-paragraph, breaking semantic units

Strategy 2: Semantic Chunking

Split text at natural boundaries — sentences, paragraphs, or sections — while staying close to a target chunk size.

import spacy

nlp = spacy.load("en_core_web_sm")

def semantic_chunk(text: str, max_tokens: int = 500) -> List[str]:
    """Split text at sentence boundaries, grouping sentences up to max_tokens."""
    doc = nlp(text)
    sentences = [sent.text.strip() for sent in doc.sents if sent.text.strip()]

    chunks = []
    current_chunk = []
    current_length = 0

    for sentence in sentences:
        sent_len = len(sentence.split())
        if current_length + sent_len > max_tokens and current_chunk:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentence]
            current_length = sent_len
        else:
            current_chunk.append(sentence)
            current_length += sent_len

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

    return chunks

Pros: Preserves semantic boundaries, cleaner retrieval Cons: Slower (requires NLP parsing), chunk sizes vary

Strategy 3: Recursive Character Splitting

This is the default strategy used by LangChain. It attempts to split on successively smaller separators (\n\n → \n → . → ) until chunks fit the target size.

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]
)

text = open("document.txt").read()
chunks = splitter.split_text(text)
print(f"Created {len(chunks)} chunks")

Pros: Good balance of structure awareness and simplicity, handles diverse document formats Cons: Character-based (not token-based), can still break semantic units at the edges

Strategy 4: Document-Aware Chunking

For structured documents (Markdown, HTML, code), split based on document structure:

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_text = open("README.md").read()
chunks = splitter.split_text(md_text)

for chunk in chunks:
    print(f"Section: {chunk.metadata}")
    print(f"Content: {chunk.page_content[:200]}...")

Pros: Perfect for structured docs (technical documentation, wikis), preserves section hierarchy in metadata Cons: Only works for structured formats

Practical chunking advice

  1. Chunk size: 256–512 tokens is the sweet spot for most use cases. Larger chunks (1024+) work when you need more context per chunk, but retrieval precision drops.
  2. Overlap: Use 10–20% overlap to avoid losing context at chunk boundaries.
  3. Test empirically: Different document types benefit from different strategies. Build an evaluation set and measure retrieval accuracy.

Advanced Techniques

Basic RAG (embed → search → generate) gets you 80% of the way. These advanced techniques push retrieval quality higher and handle edge cases that basic pipelines miss.

Re-ranking with Cross-Encoders

Vector search retrieves candidates fast — but the embedding model's similarity scores are approximate. A cross-encoder takes a (query, document) pair as input and produces a precise relevance score. It's slower but much more accurate.

The standard pattern is two-stage retrieval:

  1. Fast vector search retrieves top 50–100 candidates
  2. A re-ranker scores each candidate against the query and returns the top 5–10
# Using Cohere Rerank (API-based)
import cohere

co = cohere.Client(api_key="...")

def rerank_cohere(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
    """Re-rank documents using Cohere's rerank model."""
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=documents,
        top_n=top_n
    )
    return [
        {"index": r.index, "score": r.relevance_score, "text": documents[r.index]}
        for r in response.results
    ]

# Using a local cross-encoder (free, self-hosted)
from sentence_transformers import CrossEncoder

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

def rerank_local(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
    """Re-rank documents using a local cross-encoder model."""
    pairs = [(query, doc) for doc in documents]
    scores = cross_encoder.predict(pairs)

    ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)[:top_n]
    return [
        {"index": i, "score": float(s), "text": documents[i]}
        for i, s in ranked
    ]

Re-ranking typically improves retrieval accuracy by 5–15% on standard benchmarks. It's one of the highest-ROI improvements you can make to a RAG pipeline.

Hybrid Search

Vector embeddings are great at semantic similarity but poor at exact keyword matching. If a user searches for a specific product code, error message, or person's name, pure vector search may miss it.

Hybrid search combines:

  • Dense vector search (semantic): Embedding similarity
  • Sparse keyword search (lexical): BM25 or TF-IDF
import rank_bm25

def hybrid_search(
    query: str,
    vector_results: list[dict],
    documents: list[str],
    alpha: float = 0.5
) -> list[dict]:
    """
    Combine BM25 keyword scores with vector similarity scores.
    alpha=0.5 gives equal weight to both.
    """
    # BM25 keyword search
    tokenized_docs = [doc.lower().split() for doc in documents]
    bm25 = rank_bm25.BM25Okapi(tokenized_docs)
    bm25_scores = bm25.get_scores(query.lower().split())

    # Normalize BM25 scores to [0, 1]
    max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
    bm25_normalized = [s / max_bm25 for s in bm25_scores]

    # Normalize vector scores (assumed to be cosine similarity, already in [0, 1])
    vector_scores = {r["index"]: r["score"] for r in vector_results}

    # Combine
    combined = []
    for i in range(len(documents)):
        vec_score = vector_scores.get(i, 0)
        keyword_score = bm25_normalized[i]
        final_score = alpha * vec_score + (1 - alpha) * keyword_score
        combined.append({"index": i, "score": final_score, "text": documents[i]})

    combined.sort(key=lambda x: x["score"], reverse=True)
    return combined[:5]

Use hybrid search when your users query for specific identifiers (product names, error codes, document IDs) alongside natural language questions.

HyDE (Hypothetical Document Embedding)

HyDE is a clever technique that improves retrieval for queries that look nothing like the documents they're searching for. The idea: instead of embedding the user's raw query, first ask the LLM to generate a hypothetical answer, then embed that hypothetical answer and search with it.

Why does this work? User questions are short and interrogative. Document passages are long and declarative. The embedding similarity between "What is the refund policy?" and "Refunds are available within 30 days of purchase..." may be low — even though they're clearly about the same topic. By generating a hypothetical answer, the embedded text is closer in form to the documents.

import openai

client = openai.Client(api_key="sk-...")

def hyde_retrieve(query: str, top_k: int = 5) -> list[dict]:
    """Use HyDE to generate a hypothetical answer, then retrieve with it."""
    # Step 1: Generate hypothetical answer
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Generate a short, factual answer to the question. Do not hedge."},
            {"role": "user", "content": query}
        ],
        max_tokens=200
    )
    hypothetical_answer = response.choices[0].message.content

    # Step 2: Embed the hypothetical answer
    embedding_response = client.embeddings.create(
        model="text-embedding-3-large",
        input=hypothetical_answer
    )
    query_vector = embedding_response.data[0].embedding

    # Step 3: Vector search with the hypothetical embedding
    # (Replace with your actual vector DB call)
    results = vector_db.search(query_vector, top_k=top_k)
    return results

HyDE is especially effective for question-answer retrieval over FAQs, help centers, and knowledge bases where queries and documents have very different writing styles.

Building a Complete RAG Pipeline

Let's put everything together into a complete, runnable RAG pipeline. This example uses LangChain for orchestration, but the same concepts apply to raw implementations.

"""
Production RAG Pipeline with LangChain
Dependencies: pip install langchain langchain-openai langchain-community \
                         chromadb sentence-transformers rank-bm25 cohere
"""

import os
from dataclasses import dataclass

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough


@dataclass
class RAGConfig:
    """Configuration for the RAG pipeline."""
    chunk_size: int = 512
    chunk_overlap: int = 50
    top_k: int = 5
    embedding_model: str = "text-embedding-3-large"
    llm_model: str = "gpt-4o"
    persist_dir: str = "./chroma_db"


class RAGPipeline:
    """A complete RAG pipeline: ingest, retrieve, generate."""

    def __init__(self, config: RAGConfig):
        self.config = config
        self.embeddings = OpenAIEmbeddings(
            model=config.embedding_model,
            openai_api_key=os.environ["OPENAI_API_KEY"]
        )
        self.llm = ChatOpenAI(
            model=config.llm_model,
            temperature=0,  # Deterministic for factual answers
            openai_api_key=os.environ["OPENAI_API_KEY"]
        )
        self.vector_store = None
        self.bm25_retriever = None

    # ─── Ingestion ───────────────────────────────────────────

    def ingest_directory(self, dir_path: str):
        """Load, chunk, and index all documents in a directory."""
        print(f"Loading documents from {dir_path}...")
        loader = DirectoryLoader(
            dir_path,
            glob="**/*.{txt,md,pdf}",
            loader_cls=TextLoader,
            show_progress=True
        )
        documents = loader.load()
        print(f"Loaded {len(documents)} documents")

        # Chunk
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.config.chunk_size,
            chunk_overlap=self.config.chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )
        chunks = splitter.split_documents(documents)
        print(f"Split into {len(chunks)} chunks")

        # Index into Chroma
        self.vector_store = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.config.persist_dir
        )

        # Build BM25 retriever for hybrid search
        texts = [chunk.page_content for chunk in chunks]
        self.bm25_retriever = BM25Retriever.from_texts(texts)
        self.bm25_retriever.k = self.config.top_k

        print("Ingestion complete.")

    # ─── Retrieval ───────────────────────────────────────────

    def get_hybrid_retriever(self):
        """Combine vector and BM25 retrievers."""
        vector_retriever = self.vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": self.config.top_k}
        )
        return EnsembleRetriever(
            retrievers=[vector_retriever, self.bm25_retriever],
            weights=[0.5, 0.5]
        )

    # ─── Generation ──────────────────────────────────────────

    PROMPT_TEMPLATE = """You are a knowledgeable assistant. Answer the user's
question based ONLY on the provided context. If the context does not contain
the answer, say "I don't know based on the provided context."

Cite your sources by referencing [Doc N] where N is the document number.

Context:
{context}

Question: {question}

Answer:"""

    def ask(self, question: str) -> str:
        """Ask a question and get a grounded answer."""
        if not self.vector_store:
            raise RuntimeError("No documents ingested. Call ingest_directory() first.")

        retriever = self.get_hybrid_retriever()
        prompt = ChatPromptTemplate.from_template(self.PROMPT_TEMPLATE)

        def format_docs(docs):
            return "\n\n".join(
                f"[Doc {i+1}] {doc.page_content}" for i, doc in enumerate(docs)
            )

        # Build the RAG chain
        rag_chain = (
            {"context": retriever | format_docs, "question": RunnablePassthrough()}
            | prompt
            | self.llm
            | StrOutputParser()
        )

        return rag_chain.invoke(question)

    def ask_with_sources(self, question: str) -> dict:
        """Ask a question and return both the answer and source documents."""
        retriever = self.get_hybrid_retriever()
        retrieved_docs = retriever.invoke(question)

        prompt = ChatPromptTemplate.from_template(self.PROMPT_TEMPLATE)
        context = "\n\n".join(
            f"[Doc {i+1}] {doc.page_content}" for i, doc in enumerate(retrieved_docs)
        )

        answer = (prompt | self.llm | StrOutputParser()).invoke({
            "context": context,
            "question": question
        })

        return {
            "answer": answer,
            "sources": [
                {"content": doc.page_content, "metadata": doc.metadata}
                for doc in retrieved_docs
            ]
        }


# ─── Usage ────────────────────────────────────────────────────

if __name__ == "__main__":
    config = RAGConfig(
        chunk_size=512,
        chunk_overlap=50,
        top_k=5,
        embedding_model="text-embedding-3-large",
        llm_model="gpt-4o"
    )

    rag = RAGPipeline(config)

    # Ingest documents
    rag.ingest_directory("./knowledge_base")

    # Ask questions
    result = rag.ask_with_sources("What is our company's remote work policy?")
    print(f"Answer: {result['answer']}\n")
    print(f"Sources: {len(result['sources'])} documents retrieved")
    for i, src in enumerate(result['sources']):
        print(f"  [{i+1}] {src['metadata'].get('source', 'unknown')}")

This pipeline includes:

  • Document loading from a directory (TXT, Markdown, PDF)
  • Recursive chunking with configurable size and overlap
  • Hybrid retrieval combining vector search (Chroma) with BM25 keyword search
  • Source tracking so every answer links back to the original documents
  • Prompt engineering to constrain answers to retrieved context

Evaluation

How do you know if your RAG system is actually good? You need a systematic evaluation framework. RAGAS (Retrieval Augmented Generation Assessment) is the most widely adopted framework for evaluating RAG pipelines.

What RAGAS measures

RAGAS evaluates three components:

  1. Faithfulness: Does the answer contain only information supported by the retrieved context? (Measures hallucination.)
  2. Answer Relevancy: Is the answer relevant to the question? (Measures usefulness.)
  3. Context Precision: Did the retrieval step find the right documents? (Measures retrieval quality.)

Each metric is scored 0–1. A well-tuned system typically achieves:

  • Faithfulness: > 0.85
  • Answer Relevancy: > 0.80
  • Context Precision: > 0.75

Setting up RAGAS

"""
RAGAS evaluation for RAG pipelines.
Dependencies: pip install ragas datasets
"""

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

# Prepare your evaluation dataset
eval_data = {
    "question": [
        "What is the company's vacation policy?",
        "How do I reset my password?",
        "What is the process for expense reimbursement?",
    ],
    "answer": [
        # The answers your RAG system produced
        "Employees receive 20 days of paid vacation annually...",
        "To reset your password, go to Settings > Security...",
        "Submit expenses through the portal within 30 days...",
    ],
    "contexts": [
        # The retrieved chunks for each question
        ["The employee handbook states that full-time staff..."],
        ["Password resets can be done from the settings page..."],
        ["All business expenses must be submitted via..."],
    ],
    "ground_truth": [
        # Reference answers for context_recall
        "Full-time employees get 20 vacation days per year.",
        "Navigate to Settings > Security > Reset Password.",
        "Use the expense portal within 30 days of the expense.",
    ]
}

dataset = Dataset.from_dict(eval_data)

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

print(results)
# Output example:
# {'faithfulness': 0.88, 'answer_relevancy': 0.82,
#  'context_precision': 0.79, 'context_recall': 0.85}

Beyond RAGAS: Human evaluation

Automated metrics are useful for regression testing, but they don't capture everything. Complement RAGAS with:

  • Human spot-checks: Manually review 20–50 QA pairs weekly
  • User feedback: Add 👍/👎 buttons to your interface and track satisfaction
  • Failure analysis: Log low-rated answers and categorize the failure mode (bad retrieval, bad generation, missing data)

Building an evaluation pipeline

import json

def evaluate_rag_pipeline(rag: RAGPipeline, eval_file: str) -> dict:
    """Run a full evaluation on the RAG pipeline using a JSON eval set."""
    with open(eval_file) as f:
        test_cases = json.load(f)

    results = []
    for case in test_cases:
        # Get RAG response
        response = rag.ask_with_sources(case["question"])

        results.append({
            "question": case["question"],
            "answer": response["answer"],
            "contexts": [s["content"] for s in response["sources"]],
            "ground_truth": case["expected_answer"]
        })

    # Run RAGAS
    dataset = Dataset.from_list(results)
    scores = evaluate(
        dataset,
        metrics=[faithfulness, answer_relevancy, context_precision]
    )

    # Flag low-scoring cases for review
    for i, (result, score_row) in enumerate(zip(results, dataset)):
        if scores["faithfulness"][i] < 0.7:
            print(f"⚠️ Low faithfulness on: {result['question']}")

    return scores

Treat evaluation as a continuous process, not a one-time step. Every time you change a component (embedding model, chunk size, prompt template), re-run your evaluation suite to ensure quality hasn't regressed.

Production Deployment Patterns

Moving from a prototype to production introduces new challenges: latency, scalability, observability, and cost. Here are the patterns that work in 2026.

Pattern 1: Async Ingestion with Task Queues

Document ingestion (loading, chunking, embedding, indexing) is slow. Don't do it in the request path. Use a task queue:

"""
Async ingestion using Celery + Redis.
Documents are queued for processing and indexed in background workers.
"""
from celery import Celery

app = Celery("rag", broker="redis://localhost:6379/0")

@app.task
def ingest_document(file_path: str, metadata: dict):
    """Background task: chunk, embed, and index a single document."""
    from langchain_community.document_loaders import TextLoader
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    from langchain_openai import OpenAIEmbeddings
    from langchain_community.vectorstores import Chroma

    loader = TextLoader(file_path)
    docs = loader.load()

    splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
    chunks = splitter.split_documents(docs)

    for chunk in chunks:
        chunk.metadata.update(metadata)

    embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
    Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

    return {"file": file_path, "chunks": len(chunks)}

# API endpoint submits to queue
def upload_document(file_path: str, source: str):
    result = ingest_document.delay(file_path, {"source": source})
    return {"task_id": result.id}

Pattern 2: Caching

Cache both embeddings and LLM responses to reduce latency and cost:

import hashlib
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=1)

def cached_embed(text: str, model: str = "text-embedding-3-large") -> list[float]:
    """Cache embeddings to avoid re-computing for unchanged text."""
    key = f"embed:{model}:{hashlib.sha256(text.encode()).hexdigest()}"

    cached = redis_client.get(key)
    if cached:
        import json
        return json.loads(cached)

    # Compute embedding
    embedding = embed_openai(text)

    # Cache for 24 hours
    redis_client.setex(key, 86400, json.dumps(embedding))
    return embedding

def cached_rag_answer(question: str, rag: RAGPipeline) -> str:
    """Cache complete RAG responses for identical questions."""
    key = f"rag:{hashlib.sha256(question.encode()).hexdigest()}"

    cached = redis_client.get(key)
    if cached:
        return cached.decode()

    answer = rag.ask(question)
    redis_client.setex(key, 3600, answer)  # 1 hour TTL
    return answer

Pattern 3: Observability with Tracing

In production, you need to see what's happening inside your pipeline. Use LangSmith or Phoenix for end-to-end tracing:

import os

# Enable LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."
os.environ["LANGCHAIN_PROJECT"] = "rag-production"

# Now every LLM call, retrieval, and chain step is automatically traced.
# View traces at https://smith.langchain.com

Tracing lets you see:

  • Which chunks were retrieved for each query
  • The exact prompt sent to the LLM
  • Token counts, latency, and cost per request
  • Where errors occurred in the pipeline

Pattern 4: Streaming Responses

For user-facing applications, stream responses so users see text immediately instead of waiting for the full generation:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/ask")
def ask_endpoint(question: str):
    """Streaming endpoint for RAG responses."""
    def generate():
        retriever = rag.get_hybrid_retriever()
        docs = retriever.invoke(question)
        context = format_docs(docs)

        stream = rag.llm.stream(
            [{"role": "user", "content": f"Context: {context}\n\nQ: {question}"}]
        )

        for chunk in stream:
            data = json.dumps({"token": chunk.content})
            yield f"data: {data}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Pattern 5: Multi-Tenancy

If serving multiple customers, isolate their data:

  • Separate collections per tenant: Each customer gets their own vector index. Clean separation but more overhead.
  • Metadata filtering: One shared index with tenant_id metadata. More efficient but requires careful filtering on every query.
# Metadata filtering approach
def search_tenant(tenant_id: str, query_vector: list[float], top_k: int = 5):
    """Search with tenant isolation via metadata filtering."""
    return vector_store.search(
        query_vector,
        top_k=top_k,
        filter={"tenant_id": tenant_id}  # Only retrieve this tenant's docs
    )

Common Failure Modes and How to Fix Them

Even well-built RAG systems fail in predictable ways. Here are the most common failure modes and their fixes.

Failure 1: Hallucination Despite Retrieved Context

Symptom: The LLM generates an answer that contradicts the retrieved context, or makes up facts not in the documents.

Causes & Fixes:

  • Weak prompt instructions: Make your system prompt stricter. Use phrasing like "Answer using ONLY the provided context. If the answer is not in the context, say 'I don't know.' Do not use any prior knowledge."

  • Retrieved context is irrelevant: If the retrieval step pulls the wrong documents, the LLM has no good context to work with. Improve retrieval quality with re-ranking, hybrid search, or better chunking.

  • Model temperature too high: Set temperature=0 (or 0.1 max) for factual tasks. Higher temperatures encourage creative — and potentially hallucinated — responses.

# Strict anti-hallucination prompt
STRICT_PROMPT = """You are a fact-based assistant. Follow these rules:

1. Answer ONLY using information in the provided context.
2. If the context does not contain the answer, say: "I don't know based on the available information."
3. Do NOT use any prior knowledge, training data, or assumptions.
4. For each claim, reference the source: [Doc N].
5. If multiple sources conflict, note the conflict.

Context:
{context}

Question: {question}

Answer:"""

Failure 2: Poor Retrieval Quality

Symptom: The system retrieves documents that aren't relevant to the question, leading to unhelpful or wrong answers.

Causes & Fixes:

  • Chunks too large: Large chunks dilute relevance. Try reducing chunk size from 1024 to 512 tokens.
  • Chunks too small: Tiny chunks lose context. If chunks are under 100 tokens, the model can't tell what the passage is about. Increase chunk size or add overlap.
  • Wrong embedding model: A domain-specific corpus (medical, legal, code) may need a specialized embedding model.
  • Missing metadata: Add document titles, section headers, or summaries to chunk metadata. Many vector databases support metadata-aware search.
# Enrich chunks with context before embedding
def enrich_chunks(chunks: list[str], doc_title: str) -> list[str]:
    """Add document context to each chunk for better embeddings."""
    return [f"Document: {doc_title}\n\n{chunk}" for chunk in chunks]

Failure 3: Context Window Overflow

Symptom: Retrieved chunks plus the prompt exceed the LLM's context window, causing truncated input or errors.

Causes & Fixes:

  • Too many retrieved chunks: Reduce top_k from 10 to 5, or even 3.
  • Chunks too large: Use smaller chunks (256–512 tokens).
  • Long conversation history: Summarize older messages instead of including them all.
def truncate_context(docs: list[str], max_tokens: int = 4000) -> list[str]:
    """Truncate retrieved context to fit within token budget."""
    total_tokens = 0
    selected = []

    for doc in docs:
        doc_tokens = len(doc.split()) * 1.3  # rough tokens-to-words ratio
        if total_tokens + doc_tokens > max_tokens:
            break
        selected.append(doc)
        total_tokens += doc_tokens

    return selected

Failure 4: Slow Response Times

Symptom: End-to-end latency exceeds 5–10 seconds, creating a poor user experience.

Causes & Fixes:

  • Embedding API latency: Cache query embeddings for common questions. Or switch to a local embedding model to eliminate network calls.
  • Vector search latency: Ensure your vector database has an HNSW or IVF index. Brute-force search is O(n) and won't scale.
  • LLM generation latency: Use a faster model for simple queries (e.g., GPT-4o-mini instead of GPT-4o). Or implement a routing layer that sends complex queries to a powerful model and simple queries to a fast one.
  • Sequential instead of parallel: Embed the query and search in parallel with any pre-processing steps.
import asyncio

async def parallel_rag(question: str):
    """Run retrieval steps in parallel."""
    embedding_task = asyncio.create_task(embed_openai_async(question))
    keyword_task = asyncio.create_task(bm25_search_async(question))

    query_vector = await embedding_task
    keyword_results = await keyword_task

    vector_results = await vector_search_async(query_vector)
    combined = merge_results(vector_results, keyword_results)

    return combined

Failure 5: Stale Knowledge

Symptom: The RAG system returns outdated information because the document store hasn't been updated.

Causes & Fixes:

  • No update pipeline: Build a webhook-triggered ingestion pipeline that re-indexes documents when they change in your CMS, wiki, or database.
  • No document versioning: Store document version or timestamp in metadata and filter out stale versions.
def upsert_document(doc_id: str, content: str, version: int):
    """Update a document: delete old chunks, insert new ones."""
    # Delete existing chunks for this document
    vector_store.delete(filter={"doc_id": doc_id})

    # Insert new chunks
    chunks = semantic_chunk(content)
    for i, chunk in enumerate(chunks):
        vector_store.add(
            texts=[chunk],
            metadatas=[{"doc_id": doc_id, "version": version, "chunk_index": i}]
        )

Conclusion

RAG transforms LLMs from confident hallucinators into grounded, reliable knowledge workers. The architecture is straightforward in principle — retrieve, augment, generate — but getting it production-ready requires careful decisions at every layer:

  1. Choose the right embedding model for your domain, language, and budget.
  2. Pick a vector database that matches your scale, infrastructure, and filtering needs.
  3. Invest in chunking — it's the single highest-leverage decision for retrieval quality.
  4. Add re-ranking and hybrid search to push retrieval accuracy past 90%.
  5. Evaluate continuously with RAGAS so you catch regressions before users do.
  6. Design for production with caching, streaming, observability, and multi-tenancy from day one.

The field is evolving rapidly — multimodal RAG (retrieving images and tables alongside text), agentic RAG (LLMs that decide when to retrieve), and graph RAG (retrieving over knowledge graphs) are all pushing the boundaries. But the fundamentals in this guide will remain relevant regardless of what comes next.

Start simple. Measure everything. Iterate. Ship.

Have questions about implementing RAG in your specific use case? Drop a comment below or reach out — we love talking about retrieval architecture.