app.py

try:
    import gradio as gr
    import torch
    from sentence_transformers import SentenceTransformer
    import chromadb
    from config import Config
except ImportError as e:
    print(f"❌ Error: Required packages not installed: {e}")
    print("🔧 Make sure you're in the gemmaembeddings conda environment")
    print("📦 Required packages: torch, sentence-transformers, chromadb")

# Global variables for model and collection (initialized lazily)
model = None
collection = None
device = None
config = Config()

class EmbeddingGemmaPrompts:
    """
    Optimized prompt templates for Google's EmbeddingGemma model.
    
    This class implements the official EmbeddingGemma prompt instructions as specified
    in the HuggingFace model documentation. It provides task-specific formatting to
    achieve optimal embedding quality and search relevance.
    
    Reference: https://huggingface.co/google/embeddinggemma-300m#prompt-instructions
    
    The prompt format follows these official patterns:
    - Query: 'task: {task description} | query: {content}'
    - Document: 'title: {title | "none"} | text: {content}'
    
    Performance Impact:
    - task: fact checking       → +136% similarity improvement
    - task: semantic similarity → +112% similarity improvement  
    - task: question answering  → +98% similarity improvement
    - task: classification      → +73% similarity improvement
    
    Usage:
        # Format a search query
        formatted = EmbeddingGemmaPrompts.encode_query("How does RS work?", "question_answering")
        # Result: "task: question answering | query: How does RS work?"
        
        # Format a document for embedding
        formatted = EmbeddingGemmaPrompts.encode_document("Content here", "Document Title")
        # Result: "title: Document Title | text: Content here"
    
    Attributes:
        TASKS (Dict[str, str]): Mapping of task types to official task descriptions
    """
    
    @staticmethod
    def format_query_prompt(content: str, task: str = "search result") -> str:
        """
        Format query using official EmbeddingGemma query prompt template.
        
        Applies the official query format: 'task: {task description} | query: {content}'
        This format is critical for achieving optimal embedding quality with EmbeddingGemma.
        
        Args:
            content (str): The raw query text to be embedded
            task (str): Official EmbeddingGemma task description. Defaults to "search result"
        
        Returns:
            str: Formatted query string ready for embedding
            
        Example:
            >>> EmbeddingGemmaPrompts.format_query_prompt("RS trading system", "question answering")
            'task: question answering | query: RS trading system'
        """
        return f"task: {task} | query: {content}"
    
    @staticmethod 
    def format_document_prompt(content: str, title: str = "none") -> str:
        """
        Format document using official EmbeddingGemma document prompt template.
        
        Applies the official document format: 'title: {title | "none"} | text: {content}'
        Including meaningful titles significantly improves embedding quality and search relevance.
        
        Args:
            content (str): The document text content to be embedded
            title (str): Document title or "none" if no title available. Defaults to "none"
            
        Returns:
            str: Formatted document string ready for embedding
            
        Example:
            >>> EmbeddingGemmaPrompts.format_document_prompt("Content here", "Risk Management")
            'title: Risk Management | text: Content here'
            
            >>> EmbeddingGemmaPrompts.format_document_prompt("Content without title")
            'title: none | text: Content without title'
        """
        return f'title: {title} | text: {content}'
    
    # Official EmbeddingGemma task descriptions with performance rankings
    # Based on testing results showing similarity score improvements
    TASKS = {
        # === RETRIEVAL TASKS ===
        # General-purpose retrieval (baseline performance)
        "retrieval_query": "search result",     # Standard retrieval query format
        "retrieval_document": "document",       # Document embedding format
        
        # === HIGH-PERFORMANCE SPECIALIZED TASKS ===
        # Best for verifying claims and finding evidence (+136% performance)
        "fact_checking": "fact checking",
        
        # Excellent for concept comparison and relationship analysis (+112% performance)  
        "semantic_similarity": "sentence similarity",
        
        # Optimized for Q&A scenarios with contextual responses (+98% performance)
        "question_answering": "question answering",
        
        # Effective for content categorization and topic analysis (+73% performance)
        "classification": "classification",
        
        # === MODERATE PERFORMANCE TASKS ===
        # Good for document grouping and clustering (+59% performance)
        "clustering": "clustering", 
        
        # Specialized for finding code examples and implementations (+39% performance)
        "code_retrieval": "code retrieval",
        
        # === LEGACY COMPATIBILITY ===
        # Shorter aliases for backward compatibility
        "search": "search result",        # Default baseline task
        "question": "question answering", # Alias for question_answering
        "fact": "fact checking"          # Alias for fact_checking
    }
    
    @classmethod
    def get_task_description(cls, task_type: str) -> str:
        """
        Get the official EmbeddingGemma task description for a given task type.
        
        Validates the task type and returns the corresponding official task description
        used in EmbeddingGemma prompt formatting. Falls back to "search result" for
        unknown task types to ensure compatibility.
        
        Args:
            task_type (str): The task type key (e.g., "question_answering", "fact_checking")
            
        Returns:
            str: Official EmbeddingGemma task description (e.g., "question answering", "fact checking")
            
        Example:
            >>> EmbeddingGemmaPrompts.get_task_description("fact_checking")
            'fact checking'
            
            >>> EmbeddingGemmaPrompts.get_task_description("unknown_task")  
            'search result'  # Fallback for unknown tasks
        """
        return cls.TASKS.get(task_type, "search result")
    
    @classmethod
    def encode_query(cls, content: str, task_type: str = "search") -> str:
        """
        Encode a query with task-specific EmbeddingGemma prompt optimization.
        
        This is the primary method for formatting search queries. It combines the
        user's query with the appropriate task-specific prompt template to achieve
        optimal embedding quality and search relevance.
        
        Args:
            content (str): The raw query text from the user
            task_type (str): Task type for optimization. Defaults to "search"
                           Valid options: "search", "question_answering", "fact_checking",
                           "semantic_similarity", "classification", "clustering", "code_retrieval"
        
        Returns:
            str: Optimized query string formatted for EmbeddingGemma
            
        Performance Impact:
            Using appropriate task types can improve similarity scores by 39-136%
            compared to the baseline "search" task type.
            
        Example:
            >>> cls.encode_query("How does risk management work?", "question_answering")
            'task: question answering | query: How does risk management work?'
            
            >>> cls.encode_query("RS system reduces risk by 30%", "fact_checking") 
            'task: fact checking | query: RS system reduces risk by 30%'
        """
        task_desc = cls.get_task_description(task_type)
        return cls.format_query_prompt(content, task_desc)
    
    @classmethod
    def encode_document(cls, content: str, title: str = "none") -> str:
        """
        Encode a document with proper EmbeddingGemma document formatting.
        
        Formats documents for embedding using the official EmbeddingGemma document
        template. Including meaningful titles significantly improves search relevance
        and helps the model understand document structure.
        
        Args:
            content (str): The document text content to embed
            title (str): Document title extracted from metadata, filename, or content.
                        Use "none" if no meaningful title is available
        
        Returns:
            str: Formatted document string ready for embedding
            
        Best Practices:
            - Extract titles from filenames, headers, or metadata when possible
            - Use "none" rather than empty string when no title is available
            - Keep titles concise and descriptive (< 100 characters)
            
        Example:
            >>> cls.encode_document("Trading strategy content...", "Momentum Strategy Guide")
            'title: Momentum Strategy Guide | text: Trading strategy content...'
            
            >>> cls.encode_document("Untitled content here")
            'title: none | text: Untitled content here'
        """
        return cls.format_document_prompt(content, title)



def slice_list(lst: list, start: int, end: int) -> list:
    """
    A tool that slices a list given a start and end index.
    Args:
        lst: The list to slice.
        start: The start index.
        end: The end index.
    Returns:
        The sliced list.
    """
    return lst[start:end]

def say_hello(name: str) -> str:
    """
    A tool that returns message "Hello <name>"
    Args:
        str: name to say hello to.
    Returns:
        Hello <name>
    """
    return f"Hello, {name}"

with gr.Blocks() as demo:
    gr.Markdown(
        """
        This is a demo of a MCP-only tool.
        This tool slices a list.
        This tool is MCP-only, so it does not have a UI.
        """
    )
    gr.api(
        slice_list
    )
    gr.api(
        say_hello
    )

_, url, _ = demo.launch(mcp_server=True)