⚙️ Session 1 Advanced: RAG Architecture & Evaluation Mastery¶

# src/advanced_context.py
from typing import List, Dict, Any
import time

def _prepare_enhanced_context(self, search_results: List[Dict]) -> str:
    """Prepare context with quality assessment and source tracking."""
    if not search_results:
        return "No relevant information found in the knowledge base."

    context_parts = []
    total_confidence = 0

    for i, result in enumerate(search_results, 1):
        document = result['document']
        similarity_score = result['similarity_score']
        total_confidence += similarity_score

        source = document.metadata.get("source", "Unknown source")
        chunk_info = document.metadata.get("chunk_id", "N/A")

        # Format context with source attribution
        context_section = f"""
Source {i} (Relevance: {similarity_score}, Source: {source}, Chunk: {chunk_info}):
{document.page_content}
"""
        context_parts.append(context_section)

    # Add confidence assessment
    avg_confidence = total_confidence / len(search_results)
    confidence_note = f"\nContext Confidence: {avg_confidence:.3f} (based on {len(search_results)} sources)"

    return "\n".join(context_parts) + confidence_note

def _create_success_response(self, question: str, response: str,
                            search_results: List[Dict], processing_time: float) -> Dict[str, Any]:
    """Create comprehensive response with production metadata."""
    # Calculate confidence based on search results
    avg_similarity = sum(result['similarity_score'] for result in search_results) / len(search_results)

    # Assess response quality
    response_quality = self._assess_response_quality(response, search_results)

    return {
        "status": "success",
        "answer": response,
        "confidence": round(avg_similarity, 3),
        "quality_score": response_quality,
        "sources": [{
            "content": result['document'].page_content[:300] + "...",
            "metadata": result['document'].metadata,
            "relevance": result['similarity_score'],
            "source": result['document'].metadata.get('source', 'Unknown')
        } for result in search_results],
        "query_metadata": {
            "processing_time_ms": round(processing_time * 1000),
            "sources_used": len(search_results),
            "timestamp": time.time()
        },
        "system_stats": self.query_stats.copy()
    }

def _assess_response_quality(self, response: str, search_results: List[Dict]) -> float:
    """Assess response quality using multiple metrics."""
    quality_score = 1.0

    # Length check
    if len(response.split()) < 10:
        quality_score -= 0.3

    # Source utilization
    sources_mentioned = sum(1 for result in search_results
                          if any(word in response.lower()
                               for word in result['document'].page_content.lower().split()[:20]))
    utilization_ratio = sources_mentioned / len(search_results)
    quality_score *= (0.5 + 0.5 * utilization_ratio)

    # Uncertainty handling
    uncertainty_phrases = ["I don't know", "insufficient information", "not clear"]
    if any(phrase in response for phrase in uncertainty_phrases):
        quality_score *= 1.1  # Bonus for acknowledging uncertainty

    return round(max(0.0, min(1.0, quality_score)), 3)

def hybrid_search(self, query: str, alpha: float = 0.7) -> List[Dict]:
    """Hybrid search combining vector and keyword matching."""
    # Vector similarity search
    vector_results = self.vectorstore.similarity_search_with_score(query, k=10)

    # Simple keyword matching as backup
    all_docs = self._get_all_documents()  # In production, use proper indexing
    keyword_results = self._keyword_search(query, all_docs)

    # Combine results with weighted scoring
    combined_results = self._combine_search_results(
        vector_results, keyword_results, alpha
    )

    return combined_results[:self.config.TOP_K]

def _combine_search_results(self, vector_results, keyword_results, alpha):
    """Combine vector and keyword search with weighted scoring."""
    combined_scores = {}

    # Process vector results (alpha weight)
    for doc, vector_score in vector_results:
        doc_id = doc.metadata.get('source', str(hash(doc.page_content[:100])))
        combined_scores[doc_id] = {
            'document': doc,
            'score': alpha * (1.0 - vector_score),  # Convert distance to similarity
            'source': 'vector'
        }

    # Process keyword results ((1-alpha) weight)
    for doc, keyword_score in keyword_results:
        doc_id = doc.metadata.get('source', str(hash(doc.page_content[:100])))
        if doc_id in combined_scores:
            combined_scores[doc_id]['score'] += (1 - alpha) * keyword_score
            combined_scores[doc_id]['source'] = 'hybrid'
        else:
            combined_scores[doc_id] = {
                'document': doc,
                'score': (1 - alpha) * keyword_score,
                'source': 'keyword'
            }

    # Sort by combined score
    sorted_results = sorted(combined_scores.values(),
                          key=lambda x: x['score'], reverse=True)

    return [{
        'document': result['document'],
        'similarity_score': round(result['score'], 3),
        'search_method': result['source']
    } for result in sorted_results]

