Session 1: Basic RAG Implementation¶

# requirements.txt - 2024 Production Stack

langchain==0.1.0
langchain-community==0.0.13
langchain-openai==0.0.5
chromadb==0.4.22
tiktoken==0.5.2
python-dotenv==1.0.0
beautifulsoup4==4.12.2
requests==2.31.0
numpy==1.24.3

# src/config.py - Production Configuration Pattern

import os
from dotenv import load_dotenv

load_dotenv()

class RAGConfig:
    # API Configuration
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

    # Chunking Strategy (2024 Best Practices)
    CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", 1000))
    CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", 200))
    TOP_K = int(os.getenv("TOP_K", 5))

    # Vector Database Settings
    VECTOR_DB_PATH = "./chroma_db"
    COLLECTION_NAME = "rag_documents"

    # Model Configuration
    EMBEDDING_MODEL = "text-embedding-ada-002"
    LLM_MODEL = "gpt-3.5-turbo"

# tests/test_setup.py - Environment Validation

from src.config import RAGConfig
import chromadb
from langchain.embeddings import OpenAIEmbeddings

def test_environment():
    """Validate production environment setup."""
    config = RAGConfig()

    # Test API key configuration
    assert config.OPENAI_API_KEY, "OpenAI API key not configured"

    # Test ChromaDB initialization
    client = chromadb.PersistentClient(path=config.VECTOR_DB_PATH)
    print("✅ ChromaDB initialized successfully")

    # Test embedding model access
    embeddings = OpenAIEmbeddings(openai_api_key=config.OPENAI_API_KEY)
    test_embedding = embeddings.embed_query("test query")
    assert len(test_embedding) > 0, "Embedding generation failed"
    print("✅ Embedding model accessible")

    print("🎉 Environment setup validated successfully!")

if __name__ == "__main__":
    test_environment()

# src/document_loader.py - Production Document Loader

from typing import List, Dict, Any
import requests
from bs4 import BeautifulSoup
from langchain.schema import Document
import os

class ProductionDocumentLoader:
    """Production-grade document loader with error handling and monitoring."""

    def __init__(self):
        self.supported_formats = ['.txt', '.md', '.html', '.pdf']
        self.load_stats = {'success': 0, 'errors': 0}
        print(f"Loader initialized for formats: {self.supported_formats}")

    def load_from_file(self, file_path: str) -> List[Document]:
        """Load document with production-grade error handling."""
        if not os.path.exists(file_path):
            self.load_stats['errors'] += 1
            raise FileNotFoundError(f"File not found: {file_path}")

        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                content = file.read()

            self.load_stats['success'] += 1
            print(f"Loaded {len(content)} characters from {file_path}")

            return [Document(
                page_content=content,
                metadata={
                    "source": file_path,
                    "type": "file",
                    "char_count": len(content),
                    "load_timestamp": self._get_timestamp()
                }
            )]
        except Exception as e:
            self.load_stats['errors'] += 1
            print(f"Error loading {file_path}: {e}")
            return []

    def load_from_url(self, url: str) -> List[Document]:
        """Extract clean web content for RAG processing."""
        try:
            response = requests.get(url, timeout=30, headers={
                'User-Agent': 'RAG-System-1.0'
            })
            response.raise_for_status()

            soup = BeautifulSoup(response.content, 'html.parser')
            return self._extract_content(soup, url)

        except Exception as e:
            self.load_stats['errors'] += 1
            print(f"Web loading error {url}: {str(e)}")
            return []

    def _extract_content(self, soup, url: str) -> List[Document]:
        """Advanced HTML content cleaning pipeline."""
        # Remove noise elements
        for element in soup(["script", "style", "nav", "header", "footer", "aside"]):
            element.extract()

        # Extract and normalize text
        text = soup.get_text()
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines 
                 for phrase in line.split("  "))
        clean_text = ' '.join(chunk for chunk in chunks if chunk)

        self.load_stats['success'] += 1
        print(f"Extracted {len(clean_text)} characters from {url}")

        return [Document(
            page_content=clean_text,
            metadata={
                "source": url,
                "type": "web",
                "char_count": len(clean_text),
                "load_timestamp": self._get_timestamp()
            }
        )]

    def load_batch_with_monitoring(self, sources: List[str]) -> List[Document]:
        """Production batch processing with comprehensive monitoring."""
        all_documents = []
        self.load_stats = {'success': 0, 'errors': 0, 'skipped': 0}

        print(f"🔄 Processing {len(sources)} document sources...")

        for i, source in enumerate(sources, 1):
            print(f"[{i}/{len(sources)}] Processing: {source[:60]}...")

            try:
                # Route to appropriate loader
                if source.startswith(('http://', 'https://')):
                    docs = self.load_from_url(source)
                elif os.path.isfile(source):
                    docs = self.load_from_file(source)
                else:
                    print(f"  ⚠️ Unsupported source type: {source}")
                    self.load_stats['skipped'] += 1
                    continue

                # Validate and collect results
                if docs:
                    all_documents.extend(docs)
                    print(f"  ✅ Loaded {len(docs)} documents")
                else:
                    print(f"  ⚠️ No documents loaded from {source}")

            except Exception as e:
                self.load_stats['errors'] += 1
                print(f"  ❌ Error processing {source}: {e}")

        # Production monitoring report
        total_processed = len(sources)
        success_rate = (self.load_stats['success'] / total_processed) * 100

        print(f"\n📊 Batch Processing Report:")
        print(f"   Success Rate: {success_rate:.1f}% ({self.load_stats['success']}/{total_processed})")
        print(f"   Errors: {self.load_stats['errors']}")
        print(f"   Skipped: {self.load_stats['skipped']}")
        print(f"   Total Documents: {len(all_documents)}")

        return all_documents

