refactor: optimize OpenAI compatible endpoint batch processing

- Remove unused decord imports from chat implementation
- Add configurable max_concurrent_requests parameter (default: 32)
- Optimize file I/O by writing cache once at the end instead of per batch
- Improve thread pool sizing based on actual batch size

These changes improve performance and reduce unnecessary file I/O operations
while maintaining backward compatibility.

Reported-by:b8zhong
Github-Issue:#835

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Luodian

lmms_eval/models/chat/openai_compatible.py

@@ -7,10 +7,7 @@
 
 from lmms_eval.api.registry import register_model
 
-try:
-    from decord import VideoReader, cpu
-except ImportError:
-    pass
+# decord imports removed - not used in chat implementation
 
 from dotenv import load_dotenv
 from loguru import logger as eval_logger
@@ -32,8 +29,14 @@ def generate_until(self, requests) -> List[str]:
         res = []
 
         batch_size = getattr(self, "batch_size_per_gpu", 1)
-        batched_requests = [requests[i : i + batch_size] for i in range(0, len(requests), batch_size)]
-        pbar = tqdm(total=len(batched_requests), disable=(self.rank != 0), desc="Model Responding")
+        batched_requests = [
+            requests[i : i + batch_size] for i in range(0, len(requests), batch_size)
+        ]
+        pbar = tqdm(
+            total=len(batched_requests),
+            disable=(self.rank != 0),
+            desc="Model Responding",
+        )
 
         e2e_latency = 0
         total_tokens = 0
@@ -56,7 +59,9 @@ def generate_until(self, requests) -> List[str]:
                             continue
 
                 chat_messages_raw = doc_to_messages(self.task_dict[task][split][doc_id])
-                chat_messages: ChatMessages = ChatMessages(**{"messages": chat_messages_raw})
+                chat_messages: ChatMessages = ChatMessages(
+                    **{"messages": chat_messages_raw}
+                )
 
                 payload = {"messages": chat_messages.to_openai_messages()}
                 payload["model"] = self.model_version
@@ -75,7 +80,11 @@ def generate_until(self, requests) -> List[str]:
                 payload["max_tokens"] = gen_kwargs["max_new_tokens"]
                 payload["temperature"] = gen_kwargs["temperature"]
 
-                if "o1" in self.model_version or "o3" in self.model_version or "o4" in self.model_version:
+                if (
+                    "o1" in self.model_version
+                    or "o3" in self.model_version
+                    or "o4" in self.model_version
+                ):
                     del payload["temperature"]
                     payload.pop("max_tokens")
                     payload["reasoning_effort"] = "medium"
@@ -108,22 +117,35 @@ def process_single_request(payload, i):
 
                     except Exception as e:
                         error_msg = str(e)
-                        eval_logger.info(f"Attempt {attempt + 1}/{self.max_retries} failed with error: {error_msg}")
+                        eval_logger.info(
+                            f"Attempt {attempt + 1}/{self.max_retries} failed with error: {error_msg}"
+                        )
 
                         if attempt == self.max_retries - 1:
-                            eval_logger.error(f"All {self.max_retries} attempts failed. Last error: {error_msg}")
+                            eval_logger.error(
+                                f"All {self.max_retries} attempts failed. Last error: {error_msg}"
+                            )
                             return "", i, 0, 0
                         else:
                             time.sleep(self.timeout)
 
                 return "", i, 0, 0
 
-            tasks_to_run = [(payload, i) for i, payload in enumerate(batch_payloads) if batch_responses[i] is None]
+            tasks_to_run = [
+                (payload, i)
+                for i, payload in enumerate(batch_payloads)
+                if batch_responses[i] is None
+            ]
 
             if tasks_to_run:
-                max_workers = min(len(tasks_to_run), 32)
+                max_workers = min(
+                    len(tasks_to_run), getattr(self, "max_concurrent_requests", 32)
+                )
                 with ThreadPoolExecutor(max_workers=max_workers) as executor:
-                    future_to_index = {executor.submit(process_single_request, payload, i): i for payload, i in tasks_to_run}
+                    future_to_index = {
+                        executor.submit(process_single_request, payload, i): i
+                        for payload, i in tasks_to_run
+                    }
 
                     for future in as_completed(future_to_index):
                         response_text, i, latency, tokens = future.result()
@@ -135,8 +157,6 @@ def process_single_request(payload, i):
                 for doc_uuid, response_text in zip(batch_doc_uuids, batch_responses):
                     if response_text is not None:
                         self.response_cache[doc_uuid] = response_text
-                with open(self.response_persistent_file, "w") as f:
-                    json.dump(self.response_cache, f)
 
             res.extend([r for r in batch_responses if r is not None])
             pbar.update(1)
@@ -152,4 +172,10 @@ def process_single_request(payload, i):
         log_metrics(**metric_dict)
 
         pbar.close()
+
+        # Write cache once at the end if in continual mode
+        if self.continual_mode is True:
+            with open(self.response_persistent_file, "w") as f:
+                json.dump(self.response_cache, f)
+
         return res

lmms_eval/models/simple/openai_compatible.py

@@ -44,6 +44,7 @@ def __init__(
         max_frames_num: int = 10,
         httpx_trust_env: bool = True,
         batch_size: int = 64,
+        max_concurrent_requests: int = 32,
         **kwargs,
     ) -> None:
         """
@@ -57,16 +58,22 @@ def __init__(
         self.model_version = model_version
         self.timeout = timeout
         self.max_retries = max_retries
-        self.max_size_in_mb = max_size_in_mb  # some models have a limit on the size of the image
+        self.max_size_in_mb = (
+            max_size_in_mb  # some models have a limit on the size of the image
+        )
         self.continual_mode = continual_mode
         self.max_frames_num = max_frames_num
         if self.continual_mode:
             if response_persistent_folder is None:
-                raise ValueError("Continual mode requires a persistent path for the response. Please provide a valid path.")
+                raise ValueError(
+                    "Continual mode requires a persistent path for the response. Please provide a valid path."
+                )
 
             os.makedirs(response_persistent_folder, exist_ok=True)
             self.response_persistent_folder = response_persistent_folder
-            self.response_persistent_file = os.path.join(self.response_persistent_folder, f"{self.model_version}_response.json")
+            self.response_persistent_file = os.path.join(
+                self.response_persistent_folder, f"{self.model_version}_response.json"
+            )
 
             if os.path.exists(self.response_persistent_file):
                 with open(self.response_persistent_file, "r") as f:
@@ -81,7 +88,11 @@ def __init__(
         # settings. openai-python uses a httpx.Client with trust_env set to True. Such a
         # httpx.Client uses macOS proxy server settings. Adding httpx_trust_env option
         # allows httpx to ignore proxy server settings set by VPN clients.
-        http_client = DefaultHttpxClient(trust_env=httpx_trust_env) if not httpx_trust_env else None
+        http_client = (
+            DefaultHttpxClient(trust_env=httpx_trust_env)
+            if not httpx_trust_env
+            else None
+        )
 
         # Use provided parameters or fall back to environment variables
         api_key = api_key or os.getenv("OPENAI_API_KEY")
@@ -98,16 +109,27 @@ def __init__(
         self.client = (
             OpenAI(api_key=api_key, base_url=base_url, http_client=http_client)
             if not azure_openai
-            else AzureOpenAI(api_key=os.getenv("AZURE_OPENAI_API_KEY"), azure_endpoint=os.getenv("AZURE_OPENAI_API_BASE"), api_version=os.getenv("AZURE_OPENAI_API_VERSION"), http_client=http_client)
+            else AzureOpenAI(
+                api_key=os.getenv("AZURE_OPENAI_API_KEY"),
+                azure_endpoint=os.getenv("AZURE_OPENAI_API_BASE"),
+                api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
+                http_client=http_client,
+            )
         )
 
         accelerator = Accelerator()
         # assert self.batch_size_per_gpu == 1, "Llava currently does not support batched generation. See https://github.com/haotian-liu/LLaVA/issues/754. HF Llava also has this issue."
         if accelerator.num_processes > 1:
-            assert accelerator.distributed_type in [DistributedType.FSDP, DistributedType.MULTI_GPU, DistributedType.DEEPSPEED], "Unsupported distributed type provided. Only DDP and FSDP are supported."
+            assert accelerator.distributed_type in [
+                DistributedType.FSDP,
+                DistributedType.MULTI_GPU,
+                DistributedType.DEEPSPEED,
+            ], "Unsupported distributed type provided. Only DDP and FSDP are supported."
             self.accelerator = accelerator
             if self.accelerator.is_local_main_process:
-                eval_logger.info(f"Using {accelerator.num_processes} devices with data parallelism")
+                eval_logger.info(
+                    f"Using {accelerator.num_processes} devices with data parallelism"
+                )
             self._rank = self.accelerator.local_process_index
             self._world_size = self.accelerator.num_processes
         else:
@@ -117,6 +139,7 @@ def __init__(
 
         self.device = self.accelerator.device
         self.batch_size_per_gpu = int(batch_size)
+        self.max_concurrent_requests = max_concurrent_requests
 
     @property
     def batch_size(self):
@@ -164,11 +187,15 @@ def encode_image(self, image: Union[Image.Image, str]):
     def encode_video(self, video_path, for_get_frames_num):
         vr = VideoReader(video_path, ctx=cpu(0))
         total_frame_num = len(vr)
-        uniform_sampled_frames = np.linspace(0, total_frame_num - 1, for_get_frames_num, dtype=int)
+        uniform_sampled_frames = np.linspace(
+            0, total_frame_num - 1, for_get_frames_num, dtype=int
+        )
 
         # Ensure the last frame is included
         if total_frame_num - 1 not in uniform_sampled_frames:
-            uniform_sampled_frames = np.append(uniform_sampled_frames, total_frame_num - 1)
+            uniform_sampled_frames = np.append(
+                uniform_sampled_frames, total_frame_num - 1
+            )
 
         frame_idx = uniform_sampled_frames.tolist()
         frames = vr.get_batch(frame_idx).asnumpy()
@@ -200,9 +227,15 @@ def _collate(x):
 
         from lmms_eval import utils
 
-        re_ords = utils.Collator([reg.args for reg in requests], _collate, grouping=True)
+        re_ords = utils.Collator(
+            [reg.args for reg in requests], _collate, grouping=True
+        )
         chunks = re_ords.get_batched(n=self.batch_size, batch_fn=None)
-        num_iters = len(requests) // self.batch_size if len(requests) % self.batch_size == 0 else len(requests) // self.batch_size + 1
+        num_iters = (
+            len(requests) // self.batch_size
+            if len(requests) % self.batch_size == 0
+            else len(requests) // self.batch_size + 1
+        )
         pbar = tqdm(total=num_iters, disable=(self.rank != 0), desc="Model Responding")
 
         for chunk in chunks:
@@ -234,10 +267,24 @@ def _collate(x):
                     visuals = self.flatten(visuals)
                     imgs = []
                     for visual in visuals:
-                        if isinstance(visual, str) and (".mp4" in visual or ".avi" in visual or ".mov" in visual or ".flv" in visual or ".wmv" in visual):
+                        if isinstance(visual, str) and (
+                            ".mp4" in visual
+                            or ".avi" in visual
+                            or ".mov" in visual
+                            or ".flv" in visual
+                            or ".wmv" in visual
+                        ):
                             frames = self.encode_video(visual, self.max_frames_num)
                             imgs.extend(frames)
-                        elif isinstance(visual, str) and (".jpg" in visual or ".jpeg" in visual or ".png" in visual or ".gif" in visual or ".bmp" in visual or ".tiff" in visual or ".webp" in visual):
+                        elif isinstance(visual, str) and (
+                            ".jpg" in visual
+                            or ".jpeg" in visual
+                            or ".png" in visual
+                            or ".gif" in visual
+                            or ".bmp" in visual
+                            or ".tiff" in visual
+                            or ".webp" in visual
+                        ):
                             img = self.encode_image(visual)
                             imgs.append(img)
                         elif isinstance(visual, Image.Image):
@@ -248,9 +295,16 @@ def _collate(x):
                 payload["model"] = self.model_version
 
                 payload["messages"].append({"role": "user", "content": []})
-                payload["messages"][0]["content"].append({"type": "text", "text": context})
+                payload["messages"][0]["content"].append(
+                    {"type": "text", "text": context}
+                )
                 for img in imgs:
-                    payload["messages"][0]["content"].append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img}"}})
+                    payload["messages"][0]["content"].append(
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": f"data:image/png;base64,{img}"},
+                        }
+                    )
 
                 if "max_new_tokens" not in gen_kwargs:
                     gen_kwargs["max_new_tokens"] = 1024
@@ -288,22 +342,33 @@ def process_single_request(payload, i):
 
                     except Exception as e:
                         error_msg = str(e)
-                        eval_logger.info(f"Attempt {attempt + 1}/{self.max_retries} failed with error: {error_msg}")
+                        eval_logger.info(
+                            f"Attempt {attempt + 1}/{self.max_retries} failed with error: {error_msg}"
+                        )
 
                         if attempt == self.max_retries - 1:
-                            eval_logger.error(f"All {self.max_retries} attempts failed. Last error: {error_msg}")
+                            eval_logger.error(
+                                f"All {self.max_retries} attempts failed. Last error: {error_msg}"
+                            )
                             return "", i
                         else:
                             time.sleep(self.timeout)
 
                 return "", i
 
-            tasks_to_run = [(payload, i) for i, payload in enumerate(batch_payloads) if batch_responses[i] is None]
+            tasks_to_run = [
+                (payload, i)
+                for i, payload in enumerate(batch_payloads)
+                if batch_responses[i] is None
+            ]
 
             if tasks_to_run:
-                max_workers = min(len(tasks_to_run), 32)
+                max_workers = min(len(tasks_to_run), self.max_concurrent_requests)
                 with ThreadPoolExecutor(max_workers=max_workers) as executor:
-                    future_to_index = {executor.submit(process_single_request, payload, i): i for payload, i in tasks_to_run}
+                    future_to_index = {
+                        executor.submit(process_single_request, payload, i): i
+                        for payload, i in tasks_to_run
+                    }
 
                     for future in as_completed(future_to_index):
                         response_text, i = future.result()
@@ -313,17 +378,25 @@ def process_single_request(payload, i):
                 for doc_uuid, response_text in zip(batch_doc_uuids, batch_responses):
                     if response_text is not None:
                         self.response_cache[doc_uuid] = response_text
-                with open(self.response_persistent_file, "w") as f:
-                    json.dump(self.response_cache, f)
 
             res.extend([r for r in batch_responses if r is not None])
             pbar.update(1)
 
         pbar.close()
+
+        # Write cache once at the end if in continual mode
+        if self.continual_mode is True:
+            with open(self.response_persistent_file, "w") as f:
+                json.dump(self.response_cache, f)
+
         return res
 
     def generate_until_multi_round(self, requests) -> List[str]:
-        raise NotImplementedError("TODO: Implement multi-round generation for OpenAI compatible models")
+        raise NotImplementedError(
+            "TODO: Implement multi-round generation for OpenAI compatible models"
+        )
 
     def loglikelihood(self, requests: List[Instance]) -> List[Tuple[float, bool]]:
-        raise NotImplementedError("TODO: Implement loglikelihood for OpenAI compatible models")
+        raise NotImplementedError(
+            "TODO: Implement loglikelihood for OpenAI compatible models"
+        )