From d6871db6a3ca23ea5a2bf47e30899157f02e5fc8 Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Mon, 9 Mar 2026 11:33:38 +0800 Subject: [PATCH] [Feat]: Offline inference supports async_chunk (#1415) Signed-off-by: Sy03 <1370724210@qq.com> Signed-off-by: linyueqian Co-authored-by: Yueqian Lin <70319226+linyueqian@users.noreply.github.com> Co-authored-by: linyueqian --- .../offline_inference/qwen3_omni/README.md | 45 ++ .../qwen3_omni/end2end_async_chunk.py | 614 ++++++++++++++++++ .../run_multiple_prompts_async_chunk.sh | 23 + .../run_single_prompt_async_chunk.sh | 26 + 4 files changed, 708 insertions(+) create mode 100644 examples/offline_inference/qwen3_omni/end2end_async_chunk.py create mode 100755 examples/offline_inference/qwen3_omni/run_multiple_prompts_async_chunk.sh create mode 100755 examples/offline_inference/qwen3_omni/run_single_prompt_async_chunk.sh diff --git a/examples/offline_inference/qwen3_omni/README.md b/examples/offline_inference/qwen3_omni/README.md index b7eee8f74..b3e859253 100644 --- a/examples/offline_inference/qwen3_omni/README.md +++ b/examples/offline_inference/qwen3_omni/README.md @@ -64,6 +64,51 @@ If media file paths are not provided, the script will use default assets. Suppor - `multi_audios`: Multiple audio inputs - `mixed_modalities`: Combination of video, image, and audio inputs +### Async-chunk (offline) + +For true stage-level concurrency -- where downstream stages (Talker, Code2Wav) +start **before** the upstream stage (Thinker) finishes -- use the async_chunk +example. This requires: + +1. A stage config YAML with ``async_chunk: true`` (e.g. + ``qwen3_omni_moe_async_chunk.yaml``). +2. Hardware that matches the config (e.g. 2x H100 for the default 3-stage + config). + +The async_chunk example uses ``AsyncOmni`` instead of the synchronous ``Omni`` +class, which enables the async orchestrator to receive stage-0 intermediate +outputs and trigger downstream stages early. Chunk data flows directly between +stage workers via the in-worker ``OmniChunkTransferAdapter`` / connector, +**not** through the orchestrator. + +#### Single prompt +```bash +cd examples/offline_inference/qwen3_omni +bash run_single_prompt_async_chunk.sh +``` + +#### Multiple prompts with concurrency control +```bash +bash run_multiple_prompts_async_chunk.sh --max-in-flight 4 +``` + +#### Text-only output (skip audio generation) +```bash +python end2end_async_chunk.py --query-type text --modalities text +``` + +#### Custom stage config +```bash +python end2end_async_chunk.py \ + --query-type use_audio \ + --stage-configs-path /path/to/your_async_chunk.yaml +``` + +> **Note**: The synchronous ``end2end.py`` (using ``Omni``) is still the +> recommended entry point for non-async-chunk workflows. Only use the +> async_chunk example when you need the stage-level concurrency semantics +> described in PR #962 / #1151. + ### FAQ If you encounter error about backend of librosa, try to install ffmpeg with command below. diff --git a/examples/offline_inference/qwen3_omni/end2end_async_chunk.py b/examples/offline_inference/qwen3_omni/end2end_async_chunk.py new file mode 100644 index 000000000..965595578 --- /dev/null +++ b/examples/offline_inference/qwen3_omni/end2end_async_chunk.py @@ -0,0 +1,614 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Offline inference with async_chunk enabled via AsyncOmni. + +This script uses AsyncOmni (the async orchestrator) to run offline inference +with async_chunk semantics: downstream stages (Talker, Code2Wav) start +*before* upstream stages finish, consuming chunks as they arrive via +the in-worker OmniChunkTransferAdapter / connector. + +Compared to the synchronous ``end2end.py`` (which uses ``Omni``), this +entry point achieves true stage-level concurrency -- stage-1/2 are +actively processing while stage-0 is still generating. + +Usage +----- + python end2end_async_chunk.py --query-type use_audio \ + --stage-configs-path + +See ``--help`` for all options. +""" + +import asyncio +import logging +import os +import time +import uuid +from typing import NamedTuple + +import numpy as np +import soundfile as sf +import torch + +os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" + +import librosa +from PIL import Image +from vllm import SamplingParams +from vllm.assets.audio import AudioAsset +from vllm.assets.image import ImageAsset +from vllm.assets.video import VideoAsset, video_to_ndarrays +from vllm.multimodal.image import convert_image_mode +from vllm.utils.argparse_utils import FlexibleArgumentParser + +from vllm_omni.entrypoints.async_omni import AsyncOmni + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Query builders (reuse the patterns from end2end.py) +# --------------------------------------------------------------------------- + +default_system = ( + "You are Qwen, a virtual human developed by the Qwen Team, Alibaba " + "Group, capable of perceiving auditory and visual inputs, as well as " + "generating text and speech." +) + + +class QueryResult(NamedTuple): + inputs: dict + limit_mm_per_prompt: dict[str, int] + + +def get_text_query(question: str = None) -> QueryResult: + if question is None: + question = "Explain the system architecture for a scalable audio generation pipeline. Answer in 15 words." + prompt = ( + f"<|im_start|>system\n{default_system}<|im_end|>\n" + "<|im_start|>user\n" + f"{question}<|im_end|>\n" + "<|im_start|>assistant\n" + ) + return QueryResult(inputs={"prompt": prompt}, limit_mm_per_prompt={}) + + +def get_audio_query( + question: str = None, + audio_path: str | None = None, + sampling_rate: int = 16000, +) -> QueryResult: + if question is None: + question = "What is the content of this audio?" + prompt = ( + f"<|im_start|>system\n{default_system}<|im_end|>\n" + "<|im_start|>user\n<|audio_start|><|audio_pad|><|audio_end|>" + f"{question}<|im_end|>\n" + "<|im_start|>assistant\n" + ) + if audio_path: + if not os.path.exists(audio_path): + raise FileNotFoundError(f"Audio file not found: {audio_path}") + audio_signal, sr = librosa.load(audio_path, sr=sampling_rate) + audio_data = (audio_signal.astype(np.float32), sr) + else: + audio_data = AudioAsset("mary_had_lamb").audio_and_sample_rate + return QueryResult( + inputs={ + "prompt": prompt, + "multi_modal_data": {"audio": audio_data}, + }, + limit_mm_per_prompt={"audio": 1}, + ) + + +def get_image_query(question: str = None, image_path: str | None = None) -> QueryResult: + if question is None: + question = "What is the content of this image?" + prompt = ( + f"<|im_start|>system\n{default_system}<|im_end|>\n" + "<|im_start|>user\n<|vision_start|><|image_pad|><|vision_end|>" + f"{question}<|im_end|>\n" + "<|im_start|>assistant\n" + ) + if image_path: + if not os.path.exists(image_path): + raise FileNotFoundError(f"Image file not found: {image_path}") + pil_image = Image.open(image_path) + image_data = convert_image_mode(pil_image, "RGB") + else: + image_data = convert_image_mode(ImageAsset("cherry_blossom").pil_image, "RGB") + return QueryResult( + inputs={ + "prompt": prompt, + "multi_modal_data": {"image": image_data}, + }, + limit_mm_per_prompt={"image": 1}, + ) + + +def get_video_query( + question: str = None, + video_path: str | None = None, + num_frames: int = 16, +) -> QueryResult: + if question is None: + question = "Why is this video funny?" + prompt = ( + f"<|im_start|>system\n{default_system}<|im_end|>\n" + "<|im_start|>user\n<|vision_start|><|video_pad|><|vision_end|>" + f"{question}<|im_end|>\n" + "<|im_start|>assistant\n" + ) + if video_path: + if not os.path.exists(video_path): + raise FileNotFoundError(f"Video file not found: {video_path}") + video_frames = video_to_ndarrays(video_path, num_frames=num_frames) + else: + video_frames = VideoAsset(name="baby_reading", num_frames=num_frames).np_ndarrays + return QueryResult( + inputs={ + "prompt": prompt, + "multi_modal_data": {"video": video_frames}, + }, + limit_mm_per_prompt={"video": 1}, + ) + + +query_map = { + "text": get_text_query, + "use_audio": get_audio_query, + "use_image": get_image_query, + "use_video": get_video_query, +} + +# --------------------------------------------------------------------------- +# Core async routine +# --------------------------------------------------------------------------- + + +def clone_prompt_for_request(template: dict) -> dict: + """Shallow-clone prompt dict so concurrent requests own independent containers.""" + cloned = dict(template) + for key in ("multi_modal_data", "mm_processor_kwargs", "additional_information"): + value = template.get(key) + if isinstance(value, dict): + cloned[key] = dict(value) + elif isinstance(value, list): + cloned[key] = list(value) + return cloned + + +def _default_async_chunk_stage_configs_path() -> str | None: + """Best-effort default stage config for running Qwen3-Omni with async_chunk. + + When this example is executed from within the repository, we resolve the + default YAML path relative to this file. When installed elsewhere, the + file may not exist and callers should pass --stage-configs-path explicitly. + """ + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")) + candidate = os.path.join( + repo_root, + "vllm_omni", + "model_executor", + "stage_configs", + "qwen3_omni_moe_async_chunk.yaml", + ) + return candidate if os.path.exists(candidate) else None + + +async def run_single_request( + async_omni: AsyncOmni, + prompt: dict, + request_id: str, + sampling_params_list: list[SamplingParams] | None, + output_dir: str, + output_modalities: list[str] | None = None, + stream_audio_to_disk: bool = False, +) -> dict: + """Run one request through AsyncOmni and collect outputs. + + Returns a dict with timing information and saved file paths. + """ + t_start = time.perf_counter() + text_parts: list[str] = [] + audio_chunks: list[torch.Tensor] = [] + audio_sr: int | None = None + first_audio_ts: float | None = None + audio_list_consumed: int = 0 + audio_last_tensor: torch.Tensor | None = None + stage_0_first_output_ts: float | None = None + + samplerate = 24000 + wav_file = os.path.join(output_dir, f"output_{request_id}.wav") + sf_writer: sf.SoundFile | None = None + audio_samples_written: int = 0 + + try: + async for omni_output in async_omni.generate( + prompt=prompt, + request_id=request_id, + sampling_params_list=sampling_params_list, + output_modalities=output_modalities, + ): + if not isinstance(omni_output.request_output, list): + outputs_list = [omni_output.request_output] + else: + outputs_list = omni_output.request_output + + for output in outputs_list: + if omni_output.final_output_type == "text": + if stage_0_first_output_ts is None: + stage_0_first_output_ts = time.perf_counter() + text_output = output.outputs[0].text + if output.finished: + text_parts.append(text_output) + elif omni_output.final_output_type == "audio": + mm_out = output.outputs[0].multimodal_output + if mm_out and "audio" in mm_out: + if first_audio_ts is None: + first_audio_ts = time.perf_counter() + if audio_sr is None and "sr" in mm_out: + sr_val = mm_out["sr"] + audio_sr = sr_val.item() if hasattr(sr_val, "item") else int(sr_val) + samplerate = audio_sr + audio_data = mm_out["audio"] + if isinstance(audio_data, list): + new_chunks = audio_data[audio_list_consumed:] + audio_list_consumed = len(audio_data) + elif isinstance(audio_data, torch.Tensor): + new_chunks = [audio_data] + audio_last_tensor = audio_data + else: + new_chunks = [] + + if stream_audio_to_disk and new_chunks: + if sf_writer is None: + sf_writer = sf.SoundFile( + wav_file, + mode="w", + samplerate=samplerate, + channels=1, + subtype="FLOAT", + ) + for chunk in new_chunks: + chunk_np = chunk.float().detach().cpu().numpy().flatten() + sf_writer.write(chunk_np) + audio_samples_written += len(chunk_np) + else: + audio_chunks.extend(new_chunks) + finally: + if sf_writer is not None: + sf_writer.close() + + t_end = time.perf_counter() + result = { + "request_id": request_id, + "e2e_latency_s": t_end - t_start, + "saved_files": [], + } + + # Save text output + if text_parts: + text_file = os.path.join(output_dir, f"{request_id}.txt") + with open(text_file, "w", encoding="utf-8") as f: + f.write("\n".join(text_parts)) + result["saved_files"].append(text_file) + print( + f"[Request {request_id}] Text saved to {text_file} " + f"(stage-0 first output at {stage_0_first_output_ts - t_start:.3f}s)" + ) + + # Save audio output + if stream_audio_to_disk and audio_samples_written > 0: + result["saved_files"].append(wav_file) + result["audio_duration_s"] = audio_samples_written / samplerate + result["num_audio_chunks"] = audio_list_consumed + ttfa = (first_audio_ts - t_start) if first_audio_ts else None + result["time_to_first_audio_s"] = ttfa + ttfa_str = f"{ttfa:.3f}s" if ttfa is not None else "N/A" + print( + f"[Request {request_id}] Audio streamed to {wav_file} " + f"(duration={result['audio_duration_s']:.2f}s, " + f"TTFA={ttfa_str}, " + f"e2e={result['e2e_latency_s']:.3f}s)" + ) + elif audio_chunks or audio_last_tensor is not None: + if audio_chunks: + if len(audio_chunks) > 1: + audio_tensor = torch.cat(audio_chunks, dim=-1) + else: + audio_tensor = audio_chunks[0] + else: + audio_tensor = audio_last_tensor + audio_numpy = audio_tensor.float().detach().cpu().numpy() + if audio_numpy.ndim > 1: + audio_numpy = audio_numpy.flatten() + sf.write(wav_file, audio_numpy, samplerate=samplerate, format="WAV") + result["saved_files"].append(wav_file) + result["audio_duration_s"] = len(audio_numpy) / samplerate + result["num_audio_chunks"] = len(audio_chunks) + ttfa = (first_audio_ts - t_start) if first_audio_ts else None + result["time_to_first_audio_s"] = ttfa + ttfa_str = f"{ttfa:.3f}s" if ttfa is not None else "N/A" + print( + f"[Request {request_id}] Audio saved to {wav_file} " + f"({len(audio_chunks)} chunks, " + f"duration={result['audio_duration_s']:.2f}s, " + f"TTFA={ttfa_str}, " + f"e2e={result['e2e_latency_s']:.3f}s)" + ) + + return result + + +async def run_all(args): + """Main async entry: build prompts, create AsyncOmni, run requests.""" + # Build query + query_func = query_map[args.query_type] + if args.query_type == "use_video": + query_result = query_func( + video_path=getattr(args, "video_path", None), + num_frames=getattr(args, "num_frames", 16), + ) + elif args.query_type == "use_image": + query_result = query_func(image_path=getattr(args, "image_path", None)) + elif args.query_type == "use_audio": + query_result = query_func( + audio_path=getattr(args, "audio_path", None), + sampling_rate=getattr(args, "sampling_rate", 16000), + ) + else: + query_result = query_func() + + # Build prompt list + if args.txt_prompts is not None: + assert args.query_type == "text", "txt-prompts is only supported for text query type" + with open(args.txt_prompts, encoding="utf-8") as f: + lines = [ln.strip() for ln in f if ln.strip()] + prompts = [get_text_query(ln).inputs for ln in lines] + print(f"[Info] Loaded {len(prompts)} prompts from {args.txt_prompts}") + else: + prompts = [clone_prompt_for_request(query_result.inputs) for _ in range(args.num_prompts)] + + # Inject output modalities if specified + output_modalities = None + if args.modalities is not None: + output_modalities = args.modalities.split(",") + for prompt in prompts: + prompt["modalities"] = output_modalities + + # Create AsyncOmni + print(f"[Info] Creating AsyncOmni with stage_configs_path={args.stage_configs_path}") + async_omni = None + try: + async_omni = AsyncOmni( + model=args.model, + stage_configs_path=args.stage_configs_path, + log_stats=args.log_stats, + stage_init_timeout=args.stage_init_timeout, + ) + + # Use default sampling params from stage config (they are pre-configured + # in the YAML for each stage). + sampling_params_list = None + + output_dir = args.output_dir + os.makedirs(output_dir, exist_ok=True) + + # Run requests with concurrency control + semaphore = asyncio.Semaphore(args.max_in_flight) + request_timeout = getattr(args, "request_timeout_s", None) + stream_audio = getattr(args, "stream_audio_to_disk", False) + + async def _run_one(idx: int, prompt: dict): + async with semaphore: + request_id = f"req_{idx}_{uuid.uuid4().hex[:8]}" + coro = run_single_request( + async_omni=async_omni, + prompt=prompt, + request_id=request_id, + sampling_params_list=sampling_params_list, + output_dir=output_dir, + output_modalities=output_modalities, + stream_audio_to_disk=stream_audio, + ) + if request_timeout and request_timeout > 0: + return await asyncio.wait_for(coro, timeout=request_timeout) + return await coro + + wall_start = time.perf_counter() + tasks = [_run_one(i, p) for i, p in enumerate(prompts)] + all_results = await asyncio.gather(*tasks, return_exceptions=True) + wall_end = time.perf_counter() + + # Print summary + print("\n" + "=" * 60) + print("Summary") + print("=" * 60) + success_count = 0 + total_audio_dur = 0.0 + for r in all_results: + if isinstance(r, Exception): + print(f" [ERROR] {type(r).__name__}: {r}") + else: + success_count += 1 + total_audio_dur += r.get("audio_duration_s", 0.0) + print(f" [{r['request_id']}] e2e={r['e2e_latency_s']:.3f}s files={r['saved_files']}") + wall_time = wall_end - wall_start + print(f"\nTotal: {success_count}/{len(prompts)} succeeded") + print(f"Wall time: {wall_time:.3f}s") + if total_audio_dur > 0: + print(f"Total audio duration: {total_audio_dur:.2f}s") + print(f"Real-time factor: {total_audio_dur / wall_time:.2f}x") + print("=" * 60) + finally: + if async_omni is not None: + async_omni.shutdown() + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def parse_args(): + parser = FlexibleArgumentParser( + description=( + "Offline inference with async_chunk enabled via AsyncOmni. " + "Downstream stages start before upstream stages finish, " + "achieving true stage-level concurrency." + ) + ) + parser.add_argument( + "--model", + type=str, + default="Qwen/Qwen3-Omni-30B-A3B-Instruct", + help="Model name or path.", + ) + parser.add_argument( + "--query-type", + "-q", + type=str, + default="use_audio", + choices=query_map.keys(), + help="Query type.", + ) + parser.add_argument( + "--stage-configs-path", + type=str, + default=_default_async_chunk_stage_configs_path(), + help=( + "Path to an async_chunk stage config YAML. " + "If not set, uses the model's default config " + "(make sure it has async_chunk: true)." + ), + ) + parser.add_argument( + "--log-stats", + action="store_true", + default=False, + help="Enable writing detailed statistics.", + ) + parser.add_argument( + "--stage-init-timeout", + type=int, + default=300, + help="Timeout for initializing a single stage (seconds).", + ) + parser.add_argument( + "--output-dir", + type=str, + default="output_audio_async_chunk", + help="Directory to save output files.", + ) + parser.add_argument( + "--num-prompts", + type=int, + default=1, + help="Number of prompts to generate (duplicated from query).", + ) + parser.add_argument( + "--txt-prompts", + type=str, + default=None, + help="Path to a .txt file with one prompt per line.", + ) + parser.add_argument( + "--max-in-flight", + type=int, + default=1, + help="Maximum concurrent requests (default: 1).", + ) + parser.add_argument( + "--request-timeout-s", + type=float, + default=None, + help=( + "Per-request timeout in seconds. When set, a request that " + "exceeds this duration is cancelled and reported as an error. " + "Default: None (no timeout)." + ), + ) + parser.add_argument( + "--batch-timeout-s", + type=float, + default=None, + help=( + "Global timeout for the entire batch in seconds. When set, " + "the whole run_all() is cancelled if it exceeds this duration. " + "Default: None (no global timeout)." + ), + ) + parser.add_argument( + "--stream-audio-to-disk", + action="store_true", + default=False, + help=( + "Write audio chunks to WAV incrementally instead of " + "accumulating in memory. Useful for very long audio or " + "high --max-in-flight to reduce memory footprint." + ), + ) + parser.add_argument( + "--modalities", + type=str, + default=None, + help="Comma-separated output modalities filter (e.g. 'text', 'audio', 'text,audio').", + ) + parser.add_argument( + "--audio-path", + "-a", + type=str, + default=None, + help="Path to local audio file.", + ) + parser.add_argument( + "--image-path", + "-i", + type=str, + default=None, + help="Path to local image file.", + ) + parser.add_argument( + "--video-path", + "-v", + type=str, + default=None, + help="Path to local video file.", + ) + parser.add_argument( + "--num-frames", + type=int, + default=16, + help="Number of frames to extract from video.", + ) + parser.add_argument( + "--sampling-rate", + type=int, + default=16000, + help="Sampling rate for audio loading.", + ) + return parser.parse_args() + + +if __name__ == "__main__": + args = parse_args() + + async def _main(): + batch_timeout = getattr(args, "batch_timeout_s", None) + if batch_timeout and batch_timeout > 0: + await asyncio.wait_for(run_all(args), timeout=batch_timeout) + else: + await run_all(args) + + try: + asyncio.run(_main()) + except asyncio.TimeoutError: + print( + f"\n[TIMEOUT] Batch exceeded --batch-timeout-s=" + f"{args.batch_timeout_s}s. AsyncOmni shutdown was handled " + f"by finally block." + ) + except KeyboardInterrupt: + print("\nInterrupted by user. AsyncOmni shutdown was handled by finally block.") diff --git a/examples/offline_inference/qwen3_omni/run_multiple_prompts_async_chunk.sh b/examples/offline_inference/qwen3_omni/run_multiple_prompts_async_chunk.sh new file mode 100755 index 000000000..809054867 --- /dev/null +++ b/examples/offline_inference/qwen3_omni/run_multiple_prompts_async_chunk.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# Run multiple Qwen3-Omni requests with async_chunk enabled. +# +# Uses AsyncOmni with --max-in-flight to control request-level +# concurrency (each request still gets true stage-level concurrency +# via async_chunk). +# +# Usage: +# bash run_multiple_prompts_async_chunk.sh +# bash run_multiple_prompts_async_chunk.sh --max-in-flight 4 + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)" + +python "${SCRIPT_DIR}/end2end_async_chunk.py" \ + --query-type text \ + --txt-prompts "${SCRIPT_DIR}/text_prompts_10.txt" \ + --stage-configs-path "${REPO_ROOT}/vllm_omni/model_executor/stage_configs/qwen3_omni_moe_async_chunk.yaml" \ + --output-dir output_audio_async_chunk \ + --max-in-flight 2 \ + "$@" diff --git a/examples/offline_inference/qwen3_omni/run_single_prompt_async_chunk.sh b/examples/offline_inference/qwen3_omni/run_single_prompt_async_chunk.sh new file mode 100755 index 000000000..918c7ee4f --- /dev/null +++ b/examples/offline_inference/qwen3_omni/run_single_prompt_async_chunk.sh @@ -0,0 +1,26 @@ +#!/bin/bash +# Run a single Qwen3-Omni request with async_chunk enabled. +# +# This uses AsyncOmni (async orchestrator) so that downstream stages +# (Talker, Code2Wav) start *before* stage-0 (Thinker) finishes, +# achieving true stage-level concurrency via chunk-level streaming. +# +# Prerequisites: +# - An async_chunk stage config YAML (e.g. qwen3_omni_moe_async_chunk.yaml) +# - Hardware matching the config (e.g. 2x H100 for the default 3-stage config) +# +# Usage: +# bash run_single_prompt_async_chunk.sh +# bash run_single_prompt_async_chunk.sh --query-type text --modalities text +# bash run_single_prompt_async_chunk.sh --stage-configs-path /path/to/custom.yaml + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)" + +python "${SCRIPT_DIR}/end2end_async_chunk.py" \ + --query-type use_audio \ + --stage-configs-path "${REPO_ROOT}/vllm_omni/model_executor/stage_configs/qwen3_omni_moe_async_chunk.yaml" \ + --output-dir output_audio_async_chunk \ + "$@"