Advanced RAG Techniques

It is not enough to find the right answer. You must first learn to ask the right question.

RAG, Inquisitive AI Agent

1. Query Transformation

The user's raw query is often a poor match for the retrieval index. Queries may be vague, use different terminology than the source documents (a challenge rooted in text representation), or bundle multiple sub-questions into one. Query transformation techniques rewrite, expand, or decompose the original query to improve retrieval recall and precision.

1.1 HyDE: Hypothetical Document Embeddings

HyDE (Gao et al., 2022) takes a counterintuitive approach: instead of embedding the query directly, it first asks the LLM to generate a hypothetical answer to the query (a creative application of prompt engineering), then embeds that hypothetical answer and uses it for retrieval. The intuition is that a hypothetical answer, even if factually incorrect, will be more lexically and semantically similar to real documents that contain the actual answer than the short query itself.

# implement hyde_retrieve
# Key operations: retrieval pipeline, API interaction

from openai import OpenAI

client = OpenAI()

def hyde_retrieve(query, collection, k=5):
 """HyDE: Generate hypothetical answer, embed it, retrieve."""

 # Step 1: Generate a hypothetical document
 hypo_response = client.chat.completions.create(
 model="gpt-4o-mini",
 messages=[{
 "role": "system",
 "content": "Write a short passage that would answer the "
 "following question. Be specific and detailed."
 }, {
 "role": "user",
 "content": query
 }],
 temperature=0.7
 )
 hypothetical_doc = hypo_response.choices[0].message.content

 # Step 2: Retrieve using the hypothetical document
 results = collection.query(
 query_texts=[hypothetical_doc],
 n_results=k
 )

 return results["documents"][0], results["metadatas"][0]

1.2 Multi-Query Expansion

Multi-query expansion generates several rephrased versions of the original query, retrieves results for each variant, and merges the result sets. This approach captures different phrasings and perspectives that might match different documents in the corpus. Code Fragment 20.2.3 below puts this into practice.

# implement multi_query_retrieve
# Key operations: retrieval pipeline, API interaction

def multi_query_retrieve(query, collection, k=5, num_variants=3):
 """Generate multiple query variants and merge results."""

 # Generate query variants
 response = client.chat.completions.create(
 model="gpt-4o-mini",
 messages=[{
 "role": "system",
 "content": f"Generate {num_variants} alternative phrasings of "
 "the following search query. Return one per line."
 }, {
 "role": "user",
 "content": query
 }]
 )
 variants = response.choices[0].message.content.strip().split("\n")
 all_queries = [query] + variants

 # Retrieve for each variant
 seen_ids = set()
 merged_results = []

 for q in all_queries:
 results = collection.query(query_texts=[q], n_results=k)
 for doc, meta, doc_id in zip(
 results["documents"][0],
 results["metadatas"][0],
 results["ids"][0]
 ):
 if doc_id not in seen_ids:
 seen_ids.add(doc_id)
 merged_results.append({
 "document": doc,
 "metadata": meta
 })

 return merged_results[:k * 2] # Return expanded set

1.3 Step-Back Prompting

Step-back prompting (Zheng et al., 2023) generates a more abstract or general version of the query before retrieval. For example, the query "What was the GDP growth rate of Japan in Q3 2024?" might be stepped back to "What are the recent economic trends in Japan?" The broader query retrieves documents that provide necessary background context, which is then combined with results from the specific query. Figure 20.2.5 compares these three query transformation strategies.

2. Hybrid Retrieval: Dense + Sparse

Dense retrieval (embedding similarity) excels at semantic matching but can miss exact keyword matches. Sparse retrieval (BM25) excels at keyword matching but misses semantic relationships. Hybrid retrieval combines both signals, typically using Reciprocal Rank Fusion (RRF) to merge the ranked result lists.

2.1 BM25 for Sparse Retrieval

BM25 is a term-frequency scoring function that has been the backbone of search engines for decades. It assigns higher scores to documents containing query terms that are rare in the corpus (high IDF, or Inverse Document Frequency) and that appear frequently in the specific document (high TF, or Term Frequency), with saturation to prevent long documents from dominating. Code Fragment 20.2.4 below puts this into practice.

