Spaces:
Sleeping
Sleeping
| import time | |
| import numpy as np | |
| import tensorflow as tf | |
| import torch | |
| import torch.nn.functional as F | |
| from app.services.model_loader import model_manager | |
| from app.services.explainer import get_lr_important_words, generate_reasoning | |
| from app.utils.preprocessing import ( | |
| preprocess_for_tfidf, | |
| text_to_sequence, | |
| preprocess_for_bert | |
| ) | |
| from app.schemas.prediction import ModelPrediction | |
| def _normalize_conf(prob_pos: float, threshold: float) -> float: | |
| if prob_pos >= threshold: | |
| return 0.5 + 0.5 * ((prob_pos - threshold) / (1.0 - threshold)) if threshold < 1.0 else 1.0 | |
| else: | |
| return 0.5 + 0.5 * ((threshold - prob_pos) / threshold) if threshold > 0.0 else 1.0 | |
| def predict_lr(text: str) -> ModelPrediction: | |
| start_time = time.perf_counter() | |
| if not model_manager.models_loaded["logistic_regression"]: | |
| raise RuntimeError("LR Model not loaded") | |
| vec = model_manager.lr_vectorizer | |
| mdl = model_manager.lr_model | |
| cleaned = preprocess_for_tfidf(text) | |
| features = vec.transform([cleaned]) | |
| proba = mdl.predict_proba(features)[0] | |
| label_idx = int(np.argmax(proba)) | |
| label = "Positive" if label_idx == 1 else "Negative" | |
| confidence = float(proba[label_idx]) | |
| latency_ms = (time.perf_counter() - start_time) * 1000 | |
| top_pos, top_neg = get_lr_important_words(cleaned, vec, mdl) | |
| return ModelPrediction( | |
| name="The Statistician", | |
| label=label, | |
| confidence=confidence, | |
| latency_ms=latency_ms, | |
| top_positive_words=top_pos, | |
| top_negative_words=top_neg, | |
| reasoning=generate_reasoning("lr", confidence, latency_ms) | |
| ) | |
| def predict_lstm(text: str) -> ModelPrediction: | |
| start_time = time.perf_counter() | |
| if not model_manager.models_loaded["lstm"]: | |
| raise RuntimeError("LSTM Model not loaded") | |
| mdl = model_manager.lstm_model | |
| tok = model_manager.lstm_tokenizer | |
| cfg = model_manager.lstm_config | |
| max_len = cfg.get("max_len", 300) | |
| threshold = cfg.get("best_threshold", 0.5) | |
| seq = text_to_sequence(text, tok, max_len) | |
| tensor_seq = tf.convert_to_tensor(seq) | |
| raw = model_manager.lstm_fast_predict(tensor_seq).numpy()[0][0] | |
| label = "Positive" if raw >= threshold else "Negative" | |
| conf = _normalize_conf(float(raw), threshold) | |
| latency_ms = (time.perf_counter() - start_time) * 1000 | |
| return ModelPrediction( | |
| name="The Sequentialist", | |
| label=label, | |
| confidence=conf, | |
| latency_ms=latency_ms, | |
| reasoning=generate_reasoning("lstm", conf, latency_ms) | |
| ) | |
| def predict_bert(text: str) -> ModelPrediction: | |
| start_time = time.perf_counter() | |
| if not model_manager.models_loaded["bert"]: | |
| raise RuntimeError("BERT Model not loaded") | |
| mdl = model_manager.bert_model | |
| tok = model_manager.bert_tokenizer | |
| dev = model_manager.device | |
| threshold = model_manager.bert_threshold | |
| cleaned = preprocess_for_bert(text) | |
| inputs = tok(cleaned, return_tensors="pt", truncation=True, padding=True, max_length=128) | |
| inputs = {k: v.to(dev) for k, v in inputs.items()} | |
| with torch.no_grad(): | |
| logits = mdl(**inputs).logits | |
| proba = F.softmax(logits, dim=-1).cpu().numpy()[0] | |
| prob_pos = float(proba[1]) | |
| label = "Positive" if prob_pos >= threshold else "Negative" | |
| conf = _normalize_conf(prob_pos, threshold) | |
| latency_ms = (time.perf_counter() - start_time) * 1000 | |
| return ModelPrediction( | |
| name="The Contextualist", | |
| label=label, | |
| confidence=conf, | |
| latency_ms=latency_ms, | |
| reasoning=generate_reasoning("bert", conf, latency_ms) | |
| ) | |