Agentic Retrieval: Letting the Agent Decide What to Fetch

Traditional RAG is a two-step process: retrieve then generate. But real research isn’t that simple. You search, read, realize you need different terms, search again, find a contradiction, dig deeper. Agentic retrieval gives agents this same iterative, adaptive capability — letting them decide what to fetch, evaluate what they found, and search again if needed.

From Static to Agentic RAG

Static RAG (The Old Way)

User Query → Embed Query → Vector Search → Top-K Results → LLM Generate

Problems:

• One-shot retrieval misses relevant context
• No ability to refine searches based on initial results
• Can’t combine multiple retrieval strategies
• Fails when the answer requires synthesizing across sources

Agentic RAG (The New Way)

User Query
    ↓
Agent: What do I need to know?
    ├── Search 1: Vector similarity
    ├── Search 2: Keyword search (if vector fails)
    ├── Search 3: Structured query (if data is tabular)
    ↓
Agent: Did I find enough?
    ├── Yes → Generate answer
    └── No → Refine query, search again

Implementation with LlamaIndex

Setting Up the Agent

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core import (
    VectorStoreIndex,
    SimpleKeywordTableIndex,
    Settings,
)
from llama_index.llms.openai import OpenAI

Settings.llm = OpenAI(model="gpt-4o")

# Create multiple retrieval indexes
vector_index = VectorStoreIndex.from_documents(documents)
keyword_index = SimpleKeywordTableIndex.from_documents(documents)

# Create query engines
vector_engine = vector_index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact"
)

keyword_engine = keyword_index.as_query_engine(
    response_mode="compact"
)

Define Retrieval Tools

Give the agent multiple retrieval strategies:

from llama_index.core.tools import FunctionTool

def vector_search(query: str) -> str:
    """Search using semantic similarity. Best for conceptual queries."""
    response = vector_engine.query(query)
    return str(response)

def keyword_search(query: str) -> str:
    """Search using exact keyword matching. Best for specific terms, names, IDs."""
    response = keyword_engine.query(query)
    return str(response)

def combined_search(query: str) -> str:
    """Search using both semantic and keyword approaches. Most comprehensive."""
    vector_response = vector_engine.query(query)
    keyword_response = keyword_engine.query(query)
    
    return f"""Vector Results:
{vector_response}

Keyword Results:
{keyword_response}"""

# Wrap as tools
retrieval_tools = [
    FunctionTool.from_defaults(
        fn=vector_search,
        name="vector_search",
        description="Semantic search for conceptual queries"
    ),
    FunctionTool.from_defaults(
        fn=keyword_search,
        name="keyword_search",
        description="Keyword search for specific terms"
    ),
    FunctionTool.from_defaults(
        fn=combined_search,
        name="combined_search",
        description="Comprehensive search using both methods"
    ),
]

Create the Agentic Retriever

agent = ReActAgent.from_tools(
    tools=retrieval_tools,
    llm=Settings.llm,
    verbose=True,
    max_iterations=10,
    system_prompt="""You are a research agent that retrieves information from documents.

Your strategy:
1. Start with vector_search for conceptual understanding
2. Use keyword_search for specific terms, names, or IDs
3. Use combined_search when you need comprehensive coverage
4. If initial results are insufficient, refine your query and search again
5. Synthesize findings from multiple searches into a coherent answer

Always explain your reasoning for which search strategy you choose."""
)

# Use the agent
response = agent.chat("What are the key differences between LangChain and LlamaIndex?")

Advanced Agentic Retrieval Patterns

Multi-Hop Retrieval

When answers require information from multiple documents:

class MultiHopRetriever:
    def __init__(self, agent, max_hops: int = 3):
        self.agent = agent
        self.max_hops = max_hops
    
    def retrieve(self, query: str) -> str:
        context = []
        current_query = query
        
        for hop in range(self.max_hops):
            # Retrieve for current query
            results = self.agent.chat(
                f"Search for: {current_query}\n"
                f"Context so far: {' '.join(context)}"
            )
            
            context.append(str(results))
            
            # Ask agent if more information is needed
            follow_up = self.agent.chat(
                f"Based on what you found, is there a follow-up "
                f"question that would help answer: {query}?\n"
                f"If yes, provide the follow-up question. "
                f"If no, respond with 'DONE'."
            )
            
            if "DONE" in str(follow_up).upper():
                break
            
            current_query = str(follow_up)
        
        # Synthesize all context
        final = self.agent.chat(
            f"Original question: {query}\n"
            f"Information gathered across {len(context)} searches:\n"
            + "\n---\n".join(context) +
            "\n\nProvide a comprehensive answer."
        )
        
        return str(final)

