aishastr.live

Agentic Design Pattern: Agentic Rag

Written by

admin

in


    """
Pattern 067: Agentic RAG (Advanced)

Description:
    Agentic RAG extends basic Retrieval-Augmented Generation with autonomous decision-making
    capabilities. Unlike simple RAG that retrieves once and generates, Agentic RAG employs
    sophisticated query planning, multi-hop reasoning, dynamic retrieval strategies, and source
    validation. The agent can decompose complex queries, determine when and what to retrieve,
    route queries to appropriate sources, validate retrieved information, and iteratively
    refine its understanding through multiple retrieval-reasoning cycles.
    
    This pattern transforms RAG from a passive information lookup into an active research
    process where the agent strategically explores knowledge bases, identifies gaps, verifies
    information across sources, and constructs well-grounded answers through deliberate
    reasoning chains.

Components:
    1. Query Analyzer: Decomposes complex questions into sub-queries
    2. Query Router: Routes queries to appropriate retrieval sources
    3. Multi-Hop Reasoner: Follows chains of reasoning across retrieved documents
    4. Source Validator: Validates credibility and relevance of sources
    5. Retrieval Planner: Decides what to retrieve and when
    6. Answer Synthesizer: Combines information from multiple sources
    7. Confidence Estimator: Assesses answer quality and completeness

Architecture:
    ```
    User Query
        ↓
    Query Analyzer → [Sub-queries]
        ↓
    Query Router → [Route to sources]
        ↓
    Retrieval Planner → [Iterative retrieval strategy]
        ↓
    Multi-Hop Reasoner → [Follow reasoning chains]
        ↓
    Source Validator → [Validate & filter]
        ↓
    Answer Synthesizer → [Combine & generate]
        ↓
    Confidence Estimator → [Quality check]
        ↓
    Final Answer (or iterate)
    ```

Use Cases:
    - Complex research questions requiring multi-source synthesis
    - Scientific literature review and analysis
    - Legal document analysis with precedent lookup
    - Technical troubleshooting requiring multiple knowledge sources
    - Fact-checking and claim verification across sources
    - Exploratory data analysis with iterative refinement

Advantages:
    - Handles complex, multi-part questions effectively
    - Validates information across multiple sources
    - Provides transparent reasoning chains
    - Adapts retrieval strategy based on findings
    - Reduces hallucination through source grounding
    - Identifies and fills knowledge gaps

LangChain Implementation:
    Uses ChatOpenAI for LLM operations, custom query planning, routing logic,
    and iterative retrieval-reasoning loops. Demonstrates query decomposition,
    multi-source retrieval, validation chains, and answer synthesis.
"""

import os
from typing import List, Dict, Any, Optional, Set, Tuple
from dataclasses import dataclass, field
from enum import Enum
import time
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class QueryType(Enum):
    """Types of queries for routing."""
    FACTUAL = "factual"
    COMPARATIVE = "comparative"
    PROCEDURAL = "procedural"
    CAUSAL = "causal"
    EXPLORATORY = "exploratory"


class SourceType(Enum):
    """Types of retrieval sources."""
    DOCUMENTS = "documents"
    WEB = "web"
    DATABASE = "database"
    API = "api"
    KNOWLEDGE_GRAPH = "knowledge_graph"


class ConfidenceLevel(Enum):
    """Confidence in answer quality."""
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INSUFFICIENT = "insufficient"


@dataclass
class SubQuery:
    """Represents a decomposed sub-query."""
    query_id: str
    text: str
    query_type: QueryType
    dependencies: List[str] = field(default_factory=list)  # Other query IDs this depends on
    priority: int = 1


@dataclass
class RetrievalSource:
    """Represents a source for retrieval."""
    source_id: str
    source_type: SourceType
    name: str
    relevance_score: float = 0.0
    
    def retrieve(self, query: str) -> List[str]:
        """Simulate retrieval from this source."""
        # In real implementation, this would access actual data sources
        if self.source_type == SourceType.DOCUMENTS:
            return [
                f"[{self.name}] Document excerpt about: {query}",
                f"[{self.name}] Additional context: {query}"
            ]
        elif self.source_type == SourceType.WEB:
            return [f"[{self.name}] Web search result for: {query}"]
        elif self.source_type == SourceType.DATABASE:
            return [f"[{self.name}] Database record matching: {query}"]
        else:
            return [f"[{self.name}] Information about: {query}"]


@dataclass
class RetrievedDocument:
    """Represents a retrieved document with metadata."""
    doc_id: str
    content: str
    source: RetrievalSource
    relevance_score: float
    credibility_score: float = 0.0
    timestamp: float = field(default_factory=time.time)


@dataclass
class ReasoningStep:
    """Represents a step in multi-hop reasoning."""
    step_id: int
    question: str
    retrieved_docs: List[RetrievedDocument]
    reasoning: str
    answer: str
    confidence: float


