Retrieval Augmented Generation (RAG) has become the cornerstone of enterprise AI applications in 2025. While large language models possess impressive general knowledge, they struggle with company-specific information, recent data, and domain expertise. RAG solves this by grounding LLM responses in your actual data—documents, databases, and knowledge bases. This comprehensive guide covers everything you need to build production-ready RAG systems, from basic concepts to advanced optimization techniques.
Understanding RAG: Beyond Simple Retrieval
RAG combines the generative capabilities of large language models with precise information retrieval. Instead of relying solely on the model's training data, RAG retrieves relevant context from your knowledge base and uses it to generate accurate, grounded responses.
"RAG is not just about connecting an LLM to a search engine. It's about creating a symbiotic system where retrieval and generation enhance each other to produce responses that are both fluent and factually grounded."
— Patrick Lewis, Lead Author of the RAG Paper
Why RAG Matters in 2025
- Accuracy: Ground responses in verified, up-to-date information
- Reduced Hallucination: Provide evidence for claims, making it easier to verify responses
- Data Privacy: Keep sensitive data in your infrastructure instead of fine-tuning models
- Cost Efficiency: Cheaper than fine-tuning for most use cases
- Flexibility: Update knowledge without retraining models
- Auditability: Track which sources informed each response
RAG Architecture: The Complete Pipeline
A production RAG system consists of several interconnected components: document processing, embedding generation, vector storage, retrieval, and generation. Let's examine each component in detail.
The RAG pipeline begins long before a user asks a question. In the ingestion phase, documents are processed, chunked into appropriate segments, and converted into vector embeddings that capture their semantic meaning. These embeddings are stored in a vector database optimized for similarity search. When a query arrives, it too is converted to an embedding, and the system retrieves the most semantically similar document chunks. Finally, these retrieved chunks provide context for the language model to generate an accurate, grounded response.
Each step in this pipeline offers opportunities for optimization. Document chunking strategies affect what context is retrievable. Embedding model choice impacts semantic understanding. Vector store configuration determines search speed and accuracy. Retrieval strategies—from simple nearest-neighbor search to sophisticated multi-query approaches—affect recall and precision. The generation step must synthesize retrieved context with the query to produce coherent, accurate responses.
The following implementation demonstrates a production-ready RAG pipeline that incorporates best practices at each stage, including query transformation, hybrid search, reranking, and proper context construction:
# Complete RAG System Architecture
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import asyncio
@dataclass
class Document:
"""Represents a document chunk with metadata"""
id: str
content: str
metadata: Dict[str, Any]
embedding: Optional[List[float]] = None
@dataclass
class RetrievalResult:
"""Result from retrieval with relevance score"""
document: Document
score: float
source: str # 'vector', 'keyword', 'hybrid'
class RAGPipeline:
"""Production-ready RAG pipeline"""
def __init__(
self,
embedding_model: str = "text-embedding-3-large",
llm_model: str = "gpt-4-turbo",
vector_store: "VectorStore" = None,
reranker: "Reranker" = None
):
self.embedder = EmbeddingModel(embedding_model)
self.llm = LLMClient(llm_model)
self.vector_store = vector_store or ChromaVectorStore()
self.reranker = reranker or CohereReranker()
self.query_transformer = QueryTransformer()
async def query(
self,
question: str,
top_k: int = 10,
rerank_top_k: int = 5,
filters: Dict = None
) -> Dict[str, Any]:
"""Execute RAG query and return response with sources"""
# Step 1: Query Understanding & Transformation
enhanced_queries = await self.query_transformer.transform(question)
# Step 2: Multi-Query Retrieval
all_results = []
for query in enhanced_queries:
results = await self._retrieve(query, top_k, filters)
all_results.extend(results)
# Step 3: Deduplicate and merge results
unique_results = self._deduplicate(all_results)
# Step 4: Rerank for relevance
reranked = await self.reranker.rerank(
question,
unique_results,
top_k=rerank_top_k
)
# Step 5: Generate response with context
context = self._build_context(reranked)
response = await self._generate(question, context)
return {
"answer": response,
"sources": [{
"id": r.document.id,
"content": r.document.content[:500],
"metadata": r.document.metadata,
"relevance_score": r.score
} for r in reranked],
"query_variations": enhanced_queries
}
async def _retrieve(
self,
query: str,
top_k: int,
filters: Dict
) -> List[RetrievalResult]:
"""Hybrid retrieval combining vector and keyword search"""
# Vector search
query_embedding = await self.embedder.embed(query)
vector_results = await self.vector_store.search(
embedding=query_embedding,
top_k=top_k,
filters=filters
)
# Keyword search (BM25)
keyword_results = await self.vector_store.keyword_search(
query=query,
top_k=top_k,
filters=filters
)
# Combine with Reciprocal Rank Fusion
return self._reciprocal_rank_fusion(
vector_results,
keyword_results,
weights=[0.7, 0.3]
)
def _reciprocal_rank_fusion(
self,
*result_lists: List[RetrievalResult],
weights: List[float] = None,
k: int = 60
) -> List[RetrievalResult]:
"""Combine multiple result lists using RRF"""
if weights is None:
weights = [1.0] * len(result_lists)
scores = {}
doc_map = {}
for results, weight in zip(result_lists, weights):
for rank, result in enumerate(results):
doc_id = result.document.id
rrf_score = weight / (k + rank + 1)
if doc_id in scores:
scores[doc_id] += rrf_score
else:
scores[doc_id] = rrf_score
doc_map[doc_id] = result
# Sort by combined score
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [
RetrievalResult(
document=doc_map[doc_id].document,
score=scores[doc_id],
source="hybrid"
)
for doc_id in sorted_ids
]The implementation above demonstrates several crucial patterns. The query transformation step generates multiple query variations to improve recall—a user's question might not use the same terminology as the source documents. Hybrid retrieval combines semantic search (which understands meaning) with keyword search (which catches exact matches), leveraging the strengths of both approaches.
Reciprocal Rank Fusion (RRF) is a powerful technique for combining results from different retrieval methods. Rather than trying to normalize scores across methods (which is unreliable), RRF uses ranking positions. Documents that rank highly in multiple retrieval methods get boosted, while documents that only appear in one method are appropriately weighted. The k parameter (typically 60) controls how much weight is given to lower-ranked results.
Document Processing and Chunking
How you chunk documents significantly impacts retrieval quality. Poor chunking leads to incomplete context or irrelevant noise. Here are the strategies that work best in production.
Chunking is deceptively simple to get wrong. The intuitive approach—splitting documents at fixed character or token boundaries—often breaks sentences mid-thought, separates context from its explanation, and creates chunks that are difficult to understand in isolation. The retrieved chunk might contain an answer but lack the context needed to understand it, or conversely, might contain context but not the specific information requested.
Effective chunking requires understanding the structure of your documents. Technical documentation has different natural boundaries than legal contracts. Code files should be chunked differently than prose. Tables and structured data need special handling. The goal is to create chunks that are self-contained units of meaning—large enough to be useful, small enough to be precise.
Chunking Strategies
- Fixed-size Chunking: Simple but often breaks mid-sentence. Use with overlap.
- Semantic Chunking: Split on natural boundaries (paragraphs, sections, sentences)
- Recursive Chunking: Hierarchically split large documents while respecting structure
- Document-Type Aware: Different strategies for PDFs, code, tables, etc.
- Agentic Chunking: Use LLMs to determine optimal chunk boundaries
The choice of chunking strategy depends on your document types and retrieval requirements. Fixed-size chunking is simple to implement but risks breaking semantic units. Semantic chunking respects natural document structure but may produce uneven chunk sizes. Recursive chunking offers a good balance by progressively splitting content while trying to maintain coherence. For specialized content types like code or structured data, custom chunking logic often yields the best results.
Chunk overlap is a critical but often overlooked parameter. By including some content from adjacent chunks, overlap ensures that context spanning chunk boundaries isn't lost. However, too much overlap wastes storage and can lead to redundant retrieval results. A typical overlap of 10-20% of chunk size provides a good balance between context preservation and efficiency.
The following implementation demonstrates multiple chunking strategies, including specialized handlers for code, markdown, and general text. The semantic chunking approach uses embeddings to identify natural breakpoints where topic shifts occur:
# Advanced Chunking Strategies
import re
from typing import List, Tuple
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken
class SemanticChunker:
"""Chunk documents while preserving semantic coherence"""
def __init__(
self,
max_chunk_size: int = 512,
chunk_overlap: int = 50,
embedding_model: str = "text-embedding-3-large"
):
self.max_chunk_size = max_chunk_size
self.chunk_overlap = chunk_overlap
self.tokenizer = tiktoken.encoding_for_model("gpt-4")
self.embedder = EmbeddingModel(embedding_model)
def chunk_document(self, text: str, doc_type: str = "general") -> List[Document]:
"""Chunk document based on type"""
if doc_type == "code":
return self._chunk_code(text)
elif doc_type == "markdown":
return self._chunk_markdown(text)
elif doc_type == "table":
return self._chunk_table(text)
else:
return self._chunk_general(text)
def _chunk_general(self, text: str) -> List[Document]:
"""Smart recursive chunking for general text"""
# First, split on natural boundaries
paragraphs = self._split_paragraphs(text)
chunks = []
current_chunk = ""
current_tokens = 0
for para in paragraphs:
para_tokens = len(self.tokenizer.encode(para))
if current_tokens + para_tokens <= self.max_chunk_size:
current_chunk += para + "\n\n"
current_tokens += para_tokens
else:
if current_chunk:
chunks.append(current_chunk.strip())
# Handle paragraphs larger than max size
if para_tokens > self.max_chunk_size:
sub_chunks = self._split_large_paragraph(para)
chunks.extend(sub_chunks)
current_chunk = ""
current_tokens = 0
else:
current_chunk = para + "\n\n"
current_tokens = para_tokens
if current_chunk:
chunks.append(current_chunk.strip())
# Add overlap between chunks
return self._add_overlap(chunks)
def _chunk_code(self, code: str) -> List[Document]:
"""Chunk code while preserving function/class boundaries"""
# Split on function/class definitions
patterns = [
r'(^class\s+\w+.*?(?=^class\s|^def\s|\Z))', # Python classes
r'(^def\s+\w+.*?(?=^def\s|^class\s|\Z))', # Python functions
r'(^function\s+\w+.*?(?=^function\s|\Z))', # JavaScript functions
r'(^const\s+\w+\s*=\s*(?:async\s+)?\(.*?(?=^const\s|^function\s|\Z))',
]
chunks = []
for pattern in patterns:
matches = re.findall(pattern, code, re.MULTILINE | re.DOTALL)
for match in matches:
if len(self.tokenizer.encode(match)) <= self.max_chunk_size:
chunks.append(match)
else:
# Split large functions by logical blocks
chunks.extend(self._split_code_block(match))
return [Document(id=f"code_{i}", content=c, metadata={"type": "code"})
for i, c in enumerate(chunks)]
def _chunk_markdown(self, text: str) -> List[Document]:
"""Chunk markdown preserving header hierarchy"""
# Split on headers while keeping header context
sections = re.split(r'(^#{1,6}\s+.+$)', text, flags=re.MULTILINE)
chunks = []
current_headers = []
current_content = ""
for section in sections:
if re.match(r'^#{1,6}\s+', section):
# It's a header
level = len(re.match(r'^(#+)', section).group(1))
# Update header stack
current_headers = current_headers[:level-1] + [section]
if current_content:
chunks.append({
"headers": current_headers[:-1].copy(),
"content": current_content.strip()
})
current_content = ""
else:
current_content += section
if current_content:
chunks.append({
"headers": current_headers.copy(),
"content": current_content.strip()
})
# Create documents with header context
return [
Document(
id=f"md_{i}",
content=" > ".join(c["headers"]) + "\n\n" + c["content"],
metadata={"headers": c["headers"], "type": "markdown"}
)
for i, c in enumerate(chunks) if c["content"]
]
async def semantic_chunk(self, text: str, threshold: float = 0.5) -> List[Document]:
"""Use embeddings to find natural semantic boundaries"""
sentences = self._split_sentences(text)
# Embed all sentences
embeddings = await self.embedder.embed_batch(sentences)
# Find semantic breaks (low similarity between consecutive sentences)
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
similarity = self._cosine_similarity(embeddings[i-1], embeddings[i])
if similarity < threshold:
# Semantic break detected
chunks.append(" ".join(current_chunk))
current_chunk = [sentences[i]]
else:
current_chunk.append(sentences[i])
if current_chunk:
chunks.append(" ".join(current_chunk))
return [Document(id=f"sem_{i}", content=c, metadata={"type": "semantic"})
for i, c in enumerate(chunks)]The chunking implementation above demonstrates several sophisticated approaches. The markdown chunker preserves header hierarchy, including parent headers in chunk metadata so that retrieved content maintains its organizational context. The code chunker respects function and class boundaries, ensuring that code snippets remain complete and executable. The semantic chunker uses embedding similarity to identify natural topic shifts, creating chunks that represent coherent ideas rather than arbitrary text segments.
Embedding Models and Vector Stores
The choice of embedding model and vector database significantly impacts retrieval quality. Here's what you need to know about the latest options in 2025.
Embedding models transform text into dense vector representations that capture semantic meaning. Two pieces of text with similar meanings will have vectors that are close together in the embedding space, regardless of the specific words used. This property enables semantic search—finding relevant content based on meaning rather than keyword matching. The quality of your embeddings directly impacts retrieval accuracy.
When choosing an embedding model, consider several factors. Dimension size affects storage requirements and search speed—higher dimensions capture more nuance but require more resources. Maximum token length determines how much text can be embedded at once. Performance on benchmarks like MTEB indicates general-purpose quality, but task-specific evaluation on your data is essential. Some models excel at particular domains like code or multilingual content.
Embedding Model Comparison (2025)
The embedding model landscape has evolved significantly, with specialized models emerging for different use cases. The table below compares the leading options available in 2025:
┌─────────────────────────────────────────────────────────────────────────────┐
│ EMBEDDING MODELS COMPARISON 2025 │
├─────────────────────┬───────────┬──────────┬────────────┬──────────────────┤
│ Model │ Dimensions│ MTEB Avg │ Max Tokens │ Best For │
├─────────────────────┼───────────┼──────────┼────────────┼──────────────────┤
│ text-embedding-3-lg │ 3072 │ 64.6 │ 8191 │ General purpose │
│ text-embedding-3-sm │ 1536 │ 62.3 │ 8191 │ Cost-sensitive │
│ voyage-3 │ 1024 │ 67.1 │ 32000 │ Long documents │
│ voyage-code-3 │ 1024 │ - │ 16000 │ Code retrieval │
│ cohere-embed-v3 │ 1024 │ 66.3 │ 512 │ Multilingual │
│ jina-embeddings-v3 │ 1024 │ 65.5 │ 8192 │ Multilingual │
│ bge-m3 │ 1024 │ 66.0 │ 8192 │ Hybrid search │
│ nomic-embed-text │ 768 │ 62.4 │ 8192 │ Open source │
│ e5-mistral-7b │ 4096 │ 66.6 │ 32768 │ Long context │
└─────────────────────┴───────────┴──────────┴────────────┴──────────────────┘# Vector Store Configuration for Production
import os
from abc import ABC, abstractmethod
class VectorStoreConfig:
"""Configuration for vector stores"""
# Pinecone configuration (managed, high-scale)
PINECONE_CONFIG = {
"api_key": os.getenv("PINECONE_API_KEY"),
"environment": "us-west-2-aws",
"index_name": "rag-production",
"metric": "cosine",
"pods": 2,
"replicas": 2,
"pod_type": "p2.x1" # Performance tier
}
# Qdrant configuration (self-hosted or cloud)
QDRANT_CONFIG = {
"url": os.getenv("QDRANT_URL", "http://localhost:6333"),
"api_key": os.getenv("QDRANT_API_KEY"),
"collection_name": "documents",
"vector_size": 3072,
"distance": "Cosine",
"hnsw_config": {
"m": 16,
"ef_construct": 100
},
"quantization_config": {
"scalar": {
"type": "int8",
"always_ram": True
}
}
}
# Weaviate configuration (multimodal support)
WEAVIATE_CONFIG = {
"url": os.getenv("WEAVIATE_URL"),
"api_key": os.getenv("WEAVIATE_API_KEY"),
"class_name": "Document",
"vectorizer": "text2vec-openai",
"module_config": {
"generative-openai": {
"model": "gpt-4-turbo"
}
}
}
class HybridVectorStore:
"""Vector store with hybrid search capabilities"""
def __init__(self, config: dict):
self.qdrant = QdrantClient(**config)
self.collection = config["collection_name"]
async def upsert(self, documents: List[Document]):
"""Insert or update documents with both dense and sparse vectors"""
points = []
for doc in documents:
# Compute sparse vector for BM25-like retrieval
sparse_vector = self._compute_sparse_vector(doc.content)
points.append({
"id": doc.id,
"vector": {
"dense": doc.embedding, # Dense embedding
"sparse": sparse_vector # Sparse BM25 vector
},
"payload": {
"content": doc.content,
**doc.metadata
}
})
await self.qdrant.upsert(
collection_name=self.collection,
points=points
)
async def hybrid_search(
self,
query_embedding: List[float],
query_text: str,
top_k: int = 10,
alpha: float = 0.7, # Weight for dense vs sparse
filters: dict = None
) -> List[RetrievalResult]:
"""Perform hybrid dense + sparse search"""
sparse_query = self._compute_sparse_vector(query_text)
results = await self.qdrant.search(
collection_name=self.collection,
query_vector={
"dense": query_embedding,
"sparse": sparse_query
},
limit=top_k,
query_filter=self._build_filter(filters) if filters else None,
search_params={
"hnsw_ef": 128,
"exact": False
},
score_threshold=0.5
)
return [
RetrievalResult(
document=Document(
id=r.id,
content=r.payload["content"],
metadata={k: v for k, v in r.payload.items() if k != "content"}
),
score=r.score,
source="hybrid"
)
for r in results
]The vector store configuration above includes several production-oriented features. Quantization reduces storage requirements and speeds up search by compressing vectors, with minimal impact on accuracy. HNSW (Hierarchical Navigable Small World) indexing enables fast approximate nearest neighbor search. The hybrid store implementation supports both dense and sparse vectors, enabling combined semantic and keyword search in a single system.
Query Transformation and Multi-Query Retrieval
User queries often don't match the language used in documents. Query transformation techniques bridge this gap by generating multiple query variations that capture different aspects of the user's intent.
The vocabulary mismatch problem is one of the fundamental challenges in information retrieval. A user asking 'How do I fix a slow API?' might be looking for content that discusses 'API performance optimization,' 'reducing latency,' or 'response time improvements.' None of these phrases share keywords with the original query, yet they all represent relevant content. Query transformation addresses this by generating multiple query variations that are more likely to match relevant documents.
Several query transformation strategies have proven effective. Query rephrasing generates alternative wordings that capture the same intent. Query decomposition breaks complex questions into simpler sub-questions that can be answered individually. Hypothetical Document Embedding (HyDE) generates a hypothetical answer to the query and uses that to search—based on the insight that an answer is often more similar to source documents than the question itself. Step-back prompting generates more general queries to retrieve broader context before addressing specifics.
The following implementation demonstrates these query transformation strategies, including the powerful HyDE technique that can significantly improve retrieval for certain types of queries:
# Query Transformation Strategies
class QueryTransformer:
"""Transform queries for better retrieval"""
def __init__(self, llm_client):
self.llm = llm_client
async def transform(self, query: str) -> List[str]:
"""Generate multiple query variations"""
variations = [query] # Always include original
# 1. Generate alternative phrasings
variations.extend(await self._rephrase_query(query))
# 2. Decompose into sub-questions
variations.extend(await self._decompose_query(query))
# 3. Generate hypothetical answer (HyDE)
hyde_doc = await self._hypothetical_document(query)
variations.append(hyde_doc)
return list(set(variations)) # Deduplicate
async def _rephrase_query(self, query: str) -> List[str]:
"""Generate alternative phrasings"""
prompt = f"""
Generate 3 alternative phrasings of this search query.
Each should capture the same intent but use different words.
Query: {query}
Return only the 3 alternatives, one per line.
"""
response = await self.llm.complete(prompt)
return [line.strip() for line in response.text.split("\n") if line.strip()]
async def _decompose_query(self, query: str) -> List[str]:
"""Break complex queries into simpler sub-questions"""
prompt = f"""
If this query requires multiple pieces of information to answer,
break it into simpler sub-questions. If it's already simple,
return an empty list.
Query: {query}
Return sub-questions, one per line, or "NONE" if not needed.
"""
response = await self.llm.complete(prompt)
if "NONE" in response.text:
return []
return [line.strip() for line in response.text.split("\n") if line.strip()]
async def _hypothetical_document(self, query: str) -> str:
"""Generate a hypothetical document that would answer the query (HyDE)"""
prompt = f"""
Write a short passage that would perfectly answer this question.
Write as if you're excerpting from a document that contains the answer.
Question: {query}
Hypothetical passage:
"""
response = await self.llm.complete(prompt)
return response.text.strip()
class StepBackPrompting:
"""Use step-back prompting for complex queries"""
async def get_step_back_query(self, query: str) -> str:
"""Generate a more general query to retrieve broader context"""
prompt = f"""
Given a specific question, generate a more general "step-back" question
that would help provide context for answering the specific question.
Specific question: {query}
Step-back question:
"""
response = await self.llm.complete(prompt)
return response.text.strip()Query transformation adds latency and cost—each transformation requires an LLM call, and each variation requires a retrieval round. However, the improvement in retrieval quality often justifies these costs, particularly for complex or ambiguous queries. In production, you might apply different transformation strategies based on query complexity, using simple approaches for straightforward queries and more sophisticated techniques for complex ones.
Reranking for Precision
Initial retrieval is optimized for recall—getting all potentially relevant documents. Reranking then optimizes for precision, selecting the most relevant results for the specific query.
Embedding-based retrieval, while powerful, has limitations. Embeddings are computed independently for the query and each document, missing subtle interactions between them. A document might contain the answer to a question but in a way that doesn't produce high embedding similarity. Reranking addresses this by using more sophisticated models that consider the query and document together, providing more accurate relevance assessments.
Reranking is typically applied as a second stage after initial retrieval. The first stage retrieves a larger set of candidates (say, top 50-100) using fast vector search. The reranker then scores each candidate more carefully, selecting the top 5-10 for final use. This two-stage approach balances efficiency with accuracy—the expensive reranking is only applied to a small candidate set rather than the entire document collection.
Several reranking approaches are available, each with different tradeoffs. Dedicated reranker models like Cohere's reranker are optimized specifically for relevance scoring and offer the best quality-to-latency ratio. Cross-encoder models provide high accuracy but are slower. LLM-based reranking offers flexibility and can incorporate complex relevance criteria but is the most expensive. The following implementations demonstrate each approach:
# Reranking Strategies
from typing import List
import cohere
class CohereReranker:
"""Use Cohere's reranking model for relevance scoring"""
def __init__(self, api_key: str = None):
self.client = cohere.Client(api_key or os.getenv("COHERE_API_KEY"))
async def rerank(
self,
query: str,
results: List[RetrievalResult],
top_k: int = 5
) -> List[RetrievalResult]:
"""Rerank results using Cohere's reranker"""
if not results:
return []
documents = [r.document.content for r in results]
response = self.client.rerank(
model="rerank-v3.5",
query=query,
documents=documents,
top_n=top_k,
return_documents=False
)
reranked = []
for item in response.results:
original = results[item.index]
reranked.append(RetrievalResult(
document=original.document,
score=item.relevance_score,
source="reranked"
))
return reranked
class CrossEncoderReranker:
"""Use cross-encoder model for precise relevance scoring"""
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
results: List[RetrievalResult],
top_k: int = 5
) -> List[RetrievalResult]:
"""Rerank using cross-encoder"""
pairs = [(query, r.document.content) for r in results]
scores = self.model.predict(pairs)
# Sort by score
scored_results = list(zip(results, scores))
scored_results.sort(key=lambda x: x[1], reverse=True)
return [
RetrievalResult(
document=r.document,
score=float(s),
source="cross-encoder"
)
for r, s in scored_results[:top_k]
]
class LLMReranker:
"""Use LLM for nuanced relevance assessment"""
async def rerank(
self,
query: str,
results: List[RetrievalResult],
top_k: int = 5
) -> List[RetrievalResult]:
"""Use LLM to assess relevance"""
prompt = f"""
Given the query and documents below, rank the documents by relevance.
Return the indices of the most relevant documents in order.
Query: {query}
Documents:
{self._format_documents(results)}
Return a JSON array of document indices in order of relevance:
"""
response = await self.llm.complete(prompt, response_format="json")
indices = json.loads(response.text)
return [results[i] for i in indices[:top_k]]The choice of reranker depends on your requirements. For most production use cases, dedicated reranker models offer the best balance of quality and performance. Cross-encoders are worth considering when you need the highest accuracy and can tolerate slightly higher latency. LLM-based reranking makes sense when you need to incorporate complex relevance criteria that go beyond semantic similarity.
Advanced RAG Patterns
Beyond basic RAG, several advanced patterns can significantly improve performance for specific use cases.
Basic RAG—retrieve then generate—works well for straightforward questions but struggles with complex scenarios. What if the retrieved documents don't actually contain the answer? What if the question requires synthesizing information from multiple sources? What if different parts of the question need different types of information? Advanced RAG patterns address these challenges by adding intelligence to the retrieval and generation process.
These patterns represent the cutting edge of RAG research and are increasingly being adopted in production systems. Self-RAG adds self-reflection to decide when retrieval is needed and whether retrieved content is relevant. Corrective RAG (CRAG) includes fallback mechanisms when initial retrieval fails. Agentic RAG combines RAG with AI agent capabilities, using planning and tool use to gather information from multiple sources. Each pattern adds complexity but can dramatically improve performance for the right use cases.
1. Self-RAG: Self-Reflective Retrieval
Self-RAG introduces reflection tokens that allow the model to reason about whether it needs external information and whether retrieved documents are relevant. This self-awareness reduces unnecessary retrieval and prevents the model from being misled by irrelevant context. The pattern is particularly effective for questions that mix general knowledge with specific information needs.
class SelfRAG:
"""Self-reflective RAG that decides when to retrieve"""
async def query(self, question: str) -> str:
# Step 1: Decide if retrieval is needed
needs_retrieval = await self._assess_retrieval_need(question)
if not needs_retrieval:
# Answer directly from LLM knowledge
return await self.llm.complete(question)
# Step 2: Retrieve and assess relevance
results = await self.retriever.retrieve(question)
relevant_results = await self._filter_relevant(question, results)
if not relevant_results:
# No relevant context found, use LLM knowledge with caveat
return await self._answer_without_context(question)
# Step 3: Generate with context
response = await self._generate_with_context(question, relevant_results)
# Step 4: Self-critique and refine
critique = await self._critique_response(question, response)
if critique["needs_refinement"]:
response = await self._refine_response(
question, response, critique["issues"]
)
return response
async def _assess_retrieval_need(self, question: str) -> bool:
"""Determine if external knowledge is needed"""
prompt = f"""
Does this question require external/specific knowledge to answer accurately?
Or can it be answered from general knowledge?
Question: {question}
Respond with only: RETRIEVE or DIRECT
"""
response = await self.llm.complete(prompt)
return "RETRIEVE" in response.text2. Corrective RAG (CRAG)
class CorrectiveRAG:
"""RAG with corrective actions when retrieval fails"""
async def query(self, question: str) -> str:
results = await self.retriever.retrieve(question)
# Evaluate retrieval quality
evaluation = await self._evaluate_retrieval(question, results)
if evaluation["status"] == "correct":
# Good retrieval, proceed normally
return await self._generate(question, results)
elif evaluation["status"] == "ambiguous":
# Partial match, refine and combine
refined_results = await self._refine_retrieval(question, results)
return await self._generate(question, refined_results)
else: # "incorrect"
# Retrieval failed, use web search as backup
web_results = await self.web_search(question)
return await self._generate(question, web_results)
async def _evaluate_retrieval(
self,
question: str,
results: List[RetrievalResult]
) -> Dict:
"""Evaluate if retrieved documents can answer the question"""
prompt = f"""
Evaluate if these documents can answer the question.
Question: {question}
Documents:
{self._format_documents(results)}
Respond with JSON:
{{
"status": "correct" | "ambiguous" | "incorrect",
"confidence": 0.0-1.0,
"reasoning": "..."
}}
"""
response = await self.llm.complete(prompt, response_format="json")
return json.loads(response.text)3. Agentic RAG
class AgenticRAG:
"""RAG with agentic planning and tool use"""
def __init__(self):
self.tools = {
"search_docs": self._search_documents,
"search_web": self._search_web,
"query_database": self._query_database,
"calculate": self._calculate,
"summarize": self._summarize
}
async def query(self, question: str) -> str:
"""Use agent to plan and execute retrieval strategy"""
# Create execution plan
plan = await self._create_plan(question)
context = []
for step in plan["steps"]:
tool = step["tool"]
args = step["arguments"]
result = await self.tools[tool](**args)
context.append({
"step": step["description"],
"result": result
})
# Check if we have enough information
if await self._has_sufficient_context(question, context):
break
return await self._synthesize_answer(question, context)
async def _create_plan(self, question: str) -> Dict:
"""Create a retrieval plan"""
prompt = f"""
Create a plan to answer this question using available tools.
Question: {question}
Available tools:
- search_docs: Search internal documents
- search_web: Search the web for current information
- query_database: Query structured data
- calculate: Perform calculations
- summarize: Summarize long content
Return a JSON plan with steps:
{{
"steps": [
{{
"tool": "tool_name",
"arguments": {{...}},
"description": "Why this step"
}}
]
}}
"""
response = await self.llm.complete(prompt, response_format="json")
return json.loads(response.text)RAG Evaluation and Optimization
Measuring RAG performance requires evaluating both retrieval quality and generation quality. Here are the key metrics and how to measure them.
# RAG Evaluation Framework
from dataclasses import dataclass
from typing import List, Dict
import numpy as np
@dataclass
class EvaluationResult:
"""Results from RAG evaluation"""
retrieval_precision: float
retrieval_recall: float
retrieval_mrr: float
answer_relevance: float
answer_faithfulness: float
answer_completeness: float
latency_p50: float
latency_p95: float
class RAGEvaluator:
"""Comprehensive RAG evaluation"""
def __init__(self, llm_client):
self.llm = llm_client
async def evaluate(
self,
test_cases: List[Dict], # [{question, expected_docs, expected_answer}]
rag_pipeline: RAGPipeline
) -> EvaluationResult:
"""Run comprehensive evaluation"""
retrieval_metrics = []
generation_metrics = []
latencies = []
for case in test_cases:
start_time = time.time()
result = await rag_pipeline.query(case["question"])
latencies.append(time.time() - start_time)
# Evaluate retrieval
retrieved_ids = [s["id"] for s in result["sources"]]
expected_ids = case.get("expected_docs", [])
if expected_ids:
retrieval_metrics.append({
"precision": self._precision(retrieved_ids, expected_ids),
"recall": self._recall(retrieved_ids, expected_ids),
"mrr": self._mrr(retrieved_ids, expected_ids)
})
# Evaluate generation
gen_eval = await self._evaluate_generation(
question=case["question"],
answer=result["answer"],
context=[s["content"] for s in result["sources"]],
expected_answer=case.get("expected_answer")
)
generation_metrics.append(gen_eval)
return EvaluationResult(
retrieval_precision=np.mean([m["precision"] for m in retrieval_metrics]),
retrieval_recall=np.mean([m["recall"] for m in retrieval_metrics]),
retrieval_mrr=np.mean([m["mrr"] for m in retrieval_metrics]),
answer_relevance=np.mean([m["relevance"] for m in generation_metrics]),
answer_faithfulness=np.mean([m["faithfulness"] for m in generation_metrics]),
answer_completeness=np.mean([m["completeness"] for m in generation_metrics]),
latency_p50=np.percentile(latencies, 50),
latency_p95=np.percentile(latencies, 95)
)
async def _evaluate_generation(
self,
question: str,
answer: str,
context: List[str],
expected_answer: str = None
) -> Dict[str, float]:
"""Evaluate generation quality using LLM-as-judge"""
# Faithfulness: Is the answer supported by the context?
faithfulness_prompt = f"""
Given the context and answer, evaluate if the answer is
fully supported by the context (no hallucinations).
Context:
{chr(10).join(context)}
Answer: {answer}
Score from 0-1 where 1 means fully supported:
"""
# Relevance: Does the answer address the question?
relevance_prompt = f"""
Given the question and answer, evaluate how well the answer
addresses the question.
Question: {question}
Answer: {answer}
Score from 0-1 where 1 means perfectly relevant:
"""
# Completeness: Does the answer fully address the question?
completeness_prompt = f"""
Given the question and answer, evaluate if the answer is
complete or if important aspects are missing.
Question: {question}
Answer: {answer}
Score from 0-1 where 1 means fully complete:
"""
faithfulness = await self._get_score(faithfulness_prompt)
relevance = await self._get_score(relevance_prompt)
completeness = await self._get_score(completeness_prompt)
return {
"faithfulness": faithfulness,
"relevance": relevance,
"completeness": completeness
}RAG Optimization Checklist
✓ Use domain-specific embedding models when available
✓ Implement hybrid search (dense + sparse)
✓ Add reranking for precision
✓ Use query transformation for better recall
✓ Chunk documents appropriately for your use case
✓ Include metadata filtering for large corpora
✓ Implement caching for repeated queries
✓ Monitor and optimize latency bottlenecks
✓ Set up continuous evaluation pipeline
✓ Track hallucination rates in production
Conclusion
RAG has matured from a simple retrieval-and-generate pattern to a sophisticated system requiring careful optimization at every step. The key to success is understanding your specific use case—the nature of your documents, user queries, and accuracy requirements—and tuning each component accordingly.
In 2025, the most successful RAG implementations combine multiple techniques: hybrid search for comprehensive retrieval, reranking for precision, query transformation for better matching, and advanced patterns like Self-RAG and CRAG for reliability. With proper evaluation and continuous optimization, RAG systems can achieve the accuracy and reliability required for production enterprise applications.
Next Steps
Building production-ready RAG systems requires expertise in machine learning, information retrieval, and software engineering. At Jishu Labs, our AI engineering team has extensive experience designing and deploying RAG systems for enterprise clients across industries.
Contact us to discuss your RAG implementation needs, or explore our AI Development Services for comprehensive AI solutions.
About Michael Chen
Michael Chen is a Principal AI Engineer at Jishu Labs specializing in large-scale AI systems and natural language processing. He has architected RAG systems processing millions of documents for Fortune 500 companies and is passionate about making AI systems more accurate and trustworthy.