Query Rewriting and Expansion for Agentic Retrieval: Better Search Starts Here

Users do not ask perfect questions. They type fragments, use ambiguous terms, and expect precise answers. Query rewriting and expansion bridge the gap between what users ask and what the retrieval system needs to find. In agentic RAG, the agent itself decides how to transform queries—making this a critical capability.

Why Raw Queries Fail Retrieval

Consider a user asking: “How do I fix the bug in my app?”

Problems with this query:

1. Too vague: What bug? What app? What language?
2. No technical terms: The vector store contains “debugging,” “error handling,” “patch”
3. Single formulation: Only one chance to match relevant documents

Query rewriting addresses all three problems.

The Query Rewriting Pipeline

User Query
    ↓
[Query Analysis] → What type of question is this?
    ↓
[Query Rewriting] → Rephrase for clarity
    ↓
[Query Expansion] → Generate alternative formulations
    ↓
[Query Decomposition] → Break complex questions into sub-queries
    ↓
[Retrieval] → Search with multiple queries
    ↓
[Merge & Rank] → Combine and deduplicate results

Technique 1: Query Rewriting with LLM

The simplest approach uses the LLM to rephrase the user’s query:

from llama_index.core import PromptTemplate

rewrite_prompt = PromptTemplate(
    """You are a query rewriting expert. Given a user query, 
    rewrite it to be more specific and searchable for a vector database.

User query: {query}

Rewritten query:"""
)

async def rewrite_query(query: str, llm) -> str:
    response = await llm.complete(
        rewrite_prompt.format(query=query)
    )
    return response.text.strip()

# Example
original = "how to fix the thing"
rewritten = await rewrite_query(original, llm)
# "debugging techniques for software error resolution"

Rewriting Strategies

REWRITE_STRATEGIES = {
    "technical": {
        "instruction": "Convert natural language to technical search terms",
        "example": {
            "input": "my app is slow",
            "output": "application performance optimization latency reduction"
        }
    },
    "conceptual": {
        "instruction": "Extract the core concept from a vague question",
        "example": {
            "input": "what's that thing for",
            "output": "functionality purpose of component"
        }
    },
    "comparative": {
        "instruction": "Identify comparison entities and relationship",
        "example": {
            "input": "is X better than Y",
            "output": "comparison of X vs Y performance features use cases"
        }
    }
}

Technique 2: Query Expansion

Generate multiple query variations to increase recall:

from llama_index.core.node_postprocessor import QueryRecipientNodePostprocessor

async def expand_query(query: str, llm, num_expansions: int = 3) -> list[str]:
    """Generate multiple query variations."""
    expansion_prompt = f"""Generate {num_expansions} different search queries 
    that would find documents relevant to: "{query}"
    
    Return each query on a new line, prefixed with a number."""
    
    response = await llm.complete(expansion_prompt)
    
    queries = [query]  # Always include original
    for line in response.text.strip().split("\n"):
        cleaned = line.strip()
        if cleaned and cleaned[0].isdigit():
            # Remove numbering
            expanded = cleaned.split(".", 1)[-1].strip()
            queries.append(expanded)
    
    return queries

# Example
original = "how to deploy a Python app"
expanded = await expand_query(original, llm)
# [
#   "how to deploy a Python app",
#   "Python application deployment guide",
#   "deploying Flask Django FastAPI to production",
#   "Python app deployment Docker Kubernetes cloud"
# ]

Multi-Query Retrieval

from llama_index.core import VectorStoreIndex
from llama_index.core.retriever import VectorIndexRetriever

