fix pre-commit

Signed-off-by: jianjunzhong <jianjunzhong@foxmail.com>

Author: jianjunzhong

verl/single_controller/ray/base.py

@@ -433,6 +433,7 @@ def _init_with_detached_workers(self, worker_names, worker_handles):
 
     def _get_master_addr_port(self, pg):
         """Get master addr and port for this worker group"""
+
         def _do_get_master_addr_port(pg):
             master_addr, master_port = ray.get(
                 get_master_addr_port.options(
@@ -442,6 +443,7 @@ def _do_get_master_addr_port(pg):
                 ).remote()
             )
             return master_addr, master_port
+
         if self._master_addr is None and self._master_port is None:
             self._master_addr, self._master_port = _do_get_master_addr_port(pg)
         elif self._master_addr is not None and self._master_port is not None:

verl/utils/vllm/utils.py

@@ -12,9 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import torch
 
-from typing import Callable
 from msgspec import field
 from packaging import version as vs
 from vllm.lora.models import LoRAModel

verl/workers/rollout/vllm_rollout/__init__.py

@@ -14,7 +14,7 @@
 import os
 from importlib.metadata import PackageNotFoundError, version
 
-from .vllm_rollout import vLLMAsyncRollout  # noqa: F401
+from .vllm_rollout import ServerAdapter  # noqa: F401
 
 
 def get_version(pkg):

verl/workers/rollout/vllm_rollout/utils.py