@dataclass
class AgenticRAGResult:
    """Complete result from Agentic RAG."""
    query: str
    sub_queries: List[SubQuery]
    reasoning_chain: List[ReasoningStep]
    final_answer: str
    sources_used: List[RetrievalSource]
    confidence: ConfidenceLevel
    gaps_identified: List[str] = field(default_factory=list)


class AgenticRAGAgent:
    """
    Advanced RAG agent with query planning, multi-hop reasoning, and validation.
    """
    
    def __init__(
        self,
        max_hops: int = 3,
        min_confidence: float = 0.7,
        temperature: float = 0.3
    ):
        """
        Initialize the Agentic RAG agent.
        
        Args:
            max_hops: Maximum number of retrieval-reasoning hops
            min_confidence: Minimum confidence threshold for answer
            temperature: Temperature for LLM generation
        """
        self.max_hops = max_hops
        self.min_confidence = min_confidence
        
        # Different LLMs for different tasks
        self.query_analyzer = ChatOpenAI(temperature=0.2, model="gpt-3.5-turbo")
        self.router = ChatOpenAI(temperature=0.1, model="gpt-3.5-turbo")
        self.reasoner = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        self.validator = ChatOpenAI(temperature=0.1, model="gpt-3.5-turbo")
        self.synthesizer = ChatOpenAI(temperature=0.4, model="gpt-3.5-turbo")
        
        # Available sources (in real implementation, these would be actual data sources)
        self.sources = [
            RetrievalSource("src1", SourceType.DOCUMENTS, "Technical Documentation"),
            RetrievalSource("src2", SourceType.WEB, "Web Search"),
            RetrievalSource("src3", SourceType.DATABASE, "Knowledge Database"),
            RetrievalSource("src4", SourceType.KNOWLEDGE_GRAPH, "Entity Graph"),
        ]
        
        self.parser = StrOutputParser()
    
    def decompose_query(self, query: str) -> List[SubQuery]:
        """
        Decompose complex query into sub-queries.
        
        Args:
            query: The complex user query
            
        Returns:
            List of sub-queries with dependencies
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a query analyzer. Decompose complex questions into sub-queries.
For each sub-query, identify:
1. The specific question
2. The type (factual, comparative, procedural, causal, exploratory)
3. Dependencies on other sub-queries
4. Priority (1=highest, 3=lowest)

Format each sub-query as:
Q[id]: [question] | Type: [type] | Depends: [comma-separated ids or 'none'] | Priority: [1-3]"""),
            ("user", "Decompose this query: {query}")
        ])
        
        chain = prompt | self.query_analyzer | self.parser
        
        try:
            result = chain.invoke({"query": query})
            
            # Parse the result into SubQuery objects
            sub_queries = []
            lines = [line.strip() for line in result.split('\n') if line.strip().startswith('Q')]
            
            for line in lines:
                try:
                    # Parse format: Q1: What is X? | Type: factual | Depends: none | Priority: 1
                    parts = line.split('|')
                    if len(parts) >= 4:
                        # Extract query ID and text
                        q_part = parts[0].strip()
                        q_id = q_part.split(':')[0].strip()
                        q_text = ':'.join(q_part.split(':')[1:]).strip()
                        
                        # Extract type
                        type_str = parts[1].split(':')[1].strip().lower()
                        q_type = QueryType(type_str) if type_str in [t.value for t in QueryType] else QueryType.FACTUAL
                        
                        # Extract dependencies
                        deps_str = parts[2].split(':')[1].strip().lower()
                        dependencies = [] if deps_str == 'none' else [d.strip() for d in deps_str.split(',')]
                        
                        # Extract priority
                        priority = int(parts[3].split(':')[1].strip())
                        
                        sub_queries.append(SubQuery(
                            query_id=q_id,
                            text=q_text,
                            query_type=q_type,
                            dependencies=dependencies,
                            priority=priority
                        ))
                except Exception as e:
                    print(f"Warning: Failed to parse sub-query line: {line}")
                    continue
            
            # If parsing failed, create a single sub-query
            if not sub_queries:
                sub_queries = [SubQuery(
                    query_id="Q1",
                    text=query,
                    query_type=QueryType.EXPLORATORY,
                    dependencies=[],
                    priority=1
                )]
            
            return sub_queries
            
        except Exception as e:
            print(f"Error in query decomposition: {e}")
            # Fallback to single query
            return [SubQuery(
                query_id="Q1",
                text=query,
                query_type=QueryType.EXPLORATORY,
                dependencies=[],
                priority=1
            )]
    
    def route_query(self, sub_query: SubQuery) -> List[RetrievalSource]:
        """
        Route sub-query to appropriate sources.
        
        Args:
            sub_query: The sub-query to route
            
        Returns:
            Ranked list of sources to query
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a query router. Given a query and available sources, 
rank which sources are most relevant (1-5 scale).

Available sources:
- Technical Documentation (detailed technical specs)
- Web Search (general information, current events)
- Knowledge Database (structured facts and entities)
- Entity Graph (relationships between concepts)

Format: [Source Name]: [score] - [brief reason]"""),
            ("user", "Query: {query}\nQuery Type: {query_type}\nRank the sources:")
        ])
        
        chain = prompt | self.router | self.parser
        
        try:
            result = chain.invoke({
                "query": sub_query.text,
                "query_type": sub_query.query_type.value
            })
            
            # Parse relevance scores and update sources
            scored_sources = []
            for source in self.sources:
                if source.name in result:
                    # Extract score (look for patterns like "4" or "4/5")
                    import re
                    pattern = rf"{re.escape(source.name)}[:\s]+(\d+)"
                    match = re.search(pattern, result)
                    if match:
                        score = float(match.group(1)) / 5.0  # Normalize to 0-1
                        source.relevance_score = score
                        scored_sources.append(source)
            
            # Sort by relevance and return top sources
            scored_sources.sort(key=lambda x: x.relevance_score, reverse=True)
            return scored_sources[:2] if scored_sources else self.sources[:2]
            
        except Exception as e:
            print(f"Error in query routing: {e}")
            # Default to all sources
            return self.sources[:2]
    
    def retrieve_documents(
        self,
        sub_query: SubQuery,
        sources: List[RetrievalSource]
    ) -> List[RetrievedDocument]:
        """
        Retrieve documents from sources.
        
        Args:
            sub_query: The query to retrieve for
            sources: Sources to retrieve from
            
        Returns:
            List of retrieved documents
        """
        documents = []
        
        for source in sources:
            # Retrieve from source (simulated)
            contents = source.retrieve(sub_query.text)
            
            for i, content in enumerate(contents):
                doc = RetrievedDocument(
                    doc_id=f"{source.source_id}_doc_{i}",
                    content=content,
                    source=source,
                    relevance_score=source.relevance_score * (1.0 - i * 0.1)  # Decay for lower results
                )
                documents.append(doc)
        
        # Sort by relevance
        documents.sort(key=lambda x: x.relevance_score, reverse=True)
        return documents
    
    def validate_sources(self, documents: List[RetrievedDocument]) -> List[RetrievedDocument]:
        """
        Validate credibility and consistency of retrieved documents.
        
        Args:
            documents: Documents to validate
            
        Returns:
            Validated documents with credibility scores
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a source validator. Assess the credibility of information.
Consider:
1. Source authority
2. Information consistency across sources
3. Specificity and detail level
4. Potential biases

Rate credibility 0.0-1.0 for each source and briefly explain."""),
            ("user", "Validate these sources:\n{sources}")
        ])
        
        chain = prompt | self.validator | self.parser
        
        try:
            # Prepare sources text
            sources_text = "\n".join([
                f"- {doc.source.name}: {doc.content[:200]}..."
                for doc in documents[:3]  # Validate top 3
            ])
            
            result = chain.invoke({"sources": sources_text})
            
            # Parse credibility scores (simple heuristic)
            for doc in documents:
                # Higher credibility for documentation and databases
                if doc.source.source_type in [SourceType.DOCUMENTS, SourceType.DATABASE]:
                    doc.credibility_score = 0.9
                elif doc.source.source_type == SourceType.KNOWLEDGE_GRAPH:
                    doc.credibility_score = 0.85
                else:
                    doc.credibility_score = 0.7
            
            # Filter low credibility documents
            validated_docs = [doc for doc in documents if doc.credibility_score >= 0.6]
            return validated_docs
            
        except Exception as e:
            print(f"Error in source validation: {e}")
            # Default: assign medium credibility to all
            for doc in documents:
                doc.credibility_score = 0.75
            return documents
    
    def multi_hop_reasoning(
        self,
        sub_queries: List[SubQuery],
        context: Dict[str, str] = None
    ) -> List[ReasoningStep]:
        """
        Perform multi-hop reasoning across sub-queries.
        
        Args:
            sub_queries: Sub-queries to reason over
            context: Context from previous reasoning steps
            
        Returns:
            List of reasoning steps
        """
        reasoning_steps = []
        context = context or {}
        
        # Sort by priority and dependencies
        sorted_queries = sorted(sub_queries, key=lambda q: (q.priority, len(q.dependencies)))
        
        for step_id, sub_query in enumerate(sorted_queries, 1):
            # Check if dependencies are met
            if sub_query.dependencies:
                missing_deps = [dep for dep in sub_query.dependencies if dep not in context]
                if missing_deps:
                    print(f"Warning: Skipping {sub_query.query_id} due to missing dependencies: {missing_deps}")
                    continue
            
            # Route and retrieve
            sources = self.route_query(sub_query)
            documents = self.retrieve_documents(sub_query, sources)
            validated_docs = self.validate_sources(documents)
            
            # Build context from previous answers
            prev_context = "\n".join([
                f"{qid}: {ans}"
                for qid, ans in context.items()
            ])
            
            # Reason over retrieved documents
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a reasoning agent. Answer the question using retrieved documents.
Explain your reasoning step-by-step and provide a clear answer.
If information is insufficient, state what's missing."""),
                ("user", """Previous context:
{prev_context}

Current question: {question}

Retrieved information:
{documents}

Provide:
1. Your reasoning
2. Your answer
3. Confidence (0.0-1.0)

Format:
Reasoning: [your reasoning]
Answer: [your answer]
Confidence: [0.0-1.0]""")
            ])
            
            chain = prompt | self.reasoner | self.parser
            
            try:
                docs_text = "\n".join([
                    f"- [{doc.source.name}] {doc.content}"
                    for doc in validated_docs[:3]
                ])
                
                result = chain.invoke({
                    "prev_context": prev_context if prev_context else "None",
                    "question": sub_query.text,
                    "documents": docs_text
                })
                
                # Parse reasoning, answer, and confidence
                reasoning = ""
                answer = ""
                confidence = 0.5
                
                for line in result.split('\n'):
                    if line.startswith("Reasoning:"):
                        reasoning = line.replace("Reasoning:", "").strip()
                    elif line.startswith("Answer:"):
                        answer = line.replace("Answer:", "").strip()
                    elif line.startswith("Confidence:"):
                        try:
                            confidence = float(line.replace("Confidence:", "").strip())
                        except:
                            confidence = 0.5
                
                if not answer:
                    answer = result.split('\n')[0]  # Fallback to first line
                
                reasoning_step = ReasoningStep(
                    step_id=step_id,
                    question=sub_query.text,
                    retrieved_docs=validated_docs,
                    reasoning=reasoning if reasoning else "Analyzed retrieved documents.",
                    answer=answer,
                    confidence=confidence
                )
                
                reasoning_steps.append(reasoning_step)
                context[sub_query.query_id] = answer
                
            except Exception as e:
                print(f"Error in reasoning for {sub_query.query_id}: {e}")
                continue
        
        return reasoning_steps
    
    def synthesize_answer(
        self,
        query: str,
        reasoning_steps: List[ReasoningStep]
    ) -> Tuple[str, ConfidenceLevel, List[str]]:
        """
        Synthesize final answer from reasoning steps.
        
        Args:
            query: Original user query
            reasoning_steps: All reasoning steps
            
        Returns:
            Tuple of (final answer, confidence level, identified gaps)
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an answer synthesizer. Combine insights from multiple reasoning steps
into a comprehensive, coherent answer. Identify any gaps in knowledge.

Provide:
1. Synthesized answer
2. Overall confidence (high/medium/low/insufficient)
3. Any gaps or limitations

Format:
Answer: [comprehensive answer]
Confidence: [high/medium/low/insufficient]
Gaps: [identified gaps or 'none']"""),
            ("user", """Original query: {query}

Reasoning steps:
{steps}

Synthesize the final answer:""")
        ])
        
        chain = prompt | self.synthesizer | self.parser
        
        try:
            steps_text = "\n".join([
                f"Step {step.step_id}: {step.question}\n  Reasoning: {step.reasoning}\n  Answer: {step.answer}\n  Confidence: {step.confidence}"
                for step in reasoning_steps
            ])
            
            result = chain.invoke({
                "query": query,
                "steps": steps_text
            })
            
            # Parse result
            answer = ""
            confidence = ConfidenceLevel.MEDIUM
            gaps = []
            
            for line in result.split('\n'):
                if line.startswith("Answer:"):
                    answer = line.replace("Answer:", "").strip()
                    # Continue reading multi-line answer
                    answer_lines = [answer]
                    for next_line in result.split('\n')[result.split('\n').index(line)+1:]:
                        if next_line.startswith("Confidence:") or next_line.startswith("Gaps:"):
                            break
                        if next_line.strip():
                            answer_lines.append(next_line.strip())
                    answer = " ".join(answer_lines)
                elif line.startswith("Confidence:"):
                    conf_str = line.replace("Confidence:", "").strip().lower()
                    if "high" in conf_str:
                        confidence = ConfidenceLevel.HIGH
                    elif "low" in conf_str:
                        confidence = ConfidenceLevel.LOW
                    elif "insufficient" in conf_str:
                        confidence = ConfidenceLevel.INSUFFICIENT
                    else:
                        confidence = ConfidenceLevel.MEDIUM
                elif line.startswith("Gaps:"):
                    gaps_str = line.replace("Gaps:", "").strip()
                    if gaps_str.lower() not in ['none', 'no gaps', '']:
                        gaps = [g.strip() for g in gaps_str.split(',')]
            
            if not answer:
                answer = result.split('\n')[0]  # Fallback
            
            return answer, confidence, gaps
            
        except Exception as e:
            print(f"Error in answer synthesis: {e}")
            # Fallback: combine all reasoning step answers
            combined = " ".join([step.answer for step in reasoning_steps])
            avg_confidence = sum(step.confidence for step in reasoning_steps) / len(reasoning_steps) if reasoning_steps else 0.5
            
            if avg_confidence >= 0.8:
                conf_level = ConfidenceLevel.HIGH
            elif avg_confidence >= 0.6:
                conf_level = ConfidenceLevel.MEDIUM
            elif avg_confidence >= 0.4:
                conf_level = ConfidenceLevel.LOW
            else:
                conf_level = ConfidenceLevel.INSUFFICIENT
            
            return combined, conf_level, ["Error in synthesis process"]
    
    def query(self, user_query: str) -> AgenticRAGResult:
        """
        Process a query through the full Agentic RAG pipeline.
        
        Args:
            user_query: The user's question
            
        Returns:
            Complete Agentic RAG result with reasoning chain
        """
        print(f"\n{'='*60}")
        print(f"Processing Query: {user_query}")
        print(f"{'='*60}\n")
        
        # Step 1: Decompose query
        print("Step 1: Decomposing query...")
        sub_queries = self.decompose_query(user_query)
        print(f"  → Generated {len(sub_queries)} sub-queries")
        for sq in sub_queries:
            print(f"    • {sq.query_id}: {sq.text} (Type: {sq.query_type.value}, Priority: {sq.priority})")
        
        # Step 2: Multi-hop reasoning
        print("\nStep 2: Multi-hop reasoning...")
        reasoning_steps = self.multi_hop_reasoning(sub_queries)
        print(f"  → Completed {len(reasoning_steps)} reasoning steps")
        
        # Step 3: Synthesize answer
        print("\nStep 3: Synthesizing final answer...")
        final_answer, confidence, gaps = self.synthesize_answer(user_query, reasoning_steps)
        print(f"  → Confidence: {confidence.value}")
        
        # Collect all sources used
        sources_used = set()
        for step in reasoning_steps:
            for doc in step.retrieved_docs:
                sources_used.add(doc.source)
        
        return AgenticRAGResult(
            query=user_query,
            sub_queries=sub_queries,
            reasoning_chain=reasoning_steps,
            final_answer=final_answer,
            sources_used=list(sources_used),
            confidence=confidence,
            gaps_identified=gaps
        )


