| """Simple evaluators — clean-room reimplementation of LlamaIndex-style evals. |
| |
| Reimplemented from the *behavior* of `llama_index.core.evaluation` (MIT) — never copied. |
| A minimal set for the "did the trained model recall the training / maintain the persona" |
| question: |
| |
| - `SemanticSimilarityEvaluator` — embedding/lexical cosine (reuses `eval.imprint`). |
| - `CorrectnessEvaluator` — LLM-as-judge of a response vs a reference. |
| - `PairwiseEvaluator` — which of two responses (before vs after) better matches the persona. |
| - `GuidelineEvaluator` — does a response comply with a rubric / guidelines. |
| |
| The LLM-judge evaluators call the existing chat backend via `governance.panel.chat_once` |
| (ollama/vLLM). Pure stdlib + pydantic + httpx; base-install importable. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import re |
|
|
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| _SCORE_RE = re.compile(r"score\s*[:=]?\s*([0-5](?:\.\d+)?)", re.IGNORECASE) |
| _CHOICE_RE = re.compile(r"\b(A|B|TIE)\b", re.IGNORECASE) |
| _DEFAULT_JUDGE = "llama3.2" |
|
|
|
|
| def default_judge_model() -> str: |
| """`MINDXTRAIN_JUDGE_MODEL`, else the panel's chat model, else llama3.2 — read at call |
| time so a bankml node judges with a model it actually serves.""" |
| import os |
|
|
| from mindxtrain.governance.panel import default_chat_model |
|
|
| return os.environ.get("MINDXTRAIN_JUDGE_MODEL") or default_chat_model() |
|
|
|
|
| class EvalScore(BaseModel): |
| """A single evaluation result, score normalized to [0, 1].""" |
|
|
| model_config = ConfigDict(extra="forbid", frozen=True) |
|
|
| score: float = Field(ge=0.0, le=1.0) |
| passing: bool |
| reasoning: str = "" |
| method: str |
|
|
|
|
| def _parse_judge_score(text: str) -> tuple[float, str]: |
| """Parse a 1-5 `SCORE: N` from judge output → ([0,1], one-line reasoning).""" |
| m = _SCORE_RE.search(text) |
| raw = float(m.group(1)) if m else 2.5 |
| raw = max(1.0, min(5.0, raw)) |
| score01 = (raw - 1.0) / 4.0 |
| reasoning = next((ln.strip() for ln in text.splitlines() if ln.strip()), text.strip())[:240] |
| return round(score01, 4), reasoning |
|
|
|
|
| def _judge(model: str, system: str, user: str, *, base_url: str | None) -> tuple[float, str]: |
| """Run one LLM-judge call; returns ([0,1], reasoning). Best-effort (0.5 on error).""" |
| from mindxtrain.governance.panel import chat_once |
|
|
| try: |
| text = chat_once( |
| model, |
| [{"role": "system", "content": system}, {"role": "user", "content": user}], |
| base_url=base_url, max_tokens=300, temperature=0.0, |
| ) |
| except Exception as exc: |
| return 0.5, f"judge error: {exc}" |
| return _parse_judge_score(text) |
|
|
|
|
| class SemanticSimilarityEvaluator: |
| """Embedding/lexical similarity between two texts (reuses `eval.imprint`).""" |
|
|
| def __init__(self, threshold: float = 0.7) -> None: |
| self.threshold = threshold |
|
|
| def evaluate(self, text_a: str, text_b: str) -> EvalScore: |
| from mindxtrain.eval.imprint import _voice_similarity |
|
|
| score, method = _voice_similarity([text_a], [text_b]) |
| return EvalScore( |
| score=round(score, 4), passing=score >= self.threshold, |
| reasoning=f"similarity {score:.3f} vs threshold {self.threshold}", method=method, |
| ) |
|
|
|
|
| class CorrectnessEvaluator: |
| """LLM-as-judge: is RESPONSE correct/faithful vs REFERENCE for the QUERY?""" |
|
|
| def __init__(self, model: str | None = None, base_url: str | None = None, threshold: float = 0.6) -> None: |
| self.model, self.base_url, self.threshold = model or default_judge_model(), base_url, threshold |
|
|
| def evaluate(self, query: str, response: str, reference: str) -> EvalScore: |
| system = ( |
| "You are a strict evaluator. Score from 1 (wrong) to 5 (perfect) how well the " |
| "RESPONSE matches the REFERENCE answer for the QUERY. Give one sentence of " |
| "reasoning, then a final line 'SCORE: N'." |
| ) |
| user = f"QUERY: {query}\nREFERENCE: {reference}\nRESPONSE: {response}" |
| score, reasoning = _judge(self.model, system, user, base_url=self.base_url) |
| return EvalScore(score=score, passing=score >= self.threshold, reasoning=reasoning, |
| method=f"llm-judge:{self.model}") |
|
|
|
|
| class PairwiseEvaluator: |
| """Which of two responses better matches the persona/reference? B (after) vs A (before).""" |
|
|
| def __init__(self, model: str | None = None, base_url: str | None = None) -> None: |
| self.model, self.base_url = model or default_judge_model(), base_url |
|
|
| def evaluate(self, query: str, response_a: str, response_b: str, *, reference: str = "") -> EvalScore: |
| ref = f" The target voice/reference is: {reference}." if reference else "" |
| system = ( |
| "You compare two assistant responses (A and B) to a query." + ref + |
| " Decide which better matches the target voice. Give one sentence, then a final " |
| "line with exactly 'A', 'B', or 'TIE'." |
| ) |
| user = f"QUERY: {query}\nA: {response_a}\nB: {response_b}" |
| from mindxtrain.governance.panel import chat_once |
|
|
| try: |
| text = chat_once( |
| self.model, |
| [{"role": "system", "content": system}, {"role": "user", "content": user}], |
| base_url=self.base_url, max_tokens=200, temperature=0.0, |
| ) |
| except Exception as exc: |
| return EvalScore(score=0.5, passing=False, reasoning=f"judge error: {exc}", |
| method=f"llm-judge:{self.model}") |
| |
| choices = _CHOICE_RE.findall(text) |
| verdict = (choices[-1].upper() if choices else "TIE") |
| score = {"B": 1.0, "TIE": 0.5, "A": 0.0}[verdict] |
| reasoning = next((ln.strip() for ln in text.splitlines() if ln.strip()), text.strip())[:240] |
| return EvalScore(score=score, passing=verdict == "B", reasoning=f"verdict {verdict}: {reasoning}", |
| method=f"llm-judge:{self.model}") |
|
|
|
|
| class GuidelineEvaluator: |
| """LLM-as-judge: does RESPONSE comply with the GUIDELINES/rubric?""" |
|
|
| def __init__(self, model: str | None = None, base_url: str | None = None, threshold: float = 0.6) -> None: |
| self.model, self.base_url, self.threshold = model or default_judge_model(), base_url, threshold |
|
|
| def evaluate(self, response: str, guidelines: str) -> EvalScore: |
| system = ( |
| "You check compliance with guidelines. Score from 1 (violates) to 5 (fully " |
| "complies) how well the RESPONSE follows the GUIDELINES. One sentence, then " |
| "'SCORE: N'." |
| ) |
| user = f"GUIDELINES: {guidelines}\nRESPONSE: {response}" |
| score, reasoning = _judge(self.model, system, user, base_url=self.base_url) |
| return EvalScore(score=score, passing=score >= self.threshold, reasoning=reasoning, |
| method=f"llm-judge:{self.model}") |
|
|
|
|
| __all__ = [ |
| "CorrectnessEvaluator", |
| "EvalScore", |
| "GuidelineEvaluator", |
| "PairwiseEvaluator", |
| "SemanticSimilarityEvaluator", |
| ] |
|
|
