Comprehensive comparison for Embeddings technology in AI applications

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
import numpy as np
from typing import List, Dict, Any
import logging

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

class SemanticSearchEngine:
    """Production-ready semantic search engine using Milvus for AI-powered product search."""
    
    def __init__(self, host: str = "localhost", port: str = "19530"):
        self.collection_name = "product_embeddings"
        self.dimension = 768
        self.connect_to_milvus(host, port)
        
    def connect_to_milvus(self, host: str, port: str) -> None:
        """Establish connection to Milvus server with error handling."""
        try:
            connections.connect(alias="default", host=host, port=port)
            logger.info(f"Connected to Milvus at {host}:{port}")
        except Exception as e:
            logger.error(f"Failed to connect to Milvus: {e}")
            raise
    
    def create_collection(self) -> None:
        """Create collection with optimized schema for product search."""
        if utility.has_collection(self.collection_name):
            logger.info(f"Collection {self.collection_name} already exists")
            return
        
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="product_id", dtype=DataType.VARCHAR, max_length=100),
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dimension),
            FieldSchema(name="price", dtype=DataType.FLOAT),
            FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=50)
        ]
        
        schema = CollectionSchema(fields=fields, description="Product semantic search")
        collection = Collection(name=self.collection_name, schema=schema)
        
        index_params = {
            "metric_type": "IP",
            "index_type": "IVF_FLAT",
            "params": {"nlist": 1024}
        }
        collection.create_index(field_name="embedding", index_params=index_params)
        logger.info(f"Created collection {self.collection_name} with index")
    
    def insert_products(self, products: List[Dict[str, Any]]) -> None:
        """Batch insert product embeddings with validation."""
        if not products:
            logger.warning("No products to insert")
            return
        
        try:
            collection = Collection(self.collection_name)
            
            product_ids = [p["product_id"] for p in products]
            embeddings = [p["embedding"] for p in products]
            prices = [p["price"] for p in products]
            categories = [p["category"] for p in products]
            
            data = [product_ids, embeddings, prices, categories]
            collection.insert(data)
            collection.flush()
            logger.info(f"Inserted {len(products)} products")
        except Exception as e:
            logger.error(f"Failed to insert products: {e}")
            raise
    
    def search_similar_products(self, query_embedding: np.ndarray, 
                               top_k: int = 10, 
                               price_range: tuple = None) -> List[Dict]:
        """Search for similar products with optional filtering."""
        try:
            collection = Collection(self.collection_name)
            collection.load()
            
            search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
            
            expr = None
            if price_range:
                min_price, max_price = price_range
                expr = f"price >= {min_price} && price <= {max_price}"
            
            results = collection.search(
                data=[query_embedding.tolist()],
                anns_field="embedding",
                param=search_params,
                limit=top_k,
                expr=expr,
                output_fields=["product_id", "price", "category"]
            )
            
            search_results = []
            for hits in results:
                for hit in hits:
                    search_results.append({
                        "product_id": hit.entity.get("product_id"),
                        "price": hit.entity.get("price"),
                        "category": hit.entity.get("category"),
                        "similarity_score": hit.score
                    })
            
            logger.info(f"Found {len(search_results)} similar products")
            return search_results
        except Exception as e:
            logger.error(f"Search failed: {e}")
            raise
    
    def cleanup(self) -> None:
        """Release resources and close connection."""
        try:
            if utility.has_collection(self.collection_name):
                Collection(self.collection_name).release()
            connections.disconnect("default")
            logger.info("Disconnected from Milvus")
        except Exception as e:
            logger.error(f"Cleanup failed: {e}")

if __name__ == "__main__":
    engine = SemanticSearchEngine()
    engine.create_collection()
    
    sample_products = [
        {"product_id": "PROD001", "embedding": np.random.rand(768).tolist(), 
         "price": 29.99, "category": "electronics"},
        {"product_id": "PROD002", "embedding": np.random.rand(768).tolist(), 
         "price": 49.99, "category": "electronics"}
    ]
    
    engine.insert_products(sample_products)
    query = np.random.rand(768)
    results = engine.search_similar_products(query, top_k=5, price_range=(20, 50))
    print(f"Search results: {results}")
    engine.cleanup()