def demonstrate_agentic_rag():
    """Demonstrate Agentic RAG pattern with various scenarios."""
    
    print("="*80)
    print("AGENTIC RAG (ADVANCED) - DEMONSTRATION")
    print("="*80)
    
    agent = AgenticRAGAgent(max_hops=3, min_confidence=0.7)
    
    # Test 1: Multi-part complex query
    print("\n" + "="*80)
    print("TEST 1: Complex Multi-Part Query")
    print("="*80)
    
    result1 = agent.query(
        "What is the difference between machine learning and deep learning, "
        "and which one should I use for image classification?"
    )
    
    print("\n--- RESULT ---")
    print(f"Final Answer: {result1.final_answer}")
    print(f"Confidence: {result1.confidence.value}")
    print(f"Sources Used: {[s.name for s in result1.sources_used]}")
    print(f"Sub-queries: {len(result1.sub_queries)}")
    print(f"Reasoning Steps: {len(result1.reasoning_chain)}")
    if result1.gaps_identified:
        print(f"Gaps Identified: {', '.join(result1.gaps_identified)}")
    
    # Test 2: Causal query requiring multi-hop reasoning
    print("\n" + "="*80)
    print("TEST 2: Causal Query with Multi-Hop Reasoning")
    print("="*80)
    
    result2 = agent.query(
        "Why do neural networks need activation functions, and what happens "
        "if we remove them?"
    )
    
    print("\n--- REASONING CHAIN ---")
    for step in result2.reasoning_chain:
        print(f"\nStep {step.step_id}: {step.question}")
        print(f"  Reasoning: {step.reasoning}")
        print(f"  Answer: {step.answer}")
        print(f"  Confidence: {step.confidence:.2f}")
        print(f"  Sources: {[doc.source.name for doc in step.retrieved_docs[:2]]}")
    
    print("\n--- FINAL ANSWER ---")
    print(result2.final_answer)
    print(f"Overall Confidence: {result2.confidence.value}")
    
    # Test 3: Comparative analysis
    print("\n" + "="*80)
    print("TEST 3: Comparative Analysis Query")
    print("="*80)
    
    result3 = agent.query(
        "Compare the advantages and disadvantages of SQL vs NoSQL databases "
        "for a high-traffic social media application."
    )
    
    print("\n--- QUERY DECOMPOSITION ---")
    for sq in result3.sub_queries:
        print(f"{sq.query_id}: {sq.text}")
        print(f"  Type: {sq.query_type.value}")
        print(f"  Priority: {sq.priority}")
        if sq.dependencies:
            print(f"  Depends on: {', '.join(sq.dependencies)}")
    
    print("\n--- FINAL ANSWER ---")
    print(result3.final_answer)
    print(f"\nSources consulted: {[s.name for s in result3.sources_used]}")
    print(f"Confidence: {result3.confidence.value}")
    
    # Summary
    print("\n" + "="*80)
    print("SUMMARY: Agentic RAG (Advanced)")
    print("="*80)
    print("""
The Agentic RAG pattern demonstrates advanced information retrieval and reasoning:

Key Features Demonstrated:
1. Query Decomposition: Breaks complex queries into manageable sub-queries
2. Intelligent Routing: Routes queries to appropriate specialized sources
3. Multi-Hop Reasoning: Follows chains of logic across multiple retrieval steps
4. Source Validation: Assesses credibility and consistency of information
5. Answer Synthesis: Combines insights from multiple sources coherently
6. Confidence Estimation: Provides transparency about answer quality
7. Gap Identification: Identifies missing information or limitations

Benefits:
• Handles complex, multi-faceted questions effectively
• Provides transparent reasoning chains for interpretability
• Validates information across sources to reduce hallucination
• Adapts retrieval strategy based on intermediate findings
• Identifies knowledge gaps and uncertainty explicitly
• Grounds answers in retrieved sources with citations

Use Cases:
• Research and literature review
• Technical troubleshooting and diagnosis
• Comparative analysis and decision support
• Fact-checking and claim verification
• Legal and regulatory document analysis
• Medical diagnosis support (with appropriate validation)

Comparison with Basic RAG:
┌─────────────────────┬──────────────────┬────────────────────┐
│ Aspect              │ Basic RAG        │ Agentic RAG        │
├─────────────────────┼──────────────────┼────────────────────┤
│ Query Processing    │ Single-shot      │ Decomposed         │
│ Retrieval Strategy  │ One-time lookup  │ Iterative/adaptive │
│ Reasoning           │ Direct answer    │ Multi-hop chains   │
│ Source Handling     │ Single source    │ Multi-source       │
│ Validation          │ Minimal          │ Comprehensive      │
│ Transparency        │ Low              │ High               │
│ Complexity Handling │ Limited          │ Excellent          │
└─────────────────────┴──────────────────┴────────────────────┘

LangChain Implementation Notes:
• Multiple specialized LLMs for different tasks (analysis, routing, reasoning)
• Custom query decomposition and dependency resolution
• Source routing based on query type and content
• Iterative retrieval-reasoning loops with context accumulation
• Validation chains for credibility assessment
• Synthesis prompts for coherent answer generation
• Confidence estimation through multi-model assessment

Production Considerations:
• Implement actual vector stores and retrieval systems
• Add caching for repeated queries and sub-queries
• Implement proper source credibility scoring
• Add cost controls for multi-hop operations
• Monitor and limit reasoning depth to prevent loops
• Implement proper error handling and fallbacks
• Add human-in-the-loop for critical decisions
• Track and optimize retrieval quality metrics

Advanced Extensions:
• Adaptive retrieval depth based on query complexity
• Cross-lingual retrieval and reasoning
• Temporal reasoning for time-sensitive queries
• Graph-based knowledge integration
• Active learning from user feedback
• Explanation generation for reasoning steps
• Multi-modal retrieval (text, images, structured data)
    """)


