pattern_namer.py

import re, json, os, sys

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE)

_cached_model      = None
_cached_tokenizer  = None
_cached_model_name = None
_model_registry    = {}

NAMING_PROMPT = (
    "You are a hate speech categorization expert for Ethiopian social media.\n"
    "Analyze the shared harmful pattern in these texts and suggest a NEW specific category name.\n"
    "\nExisting categories already covered:\n"
    "- Violence & Extremism, Identity-Based Hate, Derogation & Slurs,\n"
    "  Gender-Based Hate, Stereotype & Discrimination\n"
    "\nRespond with ONLY this JSON:\n"
    "{\n"
    '  \"name\": \"<concise new category name, 2-4 words>\",\n'
    '  \"rationale\": \"<1-2 sentences: what specific harmful pattern do these texts share?>\",\n'
    '  \"confidence\": <0.0-1.0>,\n'
    '  \"sample_terms\": [\"<key term 1>\", \"<key term 2>\", \"<key term 3>\"]\n'
    "}\n"
    "If fits existing category, set confidence below 0.4."
)


def _detect_language(texts):
    am = sum(sum(1 for c in str(t) if "\u1200" <= c <= "\u137f") for t in texts)
    return "amharic" if am > len(texts) * 3 else "english"


def name_pattern(representative_texts, cluster_info=None):
    if not representative_texts:
        return None
    _self = sys.modules[__name__]
    _reg  = getattr(_self, "_model_registry", {})
    if "Qwen/Qwen2.5-1.5B-Instruct" in _reg:
        model_name = "Qwen/Qwen2.5-1.5B-Instruct"
    elif "CohereLabs/aya-expanse-8b" in _reg:
        model_name = "CohereLabs/aya-expanse-8b"
    else:
        model_name = "Qwen/Qwen2.5-1.5B-Instruct"

    sep = chr(10)
    texts_block  = sep.join(f"- {t[:200]}" for t in representative_texts[:8])
    user_content = f"Texts to analyze:{sep}{sep}{texts_block}{sep}{sep}JSON:"

    try:
        import torch
        from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
        from huggingface_hub import login as _hf_login
        hf_token = os.environ.get("HF_TOKEN", None)
        if hf_token:
            _hf_login(token=hf_token, add_to_git_credential=False)
        if not hasattr(_self, "_model_registry"):
            _self._model_registry = {}
        if model_name in _self._model_registry:
            tokenizer, model = _self._model_registry[model_name]
            print(f"  [pattern_namer] Reusing {model_name.split(chr(47))[-1]}")
        else:
            print(f"  [pattern_namer] Loading {model_name}...")
            tokenizer = AutoTokenizer.from_pretrained(model_name, token=hf_token)
            if tokenizer.pad_token is None:
                tokenizer.pad_token = tokenizer.eos_token
            tokenizer.padding_side = "left"
            try:
                bnb = BitsAndBytesConfig(
                    load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16,
                    bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True,
                )
                model = AutoModelForCausalLM.from_pretrained(
                    model_name, quantization_config=bnb, device_map="auto", token=hf_token
                )
            except Exception:
                model = AutoModelForCausalLM.from_pretrained(
                    model_name, torch_dtype=torch.bfloat16, device_map="auto", token=hf_token
                )
            model.eval()
            _self._cached_model = model
            _self._cached_tokenizer = tokenizer
            _self._cached_model_name = model_name
            _self._model_registry[model_name] = (tokenizer, model)

        messages = [
            {"role": "system", "content": NAMING_PROMPT},
            {"role": "user",   "content": user_content},
        ]
        try:
            prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
        except Exception:
            prompt = NAMING_PROMPT + chr(10) + chr(10) + user_content

        inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=1024).to(model.device)
        with torch.no_grad():
            gen_ids = model.generate(**inputs, max_new_tokens=200, do_sample=False,
                                     pad_token_id=tokenizer.pad_token_id)
        new_tokens = gen_ids[0][inputs["input_ids"].shape[1]:]
        raw = tokenizer.decode(new_tokens, skip_special_tokens=True)

        m = re.search(r"{.*}", raw, re.DOTALL)
        if m:
            data = json.loads(m.group(0))
            return {
                "name":         data.get("name", "Emerging Pattern"),
                "rationale":    data.get("rationale", ""),
                "confidence":   float(max(0.0, min(1.0, data.get("confidence", 0.5)))),
                "sample_terms": data.get("sample_terms", []),
                "model_used":   model_name.split(chr(47))[-1],
            }
        return {"name": "Emerging Pattern", "rationale": raw[:200],
                "confidence": 0.5, "sample_terms": [],
                "model_used": model_name.split(chr(47))[-1]}

    except Exception as e:
        print(f"  [pattern_namer] LLM unavailable: {e}")
        # On CPU deployment: save pattern for human review without a name
        # The admin can name it manually at /admin/patterns/
        return {
            "name": "Review Required",
            "rationale": "Pattern detected in cluster. LLM unavailable on CPU -- please name this pattern manually.",
            "confidence": 0.40,
            "sample_terms": [],
            "model_used": "cpu_fallback",
        }