&`i+ddlZddlZddlZddlZddlmZddlmZmZm Z m Z ddl m Z ddl mZmZddlmZmZmZddlmZddlmZdd lmZmZejeZGd d ZdS) N) OrderedDict)AnyCallableListSet)metrics) ReplicaIDRequestRoutingInfo)MODEL_LOAD_LATENCY_BUCKETS_MS%PUSH_MULTIPLEXED_MODEL_IDS_INTERVAL_SSERVE_LOGGER_NAME) MetricsPusher) ServeUsageTag)_get_global_client_get_internal_replica_contextcxeZdZdZdZdeegefdedefdZ de efdZ d Z d Z d edefd ZddZd S)_ModelMultiplexWrapperaA wrapper class that wraps the model load function and provides the LRU caching functionality. The model multiplexer is a wrapper class that wraps the model load function and provides the LRU caching functionality, and the model load function should be a coroutine function that takes the model ID as the first argument and returns the user-constructed model object. The model multiplexer will also ensure that the number of models on the current replica does not exceed the specified limit. The model will be unloaded in the LRU order, the model multiplexer will call the model's __del__ attribute if it exists to clean up the model resources eagerly. push_multiplexed_model_idsmodel_load_funcself_argmax_num_models_per_replicactjdt|_||_||_||_t dttj ddt|_ tj ddt|_tjdd |_tjd d d |_tjdd |_tjdd |_tjdd |_t-}|t/d|j|_|j|_|j|_d|_t?j |_!tE|_#tI|_%|j%&|j'|j(tR|j%*dS)aInitialize the model multiplexer. Args: model_load_func: the model load async function. self_arg: self argument when model_load_func is class method. max_num_models_per_replica: the maximum number of models to be loaded on the current replica. If it is -1, there is no limit for the number of models per replica. 1zMODEL_LOAD_LATENCY_BUCKET_MS: 'serve_multiplexed_model_load_latency_msz"The time it takes to load a model.) description boundaries)serve_multiplexed_model_unload_latency_msz$The time it takes to unload a model.serve_num_multiplexed_modelsz3The number of models loaded on the current replica.)r%serve_registered_multiplexed_model_idz/The model id registered on the current replica.)model_id)rtag_keys,serve_multiplexed_get_model_requests_counterz:The counter for get model requests on the current replica.'serve_multiplexed_models_unload_counterz7The counter for unloaded models on the current replica.%serve_multiplexed_models_load_counterz5The counter for loaded models on the current replica.Nzc`@serve.multiplex` can only be used within a deployment (failed to retrieve Serve replica context).F)+rMULTIPLEXED_API_USEDrecordrmodels_funcrrloggerdebugr r Histogrammodel_load_latency_msmodel_unload_latency_msGaugenum_models_gaugeregistered_model_gaugeCounterget_model_requests_countermodels_unload_countermodels_load_counterr RuntimeErrorapp_name _app_name deployment_deployment_name replica_id _replica_id_push_multiplexed_replica_infoasyncioLock_model_cache_lockset_model_load_tasksrmetrics_pusherregister_or_update_task%_PUSH_MULTIPLEXED_MODEL_IDS_TASK_NAME_push_model_ids_infor start)selfrrrcontexts g/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/serve/multiplex.py__init__z_ModelMultiplexWrapper.__init__'s *11#666!mm . % /I'  U6SUUVVV%,%6 5<4& & & " (/'8 7>4( ( ( $ !( *M! ! !  '.m 3I"' ' ' # +2/ :T+ + + '&-_ 5Q& & & "$+? 3O$ $ $  011 ?>  &.%,%7&-&85:+") ,/55+oo 33  6  % 1   !!#####returnct|j}||jt |S)zGet the model IDs of the loaded models & loading models in the replica. This is to push the model id information early to the controller, so that requests can be routed to the replica. )r@r'keysupdaterAlist)rG models_lists rI!_get_loading_and_loaded_model_idsz8_ModelMultiplexWrapper._get_loading_and_loaded_model_ids~sD $+**,,-- 41222K   rKc |jt|j|jD] }|jdd|i!|jrPt t|j | d|_dSdS#t$r(}t d|Yd}~dSd}~wwxYw)z4Push the multiplexed replica info to the controller.r tags)r:multiplexed_model_idsFzFFailed to push the multiplexed replica info to the controller. Error: N)r/r@lenr'r0r<rrecord_request_routing_infor r;rR Exceptionr)warning)rGr es rIrEz+_ModelMultiplexWrapper._push_model_ids_infos0   ! % %c$+&6&6 7 7 7 K P P+//X8N/OOOO2 <"$$@@&#'#3.2.T.T.V.V 7<333 < <    NN1-.11          sB)B// C!9CC!cKt|jdkrj |d{Vn4#t$r'}td|Yd}~nd}~wwxYwt|jdkhdSdS)zt|j|j kr!|d{Vd|_td|d|jt%j}|j$||d{V|j|<n)||j|d{V|j|<t%j|z dz}td |d |d d |j ||j||j|cdddd{VS#t0$rA}td |d||j ||d}~wwxYw#1d{VswxYwYdS)zLoad the model if it is not loaded yet, and return the user-constructed model object. Args: model_id: the model ID. Returns: The user-constructed model object. zThe model ID must be a string.zThe model ID cannot be empty.TNrzLoading model ''.@@zSuccessfully loaded model '' in .1fms.zFailed to load model 'z '. Error: )typestr TypeError ValueErrorr2incr'popr<rAaddr?rrXr^r)infor4timerr(discardr,observerZerror)rGr modelload_start_timeload_latency_msr\s rI load_modelz!_ModelMultiplexWrapper.load_models >> $ $<== = ><== = '++--- t{ " "KOOH--E$)DK !;x( ( 37D /  " & &x 0 0 0-' ' ' ' ' ' ' ' t{**;x0' ' ' ' ' ' ' ' ' ' ' ' ' ' # 7!;; ,,0OOO#33555555555>B;KK >( > > >???,00222&*ikkO},6:jj6J6J0J0J0J0J0J0J H--6:jj M877111111 H-(,y{{_'D&NOKK4h44*3444*228<<<.66GGG;x0C' ' ' ' ' ' ' ' ' ' ' ' ' ' D!LLHHHQHH*228<<<G E' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' s1:J'$E"I J$#z9_ModelMultiplexWrapper.unload_model_lru..sdrKrczSuccessfully unloaded model 'rdrerfrr rU)r3rkror'popitemr)rnhasattrinspectiscoroutinefunctionryr=get_running_looprun_in_executorr-rqr0r@)rGunload_start_timer rsunload_latency_mss rIr^z'_ModelMultiplexWrapper.unload_model_lruss "&&((( IKK+--5-99% 4444555 5) $ $ +.u}== &.00@@u}UUUUUUUUUUmmoo%%%%%%%*NEM!Y[[+<<F $,,->??? UH U U;L U U U U    #''X0F'GGGGGrK)rLN)__name__ __module__ __qualname____doc__rDrrhrintrJrrRrEr`rvr^r{rKrIrrs  -I)U$!3%*-U$U$%( U$U$U$U$n!49!!!!,FFFFFFPHHHHHHrKr)r=rloggingro collectionsrtypingrrrr ray.serverray.serve._private.commonr r ray.serve._private.constantsr r r ray.serve._private.metrics_utilsrray.serve._private.usagerray.serve.contextrr getLoggerr)rr{rKrIrsR ######++++++++++++CCCCCCCC ;:::::222222OOOOOOOO  , - -nHnHnHnHnHnHnHnHnHnHrK