if __name__ == "__main__":
    demonstrate_agentic_rag()

Agentic Design Pattern: Agentic Rag

Pattern 067: Agentic Rag – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Agentic Rag pattern implementation.

Pattern Description

Pattern 067: Agentic RAG (Advanced)

Description: Agentic RAG extends basic Retrieval-Augmented Generation with autonomous decision-making capabilities. Unlike simple RAG that retrieves once and generates, Agentic RAG employs sophisticated query planning, multi-hop reasoning, dynamic retrieval strategies, and source validation. The agent can decompose complex queries, determine when and what to retrieve, route queries to appropriate sources, validate retrieved information, and iteratively refine its understanding through multiple retrieval-reasoning cycles.

This pattern transforms RAG from a passive information lookup into an active research
process where the agent strategically explores knowledge bases, identifies gaps, verifies
information across sources, and constructs well-grounded answers through deliberate
reasoning chains.

Components: 1. Query Analyzer: Decomposes complex questions into sub-queries 2. Query Router: Routes queries to appropriate retrieval sources 3. Multi-Hop Reasoner: Follows chains of reasoning across retrieved documents 4. Source Validator: Validates credibility and relevance of sources 5. Retrieval Planner: Decides what to retrieve and when 6. Answer Synthesizer: Combines information from multiple sources 7. Confidence Estimator: Assesses answer quality and completeness

