Comprehensive comparison for Embeddings technology in AI applications

import torch
from transformers import AutoTokenizer, AutoModel
import numpy as np
from typing import List, Dict, Any
import logging
from functools import lru_cache

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class BGEEmbeddingService:
    """Production-ready BGE embedding service for semantic search."""
    
    def __init__(self, model_name: str = "BAAI/bge-base-en-v1.5", device: str = None):
        """
        Initialize BGE model for generating embeddings.
        
        Args:
            model_name: HuggingFace model identifier
            device: cuda, cpu, or None for auto-detection
        """
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        logger.info(f"Loading BGE model on {self.device}")
        
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModel.from_pretrained(model_name)
            self.model.to(self.device)
            self.model.eval()
            logger.info("BGE model loaded successfully")
        except Exception as e:
            logger.error(f"Failed to load model: {e}")
            raise
    
    def encode(self, texts: List[str], batch_size: int = 32, 
               normalize: bool = True, add_instruction: bool = False) -> np.ndarray:
        """
        Generate embeddings for input texts with batching.
        
        Args:
            texts: List of text strings to embed
            batch_size: Number of texts to process at once
            normalize: Whether to L2 normalize embeddings
            add_instruction: Add BGE instruction prefix for queries
        
        Returns:
            numpy array of shape (len(texts), embedding_dim)
        """
        if not texts:
            raise ValueError("Input texts list cannot be empty")
        
        # Add instruction prefix for query texts if needed
        if add_instruction:
            texts = [f"Represent this sentence for searching relevant passages: {text}" 
                    for text in texts]
        
        all_embeddings = []
        
        with torch.no_grad():
            for i in range(0, len(texts), batch_size):
                batch_texts = texts[i:i + batch_size]
                
                try:
                    # Tokenize with padding and truncation
                    encoded_input = self.tokenizer(
                        batch_texts,
                        padding=True,
                        truncation=True,
                        max_length=512,
                        return_tensors="pt"
                    )
                    encoded_input = {k: v.to(self.device) for k, v in encoded_input.items()}
                    
                    # Generate embeddings using CLS token pooling
                    model_output = self.model(**encoded_input)
                    embeddings = model_output[0][:, 0]  # CLS token
                    
                    # Normalize embeddings for cosine similarity
                    if normalize:
                        embeddings = torch.nn.functional.normalize(embeddings, p=2, dim=1)
                    
                    all_embeddings.append(embeddings.cpu().numpy())
                    
                except Exception as e:
                    logger.error(f"Error processing batch {i//batch_size}: {e}")
                    raise
        
        return np.vstack(all_embeddings)
    
    def compute_similarity(self, query_embedding: np.ndarray, 
                          doc_embeddings: np.ndarray) -> np.ndarray:
        """
        Compute cosine similarity between query and documents.
        
        Args:
            query_embedding: Single query embedding (1, dim)
            doc_embeddings: Document embeddings (n, dim)
        
        Returns:
            Similarity scores array of shape (n,)
        """
        return np.dot(doc_embeddings, query_embedding.T).squeeze()

# Example usage: Semantic product search API
def search_products(query: str, product_descriptions: List[str], 
                   top_k: int = 5) -> List[Dict[str, Any]]:
    """
    Search products using semantic similarity.
    
    Args:
        query: User search query
        product_descriptions: List of product descriptions
        top_k: Number of results to return
    
    Returns:
        List of top matching products with scores
    """
    try:
        service = BGEEmbeddingService()
        
        # Encode query with instruction prefix
        query_emb = service.encode([query], add_instruction=True)
        
        # Encode product descriptions
        doc_embs = service.encode(product_descriptions, batch_size=32)
        
        # Compute similarities
        scores = service.compute_similarity(query_emb, doc_embs)
        
        # Get top-k results
        top_indices = np.argsort(scores)[::-1][:top_k]
        
        results = [
            {
                "index": int(idx),
                "description": product_descriptions[idx],
                "score": float(scores[idx])
            }
            for idx in top_indices
        ]
        
        return results
        
    except Exception as e:
        logger.error(f"Search failed: {e}")
        return []