# Define HybridRetriever; implement __init__, retrieve
# Key operations: retrieval pipeline, vector search, evaluation logic

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
 """Combine dense (vector) and sparse (BM25) retrieval."""

 def __init__(self, documents, collection):
 self.documents = documents
 self.collection = collection # ChromaDB collection

 # Build BM25 index
 tokenized = [doc.lower().split() for doc in documents]
 self.bm25 = BM25Okapi(tokenized)

 def retrieve(self, query, k=5, alpha=0.5):
 """Hybrid retrieval with Reciprocal Rank Fusion.

 Args:
 alpha: Weight for dense results (1-alpha for sparse).
 """
 # Dense retrieval
 dense_results = self.collection.query(
 query_texts=[query], n_results=k * 2
 )
 dense_ids = dense_results["ids"][0]

 # Sparse retrieval (BM25)
 tokenized_query = query.lower().split()
 bm25_scores = self.bm25.get_scores(tokenized_query)
 sparse_top = np.argsort(bm25_scores)[::-1][:k * 2]

 # Reciprocal Rank Fusion
 rrf_scores = {}
 rrf_k = 60 # Standard RRF constant

 for rank, doc_id in enumerate(dense_ids):
 rrf_scores[doc_id] = rrf_scores.get(doc_id, 0)
 rrf_scores[doc_id] += alpha / (rrf_k + rank + 1)

 for rank, idx in enumerate(sparse_top):
 doc_id = f"doc_{idx}"
 rrf_scores[doc_id] = rrf_scores.get(doc_id, 0)
 rrf_scores[doc_id] += (1 - alpha) / (rrf_k + rank + 1)

 # Sort by fused score
 ranked = sorted(
 rrf_scores.items(),
 key=lambda x: x[1],
 reverse=True
 )
 return ranked[:k]

3. Re-Ranking with Cross-Encoders

Initial retrieval (whether dense, sparse, or hybrid) uses fast but approximate scoring. Re-ranking applies a more powerful but slower model to the initial candidate set. Cross-encoder models are particularly effective because they jointly encode the query and document together, enabling fine-grained interaction between query and passage tokens.

3.1 How Cross-Encoders Differ from Bi-Encoders

Bi-encoders (used for initial retrieval) encode the query and document independently, then compute similarity via dot product or cosine. This allows pre-computing document embeddings but limits interaction between query and document representations. Cross-encoders encode the query and document as a single concatenated input, enabling full token-level attention between them. This produces much more accurate relevance scores but requires running inference for every (query, document) pair, making it too slow for searching millions of documents. Figure 20.2.6 shows the architectural difference between bi-encoders and cross-encoders in a retrieval pipeline. Code Fragment 20.2.5 below puts this into practice.

3.2 Using Cohere Rerank

This snippet reranks an initial set of retrieved passages using the Cohere Rerank API.

# implement rerank_results
# Key operations: retrieval pipeline, API interaction

import cohere

co = cohere.ClientV2("YOUR_API_KEY")

def rerank_results(query, documents, top_n=5):
 """Re-rank retrieved documents using Cohere Rerank."""
 response = co.rerank(
 model="rerank-v3.5",
 query=query,
 documents=documents,
 top_n=top_n,
 return_documents=True
 )

 reranked = []
 for result in response.results:
 reranked.append({
 "text": result.document.text,
 "relevance_score": result.relevance_score,
 "original_index": result.index
 })

 return reranked

4. Contextual Retrieval

Standard chunking strips away the surrounding context that gives a chunk its meaning. A chunk reading "The company reported 15% growth" is ambiguous without knowing which company, which metric, and which time period. Contextual retrieval (Anthropic, 2024) prepends each chunk with a short contextual summary generated by an LLM, creating self-contained chunks that embed and retrieve much more accurately.

5. Self-Corrective RAG

Standard RAG blindly trusts the retrieval results and generates from whatever context is provided. Self-corrective RAG systems evaluate the quality of retrieved documents and the faithfulness of generated answers, triggering corrective actions when problems are detected.

5.1 CRAG: Corrective Retrieval-Augmented Generation

CRAG (Yan et al., 2024) adds a retrieval evaluator that classifies each retrieved document as "correct," "incorrect," or "ambiguous." If all documents are incorrect, the system falls back to web search. If documents are ambiguous, the system refines the query and re-retrieves. Only when documents are classified as correct does generation proceed normally. Figure 20.2.7 illustrates how CRAG branches into correction paths based on retrieval quality.

5.2 Self-RAG

Self-RAG (Asai et al., 2023) trains the LLM itself to generate special reflection tokens that assess whether retrieval is needed, whether retrieved passages are relevant, whether the generated response is supported by the evidence, and whether the response is useful. These self-assessments allow the model to adaptively decide when to retrieve, which passages to use, and when to regenerate.

