from __future__ import annotations import json import re import warnings from collections import Counter from pathlib import Path from typing import Any, Dict, Optional, Tuple import joblib import numpy as np import pandas as pd import streamlit as st import torch import torch.nn as nn from pyvi import ViTokenizer from sklearn.metrics import accuracy_score, precision_recall_fscore_support from torch.utils.data import DataLoader, Dataset try: from teencode_normalizer import normalize_teencode except ImportError: def normalize_teencode(t: str) -> str: return t warnings.filterwarnings("ignore") st.set_page_config( page_title="DLNLP Demo", page_icon="🧠", layout="wide", initial_sidebar_state="collapsed", ) PROJECT_ROOT = Path(__file__).resolve().parent DATA_DIR = PROJECT_ROOT / "data" MODEL_DIR = PROJECT_ROOT / "outputs" / "models" RESULT_DIR = PROJECT_ROOT / "outputs" / "results" FIG_DIR = PROJECT_ROOT / "outputs" / "figures" PHOBERT_DIR = MODEL_DIR / "phobert_base" DEBERTA_DIR = MODEL_DIR / "deberta" LABEL_MAP = {0: "CLEAN", 1: "OFFENSIVE", 2: "HATE"} LABEL_VI = {0: "Bình thường", 1: "Công kích", 2: "Thù ghét"} LABEL_COLOR = {0: "#16a34a", 1: "#f59e0b", 2: "#dc2626"} LABEL_BG = {0: "#dcfce7", 1: "#ffedd5", 2: "#fee2e2"} LABEL_TEXT = {0: "#14532d", 1: "#7c2d12", 2: "#7f1d1d"} DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") SAMPLES = { "Câu tích cực": "Bài viết này rất hữu ích, cảm ơn bạn đã chia sẻ.", "Câu công kích nhẹ": "Đọc mà chán thật, nói năng kiểu này thì ai chịu nổi.", "Câu thù ghét": "Mấy đứa đó đúng là phải biến khỏi đây ngay.", "Câu trung tính": "Hôm nay trời khá mát, mình vừa đi học về.", } class BiLSTMClassifier(nn.Module): def __init__(self, vocab_size: int, embed_dim: int, hidden_dim: int, num_classes: int, num_layers: int = 2, dropout: float = 0.4): super().__init__() self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0) self.emb_dropout = nn.Dropout(dropout) self.lstm = nn.LSTM( input_size=embed_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, bidirectional=True, dropout=dropout if num_layers > 1 else 0.0, ) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_dim * 4, num_classes) def forward(self, input_ids, lengths): emb = self.emb_dropout(self.embedding(input_ids)) packed = nn.utils.rnn.pack_padded_sequence( emb, lengths.cpu(), batch_first=True, enforce_sorted=False, ) packed_out, _ = self.lstm(packed) output, _ = nn.utils.rnn.pad_packed_sequence( packed_out, batch_first=True, total_length=input_ids.size(1), ) mask = (input_ids != 0).unsqueeze(-1) output_masked = output.masked_fill(~mask, -1e9) max_pool = output_masked.max(dim=1).values sum_pool = (output * mask).sum(dim=1) valid_len = mask.sum(dim=1).clamp(min=1) mean_pool = sum_pool / valid_len feat = torch.cat([max_pool, mean_pool], dim=1) logits = self.fc(self.dropout(feat)) return logits class TextDataset(Dataset): def __init__(self, texts, vocab, max_len): self.texts = list(texts) self.vocab = vocab self.max_len = max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): input_ids, length = encode_text(self.texts[idx], self.vocab, self.max_len) return { "input_ids": torch.tensor(input_ids, dtype=torch.long), "length": torch.tensor(length, dtype=torch.long), } @st.cache_resource(show_spinner=False) def load_svm_artifacts() -> Tuple[Any, Any]: vectorizer = joblib.load(MODEL_DIR / "tfidf_vectorizer.joblib") model = joblib.load(MODEL_DIR / "svm_model.joblib") return vectorizer, model @st.cache_resource(show_spinner=False) def load_bilstm_artifacts() -> Optional[Dict[str, Any]]: vocab_path = MODEL_DIR / "bilstm_vocab.json" model_path = MODEL_DIR / "bilstm_best.pt" metrics_path = RESULT_DIR / "bilstm_metrics.csv" if not vocab_path.exists() or not model_path.exists() or not metrics_path.exists(): return None vocab = json.loads(vocab_path.read_text(encoding="utf-8")) metrics_df = pd.read_csv(metrics_path) if metrics_df.empty: return None row = metrics_df.iloc[0] max_len = int(row.get("max_len", 64)) embed_dim = int(row.get("embed_dim", 300)) hidden_dim = int(row.get("hidden_dim", 128)) num_layers = int(row.get("num_layers", 2)) model = BiLSTMClassifier( vocab_size=len(vocab), embed_dim=embed_dim, hidden_dim=hidden_dim, num_classes=3, num_layers=num_layers, dropout=0.4, ).to(DEVICE) state_dict = torch.load(model_path, map_location=DEVICE) model.load_state_dict(state_dict) model.eval() return { "model": model, "vocab": vocab, "max_len": max_len, "embed_dim": embed_dim, "hidden_dim": hidden_dim, "num_layers": num_layers, } @st.cache_resource(show_spinner=False) def find_phobert_model_dir() -> Optional[Path]: if not PHOBERT_DIR.exists(): return None candidates = [] for pattern in ("**/config.json", "**/pytorch_model.bin", "**/model.safetensors"): for file_path in PHOBERT_DIR.glob(pattern): candidates.append(file_path.parent) if not candidates: return None unique_dirs = [] seen = set() for candidate in candidates: resolved = candidate.resolve() if resolved not in seen: seen.add(resolved) unique_dirs.append(candidate) return max(unique_dirs, key=lambda item: item.stat().st_mtime) @st.cache_resource(show_spinner=False) def load_phobert_artifacts() -> Optional[Dict[str, Any]]: model_dir = find_phobert_model_dir() if model_dir is None: return None try: from transformers.models.auto.modeling_auto import AutoModelForSequenceClassification from transformers.models.auto.tokenization_auto import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained(model_dir) model = AutoModelForSequenceClassification.from_pretrained(model_dir).to(DEVICE) model.eval() return {"tokenizer": tokenizer, "model": model, "model_dir": model_dir} except Exception: return None @st.cache_resource(show_spinner=False) def find_deberta_model_dir() -> Optional[Path]: if not DEBERTA_DIR.exists(): return None candidates = [] for pattern in ("**/config.json", "**/pytorch_model.bin", "**/model.safetensors"): for file_path in DEBERTA_DIR.glob(pattern): candidates.append(file_path.parent) if not candidates: return None unique_dirs = [] seen = set() for candidate in candidates: resolved = candidate.resolve() if resolved not in seen: seen.add(resolved) unique_dirs.append(candidate) return max(unique_dirs, key=lambda item: item.stat().st_mtime) @st.cache_resource(show_spinner=False) def load_deberta_artifacts() -> Optional[Dict[str, Any]]: model_dir = find_deberta_model_dir() if model_dir is None: return None try: from transformers import AutoModelForSequenceClassification, DebertaV2Tokenizer tokenizer = DebertaV2Tokenizer.from_pretrained(model_dir) model = AutoModelForSequenceClassification.from_pretrained(model_dir).to(DEVICE) model.eval() return {"tokenizer": tokenizer, "model": model, "model_dir": model_dir} except Exception: return None @st.cache_data(show_spinner=False) def clean_text(text: str) -> str: text = "" if text is None else str(text) text = text.replace("\n", " ").replace("\t", " ") text = re.sub(r"https?://\S+|www\.\S+", "[URL]", text) text = re.sub(r"\s+", " ", text).strip() return text @st.cache_data(show_spinner=False) def segment_text(text: str) -> str: text = "" if text is None else str(text) text = re.sub(r"\s+", " ", text.strip().lower()) return ViTokenizer.tokenize(text) @st.cache_data(show_spinner=False) def tokenize_text(text: str): return str(text).split() @st.cache_data(show_spinner=False) def encode_text(text: str, vocab: Dict[str, int], max_len: int): tokens = tokenize_text(text)[:max_len] ids = [vocab.get(token, 1) for token in tokens] if len(ids) == 0: ids = [1] length = len(ids) if length < max_len: ids += [0] * (max_len - length) return ids, length @torch.no_grad() def predict_bilstm(text: str, artifacts: Dict[str, Any]): model = artifacts["model"] vocab = artifacts["vocab"] max_len = artifacts["max_len"] seg_text = segment_text(clean_text(text)) input_ids, length = encode_text(seg_text, vocab, max_len) batch = { "input_ids": torch.tensor([input_ids], dtype=torch.long, device=DEVICE), "length": torch.tensor([length], dtype=torch.long, device=DEVICE), } logits = model(batch["input_ids"], batch["length"]) probs = torch.softmax(logits, dim=1).squeeze(0).detach().cpu().numpy() pred_id = int(np.argmax(probs)) return pred_id, probs @torch.no_grad() def predict_phobert(text: str, artifacts: Dict[str, Any]): tokenizer = artifacts["tokenizer"] model = artifacts["model"] seg_text = segment_text(clean_text(text)) encoding = tokenizer( [seg_text], truncation=True, padding=True, max_length=128, return_tensors="pt", ) encoding = {key: value.to(DEVICE) for key, value in encoding.items()} outputs = model(**encoding) logits = outputs.logits probs = torch.softmax(logits, dim=1).squeeze(0).detach().cpu().numpy() pred_id = int(np.argmax(probs)) return pred_id, probs @torch.no_grad() def predict_deberta(text: str, artifacts: Dict[str, Any]): tokenizer = artifacts["tokenizer"] model = artifacts["model"] # DeBERTa uses teencode normalizer instead of ViTokenizer normalized_text = normalize_teencode(text) clean_t = clean_text(normalized_text) encoding = tokenizer( [clean_t], truncation=True, padding=False, max_length=256, return_tensors="pt", ) encoding = {key: value.to(DEVICE) for key, value in encoding.items()} outputs = model(**encoding) logits = outputs.logits probs = torch.softmax(logits, dim=1).squeeze(0).detach().cpu().numpy() pred_id = int(np.argmax(probs)) return pred_id, probs @st.cache_resource(show_spinner=False) def load_models(): svm_vectorizer, svm_model = load_svm_artifacts() bilstm_artifacts = load_bilstm_artifacts() phobert_artifacts = load_phobert_artifacts() deberta_artifacts = load_deberta_artifacts() return { "svm": (svm_vectorizer, svm_model), "bilstm": bilstm_artifacts, "phobert": phobert_artifacts, "deberta": deberta_artifacts, } def predict_svm(text: str, artifacts: Tuple[Any, Any]): vectorizer, model = artifacts cleaned = clean_text(text) features = vectorizer.transform([cleaned]) pred_id = int(model.predict(features)[0]) if hasattr(model, "predict_proba"): probs = model.predict_proba(features)[0] elif hasattr(model, "decision_function"): scores = model.decision_function(features) if np.ndim(scores) == 1: scores = np.column_stack([-scores, scores]) probs = torch.softmax(torch.tensor(scores, dtype=torch.float32), dim=1).squeeze(0).detach().cpu().numpy() else: probs = None return pred_id, probs def top2_labels(probs: Optional[np.ndarray]) -> str: if probs is None: return "Top 2: N/A" probs = np.asarray(probs, dtype=float) if probs.size == 0: return "Top 2: N/A" order = probs.argsort()[::-1][:2] parts = [f"{LABEL_MAP.get(int(index), 'Unknown')} {probs[int(index)]:.1%}" for index in order] return f"Top 2: {' | '.join(parts)}" def _safe_label_id(label_id: Optional[int]) -> Optional[int]: if label_id is None: return None try: label_id = int(label_id) except Exception: return None return label_id if label_id in LABEL_MAP else None def _format_confidence(conf: Optional[float]) -> str: if conf is None: return "N/A" try: return f"{float(conf) * 100:.1f}%" except Exception: return "N/A" def confidence_from_probs(pred_id: Optional[int], probs: Optional[np.ndarray]) -> Optional[float]: safe_id = _safe_label_id(pred_id) if safe_id is None or probs is None: return None try: probs = np.asarray(probs, dtype=float) if probs.size == 0: return None return float(probs[safe_id]) except Exception: return None def majority_vote(pred_ids: list[int]) -> int: valid = [pred for pred in pred_ids if pred in LABEL_MAP] if not valid: return 0 return Counter(valid).most_common(1)[0][0] def consensus_status(pred_ids: list[int]) -> bool: valid = [pred for pred in pred_ids if pred in LABEL_MAP] if len(valid) <= 1: return True return len(set(valid)) == 1 def render_model_card(model_name: str, label_id: Optional[int], confidence: Optional[float], *, available: bool): safe_id = _safe_label_id(label_id) if not available: label_code = "-" label_primary = "KHÔNG KHẢ DỤNG" elif safe_id is None: label_code = "-" label_primary = "CHƯA DỰ ĐOÁN" else: label_code = LABEL_MAP[safe_id] label_primary = label_code bg = "#f1f5f9" if safe_id is None else LABEL_BG[safe_id] border = "#94a3b8" if safe_id is None else LABEL_COLOR[safe_id] text = "#0f172a" if safe_id is None else LABEL_TEXT[safe_id] conf_text = _format_confidence(confidence) if available and safe_id is not None else "N/A" st.markdown( f"""
Nhập một bình luận bất kỳ và so sánh dự đoán từ các mô hình: TF‑IDF + SVM, BiLSTM, PhoBERT và DeBERTa.