o ٷi7@sddlZddlmZddlmZddlmZddlmZddl m Z e e Z ded e d e d ed ed ededeeeefgdfdedefddZdeeefdeeefd e deeeeefdBffddZdee de fddZdS)N)Callable)Any)OmniStageTaskType)OrchestratorAggregator)get_connector_logger connectorstage_id next_stage_idreq_id next_inputssampling_paramsoriginal_promptnext_stage_queue_submit_fnmetricsreturnc  Csz_t} ||||d|tdd} |t|t|t|| \} } } | r]tj||dt|t|td}| r@| |d<||t}|| d}||||| t|dWdSWdStyy}zt d ||WYd }~dSd }~ww) z Attempts to send data via OmniConnector. Returns True if successful, False otherwise. Encapsulates the logic of preparing payload, sending via connector, sending notification, and recording metrics. z->)rstage_transition timestamp) engine_inputsr metadataT)type request_idr from_connector from_stageto_stagesent_tsconnector_metadata@@FzI[Orchestrator] OmniConnector failed for req %s: %s; falling back to queueN) timeputstrrGENERATE on_forwardfloat Exceptionloggerwarning)rr r r r r rrrt0 payload_datasuccessserialized_sizernotify_payloadt1tx_mser/a/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/distributed/omni_connectors/adapter.pytry_send_via_connectorsV "   r1task connectorsc Cs|d}|dr|d}t|}|std||dS||f}||}|rzXt}|d} |j||t|| d} t} | rWt| trM| \} } n| } t|| } nd} d } | rwt| t rw| d }| |d }|| d }||fWStd ||WdSt y}ztd|||WYd}~dSd}~wwtd||||dSd dl m }z ||d d\}}||fWSt yYdSw)z Attempts to resolve input data from either connector or IPC. Returns (engine_inputs, rx_metrics) or (None, None) if failed/skipped. rrrzN[Stage-%s] 'from_connector' is true but 'from_stage' is missing for request %s)NNr)rNrrr)rx_decode_time_msrx_transfer_byteszO[Stage-%s] Failed to get data from connector for request %s or payload is emptyzB[Stage-%s] Error retrieving data from connector for request %s: %sz>[Stage-%s] No connector found for edge %s -> %s for request %s) maybe_load_from_ipc_with_metricsengine_inputs_shm) getr r%errorr isinstancetuplelen serialize_objdictr$!vllm_omni.entrypoints.stage_utilsr6)r2r3r ridrr connector_keyr_t_startrpayload_t_endr(r*ein decode_ms rx_metricsr.r6rr/r/r0try_recv_via_connector^sb              rH prompt_idsc sdd}d}d}fddttD}|td}d}tt|dD]4}||}||d} |d} | |kr@q)| |krK|| |7}q)| |kr\|t|d kr\|d 7}q) q)||S) zCompute the length of the talker prompt ids. Args: prompt_ids: The prompt ids tensor. Returns: The length of the talker prompt ids. i\Pi"ihi#-csg|] }|kr|qSr/r/).0iim_start_token_idrIr/r0 sz4compute_talker_prompt_ids_length..rr )ranger<append) rIsystem_token_id user_token_idassistant_token_idim_start_indexes sum_user_len assistant_lenrKsr.roler/rLr0 compute_talker_prompt_ids_lengths(    r[)rcollections.abcrtypingrr?rvllm_omni.metricsr utils.loggingr__name__r%intr r>boolr1r;rHlistr[r/r/r/r0sL       L   R