Why RAG Still Matters in 2026
If you've been following AI engineering, you know that Retrieval-Augmented Generation (RAG) isn't new. But in 2026, it's more relevant than ever. Despite context windows growing to millions of tokens, RAG remains the most cost-effective, accurate, and controllable way to ground LLMs in your organization's data.
Here's the reality: stuffing your entire knowledge base into a context window is expensive, slow, and surprisingly unreliable at scale. RAG gives you surgical precision — retrieving exactly the information the model needs, when it needs it.
In this guide, we'll build a production-grade RAG pipeline from scratch. No toy examples. No "just use LangChain" hand-waving. Real engineering decisions, real code, and real tradeoffs.
The Architecture: What We're Building
A modern RAG pipeline has five core stages:
- 1Document Ingestion — Loading and preprocessing your source documents
- 2Chunking — Breaking documents into retrievable units
- 3Embedding — Converting chunks into vector representations
- 4Retrieval — Finding the most relevant chunks for a query
- 5Generation — Feeding retrieved context to the LLM for response synthesis
Let's build each one.
Step 1: Document Ingestion
Before anything else, you need to get your data into a format the pipeline can process. In production, this means handling PDFs, HTML, Markdown, Slack exports, Confluence pages, and more.
import os
from pathlib import Path
from dataclasses import dataclass
from typing import List
@dataclass
class Document:
content: str
metadata: dict
source: str
class DocumentLoader:
"""Unified document loader for multiple formats."""
def load_directory(self, path: str) -> List[Document]:
documents = []
for file_path in Path(path).rglob("*"):
if file_path.suffix == ".md":
documents.append(self._load_markdown(file_path))
elif file_path.suffix == ".txt":
documents.append(self._load_text(file_path))
elif file_path.suffix == ".pdf":
documents.append(self._load_pdf(file_path))
return documents
def _load_markdown(self, path: Path) -> Document:
content = path.read_text(encoding="utf-8")
return Document(
content=content,
metadata={"format": "markdown", "filename": path.name},
source=str(path)
)
def _load_text(self, path: Path) -> Document:
content = path.read_text(encoding="utf-8")
return Document(
content=content,
metadata={"format": "text", "filename": path.name},
source=str(path)
)
def _load_pdf(self, path: Path) -> Document:
# In production, use pymupdf or pdfplumber
import fitz # PyMuPDF
doc = fitz.open(str(path))
content = "\n".join(page.get_text() for page in doc)
return Document(
content=content,
metadata={"format": "pdf", "filename": path.name, "pages": len(doc)},
source=str(path)
)Production tip: Always preserve metadata. Source file, page number, section headers — you'll need these later for citation and debugging.
Step 2: Chunking — The Most Underrated Step
Chunking is where most RAG pipelines quietly fail. Chunk too big and your retrieval gets noisy. Chunk too small and you lose context. The right strategy depends entirely on your data.
The Three Chunking Strategies That Work
1. Recursive Character Splitting — The reliable default:
class RecursiveChunker:
"""Split text recursively by separators, preserving structure."""
def __init__(self, chunk_size: int = 512, chunk_overlap: int = 64):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.separators = ["\n\n", "\n", ". ", " "]
def chunk(self, text: str) -> List[str]:
chunks = []
self._split_recursive(text, self.separators, chunks)
return chunks
def _split_recursive(self, text: str, separators: List[str], chunks: List[str]):
if len(text) <= self.chunk_size:
if text.strip():
chunks.append(text.strip())
return
sep = separators[0] if separators else ""
parts = text.split(sep) if sep else [text[i:i+self.chunk_size] for i in range(0, len(text), self.chunk_size)]
current = ""
for part in parts:
candidate = current + sep + part if current else part
if len(candidate) > self.chunk_size and current:
chunks.append(current.strip())
# Keep overlap from end of previous chunk
overlap_text = current[-self.chunk_overlap:] if len(current) > self.chunk_overlap else current
current = overlap_text + sep + part
else:
current = candidate
if current.strip():
if len(current) > self.chunk_size and len(separators) > 1:
self._split_recursive(current, separators[1:], chunks)
else:
chunks.append(current.strip())2. Semantic Chunking — Split based on meaning shifts:
import numpy as np
from sentence_transformers import SentenceTransformer
class SemanticChunker:
"""Split text at semantic boundaries using embedding similarity."""
def __init__(self, model_name: str = "all-MiniLM-L6-v2", threshold: float = 0.5):
self.model = SentenceTransformer(model_name)
self.threshold = threshold
def chunk(self, text: str) -> List[str]:
sentences = text.split(". ")
if len(sentences) <= 1:
return [text]
# Embed all sentences
embeddings = self.model.encode(sentences)
# Find breakpoints where cosine similarity drops
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
similarity = np.dot(embeddings[i-1], embeddings[i]) / (
np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
)
if similarity < self.threshold:
# Semantic shift — start new chunk
chunks.append(". ".join(current_chunk) + ".")
current_chunk = [sentences[i]]
else:
current_chunk.append(sentences[i])
if current_chunk:
chunks.append(". ".join(current_chunk) + ".")
return chunks3. Agentic / Contextual Chunking — The 2026 approach that's gaining traction:
from openai import OpenAI
class ContextualChunker:
"""Use an LLM to add context to each chunk for better retrieval."""
def __init__(self, base_chunker: RecursiveChunker):
self.base_chunker = base_chunker
self.client = OpenAI()
def chunk_with_context(self, document: str) -> List[dict]:
# First, create base chunks
raw_chunks = self.base_chunker.chunk(document)
contextualized = []
for chunk in raw_chunks:
# Ask the LLM to situate this chunk within the document
context = self._generate_context(document[:2000], chunk)
contextualized.append({
"content": chunk,
"context": context,
"searchable": f"{context}\n\n{chunk}"
})
return contextualized
def _generate_context(self, doc_summary: str, chunk: str) -> str:
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Given this document excerpt and a specific chunk,
write a 1-2 sentence context that explains where this chunk fits
in the broader document. This will be prepended to the chunk for
better search retrieval.
Document start: {doc_summary}
Chunk: {chunk}
Context:"""
}],
max_tokens=100,
temperature=0
)
return response.choices[0].message.content.strip()Our recommendation: Start with recursive chunking (512 tokens, 64 token overlap). Move to contextual chunking when you need higher retrieval accuracy and can absorb the indexing cost.
Step 3: Embedding — Choosing the Right Model
The embedding model determines how well your retrieval works. Here's the landscape in 2026:
| Model | Dimensions | Speed | Quality | Cost |
|---|---|---|---|---|
| OpenAI text-embedding-3-large | 3072 | Fast | Excellent | $0.13/1M tokens |
| Cohere embed-v4 | 1024 | Fast | Excellent | $0.10/1M tokens |
| BGE-M3 (open source) | 1024 | Moderate | Very Good | Free (self-hosted) |
| all-MiniLM-L6-v2 | 384 | Very Fast | Good | Free (self-hosted) |
from openai import OpenAI
from typing import List
class EmbeddingService:
"""Production embedding service with batching and caching."""
def __init__(self, model: str = "text-embedding-3-large"):
self.client = OpenAI()
self.model = model
self._cache = {}
def embed_texts(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# Check cache first
uncached = [(j, t) for j, t in enumerate(batch) if t not in self._cache]
if uncached:
indices, uncached_texts = zip(*uncached) if uncached else ([], [])
response = self.client.embeddings.create(
model=self.model,
input=list(uncached_texts)
)
for idx, embedding_obj in zip(indices, response.data):
self._cache[batch[idx]] = embedding_obj.embedding
all_embeddings.extend([self._cache[t] for t in batch])
return all_embeddings
def embed_query(self, query: str) -> List[float]:
"""Embed a single query — no caching for queries."""
response = self.client.embeddings.create(
model=self.model,
input=query
)
return response.data[0].embeddingCritical insight: Always embed your queries the same way you embed your documents. If you add contextual prefixes to documents during indexing, account for that asymmetry at query time.
Step 4: Vector Store and Retrieval
You need somewhere to store and search your embeddings. Here's the honest comparison:
- Pinecone / Weaviate / Qdrant — Managed vector DBs. Great for getting started, scales well.
- pgvector (PostgreSQL) — If you already run Postgres, this is surprisingly good for < 10M vectors.
- ChromaDB — Perfect for local development and prototyping.
Here's a production retrieval implementation with pgvector:
import psycopg2
import numpy as np
from typing import List, Tuple
class VectorStore:
"""PostgreSQL + pgvector retrieval store."""
def __init__(self, connection_string: str):
self.conn = psycopg2.connect(connection_string)
self._ensure_schema()
def _ensure_schema(self):
with self.conn.cursor() as cur:
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
cur.execute("""
CREATE TABLE IF NOT EXISTS documents (
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
embedding vector(3072),
created_at TIMESTAMP DEFAULT NOW()
)
""")
cur.execute("""
CREATE INDEX IF NOT EXISTS documents_embedding_idx
ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100)
""")
self.conn.commit()
def insert(self, content: str, embedding: List[float], metadata: dict = None):
with self.conn.cursor() as cur:
cur.execute(
"INSERT INTO documents (content, embedding, metadata) VALUES (%s, %s, %s)",
(content, embedding, psycopg2.extras.Json(metadata or {}))
)
self.conn.commit()
def search(self, query_embedding: List[float], top_k: int = 5) -> List[Tuple[str, float, dict]]:
with self.conn.cursor() as cur:
cur.execute("""
SELECT content, 1 - (embedding <=> %s::vector) as similarity, metadata
FROM documents
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_embedding, query_embedding, top_k))
return [(row[0], row[1], row[2]) for row in cur.fetchall()]Hybrid Search: The Secret Weapon
Pure vector search misses keyword-exact matches. Pure keyword search misses semantic meaning. The answer? Combine both.
class HybridRetriever:
"""Combine vector similarity with BM25 keyword search."""
def __init__(self, vector_store: VectorStore, bm25_weight: float = 0.3):
self.vector_store = vector_store
self.bm25_weight = bm25_weight
def search(self, query: str, query_embedding: List[float], top_k: int = 5) -> List[dict]:
# Vector search results
vector_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
# BM25 / full-text search results
keyword_results = self._keyword_search(query, top_k=top_k * 2)
# Reciprocal Rank Fusion to merge results
return self._reciprocal_rank_fusion(
vector_results, keyword_results, top_k=top_k
)
def _keyword_search(self, query: str, top_k: int) -> List[Tuple[str, float, dict]]:
with self.vector_store.conn.cursor() as cur:
cur.execute("""
SELECT content, ts_rank(to_tsvector('english', content),
plainto_tsquery('english', %s)) as rank, metadata
FROM documents
WHERE to_tsvector('english', content) @@ plainto_tsquery('english', %s)
ORDER BY rank DESC
LIMIT %s
""", (query, query, top_k))
return [(row[0], row[1], row[2]) for row in cur.fetchall()]
def _reciprocal_rank_fusion(self, *result_lists, top_k: int = 5, k: int = 60):
scores = {}
for results in result_lists:
for rank, (content, _, metadata) in enumerate(results):
if content not in scores:
scores[content] = {"score": 0, "metadata": metadata}
scores[content]["score"] += 1 / (k + rank + 1)
sorted_results = sorted(scores.items(), key=lambda x: x[1]["score"], reverse=True)
return [{"content": c, **s} for c, s in sorted_results[:top_k]]Step 5: Reranking — The Accuracy Multiplier
Retrieval gets you candidates. Reranking orders them by actual relevance. This single step can boost answer quality by 15-30%.
from openai import OpenAI
class Reranker:
"""Cross-encoder reranker using an LLM for high-quality relevance scoring."""
def __init__(self):
self.client = OpenAI()
def rerank(self, query: str, documents: List[dict], top_k: int = 3) -> List[dict]:
scored = []
for doc in documents:
score = self._score_relevance(query, doc["content"])
scored.append({**doc, "relevance_score": score})
scored.sort(key=lambda x: x["relevance_score"], reverse=True)
return scored[:top_k]
def _score_relevance(self, query: str, document: str) -> float:
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Rate the relevance of this document to the query on a scale of 0-10.
Return ONLY a number.
Query: {query}
Document: {document[:1000]}
Relevance score:"""
}],
max_tokens=3,
temperature=0
)
try:
return float(response.choices[0].message.content.strip())
except ValueError:
return 0.0For production: Use a dedicated cross-encoder model like Cohere Rerank or a fine-tuned BERT cross-encoder. LLM-based reranking works but is slower and more expensive at scale.
Step 6: Generation — Putting It All Together
Now we assemble the complete pipeline:
from openai import OpenAI
from typing import List
class RAGPipeline:
"""Complete RAG pipeline: retrieve, rerank, generate."""
def __init__(self, retriever: HybridRetriever, reranker: Reranker, embedder: EmbeddingService):
self.retriever = retriever
self.reranker = reranker
self.embedder = embedder
self.client = OpenAI()
def query(self, user_question: str, top_k: int = 3) -> dict:
# Step 1: Embed the query
query_embedding = self.embedder.embed_query(user_question)
# Step 2: Retrieve candidates
candidates = self.retriever.search(user_question, query_embedding, top_k=top_k * 3)
# Step 3: Rerank for precision
reranked = self.reranker.rerank(user_question, candidates, top_k=top_k)
# Step 4: Build context
context = "\n\n---\n\n".join([
f"[Source: {doc.get('metadata', {}).get('source', 'unknown')}]\n{doc['content']}"
for doc in reranked
])
# Step 5: Generate answer
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": """You are a helpful assistant that answers questions
based on the provided context. Always cite your sources. If the context
doesn't contain enough information to answer, say so honestly."""
},
{
"role": "user",
"content": f"""Context:
{context}
Question: {user_question}
Please answer based on the context above. Cite specific sources when possible."""
}
],
temperature=0.1
)
return {
"answer": response.choices[0].message.content,
"sources": [doc.get("metadata", {}) for doc in reranked],
"num_candidates": len(candidates),
"num_used": len(reranked)
}Evaluation: How to Know Your RAG Actually Works
Building a RAG pipeline is one thing. Knowing if it's good enough for production is another. Here are the metrics that matter:
class RAGEvaluator:
"""Evaluate RAG pipeline quality with standard metrics."""
def __init__(self, pipeline: RAGPipeline):
self.pipeline = pipeline
self.client = OpenAI()
def evaluate_retrieval(self, test_cases: List[dict]) -> dict:
"""Measure retrieval quality: are we finding the right documents?"""
total_recall = 0
total_precision = 0
for case in test_cases:
query_embedding = self.pipeline.embedder.embed_query(case["query"])
results = self.pipeline.retriever.search(
case["query"], query_embedding, top_k=5
)
retrieved_sources = {r.get("metadata", {}).get("source") for r in results}
expected_sources = set(case["expected_sources"])
hits = retrieved_sources & expected_sources
recall = len(hits) / len(expected_sources) if expected_sources else 0
precision = len(hits) / len(retrieved_sources) if retrieved_sources else 0
total_recall += recall
total_precision += precision
n = len(test_cases)
return {
"mean_recall": total_recall / n,
"mean_precision": total_precision / n,
}
def evaluate_answer_quality(self, test_cases: List[dict]) -> dict:
"""Use LLM-as-judge to assess answer quality."""
scores = []
for case in test_cases:
result = self.pipeline.query(case["query"])
judgment = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"""Rate this answer on faithfulness (1-5) and relevance (1-5).
Question: {case["query"]}
Expected Answer: {case["expected_answer"]}
Actual Answer: {result["answer"]}
Return JSON: {{"faithfulness": X, "relevance": X}}"""
}],
temperature=0
)
scores.append(judgment.choices[0].message.content)
return {"evaluations": scores}Common Pitfalls and How to Avoid Them
After helping hundreds of teams build RAG systems, here are the mistakes we see most often:
1. Ignoring chunk boundaries. A chunk that starts mid-sentence or cuts off a code block is useless. Always respect document structure.
2. Not evaluating retrieval separately from generation. If your retrieved documents are wrong, no amount of prompt engineering will fix the answer. Measure retrieval recall independently.
3. Over-indexing everything. Not all content is created equal. Index your most valuable, authoritative content first. Quality beats quantity.
4. Skipping reranking. The difference between "top 20 by embedding similarity" and "top 3 after cross-encoder reranking" is dramatic. Don't skip this step.
5. Forgetting about freshness. Documents change. Build incremental indexing from day one, not as an afterthought.
What's Next: Advanced Patterns
Once your basic pipeline works, consider these advanced techniques:
- Query decomposition — Break complex questions into sub-queries and merge results
- Hypothetical document embeddings (HyDE) — Generate a hypothetical answer, embed that, and use it for retrieval
- Agentic RAG — Let an agent decide when and how to retrieve, with tool-use loops
- Graph RAG — Combine knowledge graphs with vector retrieval for relationship-aware answers
Build These Skills Systematically
RAG is just one piece of the AI engineering puzzle. At TrainIQ, our comprehensive AI training program covers RAG systems end-to-end in Module 5 — including hands-on labs where you build, evaluate, and optimize production pipelines. But we also cover the foundations (embeddings, LLMs, prompt engineering) and advanced topics (agents, deployment, evaluation) that make the difference between a demo and a production system.
**Explore the full TrainIQ curriculum →**
Whether you're an individual practitioner or leading a team, mastering these patterns is what separates AI-curious from AI-capable. The pipeline we built in this guide is production-ready — but there's always more to learn.