# src/text_splitter.py - Advanced Chunking System

from typing import List
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
import tiktoken

class AdvancedTextSplitter:
    """2024 production-grade chunking with multiple strategies."""

    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
        self.chunking_stats = {'total_chunks': 0, 'avg_size': 0}

    def count_tokens(self, text: str) -> int:
        """Accurate token counting using tiktoken."""
        return len(self.encoding.encode(text))

    def recursive_split(self, documents: List[Document]) -> List[Document]:
        """Enhanced recursive splitting with token awareness."""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            length_function=self.count_tokens,
            separators=["\n\n", "\n", ". ", " ", ""]  # Hierarchical separators
        )

        split_docs = text_splitter.split_documents(documents)

        # Add enhanced metadata tracking
        total_tokens = 0
        for i, doc in enumerate(split_docs):
            token_count = self.count_tokens(doc.page_content)
            total_tokens += token_count

            doc.metadata.update({
                "chunk_id": i,
                "token_count": token_count,
                "chunk_method": "recursive",
                "overlap_size": self.chunk_overlap
            })

        # Update statistics
        self.chunking_stats['total_chunks'] = len(split_docs)
        self.chunking_stats['avg_size'] = total_tokens / len(split_docs) if split_docs else 0

        print(f"Recursive chunking: {len(split_docs)} chunks, avg {self.chunking_stats['avg_size']:.0f} tokens")
        return split_docs

    def semantic_split(self, documents: List[Document]) -> List[Document]:
        """Context-aware semantic chunking (2024 best practice)."""
        semantic_chunks = []

        for doc in documents:
            content = doc.page_content

            # Split by semantic boundaries (paragraphs)
            paragraphs = content.split('\n\n')

            current_chunk = ""
            current_tokens = 0

            for paragraph in paragraphs:
                paragraph_tokens = self.count_tokens(paragraph)

                # Check if adding paragraph exceeds limit
                if current_tokens + paragraph_tokens > self.chunk_size and current_chunk:
                    # Save current chunk
                    semantic_chunks.append(self._create_semantic_chunk(
                        current_chunk.strip(), doc.metadata, current_tokens, len(semantic_chunks)
                    ))

                    # Start new chunk with overlap
                    overlap_text = self._get_overlap_text(current_chunk)
                    current_chunk = overlap_text + paragraph
                    current_tokens = self.count_tokens(current_chunk)
                else:
                    # Add paragraph to current chunk
                    separator = "\n\n" if current_chunk else ""
                    current_chunk += separator + paragraph
                    current_tokens += paragraph_tokens

            # Add final chunk
            if current_chunk:
                semantic_chunks.append(self._create_semantic_chunk(
                    current_chunk.strip(), doc.metadata, current_tokens, len(semantic_chunks)
                ))

        print(f"Semantic chunking: {len(semantic_chunks)} chunks with context preservation")
        return semantic_chunks

    def hybrid_chunk(self, documents: List[Document]) -> List[Document]:
        """Hybrid chunking: semantic when possible, recursive fallback."""
        hybrid_chunks = []

        for doc in documents:
            # Try semantic chunking first
            semantic_result = self._attempt_semantic_chunking(doc)

            if self._validate_chunk_quality(semantic_result):
                hybrid_chunks.extend(semantic_result)
                print(f"Used semantic chunking for {doc.metadata.get('source', 'unknown')}")
            else:
                # Fallback to recursive chunking
                recursive_result = self.recursive_split([doc])
                hybrid_chunks.extend(recursive_result)
                print(f"Used recursive chunking for {doc.metadata.get('source', 'unknown')}")

        return hybrid_chunks

    def _validate_chunk_quality(self, chunks: List[Document]) -> bool:
        """Validate chunk quality using 2024 best practices."""
        if not chunks:
            return False

        # Check average chunk size
        avg_size = sum(self.count_tokens(chunk.page_content) for chunk in chunks) / len(chunks)

        # Quality thresholds based on research
        return (
            200 <= avg_size <= 1500 and  # Optimal size range
            len(chunks) > 0 and          # Must produce chunks
            all(self.count_tokens(chunk.page_content) >= 100 for chunk in chunks)  # Min viable size
        )