Architecture: User Query ↓ Query Analyzer → [Sub-queries] ↓ Query Router → [Route to sources] ↓ Retrieval Planner → [Iterative retrieval strategy] ↓ Multi-Hop Reasoner → [Follow reasoning chains] ↓ Source Validator → [Validate & filter] ↓ Answer Synthesizer → [Combine & generate] ↓ Confidence Estimator → [Quality check] ↓ Final Answer (or iterate)

Use Cases: – Complex research questions requiring multi-source synthesis – Scientific literature review and analysis – Legal document analysis with precedent lookup – Technical troubleshooting requiring multiple knowledge sources – Fact-checking and claim verification across sources – Exploratory data analysis with iterative refinement

Advantages: – Handles complex, multi-part questions effectively – Validates information across multiple sources – Provides transparent reasoning chains – Adapts retrieval strategy based on findings – Reduces hallucination through source grounding – Identifies and fills knowledge gaps

LangChain Implementation: Uses ChatOpenAI for LLM operations, custom query planning, routing logic, and iterative retrieval-reasoning loops. Demonstrates query decomposition, multi-source retrieval, validation chains, and answer synthesis.

File Statistics

  • Total Lines: 860
  • Classes: 9
  • Functions: 1
  • Constants: 0

Module Documentation

"""
Pattern 067: Agentic RAG (Advanced)

Description:
    Agentic RAG extends basic Retrieval-Augmented Generation with autonomous decision-making
    capabilities. Unlike simple RAG that retrieves once and generates, Agentic RAG employs
    sophisticated query planning, multi-hop reasoning, dynamic retrieval strategies, and source
    validation. The agent can decompose complex queries, determine when and what to retrieve,
    route queries to appropriate sources, validate retrieved information, and iteratively
    refine its understanding through multiple retrieval-reasoning cycles.

    This pattern transforms RAG from a passive information lookup into an active research
    process where the agent strategically explores knowledge bases, identifies gaps, verifies
    information across sources, and constructs well-grounded answers through deliberate
    reasoning chains.

Components:
    1. Query Analyzer: Decomposes complex questions into sub-queries
    2. Query Router: Routes queries to appropriate retrieval sources
    3. Multi-Hop Reasoner: Follows chains of reasoning across retrieved documents
    4. Source Validator: Validates credibility and relevance of sources
    5. Retrieval Planner: Decides what to retrieve and when
    6. Answer Synthesizer: Combines information from multiple sources
    7. Confidence Estimator: Assesses answer quality and completeness

Architecture:
    ```
    User Query
        ↓
    Query Analyzer → [Sub-queries]
        ↓
    Query Router → [Route to sources]
        ↓
    Retrieval Planner → [Iterative retrieval strategy]
        ↓
    Multi-Hop Reasoner → [Follow reasoning chains]
        ↓
    Source Validator → [Validate & filter]
        ↓
    Answer Synthesizer → [Combine & generate]
        ↓
    Confidence Estimator → [Quality check]
        ↓
    Final Answer (or iterate)
    ```

Use Cases:
    - Complex research questions requiring multi-source synthesis
    - Scientific literature review and analysis
    - Legal document analysis with precedent lookup
    - Technical troubleshooting requiring multiple knowledge sources
    - Fact-checking and claim verification across sources
    - Exploratory data analysis with iterative refinement

Advantages:
    - Handles complex, multi-part questions effectively
    - Validates information across multiple sources
    - Provides transparent reasoning chains
    - Adapts retrieval strategy based on findings
    - Reduces hallucination through source grounding
    - Identifies and fills knowledge gaps

LangChain Implementation:
    Uses ChatOpenAI for LLM operations, custom query planning, routing logic,
    and iterative retrieval-reasoning loops. Demonstrates query decomposition,
    multi-source retrieval, validation chains, and answer synthesis.
"""

