Advanced RAG Patterns

Query Optimization

Problem: User Queries Are Messy

# Users don't write perfect queries:
"uh what was that thing about the vacation days again?"
"how many pto do we get lol"
"vacation policy ???"

→ Need to clean and optimize before retrieval!

Query Expansion

Add related terms to improve coverage

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def expand_query(query):
    prompt = f"""
    Generate 3 alternative phrasings of this query to improve search:
    
    Original: {query}
    
    Alternatives (one per line):
    """
    
    response = llm.invoke(prompt)
    alternatives = response.content.strip().split('\n')
    
    # Search with all versions
    all_results = []
    for alt_query in [query] + alternatives:
        results = vectorstore.similarity_search(alt_query, k=3)
        all_results.extend(results)
    
    # Deduplicate
    return deduplicate(all_results)

# Example:
query = "PTO policy"
# Expands to:
# - "paid time off policy"
# - "vacation leave guidelines"
# - "employee time off benefits"

Query Decomposition

Break complex queries into simple sub-queries

def decompose_and_answer(complex_query):
    # Step 1: Decompose
    decompose_prompt = f"""
    Break this into 2-3 simple sub-questions:
    Question: {complex_query}
    
    Sub-questions:
    1.
    2.
    3.
    """
    
    sub_questions = llm.invoke(decompose_prompt).content.split('\n')
    
    # Step 2: Answer each
    sub_answers = []
    for sub_q in sub_questions:
        if sub_q.strip():
            docs = vectorstore.similarity_search(sub_q, k=3)
            answer = generate_answer(sub_q, docs)
            sub_answers.append(answer)
    
    # Step 3: Synthesize
    final_prompt = f"""
    Question: {complex_query}
    
    Sub-answers:
    {'\n'.join(sub_answers)}
    
    Synthesize a complete answer:
    """
    
    return llm.invoke(final_prompt).content

# Example:
query = "Compare vacation policies for engineers vs sales, and who gets more?"
# Sub-questions:
# 1. What is the vacation policy for engineers?
# 2. What is the vacation policy for sales?
# 3. Which role receives more vacation days?

Query Rewriting

Fix vague/unclear queries

def rewrite_query(vague_query, conversation_history=None):
    context = ""
    if conversation_history:
        context = f"Previous conversation:\n{conversation_history}\n\n"
    
    prompt = f"""
    {context}
    Rewrite this vague query into a clear, specific search query:
    
    Vague: {vague_query}
    Clear:
    """
    
    return llm.invoke(prompt).content

# Examples:
rewrite_query("that thing about days off")
# → "What is the vacation day policy?"

rewrite_query("more info on that")
# → (needs conversation history)

# With context:
history = "User: What's the 401k match?\nBot: We match 6%.\nUser: more info on that"
rewrite_query("more info on that", history)
# → "What are additional details about the 401k matching policy?"

Multi-Hop Reasoning

Self-Ask Pattern

LLM asks itself follow-up questions

from langchain.chains import SequentialChain

def self_ask_rag(question):
    max_hops = 3
    current_question = question
    context = []
    
    for hop in range(max_hops):
        # Ask: "Do I need more info to answer this?"
        check_prompt = f"""
        Question: {current_question}
        Current context: {context}
        
        Can you answer this with the current context? If not, what additional question should be asked?
        
        Answer with:
        - "ANSWER: [your answer]" if you can answer
        - "ASK: [follow-up question]" if you need more info
        """
        
        response = llm.invoke(check_prompt).content
        
        if response.startswith("ANSWER:"):
            return response[7:].strip()
        
        elif response.startswith("ASK:"):
            follow_up = response[4:].strip()
            
            # Retrieve for follow-up
            docs = vectorstore.similarity_search(follow_up, k=3)
            context.append({
                "question": follow_up,
                "docs": [d.page_content for d in docs]
            })
            
            current_question = question  # Keep original question in focus
    
    # Final answer with all context
    return generate_final_answer(question, context)

# Example:
question = "How much vacation does a senior engineer with 5 years get?"

# Hop 1: ASK: "What is the base vacation policy?"
# Retrieve → "15 days base"

