local_embedder.py

import numpy as np
import torch
from typing import List, Union, Dict
from transformers import AutoTokenizer, AutoModel
from pathlib import Path
import json

class LocalEmbedder:
    def __init__(self, model_name: str, device: str = None, batch_size: int = 32):
        """
        Ініціалізація локальної моделі для ембедінгів
        
        Args:
            model_name: назва або шлях до моделі (з HuggingFace або локальна)
            device: пристрій для обчислень ('cuda', 'cpu' або None - автовибір)
            batch_size: розмір батчу для інференсу
        """
        self.model_name = model_name
        self.batch_size = batch_size
        
        # Визначення пристрою
        if device is None:
            self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
        else:
            self.device = device
            
        # Завантаження моделі та токенізатора
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name).to(self.device)
        self.model.eval()
        
        # Максимальна довжина послідовності
        self.max_length = self.tokenizer.model_max_length
        if self.max_length > 512:
            self.max_length = 512
            
    def _normalize_embeddings(self, embeddings: np.ndarray) -> np.ndarray:
        """
        L2-нормалізація ембедінгів
        
        Args:
            embeddings: матриця ембедінгів
            
        Returns:
            np.ndarray: нормалізована матриця ембедінгів
        """
        norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
        return embeddings / norms
            
    def get_embeddings(self, texts: Union[str, List[str]]) -> np.ndarray:
        """
        Отримання ембедінгів для тексту або списку текстів
        
        Args:
            texts: текст або список текстів
            
        Returns:
            np.ndarray: матриця нормалізованих ембедінгів
        """
        if isinstance(texts, str):
            texts = [texts]
            
        all_embeddings = []
        
        with torch.no_grad():
            for i in range(0, len(texts), self.batch_size):
                batch_texts = texts[i:i + self.batch_size]
                
                # Токенізація
                encoded = self.tokenizer.batch_encode_plus(
                    batch_texts,
                    padding=True,
                    truncation=True,
                    max_length=self.max_length,
                    return_tensors='pt'
                )
                
                # Переміщуємо тензори на потрібний пристрій
                input_ids = encoded['input_ids'].to(self.device)
                attention_mask = encoded['attention_mask'].to(self.device)
                
                # Отримуємо ембедінги
                outputs = self.model(
                    input_ids=input_ids,
                    attention_mask=attention_mask
                )
                
                # Використовуємо [CLS] токен як ембедінг
                embeddings = outputs.last_hidden_state[:, 0, :]
                all_embeddings.append(embeddings.cpu().numpy())
                
        # Об'єднуємо всі батчі
        embeddings = np.vstack(all_embeddings)
        
        # Нормалізуємо ембедінги
        normalized_embeddings = self._normalize_embeddings(embeddings)
        
        return normalized_embeddings
    
    def get_model_info(self) -> Dict[str, any]:
        """
        Отримання інформації про модель
        
        Returns:
            Dict: інформація про модель
        """
        return {
            'model_name': self.model_name,
            'device': self.device,
            'embedding_size': self.model.config.hidden_size,
            'max_length': self.max_length,
            'batch_size': self.batch_size
        }