class MultiQueryRetriever:
    def __init__(self, index: VectorStoreIndex, llm):
        self.index = index
        self.llm = llm
        self.base_retriever = VectorIndexRetriever(
            index=index,
            similarity_top_k=5,
        )
    
    async def retrieve(self, query: str) -> list:
        """Retrieve using multiple query formulations."""
        # Expand the query
        queries = await expand_query(query, self.llm)
        
        # Retrieve for each query
        all_nodes = {}
        for q in queries:
            results = self.base_retriever.retrieve(q)
            for node in results:
                # Deduplicate by node ID, keep highest score
                if node.node.node_id not in all_nodes or \
                   node.score > all_nodes[node.node.node_id].score:
                    all_nodes[node.node.node_id] = node
        
        # Sort by score and return top results
        ranked = sorted(
            all_nodes.values(), 
            key=lambda x: x.score, 
            reverse=True
        )
        return ranked[:10]

# Usage
retriever = MultiQueryRetriever(index, llm)
results = await retriever.retrieve("how to handle errors in Python")

Technique 3: Query Decomposition

Break complex questions into sub-questions:

class QueryDecomposer:
    def __init__(self, llm):
        self.llm = llm
    
    async def decompose(self, query: str) -> list[str]:
        """Break a complex query into sub-questions."""
        decomposition_prompt = f"""Break this complex question into simpler 
        sub-questions that can be answered independently:

"{query}"

Return each sub-question on a new line."""
        
        response = await self.llm.complete(decomposition_prompt)
        
        sub_questions = []
        for line in response.text.strip().split("\n"):
            cleaned = line.strip()
            if cleaned and ("?" in cleaned or cleaned[0].isdigit()):
                question = cleaned.split(".", 1)[-1].strip()
                sub_questions.append(question)
        
        return sub_questions if sub_questions else [query]

# Example
complex_query = "Compare the performance of Redis and Memcached for caching, 
                  including their memory efficiency and ease of setup"
sub_queries = await decomposer.decompose(complex_query)
# [
#   "What is the performance comparison between Redis and Memcached?",
#   "How does memory efficiency compare between Redis and Memcached?",
#   "Which is easier to set up, Redis or Memcached?"
# ]

Sub-Query Retrieval and Synthesis

class DecompositionRAG:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
        self.decomposer = QueryDecomposer(llm)
    
    async def query(self, original_query: str) -> str:
        # Decompose
        sub_queries = await self.decomposer.decompose(original_query)
        
        # Retrieve for each sub-query
        sub_answers = {}
        for sq in sub_queries:
            nodes = await self.retriever.retrieve(sq)
            context = "\n".join([n.node.text for n in nodes[:3]])
            answer = await self._answer_sub_query(sq, context)
            sub_answers[sq] = answer
        
        # Synthesize final answer
        synthesis_prompt = f"""Original question: {original_query}

Sub-answers:
{json.dumps(sub_answers, indent=2)}

Synthesize a comprehensive answer from the sub-answers above:"""
        
        response = await self.llm.complete(synthesis_prompt)
        return response.text.strip()
    
    async def _answer_sub_query(self, query: str, context: str) -> str:
        prompt = f"""Context: {context}

Question: {query}

Answer based on the context:"""
        response = await self.llm.complete(prompt)
        return response.text.strip()

Technique 4: HyDE (Hypothetical Document Embeddings)

Generate a hypothetical answer, then use its embedding to find similar real documents:

class HyDERetriever:
    def __init__(self, index, llm, embed_model):
        self.index = index
        self.llm = llm
        self.embed_model = embed_model
    
    async def retrieve(self, query: str, top_k: int = 5):
        # Generate hypothetical document
        hyde_prompt = f"""Write a detailed passage that would answer this question:
"{query}"

Passage:"""
        
        response = await self.llm.complete(hyde_prompt)
        hypothetical_doc = response.text.strip()
        
        # Embed the hypothetical document
        hyde_embedding = await self.embed_model.aget_text_embedding(
            hypothetical_doc
        )
        
        # Search with the embedding
        from llama_index.core.vector_stores import VectorStoreQuery
        
        query_result = self.index.vector_store.query(
            VectorStoreQuery(
                query_embedding=hyde_embedding,
                similarity_top_k=top_k,
            )
        )
        
        return query_result