6. Fusion Retrieval and Multi-Modal RAG

Fusion retrieval goes beyond combining dense and sparse signals. RAG Fusion (Raudaschl, 2023) generates multiple search queries, retrieves results for each, and applies RRF across all result sets. This approach captures diverse perspectives on the query and is particularly effective for complex, multi-faceted questions.

6.1 Multi-Modal RAG

Multi-modal RAG extends retrieval beyond text to include images, tables, charts, and diagrams. This is essential for domains where critical information is encoded visually, such as scientific papers (figures and plots), financial reports (tables and charts), or technical documentation (architecture diagrams). Vision-language models like GPT-4o and Claude can process both retrieved text and images in their context window.

7. Comparison of Advanced RAG Techniques

Measuring RAG Improvements with Ragas

After adding advanced retrieval techniques, you need to measure whether the pipeline actually improved. The ragas library (pip install ragas) provides RAG-specific metrics that evaluate both retrieval quality and generation faithfulness. See Section 29.3 for deeper coverage of RAG evaluation.

# pip install ragas
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset

eval_data = Dataset.from_dict({
 "question": ["What is the return policy for electronics?"],
 "answer": ["Electronics can be returned within 30 days..."],
 "contexts": [["Our return policy allows 30-day returns for electronics..."]],
 "ground_truth": ["Electronics: 30-day return window with receipt."],
})

results = evaluate(
 dataset=eval_data,
 metrics=[faithfulness, answer_relevancy, context_precision],
)
print(results) # {'faithfulness': 0.95, 'answer_relevancy': 0.88, ...}

pip install sentence-transformers openai numpy

from sentence_transformers import SentenceTransformer, CrossEncoder
import numpy as np
from openai import OpenAI

client = OpenAI()
bi_encoder = SentenceTransformer("all-MiniLM-L6-v2")

documents = [
 "Python supports procedural, object-oriented, and functional programming.",
 "The GIL prevents multiple threads from executing Python bytecode simultaneously.",
 "NumPy provides efficient array operations for scientific computing in Python.",
 "Pandas provides DataFrames for data manipulation and analysis.",
 "Scikit-learn offers tools for data mining and machine learning.",
 "PyTorch and TensorFlow are the dominant deep learning frameworks.",
 "Flask and Django are popular Python web frameworks.",
 "List comprehensions provide concise list creation from iterables.",
 "Virtual environments isolate project dependencies to avoid conflicts.",
 "asyncio enables asynchronous programming for concurrent I/O.",
 "Type hints improve code readability and enable static analysis with mypy.",
 "Python 3.12 introduced performance improvements via adaptive specialization.",
]

doc_embs = bi_encoder.encode(documents)
doc_norms = np.linalg.norm(doc_embs, axis=1)

def baseline_search(query, top_k=5):
 qe = bi_encoder.encode(query)
 scores = np.dot(doc_embs, qe) / (doc_norms * np.linalg.norm(qe))
 idx = np.argsort(scores)[::-1][:top_k]
 return [(documents[i], scores[i], i) for i in idx]

query = "What tools should I use for data science in Python?"
print("Baseline results:")
for doc, score, _ in baseline_search(query):
 print(f" [{score:.3f}] {doc[:70]}...")

def expand_query(query, n=3):
 """Generate alternative query phrasings using an LLM."""
 resp = client.chat.completions.create(
 model="gpt-4o-mini",
 messages=[{"role": "user",
 "content": f"Generate {n} alternative phrasings of this search query. "
 f"One per line, no numbering.\n\nQuery: {query}"}],
 temperature=0.7, max_tokens=200)
 expansions = [q.strip() for q in
 resp.choices[0].message.content.strip().split("\n") if q.strip()]
 return [query] + expansions[:n]

def expanded_search(query, top_k=5):
 """Search with all expanded queries, deduplicate by best score."""
 queries = expand_query(query)
 print(f" Expanded: {queries}")

 # TODO: Search with each query, keep best score per document
 best = {}
 for q in queries:
 for doc, score, idx in baseline_search(q, top_k=top_k):
 if idx not in best or score > best[idx][1]:
 best[idx] = (doc, score, idx)
 ranked = sorted(best.values(), key=lambda x: x[1], reverse=True)
 return ranked[:top_k]

print("\nExpanded results:")
for doc, score, _ in expanded_search(query):
 print(f" [{score:.3f}] {doc[:70]}...")

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

