Function process_single_task

examples/analysis.py:311–394

Process a single task with the optimizer.

(optimizer_type: str, benchmark_name: str, task_data: Any, task_index: int, total_tasks: int, result_saver: ExperimentResultSaver = None)

    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."""
    task_id = task_data.task_id
    task_input = task_data.input
    task_gt = task_data.ground_truth
    system_instruction = task_data.system_prompt
    start_time = time.time()

    # Combine system instruction with task input
    full_task = f"{system_instruction}\n\n{task_input}"

    logger.info(f"\n📋 Task {task_index + 1}/{total_tasks}: {task_id}")
    print(f"\n📋 Task {task_index + 1}/{total_tasks}: {task_id}")
    logger.info(f"📋 Task: {full_task[:150]}..." if len(full_task) > 150 else f"📋 Task: {full_task}")

    try:
        # Get the agent instance
        agent = await acp.get("tool_calling")
        # Create optimizer
        optimizer = create_optimizer(optimizer_type, reward_fn)

        # ！！！！！用于临时代替参考模型输出
        if optimizer_type == 'reinforce_pp':
            logger.info(f"| 🚀 Running agent to get initial solution...")
            reference_agent_response = await agent(task=full_task, files=[])
            reference_agent_response_extra_data = reference_agent_response.extra.data if reference_agent_response.extra and reference_agent_response.extra.data else None
            reference_agent_result = reference_agent_response_extra_data['final_result']
            reference_agent_reasoning = reference_agent_response_extra_data['final_reasoning']
            reference_solution = f"Result: {reference_agent_result}\nReasoning: {reference_agent_reasoning}" if reference_agent_reasoning else f"Result: {reference_agent_result}"
            logger.info(f"| ✅ Initial solution obtained")

            initial_agent_result, initial_agent_reasoning, reflecion_text, improved_solution, agent_reasoning, agent_result = await optimizer.optimize(agent=agent,
                                                                             task=full_task,
                                                                             ground_truth=task_gt,
                                                                             sft_solution=reference_solution,
                                                                             benchmark_task_id=task_id,
                                                                             files=[])
        else:
            initial_agent_result, initial_agent_reasoning, reflecion_text, improved_solution, agent_reasoning, agent_result = await optimizer.optimize(agent=agent,
                                                                             task=full_task,
                                                                             ground_truth=task_gt,
                                                                             benchmark_task_id=task_id,
                                                                             files=[])

        parse_reasoning, parse_result = parse_agent_result(agent_result)

        if parse_reasoning == '':
            parse_reasoning = agent_reasoning
        task_data.reasoning = parse_reasoning
        task_data.result = parse_result

        _ = await benchmark_manager.eval(benchmark_name, task_data)
        # Get current stats after processing this task
        stats = await benchmark_manager.stats(benchmark_name)
        if stats:
            attempted = stats.correct + stats.wrong
            accuracy_msg = f"📊 Overall Progress: {attempted}/{stats.total} | Accuracy: {stats.accuracy:.2%}"
            print(accuracy_msg)