o ٷi'N@sddlmZddlmZddlZddlZddlmZddl m Z ddl m Z ddl mZddlmZmZmZdd lmZdd lmZmZmZdd lmZdd lmZdd lmZeeZ GdddeZ!GdddeZ"dS))Callable)AnyN) init_logger)PoolingRequestOutput)RequestOutputKind) TokenizerLike)EngineCoreOutputEngineCoreRequest FinishReason)OutputProcessor)OutputProcessorOutputRequestOutputCollector RequestState) ParentRequest)IterationStats)OmniRequestOutputcseZdZdZfddZdedBdedBddfdd Zdd d Z  dd e e d e j dBde dBde eBdBdeeefdBdejdBdeeBdBffdd Z dde e de dBde eBdBdejdBdef fdd ZZS)OmniRequestStatezRequest state for omni models, tracking multimodal outputs. Extends the base RequestState with support for accumulating multimodal tensor outputs (e.g., images, audio, latents) that are produced incrementally during generation. cs"tj|i|d|_d|_dSN)super__init__mm_typemm_accumulated)selfargskwargs __class__U/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/engine/output_processor.pyr!s zOmniRequestState.__init__payloadNrreturnc s|durdSz|r|p d|_ddt|trTi}|jpd}|D].\}}|dkr/|}n |dkr9|dkr9|}t|trLfdd|D||<q$|||<q$n |jpXd}||i}|jdurj||_WdS|D]\}}||jvr}||j|<qn|j|}t|tjrt|tjr||g|j|<qnt|tjrt|tr| |qnt|trt|tr|D]=\} } | |vr| || <qt| tjrt|| tjr|| | g|| <qt| tjrt|| tr||  | q| || <qqn||j|<qnWdSt y t dYdSw) NcSs@t|tjrz |jdddWSty|YSw|S)NcpuT) non_blocking) isinstancetorchTensordetachto contiguous Exception)xrrr_to_cpu3s  z7OmniRequestState.add_multimodal_tensor.._to_cpuhidden model_outputscsi|] \}}t||qSr)str).0sksvr,rr Hsz:OmniRequestState.add_multimodal_tensor..z$Error accumulating multimodal tensor) lowerrr$dictitemsrr%r&listappendr*logger exception) rrrincoming target_keykvkeyexistingr1r2rr3radd_multimodal_tensor+sZ             z&OmniRequestState.add_multimodal_tensorc Cs.|jdurdSz|jD]w\}}t|trL|rLt|dtjrLz|dkr'Wq tj|dd|j|<Wq tyKt d|d|d|j|<Yq wt|t r|D].\}}t|tr|rt|dtjrz tj|dd||<WqUty|d||<YqUwqUq WdStyt dYdSw) zKConsolidate accumulated tensor lists into single tensors via concatenation.Nraudio)dimz#Error concatenating tensor for key z; keeping last tensorz&Error consolidating multimodal tensors) rr7r$r8r%r&catr*r:warningr6r;)rr>r?r1r2rrr_consolidate_multimodal_tensorsos4     z0OmniRequestState._consolidate_multimodal_tensors new_token_idspooling_output finish_reason stop_reasonkv_transfer_paramsrouted_expertsc s|jdur|durt||||||S|du}|jtjk}|s$|r$dS|r*||jdkra|jdus6J|sK|jdksKt |jj |j|jksKdS|jtj kra|jj |jd}t |jj |_|j } | ||||} |jduru| g} n|j|j| \} }| sdS|jj } || | ||S)aCreate a request output from generation results. Creates a RequestOutput or PoolingRequestOutput from the generated tokens and accumulated multimodal outputs. Attaches multimodal tensors to the completion output if available. Args: new_token_ids: List of newly generated token IDs pooling_output: Optional pooling output tensor finish_reason: Optional finish reason indicating why generation stopped stop_reason: Optional stop reason (token ID or stop string) kv_transfer_params: Optional KV cache transfer parameters Returns: OmniRequestOutput or PoolingRequestOutput if output should be emitted (based on finish status and output kind), None otherwise Nr) detokenizerrmake_request_output output_kindr FINAL_ONLYrHstream_intervalsent_tokens_offsetlenoutput_token_idsDELTAexternal_req_id_new_completion_output parent_req get_outputs request_id_new_request_output) rrIrJrKrLrMrNfinished final_onlyrYoutputoutputsrrrrQsD     z$OmniRequestState.make_request_output token_idsc st||||}z8|jdur?t|dst|dit|d}t|tr5|jD]\}}|||<q)W|St|d|jW|SW|St yPt dY|Sw)Nmultimodal_outputzError in _new_completion_output) rrZrhasattrsetattrgetattrr$r6r7r*r:r;) rrcrKrLrN base_outputmm_outr>r?rrrrZs&       z'OmniRequestState._new_completion_output)r NNNr)__name__ __module__ __qualname____doc__rrr/rBrHr8intr%r&r r6npndarrayrrrQrZ __classcell__rrrrrsD  D$  U rc sheZdZdZ  d/dedBdedededBffdd Zd ed e e gdfd dfd dZ   d0de dedBde dBdededBd df ddZ  d1dee dedBdedBd effdd Zde d dfddZde d dfdd Zde d dfd!d"Zde d dfd#d$Zde d dfd%d&Zde d dfd'd(Zde d dfd)d*Zde d+eed,fd ejdBfd-d.ZZ S)2MultimodalOutputProcessora Handles multimodal output processing by normalizing EngineCoreOutput before delegating to the base vLLM OutputProcessor. Strategy: - Route by EngineCoreOutput.output_type when present ("image", "text+image", "latents", "text"). - Fallback to pooling/text heuristics when output_type is absent. - Mutate EngineCoreOutput in-place to ensure vLLM's base processor can produce the correct RequestOutput/PoolingRequestOutput. - Allow custom per-modality handlers via register_handler(). rON tokenizer log_statsrTengine_core_output_typecs(tj|||di|_i|_||_dS)aInitialize the multimodal output processor. Args: tokenizer: Tokenizer for detokenizing text outputs log_stats: Whether to log statistics stream_interval: Stream interval for output generation engine_core_output_type: Optional output type specification (e.g., "image", "audio", "latents"). Used to route outputs to appropriate processors. If None, output type is inferred. )rtrurTN)rroutput_handlers_reqid_to_mm_typerv)rrtrurTrvrrrrs z"MultimodalOutputProcessor.__init__modalityhandlerr cCs||j|<dS)aRegister a custom handler for a specific modality. Allows custom processing logic for specific output modalities. The handler is called before default processing for outputs matching the specified modality. Args: modality: Modality name (e.g., "image", "audio", "latents") handler: Callable that takes an EngineCoreOutput and processes it N)rwr5)rryrzrrrregister_handlers z*MultimodalOutputProcessor.register_handlerrrequestpromptr[ request_indexqueuec Cs|j}|j|}|dur||||dStj|j||||||j|jd}|j r0|j ||j|<|r=||j |j<|j |j|dS)aAdd a new request to be processed. Creates an OmniRequestState for the request and registers it for output processing. Args: request: Engine core request to add prompt: Optional prompt string for the request parent_req: Optional parent request for parallel sampling request_index: Index of the request in the batch queue: Optional queue for collecting outputs Raises: ValueError: If the request ID is already registered N)rtr|r}r[r~rrurT)r]request_statesget_update_streaming_request_staterfrom_new_requestrtrurT_requests_drainedis_setclearparent_requestsexternal_req_idsrYr9)rr|r}r[r~rr] req_staterrr add_request&s*    z%MultimodalOutputProcessor.add_requestengine_core_outputsengine_core_timestampiteration_statscs|j|D]D}|jp d}|r||j|j<|||j|j}|dus-t|t s.q|j durK|j durK| |j t |d|jpDdd|_ qtj|||dS)Nr! output_type)rr)rxrrvr5r]_route_and_normalizerrr$rrJrPrBrgrprocess_outputs)rrrrecorrrrrrTs*   z)MultimodalOutputProcessor.process_outputsrcCst|d|jpd}||jvr)z |j||Wnty(td|Ynw|dkr4||dS|dvr?||dS|dvrJ| |dS|dvrU| |dS|dkr`| |dS|j durl| |dS| |dS) Nrr!z%Error in custom output handler for %simage)z text+imagez text,imagez image+text)latentslatent)rCspeechtext)rgrvr5rwr*r:r;_process_image_output_process_text_image_output_process_latents_output_process_audio_output_process_text_outputrJ_process_pooling_output)rrrrrrrrs(   z.MultimodalOutputProcessor._route_and_normalizecC2|jdur|j|dd}|dur||_dSdSdS)z>Ensure image tensors are surfaced via pooling_output for vLLM.N)rimages pixel_valuespixelskeysrJ _extract_from_multimodal_outputsrrtensorrrrr  z/MultimodalOutputProcessor._process_image_outputcCr)zkAllow text+image outputs. Text path stays as new_token_ids; image/latents route via pooling_output.N)rrrrrrzrrrrrrrs   z4MultimodalOutputProcessor._process_text_image_outputcCr)z6Ensure latent tensors are surfaced via pooling_output.N)rrr posteriorrrrrrrrrz1MultimodalOutputProcessor._process_latents_outputcCr)z5Ensure audio tensors are surfaced via pooling_output.N)rCaudioswavwaveform audio_pcmpcmrrrrrrrs  z/MultimodalOutputProcessor._process_audio_outputcCsdS)u=No-op; base processor will detokenize new_token_ids → text.Nrrrrrrrsz.MultimodalOutputProcessor._process_text_outputcCsJ|jdurdSt|jtjs#z t|j|_WdSty"YdSwdS)z*Optional sanity checks for pooling tensor.N)rJr$r%r& as_tensorr*rrrrrs  z1MultimodalOutputProcessor._process_pooling_outputr.cCsht|dd}t|ts dS|D]}||}t|tjr |Sq|D] }t|tjr1|Sq%dS)Nmultimodal_outputs)rgr$r6rr%r&values)rrrmmr>r?rrrrs      z:MultimodalOutputProcessor._extract_from_multimodal_outputs)rON)NrNrj)!rkrlrmrnrboolror/rrrr{r rr rr8floatrr rrrrrrrrtupler%r&rrrrrrrrssb  1 , rs)#collections.abcrtypingrnumpyrpr% vllm.loggerr vllm.outputsrvllm.sampling_paramsrvllm.tokenizersrvllm.v1.enginerr r vllm.v1.engine.output_processorr VLLMOutputProcessorr r r vllm.v1.engine.parallel_samplingrvllm.v1.metrics.statsrvllm_omni.outputsrrkr:rrsrrrrs$         ^