Purpose: This module implements the Agentic Rag pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import re
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – re: Regular expression operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: QueryType

Lines 83-89

Inherits from: Enum

Purpose:

Types of queries for routing.

Class: SourceType

Lines 92-98

Inherits from: Enum

Purpose:

Types of retrieval sources.

Class: ConfidenceLevel

Lines 101-106

Inherits from: Enum

Purpose:

Confidence in answer quality.

Class: SubQuery

Lines 110-116

Purpose:

Represents a decomposed sub-query.

Class: RetrievalSource

Lines 120-140

Purpose:

Represents a source for retrieval.

Methods:

retrieve(self, query)

Lines 127-140

Purpose:

Simulate retrieval from this source.

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D749090>

Class: RetrievedDocument

Lines 144-151

Purpose:

Represents a retrieved document with metadata.

Class: ReasoningStep

Lines 155-162

Purpose:

Represents a step in multi-hop reasoning.

Class: AgenticRAGResult

Lines 166-174

Purpose:

Complete result from Agentic RAG.

Class: AgenticRAGAgent

Lines 177-707

Purpose:

Advanced RAG agent with query planning, multi-hop reasoning, and validation.

Methods:

__init__(self, max_hops, min_confidence, temperature)

Lines 182-214

Purpose:

Initialize the Agentic RAG agent.

