Advanced RAG Techniques and Memory Concepts
Learning Objectives: By the end of this lesson, you will be able to:
- Describe and implement advanced RAG techniques such as query transformation, re-ranking, and multi-hop retrieval
- Explain the concept of persistent memory in RAG and its connection to external knowledge
- Design and build self-correcting RAG systems that reduce hallucinations
- Implement agentic RAG patterns for autonomous information retrieval
- Identify and address common challenges in advanced RAG architectures
1. Welcome and Introduction (5 minutes)
Icebreaker Activity
Have you ever needed to refine your search or ask follow-up questions to get the information you wanted? Think about the last time you used a search engine—did you get the perfect answer on your first query, or did you need to rephrase, add context, or break down your question into smaller parts? This natural human behavior of iterating on queries is exactly what advanced RAG techniques aim to replicate programmatically.
Lesson Overview
- Duration: Approximately 2.5 hours including all exercises (or 60–90 minutes when selected activities are assigned)
- Prerequisites: Familiarity with RAG basics, retrieval mechanisms, vector databases, and LLM generation pipelines
- Materials Needed: Python environment with transformers and sentence-transformers installed, RAG pipeline from previous lessons, access to vector database (ChromaDB or FAISS), sample documents for testing
2. Learning Objectives and Session Outline (5 minutes)
What We’ll Cover Today:
- Query transformation and expansion — Techniques to reformulate user queries for optimal retrieval
- Re-ranking strategies — Using cross-encoders and secondary models to improve relevance
- Iterative and multi-hop RAG — Handling complex queries that require multiple reasoning steps
- Self-correcting and self-reflecting RAG — Building systems that verify and refine their own outputs
- Agentic RAG patterns — Autonomous retrieval using frameworks like ReAct
- Persistent memory and external knowledge — Creating dynamic, updatable knowledge stores
- Hands-on exercises — Implementing these techniques in your own RAG system
Success Criteria:
By the end of this session, you should be able to implement at least two advanced RAG techniques, understand how they interconnect, and explain the architectural decisions behind memory-augmented LLM systems.
3. Foundation Knowledge (15-20 minutes)
Core Concepts
Query Transformation
Query transformation is a family of techniques that reformulate or expand user queries before retrieval to improve the quality of retrieved documents. The fundamental insight is that users often phrase questions differently than how information is stored in documents. By bridging this semantic gap, we can dramatically improve retrieval performance.
Key Techniques:
-
HyDE (Hypothetical Document Embeddings): Instead of embedding the query directly, we first generate a hypothetical answer to the query using the LLM, then embed that hypothetical answer. This works because the hypothetical answer is semantically closer to actual documents in the knowledge base.
-
Query Expansion: Adding related terms, synonyms, or paraphrases to the original query to cast a wider retrieval net. This helps capture documents that might use different terminology.
-
Query Decomposition: Breaking complex queries into simpler sub-queries that can be answered independently and then combined.
-
Step-Back Prompting: Generating a more abstract, higher-level question from the original query to retrieve broader context first.
Re-ranking
Re-ranking is a two-stage retrieval approach where an initial fast retriever (typically bi-encoder based) retrieves a larger candidate set, which is then refined by a more accurate but slower re-ranker (typically cross-encoder based).
Why Re-ranking Works:
- Bi-encoders (used in initial retrieval) encode queries and documents independently, enabling fast similarity search but missing fine-grained interactions.
- Cross-encoders process query-document pairs together, allowing rich attention-based interactions that capture nuanced relevance signals.
The trade-off is computational cost—cross-encoders are too slow to run over millions of documents, but excel at scoring a small candidate set.
Iterative/Multi-hop RAG
Complex questions often cannot be answered with a single retrieval step. Multi-hop RAG decomposes questions and performs sequential retrievals, where each step builds on information from previous steps.
Example Scenario:
- User asks: “What was the GDP growth rate of the country where the 2024 Summer Olympics were held?”
- Step 1: Retrieve documents about 2024 Summer Olympics → Learn it was held in France
- Step 2: Retrieve documents about France’s GDP growth rate → Get the answer
Self-Correcting/Self-Reflecting RAG
Self-correcting RAG systems add a verification loop where the LLM evaluates its own generated answer against the retrieved context. If inconsistencies or unsupported claims are detected, the system either retrieves additional context or regenerates the answer.
Key Components:
- Faithfulness Check: Verify every claim in the answer is supported by the context
- Relevance Check: Ensure the answer actually addresses the user’s question
- Completeness Check: Confirm all parts of the question are answered
Agentic RAG
Agentic RAG elevates the LLM from a passive generator to an active agent that decides when and how to retrieve information. Using frameworks like ReAct (Reasoning + Acting), the agent can:
- Decide whether retrieval is necessary
- Formulate optimal queries
- Evaluate retrieved results
- Determine if additional retrieval is needed
- Synthesize a final answer
This approach is particularly powerful for open-ended questions that may require exploring multiple knowledge sources.
Persistent Memory via RAG
Traditional LLMs have “frozen” parametric knowledge that becomes stale over time. RAG provides a mechanism for persistent, updatable external memory through continuous ingestion into vector databases. This creates a dynamic knowledge store that can:
- Be updated in real-time without retraining
- Scale to arbitrary knowledge sizes
- Support access control and filtering
- Maintain audit trails of knowledge sources
Key Terminology
- Query Expansion: Adding related terms, synonyms, or paraphrases to improve retrieval recall
- Cross-Encoder: A model that takes a query-document pair as input and outputs a relevance score, enabling rich interaction between query and document tokens
- Bi-Encoder: A model that encodes queries and documents independently into vectors for efficient similarity search
- Multi-hop Retrieval: Sequential retrieval steps where each step’s query is informed by previous answers
- Agentic LLM: A model configured to autonomously plan, retrieve, reason, and act to accomplish tasks
- External Memory: Knowledge stored outside the LLM’s parameters, accessible through retrieval mechanisms
- Faithfulness: The degree to which generated text is supported by the provided context
4. Detailed Explanation and Examples (25-30 minutes)
Understanding Query Transformation
Query transformation addresses a fundamental challenge in retrieval systems: the vocabulary mismatch problem. Users and documents often express the same concepts using different words, making direct matching ineffective. Let’s explore the most powerful techniques.
Query Transformation Example: HyDE (Hypothetical Document Embeddings)
HyDE is a counterintuitive but highly effective technique. Instead of searching with the query embedding, we first ask the LLM to generate a hypothetical answer, then search using that answer’s embedding.
# HyDE: Generate a hypothetical answer, embed it, and use for retrieval
from transformers import pipeline
from sentence_transformers import SentenceTransformer
import chromadb
# Initialize models
generator = pipeline(
'text-generation', model='gpt2', max_new_tokens=100,
return_full_text=False # return only the hypothetical passage
)
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
# Original user query
query = "What are the health benefits of green tea?"
# Step 1: Generate a hypothetical answer (without retrieval)
hyde_prompt = f"""Answer the following question based on your knowledge.
Question: {query}
Answer:"""
hypothetical_answer = generator(hyde_prompt)[0]['generated_text'].strip()
print(f"Hypothetical Answer: {hypothetical_answer}")
# Step 2: Embed the hypothetical answer instead of the query
hyde_embedding = embedding_model.encode(
[hypothetical_answer],
normalize_embeddings=True
)
# Step 3: Use hyde_embedding for retrieval
# The hypothetical answer is semantically closer to actual documents
# than the original question-form query
# Connect to your vector database and search
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_collection("documents")
results = collection.query(
query_embeddings=hyde_embedding.tolist(),
n_results=5
)
print(f"Retrieved {len(results['documents'][0])} documents using HyDE")
Why HyDE Works:
- Questions and answers have different linguistic structures
- Documents typically contain declarative statements (answers), not questions
- The hypothetical answer bridges this structural gap
- Even an imperfect hypothetical answer shares more vocabulary with relevant documents
Query Expansion with Multiple Paraphrases
# Query expansion using paraphrasing and synonym generation
from transformers import pipeline
from sentence_transformers import SentenceTransformer
class QueryExpander:
def __init__(self):
self.paraphraser = pipeline(
"text2text-generation",
model="Vamsi/T5_Paraphrase_Paws"
)
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def expand_query(self, query: str, n_paraphrases: int = 3) -> list[str]:
"""Generate multiple paraphrases of the original query."""
paraphrases = []
for i in range(n_paraphrases):
result = self.paraphraser(
f"paraphrase: {query}",
max_length=100,
do_sample=True,
top_k=50,
temperature=0.7 + (i * 0.1) # Vary temperature for diversity
)
paraphrases.append(result[0]['generated_text'])
return [query] + paraphrases # Include original query
def retrieve_with_expansion(
self,
query: str,
collection,
n_results_per_query: int = 3
) -> list:
"""Retrieve documents using expanded queries and deduplicate."""
expanded_queries = self.expand_query(query)
all_results = []
seen_ids = set()
for q in expanded_queries:
embedding = self.embedding_model.encode([q], normalize_embeddings=True)
results = collection.query(
query_embeddings=embedding.tolist(),
n_results=n_results_per_query
)
# Deduplicate results
for doc_id, doc, score in zip(
results['ids'][0],
results['documents'][0],
results['distances'][0]
):
if doc_id not in seen_ids:
seen_ids.add(doc_id)
all_results.append({
'id': doc_id,
'document': doc,
'score': score,
'source_query': q
})
# Sort by score (lower distance = more similar for cosine)
all_results.sort(key=lambda x: x['score'])
return all_results
# Usage
expander = QueryExpander()
query = "What is the capital of France?"
expanded = expander.expand_query(query)
print("Expanded queries:", expanded)
# Output might include:
# - "What is the capital of France?"
# - "Which city serves as France's capital?"
# - "What is France's capital city?"
# - "Name the capital of France"
Re-ranking Example with Cross-Encoders
Re-ranking is one of the most impactful improvements you can add to a RAG pipeline. It dramatically improves precision with minimal implementation complexity.
# Use a cross-encoder to re-rank retrieved chunks
from sentence_transformers import CrossEncoder, SentenceTransformer
import numpy as np
class ReRankingRetriever:
def __init__(
self,
bi_encoder_model: str = 'all-MiniLM-L6-v2',
cross_encoder_model: str = 'cross-encoder/ms-marco-MiniLM-L-6-v2'
):
# Bi-encoder for initial fast retrieval
self.bi_encoder = SentenceTransformer(bi_encoder_model)
# Cross-encoder for accurate re-ranking
self.cross_encoder = CrossEncoder(cross_encoder_model)
def retrieve_and_rerank(
self,
query: str,
collection,
initial_k: int = 20, # Retrieve more initially
final_k: int = 5 # Return fewer, better results
) -> list[dict]:
"""
Two-stage retrieval:
1. Fast bi-encoder retrieval for candidates
2. Accurate cross-encoder re-ranking for final results
"""
# Stage 1: Initial retrieval with bi-encoder
query_embedding = self.bi_encoder.encode(
[query],
normalize_embeddings=True
)
initial_results = collection.query(
query_embeddings=query_embedding.tolist(),
n_results=initial_k
)
candidates = initial_results['documents'][0]
candidate_ids = initial_results['ids'][0]
if not candidates:
return []
# Stage 2: Re-rank with cross-encoder
# Create query-document pairs for cross-encoder scoring
pairs = [[query, doc] for doc in candidates]
# Cross-encoder scores (higher = more relevant)
scores = self.cross_encoder.predict(pairs)
# Sort by cross-encoder score (descending)
ranked_indices = np.argsort(scores)[::-1]
# Build final results
final_results = []
for idx in ranked_indices[:final_k]:
final_results.append({
'id': candidate_ids[idx],
'document': candidates[idx],
'bi_encoder_rank': idx + 1,
'cross_encoder_score': float(scores[idx])
})
return final_results
# Usage demonstration
retriever = ReRankingRetriever()
query = "How does photosynthesis work in plants?"
candidates = [
"Photosynthesis is the process by which plants convert sunlight into energy.",
"Plants need water and carbon dioxide for photosynthesis.",
"The weather today is sunny with a high of 75 degrees.",
"Chlorophyll in plant leaves absorbs light energy for photosynthesis.",
"Python is a popular programming language.",
"During photosynthesis, plants release oxygen as a byproduct.",
]
# Score all candidates
pairs = [[query, c] for c in candidates]
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
scores = cross_encoder.predict(pairs)
# Display results sorted by relevance
ranked = sorted(zip(scores, candidates), reverse=True)
print("Re-ranked results:")
for score, doc in ranked:
print(f" Score: {score:.4f} | {doc[:60]}...")
Multi-hop Retrieval Example
Multi-hop retrieval is essential for complex questions that require synthesizing information from multiple sources or reasoning chains.
from dataclasses import dataclass
from typing import Optional
@dataclass
class RetrievalStep:
query: str
retrieved_docs: list[str]
extracted_info: str
reasoning: str
class MultiHopRetriever:
def __init__(self, retriever, generator):
self.retriever = retriever # Your retrieval function
self.generator = generator # Your LLM generator
def decompose_question(self, complex_question: str) -> list[str]:
"""Break down a complex question into simpler sub-questions."""
decomposition_prompt = f"""Break down this complex question into simpler sub-questions that can be answered step by step.
Complex Question: {complex_question}
Output each sub-question on a new line, numbered:
1. [First sub-question]
2. [Second sub-question]
..."""
result = self.generator(decomposition_prompt)
# Parse the numbered sub-questions
sub_questions = []
for line in result.split('\n'):
line = line.strip()
if line and line[0].isdigit():
# Remove the number prefix
question = line.split('.', 1)[-1].strip()
if question:
sub_questions.append(question)
return sub_questions
def retrieve_and_extract(
self,
query: str,
previous_context: str = ""
) -> RetrievalStep:
"""Retrieve documents and extract relevant information."""
# Incorporate previous context into the query if available
enhanced_query = query
if previous_context:
enhanced_query = f"{query} Context: {previous_context}"
# Retrieve relevant documents
docs = self.retriever(enhanced_query, n_results=3)
# Extract key information from retrieved docs
extraction_prompt = f"""Based on the following documents, extract the key information needed to answer: "{query}"
Documents:
{chr(10).join(docs)}
Previous context: {previous_context if previous_context else 'None'}
Extracted information:"""
extracted = self.generator(extraction_prompt)
return RetrievalStep(
query=query,
retrieved_docs=docs,
extracted_info=extracted,
reasoning=f"Retrieved {len(docs)} docs for: {query}"
)
def multi_hop_retrieve(
self,
question: str,
max_hops: int = 3
) -> tuple[str, list[RetrievalStep]]:
"""Perform multi-hop retrieval for a complex question."""
# Step 1: Decompose the question
sub_questions = self.decompose_question(question)
# Step 2: Iteratively retrieve for each sub-question
steps = []
accumulated_context = ""
for i, sub_q in enumerate(sub_questions[:max_hops]):
step = self.retrieve_and_extract(sub_q, accumulated_context)
steps.append(step)
accumulated_context += f"\n{step.extracted_info}"
# Step 3: Generate final answer using all accumulated context
final_prompt = f"""Answer the following question using the provided context.
Question: {question}
Context gathered through multi-hop retrieval:
{accumulated_context}
Answer:"""
final_answer = self.generator(final_prompt)
return final_answer, steps
# Example usage for multi-hop question
# Question: "What is the population of the country where the Eiffel Tower is located?"
# Step 1: "Where is the Eiffel Tower located?" → France
# Step 2: "What is the population of France?" → ~67 million
Self-Reflecting RAG Example
Self-reflection adds a verification loop that dramatically reduces hallucinations by ensuring every generated claim is grounded in the retrieved context.
class SelfReflectingRAG:
def __init__(self, retriever, generator):
self.retriever = retriever
self.generator = generator
def generate_with_reflection(
self,
question: str,
max_iterations: int = 3
) -> dict:
"""Generate an answer with self-reflection and correction."""
# Initial retrieval
context_docs = self.retriever(question, n_results=5)
context = "\n\n".join(context_docs)
# Initial generation
generation_prompt = f"""Answer the question based ONLY on the provided context.
If the context doesn't contain enough information, say so.
Context:
{context}
Question: {question}
Answer:"""
answer = self.generator(generation_prompt)
# Self-reflection loop
for iteration in range(max_iterations):
# Check if answer is faithful to context
reflection_prompt = f"""Analyze whether the following answer is fully supported by the context.
Context:
{context}
Answer: {answer}
For each claim in the answer, determine if it is:
1. SUPPORTED - directly stated or clearly implied by the context
2. NOT SUPPORTED - not found in or contradicts the context
3. PARTIALLY SUPPORTED - some aspects supported, others not
List each claim and its status, then provide an overall assessment.
If there are unsupported claims, suggest how to correct them.
Analysis:"""
reflection = self.generator(reflection_prompt)
# Check if reflection indicates problems
if "NOT SUPPORTED" not in reflection.upper() and \
"PARTIALLY SUPPORTED" not in reflection.upper():
# Answer is faithful, we're done
return {
'answer': answer,
'context': context,
'iterations': iteration + 1,
'final_reflection': reflection,
'status': 'verified'
}
# Regenerate with reflection feedback
correction_prompt = f"""The previous answer had some unsupported claims.
Context:
{context}
Previous Answer: {answer}
Reflection showing issues: {reflection}
Generate a corrected answer that ONLY includes information supported by the context:"""
answer = self.generator(correction_prompt)
return {
'answer': answer,
'context': context,
'iterations': max_iterations,
'status': 'max_iterations_reached'
}
# Usage
rag = SelfReflectingRAG(retriever_func, generator_func)
result = rag.generate_with_reflection(
"What are the side effects of aspirin?"
)
print(f"Answer (verified after {result['iterations']} iterations):")
print(result['answer'])
Common Misconceptions
-
Misconception: More retrieved documents always lead to better answers. Reality: Too many documents can introduce noise and exceed context windows. Quality over quantity is crucial—re-ranking helps select the most relevant subset.
-
Misconception: HyDE requires a perfect hypothetical answer to work. Reality: HyDE works even with imperfect or partially correct hypothetical answers because they still capture relevant vocabulary and concepts.
-
Misconception: Multi-hop retrieval is always better than single-hop. Reality: Multi-hop adds latency and complexity. Use it selectively for questions that genuinely require multiple reasoning steps.
Debugging Demonstration
Show a common error: query drift in multi-hop retrieval
# Problem: Sub-questions drift away from the original intent
# Original: "What is the GDP of the country hosting the 2024 Olympics?"
# Sub-question 1: "Where are the 2024 Olympics?" → France
# Sub-question 2 (DRIFT): "What is interesting about France?" # Too broad!
# Solution: Keep sub-questions grounded in the original query
class GroundedDecomposer:
def decompose_with_grounding(self, question: str) -> list[str]:
prompt = f"""Decompose this question into sub-questions.
IMPORTANT: Each sub-question must directly help answer the original question.
Keep sub-questions specific and focused on the original goal.
Original Question: {question}
Sub-questions (each must help answer the original):"""
# The explicit grounding instruction prevents drift
return self.generator(prompt)
5. Live Coding Demonstration: Agentic RAG with ReAct (20-25 minutes)
Agentic RAG represents the cutting edge of RAG architectures, where the LLM autonomously decides when and how to retrieve information. Let’s implement a simplified ReAct-style agent.
Example 1: ReAct Agent for RAG
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Optional
import re
class ActionType(Enum):
SEARCH = "search"
LOOKUP = "lookup"
FINISH = "finish"
@dataclass
class AgentAction:
action_type: ActionType
action_input: str
thought: str
class ReActRAGAgent:
"""
A ReAct-style agent that reasons and acts to answer questions.
ReAct Pattern:
Thought: [reasoning about what to do]
Action: [search/lookup/finish]
Action Input: [query or final answer]
Observation: [result from action]
... (repeat until finish)
"""
def __init__(
self,
retriever: Callable,
generator: Callable,
max_steps: int = 5
):
self.retriever = retriever
self.generator = generator
self.max_steps = max_steps
self.scratchpad = [] # History of thoughts and actions
def _build_prompt(self, question: str) -> str:
"""Build the prompt including scratchpad history."""
base_prompt = f"""Answer the following question by reasoning step by step.
You have access to the following actions:
- search[query]: Search the knowledge base for relevant documents
- lookup[term]: Look up a specific term in the current documents
- finish[answer]: Provide your final answer
Use this format:
Thought: [your reasoning about what to do next]
Action: [one of: search, lookup, finish]
Action Input: [the input for your action]
Question: {question}
"""
# Add scratchpad history
for entry in self.scratchpad:
base_prompt += entry + "\n"
return base_prompt
def _parse_action(self, response: str) -> Optional[AgentAction]:
"""Parse the LLM response to extract action."""
# Extract thought
thought_match = re.search(r'Thought:\s*(.+?)(?=Action:|$)', response, re.DOTALL)
thought = thought_match.group(1).strip() if thought_match else ""
# Extract action
action_match = re.search(r'Action:\s*(\w+)', response)
if not action_match:
return None
action_name = action_match.group(1).lower()
# Extract action input
input_match = re.search(r'Action Input:\s*(.+?)(?=\n|$)', response, re.DOTALL)
action_input = input_match.group(1).strip() if input_match else ""
# Map to ActionType
action_map = {
'search': ActionType.SEARCH,
'lookup': ActionType.LOOKUP,
'finish': ActionType.FINISH
}
action_type = action_map.get(action_name)
if not action_type:
return None
return AgentAction(
action_type=action_type,
action_input=action_input,
thought=thought
)
def _execute_action(self, action: AgentAction) -> str:
"""Execute the parsed action and return observation."""
if action.action_type == ActionType.SEARCH:
docs = self.retriever(action.action_input, n_results=3)
return f"Found {len(docs)} relevant documents:\n" + "\n---\n".join(docs[:3])
elif action.action_type == ActionType.LOOKUP:
# Simple keyword lookup in recent retrieved docs
recent_docs = self._get_recent_docs()
matches = [d for d in recent_docs if action.action_input.lower() in d.lower()]
if matches:
return f"Found mentions of '{action.action_input}':\n" + matches[0][:500]
return f"No specific mention of '{action.action_input}' found."
elif action.action_type == ActionType.FINISH:
return action.action_input # Final answer
return "Unknown action"
def _get_recent_docs(self) -> list[str]:
"""Extract recently retrieved documents from scratchpad."""
docs = []
for entry in self.scratchpad:
if "Found" in entry and "relevant documents" in entry:
docs.append(entry)
return docs
def run(self, question: str) -> dict:
"""Run the ReAct loop to answer a question."""
self.scratchpad = []
for step in range(self.max_steps):
# Generate next thought and action
prompt = self._build_prompt(question)
response = self.generator(prompt)
# Parse the action
action = self._parse_action(response)
if not action:
# Fallback: force a finish action
self.scratchpad.append(f"Thought: Unable to parse action, finishing.\nAction: finish\nAction Input: I could not find a clear answer.")
break
# Record thought and action
self.scratchpad.append(f"Thought: {action.thought}")
self.scratchpad.append(f"Action: {action.action_type.value}")
self.scratchpad.append(f"Action Input: {action.action_input}")
# Execute action
observation = self._execute_action(action)
# Check if finished
if action.action_type == ActionType.FINISH:
return {
'answer': action.action_input,
'steps': step + 1,
'scratchpad': self.scratchpad.copy(),
'status': 'completed'
}
# Record observation and continue
self.scratchpad.append(f"Observation: {observation}")
return {
'answer': "Max steps reached without conclusive answer",
'steps': self.max_steps,
'scratchpad': self.scratchpad.copy(),
'status': 'max_steps_exceeded'
}
# Usage example
agent = ReActRAGAgent(retriever_func, generator_func)
result = agent.run("What is the relationship between photosynthesis and cellular respiration?")
print(f"Answer found in {result['steps']} steps:")
print(result['answer'])
print("\n--- Reasoning trace ---")
for entry in result['scratchpad']:
print(entry)
Example 2: Persistent Memory System
Building a knowledge store that persists and updates over time is essential for production RAG systems.
import chromadb
from datetime import datetime
from typing import Optional
import hashlib
class PersistentMemoryRAG:
"""
A RAG system with persistent, updatable memory.
Features:
- Automatic document ingestion with deduplication
- Timestamp tracking for freshness
- Source attribution for audit trails
- Incremental updates without full reindexing
"""
def __init__(
self,
persist_directory: str = "./memory_store",
collection_name: str = "knowledge_base"
):
self.client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"}
)
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def _compute_content_hash(self, content: str) -> str:
"""Compute hash for deduplication."""
return hashlib.md5(content.encode()).hexdigest()
def _chunk_document(
self,
document: str,
chunk_size: int = 500,
overlap: int = 50
) -> list[str]:
"""Split document into overlapping chunks."""
chunks = []
start = 0
while start < len(document):
end = start + chunk_size
chunk = document[start:end]
# Try to break at sentence boundary
if end < len(document):
last_period = chunk.rfind('.')
if last_period > chunk_size * 0.5:
chunk = chunk[:last_period + 1]
end = start + last_period + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if c] # Remove empty chunks
def ingest_document(
self,
content: str,
source: str,
metadata: Optional[dict] = None
) -> dict:
"""
Ingest a new document into the knowledge base.
Handles chunking, embedding, and deduplication.
"""
chunks = self._chunk_document(content)
added_count = 0
duplicate_count = 0
for i, chunk in enumerate(chunks):
content_hash = self._compute_content_hash(chunk)
doc_id = f"{source}_{content_hash}"
# Check if already exists (deduplication)
existing = self.collection.get(ids=[doc_id])
if existing['ids']:
duplicate_count += 1
continue
# Generate embedding
embedding = self.embedding_model.encode(
[chunk],
normalize_embeddings=True
).tolist()[0]
# Prepare metadata
doc_metadata = {
'source': source,
'chunk_index': i,
'total_chunks': len(chunks),
'ingested_at': datetime.now().isoformat(),
'content_hash': content_hash
}
if metadata:
doc_metadata.update(metadata)
# Add to collection
self.collection.add(
ids=[doc_id],
embeddings=[embedding],
documents=[chunk],
metadatas=[doc_metadata]
)
added_count += 1
return {
'status': 'success',
'chunks_added': added_count,
'duplicates_skipped': duplicate_count,
'total_chunks': len(chunks),
'source': source
}
def update_knowledge(
self,
source: str,
new_content: str,
metadata: Optional[dict] = None
) -> dict:
"""
Update knowledge from a source.
Removes old content from that source and adds new content.
"""
# Delete existing documents from this source
existing = self.collection.get(
where={"source": source}
)
if existing['ids']:
self.collection.delete(ids=existing['ids'])
# Ingest new content
result = self.ingest_document(new_content, source, metadata)
result['previous_chunks_removed'] = len(existing['ids'])
return result
def query(
self,
question: str,
n_results: int = 5,
recency_weight: float = 0.1
) -> list[dict]:
"""
Query the knowledge base with optional recency weighting.
"""
query_embedding = self.embedding_model.encode(
[question],
normalize_embeddings=True
).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=n_results * 2, # Retrieve more for reranking
include=['documents', 'metadatas', 'distances']
)
# Build result list with recency adjustment
processed_results = []
for i in range(len(results['ids'][0])):
result = {
'id': results['ids'][0][i],
'document': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'similarity_score': 1 - results['distances'][0][i], # Convert distance to similarity
}
# Optional: boost recent documents
ingested_at = result['metadata'].get('ingested_at', '')
if ingested_at:
# Simple recency boost (documents from today get small boost)
result['final_score'] = result['similarity_score'] # Can add recency logic here
processed_results.append(result)
# Sort by final score and return top n
processed_results.sort(key=lambda x: x.get('final_score', x['similarity_score']), reverse=True)
return processed_results[:n_results]
def get_statistics(self) -> dict:
"""Get statistics about the knowledge base."""
all_docs = self.collection.get(include=['metadatas'])
sources = {}
for meta in all_docs['metadatas']:
source = meta.get('source', 'unknown')
sources[source] = sources.get(source, 0) + 1
return {
'total_chunks': len(all_docs['ids']),
'sources': sources,
'unique_sources': len(sources)
}
# Usage example
memory = PersistentMemoryRAG(persist_directory="./my_knowledge_base")
# Ingest initial documents
result = memory.ingest_document(
content="Python is a high-level programming language...",
source="python_docs_v3.12",
metadata={"category": "programming", "language": "en"}
)
print(f"Ingested: {result}")
# Update when new version released
result = memory.update_knowledge(
source="python_docs_v3.12",
new_content="Python 3.12 introduces new features including...",
metadata={"category": "programming", "version": "3.12"}
)
print(f"Updated: {result}")
# Query the knowledge base
results = memory.query("How do I use type hints in Python?")
for r in results:
print(f"[{r['metadata']['source']}] {r['document'][:100]}...")
6. Hands-On Practice (30-40 minutes)
Exercise 1: Basic Practice — Query Expansion (10 minutes)
Objective: Implement query expansion to improve retrieval recall
Instructions:
- Take a user query and generate 2-3 paraphrases using different techniques
- Retrieve context for each query variant
- Combine and deduplicate results
Starter Code:
from sentence_transformers import SentenceTransformer
class SimpleQueryExpander:
def __init__(self):
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def expand_query_manually(self, query: str) -> list[str]:
"""
TODO: Implement manual query expansion
Generate variations of the query:
- Synonym substitution
- Rephrasing as a statement
- Adding context words
"""
expanded = [query]
# Your code here
return expanded
def retrieve_with_expansion(self, query: str, collection, n_results: int = 5):
"""
TODO: Implement retrieval with expanded queries
1. Expand the query
2. Retrieve for each variant
3. Combine and deduplicate results
"""
pass
# Test with this query
query = "What is the capital of France?"
expander = SimpleQueryExpander()
# Expected expansions might include:
# - "Which city is the capital of France?"
# - "France's capital city"
# - "The capital of France is"
Solution:
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Any
class SimpleQueryExpander:
def __init__(self):
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def expand_query_manually(self, query: str) -> list[str]:
"""Generate variations of the query."""
expanded = [query]
# Technique 1: Rephrase as different question types
if query.lower().startswith("what is"):
# Convert "What is X?" to declarative form
subject = query[8:].rstrip("?").strip()
expanded.append(f"The {subject} is")
expanded.append(f"Definition of {subject}")
# Technique 2: Add context-seeking variations
expanded.append(f"Tell me about {query.rstrip('?')}")
# Technique 3: Keyword extraction approach
# Remove question words and articles
keywords = query.lower()
for word in ['what', 'is', 'the', 'a', 'an', '?']:
keywords = keywords.replace(word, ' ')
keywords = ' '.join(keywords.split()) # Normalize whitespace
if keywords:
expanded.append(keywords)
return expanded
def retrieve_with_expansion(
self,
query: str,
collection,
n_results: int = 5
) -> list[dict]:
"""Retrieve with expanded queries and deduplicate."""
expanded_queries = self.expand_query_manually(query)
all_results = {} # Use dict for deduplication by ID
for q in expanded_queries:
embedding = self.embedding_model.encode(
[q],
normalize_embeddings=True
)
results = collection.query(
query_embeddings=embedding.tolist(),
n_results=n_results
)
# Add to results, keeping best score for duplicates
for doc_id, doc, distance in zip(
results['ids'][0],
results['documents'][0],
results['distances'][0]
):
if doc_id not in all_results or distance < all_results[doc_id]['distance']:
all_results[doc_id] = {
'id': doc_id,
'document': doc,
'distance': distance,
'matched_query': q
}
# Sort by distance and return top n
sorted_results = sorted(all_results.values(), key=lambda x: x['distance'])
return sorted_results[:n_results]
# Test
expander = SimpleQueryExpander()
query = "What is the capital of France?"
expanded = expander.expand_query_manually(query)
print("Expanded queries:", expanded)
# Output: ['What is the capital of France?', 'The capital of France is',
# 'Definition of capital of France', 'Tell me about What is the capital of France',
# 'capital france']
Exercise 2: Intermediate Challenge — Re-ranking Pipeline (15 minutes)
Objective: Add re-ranking to your RAG pipeline and compare results
Requirements:
- Retrieve top-20 chunks using bi-encoder
- Use a cross-encoder to re-rank
- Select top-5 after re-ranking
- Compare retrieval quality before and after
Starter Code:
from sentence_transformers import SentenceTransformer, CrossEncoder
class ReRankingPipeline:
def __init__(self):
self.bi_encoder = SentenceTransformer('all-MiniLM-L6-v2')
self.cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def retrieve_only(self, query: str, collection, k: int = 5):
"""Basic retrieval without re-ranking."""
# TODO: Implement basic retrieval
pass
def retrieve_and_rerank(self, query: str, collection, initial_k: int = 20, final_k: int = 5):
"""
TODO: Implement two-stage retrieval:
1. Retrieve initial_k candidates with bi-encoder
2. Re-rank with cross-encoder
3. Return final_k results
"""
pass
def compare_results(self, query: str, collection):
"""Compare results with and without re-ranking."""
basic = self.retrieve_only(query, collection, k=5)
reranked = self.retrieve_and_rerank(query, collection, initial_k=20, final_k=5)
print("=== Without Re-ranking ===")
for i, doc in enumerate(basic):
print(f"{i+1}. {doc[:100]}...")
print("\n=== With Re-ranking ===")
for i, doc in enumerate(reranked):
print(f"{i+1}. {doc[:100]}...")
# Test with a query where re-ranking makes a difference
query = "How do neural networks learn patterns in data?"
Hints:
- Cross-encoder takes
[[query, doc1], [query, doc2], ...]as input - Cross-encoder scores are higher for more relevant pairs
- Don’t forget to sort in descending order for cross-encoder scores
Exercise 3: Advanced Challenge — Self-Correcting RAG (15 minutes)
Objective: Implement a RAG system that verifies and corrects its own outputs
Challenge: Build a self-correcting RAG system that:
- Generates an initial answer from retrieved context
- Checks each claim in the answer against the context
- Regenerates if unsupported claims are found
Starter Template:
class SelfCorrectingRAG:
def __init__(self, retriever, generator):
self.retriever = retriever
self.generator = generator
def check_faithfulness(self, answer: str, context: str) -> dict:
"""
TODO: Check if answer is faithful to context
Return: {'is_faithful': bool, 'unsupported_claims': list}
"""
pass
def generate_with_correction(
self,
question: str,
max_corrections: int = 2
) -> dict:
"""
TODO: Implement the self-correction loop
1. Retrieve context
2. Generate answer
3. Check faithfulness
4. If not faithful, regenerate with feedback
5. Return final answer with metadata
"""
pass
# Test case
question = "What are the main causes of climate change?"
# Your RAG should only include causes mentioned in retrieved documents,
# and should not hallucinate additional causes
Extension Ideas:
- Add citation tracking (which sentence came from which document)
- Implement confidence scoring for each claim
- Build a feedback loop that retrieves additional context for weak claims
7. Common Pitfalls and Troubleshooting (10 minutes)
Frequent Mistakes
-
Problem: Query drift in multi-hop retrieval Cause: Sub-questions progressively lose connection to the original question as context accumulates Solution: Always include the original question in sub-question generation prompts; validate that each sub-question directly contributes to answering the original
# Bad: Generate sub-question only from previous answer prompt = f"What follow-up question should I ask about {previous_answer}?" # Good: Keep original question in scope prompt = f"""Original question: {original_question} Previous findings: {previous_answer} What sub-question will help answer the original question?""" -
Problem: Increased latency from re-ranking or multi-hop approaches Cause: Multiple model calls (cross-encoder scoring, iterative generation) add significant latency Solution:
- Cache cross-encoder scores for frequently queried documents
- Use async/parallel processing where possible
- Set reasonable limits on retrieval iterations
- Consider lightweight re-ranking models for production
# Optimize with batching and caching import functools @functools.lru_cache(maxsize=10000) def cached_cross_encode(query: str, doc_hash: str, doc: str) -> float: return cross_encoder.predict([[query, doc]])[0] -
Problem: Re-ranking doesn’t improve results Cause: Initial retrieval is already poor, or cross-encoder model is not suited to domain Solution: Ensure bi-encoder retrieves enough candidates; fine-tune cross-encoder on domain data if needed
-
Problem: Self-correction loops indefinitely Cause: LLM cannot generate faithful answers with available context, or reflection criteria are too strict Solution: Set maximum iteration limits; provide fallback responses; verify context actually contains needed information
# Always have an escape hatch for iteration in range(max_iterations): # ... correction logic ... if iteration == max_iterations - 1: return { 'answer': "Based on available information: " + best_answer_so_far, 'status': 'partial_confidence' } -
Problem: HyDE generates misleading hypothetical answers Cause: LLM confidently generates incorrect hypotheticals that retrieve wrong documents Solution: Generate multiple hypotheticals and combine embeddings; use ensemble with original query embedding
# Ensemble approach query_emb = model.encode([query]) hyde_emb = model.encode([hypothetical_answer]) combined_emb = 0.7 * query_emb + 0.3 * hyde_emb # Weight toward original query
Debugging Tips
- Log everything: Record all queries, retrievals, and generation steps for diagnosis
- Visualize retrieval chains: Create diagrams showing multi-hop paths and where they diverge
- Compare scores: Always compare bi-encoder vs. cross-encoder rankings to understand re-ranking impact
- Test edge cases: Try simple queries, complex queries, and queries with no good matches
- Monitor latency: Track time spent in each component to identify bottlenecks
- Use synthetic evaluation: Create test cases with known answers to measure accuracy
# Debugging utility
from datetime import datetime
class RAGDebugger:
def __init__(self):
self.logs = []
def log_step(self, step_name: str, data: dict):
self.logs.append({
'step': step_name,
'timestamp': datetime.now().isoformat(),
'data': data
})
def print_trace(self):
for log in self.logs:
print(f"[{log['step']}] {log['data']}")
8. Best Practices and Professional Tips (10 minutes)
Industry Standards
- Use query expansion and re-ranking for high-precision applications where accuracy matters more than latency
- Monitor and limit retrieval/generation latency — set SLAs for each component and track percentiles
- Treat the vector database as a dynamic, updatable memory — implement versioning and rollback capabilities
- Implement proper chunking strategies — balance between context completeness and retrieval precision
- Add observability from day one — log queries, retrieved documents, generated answers, and user feedback
Architecture Recommendations
# Production RAG architecture checklist
production_checklist = {
'retrieval': [
'Bi-encoder for initial candidate retrieval',
'Cross-encoder re-ranking for top candidates',
'Query expansion for recall improvement',
'Hybrid search (semantic + keyword) for robustness'
],
'generation': [
'Self-reflection for hallucination reduction',
'Citation/attribution in responses',
'Confidence scoring for answer quality',
'Graceful degradation when context is insufficient'
],
'infrastructure': [
'Async document ingestion pipeline',
'Incremental index updates',
'Caching layer for frequent queries',
'Rate limiting and cost controls'
],
'monitoring': [
'Retrieval quality metrics (MRR, NDCG)',
'Generation faithfulness scoring',
'Latency tracking per component',
'User feedback collection'
]
}
Next Steps in Your Learning Journey
- Explore agentic frameworks such as ReAct, AutoGPT, and LangGraph for complex autonomous workflows
- Study memory-augmented LLMs and research on lifelong learning systems
- Experiment with hybrid retrieval combining semantic search with BM25 or keyword matching
- Learn about evaluation frameworks like RAGAS, TruLens, and custom evaluation pipelines
- Investigate fine-tuning retrievers and generators on domain-specific data
- Explore multimodal RAG that handles images, tables, and structured data
HyDE and Memory Boundaries
For HyDE, ask the generator for a hypothetical passage and embed only the generated passage—not the instruction, prompt wrapper, or original generation prompt. Treat the passage as a retrieval query, not as trusted evidence. Persistent memory also needs an explicit owner and access policy, retention period, conflict-resolution rule, provenance, user inspection/correction, and deletion mechanism. Do not store secrets merely because they may improve retrieval.
9. Summary and Review (5 minutes)
Key Takeaways
- Query transformation (HyDE, expansion, decomposition) bridges the semantic gap between user queries and document content
- Re-ranking with cross-encoders dramatically improves precision by enabling rich query-document interaction
- Multi-hop retrieval unlocks answers to complex questions requiring multiple reasoning steps
- Self-correcting RAG reduces hallucinations through verification and regeneration loops
- Agentic RAG empowers LLMs to autonomously decide when and how to retrieve information
- Persistent memory transforms RAG from static retrieval into dynamic, updatable knowledge systems
Quick Review Questions
-
What is query transformation and why is it useful? Query transformation reformulates user queries to better match document content. It’s useful because users often phrase questions differently than how information is stored, and transformation bridges this gap.
-
How does re-ranking improve retrieval quality? Re-ranking uses cross-encoders that process query-document pairs together, enabling attention-based interactions that capture nuanced relevance signals missed by bi-encoders.
-
What is the role of memory in RAG systems? Memory (via vector databases) provides LLMs with external, updatable knowledge that can be refreshed without retraining, scaled independently, and filtered/controlled for access.
-
When should you use multi-hop retrieval vs. single-hop? Use multi-hop when questions genuinely require synthesizing information from multiple sources or reasoning chains. For straightforward factual queries, single-hop is more efficient.
-
How does self-correction reduce hallucinations? Self-correction prompts the LLM to verify each generated claim against the context, identifying unsupported statements and regenerating with explicit grounding instructions.
For Instructors
Teaching Tips
- Start with concrete examples: Show how each technique transforms a specific query or improves a specific retrieval before diving into code
- Use visual diagrams: Multi-hop and agentic RAG are much clearer with flowcharts showing the retrieval chains
- Encourage experimentation: Have students run the same query with and without each technique to see the difference
- Address common confusions: Students often conflate HyDE with query expansion—clarify that HyDE generates a hypothetical answer while expansion adds query variants
- Build incrementally: Start with basic RAG, add re-ranking, then query expansion, then multi-hop—don’t try to implement everything at once
Assessment Ideas
- Quiz: Present code snippets and ask students to identify which advanced RAG technique is being used
- Hands-on project: Extend a basic RAG system with at least two advanced techniques and measure improvement
- Peer review: Students analyze each other’s implementations and suggest improvements
- Comparative analysis: Write a report comparing retrieval quality before and after applying advanced techniques
Adaptation Notes
- For beginners: Focus primarily on query expansion and re-ranking—these provide the most improvement for the least complexity
- For intermediate learners: Introduce multi-hop and self-correcting RAG with guided exercises
- For advanced learners: Challenge them to implement full agentic RAG with custom action spaces
- Online delivery: Provide Google Colab notebooks with pre-configured vector databases and sample documents
- In-person sessions: Use live coding with audience participation to build each component step by step