@@ -67,10 +67,9 @@ def compute_logits(
 
     model.compute_logits = MethodType(compute_logits, model)
 
+
 # copy from https://github.com/vllm-project/vllm/blob/main/examples/offline_inference/rlhf_utils.py
-def rebuild_ipc(
-    handle: tuple[Callable, tuple], device_id: int | None = None
-) -> torch.Tensor:
+def rebuild_ipc(handle: tuple[Callable, tuple], device_id: int | None = None) -> torch.Tensor:
     func, args = handle
     list_args = list(args)
     if device_id is not None:
@@ -80,13 +79,15 @@ def rebuild_ipc(
     buffer = func(*list_args)
     return buffer
 
+
 class FlattenedTensorMetadata(TypedDict):
     name: str
     shape: torch.Size
     dtype: torch.dtype
     # specify the start offset of this tensor in shared ipc_buffer tensor
     offset: int
 
+
 class vLLMColocateWorkerExtension:
     """
     The class for vLLM's worker to inherit from, in the colocate setting.
@@ -96,6 +97,7 @@ class vLLMColocateWorkerExtension:
     NOTE: we define this class in a separate module, and the main module
     should pass the full qualified name as `worker_extension_cls` argument.
     """
+
     def __new__(cls, **kwargs):
         global_rank = kwargs.get("rank", 0) + int(os.environ.get("VERL_VLLM_MULTIPROC_GLOBAL_RANK_OFFSET", "0"))
         local_rank = kwargs.get("local_rank", 0)
@@ -115,7 +117,7 @@ def __new__(cls, **kwargs):
 
     def monkey_patch_compute_logits(self, vocab_size: int):
         _monkey_patch_compute_logits(self.model_runner.model, vocab_size)
-    
+
     def _fetch_weights(self, zmq_handle: str, load: bool = True):
         from vllm.model_executor.model_loader.utils import process_weights_after_loading
 
@@ -126,14 +128,10 @@ def _fetch_weights(self, zmq_handle: str, load: bool = True):
         socket.connect(zmq_handle)
         weights_to_load = []
         while True:
-            payload: tuple[Callable, tuple] | list[FlattenedTensorMetadata] | None = (
-                socket.recv_pyobj()
-            )
+            payload: tuple[Callable, tuple] | list[FlattenedTensorMetadata] | None = socket.recv_pyobj()
             if payload is None:
                 # means the update is done
-                process_weights_after_loading(
-                    self.model_runner.model, self.model_config, self.device
-                )
+                process_weights_after_loading(self.model_runner.model, self.model_config, self.device)
                 torch.cuda.synchronize()
                 socket.send(b"")
                 break
@@ -191,10 +189,10 @@ def update_lora_weights_from_ipc(self, peft_config: dict, zmq_handles: dict[str,
             lora_tensors=dict(lora_weights),
         )
         self.add_lora(lora_request)
+        logger.info(f"vLLM load weights, loaded_params: {len(lora_weights)}")
         del lora_weights
         gc.collect()
         torch.cuda.empty_cache()
-        logger.info(f"vLLM load weights, loaded_params: {len(lora_weights)}")
 
     def report_device_id(self) -> str:
         """Report device ID for ZMQ handle."""

verl/workers/rollout/vllm_rollout/vllm_async_server.py

@@ -13,20 +13,17 @@
 # limitations under the License.
 import argparse
 import asyncio
+import inspect
 import json
 import logging
 import os
-from concurrent.futures import Future
 from pprint import pprint
-from typing import Any, Callable, Optional
+from typing import Any, Optional
 
-import cloudpickle as pickle
 import numpy as np
 import ray
 import torch
 import vllm.entrypoints.cli.serve
-import zmq
-from filelock import FileLock
 from ray.actor import ActorHandle
 from vllm import SamplingParams
 from vllm.config import LoRAConfig
@@ -271,10 +268,9 @@ async def launch_server(self, master_address: str = None, master_port: int = Non
                 server_args.append(json.dumps(v) if isinstance(v, dict) else str(v))
 
         # pass worker_extension_cls parameter for cuda-ipc based weights updating
-        server_args.extend([
-            "--worker_extension_cls",
-            "verl.workers.rollout.vllm_rollout.utils.vLLMColocateWorkerExtension"
-        ])
+        server_args.extend(
+            ["--worker_extension_cls", "verl.workers.rollout.vllm_rollout.utils.vLLMColocateWorkerExtension"]
+        )
 
         if self.replica_rank == 0:
             pprint(server_args)
@@ -336,8 +332,7 @@ async def run_server(self, args: argparse.Namespace):
         # Don't keep the dummy data in memory
         await engine_client.reset_mm_cache()
         await engine_client.collective_rpc(
-            method="monkey_patch_compute_logits",
-            kwargs={"vocab_size": len(self.model_config.tokenizer)}
+            method="monkey_patch_compute_logits", kwargs={"vocab_size": len(self.model_config.tokenizer)}
         )
 
         app = build_app(args)
@@ -376,18 +371,12 @@ async def run_headless(self, args: argparse.Namespace):
             executor_class=Executor.get_class(vllm_config),
             log_stats=not engine_args.disable_log_stats,
         )
-    
+
     async def collective_rpc(
-        self,
-        method: str,
-        timeout: Optional[float] = None,
-        args: tuple = (),
-        kwargs: Optional[dict] = None
+        self, method: str, timeout: Optional[float] = None, args: tuple = (), kwargs: Optional[dict] = None
     ):
         """Perform a collective RPC call to the inference engine."""
-        return await self.engine.collective_rpc(
-            method=method, timeout=timeout, args=args, kwargs=kwargs
-        )
+        return await self.engine.collective_rpc(method=method, timeout=timeout, args=args, kwargs=kwargs)
 
     async def generate(
         self,
@@ -582,7 +571,17 @@ def __init__(
         nnodes: int,
         cuda_visible_devices: str,
     ):
-        super().__init__(config, model_config, rollout_mode, workers, replica_rank, node_rank, gpus_per_node, nnodes, cuda_visible_devices)
+        super().__init__(
+            config,
+            model_config,
+            rollout_mode,
+            workers,
+            replica_rank,
+            node_rank,
+            gpus_per_node,
+            nnodes,
+            cuda_visible_devices,
+        )
 
 
 _rollout_worker_actor_cls = ray.remote(ServerAdapter)

verl/workers/rollout/vllm_rollout/vllm_rollout.py

@@ -26,49 +26,22 @@
 - After inference, all the parameters that doesn't belong to this pp rank is freed.
 """
 
-import asyncio
-import getpass
 import logging
 import os
-from dataclasses import asdict
-from types import MethodType
 from typing import Any, Generator, Optional
 
-import cloudpickle as pickle
 import ray
 import torch
-import torch.distributed
 import zmq
-import zmq.asyncio
-from filelock import FileLock
+from packaging import version as vs
 from torch.distributed.device_mesh import DeviceMesh
 from torch.multiprocessing.reductions import reduce_tensor
-from vllm.config import LoRAConfig
-
-try:
-    from vllm.worker.worker_base import WorkerWrapperBase
-except ModuleNotFoundError:
-    # https://github.com/vllm-project/vllm/commit/6a113d9aed8221a9c234535958e70e34ab6cac5b
-    from vllm.v1.worker.worker_base import WorkerWrapperBase
-
-from packaging import version as vs
 
 from verl import DataProto
 from verl.third_party.vllm import VLLM_SLEEP_LEVEL, get_version
-from verl.utils.device import is_npu_available
-from verl.utils.distributed import initialize_global_process_group_ray
-from verl.utils.ray_utils import ray_noset_visible_devices
-from verl.utils.vllm import TensorLoRARequest, VLLMHijack, is_version_ge
-from verl.utils.vllm.vllm_fp8_utils import apply_vllm_fp8_patches, is_fp8_model, load_quanted_weights
+from verl.utils.vllm import VLLMHijack, is_version_ge
 from verl.workers.config import HFModelConfig, RolloutConfig
 from verl.workers.rollout.base import BaseRollout
-from verl.workers.rollout.utils import get_free_port, is_valid_ipv6_address
-from verl.workers.rollout.vllm_rollout.utils import (
-    VLLM_LORA_INT_ID,
-    VLLM_LORA_NAME,
-    VLLM_LORA_PATH,
-    get_vllm_max_lora_rank,
-)
 
 logger = logging.getLogger(__file__)
 logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN"))
@@ -98,6 +71,7 @@ class ServerAdapter(BaseRollout):
     vLLM server adapter used in native async mode, serve as a client to request vLLM server
     to resume/release/update weights and kv_cache.
     """
+
     def __init__(
         self,
         config: RolloutConfig,
@@ -110,7 +84,7 @@ def __init__(
         rank = int(os.environ["RANK"])
         local_world_size = int(os.environ["RAY_LOCAL_WORLD_SIZE"])
         rollout_world_size = (
-            self.config.tensor_model_parallel_size 
+            self.config.tensor_model_parallel_size
             * self.config.data_parallel_size
             * self.config.pipeline_model_parallel_size
         )
@@ -122,7 +96,7 @@ def __init__(
             self.sleep_level = 1
         else:
             self.sleep_level = VLLM_SLEEP_LEVEL
-        
+
         # Attributes related to weight updates
         from vllm.platforms import current_platform
 
@@ -137,7 +111,7 @@ async def _execute_method(
         non_block: bool = False,
         timeout: Optional[float] = None,
         args: tuple = (),
-        kwargs: Optional[dict] = None
+        kwargs: Optional[dict] = None,
     ) -> Any:
         """Execute method on inference engine via ray.
 
@@ -184,15 +158,15 @@ async def update_weights(self, weights: Generator[tuple[str, torch.Tensor], None
                 kwargs={
                     "peft_config": peft_config,
                     "zmq_handles": self.zmq_handles,
-                }
+                },
             )
         else:
             await self._execute_method(
                 "update_weights_from_ipc",
                 non_block=True,
                 kwargs={
                     "zmq_handles": self.zmq_handles,
-                }
+                },
             )
         await self._update_weights_per_tensor(weights)
 
@@ -225,7 +199,7 @@ def set_server_handle(self, server_handle: ray.actor.ActorHandle):
         """Set vLLMHttpServer handle"""
         if self.rollout_rank == 0:
             self.server_handle = server_handle
-    
+
     def get_update_weights_zmq_handle(self) -> dict[str, str]:
         """Get ZMQ handle for weight updates."""
         suffix = f"{self.device_uuid}-{self.zmq_address_counter}"