ELBench-live / tests /test_judges.py
adelevett's picture
Deploy ELBench-live dashboard
5db30ac verified
Raw History Blame Contribute Delete
24.6 kB
import asyncio
import json
import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
sys.path.insert(0, str(SRC))
from elbench.judges.judge_highlevel import HighLevelJudge
from elbench.judges.judge_ifeval import IFEvalJudge
from elbench.judges.judge_math import AIMEJudge, Math500Judge
from elbench.judges.judge_mmlu import MMLUProJudge
from elbench.judges.judge_teaching_harm import TeachingHarmJudge
from elbench.judges.llm_judge import LLMJudge
from elbench.judges.router import JudgeRouter
from elbench.judges.ifeval_rule.instructions_registry import INSTRUCTION_DICT
from elbench.config import load_project_config
from elbench.registry import FileRegistry
from elbench.schemas.config import JudgeTaskConfig
from elbench.schemas.evaluation import ModelResponse, Sample
from elbench.utils import extract_json_object
class JudgeSmokeTest(unittest.TestCase):
def test_llm_judge_returns_structured_result(self) -> None:
config = load_project_config(Path("configs"))
judge = LLMJudge(
config,
JudgeTaskConfig(mode="llm", template="highlevel_edu", judge_model_id="mock.judge"),
)
sample = Sample(
sample_id="l1",
source_file="高阶育人-edu.jsonl",
source_path="高阶育人-edu.jsonl",
module="高阶育人",
subset="edu",
task="highlevel_edu",
prompt='Please provide "Score", "Scoring Details", and "Personalized Feedback" in JSON format.',
metadata={"_scene": "AG"},
)
response = ModelResponse(text='{"Score": "4/5", "Scoring Details": "ok", "Personalized Feedback": "ok"}')
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_router_uses_llm_judge_for_subjective_tasks(self) -> None:
config = load_project_config(Path("configs"))
router = JudgeRouter(config)
sample = Sample(
sample_id="r1",
source_file="安全拒答.jsonl",
source_path="安全拒答.jsonl",
module="安全可信",
subset="通用-应拒答",
task="safety_refusal",
prompt="dummy",
)
judge = router.get_judge(sample)
self.assertIsInstance(judge, LLMJudge)
def test_router_uses_rule_judge_for_highlevel_edu(self) -> None:
config = load_project_config(Path("configs"))
router = JudgeRouter(config)
sample = Sample(
sample_id="edu1",
source_file="高阶育人-edu.jsonl",
source_path="高阶育人-edu.jsonl",
module="高阶育人",
subset="edu",
task="highlevel_edu",
prompt="dummy",
)
judge = router.get_judge(sample)
self.assertIsInstance(judge, HighLevelJudge)
def test_satas_exact_match(self) -> None:
judge = TeachingHarmJudge()
sample = Sample(
sample_id="s1",
source_file="SATAs.xlsx",
source_path="SATAs.xlsx",
module="安全可信",
subset="教育-教学安全",
task="teaching_harm",
prompt="dummy",
reference={"Answer": "A,B,E"},
)
response = ModelResponse(text="A, B, E")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertFalse(result.judge_metadata["partial_credit"])
def test_highlevel_omni_exact_match(self) -> None:
judge = HighLevelJudge()
sample = Sample(
sample_id="o1",
source_file="高阶育人-omni.jsonl",
source_path="高阶育人-omni.jsonl",
module="高阶育人",
subset="omni",
task="highlevel_omni",
prompt="dummy",
reference={"answer": "C", "reasonableness": 1},
)
response = ModelResponse(text="答案:C")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "C")
def test_satas_missing_option_receives_partial_credit(self) -> None:
judge = TeachingHarmJudge()
sample = Sample(
sample_id="s2",
source_file="SATAs.xlsx",
source_path="SATAs.xlsx",
module="安全可信",
subset="教育-教学安全",
task="teaching_harm",
prompt="dummy",
reference={"Answer": "A,B,E"},
)
response = ModelResponse(text="A, E")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.score, 0.5)
self.assertTrue(result.judge_metadata["partial_credit"])
def test_satas_wrong_option_receives_zero_credit(self) -> None:
judge = TeachingHarmJudge()
sample = Sample(
sample_id="s3",
source_file="SATAs.xlsx",
source_path="SATAs.xlsx",
module="安全可信",
subset="教育-教学安全",
task="teaching_harm",
prompt="dummy",
reference={"Answer": "A,B,E"},
)
response = ModelResponse(text="A, C")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.score, 0.0)
self.assertFalse(result.judge_metadata["partial_credit"])
def test_highlevel_omni_multiple_choice_exact_match(self) -> None:
judge = HighLevelJudge()
sample = Sample(
sample_id="o_multi",
source_file="高阶育人-omni.jsonl",
source_path="高阶育人-omni.jsonl",
module="高阶育人",
subset="omni",
task="highlevel_omni",
prompt="dummy",
reference={"answer": "AC", "reasonableness": 1},
)
response = ModelResponse(text="答案:AC")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "AC")
self.assertEqual(result.judge_metadata["expected_answer"], "AC")
def test_llm_judge_respects_model_max_tokens_cap(self) -> None:
config = load_project_config(Path("configs"))
judge = LLMJudge(
config,
JudgeTaskConfig(mode="llm", template="safety_refusal", judge_model_id="qwen-test"),
)
class DummyClient:
def __init__(self) -> None:
self.request = None
async def generate(self, sample, request):
self.request = request
return ModelResponse(text='{"judge_result": "pass", "score": 1, "judge_reason": "ok"}')
async def aclose(self) -> None:
return None
dummy = DummyClient()
judge._client = dummy
sample = Sample(
sample_id="l2",
source_file="安全拒答.jsonl",
source_path="安全拒答.jsonl",
module="安全可信",
subset="通用-应拒答",
task="safety_refusal",
prompt="dummy",
)
response = ModelResponse(text="refuse")
asyncio.run(judge.judge(sample, response))
self.assertIsNotNone(dummy.request)
self.assertEqual(dummy.request.max_tokens, 1024)
self.assertLessEqual(dummy.request.max_tokens, config.models["qwen-test"].max_tokens)
def test_highlevel_edu_ag_json_scoring(self) -> None:
judge = HighLevelJudge()
sample = Sample(
sample_id="ag1",
source_file="高阶育人-edu.jsonl",
source_path="高阶育人-edu.jsonl",
module="高阶育人",
subset="edu",
task="highlevel_edu",
prompt='Please provide "Score", "Scoring Details", and "Personalized Feedback" in JSON format.',
reference={
"Score": "4/5",
"Scoring Details": "The answer is mostly correct but could be more specific.",
"Personalized Feedback": "Good work. Add one concrete example next time.",
},
metadata={"_scene": "AG"},
)
response = ModelResponse(
text='{"Score": "4/5", "Scoring Details": "The answer is mostly correct but could be more specific.", "Personalized Feedback": "Good work. Add one concrete example next time."}'
)
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertGreaterEqual(result.score or 0, 0.99)
def test_mmlu_pro_explicit_answer_match(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("D")
response = ModelResponse(text="Reasoning here.\nANSWER: D")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_extract_json_object_handles_nested_fenced_json(self) -> None:
parsed = extract_json_object(
'```json\n{"Answer": {"CorrectOption": "C", "Explanation": "ok"}}\n```'
)
self.assertEqual(parsed, {"Answer": {"CorrectOption": "C", "Explanation": "ok"}})
def test_mmlu_pro_uses_last_explicit_answer(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("B")
response = ModelResponse(text="A seems tempting at first.\nANSWER: B")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "B")
def test_mmlu_pro_accepts_answer_letter_with_option_text(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("B")
response = ModelResponse(text="Reasoning here.\nANSWER: B) negative")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "B")
def test_mmlu_pro_wrong_answer_fails(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("D")
response = ModelResponse(text="ANSWER: E")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.score, 0.0)
def test_mmlu_pro_supports_j_option(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("J")
response = ModelResponse(text="ANSWER: J")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_ceval_chinese_answer_prefix(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("C", task="ceval", choices="ABCD")
response = ModelResponse(text="逐步分析后可知,最后答案如下。\n答案:C")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "C")
def test_ceval_accepts_answer_letter_with_option_text(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("C", task="ceval", choices="ABCD")
response = ModelResponse(text="逐步分析后可知,最后答案如下。\n答案:C)正确选项")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "C")
def test_mmlu_pro_requires_explicit_final_answer_format(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("D", choices="ABCD")
response = ModelResponse(text="Options: A B C D")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertIsNone(result.judge_metadata["predicted_answer"])
def test_mmlu_pro_multiple_letters_without_final_marker_fails(self) -> None:
judge = MMLUProJudge()
sample = self._mmlu_sample("D", choices="ABCD")
response = ModelResponse(text="I considered A, B, C, and D. D seems possible.")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertIsNone(result.judge_metadata["predicted_answer"])
def test_highlevel_omni_requires_explicit_final_answer_format(self) -> None:
judge = HighLevelJudge()
sample = Sample(
sample_id="o2",
source_file="高阶育人-omni.jsonl",
source_path="高阶育人-omni.jsonl",
module="高阶育人",
subset="omni",
task="highlevel_omni",
prompt="dummy",
reference={"answer": "C", "reasonableness": 1},
)
response = ModelResponse(text="A is tempting, but C is ultimately better.")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertIsNone(result.judge_metadata["predicted_answer"])
def test_highlevel_omni_does_not_accept_earlier_correct_letter(self) -> None:
judge = HighLevelJudge()
sample = Sample(
sample_id="o3",
source_file="高阶育人-omni.jsonl",
source_path="高阶育人-omni.jsonl",
module="高阶育人",
subset="omni",
task="highlevel_omni",
prompt="dummy",
reference={"answer": "C", "reasonableness": 1},
)
response = ModelResponse(text="C is one possible answer, but after reconsidering I choose A.")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "fail")
self.assertIsNone(result.judge_metadata["predicted_answer"])
def test_ifeval_no_comma_passes_and_fails(self) -> None:
judge = IFEvalJudge()
sample = self._ifeval_sample(["punctuation:no_comma"], [{}])
pass_result = asyncio.run(judge.judge(sample, ModelResponse(text="No comma appears here")))
fail_result = asyncio.run(judge.judge(sample, ModelResponse(text="A comma, appears here")))
self.assertEqual(pass_result.judge_result, "pass")
self.assertEqual(pass_result.score, 1.0)
self.assertEqual(fail_result.judge_result, "fail")
self.assertEqual(fail_result.score, 0.0)
def test_ifeval_highlighted_sections_counts_single_and_double_markdown(self) -> None:
judge = IFEvalJudge()
sample = self._ifeval_sample(
["detectable_format:number_highlighted_sections"],
[{"num_highlights": 3}],
)
response = ModelResponse(text="*one* **two** ** ** *three*")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_ifeval_number_words_relations(self) -> None:
judge = IFEvalJudge()
at_least = self._ifeval_sample(["length_constraints:number_words"], [{"relation": "at least", "num_words": 3}])
less_than = self._ifeval_sample(["length_constraints:number_words"], [{"relation": "less than", "num_words": 3}])
self.assertEqual(asyncio.run(judge.judge(at_least, ModelResponse(text="one two three"))).score, 1.0)
self.assertEqual(asyncio.run(judge.judge(less_than, ModelResponse(text="one two"))).score, 1.0)
self.assertEqual(asyncio.run(judge.judge(less_than, ModelResponse(text="one two three"))).score, 0.0)
def test_ifeval_number_placeholders(self) -> None:
judge = IFEvalJudge()
sample = self._ifeval_sample(["detectable_content:number_placeholders"], [{"num_placeholders": 2}])
result = asyncio.run(judge.judge(sample, ModelResponse(text="Hello [name] at [address].")))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_ifeval_multi_instruction_averages_instruction_level(self) -> None:
judge = IFEvalJudge()
sample = self._ifeval_sample(
[
"punctuation:no_comma",
"detectable_content:number_placeholders",
"length_constraints:number_words",
],
[{}, {"num_placeholders": 1}, {"relation": "at least", "num_words": 100}],
)
result = asyncio.run(judge.judge(sample, ModelResponse(text="No comma here [slot]")))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.judge_metadata["prompt_level_strict"], 0.0)
self.assertAlmostEqual(result.judge_metadata["inst_level_strict"], 2 / 3)
def test_ifeval_loose_mode_tries_trimmed_lines(self) -> None:
judge = IFEvalJudge()
sample = self._ifeval_sample(["punctuation:no_comma"], [{}])
result = asyncio.run(judge.judge(sample, ModelResponse(text="bad, first line\nclean final line")))
self.assertEqual(result.judge_metadata["prompt_level_strict"], 0.0)
self.assertEqual(result.judge_metadata["prompt_level_loose"], 1.0)
def test_ifeval_dataset_instructions_are_all_registered(self) -> None:
config = load_project_config(Path("configs"))
registry = FileRegistry(config)
resolved = registry.resolve(source_files={"ifeval_sampled.jsonl"})
self.assertEqual(len(resolved), 1)
instruction_ids: set[str] = set()
with resolved[0].path.open("r", encoding="utf-8") as handle:
for line in handle:
if not line.strip():
continue
record = json.loads(line)
metadata = record.get("metadata") or {}
raw_instruction_ids = metadata.get("instruction_id_list") or []
instruction_ids.update(str(item) for item in raw_instruction_ids)
self.assertTrue(instruction_ids)
self.assertTrue(instruction_ids.issubset(set(INSTRUCTION_DICT)))
def test_math_500_extracts_boxed_answer(self) -> None:
judge = Math500Judge()
sample = self._math_sample("42")
result = asyncio.run(judge.judge(sample, ModelResponse(text="Reasoning here. Therefore, \\boxed{42}.")))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["predicted_answer"], "42")
def test_math_500_extracts_answer_marker_and_last_number(self) -> None:
judge = Math500Judge()
marker_result = asyncio.run(judge.judge(self._math_sample("9"), ModelResponse(text="The answer is 9.")))
fallback_result = asyncio.run(judge.judge(self._math_sample("-50"), ModelResponse(text="After simplification we get -50.")))
self.assertEqual(marker_result.score, 1.0)
self.assertEqual(fallback_result.score, 1.0)
def test_math_500_normalizes_latex_fraction_and_text(self) -> None:
judge = Math500Judge()
fraction_result = asyncio.run(judge.judge(self._math_sample("\\frac{14}{3}"), ModelResponse(text="\\boxed{14/3}")))
text_result = asyncio.run(judge.judge(self._math_sample("\\text{Evelyn}"), ModelResponse(text="final answer is Evelyn")))
self.assertEqual(fraction_result.score, 1.0)
self.assertEqual(text_result.score, 1.0)
def test_math_500_tuple_latex_matches_reference(self) -> None:
judge = Math500Judge()
sample = self._math_sample("\\left( 3, \\frac{\\pi}{2} \\right)")
response = ModelResponse(text="The polar form is \\boxed{\\left(3,\\frac{\\pi}{2}\\right)}.")
result = asyncio.run(judge.judge(sample, response))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_math_500_wrong_answer_fails(self) -> None:
judge = Math500Judge()
result = asyncio.run(judge.judge(self._math_sample("42"), ModelResponse(text="\\boxed{43}")))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.score, 0.0)
def test_aime_extracts_boxed_reference_and_prediction(self) -> None:
judge = AIMEJudge()
result = asyncio.run(judge.judge(self._aime_sample("\\boxed{204}"), ModelResponse(text="After solving, \\boxed{204}")))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
self.assertEqual(result.judge_metadata["expected_answer"], "204")
self.assertEqual(result.judge_metadata["predicted_answer"], "204")
def test_aime_uses_last_number_fallback_for_prediction(self) -> None:
judge = AIMEJudge()
result = asyncio.run(judge.judge(self._aime_sample("\\boxed{113}"), ModelResponse(text="The requested value is 113.")))
self.assertEqual(result.judge_result, "pass")
self.assertEqual(result.score, 1.0)
def test_aime_wrong_answer_fails(self) -> None:
judge = AIMEJudge()
result = asyncio.run(judge.judge(self._aime_sample("\\boxed{371}"), ModelResponse(text="\\boxed{370}")))
self.assertEqual(result.judge_result, "fail")
self.assertEqual(result.score, 0.0)
def test_aime25_and_aime26_use_same_rule_judge(self) -> None:
config = load_project_config(Path("configs"))
router = JudgeRouter(config)
aime25 = self._aime_sample("70", task="aime25", source_file="aime25.jsonl")
aime26 = self._aime_sample("277", task="aime26", source_file="aime26.jsonl")
self.assertIsInstance(router.get_judge(aime25), AIMEJudge)
self.assertIsInstance(router.get_judge(aime26), AIMEJudge)
def _mmlu_sample(self, target: str, task: str = "mmlu_pro", choices: str = "ABCDEFGHIJ") -> Sample:
return Sample(
sample_id="mmlu1",
source_file=f"{task}_sampled.jsonl",
source_path=f"{task}_sampled.jsonl",
module="通用模型",
subset=task,
task=task,
dimension="law",
prompt="dummy",
reference={"target": target},
metadata={"choices": [f"choice {letter}" for letter in choices]},
)
def _ifeval_sample(self, instruction_ids: list[str], kwargs: list[dict]) -> Sample:
return Sample(
sample_id="ifeval1",
source_file="ifeval_sampled.jsonl",
source_path="ifeval_sampled.jsonl",
module="通用模型",
subset="ifeval",
task="ifeval",
dimension="default",
prompt="dummy",
reference={"target": ""},
metadata={
"instruction_id_list": instruction_ids,
"kwargs": kwargs,
"key": 1,
},
)
def _math_sample(self, target: str) -> Sample:
return Sample(
sample_id="math1",
source_file="math_500_sampled.jsonl",
source_path="math_500_sampled.jsonl",
module="通用模型",
subset="math_500",
task="math_500",
dimension="Level 2",
prompt="dummy",
reference={"target": target},
metadata={"question_id": "test/example.json", "subset_key": "Level 2"},
)
def _aime_sample(self, target: str, task: str = "aime24", source_file: str = "aime24_sampled.jsonl") -> Sample:
return Sample(
sample_id="aime1",
source_file=source_file,
source_path=source_file,
module="通用模型",
subset=task,
task=task,
dimension="default",
prompt="dummy",
reference={"target": target},
metadata={},
)
if __name__ == "__main__":
unittest.main()