Agentic RAG: When Retrieval Meets Autonomous Reasoning

User Query
    ↓
[Query Analyzer] → Classify complexity, identify sub-questions
    ↓
[Strategy Planner] → Determine retrieval approach
    ↓
[Retrieval Agent] ←→ [Vector Store / Search APIs]
    ↓                      ↑
[Result Evaluator] ────────┘ (iterate if insufficient)
    ↓
[Synthesis Agent] → Combine findings, resolve conflicts
    ↓
[Response Generator] → Format final answer with citations
    ↓
[Self-Critic] → Verify answer quality, trigger refinement

from openai import OpenAI
from typing import List, Dict, Tuple
from enum import Enum
import re

class QueryComplexity(Enum):
    SIMPLE = "simple"  # Single-shot RAG
    MULTI_FACETED = "multi_faceted"  # Parallel retrieval
    MULTI_HOP = "multi_hop"  # Sequential retrieval
    EXPLORATORY = "exploratory"  # Iterative expansion
    VERIFICATION = "verification"  # Multi-source check

class QueryAnalyzer:
    """
    Analyze query complexity and decompose into sub-questions.

    This is the routing layer that decides whether to use simple RAG
    or activate the full agentic system.
    """

    def __init__(self):
        self.client = OpenAI()

        # Keywords that indicate different query types
        self.comparison_keywords = ['compare', 'vs', 'versus', 'difference between', 'better']
        self.analysis_keywords = ['analyze', 'why', 'how did', 'what caused', 'drivers', 'factors']
        self.multi_hop_keywords = ['who manages', 'which team', 'related to', 'leads to']
        self.exploratory_keywords = ['trends', 'insights', 'overview', 'landscape', 'emerging']

    def analyze(self, query: str) -> Dict:
        """
        Analyze query to determine complexity and strategy.

        Returns:
            Dict with:
            - complexity: QueryComplexity enum
            - sub_questions: List of decomposed questions (if complex)
            - strategy: Recommended retrieval strategy
            - reasoning: Why this classification
        """
        # Classify complexity
        complexity = self._classify_complexity(query)

        # Decompose if needed
        sub_questions = []
        dependencies = []

        if complexity != QueryComplexity.SIMPLE:
            sub_questions, dependencies = self._decompose_query(query)

        # Determine strategy
        strategy = self._determine_strategy(complexity, sub_questions)

        return {
            'complexity': complexity,
            'sub_questions': sub_questions,
            'dependencies': dependencies,
            'strategy': strategy,
            'use_agentic': complexity != QueryComplexity.SIMPLE
        }

    def _classify_complexity(self, query: str) -> QueryComplexity:
        """
        Classify query complexity using rule-based + LLM approach.

        Rule-based catches common patterns quickly.
        LLM handles edge cases and nuanced queries.
        """
        query_lower = query.lower()

        # Rule-based classification (fast path)
        if any(kw in query_lower for kw in self.verification_keywords):
            return QueryComplexity.VERIFICATION

        if any(kw in query_lower for kw in self.multi_hop_keywords):
            return QueryComplexity.MULTI_HOP

        if any(kw in query_lower for kw in self.exploratory_keywords):
            return QueryComplexity.EXPLORATORY

        if any(kw in query_lower for kw in self.comparison_keywords):
            return QueryComplexity.MULTI_FACETED

        if any(kw in query_lower for kw in self.analysis_keywords):
            return QueryComplexity.MULTI_FACETED

        # Check for compound questions (multiple "and"/"or")
        if query_lower.count(' and ') >= 2 or query_lower.count(' or ') >= 2:
            return QueryComplexity.MULTI_FACETED

        # Check for question chaining ("... and what...", "... and why...")
        if re.search(r'and (what|why|how|when|where|who)', query_lower):
            return QueryComplexity.MULTI_HOP

        # LLM-based classification for ambiguous cases
        return self._llm_classify(query)

    def _llm_classify(self, query: str) -> QueryComplexity:
        """Use LLM to classify query complexity."""
        prompt = f"""Classify this query's complexity:

Query: {query}

Classifications:
- SIMPLE: Single factual question, one retrieval likely sufficient
- MULTI_FACETED: Multiple independent aspects, needs parallel searches
- MULTI_HOP: Requires chaining searches (answer to Q1 informs Q2)
- EXPLORATORY: Open-ended, requires iterative expansion
- VERIFICATION: Needs cross-checking across multiple sources

Return only the classification name.
"""

        response = self.client.chat.completions.create(
            model="gpt-4-turbo-preview",
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )

        classification = response.choices[0].message.content.strip().upper()

        try:
            return QueryComplexity[classification]
        except KeyError:
            # Default to SIMPLE if unclear
            return QueryComplexity.SIMPLE

    def _decompose_query(self, query: str) -> Tuple[List[Dict], List[Tuple]]:
        """
        Decompose complex query into sub-questions with dependencies.

        Returns:
            Tuple of (sub_questions, dependencies)
            sub_questions: List of dicts with 'id', 'question', 'type'
            dependencies: List of (question_id, depends_on_id) tuples
        """
        prompt = f"""Decompose this complex query into atomic sub-questions.

Query: {query}

Requirements:
1. Each sub-question should be answerable independently
2. Identify dependencies (questions that must be answered first)
3. Order questions by dependency chain
4. Mark the type of each question (factual, comparison, temporal, causal)

Return JSON format:
{{
    "sub_questions": [
        {{
            "id": "q1",
            "question": "...",
            "type": "factual|comparison|temporal|causal",
            "depends_on": []  // IDs of questions that must be answered first
        }},
        ...
    ],
    "reasoning": "Why this decomposition?"
}}

JSON Response:"""

        response = self.client.chat.completions.create(
            model="gpt-4-turbo-preview",
            temperature=0,
            response_format={"type": "json_object"},
            messages=[{"role": "user", "content": prompt}]
        )

        import json
        result = json.loads(response.choices[0].message.content)

        sub_questions = result['sub_questions']

        # Build dependency list
        dependencies = []
        for sq in sub_questions:
            for dep_id in sq.get('depends_on', []):
                dependencies.append((sq['id'], dep_id))

        return sub_questions, dependencies

    def _determine_strategy(
        self,
        complexity: QueryComplexity,
        sub_questions: List[Dict]
    ) -> Dict:
        """Determine the optimal retrieval strategy."""

        if complexity == QueryComplexity.SIMPLE:
            return {
                'type': 'single_shot',
                'parallel': False,
                'max_iterations': 1
            }

        elif complexity == QueryComplexity.MULTI_FACETED:
            return {
                'type': 'parallel_retrieval',
                'parallel': True,
                'max_iterations': 1,
                'merge_strategy': 'interleave'
            }

        elif complexity == QueryComplexity.MULTI_HOP:
            return {
                'type': 'sequential_retrieval',
                'parallel': False,
                'max_iterations': len(sub_questions),
                'confidence_threshold': 0.7  # Move to next hop if confident enough
            }

        elif complexity == QueryComplexity.EXPLORATORY:
            return {
                'type': 'iterative_expansion',
                'parallel': False,
                'max_iterations': 5,
                'expansion_strategy': 'follow_references'
            }

        elif complexity == QueryComplexity.VERIFICATION:
            return {
                'type': 'multi_source_verification',
                'parallel': True,
                'min_sources': 3,
                'consistency_threshold': 0.8
            }

        return {'type': 'single_shot'}