# src/vector_store.py - Enterprise Vector Storage

from typing import List, Optional, Dict, Any
import chromadb
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from src.config import RAGConfig
import time

class ProductionVectorStore:
    """Enterprise-grade vector storage with monitoring."""

    def __init__(self, config: RAGConfig):
        self.config = config
        self.embeddings = OpenAIEmbeddings(
            openai_api_key=config.OPENAI_API_KEY,
            model=config.EMBEDDING_MODEL
        )
        self.vectorstore = None
        self.operation_stats = {'adds': 0, 'searches': 0, 'errors': 0}
        self._initialize_store()

    def _initialize_store(self):
        """Initialize ChromaDB with production settings."""
        try:
            # Attempt to load existing collection
            self.vectorstore = Chroma(
                persist_directory=self.config.VECTOR_DB_PATH,
                embedding_function=self.embeddings,
                collection_name=self.config.COLLECTION_NAME
            )

            # Verify collection exists and get stats
            collection = self.vectorstore._collection
            doc_count = collection.count()
            print(f"Loaded existing vector store with {doc_count} documents")

        except Exception as e:
            print(f"Creating new vector store: {e}")
            self.vectorstore = Chroma(
                persist_directory=self.config.VECTOR_DB_PATH,
                embedding_function=self.embeddings,
                collection_name=self.config.COLLECTION_NAME
            )
            print("New vector store initialized")

    def add_documents_batch(self, documents: List[Document], batch_size: int = 100) -> Dict[str, Any]:
        """Production batch indexing with performance optimization."""
        if not documents:
            return {"indexed": 0, "errors": 0, "time_taken": 0}

        start_time = time.time()
        indexed_count = 0
        error_count = 0

        print(f"Indexing {len(documents)} documents in batches of {batch_size}...")

        # Process in batches for memory efficiency
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            batch_ids = [f"doc_{i+j}_{hash(doc.page_content[:100])}" 
                        for j, doc in enumerate(batch)]

            try:
                self.vectorstore.add_documents(batch, ids=batch_ids)
                indexed_count += len(batch)
                self.operation_stats['adds'] += len(batch)
                print(f"  Indexed batch {i//batch_size + 1}: {len(batch)} documents")

            except Exception as e:
                error_count += len(batch)
                self.operation_stats['errors'] += 1
                print(f"  Error in batch {i//batch_size + 1}: {e}")

        # Persist changes
        self.vectorstore.persist()

        end_time = time.time()
        processing_time = end_time - start_time

        results = {
            "indexed": indexed_count,
            "errors": error_count,
            "time_taken": processing_time,
            "docs_per_second": indexed_count / processing_time if processing_time > 0 else 0
        }

        print(f"Indexing complete: {indexed_count} docs, {processing_time:.1f}s, {results['docs_per_second']:.1f} docs/sec")
        return results

    def advanced_search(self, query: str, filters: Optional[Dict] = None, 
                       k: Optional[int] = None, score_threshold: float = 0.7) -> List[Dict]:
        """Production search with filtering and quality controls."""
        k = k or self.config.TOP_K
        start_time = time.time()

        try:
            # Perform search with metadata filtering
            if filters:
                results = self.vectorstore.similarity_search_with_score(
                    query=query,
                    k=k * 2,  # Get more results for filtering
                    filter=filters
                )
            else:
                results = self.vectorstore.similarity_search_with_score(
                    query=query,
                    k=k
                )

            # Apply score threshold filtering
            filtered_results = [
                (doc, score) for doc, score in results 
                if (1.0 - score) >= score_threshold  # Convert distance to similarity
            ]

            # Limit to requested count
            final_results = filtered_results[:k]

            self.operation_stats['searches'] += 1
            search_time = time.time() - start_time

            return [{
                'document': doc,
                'similarity_score': round(1.0 - score, 3),
                'metadata': doc.metadata,
                'content_preview': doc.page_content[:200] + "..."
            } for doc, score in final_results]

        except Exception as e:
            self.operation_stats['errors'] += 1
            print(f"Search error: {e}")
            return []

    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/rag_system.py - Production RAG System

