Function get_all_incorrect_tasks

examples/analysis.py:263–309

Get all incorrect tasks by combining experiment results with benchmark data. Args: experiment_file: Path to the experiment results JSON file benchmark_name: Name of the benchmark to load tasks from Returns: List of task_data objects for incorrect samples

(experiment_file: str, benchmark_name: str)

    return tasks

async def get_all_incorrect_tasks(experiment_file: str, benchmark_name: str) -> List[Any]:
    """
    Get all incorrect tasks by combining experiment results with benchmark data.

    Args:
        experiment_file: Path to the experiment results JSON file
        benchmark_name: Name of the benchmark to load tasks from

    Returns:
        List of task_data objects for incorrect samples
    """
    # Load experiment results to find incorrect task_ids
    print(f"Loading experiment results from: {experiment_file}")
    with open(experiment_file, 'r', encoding='utf-8') as f:
        experiment_data = json.load(f)

    all_results = experiment_data.get('results', [])
    incorrect_samples = [sample for sample in all_results if not sample.get('correct', True)]

    # Extract incorrect task_ids
    incorrect_task_ids = {sample['task_id'] for sample in incorrect_samples}
    print(f"Found {len(incorrect_task_ids)} incorrect task_ids")

    all_tasks = await get_all_tasks(benchmark_name)

    # Create task_id to task mapping
    task_map = {}
    for task in all_tasks:
        task_id = getattr(task, 'task_id', None)
        if task_id:
            task_map[task_id] = task

    # Find incorrect task_data objects
    incorrect_tasks = []
    found_count = 0

    for task_id in incorrect_task_ids:
        if task_id in task_map:
            incorrect_tasks.append(task_map[task_id])
            found_count += 1
            print(f"Found incorrect task: {task_id}")
        else:
            print(f"Warning: Could not find task_data for task_id {task_id}")

    print(f"Successfully retrieved {found_count} incorrect task_data objects out of {len(incorrect_task_ids)}")

    return incorrect_tasks

async def process_single_task(optimizer_type: str, benchmark_name: str, task_data: Any, task_index: int, total_tasks: int, result_saver: ExperimentResultSaver = None):
    """Process a single task with the optimizer."""