Comprehensive comparison for AI technology in applications

import chromadb
from chromadb.config import Settings
from chromadb.utils import embedding_functions
import os
from typing import List, Dict, Optional
import logging

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

class DocumentSearchService:
    """Production-grade document search service using ChromaDB for semantic search."""
    
    def __init__(self, persist_directory: str = "./chroma_db"):
        """Initialize ChromaDB client with persistence and error handling."""
        try:
            self.client = chromadb.PersistentClient(
                path=persist_directory,
                settings=Settings(
                    anonymized_telemetry=False,
                    allow_reset=False
                )
            )
            
            # Use OpenAI embeddings with fallback to sentence transformers
            api_key = os.getenv("OPENAI_API_KEY")
            if api_key:
                self.embedding_function = embedding_functions.OpenAIEmbeddingFunction(
                    api_key=api_key,
                    model_name="text-embedding-ada-002"
                )
            else:
                self.embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(
                    model_name="all-MiniLM-L6-v2"
                )
            
            logger.info("ChromaDB client initialized successfully")
        except Exception as e:
            logger.error(f"Failed to initialize ChromaDB: {str(e)}")
            raise
    
    def create_or_get_collection(self, collection_name: str):
        """Create or retrieve a collection with proper error handling."""
        try:
            collection = self.client.get_or_create_collection(
                name=collection_name,
                embedding_function=self.embedding_function,
                metadata={"hnsw:space": "cosine"}
            )
            logger.info(f"Collection '{collection_name}' ready")
            return collection
        except Exception as e:
            logger.error(f"Error with collection '{collection_name}': {str(e)}")
            raise
    
    def add_documents(self, collection_name: str, documents: List[str], 
                     metadatas: Optional[List[Dict]] = None, ids: Optional[List[str]] = None):
        """Add documents to collection with validation and error handling."""
        if not documents:
            raise ValueError("Documents list cannot be empty")
        
        if ids and len(ids) != len(documents):
            raise ValueError("IDs length must match documents length")
        
        if metadatas and len(metadatas) != len(documents):
            raise ValueError("Metadatas length must match documents length")
        
        try:
            collection = self.create_or_get_collection(collection_name)
            
            # Generate IDs if not provided
            if not ids:
                ids = [f"doc_{i}" for i in range(len(documents))]
            
            collection.add(
                documents=documents,
                metadatas=metadatas,
                ids=ids
            )
            logger.info(f"Added {len(documents)} documents to '{collection_name}'")
            return {"status": "success", "count": len(documents)}
        except Exception as e:
            logger.error(f"Failed to add documents: {str(e)}")
            raise
    
    def search(self, collection_name: str, query: str, n_results: int = 5, 
              filter_metadata: Optional[Dict] = None) -> List[Dict]:
        """Perform semantic search with optional metadata filtering."""
        if not query or not query.strip():
            raise ValueError("Query cannot be empty")
        
        if n_results < 1:
            raise ValueError("n_results must be at least 1")
        
        try:
            collection = self.create_or_get_collection(collection_name)
            
            results = collection.query(
                query_texts=[query],
                n_results=min(n_results, collection.count()),
                where=filter_metadata
            )
            
            # Format results for API response
            formatted_results = []
            for i in range(len(results['ids'][0])):
                formatted_results.append({
                    "id": results['ids'][0][i],
                    "document": results['documents'][0][i],
                    "metadata": results['metadatas'][0][i] if results['metadatas'][0][i] else {},
                    "distance": results['distances'][0][i]
                })
            
            logger.info(f"Search completed: {len(formatted_results)} results")
            return formatted_results
        except Exception as e:
            logger.error(f"Search failed: {str(e)}")
            raise

# Example usage in a FastAPI endpoint
if __name__ == "__main__":
    # Initialize service
    search_service = DocumentSearchService()
    
    # Add sample product documents
    products = [
        "Wireless Bluetooth headphones with noise cancellation",
        "USB-C charging cable for fast charging",
        "Laptop stand with adjustable height and angle"
    ]
    
    metadatas = [
        {"category": "electronics", "price": 99.99, "in_stock": True},
        {"category": "accessories", "price": 15.99, "in_stock": True},
        {"category": "accessories", "price": 45.00, "in_stock": False}
    ]
    
    # Add documents
    search_service.add_documents(
        collection_name="products",
        documents=products,
        metadatas=metadatas,
        ids=["prod_001", "prod_002", "prod_003"]
    )
    
    # Search with filtering
    results = search_service.search(
        collection_name="products",
        query="headphone audio device",
        n_results=3,
        filter_metadata={"in_stock": True}
    )
    
    print("Search Results:")
    for result in results:
        print(f"- {result['document']} (distance: {result['distance']:.3f})")