Spaces:
Running
Running
| """ | |
| Abstract -> ASJC subject codes, for populating ResearchBee's recommendation filter. | |
| from asjc_classifier import AsjcClassifier | |
| clf = AsjcClassifier("asjc_model.joblib") # local model only | |
| clf = AsjcClassifier("asjc_model.joblib", llm=True) # + LLM adjudication | |
| result = clf.classify(title=..., abstract=...) | |
| result.codes # ['3309', '1710'] — feed straight to recommend(asjc=...) | |
| result.method # 'linear+llm' | 'linear-topk' | |
| result.candidates # what the LLM was asked to choose between | |
| Architecture, and why it is not the obvious one | |
| ----------------------------------------------- | |
| The obvious design is "run the linear model, escalate to an LLM when confidence | |
| is low". Measured on the held-out abstract set, that design cannot work: the | |
| top-1 probability averages 0.367 when the prediction is correct and 0.330 when | |
| it is wrong. There is no threshold separating them — the model is wrong often | |
| and unbothered about it. | |
| What the model *is* good at is narrowing. A correct code appears within: | |
| top-1 30% | |
| top-3 70% | |
| top-5 95% <- and flat thereafter | |
| So it is used as a candidate generator, not a decider. The linear model cuts 334 | |
| codes to 5 with ~95% recall; the LLM picks among those 5. That inverts the usual | |
| cost argument in our favour — the prompt carries five labelled options rather | |
| than a 334-entry taxonomy, so the call is small, fast and hard to get wrong. | |
| Degradation without an API key is graceful: `codes` becomes the top 3, which the | |
| recommender applies as an OR filter. Recall is 70% rather than 95%, and the | |
| candidate pool is broader, but nothing breaks. | |
| Retrain after the OpenAlex harvest | |
| ---------------------------------- | |
| The shipped model is trained on journal titles, because that is the labelled text | |
| available without a network. It is applied to abstracts. `harvest_openalex.py` | |
| produces the matched corpus; retraining on it and re-running `asjc_eval.jsonl` | |
| measures the gain on the same 20 abstracts, so the upgrade is quantified rather | |
| than assumed. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| import joblib | |
| import numpy as np | |
| from scipy import sparse | |
| __all__ = ["AsjcClassifier", "AsjcResult"] | |
| #: Where the linear model's recall saturates. Measured, not guessed. | |
| DEFAULT_K = 5 | |
| #: Codes returned when no LLM is available. top-3 = 70% recall. | |
| FALLBACK_K = 3 | |
| class AsjcResult: | |
| codes: list[str] | |
| names: list[str] | |
| method: str | |
| candidates: list[dict] = field(default_factory=list) | |
| llm_error: str | None = None | |
| def __bool__(self) -> bool: | |
| return bool(self.codes) | |
| _PROMPT = """You are classifying an academic manuscript into Scopus ASJC subject areas. | |
| A statistical model has narrowed 334 possible subject areas to the {n} candidates \ | |
| below. One or more of them is correct. | |
| Candidates: | |
| {candidates} | |
| Manuscript title: {title} | |
| Abstract: {abstract} | |
| Choose the 1-3 candidates that best describe where this manuscript belongs, most \ | |
| appropriate first. Choose only from the candidate list — do not invent codes. If \ | |
| the manuscript is genuinely interdisciplinary, return more than one; if it clearly \ | |
| belongs to a single area, return just that one. | |
| Respond with JSON only, no other text: | |
| {{"codes": ["NNNN", ...], "reason": "one short sentence"}}""" | |
| class AsjcClassifier: | |
| def __init__( | |
| self, | |
| model_path: str | Path = "asjc_model.joblib", | |
| llm: bool = False, | |
| llm_model: str | None = None, | |
| api_key: str | None = None, | |
| ): | |
| self.bundle = joblib.load(model_path) | |
| self.classes: np.ndarray = self.bundle["classes"] | |
| self.names: dict[str, str] = self.bundle["names"] | |
| self.parent_of: dict[str, str] = self.bundle["parent_of"] | |
| self.trained_on: str = self.bundle.get("corpus", "unknown") | |
| self._client = None | |
| # Model id is configuration, not a constant — set ASJC_LLM_MODEL to | |
| # whichever model you are entitled to. Verify it against your account's | |
| # available models rather than trusting a default baked in here. | |
| self.llm_model = llm_model or os.getenv("ASJC_LLM_MODEL", "claude-sonnet-4-5") | |
| if llm: | |
| key = api_key or os.getenv("ANTHROPIC_API_KEY") | |
| if key: | |
| try: | |
| import anthropic | |
| self._client = anthropic.Anthropic(api_key=key) | |
| except ImportError: | |
| pass # falls back to top-k; reported in AsjcResult.method | |
| # -- stage 1: narrow ---------------------------------------------------- | |
| def candidates(self, text: str, k: int = DEFAULT_K) -> list[dict]: | |
| """The linear model's top-k, with names attached. ~95% recall at k=5.""" | |
| X = sparse.hstack([ | |
| self.bundle["word"].transform([text]), | |
| self.bundle["char"].transform([text]), | |
| ]).tocsr() | |
| p = self.bundle["clf"].predict_proba(X)[0] | |
| order = np.argsort(-p)[:k] | |
| return [ | |
| { | |
| "code": str(self.classes[i]), | |
| "name": self.names.get(str(self.classes[i]), str(self.classes[i])), | |
| "score": round(float(p[i]), 4), | |
| } | |
| for i in order | |
| ] | |
| # -- stage 2: adjudicate ------------------------------------------------ | |
| def classify( | |
| self, | |
| title: str = "", | |
| abstract: str = "", | |
| k: int = DEFAULT_K, | |
| max_codes: int = 3, | |
| ) -> AsjcResult: | |
| text = f"{title} {title} {abstract}".strip() | |
| if not text: | |
| return AsjcResult([], [], "empty") | |
| cands = self.candidates(text, k=k) | |
| if not cands: | |
| return AsjcResult([], [], "no-candidates") | |
| if self._client is None: | |
| picked = [c["code"] for c in cands[:FALLBACK_K]] | |
| return AsjcResult(picked, [self.names.get(c, c) for c in picked], | |
| "linear-topk", cands) | |
| listing = "\n".join(f" {c['code']} {c['name']}" for c in cands) | |
| prompt = _PROMPT.format( | |
| n=len(cands), candidates=listing, | |
| title=title or "(not supplied)", | |
| abstract=(abstract or "(not supplied)")[:4000], | |
| ) | |
| try: | |
| msg = self._client.messages.create( | |
| model=self.llm_model, | |
| max_tokens=300, | |
| temperature=0, | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| raw = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text") | |
| picked = self._parse(raw, {c["code"] for c in cands})[:max_codes] | |
| if picked: | |
| return AsjcResult(picked, [self.names.get(c, c) for c in picked], | |
| "linear+llm", cands) | |
| err = "LLM returned no valid code from the candidate set" | |
| except Exception as e: # noqa: BLE001 | |
| err = f"{type(e).__name__}: {e}" | |
| # Any LLM failure degrades to the statistical answer rather than raising. | |
| # A broader-than-ideal subject filter is a far better outcome for the | |
| # caller than a recommendation request that errors out. | |
| picked = [c["code"] for c in cands[:FALLBACK_K]] | |
| return AsjcResult(picked, [self.names.get(c, c) for c in picked], | |
| "linear-topk", cands, llm_error=err) | |
| def _parse(raw: str, allowed: set[str]) -> list[str]: | |
| """ | |
| Pull codes out of the reply and hard-filter to the candidate set. | |
| The filter is the point: it makes an invented or hallucinated code | |
| structurally impossible to return, so the worst case is fewer codes | |
| rather than a wrong one that silently poisons the journal filter. | |
| """ | |
| codes: list[str] = [] | |
| m = re.search(r"\{.*\}", raw, re.S) | |
| if m: | |
| try: | |
| codes = [str(c).strip() for c in json.loads(m.group(0)).get("codes", [])] | |
| except (json.JSONDecodeError, AttributeError, TypeError): | |
| codes = [] | |
| if not codes: | |
| codes = re.findall(r"\b\d{4}\b", raw) | |
| out: list[str] = [] | |
| for c in codes: | |
| if c in allowed and c not in out: | |
| out.append(c) | |
| return out | |