"""NeuroJenML evaluation harness. Replaces the old benchmark, which scored *whether a human typed an answer* and, at best, regex-scraped a base model. Two real capabilities here: 1. GENERATION — the model under test answers each held-out query itself. 2. JUDGING — an LLM judge scores each answer against a ground-truth reference using a strict rubric and returns structured JSON (no scraping). Both use the current Hugging Face Router API (OpenAI-compatible chat completions) — NOT the deprecated api-inference serverless endpoint. Backends are swappable (see ARCHITECTURE.md section 2): today we query hosted models via the router; a Kaggle/endpoint backend can serve a fine-tuned adapter later by implementing `generate()`. """ from __future__ import annotations import os import json import re from typing import Optional # ── Domain-drift detection ──────────────────────────────────────────────────── # Keywords that should NOT appear prominently in a neuro/AD response unless the # query explicitly targets those domains. We check for cardiovascular and # gut-microbiome contamination — the two known noise sources from the training set. _CARDIO_KEYWORDS = frozenset({ "myocardial infarction", "mi ", " mi,", "heart attack", "hydroxytyrosol", "atherosclerosis", "coronary", "statin", "ldl", "hdl", "cardiomyopathy", "cardiac", "platelet aggregation", "thrombus", "thrombosis", }) _GUT_KEYWORDS = frozenset({ "gut microbiome", "microbiota", "lactobacillus", "bifidobacterium", "short-chain fatty acid", "scfa", "colonic", "intestinal permeability", "leaky gut", "dysbiosis", }) # Keywords that signal a response is likely on-target for neuro/AD queries. _NEURO_KEYWORDS = frozenset({ "alzheimer", "amyloid", "tau", "neuroinflammation", "microglia", "synapse", "hippocamp", "cortex", "bbq", "blood-brain barrier", "neurodegeneration", "cognitive", "dementia", }) # HTML / structural artifact patterns left over after sanitization failures. _ARTIFACT_RE = re.compile( r"(<[a-z/][^>]*>)" # residual HTML tags r"|(\\u003[ce])" # unicode-escaped brackets r"|([}\]\"]{3,})" # cascade artefacts }}}}, ]]]] r"|(/>)" # stray self-closing , re.IGNORECASE ) def _response_quality_score(text: str, query: str = "") -> float: """Quality signal for a generated answer (0.0–1.0). Extends the original unique-ratio/length check with two additional penalty dimensions that the original missed: domain_drift_penalty Fires when a response to a neuro/AD query contains cardiovascular or gut-microbiome keywords with no matching neuro anchors. The observed symptom was: query about hypertension → response describes Hydroxytyrosol for APP/PS1 mice. Penalty: up to −0.35. artifact_penalty Counts residual HTML tags, unicode bracket escapes, and cascade artefacts (}}}}, ]]]]). Each hit deducts 0.05, capped at −0.30. """ if not text: return 0.0 cleaned = re.sub(r"\s+", " ", text).strip() words = cleaned.split() if not words: return 0.0 # Base: unique-word ratio (repetition guard). # Scaled by 2x so severe repetition (unique_ratio -> 0, e.g. an # attention-sink loop repeating one phrase) can fully cancel out # length_score (max 1.0) instead of being capped at -0.5 — a long # degenerate loop must not out-score a short unique answer. repetition_penalty = 0.0 if len(words) > 8: unique_ratio = len(set(words)) / len(words) repetition_penalty = max(0.0, 0.5 - unique_ratio) * 2 length_score = min(1.0, len(words) / 25.0) base = length_score - repetition_penalty # Domain-drift penalty — only meaningful when query is provided. domain_drift_penalty = 0.0 if query: q_lower = query.lower() r_lower = text.lower() # Determine whether the query is neuro-focused. is_neuro_query = any(kw in q_lower for kw in _NEURO_KEYWORDS) if is_neuro_query: has_neuro_anchor = any(kw in r_lower for kw in _NEURO_KEYWORDS) has_cardio_hit = any(kw in r_lower for kw in _CARDIO_KEYWORDS) has_gut_hit = any(kw in r_lower for kw in _GUT_KEYWORDS) contamination_hits = (1 if has_cardio_hit else 0) + (1 if has_gut_hit else 0) if contamination_hits > 0 and not has_neuro_anchor: # Full contamination: off-domain content, no neuro grounding at all. domain_drift_penalty = 0.35 elif contamination_hits > 0: # Partial contamination: some neuro content but off-domain leakage. domain_drift_penalty = 0.15 * contamination_hits # Artifact penalty — count distinct artifact matches. artifact_hits = len(_ARTIFACT_RE.findall(text)) artifact_penalty = min(0.30, artifact_hits * 0.05) score = base - domain_drift_penalty - artifact_penalty return round(max(0.0, min(1.0, score)), 3) import httpx HF_ROUTER_URL = "https://router.huggingface.co/v1/chat/completions" # Default models (overridable via env). Judge should be a strong instruct model. DEFAULT_GEN_MODEL = os.getenv("EVAL_GEN_MODEL", "google/gemma-2-9b-it") DEFAULT_JUDGE_MODEL = os.getenv("EVAL_JUDGE_MODEL", "meta-llama/Llama-3.3-70B-Instruct") SYSTEM_PROMPT = ( "You are a systemic Alzheimer's disease reasoning model. Given a question, " "answer concisely and mechanistically, linking peripheral (body) factors to " "central (brain) pathology through explicit biological mechanisms." ) JUDGE_RUBRIC = ( "You are a strict biomedical grader. Compare a model ANSWER to a ground-truth " "REFERENCE for an Alzheimer's research question. Score 0-10 on:\n" " - mechanistic_accuracy: are the stated mechanisms correct and aligned with the reference?\n" " - completeness: does it cover the key peripheral->central links in the reference?\n" " - faithfulness: does it avoid fabricated or contradicted claims?\n" "Return ONLY JSON: {\"mechanistic_accuracy\":N,\"completeness\":N," "\"faithfulness\":N,\"overall\":N,\"reason\":\"...\"} where overall is the 0-10 mean." ) class HFRouterBackend: """Generation backend that queries a hosted model via the HF Router API.""" def __init__(self, model_id: str, token: Optional[str] = None): self.model_id = model_id self.token = token or os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_TOKEN") def is_configured(self) -> bool: return bool(self.token) async def chat(self, messages: list, max_tokens: int = 512, temperature: float = 0.2) -> str: if not self.token: raise RuntimeError("HF_TOKEN not set — cannot call the Hugging Face Inference API") import asyncio # ── Tier 1: Dedicated Inference Endpoint (if configured) ── endpoint_url = os.getenv("HF_INFERENCE_ENDPOINT", "") if endpoint_url: from huggingface_hub import AsyncInferenceClient client = AsyncInferenceClient(base_url=endpoint_url, api_key=self.token) resp = await client.chat.completions.create( model=self.model_id, messages=messages, max_tokens=max_tokens, temperature=temperature, ) return resp.choices[0].message.content # ── Tier 2: Free HF Serverless (direct HTTP POST) ── serverless_url = "https://router.huggingface.co/v1/chat/completions" headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "x-wait-for-model": "true", } payload = { "model": self.model_id, "messages": messages, "max_tokens": max_tokens, "temperature": temperature, } max_attempts = 4 last_error = None async with httpx.AsyncClient(timeout=120.0) as http: for attempt in range(max_attempts): resp = await http.post(serverless_url, headers=headers, json=payload) if resp.status_code == 200: data = resp.json() return data["choices"][0]["message"]["content"] if resp.status_code == 503: try: body = resp.json() wait = min(body.get("estimated_time", 30), 60) except Exception: wait = 30 await asyncio.sleep(wait) continue last_error = f"HF Serverless {resp.status_code}: {resp.text[:200]}" break # ── Tier 3: SDK fallback (provider-routed) ── if last_error: from huggingface_hub import AsyncInferenceClient client = AsyncInferenceClient(api_key=self.token) resp = await client.chat.completions.create( model=self.model_id, messages=messages, max_tokens=max_tokens, temperature=temperature, ) return resp.choices[0].message.content raise RuntimeError(last_error or "All inference tiers exhausted") async def generate(self, query: str) -> str: return await self.chat( [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": query}], max_tokens=400, ) def _parse_judge_json(text: str) -> dict: """Robustly extract the judge's JSON object.""" try: return json.loads(text) except Exception: pass match = re.search(r"\{.*\}", text, re.DOTALL) if match: try: return json.loads(match.group(0)) except Exception: pass return {} async def judge_answer(judge: HFRouterBackend, query: str, answer: str, reference: str) -> dict: """Score a single answer against its reference. Returns rubric dict (0-10).""" user = ( f"QUESTION:\n{query}\n\nMODEL ANSWER:\n{answer}\n\n" f"GROUND-TRUTH REFERENCE:\n{reference}" ) raw = await judge.chat( [{"role": "system", "content": JUDGE_RUBRIC}, {"role": "user", "content": user}], max_tokens=300, temperature=0.0, ) parsed = _parse_judge_json(raw) # Derive overall if the judge omitted it. if "overall" not in parsed: parts = [parsed.get(k) for k in ("mechanistic_accuracy", "completeness", "faithfulness") if isinstance(parsed.get(k), (int, float))] parsed["overall"] = round(sum(parts) / len(parts), 2) if parts else 0 return parsed async def run_evaluation( queries: list, model_ref: str, answers: Optional[dict] = None, gen_model: Optional[str] = None, judge_model: Optional[str] = None, ) -> dict: """Evaluate a model on a held-out query set. - If `answers` is provided (e.g. from a fine-tuned adapter run elsewhere), those are judged directly. - Otherwise the configured generation model produces answers itself. Returns per-item rubric scores plus an aggregate (0-100) for the dashboard. """ token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_TOKEN") if not token: return {"error": "HF_TOKEN not configured", "items": [], "aggregate": None} generator = HFRouterBackend(gen_model or DEFAULT_GEN_MODEL, token) judge = HFRouterBackend(judge_model or DEFAULT_JUDGE_MODEL, token) answers = answers or {} items = [] for q in queries: qid = str(q["id"]) try: answer = answers.get(qid) or await generator.generate(q["query"]) rubric = await judge_answer(judge, q["query"], answer, q["reference"]) quality_score = _response_quality_score(answer or "", query=q["query"]) artifact_hits = len(_ARTIFACT_RE.findall(answer or "")) items.append({ "id": q["id"], "query": q["query"], "category": q.get("category"), "answer": answer, "rubric": rubric, "score": float(rubric.get("overall", 0)), "quality_score": quality_score, "artifact_count": artifact_hits, # drift_risk is True when the quality score was penalised by # domain-drift detection (cardio / gut content in a neuro query). "drift_risk": quality_score < _response_quality_score(answer or ""), }) except Exception as e: items.append({ "id": q["id"], "query": q["query"], "category": q.get("category"), "answer": None, "rubric": {}, "score": 0.0, "error": str(e)[:200], }) scored = [it["score"] for it in items if it.get("answer")] aggregate = round((sum(scored) / len(scored)) * 10, 1) if scored else 0.0 # Per-category breakdown for the performance dashboard. by_cat: dict = {} for it in items: cat = it.get("category") or "uncategorized" by_cat.setdefault(cat, []).append(it["score"]) category_scores = {c: round((sum(v) / len(v)) * 10, 1) for c, v in by_cat.items() if v} quality_scores = [it.get("quality_score", 0.0) for it in items if it.get("answer")] avg_quality = round(sum(quality_scores) / len(quality_scores), 3) if quality_scores else 0.0 drift_flagged = sum(1 for it in items if it.get("drift_risk")) total_artifacts = sum(it.get("artifact_count", 0) for it in items) return { "model_ref": model_ref, "gen_model": gen_model or DEFAULT_GEN_MODEL, "judge_model": judge_model or DEFAULT_JUDGE_MODEL, "items": items, "aggregate": aggregate, "category_scores": category_scores, "answered": len(scored), "total": len(items), "avg_quality_score": avg_quality, # Guardrail counters — non-zero values indicate training data issues # that survived into inference: domain drift or residual artifacts. "drift_flagged_count": drift_flagged, "total_artifact_count": total_artifacts, }