# Usage
analyzer = QueryAnalyzer()

# Simple query
result1 = analyzer.analyze("What is our refund policy?")
print(f"Complexity: {result1['complexity']}")
print(f"Use agentic: {result1['use_agentic']}")

# Complex query
result2 = analyzer.analyze(
    "How did our customer satisfaction change after we implemented "
    "the new support system, and what were the main drivers?"
)
print(f"\nComplexity: {result2['complexity']}")
print(f"Sub-questions: {len(result2['sub_questions'])}")
for sq in result2['sub_questions']:
    print(f"  - [{sq['id']}] {sq['question']}")
    if sq['depends_on']:
        print(f"    Depends on: {sq['depends_on']}")
print(f"Strategy: {result2['strategy']}")

1. Consider the current sub-question
2. Decide which search tool(s) to use
3. Execute search, receive results
4. Evaluate: Do I have enough information?
   - If yes: Move to next sub-question or synthesis
   - If no: Reformulate query or try different tool
5. Repeat until satisfied or max iterations reached

Question: "What was the revenue impact of our European expansion?"

Thought: I need to find information about European expansion and its revenue impact. Let me start with a search for European expansion timeline.

Action: vector_search("European expansion timeline launch")
Observation: [Retrieved 3 documents about EU market entry in Q2 2023...]

Thought: Good, I found when we expanded. Now I need revenue data before and after Q2 2023.

Action: vector_search("revenue Europe Q1 Q2 Q3 2023 comparison")
Observation: [Retrieved financial reports showing revenue trends...]

Thought: I have timeline and revenue data. Let me verify with another source.

Action: keyword_search("European revenue 2023 quarterly")
Observation: [Retrieved quarterly earnings call transcript...]

