File size: 1,736 Bytes
62d0c8c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Theme prediction without the full BERTopic stack: embed the review, then cosine-match it
against the exported topic vectors (artifacts/topic_index.*). Pure functions (no Streamlit) so
they can be unit-tested. The heavy model loads once and is cached as a module-level singleton.
"""
import os
import json
import numpy as np
from utils.paths import ARTIFACTS
from utils.preprocess import clean_review
_E = None # normalized topic-vector matrix (n_topics x dim)
_meta = None # list of {topic, theme, sentiment}
_model = None # SentenceTransformer
def _load():
global _E, _meta, _model
if _model is None:
from sentence_transformers import SentenceTransformer
_E = np.load(os.path.join(ARTIFACTS, "topic_index.npz"))["embeddings"]
cfg = json.load(open(os.path.join(ARTIFACTS, "topic_index.json"), encoding="utf-8"))
_meta = cfg["topics"]
_model = SentenceTransformer(cfg["model_name"])
return _E, _meta, _model
def predict_theme(text: str, topk: int = 3):
"""Return (cleaned_text, [(sentiment, theme, similarity), ...]) ranked best-first.
Similarity is aggregated to theme level (best matching topic per theme)."""
E, meta, model = _load()
cleaned = clean_review(text)
if not cleaned:
return cleaned, []
q = model.encode([cleaned])[0]
q = q / (np.linalg.norm(q) + 1e-9)
sims = E @ q # cosine (E is pre-normalized)
best = {}
for i, t in enumerate(meta):
key = (t["sentiment"], t["theme"])
if key not in best or sims[i] > best[key]:
best[key] = float(sims[i])
ranked = sorted(best.items(), key=lambda kv: -kv[1])
return cleaned, [(s, th, sim) for (s, th), sim in ranked[:topk]]
|