o i@sTddlZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl Z ddl mZddlmZddlmZddlmZmZddlmZdd lmZdd lmZmZdd lmZdd lm Z dd l!m"Z"m#Z#ddl$m%Z%m&Z&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-m.Z.ddl/m0Z0m1Z1ddl2m3Z3ddl4m5Z5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDddlEmFZFmGZGddlHmIZIddlJmKZKmLZLddlMmNZNdd lOmPZPmQZQdd!lRmSZSdd"lTmUZUdd#lVmWZWmXZXmYZYdd$lZm[Z[dd%l\m]Z]ee^Z_Gd&d'd'e`ZaGd(d)d)eZbdS)*N)AsyncGeneratorIterableMapping)copy)Any) TokensPrompt) VllmConfig)WeightTransferInitRequestWeightTransferUpdateRequest)AsyncEngineArgs) EngineClient) PromptTypeStreamingInput) init_logger) LoRARequest)MULTIMODAL_REGISTRYMultiModalRegistry)STREAM_FINISHEDPoolingRequestOutput RequestOutput)get_io_processor) PoolingParams) BaseRenderer merge_kwargs) DictPrompt TokPrompt)extract_prompt_components)RequestOutputKindSamplingParams) SupportedTask) TokenizerLike) init_tracer)(maybe_register_config_serialize_by_value) UsageContext)cancel_task_threadsafe)as_list)EngineCoreRequest PauseMode)EngineCoreClient)EngineDeadErrorEngineGenerateError)InputProcessor)OutputProcessorRequestOutputCollector) ParentRequest)Executor)StatLoggerFactoryStatLoggerManager!load_stat_logger_plugin_factories)shutdown_prometheus)IterationStatscs&eZdZdZdeffdd ZZS)InputStreamErrorzWrapper for errors from the input stream generator. This is used to propagate errors from the user's input generator without wrapping them in EngineGenerateError. causecs||_tt|dSN)r6super__init__str)selfr6 __class__N/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm/v1/engine/async_llm.pyr9BszInputStreamError.__init__)__name__ __module__ __qualname____doc__ Exceptionr9 __classcell__r>r>r<r?r5;sr5c@seZdZejeddddddddf dedeede d ed e d e d e d e de e dBde de eefdBdededdfddZedejdddddddf ded e d ede e dBde de de de eefdBdededdfddZedejdfded e d ede e dBddf ddZddZdd Zdeed!ffd"d#Z       dd$ed%eeBeBeBeedfBd&e e!Bd'e"dBd(e#dBd)e ee$fdBd*e%eefdBd+ed,edBd-edBde&fd.d/Z'd0ed%edBd1e(dBd2ed3e&f d4d5Z)      dd$ed6eedfd7e e!Bd'e"dBd(e#dBd)e ee$fdBd*e%eefdBd+ed,edBde&fd8d9Z*e+d&e e!Bfd:d;Z,ddddddd<d%eeBeBeBeedfBd7e d$ed-edBd(e#dBd)e ee$fdBd*e%eefdBd+ed,edBdee-dffd=d>Z.d?d@Z/ dd$ee0eBdAe ddfdBdCZ1dDdddEdFe2dGe dBdHe ddfdIdJZ3ddKdLZ4de fdMdNZ5    dd%eeBeBdOe!d$ed(e#dBd*e%eefdBd+ed)e ee$fdBdee6dffdPdQZ7e8de9dBfdRdSZ:de9fdTdUZ;e8deddZd[Z?dd\d]Z@dd^d_ZAdd`daZBddbdcZC ddde dee de fdfdgZDddhdiZEddjeddfdkdlZFddme edBddfdndoZGde fdpdqZHd(e#de fdrdsZIdtede fdudvZJdeKefdwdxZLdtede fdydzZM  { dd|ed}e"dBd~ede dBfddZNddefddZO ddedefddZPe8de fddZQe8de fddZRe8de fddZSe8deTfddZUd0eVddfddZWd0eXddfddZYdS)AsyncLLMFTNr vllm_configexecutor_class log_stats usage_context mm_registryuse_cached_outputs log_requestsstart_engine_loop stat_loggersaggregate_engine_loggingclient_addresses client_count client_indexreturncCst|j|_||_|j|_|jj}|durtd|||_t| p"g}|t t |}|p1||_ |s<|rr>r?r9Hs%            zAsyncLLM.__init__enable_log_requestsdisable_log_statsc Cs&||t||||| |||| | d S)N) rHrIrOrPrNrJrQrKrRrSrT)r/ get_class) clsrHrOrKrPrrQrrRrSrTr>r>r?from_vllm_configszAsyncLLM.from_vllm_config engine_argsc Cs0||}t|}||||j|j |||dS)z'Create an AsyncLLM from the EngineArgs.)rHrIrNrJrOrKrP)create_engine_configr/rrr)rrrOrKrPrHrIr>r>r?from_engine_argss zAsyncLLM.from_engine_argscCs |dSr7)shutdownr;r>r>r?__del__s zAsyncLLM.__del__cCsZtt|dd}r|t|dd}r|t|dd}|dur+t|dSdS)z2Shutdown, cleaning up the background proc and IPC.rqNrirz)r3getattrrcloser$)r;rqrihandlerr>r>r?rs  zAsyncLLM.shutdown.cs$t|ds|jIdH|_|jS)N_supported_tasks)hasattrrqget_supported_tasks_asyncrrr>r>r?get_supported_taskss zAsyncLLM.get_supported_tasks request_idpromptparams arrival_time lora_requesttokenization_kwargs trace_headersprioritydata_parallel_rank prompt_textc  s,jrtt|t} jjjr| s|jrtd|j dur9t |j } t j d| dtddt|t|j d}t|trN||||||||| IdHSt|tr`|} || jkr_tdn&| durhtd jj||||||||| IdHd } tj|\} }}j| j4IdHjfd d IdHWdIdHn 1IdHswYt |j!| j}| j"}| s|j#d krو$| | dd|IdH|S|}t|t%sJt&| }t'|j#D](}|(|\}}||j#d kr| nt)| }||_||_*$|| |||IdHq|S)z Add new request to the AsyncLLM.z--kv-sharing-fast-prefill produces incorrect logprobs for prompt tokens, please disable it when the requests need prompt logprobsNz+The `truncate_prompt_tokens` parameter in `z_` is deprecated and will be removed in v0.16. Please pass it via `tokenization_kwargs` instead. stacklevel)truncate_prompt_tokenszAsyncLLM.add_request() was passed a request_id parameter that does not match the EngineCoreRequest.request_id attribute. The latter will be used, and the former will be ignored.z6should only provide prompt_text with EngineCoreRequest)rrrrrrsupported_taskscsj Sr7)rxr>rr>r?sz&AsyncLLM.add_request..rGr)+erroredr) isinstancerrH cache_configkv_sharing_fast_prefillprompt_logprobs ValueErrorrtyper@warningswarnDeprecationWarningrdictr_add_streaming_input_requestr&rrg warning_onceriprocess_inputsrrraassign_request_idr|rwwait_forr- output_kindrn _add_requestrr.rangeget_child_inforsampling_params)r;rrrrrrrrrr is_pooling params_typerequestr[queue parent_paramsparent_requestidx child_params child_requestr>rr? add_requests          (  zAsyncLLM.add_requestr parent_reqindexrcsD|j||||||j|IdH|jr td|jdSdS)NzAdded request %s.)rnrrqadd_request_asyncrNrgrhr)r;rrrrrr>r>r?rs  zAsyncLLM._add_request input_streamrc st|||||| djsd_jjdtdgddjjt j fdd}  t | _S) N)rrrrrrTr)prompt_token_ids)rrrc s<d}zzJ2zD3dHW}|j}|r|n}jjd|j|dd}|_|jdur5tdtj |j\}}} ||ddIdHq6Wn&t j t fy\d}Yntyt}z t|WYd}~nd}~wwWd_|s dddIdHdSdSd_|s dddIdHww)NFT)rrr resumablez0prompt_embeds not supported for streaming inputsrr>)r)_validate_streaming_input_sampling_paramsrirrexternal_req_id prompt_embedsrrrarruCancelledError GeneratorExitrDputr5_input_stream_task) cancelled input_chunkspreqrr[error final_reqrinputsinternal_req_idrrrr;r>r? handle_inputssR   z.handle_inputsr>)rr skip_cloneclonerirrrrr-rr|ru create_taskr) r;rrrrrrrrrrr>rr?rs6     (z%AsyncLLM._add_streaming_input_requestcCs2t|tr|jdks|jtjks|jrtddS)NrGzrInput streaming not currently supported for pooling models, n > 1, request_kind = FINAL_ONLY or with stop strings.)rrrrr FINAL_ONLYstopr)rr>r>r?rs  z2AsyncLLM._validate_streaming_input_sampling_params)rrrrrrc Cs,d} zz5|j|||||||| |d IdH} d} | s9| p%| IdH} t| ts-J| j} | tur7| V| rWntjt fy]| durS|j | j ddIdH|j r\t d|tym|j rlt d|ty} z |j r~t d|| d} ~ wty} z| dur|j | j ddIdH|j rt d || | j| d} ~ wty} zF| dur|j | j ddIdH|j rz | jjd | }Wnty}z| jjd |jj}WYd}~nd}~wwt d ||t| d} ~ wwW| dur | dSdS| dur| ww) aj Main function called by the API server to kick off a request * 1) Making an AsyncStream corresponding to the Request. * 2) Processing the Input. * 3) Adding the Request to the Detokenizer. * 4) Adding the Request to the EngineCore (separate process). A separate output_handler loop runs in a background AsyncIO task, pulling outputs from EngineCore and putting them into the per-request AsyncStream. The caller of generate() iterates the returned AsyncGenerator, returning the RequestOutput back to the caller. N)rrrrrrFTinternalRequest %s aborted. Request %s failed (engine dead).z$Request %s failed (bad request): %s.z$Request %s failed (input error): %s.z: z-: error during printing an exception of classzRequest %s failed due to %s.)r get_nowaitgetrrfinishedrrurrabortrrNrgrhr)rr5r6rDr=r@r*r)r;rrrrrrrrrqroutese2r>r>r?generates         zAsyncLLM.generatecsZ|jdurdS|j|j|j|j|jtjfdd}t ||_dS)zBBackground loop: pulls from EngineCore and pushes to AsyncStreams.Nc s zg IdH}t|j}r|rtnd}|j}td|D]1}|}|||}||j|}|jr:J||krFt dIdH|j rR |j IdHq! |j rhj|j|j |dqty}ztd|WYd}~dSd}~ww)NTr) engine_idxscheduler_statsiteration_statsmm_cache_statszAsyncLLM output_handler failed.)get_output_asynclenoutputsr4rprocess_outputs timestamprequest_outputsrusleep reqs_to_abortabort_requests_asyncupdate_scheduler_statsrrecord engine_index stat_mm_cacherDrg exceptionpropagate_error) r num_outputsrengine_core_outputsstartend outputs_sliceprocessed_outputsr chunk_sizerqrirJrrrnr>r?rzsH     - z4AsyncLLM._run_output_handler..output_handler) rzrqrnrJrrrienvsVLLM_V1_OUTPUT_PROC_CHUNK_SIZErur)r;rzr>rr?r|s 3zAsyncLLM._run_output_handlerrcsZt|tr |fnt|}|j||}|j|IdH|jr+t dd |dSdS)z2Abort RequestId in OutputProcessor and EngineCore.NzAborted request(s) %s.,) rr:r%rnabort_requestsrqrrNrgrhjoin)r;rr request_idsall_request_idsr>r>r?rszAsyncLLM.abortr)modewait_for_inflight_requests clear_cacherrrc s&|r tjdtddd}|dkr|jIdHn^|jdkr#td|j4IdHC|jscd |_|d krJt |j j }|rI|j |d d IdHn|dkr\|j r[|j IdHntd |WdIdHn 1IdHsswY|r|IdH|IdH|IdHdSdS) aQ Pause generation to allow model weight updates. New generation/encoding requests are blocked until resume. Args: mode: How to handle in-flight requests: - ``"abort"``: Abort all in-flight requests immediately (default). - ``"wait"``: Wait for in-flight requests to complete. - ``"keep"``: Freeze requests in queue; they resume on :meth:`resume_generation`. wait_for_inflight_requests: DEPRECATED: use mode argument. Whether to wait for in-flight requests to complete before pausing. clear_cache: Whether to clear KV cache and prefix cache after draining. Set to ``False`` to preserve cache for faster resume. Default is ``True`` (clear caches). z~The `wait_for_inflight_requests` parameter in `AsyncLLM.pause_generation()` is deprecated. Please use `mode` argument instead.rrwaitkeepNrGzUpause_generation is not supported with --api-server-count > 1 when mode is not 'keep'TrrzInvalid mode: )rrrrqpause_scheduler_asyncryNotImplementedErrorrwrxrdrnrequest_stateskeysrhas_unfinished_requestswait_for_requests_to_drainrreset_prefix_cachereset_mm_cachereset_encoder_cache)r;rrrrr>r>r?pause_generationsB  (zAsyncLLM.pause_generationc sb|j4IdH|jIdHd|_|jWdIdHdS1IdHs*wYdS)z1Resume generation after :meth:`pause_generation`.NF)rwrqresume_scheduler_asyncrx notify_allrr>r>r?resume_generations  .zAsyncLLM.resume_generationc sF|j4IdH|jWdIdHS1IdHswYdS)z.Return whether the engine is currently paused.N)rwrxrr>r>r? is_paused#s0zAsyncLLM.is_pausedpooling_paramsc Csfd}zz/|j|||||||dIdH}d} | s2|p"|IdH} t| ts*J| j} | V| rWnhtjyT|durJ|j|j ddIdH|j rSt d|t yd|j rct d|tyt|j rst d|ty} z|dur|j|j ddIdH|j rt d |t| d} ~ wwW|dur|dSdS|dur|ww) a2 Main function called by the API server to kick off a request * 1) Making an AsyncStream corresponding to the Request. * 2) Processing the Input. * 3) Adding the Request to the EngineCore (separate process). A separate output_handler loop runs in a background AsyncIO task, pulling outputs from EngineCore and putting them into the per-request AsyncStream. The caller of generate() iterates the returned AsyncGenerator, returning the RequestOutput back to the caller. N)rrrrFTrrrz Request %s failed (bad request).zRequest %s failed.)rrrrrrrurrrrNrgrhr)rrDr*r) r;rr0rrrrrrrrrr>r>r?encode)s`           zAsyncLLM.encodecC|jjSr7)rirlrr>r>r?rlzzAsyncLLM.tokenizercCs |jSr7)ri get_tokenizerrr>r>r?r4~s zAsyncLLM.get_tokenizercCr2r7)rirendererrr>r>r?r5r3zAsyncLLM.renderercs|jjduSr7)rbrcrr>r>r?is_tracing_enableds zAsyncLLM.is_tracing_enabledcs|jr |jdSdSr7)rrlogrr>r>r? do_log_statsszAsyncLLM.do_log_statscstd|jr |jdS)NzCalled check_health.)rgdebugr dead_errorrr>r>r? check_healths  zAsyncLLM.check_healthcB|jdg}|jdur|t|jjtj|IdHdS)NT)rq profile_asyncrappendru to_threadrgatherr;corosr>r>r? start_profile  zAsyncLLM.start_profilecr<)NF)rqr=rr>rur?rr@rAr>r>r? stop_profilerDzAsyncLLM.stop_profilecs |j|jIdHdSr7)riclear_mm_cacherqreset_mm_cache_asyncrr>r>r?r)s zAsyncLLM.reset_mm_cachereset_running_requestsreset_connectorcs|j||IdHSr7)rqreset_prefix_cache_async)r;rHrIr>r>r?r(s zAsyncLLM.reset_prefix_cachecs|jIdHdSr7)rqreset_encoder_cache_asyncrr>r>r?r*szAsyncLLM.reset_encoder_cachelevelcsB|IdH|j|IdH|jdur|jd|dSdS)NrG)r(rq sleep_asyncrrrecord_sleep_state)r;rLr>r>r?rs  zAsyncLLM.sleeptagscs4|j|IdH|jdur|jdddSdS)Nr)rq wake_up_asyncrrrN)r;rOr>r>r?wake_ups  zAsyncLLM.wake_upcs|jIdHSr7)rqis_sleeping_asyncrr>r>r? is_sleepingszAsyncLLM.is_sleepingc|j|IdHS)zr>r?add_lorazAsyncLLM.add_loralora_idcrT)z&Remove an already loaded LoRA adapter.N)rqremove_lora_asyncr;rXr>r>r? remove_lorarWzAsyncLLM.remove_loracs|jIdHS)zList all registered adapters.N)rqlist_loras_asyncrr>r>r? list_lorasszAsyncLLM.list_lorascrT)z&Prevent an adapter from being evicted.N)rqpin_lora_asyncrZr>r>r?pin_lorarWzAsyncLLM.pin_lorar>methodtimeoutargskwargscs|j||||IdHS)zB Perform a collective RPC call to the given path. N)rqcollective_rpc_async)r;r`rarbrcr>r>r?collective_rpcs  zAsyncLLM.collective_rpc, drain_timeoutcslt}t||kr.|jstddStdtdIdHt||ks td|d)z$Wait for all requests to be drained.z,Engines are idle, requests have been drainedNz;Engines are still running, waiting for requests to drain...rGzTimeout reached after z' seconds waiting for requests to drain.)timerqdp_engines_runningrgrhrur TimeoutError)r;rg start_timer>r>r?r's    z#AsyncLLM.wait_for_requests_to_drainnew_data_parallel_sizecs|jjj}||krtd|dStd|||IdHtd||j|IdH||jj_||krJ|jrLt |jt t |dd|_ dSdSdS)a Scale up or down the data parallel size by adding or removing engine cores. Args: new_data_parallel_size: The new number of data parallel workers drain_timeout: Maximum time to wait for requests to drain (seconds) z0Data parallel size is already %s, skipping scaleNz@Waiting for requests to drain before scaling up to %s engines...z?Requests have been drained, proceeding with scale to %s engines)rHrWrX) rHparallel_configdata_parallel_sizergrhr'rqscale_elastic_eprJr1rdrrr)r;rlrgold_data_parallel_sizer>r>r?ros4    zAsyncLLM.scale_elastic_epcCs|jdup |j Sr7)rzdonerr>r>r? is_runningszAsyncLLM.is_runningcCs|jSr7)rrr>r>r? is_stoppedzAsyncLLM.is_stoppedcCs|jjjp|j Sr7)rq resources engine_deadrrrr>r>r?rszAsyncLLM.erroredcCstSr7)r)rr>r>r?r: rtzAsyncLLM.dead_errorcsNddlm}t||r|j}n tdt||jdd|idIdHdS)z Initialize weight transfer for RL training. Args: request: Weight transfer initialization request with backend-specific info r)r z(Expected WeightTransferInitRequest, got init_weight_transfer_engine init_inforcN)%vllm.distributed.weight_transfer.baser rrx TypeErrorrre)r;rr init_info_dictr>r>r?rw$s z$AsyncLLM.init_weight_transfer_enginecsBt|tr |j}n tdt||jdd|idIdHdS)z Batched weight update for RL training. Args: request: Weight update request with backend-specific update info z*Expected WeightTransferUpdateRequest, got update_weights update_inforyN)rr r~r{rre)r;rupdate_info_dictr>r>r?r}:s  zAsyncLLM.update_weights)NNNNrNN)NNNNrN)F)rUN)NNrN)FF)rGr7)Nr>N)rf)Zr@rArBr#ENGINE_CONTEXTrrrr/rfrrdr0rr:intr9 classmethodrr rrrtuplerrr&r rrrrrrfloatrrrr-rr.rr staticmethodrrrr|rrr'r+r.r/rr1propertyr rlr4rr5r6r8r;rCrEr)r(r*rrQrSrVr[setr]r_rer'rorrrsr BaseExceptionr:r rwr r}r>r>r>r?rFGsp                         U        nE   B   Q        * rF)crurrrhrcollections.abcrrrrtypingrrZ vllm.envsrvllmr vllm.configrrzr r vllm.engine.arg_utilsr vllm.engine.protocolr vllm.inputsr r vllm.loggerrvllm.lora.requestrvllm.multimodalrr vllm.outputsrrrvllm.plugins.io_processorsrvllm.pooling_paramsrvllm.renderersrrvllm.renderers.inputsrr vllm.renderers.inputs.preprocessrvllm.sampling_paramsrr vllm.tasksrvllm.tokenizersr vllm.tracingr!vllm.transformers_utils.configr"vllm.usage.usage_libr#vllm.utils.async_utilsr$vllm.utils.collection_utilsr%vllm.v1.enginer&r'vllm.v1.engine.core_clientr(vllm.v1.engine.exceptionsr)r*vllm.v1.engine.input_processorr+vllm.v1.engine.output_processorr,r- vllm.v1.engine.parallel_samplingr.vllm.v1.executorr/vllm.v1.metrics.loggersr0r1r2vllm.v1.metrics.prometheusr3vllm.v1.metrics.statsr4r@rgrDr5rFr>r>r>r?s\