Comprehensive comparison for Prompt Engineering technology in AI applications

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

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

class CustomerSupportAssistant:
    """Production-ready customer support assistant using Haystack PromptHub patterns."""
    
    def __init__(self, api_key: str):
        """Initialize the assistant with document store and pipeline."""
        if not api_key:
            raise ValueError("OpenAI API key is required")
        
        # Initialize document store with product knowledge base
        self.document_store = InMemoryDocumentStore()
        self._load_knowledge_base()
        
        # Build RAG pipeline with prompt engineering best practices
        self.pipeline = self._build_pipeline(api_key)
        
    def _load_knowledge_base(self):
        """Load product documentation into document store."""
        docs = [
            Document(content="Our return policy allows returns within 30 days of purchase with original receipt."),
            Document(content="Shipping takes 3-5 business days for standard delivery, 1-2 days for express."),
            Document(content="Technical support is available 24/7 via phone at 1-800-SUPPORT or email [email protected]."),
            Document(content="Product warranty covers manufacturing defects for 1 year from purchase date."),
            Document(content="To reset your password, click 'Forgot Password' on the login page and follow email instructions.")
        ]
        self.document_store.write_documents(docs)
        logger.info(f"Loaded {len(docs)} documents into knowledge base")
    
    def _build_pipeline(self, api_key: str) -> Pipeline:
        """Construct RAG pipeline with optimized prompt template."""
        # Define production-grade prompt template with clear instructions
        prompt_template = """
You are a professional customer support assistant. Use the provided context to answer the customer's question accurately and helpfully.

Context Information:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

Customer Question: {{ question }}

Instructions:
1. Answer based on the context provided
2. If the context doesn't contain relevant information, politely state you need to escalate
3. Be concise, friendly, and professional
4. Include specific details from the context when applicable

Answer:
"""
        
        # Initialize pipeline components
        retriever = InMemoryBM25Retriever(document_store=self.document_store, top_k=3)
        prompt_builder = PromptBuilder(template=prompt_template)
        llm = OpenAIGenerator(api_key=api_key, model="gpt-4", generation_kwargs={"temperature": 0.3})
        
        # Assemble pipeline
        pipeline = Pipeline()
        pipeline.add_component("retriever", retriever)
        pipeline.add_component("prompt_builder", prompt_builder)
        pipeline.add_component("llm", llm)
        
        # Connect components
        pipeline.connect("retriever.documents", "prompt_builder.documents")
        pipeline.connect("prompt_builder.prompt", "llm.prompt")
        
        logger.info("Pipeline constructed successfully")
        return pipeline
    
    def answer_question(self, question: str) -> Dict[str, Any]:
        """Process customer question and return answer with metadata."""
        try:
            if not question or len(question.strip()) == 0:
                raise ValueError("Question cannot be empty")
            
            logger.info(f"Processing question: {question[:50]}...")
            
            # Run pipeline
            result = self.pipeline.run({
                "retriever": {"query": question},
                "prompt_builder": {"question": question}
            })
            
            # Extract and validate response
            answer = result["llm"]["replies"][0] if result["llm"]["replies"] else "Unable to generate response"
            
            return {
                "success": True,
                "answer": answer,
                "sources_used": len(result["retriever"]["documents"]),
                "question": question
            }
            
        except Exception as e:
            logger.error(f"Error processing question: {str(e)}")
            return {
                "success": False,
                "error": str(e),
                "answer": "I apologize, but I'm experiencing technical difficulties. Please contact our support team directly."
            }

# Example usage in production API endpoint
if __name__ == "__main__":
    api_key = os.getenv("OPENAI_API_KEY")
    assistant = CustomerSupportAssistant(api_key=api_key)
    
    # Simulate customer queries
    response = assistant.answer_question("What is your return policy?")
    print(f"Success: {response['success']}")
    print(f"Answer: {response['answer']}")