Technique 5: Step-Back Prompting

Ask a higher-level question first, then use the answer to inform retrieval:

async def step_back_retrieve(query: str, llm, retriever):
    """Use step-back prompting for better retrieval."""
    # Generate a higher-level question
    step_back_prompt = f"""Given this specific question, generate a more 
    general, background question that would help answer it:

Specific: {query}

General:"""
    
    response = await llm.complete(step_back_prompt)
    general_query = response.text.strip()
    
    # Retrieve with both queries
    specific_results = await retriever.retrieve(query)
    general_results = await retriever.retrieve(general_query)
    
    # Combine results
    all_results = specific_results + general_results
    
    # Deduplicate
    seen = set()
    unique_results = []
    for node in all_results:
        if node.node.node_id not in seen:
            seen.add(node.node.node_id)
            unique_results.append(node)
    
    return unique_results[:10]

Putting It All Together

Complete Agentic RAG Pipeline

class AgenticRAG:
    def __init__(self, index, llm):
        self.index = index
        self.llm = llm
        self.retriever = VectorIndexRetriever(
            index=index, similarity_top_k=5
        )
        self.multi_query = MultiQueryRetriever(index, llm)
        self.decomposer = QueryDecomposer(llm)
    
    async def query(self, user_query: str) -> dict:
        # Analyze query complexity
        is_complex = await self._is_complex(user_query)
        
        if is_complex:
            # Decompose and retrieve
            sub_queries = await self.decomposer.decompose(user_query)
            results = {}
            for sq in sub_queries:
                results[sq] = await self.multi_query.retrieve(sq)
            
            # Synthesize
            answer = await self._synthesize_complex(user_query, results)
            return {
                "answer": answer,
                "strategy": "decomposition",
                "sub_queries": sub_queries,
            }
        else:
            # Multi-query retrieval
            results = await self.multi_query.retrieve(user_query)
            answer = await self._synthesize_simple(user_query, results)
            return {
                "answer": answer,
                "strategy": "multi_query",
            }
    
    async def _is_complex(self, query: str) -> bool:
        """Determine if query needs decomposition."""
        prompt = f"""Is this a complex question that needs to be broken down?
Answer "yes" or "no" only.

Question: {query}"""
        
        response = await self.llm.complete(prompt)
        return "yes" in response.text.lower()

Evaluation

Measure the impact of query rewriting on retrieval quality:

async def evaluate_query_strategies(test_cases, retriever, strategies):
    """Compare different query rewriting strategies."""
    results = {}
    
    for strategy_name, strategy_fn in strategies.items():
        hits = 0
        total = len(test_cases)
        
        for query, expected_docs in test_cases:
            # Apply strategy
            if strategy_fn:
                queries = await strategy_fn(query)
            else:
                queries = [query]
            
            # Retrieve
            retrieved = set()
            for q in queries:
                nodes = await retriever.retrieve(q)
                retrieved.update([n.node.node_id for n in nodes[:5]])
            
            # Check hits
            if retrieved.intersection(expected_docs):
                hits += 1
        
        results[strategy_name] = hits / total
    
    return results

# Strategies to compare
strategies = {
    "none": None,
    "rewrite": rewrite_query,
    "expand": expand_query,
    "decompose": decomposer.decompose,
}

Key Takeaways

1. Users ask vague questions: Query rewriting bridges the gap
2. Multiple formulations increase recall: Expansion catches more relevant documents
3. Complex queries need decomposition: Break them into answerable sub-questions
4. HyDE uses generated content as search queries: Sometimes the answer knows where to look
5. Measure everything: Track retrieval quality before and after rewriting

Query rewriting is not a nice-to-have—it is essential for production RAG systems. The techniques in this guide give you the tools to handle any user query, no matter how imprecise.

References:

• LlamaIndex Query Transformations
• HyDE Paper
• Multi-Query Retrieval