o ٷi@sddlZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z mZddlmZmZmZddlZddlZddlZddlmZddlmZddlmZddlmZdd lm Z dd l!m"Z"dd l#m$Z$m%Z%dd l&m'Z'dd l(m)Z)ddl*m+Z+m,Z,m-Z-ddl.m/Z/ddl0m1Z1m2Z2ddl0m3Z4ddl5m6Z6m7Z7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@mAZAddlBmCZCddlDmEZEeeFZG    d&dejHdBdejIdBdejJdBfddZKddZLdeMfd d!ZNGd"d#d#ZOGd$d%d%eOZPdS)'N)Callable GeneratorSequence)ThreadPoolExecutor as_completed)AnyLiteraloverload) OmegaConf)tqdm)SamplingParams) init_logger)make_zmq_socket)get_engine_client_zmq_addr)get_stage_connector_config"initialize_orchestrator_connectors)try_send_via_connector) resolve_omni_kv_config_for_stage)create_placement_groupget_ray_queue_class try_close_ray OmniStage) SHUTDOWN_TASKOmniStageTaskType)maybe_load_from_ipc)get_final_stage_id_for_e2einject_omni_kv_configload_and_resolve_stage_configs)ZmqQueue)OmniDiffusionSamplingParamsOmniPromptTypeOmniSamplingParams)OrchestratorAggregatorStageRequestStats)!download_weights_from_hf_specific)OmniRequestOutputhandshake_stopzmq_handshake_sockethandshake_threadc CsB|rm|D]1}z|tWnty'} ztd| WYd} ~ nd} ~ wwt|dd} t| r5| q|D]}t|dd} t| rG| q8|D]"} z| WqJtyl} ztd| WYd} ~ qJd} ~ wwt||dury| |dur|j dd| rtd|dur| d|dur| dSdS) z7Weak reference cleanup function for OmniBase instances.z5Failed to send shutdown signal to stage input queue: NclosezFailed to stop stage worker: g@timeoutzCHandshake server thread did not terminate gracefully within timeoutr) put_nowaitr Exceptionloggerwarninggetattrcallablestop_stage_workerrsetjoinis_aliver*term) stage_liststage_in_queuesstage_out_queuesray_pgzmq_ctxr'r(r)qeclose_fnstagerAN/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/entrypoints/omni.py_weak_close_cleanup8sH        rCcCs|SNrA)model_idrArArB_dummy_snapshot_downloadfsrFreturncCsvtj|r|Stjddrddlm}||Sz t|ddgddW|Stj j y:t d|d Y|Sw) NVLLM_USE_MODELSCOPEFr)snapshot_download*T)model_name_or_path cache_dirallow_patterns require_allzRepository not found for 'z'.) ospathexistsenvironget modelscope.hub.snapshot_downloadrIr%huggingface_huberrorsRepositoryNotFoundErrorr/r0)rErIrArArBomni_snapshot_downloadjs"  rXc @seZdZdZdededdfddZdedBdeeefdBfd d ZdedBd edBdedBfd d Z deeefdeeeffddZ dedeeefde ee effddZ dedeeefddfddZde defddZdeddfddZdededeeefddfddZd8deddfd d!Zdedefd"d#Zd9d$e edBddfd%d&Zd9d$e edBdefd'd(Zd:d)d*Zd+edeeeffd,d-Zd:d.d/Zd:d0d1Zd8dedefd2d3Zedefd4d5Zedefd6d7Z dS);OmniBaseaBase class for serving Omni models. Args: model: Model name or path to load. **kwargs: Arbitrary keyword arguments. - stage_configs_path: Optional path to YAML file containing stage configurations. If None, configurations are loaded from the model. - log_stats: Whether to enable statistics logging be written to files with stage-specific suffixes. - stage_init_timeout: Per-stage init watchdog (seconds). Measured from when the previous stage finished (possibly a prior Omni run with GPU reuse/overlap) to when the current stage starts to initialize. - shm_threshold_bytes: Threshold in bytes for using shared memory for IPC. Objects larger than this threshold will use shared memory. - worker_backend: Backend for worker processes. Default is "multi_process". - ray_address: Address of Ray cluster for Ray backend, if using Ray backend. - batch_timeout: Timeout in seconds for batching requests within a stage - init_timeout: Timeout in seconds for waiting for all stages to initialize - Additional keyword arguments passed to stage engines. modelkwargsrGNcKst|}||d<g|_g|_g|_t|_d|_d|_d|_d|_ d|_ d|_ d|_ d|_ d|_i|_t|_d|_td||||dS)NrZzInitializing stages for model: )rXr8_stage_in_queues_stage_out_queuesr4 _stages_ready_ray_pg _queue_cls_ctx_zmq_ctx_zmq_master_address_zmq_master_port_zmq_handshake_socket_handshake_thread_handshake_stop_handshake_endpoints_handshake_seen_single_stage_idr/info_initialize_stagesselfrZr[rArArB__init__s(zOmniBase.__init__ cache_backendc Cs4|dkrdddddddddd S|d krd d iSdS) N cache_ditrgQ?Fdynamic) Fn_compute_blocksBn_compute_blocksmax_warmup_stepsresidual_diff_thresholdmax_continuous_cached_stepsenable_taylorseertaylorseer_orderscm_steps_mask_policyscm_steps_policy tea_cache rel_l1_threshg?rA)rnrprArArB_get_default_cache_configs z"OmniBase._get_default_cache_config cache_configcCsZt|trzt|}Wntjytdd}Ynw|dur+|dvr+||}|S)Nz*Invalid cache_config JSON, using defaults.)Nnone) isinstancestrjsonloadsJSONDecodeErrorr/r0r)rnrprrArArB_normalize_cache_configs   z OmniBase._normalize_cache_configc Csd|vr t|d|d<|dd}|||dd}d}d|vr6|dj}td|D] }|d |7}q,d d d |dd ti|||dd ddg}d |d dd<|S)z-Create default diffusion stage configuration.dtyperprrN0parallel_configrr,r diffusionT)processdevicesmax_batch_size)rprimage)stage_id stage_typeruntime engine_args final_outputfinal_output_typer model_stage)rrSr world_sizeranger create)rnr[rprr num_devicesidefault_stage_cfgrArArB#_create_default_diffusion_stage_cfgs:  z,OmniBase._create_default_diffusion_stage_cfgc sZdd}dd_t||fddd\}}|D]}zpt|dddkr+Wqt|d r5|jdur;ti|_d durTt|jd rN|jjdurTd |j_d }|durbd }|durvt|jd rr|jj durv||j_ d }|durt|jd r|jj dur||j_ Wqt y} z t d| WYd} ~ qd} ~ ww||fS)zJResolve stage configs and inject defaults shared by orchestrator/headless.stage_configs_pathNtts_max_instructions_lengthcs SrD)rrAr[rnrArB s z1OmniBase._resolve_stage_configs..)default_stage_cfg_factoryrrr lora_path lora_scalestatic_lora_scalequantization_configz*Failed to inject LoRA config for stage: %s)rSrrr1hasattrrr rrrrr.r/r0) rnrZr[r config_path stage_configscfgrrr>rArrB_resolve_stage_configssF          zOmniBase._resolve_stage_configsc s`|dd|dd}|dd}|dd}|d d }|d d }|d d}|dd _|dd _jd urEd_td|dd _||\__t j||d\_ _ t |_ |_|_|_j_dtttfdtttfffdd tttjtdtpdd%fddtjD} g} t| D] } | | qWd n1swY| j!dddd d| D_"d!dj"D_#d"dj"D_$td#j%d$tj"d%jd&krt&_'n t()d'_*fd(d_'td)t_+td)t|_,-|j.|d*d S)+z!Initialize stage list management.stage_init_timeoutshm_threshold_bytesi init_timeouti,worker_backend multi_process ray_addressN batch_timeout log_statsFromni_master_addressz 127.0.0.1zDNo omni_master_address provided, defaulting to localhost (127.0.0.1)omni_master_port)rridx_cfgrGcs|\}}|t|dfS)Nrr)ridxrrrArB _build_stageNsz1OmniBase._initialize_stages.._build_stagerr) max_workerscs g|] \}}||fqSrA)submit).0rr)rexecutorrArB Ss z/OmniBase._initialize_stages..cSs|dS)NrrA)xrArArBrWsz-OmniBase._initialize_stages..)keycSsg|]\}}|qSrArA)r_strArArBrXscSg|]}|jqSrA)default_sampling_paramsrrrArArBrYcSrrA)rrrArArBrZr[z ] Loaded z stagesrayspawncsjjddS)Nr)maxsize)raQueuerArnrArBrasrr+)/rSrjrcr/rkrdrrrromni_transfer_config connectorsboolrrrr_is_async_chunk_enable async_chunktupleintrrrminlenmaxrO cpu_count enumeraterappendresultsortr8default_sampling_params_listoutput_modalities_namerr`mp get_contextra_stage_init_timeout_shm_threshold_bytes _start_stages_wait_for_stages_ready) rnrZr[rrrrrrfuturesresultsfutrA)rrrnrrBrl,sV          &&      zOmniBase._initialize_stages stage_argscCs t|ddd}tt|ddS)zget async chunk flagrrNrF)r1r)rnrrrArArBriszOmniBase._is_async_chunk_enablec Cs@|jdkrtt|j|jdd|_nI|jdurt|_t|j }i|_ |j du}t |D])}t ||jd}t ||jd}||f|j |<td|jd|d|d |q-|tt|jD]\}}|jdkrt|} |} n|j |\}}t|jtj|d } t|jtj|d } |j| |j| || | t|j|} zt|j|\} } }| rt || | |Wnt!y}z td ||WYd}~nd}~ww|j dur|t"|j krt#d|jd |d qb|j$||j%|j&|jdkr|j'nd|j(| |j|j|j dur dndd td|jd|dqbdS)zStart all stage processes.rPACK)number_of_stagesaddressstrategyN) local_onlyhostrz ] Allocated endpoints for stage-z: in=z, out=)bindz?[Omni] Failed to inject omni connector config into stage-%s: %sz#] Skipping initialization of stage-z& worker due to single_stage_id settingTF)is_asyncrctxrconnectors_configrray_placement_groupignore_runtime_config] Stage-z process started))rrrr8rr_rbzmqContextrrhrjrrrcr/debugrstart_handshake_serverrrr`rPUSHPULLr\rr] attach_queuesrrrrr.rrkinit_stage_workerrrrar)rnrZ total_stagesrsid in_endpoint out_endpointrr@in_qout_qstage_connectors_config omni_conn_cfg omni_fromomni_tor>rArArBrnsx             zOmniBase._start_stagesr@rrcCs*|j|td|jd|ddS)Nrrz reported ready)r^addr/rkr)rnr@rrrArArB_process_stage_readys zOmniBase._process_stage_readyxr,c Cs|jdur|jdkr||}t|j}ttdt|}t d|j d|d|dt|j |krxt|krxd}t |jD] \}}||j vrMqC| }rcd }|d d krc||||qC|sktd t|j |krxt|ks&z2OmniBase._wait_for_stages_ready..z7] Stage initialization timeout. Troubleshooting Steps: )rjr_wait_for_handshakesrr8timerrr/rkrr^r try_collectrSrsleepsortedr4rr0r5) rnr, num_stagesdeadline progressedrr@r not_ready suggestionsformatted_suggestionsrArArBrsF       zOmniBase._wait_for_stages_readycCs^|j|}|jdkr dSt|jdd}|durdSt|dd}|dur%dSt|dd}|duS)z2Check if profiler config is set for a given stage.rTrNFprofiler_configprofiler)r8rr1 stage_config)rnrr@rr%r&rArArB_is_profiler_enableds    zOmniBase._is_profiler_enabledstagesc Cs|dur ttt|j}|D]H}|t|jkrW||s&td|j|qz|j|dt j itd|j|Wqt yV}zt d|j||WYd}~qd}~wwqdS)azStart profiling for specified stages. Sends start_profile command to stage workers. Profiling must be enabled via VLLM_TORCH_PROFILER_DIR environment variable. Args: stages: List of stage IDs to start profiling. If None, starts profiling for all stages that have profiling enabled. Example: >>> # Profile all stages >>> omni.start_profile() >>> outputs = omni.generate(prompts, sampling_params) >>> omni.stop_profile() >>> # Profile only stage 0 and 2 >>> omni.start_profile(stages=[0, 2]) NzA[%s] Skipping start_profile for stage-%s: profiler config not setrz#[%s] Sent start_profile to stage-%sz1[%s] Failed to send start_profile to stage-%s: %s) listrrr8r(r/rkrrrPROFILER_STARTr.r0)rnr)rr>rArArB start_profiles2 zOmniBase.start_profilec Cs&|dur ttt|j}ggd}|D]}|t|jkr||s+td|j|q|j|}t|drtd|j|| }t |t r| dpO| d}| dpY| d }t d |jd |d ||r~t d |jd |d t||rt d |jd |dt||rt |tr|d|n t |tr|d||rt |tr|d |qt |tr|d |qtd |jd |dqtd |jd |dt|qtd|j||dtjiqtd |jdt|ddt|d d|S)z} Synchronously stop profiling for specified stages and collect the file paths for traces and tables. N)tracestablesz@[%s] Skipping stop_profile for stage-%s: profiler config not set stop_profilez5[%s] Requesting profile data collection from stage-%stracer-tabler.rrz returned: z traces type: z tables type: z returned no table dataz returned non-dict data: zO[%s] Stage-%s does not support synchronous stop_profile. Falling back to async.rz ] Collected z trace(s) and z table(s))r*rrr8r(r/rkrrr/rdictrSrkeysrrrextendr0rr PROFILER_STOP)rnr) all_resultsrr@ stage_datar-r.rArArBr/-s`     """    $(zOmniBase.stop_profilecCst|dr |dSdS)z1Close all stage processes and clean up resources._weak_finalizerN)rr8rrArArBr*vs  zOmniBase.closerc Cst|tr |ddkrdddSz t|d}Wnttfy6}zdd|dWYd}~Sd}~ww|j|}|durIdd |dS|j||\}}t d |j |d ||d S) zProcess incoming handshake message and generate response. Args: msg: Decoded message from client Returns: Response dictionary with ok status and either endpoints or error r handshakeFzinvalid handshake payload)okerrorrzinvalid stage_id: Nzunknown stage_id: z%[%s] Handshake received from stage-%sT)r:rr) rr2rSr TypeError ValueErrorrhrirr/rkr)rnrrr> endpointsrrrArArB_process_handshake_message{s,   z#OmniBase._process_handshake_messagec st}|jtjzGjsK|d}tfdd|D}|s%q t j j }t j |}j|jrW|jdSW|jdS|jw)zAMain loop for handshake server - polls for messages and responds.ic3s(|]\}}|jko|tjkVqdSrD)rerPOLLIN)rsockeventrrArBrrz6OmniBase._run_handshake_server_loop..N)rPollerregisterrer@rgis_setpollanymsgspecmsgpackdecoderecvencoder?send unregister)rnpollerevents has_messagerresponserArrB_run_handshake_server_loops      z#OmniBase._run_handshake_server_loopcCs|jdus |jdur dS|jr|jdurdStd|jt|jd}t|_t |j|t j ddd|_ tj |jddd|_|jdS) aStart the ZMQ handshake server. The handshake server allows distributed stages to discover their queue endpoints by querying the orchestrator with their stage_id. Skips starting if the server is already running or ZMQ is not initialized. NF)rrportTi)rlingerzzmq-handshake-server)targetdaemonname)rfrbrcrdrr threadingEventrgrrREPreThreadrSstart)rnendpointrArArBrs zOmniBase.start_handshake_serverc Cst|j}tt|t|jh}|s|Sttdt|}t d|j d|d|d| |j sMt|krMt d| |j sMt|ks>> omni = Omni(model="Qwen/Qwen2.5-Omni-7B") >>> outputs = omni.generate(prompts="Hello, world!", sampling_params_list=[SamplingParams()]) >>> print(outputs) rZr[rGNc sFtj|fi|t|t|j|j|j|j|j |j |j |j |_ dSrD)superroweakreffinalizerCr8r\r]r_rbrgrerfr8rm __class__rArBros z Omni.__init__promptssampling_params_list py_generatorTcCdSrDrArnrprqrrrArArBgenerate+z Omni.generateF)rrcCrsrDrArtrArArBru4rv)rruse_tqdmrw.c Cs|dur|j}n"t|ts*g}|jD]}|j}||jkr"||q||q|}z|r4|||WSt||||}|WStyW} z t d| | | d} ~ ww)a*Generate outputs for the given prompts. Orchestrates the multi-stage pipeline based on YAML configuration. Each stage will use OmniLLM or OmniDiffusion based on stage_type. Args: prompts: Input prompt(s) for generation. sampling_params_list: Optional list of per-stage parameters. py_generator: Whether the returned result(s) are wrapped in a generator instead of a list. use_tqdm: Whether to use tqdm progress bar Returns: List of OmniRequestOutput objects, one for each input prompt. Each output contains the stage_id, final_output_type, and the request_output from the final stage. Raises: ValueError: If sampling_params_list is None or has incorrect length. N+[Orchestrator] Failed to run generation: %s) rrrror_run_generation_with_generatorr*_run_generationr.r/ exceptionr*) rnrprqrrrwper_stage_paramsdefault_stage_spdefault_sp_typeoutputsr>rArArBru=s*      c cs`|||}z#z|EdHWnty#}ztd||d}~wwW|dS|w)zIRun generation through all stages in the pipeline and return a generator.Nrx)rzr.r/r{r*)rnrprqgenr>rArArBryss  z#Omni._run_generation_with_generatorc3cshtd|jd|durtdt|t|jkr+tdt|jdt|tt|j|D]"\}\}}|jdkr@t nt }t ||sUtd|d |d |j q3t |t s`t |tsd|g}nt|}t|j} d d tt|D} d dt| |D} i} t} i}| D]\}}t |tr|dd}nd}t||j|j}|||<qt| |j| |}| }|rt|r|nt}||dd}td|jdt|d|jdpt|jd<| D])\}}|d}|||d}|jd|t| |<td|jd|dqd}|r3t|r|nt}|t|dddddddddd}t|gdg| d}d}t|}td|jd |d!| ||krd"}t|jD]\}}|}|durpq`d}|d#}d$|vrtd|jd%|d&|d'|d$q`|d(d)krtd*q`t |d+d,d-} t!|j"|pd.t|j"|<zl|d/}!|!dur#|j#|||!j$7<|%|j|| |&|||!|j'|r#|j(d0pd1}"t)|j*}#|#|"}$|j|}%|%d2krd3nd4}&|j+dkr|j,|j+}'nd}'d5|d6|&d7|$dd8|'d9d: |_-Wn%t.yI}(zt/d|jd;|d<|d'|(WYd}(~(nd}(~(wwtd|jd=|d>|d?|0| t1|d@d"r td|jdA|dB|z|||kr|2||| || Wn%t.y}(zt/d|jdC|dD|d'|(WYd}(~(nd}(~(wwt3||j'| dE})z-t4| dFr| j5nt | tr| rt4| ddFr| dj5nd"}*|*r|6|)||Wn%t.y }(zt/d|jdG|dD|d'|(WYd}(~(nd}(~(ww|)V|d}+|+||kr|j|+},z#|7|||,8|j| |g}-Wdn 1s:wYWn&t.yf}(zt/d|jdH|dD|+d'|(WYd}(~(q`d}(~(ww||+}.t |t |+f}/|j9|/}0d"}1|0rt:|0||+||-|.| ||j|+j|dI }1|1st;d|jdJ|dK|+dLtd|jdM|dK|+||+d7<q`|d7}|r|j||}2|2d2krd3ndN|_<|=dtd|jdA|dO|dP|dQ q`|stdR||ksYtd|jdS|r |>z|?WdSt.y3}(zt/d|jdT|(WYd}(~(dSd}(~(ww)Uz2Run generation through all stages in the pipeline.rz] generate() calledNz9sampling_params_list is required for pipelined generationz Expected z sampling params, got rz'Expected sampling parameters with type z in stage z, got cSsg|] }|dtqS)r)uuiduuid4)rrrArArBrsz(Omni._run_generation..cSsi|]\}}||qSrArA)rridprArArB sz(Omni._run_generation.. modalitieszAdding requests)descz ] Seeding z requests into stage-0r) request_id engine_inputssampling_paramsz] Enqueued request z to stage-0zProcessed promptsTzest. speed input: z.2fz unit/s, output: z unit/s)totalr dynamic_ncolspostfixrrz+] Entering scheduling loop: total_requests=z , stages=Frr;z] Stage z error on request z: rrrengine_outputsengine_outputs_shm)obj_keyshm_keygmetricselapsedgư>rimgtokzest. speed stage- z/s: z, avg e2e_lat: z.1fmsz&] Failed to process metrics for stage z, req rz completed request z; forwarding or finalizingrz ] Request z finalized at stage-z*] Finalize request handling error for req z at stage )rrrequest_outputfinishedz)] Failed to record audio metrics for req z&] Process engine inputs error for req ) connectorr next_stage_idreq_id next_inputsroriginal_promptnext_stage_queue_submit_fnrz] Failed to send request z to stage-zZ via connector. Configure a connector for this edge or inspect connector logs for details.z] Forwarded request reqz fully completed (r)g{Gzt?z] All requests completedz] Failed to build/log summary: )@r/rrr=rr8rziprr r rrorrr*rritemsr2rSrrr#rr2r stage_first_tsrrr;r_loadr stage_last_tsaccumulated_gen_time_msstage_gen_time_msaccumulate_diffusion_metricson_stage_metricsr format_dictsumstage_total_tokens e2e_count e2e_total_msrr.r{set_engine_outputsr1on_finalize_requestr&rrrecord_audio_generated_framesstage_postprocess_timerprocess_engine_inputsrr RuntimeErrorunitupdater*build_and_log_summary)3rnrprqrwrr@spExpectedSPTyperequest_promptsr request_idsrequest_id_to_prompt _req_start_ts_wall_start_tsfinal_stage_id_to_promptrpromptprompt_modalitiesfinal_stage_id_for_e2erit tqdm_funcrsp0taskpbarremaining_by_stagecompleted_requeststotal_requests made_progressrrr_mr total_outout_spdmodalityravg_latr>output_to_yieldrr next_stagersp_next connector_keyrsent_via_connector final_modrArArBrzs                                 $&zOmni._run_generationcCrc)N OrchestratorrArrArArBrrdz Omni._namerD)T)rerfrgrhrrror r!rr"rrr&rur*rrr ryrzrir __classcell__rArArnrBrjs~        6      rj)NNNN)QrmultiprocessingrrOrYrrrlcollections.abcrrrconcurrent.futuresrrtypingrrr rUmsgspec.msgpackrHr omegaconfr tqdm.autor vllmr vllm.loggerr vllm.utils.network_utilsr vllm.v1.utilsr%vllm_omni.distributed.omni_connectorsrr-vllm_omni.distributed.omni_connectors.adapterr:vllm_omni.distributed.omni_connectors.utils.initializationr%vllm_omni.distributed.ray_utils.utilsrrr vllm_omni.entrypoints.omni_stager!vllm_omni.entrypoints.stage_utilsrrrrvllm_omni.entrypoints.utilsrrrvllm_omni.entrypoints.zmq_utilsrvllm_omni.inputs.datar r!r"vllm_omni.metricsr#r$2vllm_omni.model_executor.model_loader.weight_utilsr%vllm_omni.outputsr&rer/rZSocketr\rCrFrrXrYrjrArArArBsh              .}