Args:
    max_hops: Maximum number of retrieval-reasoning hops
    min_confidence: Minimum confidence threshold for answer
    temperature: Temperature for LLM generation

Parameters: – max_hops: Max hops – min_confidence: Min confidence – temperature: Sampling temperature (0-1)

decompose_query(self, query)

Lines 216-301

Purpose:

Decompose complex query into sub-queries.

Args:
    query: The complex user query

Returns:
    List of sub-queries with dependencies

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D7F5DD0>

route_query(self, sub_query)

Lines 303-355

Purpose:

Route sub-query to appropriate sources.

Args:
    sub_query: The sub-query to route

Returns:
    Ranked list of sources to query

Parameters: – sub_query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D73A950>

retrieve_documents(self, sub_query, sources)

Lines 357-389

Purpose:

Retrieve documents from sources.

Args:
    sub_query: The query to retrieve for
    sources: Sources to retrieve from

Returns:
    List of retrieved documents

Parameters: – sub_query: Input query or question – sources: Sources

Returns: <ast.Subscript object at 0x000001A06D606550>

validate_sources(self, documents)

Lines 391-443

Purpose:

Validate credibility and consistency of retrieved documents.

Args:
    documents: Documents to validate

Returns:
    Validated documents with credibility scores

Parameters: – documents: Documents