def rerank(query, candidates, top_k=3):
 """Rerank candidates using cross-encoder scores."""
 pairs = [(query, doc) for doc, _, _ in candidates]
 scores = reranker.predict(pairs)
 reranked = sorted(zip(scores, candidates), reverse=True)
 return [(doc, float(s), idx) for s, (doc, _, idx) in reranked[:top_k]]

# Compare before and after reranking
baseline = baseline_search(query, top_k=8)
print("Before reranking (top 5):")
for doc, score, _ in baseline[:5]:
 print(f" [{score:.3f}] {doc[:70]}...")

reranked = rerank(query, baseline, top_k=5)
print("\nAfter reranking (top 5):")
for doc, score, _ in reranked:
 print(f" [{score:.3f}] {doc[:70]}...")

eval_queries = {
 "Data science tools in Python": [2, 3, 4],
 "How to speed up Python": [1, 9, 11],
 "Web development in Python": [6],
 "Managing Python dependencies": [8],
}

def recall_at_k(retrieved, relevant, k):
 return len(set(retrieved[:k]) & set(relevant)) / len(relevant)

for query, relevant in eval_queries.items():
 b = baseline_search(query, 5)
 b_idx = [i for _, _, i in b]

 e = expanded_search(query, 8)
 e_idx = [i for _, _, i in e[:5]]

 r = rerank(query, e, 5)
 r_idx = [i for _, _, i in r]

 print(f"\nQuery: {query}")
 print(f" Baseline R@5: {recall_at_k(b_idx, relevant, 5):.2f}")
 print(f" Expanded R@5: {recall_at_k(e_idx, relevant, 5):.2f}")
 print(f" Reranked R@5: {recall_at_k(r_idx, relevant, 5):.2f}")

from sentence_transformers import SentenceTransformer, CrossEncoder
from openai import OpenAI
import numpy as np

client = OpenAI()
bi = SentenceTransformer("all-MiniLM-L6-v2")
ce = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

docs = [
 "Python supports procedural, OOP, and functional programming.",
 "The GIL prevents multi-threaded Python bytecode execution.",
 "NumPy provides efficient array operations for scientific computing.",
 "Pandas provides DataFrames for data manipulation.",
 "Scikit-learn offers ML and data mining tools.",
 "PyTorch and TensorFlow are deep learning frameworks.",
 "Flask and Django are Python web frameworks.",
 "List comprehensions create lists concisely.",
 "Virtual environments isolate dependencies.",
 "asyncio enables async programming for concurrent I/O.",
 "Type hints improve readability and enable mypy.",
 "Python 3.12 has performance improvements.",
]

embs = bi.encode(docs)
norms = np.linalg.norm(embs, axis=1)

def search(q, k=5):
 qe = bi.encode(q)
 s = np.dot(embs, qe)/(norms*np.linalg.norm(qe))
 idx = np.argsort(s)[::-1][:k]
 return [(docs[i],s[i],i) for i in idx]

def expand(q, n=3):
 r = client.chat.completions.create(model="gpt-4o-mini",
 messages=[{"role":"user","content":f"Generate {n} alternative phrasings:\n{q}"}],
 temperature=0.7, max_tokens=200)
 return [q]+[l.strip() for l in r.choices[0].message.content.strip().split("\n") if l.strip()][:n]

def expanded_search(q, k=5):
 best = {}
 for eq in expand(q):
 for d,s,i in search(eq, k):
 if i not in best or s>best[i][1]: best[i]=(d,s,i)
 return sorted(best.values(), key=lambda x:x[1], reverse=True)[:k]

def rerank(q, cands, k=3):
 pairs = [(q,d) for d,_,_ in cands]
 scores = ce.predict(pairs)
 return [(d,float(s),i) for s,(d,_,i) in sorted(zip(scores,cands), reverse=True)[:k]]

def recall(ret, rel, k):
 return len(set(ret[:k])&set(rel))/len(rel)

for q,rel in {"Data science tools":[2,3,4],"Speed up Python":[1,9,11],
 "Web dev in Python":[6],"Managing dependencies":[8]}.items():
 b=[i for _,_,i in search(q,5)]
 e=[i for _,_,i in expanded_search(q,8)[:5]]
 r=[i for _,_,i in rerank(q, expanded_search(q,8), 5)]
 print(f"{q}: base={recall(b,rel,5):.2f} exp={recall(e,rel,5):.2f} rr={recall(r,rel,5):.2f}")