# src/evaluation_framework.py
import time
import json
import statistics
from typing import List, Dict, Any
from src.interactive_rag import ProductionRAGInterface

class RAGEvaluationFramework:
    """Comprehensive evaluation for production RAG systems."""

    def __init__(self, rag_interface: ProductionRAGInterface):
        self.rag_interface = rag_interface
        self.evaluation_results = {}

    def run_comprehensive_evaluation(self, test_cases: List[Dict]) -> Dict[str, Any]:
        """Execute full evaluation suite with production metrics."""
        print("🔬 Starting comprehensive RAG evaluation...")

        results = {
            'performance_metrics': self.evaluate_performance(test_cases),
            'retrieval_quality': self.evaluate_retrieval_quality(test_cases),
            'response_quality': self.evaluate_response_quality(test_cases),
            'system_reliability': self.evaluate_system_reliability(test_cases)
        }

        # Calculate overall system score
        results['overall_score'] = self._calculate_overall_score(results)

        return results

def evaluate_performance(self, test_cases: List[Dict]) -> Dict[str, float]:
    """Evaluate system performance metrics."""
    response_times = []
    memory_usage = []

    print("⏱️ Testing performance metrics...")

    for i, case in enumerate(test_cases):
        start_time = time.time()

        # Process query
        result = self.rag_interface.rag_system.process_query(case['question'])

        end_time = time.time()
        response_times.append(end_time - start_time)

        if i % 10 == 0:
            print(f"  Processed {i+1}/{len(test_cases)} test queries")

    return {
        'avg_response_time': statistics.mean(response_times),
        'median_response_time': statistics.median(response_times),
        'p95_response_time': sorted(response_times)[int(0.95 * len(response_times))],
        'min_response_time': min(response_times),
        'max_response_time': max(response_times)
    }

def evaluate_retrieval_quality(self, test_cases: List[Dict]) -> Dict[str, float]:
    """Evaluate retrieval accuracy using ground truth data."""
    precision_scores = []
    recall_scores = []

    print("Testing retrieval quality...")

    for case in test_cases:
        if 'expected_sources' not in case:
            continue

        question = case['question']
        expected_sources = set(case['expected_sources'])

        # Get RAG system response
        result = self.rag_interface.rag_system.process_query(question)

        if result['status'] != 'success':
            precision_scores.append(0.0)
            recall_scores.append(0.0)
            continue

        # Extract retrieved sources
        retrieved_sources = set([
            source['source'] for source in result['sources']
        ])

        # Calculate precision and recall
        if retrieved_sources:
            intersection = expected_sources & retrieved_sources
            precision = len(intersection) / len(retrieved_sources)
            recall = len(intersection) / len(expected_sources) if expected_sources else 0
        else:
            precision = recall = 0.0

        precision_scores.append(precision)
        recall_scores.append(recall)

    avg_precision = statistics.mean(precision_scores) if precision_scores else 0
    avg_recall = statistics.mean(recall_scores) if recall_scores else 0
    f1_score = (2 * avg_precision * avg_recall) / (avg_precision + avg_recall) if (avg_precision + avg_recall) > 0 else 0

    return {
        'precision': avg_precision,
        'recall': avg_recall,
        'f1_score': f1_score,
        'total_evaluated': len(precision_scores)
    }

def evaluate_response_quality(self, test_cases: List[Dict]) -> Dict[str, float]:
    """Evaluate response quality using multiple criteria."""
    quality_scores = []
    coherence_scores = []
    source_usage_scores = []

    print("📝 Testing response quality...")

    for case in test_cases:
        result = self.rag_interface.rag_system.process_query(case['question'])

        if result['status'] != 'success':
            quality_scores.append(0.0)
            continue

        answer = result['answer']
        sources = result['sources']

        # Quality assessment
        quality_score = self._assess_answer_quality(answer, case.get('expected_answer', ''))
        quality_scores.append(quality_score)

        # Coherence assessment
        coherence = self._assess_coherence(answer)
        coherence_scores.append(coherence)

        # Source usage assessment
        source_usage = self._assess_source_usage(answer, sources)
        source_usage_scores.append(source_usage)

    return {
        'avg_quality_score': statistics.mean(quality_scores) if quality_scores else 0,
        'avg_coherence_score': statistics.mean(coherence_scores) if coherence_scores else 0,
        'avg_source_usage': statistics.mean(source_usage_scores) if source_usage_scores else 0
    }

