From 959b3ca8c8a775879aa20e734167ee6947b82b5c Mon Sep 17 00:00:00 2001 From: Junhong Liu Date: Wed, 4 Feb 2026 15:27:30 +0800 Subject: [PATCH] [Bugfix] E2E metric incorrect qwen3-omni with async chunk feature (#1018) Signed-off-by: Junhong Liu <98734602+LJH-LBJ@users.noreply.github.com> Signed-off-by: Junhong Liu Co-authored-by: Hongsheng Liu --- vllm_omni/entrypoints/async_omni.py | 58 +++++++++++++---------------- 1 file changed, 26 insertions(+), 32 deletions(-) diff --git a/vllm_omni/entrypoints/async_omni.py b/vllm_omni/entrypoints/async_omni.py index 144c73f3a..1545e17a1 100644 --- a/vllm_omni/entrypoints/async_omni.py +++ b/vllm_omni/entrypoints/async_omni.py @@ -345,14 +345,22 @@ class AsyncOmni(OmniBase): ): yield output - logger.debug(f"[{self._name}] All requests completed") - - # Summarize and print stats + logger.debug(f"[{self._name}] Request {request_id} finalized at stage-{final_stage_id_for_e2e}") try: + # Finalize E2E metrics if not already done + if str(request_id) not in metrics.e2e_done: + metrics.on_finalize_request( + final_stage_id_for_e2e, + request_id, + _req_start_ts.get(request_id, _wall_start_ts), + ) + + logger.debug(f"[{self._name}] All requests completed") + # Summarize and print stats summary = metrics.build_and_log_summary(final_stage_id_for_e2e) logger.info("[Summary] %s", pformat(summary, sort_dicts=False)) except Exception as e: - logger.exception(f"[{self._name}] Failed to build/log summary: {e}") + logger.exception(f"[{self._name}] Request {request_id} Failed to finalized/build/log summary: {e}") finally: self.request_states.pop(request_id, None) except (asyncio.CancelledError, GeneratorExit): @@ -382,9 +390,11 @@ class AsyncOmni(OmniBase): except asyncio.QueueEmpty: await asyncio.sleep(0.001) continue - engine_outputs, finished, output_to_yield = self._process_single_result( - result, stage, stage_id, metrics, req_start_ts, wall_start_ts, final_stage_id_for_e2e + result, + stage, + stage_id, + metrics, ) if submit_flag and stage_id == 0: submit_flag = False @@ -421,9 +431,11 @@ class AsyncOmni(OmniBase): while not finished: result = await req_state.queue.get() assert stage_id == req_state.stage_id - req_id = result.get("request_id") engine_outputs, finished, output_to_yield = self._process_single_result( - result, stage, stage_id, metrics, req_start_ts, wall_start_ts, final_stage_id_for_e2e + result, + stage, + stage_id, + metrics, ) if output_to_yield: yield output_to_yield @@ -432,7 +444,7 @@ class AsyncOmni(OmniBase): stage.set_engine_outputs(engine_outputs) # Forward to next stage if there is one next_stage_id = stage_id + 1 - if next_stage_id <= final_stage_id_for_e2e and finished: + if next_stage_id <= final_stage_id_for_e2e: next_stage: OmniStage = self.stage_list[next_stage_id] next_inputs = next_stage.process_engine_inputs(self.stage_list, prompt) sp_next: SamplingParams = sampling_params_list[next_stage_id] @@ -447,7 +459,7 @@ class AsyncOmni(OmniBase): connector=connector, stage_id=stage_id, next_stage_id=next_stage_id, - req_id=req_id, + req_id=request_id, next_inputs=next_inputs, sampling_params=sp_next, original_prompt=prompt, @@ -461,14 +473,14 @@ class AsyncOmni(OmniBase): # because continuing would cause the request to be silently dropped # and the orchestrator to hang waiting for completion. error_msg = ( - f"[{self._name}] Failed to send request {req_id} to stage-{next_stage_id} via connector. " + f"[{self._name}] Failed to send request {request_id} to stage-{next_stage_id} via connector. " "Configure a connector for this edge or inspect connector logs for details." ) logger.error(error_msg) raise RuntimeError(error_msg) - logger.debug(f"[{self._name}] Forwarded request {req_id} to stage-{next_stage_id}") + logger.debug(f"[{self._name}] Forwarded request {request_id} to stage-{next_stage_id}") else: - logger.debug(f"[{self._name}] Request {req_id} fully completed") + logger.debug(f"[{self._name}] Request {request_id} fully completed") def _process_single_result( self, @@ -476,9 +488,6 @@ class AsyncOmni(OmniBase): stage: OmniStage, stage_id: int, metrics: OrchestratorMetrics, - req_start_ts: dict[int, float], - wall_start_ts: float, - final_stage_id_for_e2e: int, ) -> tuple[Any, bool, OmniRequestOutput | None]: """ Process a single result dictionary from a stage. @@ -517,23 +526,8 @@ class AsyncOmni(OmniBase): ) output_to_yield = None + if getattr(stage, "final_output", False): - logger.debug(f"[{self._name}] Request {req_id} finalized at stage-{stage_id}") - - # Finalize request metrics if this is the E2E final stage and it's finished - try: - rid_key = str(req_id) - if stage_id == final_stage_id_for_e2e and rid_key not in metrics.e2e_done and finished: - metrics.on_finalize_request( - stage_id, - req_id, - req_start_ts.get(req_id, wall_start_ts), - ) - except Exception as e: - logger.exception( - f"[{self._name}] Finalize request handling error for req {req_id} at stage {stage_id}: {e}", - ) - # Construct output to yield images = [] if stage.final_output_type == "image":