Returns: <ast.Subscript object at 0x000001A06D65A5D0>

multi_hop_reasoning(self, sub_queries, context)

Lines 445-558

Purpose:

Perform multi-hop reasoning across sub-queries.

Args:
    sub_queries: Sub-queries to reason over
    context: Context from previous reasoning steps

Returns:
    List of reasoning steps

Parameters: – sub_queries: Sub queries – context: Context

Returns: <ast.Subscript object at 0x000001A06D15DDD0>

synthesize_answer(self, query, reasoning_steps)

Lines 560-660

Purpose:

Synthesize final answer from reasoning steps.

Args:
    query: Original user query
    reasoning_steps: All reasoning steps

Returns:
    Tuple of (final answer, confidence level, identified gaps)

Parameters: – query: Input query or question – reasoning_steps: Reasoning steps

Returns: <ast.Subscript object at 0x000001A06D7E6510>

query(self, user_query)

Lines 662-707

Purpose:

Process a query through the full Agentic RAG pipeline.

Args:
    user_query: The user's question

Returns:
    Complete Agentic RAG result with reasoning chain

Parameters: – user_query: Input query or question

Returns: AgenticRAGResult

Functions

demonstrate_agentic_rag()

Lines 710-855

Purpose:

Demonstrate Agentic RAG pattern with various scenarios.

Key Concepts Summary

Pattern Overview

The Agentic Rag pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

  1. Agent Class: Main implementation of the pattern logic
  2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
  3. Tool Integration: Optional tools for extended functionality
  4. Memory Management: State handling across interactions
  5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

Best Practices Demonstrated

1. Environment Configuration

  • ✅ Use .env files for API keys
  • ✅ Never hardcode credentials
  • ✅ Validate environment variables on startup

2. Error Handling

  • ✅ Graceful degradation on failures
  • ✅ Informative error messages
  • ✅ Logging for debugging

3. Prompt Engineering

  • ✅ Clear, specific instructions
  • ✅ Structured output formats
  • ✅ Few-shot examples when beneficial

4. Code Organization

  • ✅ Separation of concerns (class-based design)
  • ✅ Type hints for clarity
  • ✅ Docstrings for documentation

5. Performance Optimization

  • ✅ Appropriate temperature settings
  • ✅ Token budget management
  • ✅ Caching where applicable

LangChain Components Used

  • ChatOpenAI: OpenAI model interface
  • Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

Performance Considerations

Computational Cost

  • LLM Calls: Primary cost factor (based on tokens)
  • Latency: Depends on model and input size
  • Memory: Minimal for most patterns

Optimization Strategies

  1. Prompt Optimization: Minimize tokens while maintaining clarity
  2. Caching: Cache repeated queries/responses
  3. Model Selection: Use smaller models when appropriate
  4. Batch Processing: Group similar requests
  5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

  • GPT-3.5-turbo: ~$0.002 per request
  • GPT-4: ~$0.03 per request
  • GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

  1. Monitoring: Add logging, metrics, tracing
  2. Validation: Input/output schema validation
  3. Rate Limiting: Implement request throttling
  4. Caching: Redis or in-memory cache for responses
  5. Async Support: Non-blocking operations for scalability

Feature Extensions

  1. Multi-Model Support: Switch between providers
  2. Fallback Models: Graceful degradation strategy
  3. Custom Tools: Domain-specific function calling
  4. Advanced Memory: Vector stores for long-term memory
  5. Evaluation: Automated testing and benchmarking

Code Quality

  1. Testing: Unit tests, integration tests
  2. Documentation: API docs, usage examples
  3. Type Safety: Stricter type hints
  4. Configuration: Externalized config files

Related Patterns

  • Pattern 066
  • Pattern 068

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

Summary

The Agentic Rag pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

File: 067_agentic_rag.py
Total Lines: 860
Pattern Category: AI Agent Design Pattern #067

Generated from actual implementation code
For educational and reference purposes

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts