| """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} |
|
|
