import logging import sys import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from elbench.config import load_project_config # noqa: E402 from elbench.basic_education_runtime.cli.eval import _metric_fields_from_evals # noqa: E402 from elbench.basic_education_runtime.entity import AgentConfig, AgentMemoryConfig, ExportFormat, Prompt # noqa: E402 from elbench.basic_education_runtime.router import any_keyword_route # noqa: E402 from elbench.basic_education_runtime.utils import content_to_text, remove_think # noqa: E402 from elbench.execution.basic_education import ( # noqa: E402 BasicEducationExecutor, aggregate_numeric_score, count_runtime_tasks, load_basic_education_config, ) class BasicEducationBridgeTest(unittest.TestCase): def test_count_iter_tasks(self) -> None: cfg = { "tasks": { "mode": "iter", "content": [{"q": 1}, {"q": 2}, {"q": 3}], } } self.assertEqual(count_runtime_tasks(cfg), 3) def test_count_union_tasks(self) -> None: cfg = { "tasks": { "mode": "union", "content": { "image": ["a", "b", "c"], "question": ["q1", "q2", "q3", "q4", "q5"], }, } } self.assertEqual(count_runtime_tasks(cfg), 15) def test_numeric_score_aggregation(self) -> None: score, details = aggregate_numeric_score( { "a": 4, "b": {"x": 2.0, "y": 3}, "c": [1, {"z": 5}], "d": "skip", } ) self.assertIsNotNone(score) self.assertAlmostEqual(score or 0, (4 + 2 + 3 + 1 + 5) / 5) self.assertEqual(details["a"], 4.0) self.assertEqual(details["b.x"], 2.0) self.assertEqual(details["b.y"], 3.0) self.assertEqual(details["c[0]"], 1.0) self.assertEqual(details["c[1].z"], 5.0) def test_basic_education_config_declares_45_tasks(self) -> None: config = load_basic_education_config( config_root=ROOT / "configs", project_root=ROOT, data_root=ROOT / "data" / "benchmark_root", ) self.assertTrue(config.enabled) expected = sum(int(s.expected_tasks or 0) for s in config.scenarios if s.enabled) self.assertEqual(expected, 45) base_dir = ROOT / "data" / "benchmark_root" / "基本教育" for scenario in config.scenarios: self.assertIn(str(base_dir), str(scenario.template_path)) self.assertTrue(scenario.template_path.exists()) def test_mock_runtime_payload_does_not_require_api_base(self) -> None: project_config = load_project_config(ROOT / "configs") executor = BasicEducationExecutor( project_config=project_config, model_config=project_config.models["mock.default"], logger=logging.getLogger("test-basic-education"), ) payload = executor._build_runtime_model_payload(project_config.models["mock.default"]) self.assertEqual(payload["type"], "mock") self.assertEqual(payload["model"], "mock-echo") self.assertIsNone(payload["api_base"]) self.assertEqual(payload["kargs"]["prefix"], "[MOCK]") def test_guided_scenario_splits_teacher_student_and_eval_models(self) -> None: project_config = load_project_config(ROOT / "configs") executor = BasicEducationExecutor( project_config=project_config, model_config=project_config.models["gpt-5.4"], logger=logging.getLogger("test-basic-education"), ) executor.config.test_endpoint_model_id = "qwen-test" scenario = next( item for item in executor.config.scenarios if item.scenario_id == "guided_problem_solving_teaching" ) template_data = executor._read_yaml_file(scenario.template_path) with patch("elbench.execution.basic_education.get_api_key", return_value="test-key"): rendered = executor._render_runtime_template( template_data=template_data, scenario=scenario, memory_path=ROOT / "tmp_test_artifacts" / "basic_education_guided_memory", max_samples=1, ) models_section = rendered["models"] self.assertEqual(models_section["teacher"]["model"], "gpt-5.4") self.assertEqual( models_section["teacher"]["api_base"], project_config.models["gpt-5.4"].api_base, ) self.assertEqual(models_section["stu"]["model"], project_config.models["qwen-test"].model_name) self.assertEqual( models_section["stu"]["api_base"], project_config.models["qwen-test"].api_base, ) self.assertEqual(models_section["eval"]["model"], project_config.models["qwen-test"].model_name) self.assertEqual(models_section["stu"]["kargs"]["max_tokens"], 512) self.assertEqual(models_section["eval"]["kargs"]["max_tokens"], 1024) self.assertEqual( models_section["eval"]["kargs"]["extra_body"], { "chat_template_kwargs": {"enable_thinking": False}, "served_model_name": "qwen3.6-35B-sft-v1", }, ) self.assertEqual(rendered["globals"]["recursion_limit"], 20) self.assertEqual(rendered["evaluation"]["model"], "eval") def test_runtime_template_uses_target_model_concurrency_cap(self) -> None: project_config = load_project_config(ROOT / "configs") executor = BasicEducationExecutor( project_config=project_config, model_config=project_config.models["innospark-235b"], logger=logging.getLogger("test-basic-education"), ) scenario = next( item for item in executor.config.scenarios if item.scenario_id == "contextualized_question_generation" ) template_data = executor._read_yaml_file(scenario.template_path) with patch("elbench.execution.basic_education.get_api_key", return_value="test-key"): rendered = executor._render_runtime_template( template_data=template_data, scenario=scenario, memory_path=ROOT / "tmp_test_artifacts" / "basic_education_question_memory", max_samples=1, ) self.assertEqual(rendered["globals"]["concurrency"], 1) def test_content_to_text_ignores_reasoning_only_blocks(self) -> None: content = [{"type": "reasoning", "summary": [], "text": "hidden"}] self.assertEqual(content_to_text(content), "") def test_content_to_text_extracts_text_blocks(self) -> None: content = [ {"type": "reasoning", "summary": [], "text": "hidden"}, {"type": "text", "text": "课堂结束 "}, ] self.assertEqual(content_to_text(remove_think(content)), "课堂结束 ") def test_keyword_route_handles_openai_content_blocks(self) -> None: route, _ = any_keyword_route([""], exists_to="END", else_to="student") state = { "messages": [ type("Msg", (), {"content": [{"type": "text", "text": "下课。"}]})(), ] } self.assertTrue(route(state)) def test_agent_skips_messages_that_become_empty_after_reasoning_cleanup(self) -> None: import asyncio from langchain_core.messages import AIMessage, HumanMessage from elbench.basic_education_runtime import config as runtime_config runtime_config.CONFIG = SimpleNamespace( globals=SimpleNamespace( retry=SimpleNamespace(attempt=1, interval=0), model_call_timeout_seconds=300, ) ) from elbench.basic_education_runtime.agent import _init_agent_from_dict class CapturingModel: def __init__(self) -> None: self.messages = [] async def ainvoke(self, messages): self.messages = messages return AIMessage(content="ok") model = CapturingModel() agent = _init_agent_from_dict( AgentConfig( model="teacher", prompt=[Prompt(role="system", content="Teach.")], memory=AgentMemoryConfig(enable=False, keep_turns=3), ), {"teacher": model}, "teacher", ) state = { "messages": [ AIMessage( content=[{"type": "reasoning", "summary": [], "text": "hidden"}], name="teacher", ), HumanMessage(content="Please continue.", name="student"), ] } asyncio.run(agent(state)) contents = [getattr(message, "content", "") for message in model.messages] self.assertEqual(contents, ["Teach.", "Please continue."]) class BasicEducationEvalCliHelpersTest(unittest.TestCase): def test_metric_fields_from_evals_ignores_empty_results(self) -> None: self.assertEqual(_metric_fields_from_evals([{}, {}]), []) self.assertEqual(_metric_fields_from_evals([{}, {"a": 1, "b": 2}]), ["a", "b"]) def test_evaluation_messages_add_user_prompt_when_template_only_has_system(self) -> None: from elbench.basic_education_runtime import config as runtime_config runtime_config.CONFIG = SimpleNamespace( globals=SimpleNamespace( retry=SimpleNamespace(attempt=1, interval=0), model_call_timeout_seconds=300, ) ) from elbench.basic_education_runtime.evaluation import _build_evaluation_messages exported = ExportFormat( task={"query": "Explain decimals"}, messages=[Prompt(role="teacher", content="Decimals are parts of one.")], ) messages = _build_evaluation_messages( system_prompt="Evaluate this output:\n{messages.as_dialog()}\nTask: {task.query}", other_prompts=[], exported_result=exported, ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0]["role"], "user") self.assertIn("teacher: Decimals are parts of one.", messages[0]["content"]) self.assertIn("Task: Explain decimals", messages[0]["content"]) def test_replace_template_preserves_non_placeholder_braces(self) -> None: exported = ExportFormat( task={"query": "Solve the set example"}, messages=[Prompt(role="teacher", content="The answer set is {1}.")], ) rendered = exported.replace_template( "Evaluate {task.query}\n{messages.as_dialog()}\nDo not treat {1} as a template." ) self.assertIn("Evaluate Solve the set example", rendered) self.assertIn("teacher: The answer set is {1}.", rendered) self.assertIn("Do not treat {1} as a template.", rendered) if __name__ == "__main__": unittest.main()