Thought: Multiple sources confirm ~23% revenue increase attributed to EU. I have enough to answer.

Answer: The European expansion in Q2 2023 contributed to a 23% increase in revenue...

from openai import OpenAI
from typing import List, Dict, Callable, Optional
import re
import json

class ReActAgent:
    """
    ReAct-style agentic RAG: Reasoning + Acting in iterative loop.

    The agent:
    1. Observes the question and context
    2. Thinks about what to do next
    3. Acts using available tools
    4. Observes the results
    5. Repeats until confident in answer

    This implements the core agentic RAG pattern.
    """

    def __init__(
        self,
        retriever,  # HybridRetriever from previous code
        model: str = "gpt-4-turbo-preview",
        max_iterations: int = 5
    ):
        self.client = OpenAI()
        self.retriever = retriever
        self.model = model
        self.max_iterations = max_iterations

        # Define available tools
        self.tools = {
            'vector_search': self._vector_search,
            'keyword_search': self._keyword_search,
            'hybrid_search': self._hybrid_search,
            'get_document': self._get_document,
        }

    def run(self, query: str) -> Dict:
        """
        Run the ReAct loop to answer the query.

        Returns:
            Dict with:
            - answer: Final answer
            - reasoning_trace: List of thought-action-observation steps
            - total_iterations: Number of loops
            - confidence: Estimated confidence
        """
        # Initialize reasoning trace
        trace = []
        messages = [
            {"role": "system", "content": self._get_system_prompt()},
            {"role": "user", "content": f"Question: {query}"}
        ]

        iteration = 0
        answer = None

        while iteration < self.max_iterations and not answer:
            iteration += 1

            # Get agent's next thought + action
            response = self.client.chat.completions.create(
                model=self.model,
                temperature=0,
                messages=messages
            )

            agent_message = response.choices[0].message.content

            # Parse the response
            thought, action, action_input, final_answer = self._parse_agent_response(
                agent_message
            )

            # Log to trace
            step = {
                'iteration': iteration,
                'thought': thought,
                'action': action,
                'action_input': action_input
            }

            # Check if agent provided final answer
            if final_answer:
                answer = final_answer
                step['final_answer'] = answer
                trace.append(step)
                break

            # Execute action
            if action and action in self.tools:
                try:
                    observation = self.tools[action](action_input)
                    step['observation'] = observation

                    # Add observation to conversation
                    messages.append({"role": "assistant", "content": agent_message})
                    messages.append({
                        "role": "user",
                        "content": f"Observation: {observation}"
                    })

                except Exception as e:
                    step['observation'] = f"Error: {str(e)}"
                    messages.append({"role": "assistant", "content": agent_message})
                    messages.append({
                        "role": "user",
                        "content": f"Observation: Error executing {action}: {str(e)}"
                    })

            else:
                step['observation'] = f"Unknown action: {action}"
                messages.append({"role": "assistant", "content": agent_message})
                messages.append({
                    "role": "user",
                    "content": f"Observation: Unknown action '{action}'. Available actions: {list(self.tools.keys())}"
                })

            trace.append(step)

        # If no answer after max iterations, force generation
        if not answer:
            messages.append({
                "role": "user",
                "content": "You've reached the iteration limit. Based on what you've gathered, provide your final answer."
            })

            response = self.client.chat.completions.create(
                model=self.model,
                temperature=0,
                messages=messages
            )

            answer = response.choices[0].message.content

        # Estimate confidence based on retrieval quality
        confidence = self._estimate_confidence(trace)

        return {
            'answer': answer,
            'reasoning_trace': trace,
            'total_iterations': iteration,
            'confidence': confidence
        }

    def _get_system_prompt(self) -> str:
        """Define agent behavior and available tools."""
        return """You are a research agent that answers questions using a ReAct (Reasoning + Acting) approach.

Available tools:
- vector_search(query): Semantic search for relevant documents
- keyword_search(query): Keyword-based search (good for exact terms, acronyms)
- hybrid_search(query): Combined semantic + keyword search
- get_document(doc_id): Retrieve full document by ID

Format your response as:

Thought: [Your reasoning about what to do next]
Action: [tool_name]
Action Input: [input to the tool]

Or, when you have enough information:

Thought: [Your reasoning about why you can now answer]
Final Answer: [Your complete answer with citations]

Instructions:
1. Think step-by-step about what information you need
2. Use tools iteratively to gather information
3. If initial results are insufficient, reformulate and search again
4. Verify important facts across multiple sources when possible
5. Provide a Final Answer only when confident
6. Cite sources in your answer using [doc_id] notation

Begin!"""

    def _parse_agent_response(self, response: str) -> tuple:
        """
        Parse agent response to extract thought, action, and answer.

        Returns:
            (thought, action, action_input, final_answer)
        """
        # Extract thought
        thought_match = re.search(r'Thought:\s*(.*?)(?=\n|Action:|Final Answer:|$)', response, re.DOTALL)
        thought = thought_match.group(1).strip() if thought_match else None

        # Check for final answer
        final_answer_match = re.search(r'Final Answer:\s*(.*?)$', response, re.DOTALL)
        if final_answer_match:
            return thought, None, None, final_answer_match.group(1).strip()

        # Extract action
        action_match = re.search(r'Action:\s*(\w+)', response)
        action = action_match.group(1) if action_match else None

        # Extract action input
        action_input_match = re.search(r'Action Input:\s*(.*?)(?=\n|$)', response)
        action_input = action_input_match.group(1).strip() if action_input_match else None

        return thought, action, action_input, None

    # Tool implementations
    def _vector_search(self, query: str) -> str:
        """Execute vector search and format results."""
        results = self.retriever.retrieve(query)

        formatted = []
        for i, result in enumerate(results[:5], 1):  # Top 5
            formatted.append(
                f"[doc_{i}] (score: {result['final_score']:.2f})\n"
                f"{result['text'][:300]}..."
            )

        return "\n\n".join(formatted) if formatted else "No results found."

    def _keyword_search(self, query: str) -> str:
        """Execute keyword search (BM25)."""
        # Use BM25 component of hybrid retriever
        results = self.retriever.bm25_retriever.get_relevant_documents(query)

        formatted = []
        for i, result in enumerate(results[:5], 1):
            formatted.append(
                f"[doc_{i}]\n{result.page_content[:300]}..."
            )

        return "\n\n".join(formatted) if formatted else "No results found."

    def _hybrid_search(self, query: str) -> str:
        """Execute hybrid search."""
        return self._vector_search(query)  # Our retriever is already hybrid

    def _get_document(self, doc_id: str) -> str:
        """Retrieve full document by ID."""
        # Implementation depends on your document store
        return f"Full content of {doc_id}: [document content here]"

    def _estimate_confidence(self, trace: List[Dict]) -> float:
        """
        Estimate confidence based on reasoning trace.

        Factors:
        - Number of sources consulted
        - Retrieval scores
        - Whether verification was done
        """
        if not trace:
            return 0.0

        # More iterations with good results = higher confidence
        successful_retrievals = sum(
            1 for step in trace
            if step.get('observation') and 'No results found' not in step.get('observation', '')
        )

        # Base confidence on successful retrievals
        confidence = min(successful_retrievals / 3, 1.0)  # Max out at 3 good retrievals

        # Bonus for verification (multiple searches on same topic)
        actions = [step.get('action') for step in trace]
        if len(set(actions)) > 2:  # Used multiple different tools
            confidence *= 1.1

        return min(confidence, 1.0)


