Spaces:
Running
Running
| import os | |
| import sys | |
| import time | |
| import pickle | |
| import numpy as np | |
| import pandas as pd | |
| from tanglish_sentiment_pipeline import ( | |
| resolve_path, | |
| load_dataset, | |
| normalize_text, | |
| CongMoeClassifier, | |
| TARGET_CLASSES, | |
| EMOJI_MAP | |
| ) | |
| MODEL_FILE = resolve_path("tanglish_congmoe_model.pkl") | |
| DATASET_FILE = resolve_path("tanglish_test_20000_with_labels.csv") | |
| STATE_EMOJIS = { | |
| "Positive": "🌟 POSITIVE (Praise / High Excitement)", | |
| "Negative": "💔 NEGATIVE (Disappointment / Criticism)", | |
| "Mixed_feelings": "⚖️ MIXED FEELINGS (Conflicted / Nuanced Review)", | |
| "Unknown_state": "❓ UNKNOWN STATE (Question / Inquiry / Neutral)" | |
| } | |
| def get_or_train_model(): | |
| """Loads the pre-trained model or trains and saves it if not already present.""" | |
| if os.path.exists(MODEL_FILE): | |
| print(f"[*] Loading pre-trained ConG-MoE v2 model from '{MODEL_FILE}'...") | |
| t0 = time.perf_counter() | |
| with open(MODEL_FILE, "rb") as f: | |
| model = pickle.load(f) | |
| print(f"[*] Model loaded in {(time.perf_counter() - t0)*1000:.1f} ms!\n") | |
| return model | |
| print(f"[*] Pre-trained model not found. Training once on {DATASET_FILE}...") | |
| df, text_col, label_col = load_dataset(DATASET_FILE) | |
| X_train_raw = df[text_col] | |
| y_train = df[label_col] | |
| print("[*] Normalizing text representations...") | |
| X_train_norm = X_train_raw.apply(normalize_text) | |
| print("[*] Training ConG-MoE v2 architecture...") | |
| t0 = time.perf_counter() | |
| model = CongMoeClassifier(confidence_threshold=0.65, C=0.5, random_state=42) | |
| model.fit(X_train_norm, y_train, X_raw=X_train_raw) | |
| train_time = time.perf_counter() - t0 | |
| print(f"[*] Training finished in {train_time:.2f} seconds.") | |
| print(f"[*] Saving model to '{MODEL_FILE}' for instant future startup...") | |
| with open(MODEL_FILE, "wb") as f: | |
| pickle.dump(model, f) | |
| print("[*] Model successfully cached!\n") | |
| return model | |
| def predict_comment(raw_text: str, model: CongMoeClassifier): | |
| """Predicts the sentiment state of a user comment using ConG-MoE v2.""" | |
| t0 = time.perf_counter() | |
| clean_text = normalize_text(raw_text) | |
| # 1. Base statistical pipeline prediction & probabilities | |
| probs = model.base_pipeline.predict_proba([clean_text])[0] | |
| classes = model.base_pipeline.classes_ | |
| prob_dict = {cls: float(p) for cls, p in zip(classes, probs)} | |
| base_pred = classes[probs.argmax()] | |
| confidence = float(probs.max()) | |
| # 2. Check Expert Heads & Linguistic Markers | |
| # Extended pivot check | |
| extended_pivots = {"but", "aana", "aanal", "aanaa", "anaa", "ana", "irundhalum", "irunthalum", "yet", "though", "however", "analum"} | |
| has_pivot = any(p in clean_text.split() for p in extended_pivots) | |
| # Counterfactual regret markers ('could have been better') | |
| regret_markers = {"irunthurukalam", "irunthirukalam", "irundhurukalam", "irundhirukalam", "irukalam", "irunthrukalam", "panniirukalam"} | |
| has_regret = any(m in clean_text for m in regret_markers) or ("inno" in clean_text and "best" in clean_text) or ("innum" in clean_text and "nalla" in clean_text) | |
| pivot_pred = model.pivot_head.predict_sample(raw_text) | |
| emoji_pred = model.emoji_head.predict_sample(raw_text) | |
| # Affective terms | |
| pos_terms = {"sema", "mass", "super", "nalla", "best", "verithanam", "thalaiva", "level", "good", "love", "arputham", "marana"} | |
| neg_terms = {"worst", "waste", "mokka", "bore", "flaw", "bad", "disaster", "terrible", "hate", "daridram", "kevalam"} | |
| words_set = set(clean_text.split()) | |
| has_pos_term = any(t in words_set for t in pos_terms) | |
| has_neg_term = any(t in words_set for t in neg_terms) | |
| has_sad_emoji = any(e in raw_text for e in ["😢", "😭", "😡", "👎", "🤮", "💔", "🤦", "🤦♂️", "😠"]) | |
| has_happy_emoji = any(e in raw_text for e in ["😊", "😍", "🎉", "🔥", "❤️", "👍", "👏", "💯", "💥", "🤩"]) | |
| is_emoji_dissonance = (has_pos_term and has_sad_emoji) or (has_neg_term and has_happy_emoji) | |
| # 3. Gating Decision Logic | |
| override_reason = None | |
| if (has_pivot and has_pos_term and has_neg_term) or (has_pivot and has_regret) or (has_regret and "acting" in clean_text): | |
| final_state = "Mixed_feelings" | |
| override_reason = "Contrastive pivot ('aana'/'but') connecting positive and negative clauses" | |
| elif is_emoji_dissonance and confidence < 0.85: | |
| final_state = "Mixed_feelings" | |
| override_reason = "Affective dissonance detected between text and emojis" | |
| elif confidence < model.confidence_threshold and (pivot_pred == "Mixed_feelings" or (emoji_pred == "Mixed_feelings" and not (has_neg_term and has_sad_emoji))): | |
| final_state = "Mixed_feelings" | |
| override_reason = "Contrastive pivot or discordance detected by specialist head" | |
| elif ("?" in raw_text or any(q in words_set for q in ["eppo", "eppadi", "enna", "ethu", "when", "why", "who", "what"])) and (base_pred == "Unknown_state" or prob_dict.get("Unknown_state", 0) > 0.15 or confidence < 0.60): | |
| final_state = "Unknown_state" | |
| override_reason = "Inquiry syntax ('?') and interrogative token detected" | |
| elif has_neg_term and not has_pos_term and (base_pred != "Negative"): | |
| # Strong negative keywords should not fall to Positive | |
| final_state = "Negative" | |
| override_reason = "Explicit negative evaluation keywords detected" | |
| else: | |
| final_state = base_pred | |
| latency_ms = (time.perf_counter() - t0) * 1000 | |
| return { | |
| "raw_text": raw_text, | |
| "clean_text": clean_text, | |
| "final_state": final_state, | |
| "base_prediction": base_pred, | |
| "confidence": confidence, | |
| "override_reason": override_reason, | |
| "prob_dict": prob_dict, | |
| "latency_ms": latency_ms | |
| } | |
| def print_prediction_result(res: dict): | |
| state_str = STATE_EMOJIS.get(res['final_state'], res['final_state']) | |
| print("\n" + "─" * 70) | |
| print(f" Input Comment : \"{res['raw_text']}\"") | |
| print(f" Processed As : \"{res['clean_text']}\"") | |
| print(f" Predicted State: {state_str}") | |
| print("─" * 70) | |
| print(f" Base Model Prediction : {res['base_prediction']} (Confidence: {res['confidence']*100:.1f}%)") | |
| if res['override_reason']: | |
| print(f" Specialist Intervention: [YES] Overridden to Mixed feelings") | |
| print(f" Reason : {res['override_reason']}") | |
| else: | |
| print(f" Specialist Intervention: [NO] Base prediction accepted (Confidence >= 65% or no conflict)") | |
| print(f" Inference Latency : {res['latency_ms']:.3f} milliseconds") | |
| print("\n Class Probability Breakdown:") | |
| for cls in TARGET_CLASSES: | |
| p = res['prob_dict'].get(cls, 0.0) | |
| bar = "█" * int(p * 30) | |
| print(f" - {cls:<16}: {p*100:5.1f}% |{bar:<30}|\n") | |
| print("─" * 70 + "\n") | |
| def main(): | |
| model = get_or_train_model() | |
| # If the user passed a comment directly as a command-line argument: | |
| if len(sys.argv) > 1: | |
| user_comment = " ".join(sys.argv[1:]) | |
| res = predict_comment(user_comment, model) | |
| print_prediction_result(res) | |
| return | |
| # Otherwise, run interactive prompt loop: | |
| print("=" * 70) | |
| print(" TANGLISH SENTIMENT PREDICTOR (ConG-MoE v2 Interactive CLI)") | |
| print("=" * 70) | |
| print("Type any Tanglish comment below and hit Enter.") | |
| print("Commands: 'exit' or 'q' to quit, 'samples' to test built-in examples.\n") | |
| sample_comments = [ | |
| "Thalaivar entry vera level mass 🔥 padam blockbuster confirmed!", | |
| "Padam romba mokka da timing waste don't go theatre 🤮 worst movie", | |
| "Acting sema mass aana screenplay romba worst", | |
| "first half nalla irukku aana climax mokka 🤮", | |
| "GOAT trailer vera level bro... eppo ott release date???" | |
| ] | |
| while True: | |
| try: | |
| user_input = input("Enter Tanglish Comment > ").strip() | |
| except (EOFError, KeyboardInterrupt): | |
| print("\nExiting. Vanakkam!") | |
| break | |
| if not user_input: | |
| continue | |
| if user_input.lower() in ["q", "quit", "exit"]: | |
| print("Exiting predictor. Vanakkam!") | |
| break | |
| if user_input.lower() == "samples": | |
| print("\n--- Running Built-In Sample Comments ---") | |
| for sample in sample_comments: | |
| res = predict_comment(sample, model) | |
| print_prediction_result(res) | |
| continue | |
| res = predict_comment(user_input, model) | |
| print_prediction_result(res) | |
| if __name__ == "__main__": | |
| main() | |