Spaces:
Running
Running
| """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, | |
| } | |