# Hop 2: ASK: "Are there seniority bonuses?"
# Retrieve → "2 extra days per 3 years"

# Hop 3: ASK: "Are there role-specific bonuses?"
# Retrieve → "Engineers get +3 days"

# ANSWER: "Senior engineer with 5 years: 15 + 2 + 3 = 20 days"

Agentic RAG

RAG with Tools

Give RAG system access to tools (calculator, web search, etc.)

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# Define tools
def retrieval_tool(query: str) -> str:
    """Search company knowledge base"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n".join([d.page_content for d in docs])

def calculator_tool(expression: str) -> str:
    """Calculate mathematical expressions"""
    try:
        return str(eval(expression))
    except:
        return "Invalid expression"

tools = [
    Tool(
        name="Knowledge Base",
        func=retrieval_tool,
        description="Search company documents for policies, procedures, and information"
    ),
    Tool(
        name="Calculator",
        func=calculator_tool,
        description="Perform mathematical calculations. Input should be a valid Python expression."
    )
]

# Create agent
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Query requiring both retrieval AND calculation
result = agent_executor.invoke({
    "input": "If I started 3 years ago and the base is 15 days with 2 extra per 3 years, how many vacation days do I have?"
})

# Agent thinks:
# 1. Use Knowledge Base tool → Find "15 days base, 2 extra per 3 years"
# 2. Use Calculator tool → Compute "15 + 2 = 17"
# 3. Answer: "17 vacation days"

Self-Correction

Agent checks and fixes its own mistakes

def rag_with_self_correction(query, max_attempts=2):
    for attempt in range(max_attempts):
        # Generate answer
        docs = vectorstore.similarity_search(query, k=5)
        context = "\n".join([d.page_content for d in docs])
        
        answer_prompt = f"""
        Context: {context}
        Question: {query}
        Answer:
        """
        answer = llm.invoke(answer_prompt).content
        
        # Self-critique
        critique_prompt = f"""
        Question: {query}
        Answer: {answer}
        Context: {context}
        
        Is this answer:
        1. Accurate based on the context?
        2. Complete (addresses all parts of question)?
        3. Supported by evidence?
        
        If NO to any, explain what's wrong. If YES to all, say "CORRECT".
        
        Assessment:
        """
        
        critique = llm.invoke(critique_prompt).content
        
        if "CORRECT" in critique:
            return answer
        
        # If not correct, use critique to improve
        if attempt < max_attempts - 1:
            # Revise query based on critique
            revised_query = f"{query}\n\nNote: Previous attempt was wrong because: {critique}"
            query = revised_query
    
    return answer  # Return best attempt

# Example:
query = "What's the parental leave policy?"

# Attempt 1: Returns info about vacation (wrong retrieval)
# Critique: "Answer discusses vacation, not parental leave"
# Attempt 2: Retrieves correct docs, returns parental leave policy ✓

Evaluation & Quality Metrics

Retrieval Quality

def evaluate_retrieval(test_queries_with_expected_docs):
    """
    Metrics:
    - Precision@K: % of retrieved docs that are relevant
    - Recall@K: % of relevant docs that were retrieved
    - MRR (Mean Reciprocal Rank): 1/rank of first relevant doc
    """
    
    precisions, recalls, reciprocal_ranks = [], [], []
    
    for query, expected_doc_ids in test_queries_with_expected_docs:
        # Retrieve
        retrieved = vectorstore.similarity_search(query, k=5)
        retrieved_ids = [d.metadata['chunk_id'] for d in retrieved]
        
        # Calculate metrics
        relevant_retrieved = set(retrieved_ids) & set(expected_doc_ids)
        
        precision = len(relevant_retrieved) / len(retrieved_ids)
        recall = len(relevant_retrieved) / len(expected_doc_ids)
        
        # Find rank of first relevant doc
        for rank, doc_id in enumerate(retrieved_ids, 1):
            if doc_id in expected_doc_ids:
                reciprocal_ranks.append(1/rank)
                break
        else:
            reciprocal_ranks.append(0)
        
        precisions.append(precision)
        recalls.append(recall)
    
    return {
        "precision@5": np.mean(precisions),
        "recall@5": np.mean(recalls),
        "MRR": np.mean(reciprocal_ranks)
    }

Answer Quality

def evaluate_answer_quality(test_qa_pairs):
    """
    Use LLM-as-judge to evaluate answers
    """
    
    scores = []
    
    for query, expected_answer in test_qa_pairs:
        # Generate answer
        generated_answer = rag.query(query)['answer']
        
        # LLM judges quality
        judge_prompt = f"""
        Question: {query}
        Expected: {expected_answer}
        Generated: {generated_answer}
        
        Rate the generated answer on:
        1. Accuracy (0-10): Factually correct?
        2. Completeness (0-10): Addresses all aspects?
        3. Relevance (0-10): Stays on topic?
        
        Return as JSON: {{"accuracy": X, "completeness": Y, "relevance": Z}}
        """
        
        scores_json = llm.invoke(judge_prompt).content
        scores.append(json.loads(scores_json))
    
    return {
        "avg_accuracy": np.mean([s['accuracy'] for s in scores]),
        "avg_completeness": np.mean([s['completeness'] for s in scores]),
        "avg_relevance": np.mean([s['relevance'] for s in scores])
    }

Production Challenges

Latency Optimization

# Problem: RAG can be slow
# Retrieval: 50-100ms
# LLM generation: 500-2000ms
# Total: 550-2100ms

# Solutions:

# 1. Cache frequent queries
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_retrieval(query):
    return tuple(vectorstore.similarity_search(query, k=5))

# 2. Streaming responses
def stream_rag_response(query):
    docs = vectorstore.similarity_search(query, k=5)
    context = "\n".join([d.page_content for d in docs])
    
    for chunk in llm.stream(f"Context: {context}\n\nQuestion: {query}\n\nAnswer:"):
        yield chunk  # Send to user immediately

# 3. Parallel retrieval
import asyncio

async def parallel_retrieve(queries):
    tasks = [vectorstore.asimilarity_search(q, k=3) for q in queries]
    return await asyncio.gather(*tasks)

# 4. Smaller models for simple queries
def adaptive_model_selection(query_complexity):
    if query_complexity == "simple":
        return ChatOpenAI(model="gpt-4o-mini")  # Fast & cheap
    else:
        return ChatOpenAI(model="gpt-4o")  # Slow & accurate

Cost Optimization

# Costs:
# - Embedding: $0.02 per 1M tokens (one-time indexing)
# - Retrieval: Free (local vector search)
# - LLM generation: $0.15-$10 per 1M tokens

# Optimization strategies:

# 1. Reduce context size
def smart_context_selection(docs, max_tokens=2000):
    # Instead of top-5, dynamically select until token limit
    selected = []
    total_tokens = 0
    
    for doc in docs:
        doc_tokens = len(doc.page_content) // 4
        if total_tokens + doc_tokens > max_tokens:
            break
        selected.append(doc)
        total_tokens += doc_tokens
    
    return selected

# 2. Use cheaper models when possible
def tiered_generation(query_complexity, docs):
    if query_complexity == "simple":
        llm = ChatOpenAI(model="gpt-4o-mini")  # $0.15/1M
    else:
        llm = ChatOpenAI(model="gpt-4o")  # $2.50/1M
    
    return llm.invoke(format_prompt(query, docs))

# 3. Batch processing for offline queries
def batch_process(queries, batch_size=10):
    # Process multiple queries in one LLM call
    combined_prompt = "\n\n".join([
        f"Q{i}: {q}" for i, q in enumerate(queries)
    ])
    
    # Single API call instead of multiple
    return llm.invoke(combined_prompt)

Quality Monitoring

class RAGMonitor:
    def __init__(self):
        self.query_log = []
        self.feedback_log = []
    
    def log_query(self, query, retrieved_docs, answer, latency):
        self.query_log.append({
            "timestamp": datetime.now(),
            "query": query,
            "num_docs_retrieved": len(retrieved_docs),
            "answer_length": len(answer),
            "latency_ms": latency
        })
    
    def log_feedback(self, query, answer, thumbs_up):
        self.feedback_log.append({
            "query": query,
            "answer": answer,
            "positive": thumbs_up
        })
    
    def get_metrics(self):
        return {
            "avg_latency": np.mean([q['latency_ms'] for q in self.query_log]),
            "satisfaction_rate": np.mean([f['positive'] for f in self.feedback_log]),
            "queries_per_day": len(self.query_log) / 7,
            "avg_docs_retrieved": np.mean([q['num_docs_retrieved'] for q in self.query_log])
        }
    
    def detect_issues(self):
        issues = []
        
        # High latency
        if np.mean([q['latency_ms'] for q in self.query_log[-100:]]) > 2000:
            issues.append("High latency detected")
        
        # Low satisfaction
        recent_satisfaction = np.mean([f['positive'] for f in self.feedback_log[-50:]])
        if recent_satisfaction < 0.7:
            issues.append("Low user satisfaction")
        
        # No relevant docs found
        no_results_rate = sum(1 for q in self.query_log if q['num_docs_retrieved'] == 0) / len(self.query_log)
        if no_results_rate > 0.1:
            issues.append("High rate of queries with no results")
        
        return issues

monitor = RAGMonitor()

# In production:
start = time.time()
docs = vectorstore.similarity_search(query, k=5)
answer = generate_answer(query, docs)
latency = (time.time() - start) * 1000

monitor.log_query(query, docs, answer, latency)

# Check health
issues = monitor.detect_issues()
if issues:
    alert_team(issues)

Production Checklist

✅ Indexing:
   - Document processing pipeline automated
   - Incremental updates (not full reindex)
   - Chunking strategy tested and validated
   - Metadata extraction comprehensive

✅ Retrieval:
   - Hybrid search (dense + sparse)
   - Re-ranking for critical queries
   - Caching for common queries
   - Fallback for no results

✅ Generation:
   - Prompt engineering tested
   - Citation/source tracking
   - Hallucination detection
   - Streaming for better UX

✅ Evaluation:
   - Test set with ground truth
   - Retrieval metrics (Precision@K, Recall@K)
   - Answer quality metrics
   - Regular eval runs

✅ Monitoring:
   - Latency tracking
   - Cost tracking (embedding + LLM)
   - User feedback collection
   - Error rate monitoring

✅ Safety:
   - Input validation (query length, injection attempts)
   - Output filtering (no PII leakage)
   - Rate limiting per user
   - Access control to documents

Tóm Tắt

mindmap
  root((Advanced<br/>RAG))
    Query Optimization
      Expansion
        Alternative phrasings
        Better coverage
      Decomposition
        Break complex queries
        Multi-step reasoning
      Rewriting
        Fix vague queries
        Use conversation context
    Multi-Hop Reasoning
      Self-Ask
        LLM asks follow-ups
        Iterative retrieval
      Chain multiple retrievals
        Combine information
    Agentic RAG
      Tool Use
        Calculator Web search
        Dynamic tool selection
      Self-Correction
        Critique own answer
        Retry if wrong
    Production
      Latency
        Caching Streaming
        Parallel retrieval
      Cost
        Smaller models
        Smart context selection
      Quality
        Monitoring Metrics
        User feedback

Key Takeaways:

✅ Query optimization:

Expansion: Generate alternative phrasings (+15% recall)
Decomposition: Break complex into simple sub-queries
Rewriting: Fix vague queries using context

✅ Multi-hop reasoning:

Self-ask pattern: LLM asks itself follow-ups
Combine info from multiple retrievals
Essential for complex questions

✅ Agentic RAG:

Give tools (calculator, web search, etc.)
Self-correction: Critique and improve answers
+20% accuracy on complex tasks

✅ Production challenges:

Latency: Cache + stream + parallel retrieval
Cost: Tiered models + smart context selection
Quality: Monitor metrics + user feedback

✅ Evaluation metrics:

Retrieval: Precision@K, Recall@K, MRR
Answer: Accuracy, completeness, relevance
Use LLM-as-judge for automated evaluation