RAG System — Build Retrieval-Augmented Generation from Scratch in 2025
Advertisement
Introduction
Why This Matters
Retrieval-Augmented Generation (RAG) is the dominant architecture for deploying LLMs in enterprise environments. It solves the two biggest problems with raw LLMs: knowledge cutoff (models do not know recent events) and hallucination (models fabricate facts with confidence). RAG grounds every response in documents you control, making it accurate, auditable, and safe for production.
The global RAG market is growing rapidly as companies replace static chatbots with knowledge-retrieval systems that can answer questions about internal docs, support tickets, codebases, and proprietary data. Understanding how to build, evaluate, and optimize RAG systems is one of the highest-leverage skills an AI engineer can develop in 2025.
Basic RAG is easy to demo. Production RAG is hard to get right. This guide covers both — from the minimal working implementation to the architectural decisions that determine quality at scale.
RAG Architecture Overview
A RAG system has two phases:
Indexing phase (offline):
- Load documents from sources (PDFs, databases, APIs)
- Split into chunks with appropriate size and overlap
- Generate vector embeddings for each chunk
- Store embeddings and metadata in a vector database
Querying phase (online):
- Embed the user query using the same embedding model
- Retrieve the top-k most similar chunks via approximate nearest neighbor search
- Optionally rerank retrieved chunks for precision
- Construct a prompt combining the query and retrieved context
- Generate a response using the LLM
Minimal Working RAG in Python
from openai import OpenAI
import numpy as np
import json
client = OpenAI()
def embed(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def cosine_similarity(a: list, b: list) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
# Index documents
documents = [
"Our refund policy allows returns within 30 days of purchase.",
"Enterprise plans include 99.9% SLA and dedicated support.",
"API rate limits are 1000 requests per minute on the Pro plan.",
]
doc_embeddings = [(doc, embed(doc)) for doc in documents]
def retrieve(query: str, top_k: int = 2) -> list[str]:
query_emb = embed(query)
scored = [(doc, cosine_similarity(query_emb, emb)) for doc, emb in doc_embeddings]
scored.sort(key=lambda x: x[1], reverse=True)
return [doc for doc, _ in scored[:top_k]]
def answer(query: str) -> str:
context = "\n".join(retrieve(query))
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "Answer based only on the provided context. Say 'I don't know' if the context doesn't contain the answer."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
]
)
return response.choices[0].message.content
print(answer("What is the refund window?"))Chunking Strategies
Chunking strategy is the most impactful decision in RAG quality. Chunk too large and retrieval is noisy; chunk too small and context is incomplete.
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
MarkdownHeaderTextSplitter,
SentenceTransformersTokenTextSplitter,
)
# Standard recursive splitter — good default
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " ", ""]
)
# Markdown-aware — preserves document structure
md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "h1"), ("##", "h2"), ("###", "h3")
]
)
# Token-aware — consistent chunk sizes for embedding models
token_splitter = SentenceTransformersTokenTextSplitter(
chunk_overlap=20,
tokens_per_chunk=256
)Chunking heuristics:
- Technical documentation: 512–1024 tokens with 10–15% overlap
- Legal/compliance documents: 256–512 tokens, sentence-boundary aware
- Code: function-level splitting, not character-level
- Conversational FAQ: keep question-answer pairs together as single chunks
Embedding Models
The embedding model determines retrieval quality. OpenAI embeddings are best for English; multilingual models handle non-English content better.
from openai import OpenAI
from sentence_transformers import SentenceTransformer
# OpenAI — best quality for English, API-based
client = OpenAI()
def openai_embed(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-large", # 3072 dimensions, highest quality
input=texts
)
return [item.embedding for item in response.data]
# Local model — free, private, no API calls
local_model = SentenceTransformer("BAAI/bge-large-en-v1.5") # SOTA open-source
def local_embed(texts: list[str]) -> np.ndarray:
return local_model.encode(texts, normalize_embeddings=True)Advanced Retrieval: Hybrid Search
Combining dense vector search with sparse keyword search (BM25) consistently improves retrieval quality:
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# Dense retriever (semantic)
vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings())
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# Sparse retriever (keyword)
bm25_retriever = BM25Retriever.from_documents(chunks, k=5)
# Hybrid: combine with Reciprocal Rank Fusion
ensemble = EnsembleRetriever(
retrievers=[dense_retriever, bm25_retriever],
weights=[0.6, 0.4] # weight semantic higher
)
results = ensemble.invoke("API rate limit exceeded error")Reranking for Precision
Retrieve a wider candidate set, then rerank with a cross-encoder for precision:
from sentence_transformers import CrossEncoder
from langchain_core.documents import Document
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")
def rerank(query: str, docs: list[Document], top_k: int = 3) -> list[Document]:
pairs = [(query, doc.page_content) for doc in docs]
scores = reranker.predict(pairs)
ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in ranked[:top_k]]
# First retrieve 10 candidates, then rerank to top 3
candidates = ensemble.invoke(query)
top_docs = rerank(query, candidates, top_k=3)Reranking adds 30–50ms latency but measurably improves answer quality for domain-specific queries.
RAG with Citations
Returning source citations builds user trust and enables verification:
from openai import OpenAI
client = OpenAI()
def rag_with_citations(query: str, docs: list) -> dict:
context_parts = []
for i, doc in enumerate(docs):
context_parts.append(f"[{i+1}] {doc.page_content}\nSource: {doc.metadata.get('source', 'unknown')}")
context = "\n\n".join(context_parts)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"Answer the question using only the provided context. "
"Cite sources using [1], [2], etc. after each claim. "
"If the context doesn't answer the question, say so."
)
},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
]
)
return {
"answer": response.choices[0].message.content,
"sources": [doc.metadata.get("source") for doc in docs]
}Evaluating RAG Quality
Use RAGAS to measure your pipeline objectively:
# pip install ragas
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
eval_data = {
"question": ["What is the refund policy?"],
"answer": ["Returns are allowed within 30 days."],
"contexts": [["Our refund policy allows returns within 30 days of purchase."]],
"ground_truth": ["The refund window is 30 days."]
}
dataset = Dataset.from_dict(eval_data)
results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
print(results)Target scores: faithfulness > 0.9, context precision > 0.8.
Common Mistakes / Pitfalls
- Using one-size-fits-all chunk sizes — different document types need different chunking strategies
- Not adding metadata to chunks — source, date, section title enable filtered retrieval and citations
- Skipping reranking — retrieving k=3 directly from the vector store often returns noisy results
- Embedding the query without preprocessing — removing stopwords and normalizing text improves recall
- Not evaluating retrieval separately from generation — you need to know if poor answers come from bad retrieval or bad generation
Best Practices
- Always store metadata (source file, page number, section, date) alongside embeddings
- Use hybrid search (vector + BM25) as your default retrieval strategy — it outperforms pure vector search on most benchmarks
- Implement a reranker as a second-stage filter when precision matters more than recall
- Cache query embeddings for frequently asked questions to reduce latency and cost
- Evaluate on a held-out Q&A set using RAGAS before every significant change to the pipeline
Key Takeaways
- RAG solves the two core LLM limitations: knowledge cutoff and hallucination by grounding responses in retrieved documents
- The indexing pipeline has four stages: load, chunk, embed, store — each decision affects overall system quality
- Chunking strategy is the most impactful tuning lever — chunk size and overlap must match your document type
- Hybrid search combining dense vector retrieval with sparse BM25 keyword search outperforms either method alone
- Cross-encoder reranking as a second-pass filter significantly improves precision over first-pass vector retrieval
- Metadata stored alongside embeddings enables source citations, filtered retrieval, and date-aware search
- RAGAS provides standardized metrics (faithfulness, answer relevancy, context precision, recall) for objective evaluation
- Production RAG quality is determined more by retrieval engineering than by the choice of LLM for generation
Advertisement