Query Rewriting

Let the agent reformulate queries when initial searches fail:

class AdaptiveRetriever:
    def __init__(self, vector_engine, llm):
        self.vector_engine = vector_engine
        self.llm = llm
    
    def retrieve_with_adaptation(
        self,
        query: str,
        max_retries: int = 3,
        relevance_threshold: float = 0.7
    ) -> str:
        current_query = query
        
        for attempt in range(max_retries):
            results = self.vector_engine.query(current_query)
            
            # Evaluate relevance
            relevance = self._evaluate_relevance(str(results), query)
            
            if relevance >= relevance_threshold:
                return str(results)
            
            # Rewrite query for next attempt
            current_query = self._rewrite_query(
                query, current_query, str(results), attempt
            )
        
        return f"Best results after {max_retries} attempts: {str(results)}"
    
    def _rewrite_query(
        self, original: str, current: str, results: str, attempt: int
    ) -> str:
        rewrite_prompt = f"""The following search query didn't return relevant results.

Original question: {original}
Current query: {current}
Attempt: {attempt + 1}

Results found:
{results[:500]}

Rewrite the search query to be more effective. Consider:
- Using different terminology
- Breaking into smaller concepts
- Adding specific context from the results"""
        
        response = self.llm.complete(rewrite_prompt)
        return str(response).strip()

Context Budgeting

Manage token budgets across multiple retrievals:

class ContextBudgetedRetriever:
    def __init__(self, total_budget: int = 8000):
        self.total_budget = total_budget
        self.used_budget = 0
    
    def retrieve_within_budget(
        self,
        retrieval_functions: list[callable],
        query: str
    ) -> str:
        results = []
        
        for func in retrieval_functions:
            remaining = self.total_budget - self.used_budget
            
            if remaining <= 0:
                break
            
            # Estimate token usage
            result = func(query)
            estimated_tokens = len(result.split()) * 1.3  # Rough estimate
            
            if estimated_tokens <= remaining:
                results.append(result)
                self.used_budget += estimated_tokens
        
        return "\n---\n".join(results)

Evaluation Metrics

Measure agentic retrieval quality:

@dataclass
class RetrievalMetrics:
    query_count: int
    total_retrieved: int
    relevant_retrieved: int
    irrelevant_retrieved: int
    answer_quality: float  # 0-1
    
    @property
    def precision(self) -> float:
        if self.total_retrieved == 0:
            return 0.0
        return self.relevant_retrieved / self.total_retrieved
    
    @property
    def recall(self) -> float:
        # Would need ground truth
        pass
    
    @property
    def efficiency(self) -> float:
        """Lower is better - ratio of queries needed."""
        return self.query_count / max(self.total_retrieved, 1)

def evaluate_agentic_retrieval(
    agent,
    test_cases: list[dict]
) -> RetrievalMetrics:
    metrics = {"queries": 0, "retrieved": 0, "relevant": 0}
    
    for case in test_cases:
        response = agent.chat(case["question"])
        # Count retrievals and evaluate relevance
        # ... (implementation depends on your tracing setup)
    
    return RetrievalMetrics(**metrics)

When to Use Agentic Retrieval

Use it when:

• Queries are complex and require multiple searches
• The information spans multiple documents or sources
• Initial retrieval often misses relevant context
• You need to synthesize across different data types

Skip it when:

• Queries are simple and direct
• A single retrieval step suffices
• Latency is critical (agentic retrieval is slower)
• Costs need to be minimized

Conclusion

Agentic retrieval transforms RAG from a static pipeline into an adaptive research process. By letting agents decide what to search, when to refine, and how to combine results, you get answers that static retrieval simply can’t produce. Start with a ReAct agent and multiple retrieval tools, add multi-hop capabilities for complex queries, and measure quality with retrieval-specific metrics.