add collective_rpc to llm engine

Signed-off-by: Yinghai Lu <[email protected]>

Author: yinghai

vllm/engine/async_llm_engine.py

@@ -1238,6 +1238,12 @@ async def is_sleeping(self) -> bool:
     async def add_lora(self, lora_request: LoRARequest) -> None:
         self.engine.add_lora(lora_request)
 
+    async def collective_rpc(self, method: str, timeout: float | None = None, args: tuple = (), kwargs: dict | None = None):
+        """
+        Perform a collective RPC call to the given path.
+        """
+        return await self.engine_core.collective_rpc_async(method, timeout, args, kwargs)
+
 
 # TODO(v1): Remove this class proxy when V1 goes default.
 if envs.is_set("VLLM_USE_V1") and envs.VLLM_USE_V1:

vllm/v1/engine/async_llm.py

@@ -490,6 +490,12 @@ async def pin_lora(self, lora_id: int) -> bool:
         """Prevent an adapter from being evicted."""
         return await self.engine_core.pin_lora_async(lora_id)
 
+    async def collective_rpc(self, method: str, timeout: float | None = None, args: tuple = (), kwargs: dict | None = None):
+        """
+        Perform a collective RPC call to the given path.
+        """
+        return await self.engine_core.collective_rpc_async(method, timeout, args, kwargs)
+
     @property
     def is_running(self) -> bool:
         # Is None before the loop is started.