Function test_math_benchmark

examples/run_benchmark.py:134–358

Test the benchmark manager specifically for Math/AIME using a REAL model. Uses response_format for structured output with concurrent processing. Args: benchmark_name: Name of the benchmark to test max_concurrency: Maximum number of concurrent tasks to process re

(benchmark_name: str = "aime25", max_concurrency: int = 5, result_saver: Optional[BenchmarkResultSaver] = None)

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

async def test_math_benchmark(benchmark_name: str = "aime25", max_concurrency: int = 5, result_saver: Optional[BenchmarkResultSaver] = None):
    """
    Test the benchmark manager specifically for Math/AIME using a REAL model.
    Uses response_format for structured output with concurrent processing.

    Args:
        benchmark_name: Name of the benchmark to test
        max_concurrency: Maximum number of concurrent tasks to process
        result_saver: Optional result saver for recording results
    """
    print(f"🧪 Testing benchmark manager with benchmark: {benchmark_name}")
    print(f"🤖 Using Model: {TARGET_MODEL}")
    print(f"⚡ Max Concurrency: {max_concurrency}")
    
    # 定义保存目录
    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. 重置并收集所有任务
    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

    # 收集所有任务
    print(f"📦 Collecting all tasks...")
    all_tasks = []
    while task is not None:
        all_tasks.append(task)
        task = await benchmark_manager.step(benchmark_name)
    
    total_tasks = len(all_tasks)
    print(f"✅ Collected {total_tasks} tasks. Starting concurrent processing...")

    # Update result saver with actual task count
    if result_saver:
        result_saver.update_total_tasks(total_tasks)

    # 创建 Semaphore 限制并发数
    semaphore = asyncio.Semaphore(max_concurrency)
    
    # 用于跟踪进度
    completed_count = 0
    completed_lock = asyncio.Lock()
    
    async def process_single_task(task: Task, result_saver: Optional[BenchmarkResultSaver] = None) -> Task:
        """处理单个任务的协程函数"""
        nonlocal completed_count  # 必须在函数开始处声明
        task_id = task.task_id
        start_time = time.time()
        
        async with semaphore:  # 使用 Semaphore 限制并发
            try:
                print(f"\n" + "="*50)