Function test_math_benchmark

tests/test_benchmark.py:42–159

Test the benchmark manager specifically for Math/AIME using a REAL model. Uses response_format for structured output.

(benchmark_name: str = "aime25")

    return re.sub(r'[\\/*?:"<>|]', '', name).strip()

async def test_math_benchmark(benchmark_name: str = "aime25"):
    """
    Test the benchmark manager specifically for Math/AIME using a REAL model.
    Uses response_format for structured output.
    """
    print(f"🧪 Testing benchmark manager with benchmark: {benchmark_name}")
    print(f"🤖 Using Model: {TARGET_MODEL}")
    
    # Define save directory
    save_dir = os.path.join(config.workdir, "benchmark", benchmark_name)
    if not os.path.exists(save_dir):
        os.makedirs(save_dir, exist_ok=True)
        print(f"📁 Created output directory: {save_dir}")
    
    # 1. Reset and get first task
    print(f"🔄 Resetting progress for {benchmark_name}...")
    task = await benchmark_manager.reset(benchmark_name)
    
    if not task:
        logger.warning("⚠️ No tasks available to run (Dataset empty or all finished).")
        return

    # ==========================================
    # Loop Logic
    # ==========================================
    while task is not None:
        task_id = task.task_id
        start_time = time.time()
        
        try:
            print(f"\n" + "="*50)
            print(f"🚀 Processing Task ID: {task_id}")
            print("="*50)

            # --- 1. Prepare Prompt ---
            question_text = task.input
            
            # Get system_prompt directly from task
            system_prompt_text = task.system_prompt
            
            logger.info(f"| 📋 [Task {task_id}] Input length: {len(question_text)}")

            messages = [
                SystemMessage(content=system_prompt_text),
                HumanMessage(content=question_text)
            ]

            # --- 2. Model Inference (Structured Output) ---
            print(f"⏳ [Task {task_id}] Model inferencing (Structured)...")
            
            try:
                # Call model_manager and pass response_format
                response = await model_manager(
                    model=TARGET_MODEL,
                    messages=messages,
                    response_format=Response,
                )