Comprehensive comparison for Embeddings technology in AI applications

import os
import logging
from typing import List, Dict, Optional
from gensim.models.doc2vec import Doc2Vec, TaggedDocument
import numpy as np
from flask import Flask, request, jsonify
from functools import lru_cache
import pickle

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

app = Flask(__name__)

class DocumentSimilarityService:
    def __init__(self, model_path: str):
        self.model_path = model_path
        self.model: Optional[Doc2Vec] = None
        self.document_store: Dict[str, str] = {}
        self._load_model()
    
    def _load_model(self):
        try:
            if os.path.exists(self.model_path):
                self.model = Doc2Vec.load(self.model_path)
                logger.info(f"Model loaded from {self.model_path}")
            else:
                logger.warning("No existing model found. Will train new model.")
        except Exception as e:
            logger.error(f"Error loading model: {str(e)}")
            raise
    
    def train_model(self, documents: List[Dict[str, str]], vector_size: int = 100, 
                   min_count: int = 2, epochs: int = 40):
        try:
            tagged_docs = []
            for idx, doc in enumerate(documents):
                doc_id = doc.get('id', str(idx))
                text = doc.get('text', '').lower().split()
                self.document_store[doc_id] = doc.get('text', '')
                tagged_docs.append(TaggedDocument(words=text, tags=[doc_id]))
            
            self.model = Doc2Vec(
                vector_size=vector_size,
                min_count=min_count,
                epochs=epochs,
                dm=1,
                workers=4,
                window=5,
                alpha=0.025,
                min_alpha=0.00025
            )
            
            self.model.build_vocab(tagged_docs)
            self.model.train(tagged_docs, total_examples=self.model.corpus_count, 
                           epochs=self.model.epochs)
            
            self.model.save(self.model_path)
            logger.info(f"Model trained and saved to {self.model_path}")
            return True
        except Exception as e:
            logger.error(f"Error training model: {str(e)}")
            return False
    
    @lru_cache(maxsize=1000)
    def get_document_vector(self, doc_id: str) -> Optional[np.ndarray]:
        try:
            if self.model and doc_id in self.model.dv:
                return self.model.dv[doc_id]
            return None
        except Exception as e:
            logger.error(f"Error getting document vector: {str(e)}")
            return None
    
    def infer_vector(self, text: str) -> Optional[np.ndarray]:
        try:
            if not self.model:
                raise ValueError("Model not initialized")
            tokens = text.lower().split()
            return self.model.infer_vector(tokens, epochs=20)
        except Exception as e:
            logger.error(f"Error inferring vector: {str(e)}")
            return None
    
    def find_similar_documents(self, text: str, top_n: int = 5) -> List[Dict]:
        try:
            vector = self.infer_vector(text)
            if vector is None:
                return []
            
            similar = self.model.dv.most_similar([vector], topn=top_n)
            results = []
            for doc_id, score in similar:
                results.append({
                    'document_id': doc_id,
                    'similarity_score': float(score),
                    'text': self.document_store.get(doc_id, '')
                })
            return results
        except Exception as e:
            logger.error(f"Error finding similar documents: {str(e)}")
            return []

service = DocumentSimilarityService('doc2vec_model.bin')

@app.route('/api/train', methods=['POST'])
def train():
    try:
        data = request.get_json()
        documents = data.get('documents', [])
        if not documents:
            return jsonify({'error': 'No documents provided'}), 400
        
        success = service.train_model(documents)
        if success:
            return jsonify({'message': 'Model trained successfully'}), 200
        return jsonify({'error': 'Training failed'}), 500
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/similar', methods=['POST'])
def find_similar():
    try:
        data = request.get_json()
        query_text = data.get('text', '')
        top_n = data.get('top_n', 5)
        
        if not query_text:
            return jsonify({'error': 'No text provided'}), 400
        
        results = service.find_similar_documents(query_text, top_n)
        return jsonify({'similar_documents': results}), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/health', methods=['GET'])
def health():
    return jsonify({'status': 'healthy', 'model_loaded': service.model is not None}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)