Function evaluate_batch

eval/src/vllm_request.py:168–269

(
    batch_data,
    api_url,
    image_root=None,
    output_file="./results.json",
    error_file=None,
    max_workers=None,
    max_retries=10,
    timeout_base=160,
    backoff_base=2,
    backoff_cap=10,
    flush_every=20,
    log_stats=None,
)

def evaluate_batch(
    batch_data,
    api_url,
    image_root=None,
    output_file="./results.json",
    error_file=None,
    max_workers=None,
    max_retries=10,
    timeout_base=160,
    backoff_base=2,
    backoff_cap=10,
    flush_every=20,
    log_stats=None,
):
    success_count = 0
    total_result = []
    durations = []
    output_buffer = []
    error_buffer = []

    if max_workers is None:
        max_workers = int(os.getenv("VLLM_MAX_WORKERS", "16"))
    max_workers = max(1, min(max_workers, len(batch_data)))

    client = VLMessageClient(
        api_url,
        timeout_base=timeout_base,
        max_retries=max_retries,
        backoff_base=backoff_base,
        backoff_cap=backoff_cap,
        pool_maxsize=max_workers,
    )

    if log_stats is None:
        log_stats = os.getenv("VLLM_LOG_STATS", "0") == "1"

    def flush_buffer(buffer, path):
        if not buffer or not path:
            return
        with open(path, "a", encoding="utf-8") as f:
            for row in buffer:
                f.write(json.dumps(row, ensure_ascii=False) + "\n")
        buffer.clear()

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        index = 0
        for item in batch_data:
            if "idx" not in item:
                item["idx"] = str(index)
                index += 1
            futures.append(
                executor.submit(
                    client.process_item,
                    item=item,
                    image_root=image_root,
                )
            )