import gc import random import threading import time import warnings from src.core.config import settings class TextCoherenceEvaluator: """ Clase para evaluar la coherencia de un texto mediante un modelo fill-mask, con carga perezosa y descarga automática tras periodo de inactividad. """ IRRELEVANT_COMMENTS = {"sin comentarios", "na", "n/a", "no comment", "ninguno", "ninguna", ".", "...", ".."} _UNLOAD_DELAY = settings.ML_IDLE_UNLOAD_SECONDS _instance: "TextCoherenceEvaluator | None" = None _instance_lock = threading.Lock() @classmethod def get_instance(cls) -> "TextCoherenceEvaluator": with cls._instance_lock: if cls._instance is None: cls._instance = cls() return cls._instance @classmethod def release_all(cls) -> None: with cls._instance_lock: evaluator = cls._instance cls._instance = None if evaluator is not None: evaluator.close() def __init__(self, model_name: str = "roberta-base", top_k: int = 20, batch_size: int = 8): warnings.filterwarnings("ignore", category=UserWarning) # Parámetros y estado inicial self.model_name = model_name self.top_k = top_k self.batch_size = batch_size self.device_type = "cpu" self.device = None self.tokenizer = None self.model = None self.fill_mask = None self._last_used = None self._timer = None self._torch = None self._nltk = None self._pipeline_factory = None self._auto_tokenizer = None self._auto_model = None def _ensure_runtime(self) -> None: if self._torch is not None and self._nltk is not None: return import nltk import torch from transformers import AutoModelForMaskedLM, AutoTokenizer, logging as transformers_logging, pipeline transformers_logging.set_verbosity_error() nltk.download('punkt', quiet=True) nltk.download('averaged_perceptron_tagger', quiet=True) self._torch = torch self._nltk = nltk self._pipeline_factory = pipeline self._auto_tokenizer = AutoTokenizer self._auto_model = AutoModelForMaskedLM self.device_type = self._get_device_type() self.device = torch.device(self.device_type) def _get_device_type(self) -> str: torch = self._torch if torch is None: print("Torch no disponible. Usando modo fallback sin modelo de coherencia.") return "cpu" if torch.cuda.is_available(): print("Usando CUDA.") return "cuda" elif torch.backends.mps.is_available(): print("Usando GPU M1 (MPS).") return "mps" else: print("Usando CPU.") return "cpu" def _load_model(self): """Carga el tokenizer, modelo y pipeline si no están cargados.""" self._ensure_runtime() torch = self._torch if torch is None: self.fill_mask = None self.model = None self.tokenizer = None return if self.model is None: # Tokenizer y modelo self.tokenizer = self._auto_tokenizer.from_pretrained(self.model_name) self.model = self._auto_model.from_pretrained(self.model_name) self.model.to(self.device) # Cuantización dinámica en CPU if self.device_type == 'cpu': supported = torch.backends.quantized.supported_engines for engine in ("fbgemm", "qnnpack"): if engine in supported: torch.backends.quantized.engine = engine self.model = torch.quantization.quantize_dynamic( self.model, {torch.nn.Linear}, dtype=torch.qint8 ) break # Pipeline fill-mask device_arg = 0 if self.device_type == 'cuda' else -1 self.fill_mask = self._pipeline_factory( "fill-mask", model=self.model, tokenizer=self.tokenizer, device=device_arg, top_k=self.top_k ) # Actualizar tiempo de último uso y reprogramar descarga self._last_used = time.time() self._schedule_unload() def _schedule_unload(self): """Programa la descarga del modelo tras un periodo de inactividad.""" if self._timer: self._timer.cancel() self._timer = threading.Timer(self._UNLOAD_DELAY, self._unload_model) self._timer.daemon = True self._timer.start() def _unload_model(self): """Libera modelo, tokenizer y pipeline de memoria.""" if self._timer: self._timer.cancel() self._timer = None try: del self.fill_mask, self.model, self.tokenizer except Exception: pass self.fill_mask = None self.model = None self.tokenizer = None if self._torch is not None and self.device_type == 'cuda': self._torch.cuda.empty_cache() self._torch = None self._nltk = None self._pipeline_factory = None self._auto_tokenizer = None self._auto_model = None gc.collect() def close(self) -> None: self._unload_model() def is_irrelevant_comment(self, text: str) -> bool: return text.strip().lower() in self.IRRELEVANT_COMMENTS def get_significant_word_indices(self, tokens: list[str]) -> list[int]: self._ensure_runtime() words = self.tokenizer.convert_tokens_to_string(tokens).split() pos_tags = self._nltk.pos_tag(words) return [i for i, (_, pos) in enumerate(pos_tags) if pos.startswith(('NN', 'VB'))] def validate(self, text: str) -> float: """ Evalúa la coherencia del texto: 1. Carga perezosa del modelo 2. Tokeniza y enmascara palabras significativas 3. Inferencia por lotes 4. Descarga automática tras inactividad """ text = text.strip() if not text or self.is_irrelevant_comment(text): return 0.0 # Asegurar que el modelo esté cargado self._load_model() if self.fill_mask is None or self.tokenizer is None: return 0.0 sentences = self._nltk.tokenize.sent_tokenize(text) masked_texts, original_words = [], [] total_tokens = 0 for sentence in sentences: tokens = self.tokenizer.tokenize(sentence) total_tokens += len(tokens) if total_tokens > self.tokenizer.model_max_length: break if not tokens: continue sig_idxs = self.get_significant_word_indices(tokens) idx = random.choice(sig_idxs) if sig_idxs else random.randrange(len(tokens)) original = tokens[idx] masked = tokens.copy() masked[idx] = self.tokenizer.mask_token masked_texts.append(self.tokenizer.convert_tokens_to_string(masked)) original_words.append(original) if not masked_texts: return 0.0 results = self.fill_mask(masked_texts, batch_size=self.batch_size) scores = [] for orig, res in zip(original_words, results): items = res if isinstance(res, list) else [res] score = sum(r['score'] for r in items if r['token_str'].strip() == orig) scores.append(score) return float(sum(scores) / len(scores)) if scores else 0.0