def _assess_answer_quality(self, answer: str, expected: str) -> float:
    """Assess answer quality against expected response."""
    if not answer or len(answer.strip()) < 10:
        return 0.2

    quality_score = 0.5  # Base score for valid response

    # Length appropriateness
    word_count = len(answer.split())
    if 20 <= word_count <= 200:
        quality_score += 0.2

    # Uncertainty handling
    if any(phrase in answer.lower() for phrase in
           ['not sure', 'unclear', 'insufficient information', "don't know"]):
        quality_score += 0.2

    # Specificity bonus
    if any(char.isdigit() for char in answer) or any(word in answer.lower()
           for word in ['specific', 'exactly', 'precisely']):
        quality_score += 0.1

    return min(1.0, quality_score)

# src/interactive_rag.py - Advanced Interface
from src.rag_system import ProductionRAGSystem
from src.document_loader import ProductionDocumentLoader
from src.text_splitter import AdvancedTextSplitter
from src.config import RAGConfig
import json

class ProductionRAGInterface:
    """Production RAG interface with comprehensive monitoring."""

    def __init__(self):
        self.config = RAGConfig()
        self.rag_system = ProductionRAGSystem(self.config)
        self.document_loader = ProductionDocumentLoader()
        self.text_splitter = AdvancedTextSplitter(
            chunk_size=self.config.CHUNK_SIZE,
            chunk_overlap=self.config.CHUNK_OVERLAP
        )
        self.session_stats = {'queries': 0, 'documents_loaded': 0}

    def load_and_index_documents(self, sources: List[str]) -> Dict[str, Any]:
        """Load documents with comprehensive monitoring."""
        print("🔄 Starting document processing pipeline...")

        # Load documents
        documents = self.document_loader.load_batch_with_monitoring(sources)
        if not documents:
            return {"status": "error", "message": "No documents loaded"}

        # Chunk documents
        print("🔪 Processing document chunks...")
        chunks = self.text_splitter.hybrid_chunk(documents)

        # Index in vector store
        print("📚 Indexing in vector database...")
        indexing_results = self.rag_system.vector_store.add_documents_batch(chunks)

        # Update session statistics
        self.session_stats['documents_loaded'] += len(documents)

        return {
            "status": "success",
            "documents_processed": len(documents),
            "chunks_created": len(chunks),
            "indexing_results": indexing_results,
            "loader_stats": self.document_loader.load_stats,
            "chunking_stats": self.text_splitter.chunking_stats
        }

def start_enhanced_chat(self):
    """Production chat interface with comprehensive features."""
    print("=" * 70)
    print("🤖 Production RAG System - Enterprise Edition")
    print("=" * 70)
    print("Features: Advanced chunking, hybrid search, quality monitoring")
    print("Commands: 'quit', 'stats', 'help', or ask any question")
    print("-" * 70)

    while True:
        try:
            user_input = input("\n📝 Your question: ").strip()

            if user_input.lower() in ['quit', 'exit']:
                self._display_session_summary()
                break
            elif user_input.lower() == 'stats':
                self._display_system_stats()
                continue
            elif user_input.lower() == 'help':
                self._display_help()
                continue
            elif not user_input:
                print("Please enter a question or command.")
                continue

            # Process query with full monitoring
            print("\n🔍 Processing query with advanced pipeline...")
            result = self.rag_system.process_query(user_input)
            self.session_stats['queries'] += 1

            self._display_enhanced_result(result)

        except KeyboardInterrupt:
            print("\n👋 Session terminated by user")
            break
        except Exception as e:
            print(f"❌ System error: {str(e)}")

