Comprehensive comparison for RAG Framework technology in AI applications

from haystack import Pipeline
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack import Document
from typing import List, Dict, Any
import logging
import os

# Configure logging for production
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductSupportRAG:
    """Production-ready RAG system for product support queries."""
    
    def __init__(self, api_key: str = None):
        """Initialize the RAG pipeline with document store and components."""
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("OpenAI API key must be provided")
        
        # Initialize document store
        self.document_store = InMemoryDocumentStore()
        self.pipeline = None
        
    def index_documents(self, documents: List[Dict[str, str]]) -> None:
        """Index product documentation into the document store."""
        try:
            docs = [
                Document(content=doc["content"], meta=doc.get("meta", {}))
                for doc in documents
            ]
            self.document_store.write_documents(docs)
            logger.info(f"Successfully indexed {len(docs)} documents")
        except Exception as e:
            logger.error(f"Error indexing documents: {str(e)}")
            raise
    
    def build_pipeline(self) -> Pipeline:
        """Build the RAG pipeline with retriever, prompt builder, and generator."""
        try:
            # Define the prompt template
            template = """
            You are a helpful product support assistant. Use the following context to answer the question.
            If you cannot answer based on the context, say so clearly.
            
            Context:
            {% for document in documents %}
            {{ document.content }}
            {% endfor %}
            
            Question: {{ question }}
            
            Answer:
            """
            
            # Initialize components
            retriever = InMemoryBM25Retriever(document_store=self.document_store, top_k=3)
            prompt_builder = PromptBuilder(template=template)
            generator = OpenAIGenerator(
                api_key=self.api_key,
                model="gpt-3.5-turbo",
                generation_kwargs={"max_tokens": 500, "temperature": 0.3}
            )
            
            # Build pipeline
            pipeline = Pipeline()
            pipeline.add_component("retriever", retriever)
            pipeline.add_component("prompt_builder", prompt_builder)
            pipeline.add_component("llm", generator)
            
            # Connect components
            pipeline.connect("retriever.documents", "prompt_builder.documents")
            pipeline.connect("prompt_builder", "llm")
            
            self.pipeline = pipeline
            logger.info("Pipeline built successfully")
            return pipeline
            
        except Exception as e:
            logger.error(f"Error building pipeline: {str(e)}")
            raise
    
    def query(self, question: str) -> Dict[str, Any]:
        """Execute a query through the RAG pipeline with error handling."""
        if not self.pipeline:
            raise RuntimeError("Pipeline not built. Call build_pipeline() first.")
        
        if not question or not question.strip():
            raise ValueError("Question cannot be empty")
        
        try:
            result = self.pipeline.run({
                "retriever": {"query": question},
                "prompt_builder": {"question": question}
            })
            
            response = {
                "answer": result["llm"]["replies"][0] if result["llm"]["replies"] else "No answer generated",
                "retrieved_docs": len(result.get("retriever", {}).get("documents", [])),
                "success": True
            }
            
            logger.info(f"Query processed successfully: {question[:50]}...")
            return response
            
        except Exception as e:
            logger.error(f"Error processing query: {str(e)}")
            return {
                "answer": "An error occurred processing your request.",
                "error": str(e),
                "success": False
            }

# Example usage
if __name__ == "__main__":
    # Sample product documentation
    product_docs = [
        {"content": "Our Premium plan costs $49/month and includes unlimited API calls."},
        {"content": "To reset your password, click 'Forgot Password' on the login page."},
        {"content": "We offer 24/7 customer support via email at [email protected]."}
    ]
    
    # Initialize and setup RAG system
    rag = ProductSupportRAG()
    rag.index_documents(product_docs)
    rag.build_pipeline()
    
    # Query the system
    response = rag.query("How much does the Premium plan cost?")
    print(f"Answer: {response['answer']}")