# Usage Example
from building_production_rag import HybridRetriever  # From previous post

# Initialize retriever (from previous code)
retriever = HybridRetriever(...)

# Initialize ReAct agent
agent = ReActAgent(retriever=retriever, max_iterations=5)

# Run on complex query
result = agent.run(
    "What was the revenue impact of our European expansion and how does it compare to our Asian markets?"
)

print("Answer:", result['answer'])
print(f"\nIterations: {result['total_iterations']}")
print(f"Confidence: {result['confidence']:.2%}")

print("\nReasoning Trace:")
for step in result['reasoning_trace']:
    print(f"\n--- Iteration {step['iteration']} ---")
    if step.get('thought'):
        print(f"Thought: {step['thought']}")
    if step.get('action'):
        print(f"Action: {step['action']}({step.get('action_input', '')})")
    if step.get('observation'):
        print(f"Observation: {step['observation'][:200]}...")

Query: "Analyze the competitive landscape for our new product launch"

Plan:
1. Identify our new product and its key features
2. Find direct competitors in this product category
3. Compare pricing across competitors
4. Analyze competitor strengths and weaknesses
5. Identify market gaps and opportunities
6. Synthesize competitive positioning recommendation

[Generate initial answer]
↓
[Self-critique]
- Is every claim supported by retrieved evidence?
- Are there logical gaps in the reasoning?
- Did I miss any relevant aspect of the question?
- Is my confidence level appropriate?
↓
[If issues found]
- Identify what additional information is needed
- Retrieve more context
- Regenerate improved answer
↓
[Final answer with confidence score]