o Ti;@sPddlZddlZddlZddlZddlZddlmZddlmZm Z ddl m Z ddl Z ddl m Z ddlmZmZmZerIddlZddlmZeradd lmZdd lmZeradd lmZeeZGd d d ZedkrddlmZeZ e j!dde j"ddgddedZ#e$de#ddl%m&Z&e&'d(dZ)e *e)dSdS)N)BytesIO)OptionalUnion)urlparse)nn)is_requests_availableis_vllm_ascend_availableis_vllm_available)ConnectionError)PyNcclCommunicator)StatelessProcessGroup)PyHcclCommunicatorc@s eZdZdZ     d4deeded ed ed ef d d Zd5dedefddZ          d6de edee dededededededed eed!ee d"e e efd#d$Z d7d&e ejeeffd'd(Zd)ed*ejfd+d,Zd-ejfd.d/Zd0d1Zd2d3ZdS)8 VLLMClienta A client class to interact with a vLLM server. This class provides methods to generate completions, initialize and manage weight update groups, and update model weights in a distributed setting. Before using it, start the vLLM server with `trl vllm-serve`. Args: base_url (`str` or `None`, *optional*, defaults to `None`): Base URL for the vLLM server (e.g., `"http://localhost:8000"`). If provided, `host` and `server_port` are ignored. host (`str`, *optional*, defaults to `"0.0.0.0"`): IP address of the vLLM server. Ignored if `base_url` is provided. server_port (`int`, *optional*, defaults to `8000`): Port number of the vLLM server. Ignored if `base_url` is provided. group_port (`int`, *optional*, defaults to `51216`): Port number for the weight update group. connection_timeout (`float`, *optional*, defaults to `0.0`): Total timeout duration in seconds to wait for the server to be up. If the server is not up after the timeout, a `ConnectionError` is raised. Examples: Run the vLLM server with the model `Qwen/Qwen2.5-7B`: ``` $ trl vllm-serve --model Qwen/Qwen2.5-7B ... INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) ``` Use the client to generate completions and update model weights: ```python >>> from trl.extras.vllm_client import VLLMClient >>> client = VLLMClient() >>> client.generate(["Hello, AI!", "Tell me a joke"]) [[2980, 498, 1492, 752, 448, 264, 13027, 8645, 30, 358, 2776, 4460, 311, 3270, 264, 2025], [911, 7988, 1251, 382, 3838, 653, 498, 1618, 4325, 879, 2581, 20027, 264, 21428, 30, 362]] >>> from transformers import AutoModelForCausalLM >>> model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2.5-7B", device_map="cuda") >>> client.init_communicator(device="cuda") >>> client.update_model_params(model) ``` There are several ways to initialize the client: ```python VLLMClient(base_url="http://localhost:8000") VLLMClient(base_url="http://192.168.1.100:8000") VLLMClient(host="localhost", server_port=8000) VLLMClient(host="192.168.1.100", server_port=8000) ``` N0.0.0.0@base_urlhost server_port group_portconnection_timeoutcCststdtstdt|_|dur4t|}t|j |_ |j p&d}|d|j |j |_n||_ ||_d|j d|j|_||_||dS)NzIrequests is not installed. Please install it with `pip install requests`.zAvLLM is not installed. Please install it with `pip install vllm`.httpz://zhttp://:)r ImportErrorr requestsSessionsessionrsocket gethostbynamehostnamerschemenetlocpathrrr check_server)selfrrrrr parsed_urlr"r(J/home/ubuntu/.local/lib/python3.10/site-packages/trl/extras/vllm_client.py__init__hs  zVLLMClient.__init__@ total_timeoutretry_intervalc Cs|jd}t} zt|}Wn+tjjy=}zt|}||kr3td|jd|d|WYd}~nd}~ww|jdkrUd|jvrN|jd|_ t d dSt d |d t |q ) a Check server availability with retries on failure, within a total timeout duration. If the server is not up after the total timeout duration, raise a `ConnectionError`. Args: retry_interval (`float`, *optional*, defaults to `2.0`): Interval in seconds between retries. total_timeout (`float`, *optional*, defaults to `0.0`): Total timeout duration in seconds. z/health/Tz$The vLLM server can't be reached at z after zF seconds. Make sure the server is running by running `trl vllm-serve`.NzX-Forwarded-Forz Server is up!z"Server is not up yet. Retrying in z seconds...) rtimerget exceptionsRequestExceptionr status_codeheadersrloggerinfosleep)r&r,r-url start_timeresponseexc elapsed_timer(r(r)r%s0       zVLLMClient.check_server?promptsimagesnrepetition_penalty temperaturetop_ptop_kmin_p max_tokensguided_decoding_regexgeneration_kwargsreturnc s|jd} dd|rfdd|Dnd}|jj| ||||||||| | | p(id d} | jd kr8| d Std | jd | j) a Generates model completions for the provided prompts. Args: prompts (`list[str]`): List of text prompts for which the model will generate completions. images (`list[PIL.Image]` or `None`, *optional*, defaults to `None`): List of PIL Images to send along with the prompts. n (`int`, *optional*, defaults to `1`): Number of completions to generate for each prompt. repetition_penalty (`float`, *optional*, defaults to `1.0`): Parameter for repetition penalty. 1.0 means no penalty. temperature (`float`, *optional*, defaults to `1.0`): Temperature parameter for sampling. Higher values increase diversity. top_p (`float`, *optional*, defaults to `1.0`): Top-p sampling parameter.`1.0` means no truncation. top_k (`int`, *optional*, defaults to `-1`): Top-k sampling parameter. `-1` means no truncation. min_p (`float`, *optional*, defaults to `0.0`): Minimum probability for sampling. max_tokens (`int`, *optional*, defaults to `16`): Maximum number of tokens to generate for each prompt. guided_decoding_regex (`str` or `None`, *optional*, defaults to `None`): Regular expression to guide the decoding process. generation_kwargs (`dict` or `None`, *optional*, defaults to `None`): Additional generation parameters to pass to the vLLM `SamplingParams`. This can include parameters like `seed`, `frequency_penalty`, etc. If it contains keys that conflict with the other parameters, they will override them. Returns: `list[list[int]]`: List of lists of token IDs representing the model-generated completions for each prompt. z /generate/cSs,t}|j|dd|}t|dS)NPNG)formatzutf-8)rsavegetvaluebase64 b64encodedecode)imagebuffer img_bytesr(r(r) pil_to_base64sz*VLLMClient.generate..pil_to_base64csg|]}|qSr(r().0imgrWr(r) sz'VLLMClient.generate..N) rArBrCrDrErFrGrHrIrJrKjsonr.completion_idsRequest failed: , )rrpostr3r] Exceptiontext)r&rArBrCrDrErFrGrHrIrJrKr8r:r(rZr)generates* /  zVLLMClient.generaterdevicecCs|jd}t|}|jdkr|d}n td|jd|j|d}||_|jd}tt j |j }|j j|d|j||d d }|jdkrXtd|jd|jtd tj|j|j|j|d }t||d |_t|jdS)a Initializes the weight update group in a distributed setup for model synchronization. Args: device (`torch.device`, `str`, or `int`, *optional*, defaults to `0`): Device of trainer main process. It's the device that will be used for the weights synchronization. Can be a `torch.device` object, a string like `'cuda:0'`, or an integer device index. z/get_world_size/r. world_sizer_r`r=z/init_communicator/r)rportrfclient_device_uuidr\g?)rrgrankrfreN)rrr0r3r]rbrcristrtorchcudaget_device_propertiesuuidrrarr/r7r createrr pynccl_commatexitregisterclose_communicator)r&rer8r:vllm_world_sizerfrhpgr(r(r)init_communicators.     zVLLMClient.init_communicatornameweightscCs~t|jt|j}}|jd}|jj||||dd}|jdkr.td|jd|j |j j ||j d|j j dS) a0 Updates a specific named parameter in the model and broadcasts it to other processes. Args: name (`str`): Name of the layer whose weights are being updated. weights (`torch.Tensor`): Tensor containing the updated weights. z/update_named_param/)rxdtypeshaper\r.r_r`)srcN)rkrztupler{rrrar3rbrcrq broadcastrigroupbarrier)r&rxryrzr{r8r:r(r(r)update_named_param(s  zVLLMClient.update_named_parammodelcCs$|D] \}}|||jqdS)z Updates all parameters of the given model by calling `update_named_param` for each parameter in the model. Args: model (`nn.Module`): Model whose parameters (weights/biases) are to be updated. N)named_parametersrdata)r&rrxparamr(r(r)update_model_params<szVLLMClient.update_model_paramscCs>|jd}|j|}|jdkrtd|jd|jdS)z8 Resets the prefix cache for the model. z/reset_prefix_cache/r.r_r`N)rrrar3rbrcr&r8r:r(r(r)reset_prefix_cacheHs   zVLLMClient.reset_prefix_cachecCsX|jd}z|j|}Wn tyYdSw|jdkr*td|jd|jdS)zW Closes the weight update group and cleans up the communication group. z/close_communicator/r.r_r`N)rrrar r3rbrcrr(r(r)rtQs   zVLLMClient.close_communicator)Nrrrr)rr+) Nr=r>r>r>r?rr@NN)r)__name__ __module__ __qualname____doc__rrkintfloatr*r%listdictrdrrlrerwTensorrrModulerrrtr(r(r(r)r.sv; '     O1 r__main__)SamplingParamsrmrjz Hello, AI!zTell me a joke )rCrIsampling_paramsz Responses:)AutoModelForCausalLMzQwen/Qwen2.5-7B)+rrrQloggingrr/iortypingrr urllib.parserrlr import_utilsrr r rr ,vllm.distributed.device_communicators.pyncclr vllm.distributed.utilsr 3vllm_ascend.distributed.device_communicators.pyhcclr getLoggerrr5rvllmrclientrwrd responsesprint transformersrfrom_pretrainedtorrr(r(r(r)sB        6