""" 🔮 ADVANCED RAG SYSTEM - Multi-Strategy Retrieval Augmented Generation ====================================================================== Silicon Valley-grade RAG with multiple strategies that auto-adapt to query type. RAG Variants: - Basic RAG: Simple retrieval + generation - Corrective RAG: Self-correction on low confidence - Self-RAG: Reflection and refinement loop - Agentic RAG: Tool-using with reasoning - Multi-RAG: Multiple sources fusion (RRF) - Adaptive RAG: Auto-selects best strategy Features: - Confidence scoring - Source attribution - Retrieval quality assessment - Self-correction loops """ import logging from enum import Enum from typing import Dict, Any, List, Optional, Tuple, Callable from dataclasses import dataclass, field import numpy as np from datetime import datetime logger = logging.getLogger(__name__) # LLM for generation try: from core.llm import chat as llm_chat LLM_AVAILABLE = True except ImportError: LLM_AVAILABLE = False def llm_chat(*args, **kwargs): return "LLM not available" # Vector store for retrieval try: from vector.store_faiss import FaissStore VECTOR_AVAILABLE = True except ImportError: VECTOR_AVAILABLE = False # ============================================================================= # RAG TYPES # ============================================================================= class RAGType(Enum): """Types of RAG strategies.""" BASIC = "basic" # Simple retrieve + generate CORRECTIVE = "corrective" # Self-correct on low quality SELF_RAG = "self_rag" # Reflection loop AGENTIC = "agentic" # Tool-using agent MULTI_RAG = "multi_rag" # Multiple source fusion ADAPTIVE = "adaptive" # Auto-select best @dataclass class RetrievalResult: """Result from retrieval step.""" documents: List[Dict[str, Any]] scores: List[float] source: str query_used: str relevance_score: float = 0.0 @dataclass class RAGResult: """Complete RAG result.""" answer: str sources: List[str] confidence: float rag_type: RAGType retrieval_results: List[RetrievalResult] = field(default_factory=list) reasoning_trace: List[str] = field(default_factory=list) corrections_made: int = 0 # ============================================================================= # RETRIEVAL QUALITY ASSESSMENT # ============================================================================= class RetrievalQualityAssessor: """Assess quality of retrieved documents.""" @staticmethod def assess_relevance(query: str, documents: List[Dict], scores: List[float]) -> float: """ Calculate overall relevance score. Combines: - Semantic similarity scores - Query term overlap - Document freshness (if available) """ if not documents or not scores: return 0.0 # Base score from semantic similarity avg_score = np.mean(scores) if scores else 0.0 # Query term overlap bonus query_terms = set(query.lower().split()) term_overlap = 0.0 for doc in documents[:3]: # Top 3 docs if isinstance(doc, dict): content = doc.get('content', '') or doc.get('text', '') else: content = str(doc) doc_terms = set(content.lower().split()) overlap = len(query_terms & doc_terms) / len(query_terms) if query_terms else 0 term_overlap += overlap term_overlap = term_overlap / min(len(documents), 3) if documents else 0 # Combined score (weighted) final_score = (avg_score * 0.7) + (term_overlap * 0.3) return min(final_score, 1.0) @staticmethod def needs_correction(relevance_score: float, threshold: float = 0.5) -> bool: """Check if retrieval quality is below threshold.""" return relevance_score < threshold # ============================================================================= # BASIC RAG # ============================================================================= class BasicRAG: """ Simple Retrieve-and-Generate RAG. Flow: Query → Retrieve → Generate """ def __init__(self, user_id: str): self.user_id = user_id def retrieve(self, query: str, k: int = 5) -> RetrievalResult: """Retrieve relevant documents.""" try: # Try RAG search from core.rag import rag_search context, sources = rag_search(self.user_id, query, k=k) # Parse into documents docs = [{"content": context, "source": s} for s in sources] if sources else [] scores = [0.8] * len(docs) # Approximate scores return RetrievalResult( documents=docs, scores=scores, source="vector_store", query_used=query, relevance_score=0.8 if docs else 0.0 ) except Exception as e: logger.warning(f"Basic RAG retrieval failed: {e}") return RetrievalResult( documents=[], scores=[], source="none", query_used=query, relevance_score=0.0 ) def generate(self, query: str, context: str) -> str: """Generate answer from context.""" if not context: return "No relevant information found in your data." prompt = f"""Answer the question using ONLY the provided context. CONTEXT: {context} QUESTION: {query} Answer directly and concisely. If the answer isn't in the context, say so.""" return llm_chat(prompt, temperature=0.3, max_tokens=500) def process(self, query: str) -> RAGResult: """Run basic RAG pipeline.""" # Retrieve retrieval = self.retrieve(query) # Build context context = "\n\n".join([ d.get('content', '') for d in retrieval.documents ]) # Generate answer = self.generate(query, context) return RAGResult( answer=answer, sources=[d.get('source', 'Unknown') for d in retrieval.documents], confidence=retrieval.relevance_score, rag_type=RAGType.BASIC, retrieval_results=[retrieval] ) # ============================================================================= # CORRECTIVE RAG # ============================================================================= class CorrectiveRAG(BasicRAG): """ RAG with self-correction on low-quality retrieval. Flow: Query → Retrieve → Assess → (Reformulate if needed) → Generate """ def __init__(self, user_id: str, quality_threshold: float = 0.5): super().__init__(user_id) self.quality_threshold = quality_threshold self.assessor = RetrievalQualityAssessor() def reformulate_query(self, original_query: str, context: str = "") -> str: """Reformulate query for better retrieval.""" prompt = f"""The search query didn't return good results. Original Query: {original_query} Rewrite this query to be more specific and likely to find relevant data. Focus on key entities and metrics. Output ONLY the new query.""" return llm_chat(prompt, temperature=0.3, max_tokens=100) def process(self, query: str, max_corrections: int = 2) -> RAGResult: """Run corrective RAG with potential query reformulation.""" corrections = 0 current_query = query all_retrievals = [] reasoning = [f"Original query: {query}"] while corrections <= max_corrections: # Retrieve retrieval = self.retrieve(current_query) all_retrievals.append(retrieval) # Assess quality relevance = self.assessor.assess_relevance( current_query, retrieval.documents, retrieval.scores ) retrieval.relevance_score = relevance reasoning.append(f"Retrieval {corrections + 1}: relevance={relevance:.2f}") # Check if good enough if relevance >= self.quality_threshold or corrections >= max_corrections: break # Reformulate and retry reasoning.append("Low quality - reformulating query...") current_query = self.reformulate_query(query) reasoning.append(f"New query: {current_query}") corrections += 1 # Build context from best retrieval best_retrieval = max(all_retrievals, key=lambda r: r.relevance_score) context = "\n\n".join([ d.get('content', '') for d in best_retrieval.documents ]) # Generate answer = self.generate(query, context) # Use original query for answer return RAGResult( answer=answer, sources=[d.get('source', 'Unknown') for d in best_retrieval.documents], confidence=best_retrieval.relevance_score, rag_type=RAGType.CORRECTIVE, retrieval_results=all_retrievals, reasoning_trace=reasoning, corrections_made=corrections ) # ============================================================================= # SELF-RAG # ============================================================================= class SelfRAG(BasicRAG): """ RAG with self-reflection and iterative refinement. Flow: Query → Retrieve → Generate → Reflect → (Refine if needed) """ def reflect_on_answer(self, query: str, answer: str, context: str) -> Tuple[bool, str]: """ Reflect on generated answer quality. Returns: (is_good, feedback) """ prompt = f"""Evaluate this answer for accuracy and completeness. QUESTION: {query} CONTEXT USED: {context[:500]}... ANSWER: {answer} Rate the answer: 1. Does it directly answer the question? (Yes/No) 2. Is it based on the context? (Yes/No) 3. Is it complete? (Yes/No) 4. Confidence (0-100%)? Format: RATING: good/needs_improvement FEEDBACK: [one line of feedback]""" response = llm_chat(prompt, temperature=0.2, max_tokens=100) is_good = "good" in response.lower() and "needs_improvement" not in response.lower() return is_good, response def refine_answer(self, query: str, current_answer: str, feedback: str, context: str) -> str: """Refine answer based on feedback.""" prompt = f"""Improve this answer based on feedback. QUESTION: {query} CURRENT ANSWER: {current_answer} FEEDBACK: {feedback} CONTEXT: {context[:500]}... Provide an improved answer that addresses the feedback. Be direct and accurate.""" return llm_chat(prompt, temperature=0.3, max_tokens=500) def process(self, query: str, max_refinements: int = 2) -> RAGResult: """Run Self-RAG with reflection loop.""" reasoning = [f"Query: {query}"] # Retrieve retrieval = self.retrieve(query) context = "\n\n".join([ d.get('content', '') for d in retrieval.documents ]) # Generate initial answer answer = self.generate(query, context) reasoning.append(f"Initial answer generated") # Reflection loop refinements = 0 while refinements < max_refinements: is_good, feedback = self.reflect_on_answer(query, answer, context) reasoning.append(f"Reflection {refinements + 1}: {'good' if is_good else 'needs improvement'}") if is_good: break # Refine answer = self.refine_answer(query, answer, feedback, context) reasoning.append(f"Refined answer") refinements += 1 return RAGResult( answer=answer, sources=[d.get('source', 'Unknown') for d in retrieval.documents], confidence=0.9 if refinements == 0 else 0.8, rag_type=RAGType.SELF_RAG, retrieval_results=[retrieval], reasoning_trace=reasoning, corrections_made=refinements ) # ============================================================================= # AGENTIC RAG # ============================================================================= class AgenticRAG(BasicRAG): """ RAG with tool-using agent capabilities. Flow: Query → Plan → Execute Tools → Retrieve → Generate """ def __init__(self, user_id: str): super().__init__(user_id) self.tools = self._register_tools() def _register_tools(self) -> Dict[str, Callable]: """Register available tools.""" return { "search_data": self._tool_search_data, "calculate": self._tool_calculate, "aggregate": self._tool_aggregate, "filter": self._tool_filter } def _tool_search_data(self, query: str) -> str: """Search user data.""" retrieval = self.retrieve(query) return "\n".join([d.get('content', '')[:200] for d in retrieval.documents[:3]]) def _tool_calculate(self, expression: str) -> str: """Calculate a mathematical expression.""" try: # Safe eval of math expressions allowed = {"abs", "round", "min", "max", "sum"} result = eval(expression, {"__builtins__": {}}, allowed) return str(result) except: return "Calculation error" def _tool_aggregate(self, column: str, operation: str = "sum") -> str: """Aggregate data (placeholder - would use actual DataFrame).""" return f"Aggregation of {column} with {operation}" def _tool_filter(self, condition: str) -> str: """Filter data (placeholder).""" return f"Filtered data with condition: {condition}" def plan_actions(self, query: str) -> List[Dict[str, str]]: """Plan which tools to use.""" prompt = f"""You are an AI agent. Plan actions to answer this question. QUESTION: {query} AVAILABLE TOOLS: - search_data: Search through user's data - calculate: Perform calculations - aggregate: Sum, average, count data - filter: Filter data by condition Output a plan as a list of actions: 1. TOOL: [tool_name], INPUT: [input] 2. TOOL: [tool_name], INPUT: [input] ... Keep it simple - 1-3 actions max.""" response = llm_chat(prompt, temperature=0.2, max_tokens=200) # Parse actions (simplified) actions = [] for line in response.split('\n'): if 'TOOL:' in line and 'INPUT:' in line: try: parts = line.split('TOOL:')[1].split('INPUT:') tool = parts[0].strip().strip(',').lower() inp = parts[1].strip() if len(parts) > 1 else "" if tool in self.tools: actions.append({"tool": tool, "input": inp}) except: pass # Default action if parsing fails if not actions: actions = [{"tool": "search_data", "input": query}] return actions def process(self, query: str) -> RAGResult: """Run Agentic RAG with tool execution.""" reasoning = [f"Query: {query}"] # Plan actions actions = self.plan_actions(query) reasoning.append(f"Planned {len(actions)} actions") # Execute tools tool_results = [] for action in actions: tool_name = action["tool"] tool_input = action["input"] if tool_name in self.tools: result = self.tools[tool_name](tool_input) tool_results.append(f"{tool_name}: {result}") reasoning.append(f"Executed {tool_name}") # Build context from tool results context = "\n\n".join(tool_results) # Generate final answer answer = self.generate(query, context) return RAGResult( answer=answer, sources=["Agent Tools", "User Data"], confidence=0.85, rag_type=RAGType.AGENTIC, reasoning_trace=reasoning ) # ============================================================================= # MULTI-RAG (Multiple Sources with RRF Fusion) # ============================================================================= class MultiRAG(BasicRAG): """ RAG that fuses results from multiple retrieval sources. Uses Reciprocal Rank Fusion (RRF) for merging ranked lists. """ def rrf_fusion(self, ranked_lists: List[List[Dict]], k: int = 60) -> List[Dict]: """ Reciprocal Rank Fusion of multiple ranked lists. RRF Score = Σ 1/(k + rank) """ doc_scores = {} for ranked_list in ranked_lists: for rank, doc in enumerate(ranked_list, 1): doc_id = doc.get('content', '')[:100] # Use content prefix as ID if doc_id not in doc_scores: doc_scores[doc_id] = {"doc": doc, "score": 0} doc_scores[doc_id]["score"] += 1 / (k + rank) # Sort by fused score sorted_docs = sorted( doc_scores.values(), key=lambda x: x["score"], reverse=True ) return [item["doc"] for item in sorted_docs] def process(self, query: str) -> RAGResult: """Run Multi-RAG with source fusion.""" reasoning = [f"Query: {query}"] # Retrieve from multiple sources (in production, these would be different sources) retrieval1 = self.retrieve(query, k=5) retrieval2 = self.retrieve(query + " data analysis", k=5) # Variant query reasoning.append(f"Retrieved from 2 sources") # Fuse results using RRF fused_docs = self.rrf_fusion([ retrieval1.documents, retrieval2.documents ]) reasoning.append(f"Fused to {len(fused_docs)} unique documents") # Build context context = "\n\n".join([ d.get('content', '') for d in fused_docs[:5] ]) # Generate answer = self.generate(query, context) return RAGResult( answer=answer, sources=list(set([d.get('source', 'Unknown') for d in fused_docs[:5]])), confidence=0.9, rag_type=RAGType.MULTI_RAG, retrieval_results=[retrieval1, retrieval2], reasoning_trace=reasoning ) # ============================================================================= # ADAPTIVE RAG (Auto-Selects Best Strategy) # ============================================================================= class AdaptiveRAG: """ Master RAG that automatically selects the best strategy. Selection criteria: - Query complexity - Data availability - Historical performance """ def __init__(self, user_id: str): self.user_id = user_id self.rag_engines = { RAGType.BASIC: BasicRAG(user_id), RAGType.CORRECTIVE: CorrectiveRAG(user_id), RAGType.SELF_RAG: SelfRAG(user_id), RAGType.AGENTIC: AgenticRAG(user_id), RAGType.MULTI_RAG: MultiRAG(user_id) } def select_strategy(self, query: str) -> RAGType: """Select best RAG strategy for query.""" q_lower = query.lower() # Agentic for action/calculation queries if any(kw in q_lower for kw in ['calculate', 'compute', 'find all', 'filter', 'aggregate']): return RAGType.AGENTIC # Multi-RAG for broad queries if any(kw in q_lower for kw in ['overview', 'summary', 'everything about', 'all information']): return RAGType.MULTI_RAG # Self-RAG for complex reasoning if any(kw in q_lower for kw in ['explain', 'why', 'analyze', 'compare', 'relationship']): return RAGType.SELF_RAG # Corrective for precise lookups if any(kw in q_lower for kw in ['exact', 'specific', 'what is the', 'tell me']): return RAGType.CORRECTIVE # Default to basic return RAGType.BASIC def process(self, query: str) -> RAGResult: """Run adaptive RAG with automatic strategy selection.""" # Select strategy strategy = self.select_strategy(query) logger.info(f"🔮 Adaptive RAG selected: {strategy.value}") # Get appropriate engine engine = self.rag_engines.get(strategy, self.rag_engines[RAGType.BASIC]) # Execute result = engine.process(query) # Override type to indicate adaptive selection result.reasoning_trace.insert(0, f"Adaptive RAG selected: {strategy.value}") return result # ============================================================================= # CONVENIENCE FUNCTIONS # ============================================================================= def adaptive_rag_query(user_id: str, query: str) -> Dict[str, Any]: """Quick function for adaptive RAG query.""" rag = AdaptiveRAG(user_id) result = rag.process(query) return { "answer": result.answer, "sources": result.sources, "confidence": result.confidence, "rag_type": result.rag_type.value, "reasoning": result.reasoning_trace } def basic_rag_query(user_id: str, query: str) -> str: """Quick function for basic RAG query.""" rag = BasicRAG(user_id) result = rag.process(query) return result.answer # Module exports __all__ = [ 'RAGType', 'RAGResult', 'RetrievalResult', 'BasicRAG', 'CorrectiveRAG', 'SelfRAG', 'AgenticRAG', 'MultiRAG', 'AdaptiveRAG', 'adaptive_rag_query', 'basic_rag_query' ]