"""LLM-as-judge grading: one semantic judge for every question type and axis.""" from __future__ import annotations import logging from .llm_client import MAX_ATTEMPTS from .prompts import JUDGE_SYS, JUDGE_USER, qtext logger = logging.getLogger(__name__) def gold_text(q_raw: dict) -> str: """Return the question's `answer`, or the correct options of a multiple-choice question.""" if q_raw.get("answer"): return q_raw["answer"] cor = [o for o in q_raw.get("options", []) if o.get("is_correct")] return "; ".join(f"{o['label']}. {o['content']}" for o in cor) if cor else str(q_raw.get("correct_option", "")) def score(client, q, cand: str): """(ok, verdict, detail) of one answer; the verdict is 'match', 'no_match', or '?' after MAX_ATTEMPTS failed judge calls.""" msg = JUDGE_USER.format(question=qtext(q.raw), gold=gold_text(q.raw), cand=cand) last_err = "no exception recorded" for _ in range(MAX_ATTEMPTS): try: v = client.complete_json(JUDGE_SYS, msg).get("match") if isinstance(v, str): v = v.strip().lower() in ("true", "yes", "1") if v is None: continue return bool(v), ("match" if v else "no_match"), {} except Exception as e: last_err = f"{type(e).__name__}: {e}"[:200] logger.warning("[judge] attempt failed (%s), retrying", last_err) logger.error("[judge] gave up after %d attempts; scoring as verdict '?': %s", MAX_ATTEMPTS, last_err) return False, "?", {"judge_error": last_err}