Function run_evaluation

evaluation/utils.py:130–267

(
    dataset: pd.DataFrame,
    metadata: EvalMetadata | None,
    output_file: str,
    num_workers: int,
    process_instance_func: Callable[
        [pd.Series, EvalMetadata, bool], Awaitable[EvalOutput]
    ],
    max_retries: int = 3,  # number of retries for each instance
)

    #     queue.close()
    
def run_evaluation(
    dataset: pd.DataFrame,
    metadata: EvalMetadata | None,
    output_file: str,
    num_workers: int,
    process_instance_func: Callable[
        [pd.Series, EvalMetadata, bool], Awaitable[EvalOutput]
    ],
    max_retries: int = 3,  # number of retries for each instance
):
    logger = LoggerManager.get_logger()
    use_multiprocessing = num_workers > 1

    if metadata is not None:
        logger.info(
            f'Evaluation started with Agent {metadata.agent_func}\n'
        )
    else:
        logger.info('Running evaluation without metadata.', title='Warning', color='red')
        logger.info(f'Evaluation started with {num_workers} workers.')

    total_instances = len(dataset)
    pbar = tqdm(total=total_instances, desc='Instances processed')
    output_fp = open(output_file, 'a')

    try:
        if use_multiprocessing:
            # 使用队列来收集结果
            results_queue = mp.Queue()
            active_processes = []
            instances_iter = dataset.iterrows()
            instances_completed = 0
            
            while instances_completed < total_instances:
                # 启动新进程，直到达到worker数量限制
                while len(active_processes) < num_workers and instances_completed < total_instances:
                    try:
                        _, instance = next(instances_iter)
                        # 创建非守护进程
                        p = mp.Process(
                            target=_process_and_queue,
                            args=(process_instance_func, instance, metadata, True, max_retries, results_queue),
                            daemon=False  # 关键：设置为非守护进程
                        )
                        p.start()
                        time.sleep(3)
                        active_processes.append((p, time.time()))  # 记录进程启动时间
                    except StopIteration:
                        break

                # 检查完成的进程
                for p, start_time in active_processes[:]:
                    if not p.is_alive():
                        try:
                            # 给进程1分钟时间来清理资源
                            p.join(timeout=60)
                            if p.is_alive():
                                logger.warning(f"Process {p.pid} cleanup timeout, force terminating...")