from typing import List, Dict, Any, Optional
from langchain.chat_models import ChatOpenAI
from langchain.schema import Document
from langchain.prompts import PromptTemplate
from src.config import RAGConfig
from src.vector_store import ProductionVectorStore
import time
import json

class ProductionRAGSystem:
    """Enterprise-grade RAG system with comprehensive monitoring."""

    def __init__(self, config: RAGConfig):
        self.config = config
        self.vector_store = ProductionVectorStore(config)

        # Initialize LLM with production settings
        self.llm = ChatOpenAI(
            openai_api_key=config.OPENAI_API_KEY,
            model_name=config.LLM_MODEL,
            temperature=0.2,  # Lower temperature for consistent responses
            request_timeout=30  # Production timeout
        )

        self.prompt_template = self._create_rag_prompt()
        self.query_stats = {'total_queries': 0, 'avg_response_time': 0, 'errors': 0}

    def _create_rag_prompt(self) -> PromptTemplate:
        """Production-grade RAG prompt with error handling."""
        template = """You are a helpful AI assistant providing accurate information based on retrieved context.

IMPORTANT GUIDELINES:
- Answer ONLY based on the provided context
- If the context doesn't contain sufficient information, clearly state this
- Provide source references when possible
- Be concise but complete
- If you're uncertain, express the level of confidence

Context Sources:
{context}

User Question: {question}

Instructions: Analyze the provided context carefully and provide an accurate, helpful response. If the context is insufficient, explain what additional information would be needed.

Response:"""

        return PromptTemplate(
            input_variables=["context", "question"],
            template=template
        )

    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 process_query(self, question: str, filters: Optional[Dict] = None, 
                     k: Optional[int] = None) -> Dict[str, Any]:
        """Complete RAG pipeline with comprehensive monitoring."""
        start_time = time.time()

        try:
            # Step 1: Input validation
            if not question or len(question.strip()) < 3:
                return self._create_error_response("Question too short or empty")

            # Step 2: Retrieve relevant documents
            search_results = self.vector_store.advanced_search(
                query=question,
                filters=filters,
                k=k or self.config.TOP_K,
                score_threshold=0.6  # Production quality threshold
            )

            if not search_results:
                return self._create_no_results_response(question)

            # Step 3: Prepare context
            context = self._prepare_enhanced_context(search_results)

            # Step 4: Generate response
            prompt = self.prompt_template.format(
                context=context,
                question=question
            )

            response = self.llm.predict(prompt)

            # Step 5: Calculate metrics
            processing_time = time.time() - start_time
            self._update_query_stats(processing_time)

            return self._create_success_response(
                question, response, search_results, processing_time
            )

        except Exception as e:
            self.query_stats['errors'] += 1
            return self._create_error_response(f"Processing error: {str(e)}")

    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)

# src/interactive_rag.py - Production 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 _display_system_stats(self):
        """Display comprehensive system statistics."""
        print("\n📊 **System Statistics**")
        print("-" * 40)
        print(f"Session Queries: {self.session_stats['queries']}")
        print(f"Documents Loaded: {self.session_stats['documents_loaded']}")
        print(f"Vector Store Stats: {self.rag_system.vector_store.operation_stats}")
        print(f"Query Stats: {self.rag_system.query_stats}")

# tests/evaluation_framework.py - Production Testing

import time
import json
from typing import List, Dict, Any
from src.interactive_rag import ProductionRAGInterface
import statistics

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)

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()

# Your Implementation Template

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