def _display_enhanced_result(self, result: Dict[str, Any]):
    """Display results with comprehensive information."""
    if result['status'] == 'success':
        print(f"\n🤖 **Answer** (Confidence: {result['confidence']}, Quality: {result['quality_score']})")
        print("-" * 50)
        print(result['answer'])

        print(f"\n📚 **Sources** ({result['query_metadata']['sources_used']} documents)")
        print("-" * 50)
        for i, source in enumerate(result['sources'], 1):
            print(f"{i}. Relevance: {source['relevance']:.3f}")
            print(f"   Source: {source['source']}")
            print(f"   Preview: {source['content']}")
            print()

        print(f"⏱️ **Performance**: {result['query_metadata']['processing_time_ms']}ms")
    else:
        print(f"\n❌ **Error**: {result['message']}")

def create_evaluation_test_cases() -> List[Dict]:
    """Create comprehensive test cases for RAG evaluation."""
    return [
        {
            'question': 'What is artificial intelligence?',
            'expected_sources': ['https://en.wikipedia.org/wiki/Artificial_intelligence'],
            'category': 'definitional',
            'difficulty': 'easy'
        },
        {
            'question': 'How do neural networks learn from data?',
            'expected_sources': ['https://en.wikipedia.org/wiki/Machine_learning'],
            'category': 'technical',
            'difficulty': 'medium'
        },
        {
            'question': 'What are the ethical implications of AI in healthcare?',
            'expected_sources': ['https://en.wikipedia.org/wiki/Artificial_intelligence'],
            'category': 'analytical',
            'difficulty': 'hard'
        }
        # Add more test cases for comprehensive evaluation
    ]

def run_production_evaluation():
    """Execute production evaluation suite."""
    # Initialize RAG system
    rag = ProductionRAGInterface()

    # Sample documents for testing
    test_sources = [
        "https://en.wikipedia.org/wiki/Artificial_intelligence",
        "https://en.wikipedia.org/wiki/Machine_learning",
        "https://en.wikipedia.org/wiki/Natural_language_processing"
    ]

    # Load and index documents
    print("Setting up test environment...")
    indexing_result = rag.load_and_index_documents(test_sources)

    if indexing_result['status'] != 'success':
        print("Failed to set up test environment")
        return

    # Create evaluation framework
    evaluator = RAGEvaluationFramework(rag)

    # Run comprehensive evaluation
    test_cases = create_evaluation_test_cases()
    results = evaluator.run_comprehensive_evaluation(test_cases)

    # Display results
    print("\n" + "="*60)
    print("PRODUCTION RAG EVALUATION RESULTS")
    print("="*60)

    print(f"Overall System Score: {results['overall_score']:.3f}")
    print(f"\nPerformance Metrics:")
    print(f"  Average Response Time: {results['performance_metrics']['avg_response_time']:.3f}s")
    print(f"  95th Percentile: {results['performance_metrics']['p95_response_time']:.3f}s")

    print(f"\nRetrieval Quality:")
    print(f"  Precision: {results['retrieval_quality']['precision']:.3f}")
    print(f"  Recall: {results['retrieval_quality']['recall']:.3f}")
    print(f"  F1 Score: {results['retrieval_quality']['f1_score']:.3f}")

    print(f"\nResponse Quality:")
    print(f"  Quality Score: {results['response_quality']['avg_quality_score']:.3f}")
    print(f"  Source Usage: {results['response_quality']['avg_source_usage']:.3f}")

    return results

if __name__ == "__main__":
    results = run_production_evaluation()

# Domain specialization template for RAG systems
class DomainSpecificRAG(ProductionRAGSystem):
    """Specialized RAG system for [YOUR DOMAIN]."""

    def __init__(self, config: RAGConfig):
        super().__init__(config)
        self.domain_config = self._setup_domain_config()

    def _setup_domain_config(self) -> Dict[str, Any]:
        """Configure domain-specific settings."""
        return {
            'chunk_strategy': 'semantic',  # or 'hierarchical', 'hybrid'
            'quality_threshold': 0.8,     # Higher for critical domains
            'source_validation': True,     # Enable for medical/legal
            'terminology_boost': ['domain', 'specific', 'terms']
        }

    def _create_domain_prompt(self) -> PromptTemplate:
        """Create domain-specialized prompt template."""
        # Customize based on your chosen domain
        pass

    def process_domain_query(self, question: str) -> Dict[str, Any]:
        """Domain-specific query processing with specialized validation."""
        # Add domain-specific preprocessing
        # Apply domain validation rules
        # Return enhanced results
        pass