| """Zero-shot complaint classifier.
|
|
|
| Zero-shot (NLI) is what makes runtime labels possible: the candidate labels are
|
| an *input* to every inference call, so adding "Street Lighting" in the UI takes
|
| effect on the very next complaint -- no retraining, no restart.
|
|
|
| If transformers/torch are unavailable or the model cannot be downloaded, a
|
| keyword-overlap fallback keeps the app usable; every response reports which
|
| engine produced it.
|
| """
|
| from __future__ import annotations
|
|
|
| import logging
|
| import math
|
| import re
|
| import threading
|
| import time
|
| from typing import Any
|
|
|
| from .config import (
|
| CONFIDENCE_THRESHOLD,
|
| HYPOTHESIS_TEMPLATE,
|
| LEXICAL_WEIGHT,
|
| MODEL_NAME,
|
| )
|
|
|
| log = logging.getLogger("classifier")
|
|
|
|
|
|
|
|
|
| _load_lock = threading.Lock()
|
| _infer_lock = threading.Lock()
|
| _pipe: Any = None
|
| _state: dict[str, Any] = {
|
| "status": "not_loaded",
|
| "engine": "none",
|
| "model": MODEL_NAME,
|
| "error": "",
|
| "load_seconds": 0.0,
|
| }
|
|
|
|
|
| MAX_CANDIDATE_WORDS = 16
|
| _WORD_RE = re.compile(r"[a-z0-9']+")
|
| _STOPWORDS = {
|
| "the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "to", "of",
|
| "in", "on", "at", "for", "with", "and", "or", "not", "no", "my", "our", "i", "we",
|
| "it", "this", "that", "there", "here", "from", "by", "as", "has", "have", "had",
|
| "do", "does", "did", "but", "so", "very", "please", "sir", "madam", "any", "some",
|
| }
|
|
|
|
|
| def state() -> dict[str, Any]:
|
| return dict(_state)
|
|
|
|
|
| def is_ready() -> bool:
|
| return _state["status"] == "ready"
|
|
|
|
|
| def load_model(force: bool = False) -> dict[str, Any]:
|
| """Load the zero-shot pipeline. Safe to call repeatedly and concurrently."""
|
| global _pipe
|
| with _load_lock:
|
| if _pipe is not None and not force:
|
| return state()
|
| _state.update(status="loading", error="")
|
| started = time.perf_counter()
|
| try:
|
| from transformers import pipeline as hf_pipeline
|
|
|
| try:
|
| import torch
|
|
|
| torch.set_num_threads(max(1, (torch.get_num_threads() or 2) - 1))
|
| except Exception:
|
| pass
|
|
|
| _pipe = hf_pipeline(
|
| "zero-shot-classification",
|
| model=MODEL_NAME,
|
| device=-1,
|
| )
|
| _state.update(
|
| status="ready",
|
| engine=f"zero-shot:{MODEL_NAME}",
|
| load_seconds=round(time.perf_counter() - started, 2),
|
| error="",
|
| )
|
| log.info("Loaded %s in %.2fs", MODEL_NAME, _state["load_seconds"])
|
| except Exception as exc:
|
| _pipe = None
|
| _state.update(
|
| status="failed",
|
| engine="keyword-fallback",
|
| error=f"{type(exc).__name__}: {exc}",
|
| load_seconds=round(time.perf_counter() - started, 2),
|
| )
|
| log.warning("Model load failed (%s); using keyword fallback.", exc)
|
| return state()
|
|
|
|
|
| def _candidate_for(label: dict[str, Any]) -> str:
|
| """The NLI hypothesis phrase for a label -- the label NAME, kept short.
|
|
|
| Feeding the description in here instead was measurably worse: on a 6-case
|
| benchmark, descriptions as candidates scored 0/6 against 3/6 for bare names.
|
| An NLI hypothesis wants a clean noun phrase ("Water Supply"), not a
|
| comma-separated keyword dump. Descriptions are still used -- as a lexical
|
| prior blended into the scores, see _blend().
|
| """
|
| name = (label.get("name") or "").strip()
|
| words = name.split()
|
| if len(words) > MAX_CANDIDATE_WORDS:
|
| name = " ".join(words[:MAX_CANDIDATE_WORDS])
|
| return name
|
|
|
|
|
| def _blend(
|
| model_scores: list[tuple[str, float]],
|
| text: str,
|
| labels: list[dict[str, Any]],
|
| weight: float,
|
| multi_label: bool = False,
|
| ) -> list[tuple[str, float]]:
|
| """Mix the model scores with a keyword prior built from label descriptions.
|
|
|
| The descriptions carry real domain signal ("burst pipe", "lamp post") that a
|
| small NLI model does not get from the label name alone. Blending keeps the
|
| model in charge while letting those hints break ties.
|
|
|
| Skipped entirely in multi-label mode: there each score is an INDEPENDENT
|
| probability, while the lexical prior is a distribution summing to 1 across
|
| labels. Mixing the two would systematically depress every score and the
|
| renormalisation below would destroy the independence the caller asked for.
|
| """
|
| if weight <= 0 or multi_label:
|
| return model_scores
|
|
|
| lexical = dict(_keyword_scores(text, labels))
|
|
|
| if max(lexical.values(), default=0.0) - min(lexical.values(), default=0.0) < 1e-6:
|
| return model_scores
|
|
|
| blended = [
|
| (name, (1.0 - weight) * score + weight * lexical.get(name, 0.0))
|
| for name, score in model_scores
|
| ]
|
| total = sum(s for _, s in blended)
|
| if total > 0:
|
| blended = [(n, s / total) for n, s in blended]
|
| return blended
|
|
|
|
|
| def _tokens(text: str) -> set[str]:
|
| return {w for w in _WORD_RE.findall(text.lower()) if w not in _STOPWORDS and len(w) > 2}
|
|
|
|
|
| def _keyword_scores(text: str, labels: list[dict[str, Any]]) -> list[tuple[str, float]]:
|
| """Fallback scorer: overlap between complaint tokens and label tokens."""
|
| text_tokens = _tokens(text)
|
| raw = []
|
| for label in labels:
|
| label_tokens = _tokens(f"{label.get('name','')} {label.get('description','')}")
|
| if not label_tokens:
|
| raw.append(0.0)
|
| continue
|
| hits = sum(1 for t in label_tokens if t in text_tokens)
|
|
|
| near = sum(
|
| 1 for t in label_tokens
|
| if t not in text_tokens and any(t.startswith(u[:4]) or u.startswith(t[:4])
|
| for u in text_tokens if len(u) > 3)
|
| )
|
| raw.append(hits + 0.35 * near)
|
|
|
| if max(raw, default=0.0) <= 0:
|
| even = 1.0 / len(labels) if labels else 0.0
|
| return [(l["name"], even) for l in labels]
|
|
|
| exps = [math.exp(s) for s in raw]
|
| total = sum(exps)
|
| return sorted(
|
| ((l["name"], e / total) for l, e in zip(labels, exps)),
|
| key=lambda p: p[1],
|
| reverse=True,
|
| )
|
|
|
|
|
| def classify(
|
| text: str,
|
| labels: list[dict[str, Any]],
|
| multi_label: bool = False,
|
| threshold: float | None = None,
|
| ) -> dict[str, Any]:
|
| """Classify `text` against `labels` (list of {name, description}).
|
|
|
| Returns predicted label, per-label scores, engine used and latency.
|
| """
|
| threshold = CONFIDENCE_THRESHOLD if threshold is None else threshold
|
| text = (text or "").strip()
|
| started = time.perf_counter()
|
|
|
| if not text:
|
| raise ValueError("Complaint text is empty.")
|
| if not labels:
|
| raise ValueError("No active labels configured. Add at least one label first.")
|
|
|
| if _pipe is None and _state["status"] == "not_loaded":
|
| load_model()
|
|
|
| pipe = _pipe
|
| if pipe is not None:
|
|
|
|
|
| candidates = list(dict.fromkeys(_candidate_for(l) for l in labels if _candidate_for(l)))
|
| try:
|
| with _infer_lock:
|
| out = pipe(
|
| text,
|
| candidate_labels=candidates,
|
| hypothesis_template=HYPOTHESIS_TEMPLATE,
|
| multi_label=multi_label,
|
| )
|
| scored = [(lbl, float(score))
|
| for lbl, score in zip(out["labels"], out["scores"])]
|
| scored = _blend(scored, text, labels, LEXICAL_WEIGHT, multi_label)
|
| engine = _state["engine"]
|
| except Exception as exc:
|
| log.warning("Inference failed (%s); falling back to keywords.", exc)
|
| scored = _keyword_scores(text, labels)
|
| engine = "keyword-fallback (inference error)"
|
| else:
|
| scored = _keyword_scores(text, labels)
|
| engine = "keyword-fallback"
|
|
|
| scored.sort(key=lambda p: p[1], reverse=True)
|
| top_label, top_score = scored[0]
|
| took_ms = int((time.perf_counter() - started) * 1000)
|
|
|
| return {
|
| "predicted_label": top_label,
|
| "confidence": round(top_score, 4),
|
| "confident": top_score >= threshold,
|
| "threshold": threshold,
|
| "scores": [{"label": n, "score": round(s, 4)} for n, s in scored],
|
| "engine": engine,
|
| "multi_label": multi_label,
|
| "took_ms": took_ms,
|
| }
|
|
|
