diff --git a/docs/api/README.md b/docs/api/README.md index cf40a9577..4fa85cdc6 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -4,9 +4,8 @@ Main entry points for vLLM-Omni inference and serving. - -- [vllm_omni.entrypoints.async_omni_diffusion.AsyncOmniDiffusion][] - [vllm_omni.entrypoints.async_omni.AsyncOmni][] +- [vllm_omni.entrypoints.async_omni_diffusion.AsyncOmniDiffusion][] - [vllm_omni.entrypoints.async_omni_llm.AsyncOmniLLM][] - [vllm_omni.entrypoints.chat_utils.OmniAsyncMultiModalContentParser][] - [vllm_omni.entrypoints.chat_utils.OmniAsyncMultiModalItemTracker][] @@ -14,11 +13,14 @@ Main entry points for vLLM-Omni inference and serving. - [vllm_omni.entrypoints.cli.serve.OmniServeCommand][] - [vllm_omni.entrypoints.client_request_state.ClientRequestState][] - [vllm_omni.entrypoints.log_utils.OrchestratorMetrics][] +- [vllm_omni.entrypoints.log_utils.StageRequestMetrics][] +- [vllm_omni.entrypoints.log_utils.StageStats][] - [vllm_omni.entrypoints.omni.Omni][] +- [vllm_omni.entrypoints.omni.OmniBase][] - [vllm_omni.entrypoints.omni_diffusion.OmniDiffusion][] - [vllm_omni.entrypoints.omni_llm.OmniLLM][] -- [vllm_omni.entrypoints.omni_llm.OmniStageLLM][] - [vllm_omni.entrypoints.omni_stage.OmniStage][] +- [vllm_omni.entrypoints.stage_utils.OmniStageTaskType][] ## Inputs @@ -34,6 +36,7 @@ Input data structures for multi-modal inputs. Engine classes for offline and online inference. +- [vllm_omni.diffusion.diffusion_engine.BackgroundResources][] - [vllm_omni.diffusion.diffusion_engine.DiffusionEngine][] - [vllm_omni.engine.AdditionalInformationEntry][] - [vllm_omni.engine.AdditionalInformationPayload][] @@ -59,6 +62,7 @@ Core scheduling and caching components. Model execution components. +- [vllm_omni.model_executor.custom_process_mixin.CustomProcessMixin][] - [vllm_omni.model_executor.models.output_templates.OmniOutput][] - [vllm_omni.model_executor.models.qwen2_5_omni.qwen2_5_omni.Qwen2_5OmniForConditionalGeneration][] - [vllm_omni.model_executor.models.qwen2_5_omni.qwen2_5_omni_talker.Qwen2_5OmniTalkerForConditionalGeneration][] @@ -108,6 +112,7 @@ Worker classes and model runners for distributed inference. - [vllm_omni.worker.gpu_generation_model_runner.GPUGenerationModelRunner][] - [vllm_omni.worker.gpu_generation_worker.GPUGenerationWorker][] - [vllm_omni.worker.gpu_model_runner.OmniGPUModelRunner][] +- [vllm_omni.worker.npu.npu_ar_model_runner.ExecuteModelState][] - [vllm_omni.worker.npu.npu_ar_model_runner.NPUARModelRunner][] - [vllm_omni.worker.npu.npu_ar_worker.NPUARWorker][] - [vllm_omni.worker.npu.npu_generation_model_runner.NPUGenerationModelRunner][] diff --git a/docs/user_guide/examples/offline_inference/image_to_image.md b/docs/user_guide/examples/offline_inference/image_to_image.md index ab44de854..78c7f11fc 100644 --- a/docs/user_guide/examples/offline_inference/image_to_image.md +++ b/docs/user_guide/examples/offline_inference/image_to_image.md @@ -19,11 +19,11 @@ Then run: ```bash python image_edit.py \ - --image qwen_bear.png \ + --image qwen-bear.png \ --prompt "Let this mascot dance under the moon, surrounded by floating stars and poetic bubbles such as 'Be Kind'" \ --output output_image_edit.png \ --num_inference_steps 50 \ - --cfg_scale 4.0 \ + --cfg_scale 4.0 ``` ### Multiple Image Editing (Qwen-Image-Edit-2509) diff --git a/docs/user_guide/examples/offline_inference/qwen2_5_omni.md b/docs/user_guide/examples/offline_inference/qwen2_5_omni.md index b9709a241..7a12e0e14 100644 --- a/docs/user_guide/examples/offline_inference/qwen2_5_omni.md +++ b/docs/user_guide/examples/offline_inference/qwen2_5_omni.md @@ -70,6 +70,13 @@ If media file paths are not provided, the script will use default assets. Suppor - `use_audio_in_video`: Extract audio from video - `text`: Text-only query +### FAQ + +If you encounter error about backend of librosa, try to install ffmpeg with command below. +``` +sudo apt update +sudo apt install ffmpeg +``` ## Example materials diff --git a/docs/user_guide/examples/offline_inference/qwen3_omni.md b/docs/user_guide/examples/offline_inference/qwen3_omni.md index 05869d17f..a3abf89a5 100644 --- a/docs/user_guide/examples/offline_inference/qwen3_omni.md +++ b/docs/user_guide/examples/offline_inference/qwen3_omni.md @@ -49,6 +49,12 @@ python end2end.py --query-type use_image --image-path /path/to/image.jpg # Use local audio file python end2end.py --query-type use_audio --audio-path /path/to/audio.wav + +# Combine multiple local media files +python end2end.py --query-type mixed_modalities \ + --video-path /path/to/video.mp4 \ + --image-path /path/to/image.jpg \ + --audio-path /path/to/audio.wav ``` If media file paths are not provided, the script will use default assets. Supported query types: @@ -56,7 +62,16 @@ If media file paths are not provided, the script will use default assets. Suppor - `use_image`: Image input - `use_audio`: Audio input - `text`: Text-only query +- `multi_audios`: Multiple audio inputs +- `mixed_modalities`: Combination of video, image, and audio inputs +### FAQ + +If you encounter error about backend of librosa, try to install ffmpeg with command below. +``` +sudo apt update +sudo apt install ffmpeg +``` ## Example materials diff --git a/docs/user_guide/examples/offline_inference/text_to_image.md b/docs/user_guide/examples/offline_inference/text_to_image.md index cf5cdf860..5d0ffc226 100644 --- a/docs/user_guide/examples/offline_inference/text_to_image.md +++ b/docs/user_guide/examples/offline_inference/text_to_image.md @@ -19,7 +19,7 @@ if __name__ == "__main__": omni = Omni(model="Qwen/Qwen-Image") prompt = "a cup of coffee on the table" outputs = omni.generate(prompt) - images = outputs[0].request_output[0]["images"] + images = list(outputs)[0].request_output[0].images images[0].save("coffee.png") ``` @@ -37,7 +37,7 @@ if __name__ == "__main__": ] outputs = omni.generate(prompts) for i, output in enumerate(outputs): - image = output.request_output[0]["images"][0].save(f"{i}.jpg") + image = output.request_output[0].images[0].save(f"{i}.jpg") ``` ## Local CLI Usage diff --git a/docs/user_guide/examples/online_serving/qwen2_5_omni.md b/docs/user_guide/examples/online_serving/qwen2_5_omni.md index bc7fc412c..c59306d1f 100644 --- a/docs/user_guide/examples/online_serving/qwen2_5_omni.md +++ b/docs/user_guide/examples/online_serving/qwen2_5_omni.md @@ -37,10 +37,11 @@ The Python client supports the following command-line arguments: - `--query-type` (or `-q`): Query type (default: `mixed_modalities`). Options: `mixed_modalities`, `use_audio_in_video`, `multi_audios`, `text` - `--video-path` (or `-v`): Path to local video file or URL. If not provided and query-type uses video, uses default video URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs. Example: `--video-path /path/to/video.mp4` or `--video-path https://example.com/video.mp4` -- `--image-path` (or `-i`): Path to local image file or URL. If not provided and query-type uses image, uses default image URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common image formats: JPEG, PNG, GIF. Example: `--image-path /path/to/image.jpg` or `--image-path https://example.com/image.png` +- `--image-path` (or `-i`): Path to local image file or URL. If not provided and query-type uses image, uses default image URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common image formats: JPEG, PNG, GIF, WebP. Example: `--image-path /path/to/image.jpg` or `--image-path https://example.com/image.png` - `--audio-path` (or `-a`): Path to local audio file or URL. If not provided and query-type uses audio, uses default audio URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common audio formats: MP3, WAV, OGG, FLAC, M4A. Example: `--audio-path /path/to/audio.wav` or `--audio-path https://example.com/audio.mp3` - `--prompt` (or `-p`): Custom text prompt/question. If not provided, uses default prompt for the selected query type. Example: `--prompt "What are the main activities shown in this video?"` + For example, to use mixed modalities with all local files: ```bash @@ -59,7 +60,6 @@ bash run_curl_multimodal_generation.sh mixed_modalities ``` ## Modality control - You can control output modalities to specify which types of output the model should generate. This is useful when you only need text output and want to skip audio generation stages for better performance. ### Supported modalities @@ -137,6 +137,13 @@ response = client.chat.completions.create( # Response contains two choices: one with text, one with audio print(response.choices[0].message.content) # Text response print(response.choices[1].message.audio) # Audio response + +## Streaming Output +If you want to enable streaming output, please set the argument as below. The final output will be obtained just after generated by corresponding stage. Now we only support text streaming output. Other modalities can output normally. +```bash +python openai_chat_completion_client_for_multimodal_generation.py \ + --query-type mixed_modalities \ + --stream ``` ## Run Local Web UI Demo diff --git a/docs/user_guide/examples/online_serving/qwen3_omni.md b/docs/user_guide/examples/online_serving/qwen3_omni.md index f6920d7ff..5ea4818a6 100644 --- a/docs/user_guide/examples/online_serving/qwen3_omni.md +++ b/docs/user_guide/examples/online_serving/qwen3_omni.md @@ -38,10 +38,11 @@ The Python client supports the following command-line arguments: - `--query-type` (or `-q`): Query type (default: `use_video`). Options: `text`, `use_audio`, `use_image`, `use_video` - `--model` (or `-m`): Model name/path (default: `Qwen/Qwen3-Omni-30B-A3B-Instruct`) - `--video-path` (or `-v`): Path to local video file or URL. If not provided and query-type is `use_video`, uses default video URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs. Example: `--video-path /path/to/video.mp4` or `--video-path https://example.com/video.mp4` -- `--image-path` (or `-i`): Path to local image file or URL. If not provided and query-type is `use_image`, uses default image URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common image formats: JPEG, PNG, GIF. Example: `--image-path /path/to/image.jpg` or `--image-path https://example.com/image.png`. +- `--image-path` (or `-i`): Path to local image file or URL. If not provided and query-type is `use_image`, uses default image URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common image formats: JPEG, PNG, GIF, WebP. Example: `--image-path /path/to/image.jpg` or `--image-path https://example.com/image.png` - `--audio-path` (or `-a`): Path to local audio file or URL. If not provided and query-type is `use_audio`, uses default audio URL. Supports local file paths (automatically encoded to base64) or HTTP/HTTPS URLs and common audio formats: MP3, WAV, OGG, FLAC, M4A. Example: `--audio-path /path/to/audio.wav` or `--audio-path https://example.com/audio.mp3` - `--prompt` (or `-p`): Custom text prompt/question. If not provided, uses default prompt for the selected query type. Example: `--prompt "What are the main activities shown in this video?"` + For example, to use a local video file with custom prompt: ```bash @@ -67,7 +68,6 @@ sudo apt install ffmpeg ``` ## Modality control - You can control output modalities to specify which types of output the model should generate. This is useful when you only need text output and want to skip audio generation stages for better performance. ### Supported modalities @@ -145,6 +145,13 @@ response = client.chat.completions.create( # Response contains two choices: one with text, one with audio print(response.choices[0].message.content) # Text response print(response.choices[1].message.audio) # Audio response + +## Streaming Output +If you want to enable streaming output, please set the argument as below. The final output will be obtained just after generated by corresponding stage. Now we only support text streaming output. Other modalities can output normally. +```bash +python openai_chat_completion_client_for_multimodal_generation.py \ + --query-type use_image \ + --stream ``` ## Run Local Web UI Demo diff --git a/examples/offline_inference/image_to_image/image_edit.py b/examples/offline_inference/image_to_image/image_edit.py index 20921a8d0..bf4348e4c 100644 --- a/examples/offline_inference/image_to_image/image_edit.py +++ b/examples/offline_inference/image_to_image/image_edit.py @@ -63,6 +63,7 @@ from PIL import Image from vllm_omni.diffusion.data import DiffusionParallelConfig, logger from vllm_omni.entrypoints.omni import Omni +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import detect_device_type, is_npu @@ -340,21 +341,23 @@ def main(): # Print profiling results print(f"Total generation time: {generation_time:.4f} seconds ({generation_time * 1000:.2f} ms)") + # omni.generate() returns Generator[OmniRequestOutput, None, None], convert to list + outputs = list(outputs) if not outputs: raise ValueError("No output generated from omni.generate()") logger.info("Outputs: %s", outputs) # Extract images from OmniRequestOutput - # omni.generate() returns list[OmniRequestOutput], extract images from request_output[0]['images'] + # omni.generate() returns Generator[OmniRequestOutput, None, None], extract images from request_output[0].images first_output = outputs[0] if not hasattr(first_output, "request_output") or not first_output.request_output: raise ValueError("No request_output found in OmniRequestOutput") req_out = first_output.request_output[0] - if not isinstance(req_out, dict) or "images" not in req_out: + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out["images"] + images = req_out.images if not images: raise ValueError("No images found in request_output") diff --git a/examples/offline_inference/image_to_image/image_to_image.md b/examples/offline_inference/image_to_image/image_to_image.md index 68327b500..87bb8b68d 100644 --- a/examples/offline_inference/image_to_image/image_to_image.md +++ b/examples/offline_inference/image_to_image/image_to_image.md @@ -16,11 +16,11 @@ Then run: ```bash python image_edit.py \ - --image qwen_bear.png \ + --image qwen-bear.png \ --prompt "Let this mascot dance under the moon, surrounded by floating stars and poetic bubbles such as 'Be Kind'" \ --output output_image_edit.png \ --num_inference_steps 50 \ - --cfg_scale 4.0 \ + --cfg_scale 4.0 ``` ### Multiple Image Editing (Qwen-Image-Edit-2509) diff --git a/examples/offline_inference/image_to_video/image_to_video.py b/examples/offline_inference/image_to_video/image_to_video.py index 3b1b2e84c..dac36393d 100644 --- a/examples/offline_inference/image_to_video/image_to_video.py +++ b/examples/offline_inference/image_to_video/image_to_video.py @@ -26,6 +26,7 @@ import PIL.Image import torch from vllm_omni.entrypoints.omni import Omni +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import detect_device_type, is_npu @@ -102,6 +103,7 @@ def main(): flow_shift=args.flow_shift, ) + # omni.generate() returns Generator[OmniRequestOutput, None, None] frames = omni.generate( args.prompt, negative_prompt=args.negative_prompt, @@ -115,6 +117,32 @@ def main(): num_frames=args.num_frames, ) + # Extract video frames from OmniRequestOutput + frames = list(frames) + if isinstance(frames, list) and len(frames) > 0: + first_item = frames[0] + + # Check if it's an OmniRequestOutput + if hasattr(first_item, "final_output_type"): + if first_item.final_output_type != "image": + raise ValueError( + f"Unexpected output type '{first_item.final_output_type}', expected 'image' for video generation." + ) + + # Pipeline mode: extract from nested request_output + if hasattr(first_item, "is_pipeline_output") and first_item.is_pipeline_output: + if isinstance(first_item.request_output, list) and len(first_item.request_output) > 0: + inner_output = first_item.request_output[0] + if isinstance(inner_output, OmniRequestOutput) and hasattr(inner_output, "images"): + frames = inner_output.images[0] if inner_output.images else None + if frames is None: + raise ValueError("No video frames found in output.") + # Diffusion mode: use direct images field + elif hasattr(first_item, "images") and first_item.images: + frames = first_item.images + else: + raise ValueError("No video frames found in OmniRequestOutput.") + output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/examples/offline_inference/qwen2_5_omni/end2end.py b/examples/offline_inference/qwen2_5_omni/end2end.py index 6c11e6a7b..39ebc017f 100644 --- a/examples/offline_inference/qwen2_5_omni/end2end.py +++ b/examples/offline_inference/qwen2_5_omni/end2end.py @@ -376,12 +376,12 @@ def main(args): for i, prompt in enumerate(prompts): prompt["modalities"] = output_modalities - omni_outputs = omni_llm.generate(prompts, sampling_params_list) + omni_generator = omni_llm.generate(prompts, sampling_params_list) # Determine output directory: prefer --output-dir; fallback to --output-wav output_dir = args.output_dir if getattr(args, "output_dir", None) else args.output_wav os.makedirs(output_dir, exist_ok=True) - for stage_outputs in omni_outputs: + for stage_outputs in omni_generator: if stage_outputs.final_output_type == "text": for output in stage_outputs.request_output: request_id = output.request_id diff --git a/examples/offline_inference/qwen3_omni/end2end.py b/examples/offline_inference/qwen3_omni/end2end.py index a979452e4..3e5ef8106 100644 --- a/examples/offline_inference/qwen3_omni/end2end.py +++ b/examples/offline_inference/qwen3_omni/end2end.py @@ -319,12 +319,12 @@ def main(args): for i, prompt in enumerate(prompts): prompt["modalities"] = output_modalities - omni_outputs = omni_llm.generate(prompts, sampling_params_list) + omni_generator = omni_llm.generate(prompts, sampling_params_list) # Determine output directory: prefer --output-dir; fallback to --output-wav output_dir = args.output_dir if getattr(args, "output_dir", None) else args.output_wav os.makedirs(output_dir, exist_ok=True) - for stage_outputs in omni_outputs: + for stage_outputs in omni_generator: if stage_outputs.final_output_type == "text": for output in stage_outputs.request_output: request_id = output.request_id diff --git a/examples/offline_inference/text_to_image/README.md b/examples/offline_inference/text_to_image/README.md index 2403c4139..ece7bbaa9 100644 --- a/examples/offline_inference/text_to_image/README.md +++ b/examples/offline_inference/text_to_image/README.md @@ -16,7 +16,7 @@ if __name__ == "__main__": omni = Omni(model="Qwen/Qwen-Image") prompt = "a cup of coffee on the table" outputs = omni.generate(prompt) - images = outputs[0].request_output[0]["images"] + images = list(outputs)[0].request_output[0].images images[0].save("coffee.png") ``` @@ -34,7 +34,7 @@ if __name__ == "__main__": ] outputs = omni.generate(prompts) for i, output in enumerate(outputs): - image = output.request_output[0]["images"][0].save(f"{i}.jpg") + image = output.request_output[0].images[0].save(f"{i}.jpg") ``` ## Local CLI Usage diff --git a/examples/offline_inference/text_to_image/gradio_demo.py b/examples/offline_inference/text_to_image/gradio_demo.py index 42659e058..8d4ff2b20 100644 --- a/examples/offline_inference/text_to_image/gradio_demo.py +++ b/examples/offline_inference/text_to_image/gradio_demo.py @@ -5,6 +5,7 @@ import gradio as gr import torch from vllm_omni.entrypoints.omni import Omni +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import detect_device_type, is_npu ASPECT_RATIOS: dict[str, tuple[int, int]] = { @@ -98,7 +99,7 @@ def build_demo(args: argparse.Namespace) -> gr.Blocks: if num_images not in {1, 2, 3, 4}: raise gr.Error("Number of images must be 1, 2, 3, or 4.") generator = torch.Generator(device=device).manual_seed(seed) - images = omni.generate( + outputs = omni.generate( prompt.strip(), height=height, width=width, @@ -107,7 +108,20 @@ def build_demo(args: argparse.Namespace) -> gr.Blocks: num_inference_steps=num_steps, num_outputs_per_prompt=num_images, ) - return [img for img in images[:num_images]] + images_outputs = [] + for output in outputs: + req_out = output.request_output[0] + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): + raise ValueError("Invalid request_output structure or missing 'images' key") + images = req_out.images + if not images: + raise ValueError("No images found in request_output") + # Extend the list with individual images (not append the entire list) + images_outputs.extend(images) + if len(images_outputs) >= num_images: + break + # Return only the requested number of images + return images_outputs[:num_images] with gr.Blocks( title="vLLM-Omni Web Serving Demo", diff --git a/examples/offline_inference/text_to_image/text_to_image.py b/examples/offline_inference/text_to_image/text_to_image.py index 170eb69ad..ded976f3d 100644 --- a/examples/offline_inference/text_to_image/text_to_image.py +++ b/examples/offline_inference/text_to_image/text_to_image.py @@ -9,6 +9,7 @@ import torch from vllm_omni.diffusion.data import DiffusionParallelConfig, logger from vllm_omni.entrypoints.omni import Omni +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import detect_device_type, is_npu @@ -144,8 +145,9 @@ def main(): print(f"Total generation time: {generation_time:.4f} seconds ({generation_time * 1000:.2f} ms)") # Extract images from OmniRequestOutput - # omni.generate() returns list[OmniRequestOutput], extract images from the first output - if not outputs or len(outputs) == 0: + # omni.generate() returns Generator[OmniRequestOutput, None, None], convert to list + outputs = list(outputs) + if not outputs: raise ValueError("No output generated from omni.generate()") logger.info(f"Outputs: {outputs}") @@ -155,10 +157,10 @@ def main(): raise ValueError("No request_output found in OmniRequestOutput") req_out = first_output.request_output[0] - if not isinstance(req_out, dict) or "images" not in req_out: + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out["images"] + images = req_out.images if not images: raise ValueError("No images found in request_output") @@ -175,8 +177,6 @@ def main(): img.save(save_path) print(f"Saved generated image to {save_path}") - omni.close() - if __name__ == "__main__": main() diff --git a/examples/offline_inference/text_to_video/text_to_video.py b/examples/offline_inference/text_to_video/text_to_video.py index c1aa60d6e..97797f736 100644 --- a/examples/offline_inference/text_to_video/text_to_video.py +++ b/examples/offline_inference/text_to_video/text_to_video.py @@ -8,6 +8,7 @@ import numpy as np import torch from vllm_omni.entrypoints.omni import Omni +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import detect_device_type, is_npu @@ -53,6 +54,7 @@ def main(): flow_shift=args.flow_shift, ) + # omni.generate() returns Generator[OmniRequestOutput, None, None] frames = omni.generate( args.prompt, negative_prompt=args.negative_prompt, @@ -65,6 +67,32 @@ def main(): num_frames=args.num_frames, ) + # Extract video frames from OmniRequestOutput + frames = list(frames) + if isinstance(frames, list) and len(frames) > 0: + first_item = frames[0] + + # Check if it's an OmniRequestOutput + if hasattr(first_item, "final_output_type"): + if first_item.final_output_type != "image": + raise ValueError( + f"Unexpected output type '{first_item.final_output_type}', expected 'image' for video generation." + ) + + # Pipeline mode: extract from nested request_output + if hasattr(first_item, "is_pipeline_output") and first_item.is_pipeline_output: + if isinstance(first_item.request_output, list) and len(first_item.request_output) > 0: + inner_output = first_item.request_output[0] + if isinstance(inner_output, OmniRequestOutput) and hasattr(inner_output, "images"): + frames = inner_output.images[0] if inner_output.images else None + if frames is None: + raise ValueError("No video frames found in output.") + # Diffusion mode: use direct images field + elif hasattr(first_item, "images") and first_item.images: + frames = first_item.images + else: + raise ValueError("No video frames found in OmniRequestOutput.") + output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) try: diff --git a/examples/online_serving/qwen2_5_omni/README.md b/examples/online_serving/qwen2_5_omni/README.md index 56e7b5b36..5fcf69df3 100644 --- a/examples/online_serving/qwen2_5_omni/README.md +++ b/examples/online_serving/qwen2_5_omni/README.md @@ -57,13 +57,92 @@ bash run_curl_multimodal_generation.sh mixed_modalities ``` ## Modality control -If you want to control output modalities, e.g. only output text, you can run the command below: +You can control output modalities to specify which types of output the model should generate. This is useful when you only need text output and want to skip audio generation stages for better performance. + +### Supported modalities + +| Modalities | Output | +|------------|--------| +| `["text"]` | Text only | +| `["audio"]` | Text + Audio | +| `["text", "audio"]` | Text + Audio | +| Not specified | Text + Audio (default) | + +### Using curl + +#### Text only + +```bash +curl http://localhost:8091/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "Qwen/Qwen2.5-Omni-7B", + "messages": [{"role": "user", "content": "Describe vLLM in brief."}], + "modalities": ["text"] + }' +``` + +#### Text + Audio + +```bash +curl http://localhost:8091/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "Qwen/Qwen2.5-Omni-7B", + "messages": [{"role": "user", "content": "Describe vLLM in brief."}], + "modalities": ["audio"] + }' +``` + +### Using Python client + ```bash python openai_chat_completion_client_for_multimodal_generation.py \ --query-type mixed_modalities \ --modalities text ``` +### Using OpenAI Python SDK + +#### Text only + +```python +from openai import OpenAI + +client = OpenAI(base_url="http://localhost:8091/v1", api_key="EMPTY") + +response = client.chat.completions.create( + model="Qwen/Qwen2.5-Omni-7B", + messages=[{"role": "user", "content": "Describe vLLM in brief."}], + modalities=["text"] +) +print(response.choices[0].message.content) +``` + +#### Text + Audio + +```python +from openai import OpenAI + +client = OpenAI(base_url="http://localhost:8091/v1", api_key="EMPTY") + +response = client.chat.completions.create( + model="Qwen/Qwen2.5-Omni-7B", + messages=[{"role": "user", "content": "Describe vLLM in brief."}], + modalities=["audio"] +) +# Response contains two choices: one with text, one with audio +print(response.choices[0].message.content) # Text response +print(response.choices[1].message.audio) # Audio response + +## Streaming Output +If you want to enable streaming output, please set the argument as below. The final output will be obtained just after generated by corresponding stage. Now we only support text streaming output. Other modalities can output normally. +```bash +python openai_chat_completion_client_for_multimodal_generation.py \ + --query-type mixed_modalities \ + --stream +``` + ## Run Local Web UI Demo This Web UI demo allows users to interact with the model through a web browser. diff --git a/examples/online_serving/qwen2_5_omni/gradio_demo.py b/examples/online_serving/qwen2_5_omni/gradio_demo.py index 82775ac95..a6d3a67e8 100644 --- a/examples/online_serving/qwen2_5_omni/gradio_demo.py +++ b/examples/online_serving/qwen2_5_omni/gradio_demo.py @@ -23,16 +23,18 @@ SUPPORTED_MODELS: dict[str, dict[str, Any]] = { "top_p": 1.0, "top_k": -1, "max_tokens": 2048, + "seed": SEED, "detokenize": True, "repetition_penalty": 1.1, }, "talker": { - "temperature": 0.0, - "top_p": 1.0, - "top_k": -1, + "temperature": 0.9, + "top_p": 0.8, + "top_k": 40, "max_tokens": 2048, + "seed": SEED, "detokenize": True, - "repetition_penalty": 1.1, + "repetition_penalty": 1.05, "stop_token_ids": [8294], }, "code2wav": { @@ -40,6 +42,7 @@ SUPPORTED_MODELS: dict[str, dict[str, Any]] = { "top_p": 1.0, "top_k": -1, "max_tokens": 2048, + "seed": SEED, "detokenize": True, "repetition_penalty": 1.1, }, @@ -241,10 +244,12 @@ def run_inference_api( video_file: str | None = None, use_audio_in_video: bool = False, output_modalities: str | None = None, + stream: bool = False, ): """Run inference using OpenAI API client with multimodal support.""" if not user_prompt.strip() and not audio_file and not image_file and not video_file: - return "Please provide at least a text prompt or multimodal input.", None + yield "Please provide at least a text prompt or multimodal input.", None + return try: # Build message content list @@ -324,7 +329,7 @@ def run_inference_api( extra_body["mm_processor_kwargs"] = mm_processor_kwargs # Parse output modalities - if output_modalities is not None: + if output_modalities and output_modalities.strip(): output_modalities_list = [m.strip() for m in output_modalities.split(",")] else: output_modalities_list = None @@ -335,29 +340,71 @@ def run_inference_api( model=model, modalities=output_modalities_list, extra_body=extra_body, + stream=stream, ) - # Extract outputs - text_outputs: list[str] = [] - audio_output = None + if not stream: + # Non-streaming mode: extract outputs and yield once + text_outputs: list[str] = [] + audio_output = None - for choice in chat_completion.choices: - if choice.message.content: - text_outputs.append(choice.message.content) - if choice.message.audio: - # Decode base64 audio - audio_data = base64.b64decode(choice.message.audio.data) - # Load audio from bytes - audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) - # Convert to mono if needed - if audio_np.ndim > 1: - audio_np = audio_np[:, 0] - audio_output = (int(sample_rate), audio_np.astype(np.float32)) + for choice in chat_completion.choices: + if choice.message.content: + text_outputs.append(choice.message.content) + if choice.message.audio: + # Decode base64 audio + audio_data = base64.b64decode(choice.message.audio.data) + # Load audio from bytes + audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) + # Convert to mono if needed + if audio_np.ndim > 1: + audio_np = audio_np[:, 0] + audio_output = (int(sample_rate), audio_np.astype(np.float32)) + + text_response = "\n\n".join(text_outputs) if text_outputs else "No text output." + yield text_response, audio_output + else: + # Streaming mode: yield incremental updates + text_content = "" + audio_output = None + + for chunk in chat_completion: + for choice in chunk.choices: + if hasattr(choice, "delta"): + content = getattr(choice.delta, "content", None) + else: + content = None + + # Handle audio modality + if getattr(chunk, "modality", None) == "audio" and content: + try: + # Decode base64 audio + audio_data = base64.b64decode(content) + # Load audio from bytes + audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) + # Convert to mono if needed + if audio_np.ndim > 1: + audio_np = audio_np[:, 0] + audio_output = (int(sample_rate), audio_np.astype(np.float32)) + # Yield current text and audio + yield text_content if text_content else "", audio_output + except Exception: # pylint: disable=broad-except + # If audio processing fails, just yield text + yield text_content if text_content else "", None + + # Handle text modality + elif getattr(chunk, "modality", None) == "text": + if content: + text_content += content + # Yield updated text content (keep existing audio if any) + yield text_content, audio_output + + # Final yield with accumulated text and last audio (if any) + yield text_content if text_content else "No text output.", audio_output - text_response = "\n\n".join(text_outputs) if text_outputs else "No text output." - return text_response, audio_output except Exception as exc: # pylint: disable=broad-except - return f"Inference failed: {exc}", None + error_msg = f"Inference failed: {exc}" + yield error_msg, None def build_interface( @@ -374,8 +421,10 @@ def build_interface( video_file: str | None, use_audio_in_video: bool, output_modalities: str | None = None, + stream: bool = False, ): - return run_inference_api( + # Always yield from the API function to maintain consistent generator behavior + yield from run_inference_api( client, model, sampling_params_dict, @@ -385,6 +434,7 @@ def build_interface( video_file, use_audio_in_video, output_modalities, + stream, ) css = """ @@ -455,8 +505,15 @@ def build_interface( with gr.Row(): output_modalities = gr.Textbox( label="Output Modalities", + value=None, placeholder="For example: text, image, video. Use comma to separate multiple modalities.", lines=1, + scale=2, + ) + stream_checkbox = gr.Checkbox( + label="Stream output", + value=False, + info="Enable streaming to see output as it's generated.", scale=1, ) @@ -474,7 +531,15 @@ def build_interface( generate_btn.click( fn=run_inference, - inputs=[input_box, audio_input, image_input, video_input, use_audio_in_video_checkbox, output_modalities], + inputs=[ + input_box, + audio_input, + image_input, + video_input, + use_audio_in_video_checkbox, + output_modalities, + stream_checkbox, + ], outputs=[text_output, audio_output], ) demo.queue() diff --git a/examples/online_serving/qwen2_5_omni/openai_chat_completion_client_for_multimodal_generation.py b/examples/online_serving/qwen2_5_omni/openai_chat_completion_client_for_multimodal_generation.py index e5ef42c8c..a25e97ebf 100644 --- a/examples/online_serving/qwen2_5_omni/openai_chat_completion_client_for_multimodal_generation.py +++ b/examples/online_serving/qwen2_5_omni/openai_chat_completion_client_for_multimodal_generation.py @@ -354,19 +354,43 @@ def run_multimodal_generation(args) -> None: model=model_name, modalities=output_modalities, extra_body=extra_body, + stream=args.stream, ) count = 0 - for choice in chat_completion.choices: - if choice.message.audio: - audio_data = base64.b64decode(choice.message.audio.data) - audio_file_path = f"audio_{count}.wav" - with open(audio_file_path, "wb") as f: - f.write(audio_data) - print(f"Audio saved to {audio_file_path}") - count += 1 - elif choice.message.content: - print("Chat completion output from text:", choice.message.content) + if not args.stream: + for choice in chat_completion.choices: + if choice.message.audio: + audio_data = base64.b64decode(choice.message.audio.data) + audio_file_path = f"audio_{count}.wav" + with open(audio_file_path, "wb") as f: + f.write(audio_data) + print(f"Audio saved to {audio_file_path}") + count += 1 + elif choice.message.content: + print("Chat completion output from text:", choice.message.content) + else: + printed_content = False + for chunk in chat_completion: + for choice in chunk.choices: + if hasattr(choice, "delta"): + content = getattr(choice.delta, "content", None) + else: + content = None + + if getattr(chunk, "modality", None) == "audio" and content: + audio_data = base64.b64decode(content) + audio_file_path = f"audio_{count}.wav" + with open(audio_file_path, "wb") as f: + f.write(audio_data) + print(f"\nAudio saved to {audio_file_path}") + count += 1 + + elif getattr(chunk, "modality", None) == "text": + if not printed_content: + printed_content = True + print("\ncontent:", end="", flush=True) + print(content, end="", flush=True) def parse_args(): @@ -413,6 +437,11 @@ def parse_args(): default=None, help="Output modalities to use for the prompts.", ) + parser.add_argument( + "--stream", + action="store_true", + help="Stream the response.", + ) return parser.parse_args() diff --git a/examples/online_serving/qwen3_omni/README.md b/examples/online_serving/qwen3_omni/README.md index aa141a505..0a57738b4 100644 --- a/examples/online_serving/qwen3_omni/README.md +++ b/examples/online_serving/qwen3_omni/README.md @@ -65,13 +65,92 @@ sudo apt install ffmpeg ``` ## Modality control -If you want to control output modalities, e.g. only output text, you can run the command below: +You can control output modalities to specify which types of output the model should generate. This is useful when you only need text output and want to skip audio generation stages for better performance. + +### Supported modalities + +| Modalities | Output | +|------------|--------| +| `["text"]` | Text only | +| `["audio"]` | Text + Audio | +| `["text", "audio"]` | Text + Audio | +| Not specified | Text + Audio (default) | + +### Using curl + +#### Text only + +```bash +curl http://localhost:8091/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct", + "messages": [{"role": "user", "content": "Describe vLLM in brief."}], + "modalities": ["text"] + }' +``` + +#### Text + Audio + +```bash +curl http://localhost:8091/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct", + "messages": [{"role": "user", "content": "Describe vLLM in brief."}], + "modalities": ["audio"] + }' +``` + +### Using Python client + ```bash python openai_chat_completion_client_for_multimodal_generation.py \ --query-type use_image \ --modalities text ``` +### Using OpenAI Python SDK + +#### Text only + +```python +from openai import OpenAI + +client = OpenAI(base_url="http://localhost:8091/v1", api_key="EMPTY") + +response = client.chat.completions.create( + model="Qwen/Qwen3-Omni-30B-A3B-Instruct", + messages=[{"role": "user", "content": "Describe vLLM in brief."}], + modalities=["text"] +) +print(response.choices[0].message.content) +``` + +#### Text + Audio + +```python +from openai import OpenAI + +client = OpenAI(base_url="http://localhost:8091/v1", api_key="EMPTY") + +response = client.chat.completions.create( + model="Qwen/Qwen3-Omni-30B-A3B-Instruct", + messages=[{"role": "user", "content": "Describe vLLM in brief."}], + modalities=["audio"] +) +# Response contains two choices: one with text, one with audio +print(response.choices[0].message.content) # Text response +print(response.choices[1].message.audio) # Audio response + +## Streaming Output +If you want to enable streaming output, please set the argument as below. The final output will be obtained just after generated by corresponding stage. Now we only support text streaming output. Other modalities can output normally. +```bash +python openai_chat_completion_client_for_multimodal_generation.py \ + --query-type use_image \ + --stream +``` + ## Run Local Web UI Demo This Web UI demo allows users to interact with the model through a web browser. diff --git a/examples/online_serving/qwen3_omni/gradio_demo.py b/examples/online_serving/qwen3_omni/gradio_demo.py index eee0c75f6..76c4e311a 100644 --- a/examples/online_serving/qwen3_omni/gradio_demo.py +++ b/examples/online_serving/qwen3_omni/gradio_demo.py @@ -244,10 +244,11 @@ def run_inference_api( video_file: str | None = None, use_audio_in_video: bool = False, output_modalities: str | None = None, + stream: bool = False, ): """Run inference using OpenAI API client with multimodal support.""" if not user_prompt.strip() and not audio_file and not image_file and not video_file: - return "Please provide at least a text prompt or multimodal input.", None + yield "Please provide at least a text prompt or multimodal input.", None try: # Build message content list @@ -327,7 +328,7 @@ def run_inference_api( extra_body["mm_processor_kwargs"] = mm_processor_kwargs # Parse output modalities - if output_modalities is not None: + if output_modalities and output_modalities.strip(): output_modalities_list = [m.strip() for m in output_modalities.split(",")] else: output_modalities_list = None @@ -338,29 +339,71 @@ def run_inference_api( model=model, modalities=output_modalities_list, extra_body=extra_body, + stream=stream, ) - # Extract outputs - text_outputs: list[str] = [] - audio_output = None + if not stream: + # Non-streaming mode: extract outputs and yield once + text_outputs: list[str] = [] + audio_output = None - for choice in chat_completion.choices: - if choice.message.content: - text_outputs.append(choice.message.content) - if choice.message.audio: - # Decode base64 audio - audio_data = base64.b64decode(choice.message.audio.data) - # Load audio from bytes - audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) - # Convert to mono if needed - if audio_np.ndim > 1: - audio_np = audio_np[:, 0] - audio_output = (int(sample_rate), audio_np.astype(np.float32)) + for choice in chat_completion.choices: + if choice.message.content: + text_outputs.append(choice.message.content) + if choice.message.audio: + # Decode base64 audio + audio_data = base64.b64decode(choice.message.audio.data) + # Load audio from bytes + audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) + # Convert to mono if needed + if audio_np.ndim > 1: + audio_np = audio_np[:, 0] + audio_output = (int(sample_rate), audio_np.astype(np.float32)) + + text_response = "\n\n".join(text_outputs) if text_outputs else "No text output." + yield text_response, audio_output + else: + # Streaming mode: yield incremental updates + text_content = "" + audio_output = None + + for chunk in chat_completion: + for choice in chunk.choices: + if hasattr(choice, "delta"): + content = getattr(choice.delta, "content", None) + else: + content = None + + # Handle audio modality + if getattr(chunk, "modality", None) == "audio" and content: + try: + # Decode base64 audio + audio_data = base64.b64decode(content) + # Load audio from bytes + audio_np, sample_rate = sf.read(io.BytesIO(audio_data)) + # Convert to mono if needed + if audio_np.ndim > 1: + audio_np = audio_np[:, 0] + audio_output = (int(sample_rate), audio_np.astype(np.float32)) + # Yield current text and audio + yield text_content if text_content else "", audio_output + except Exception: # pylint: disable=broad-except + # If audio processing fails, just yield text + yield text_content if text_content else "", None + + # Handle text modality + elif getattr(chunk, "modality", None) == "text": + if content: + text_content += content + # Yield updated text content (keep existing audio if any) + yield text_content, audio_output + + # Final yield with accumulated text and last audio (if any) + yield text_content if text_content else "No text output.", audio_output - text_response = "\n\n".join(text_outputs) if text_outputs else "No text output." - return text_response, audio_output except Exception as exc: # pylint: disable=broad-except - return f"Inference failed: {exc}", None + error_msg = f"Inference failed: {exc}" + yield error_msg, None def build_interface( @@ -377,8 +420,10 @@ def build_interface( video_file: str | None, use_audio_in_video: bool, output_modalities: str | None = None, + stream: bool = False, ): - return run_inference_api( + # Always yield from the API function to maintain consistent generator behavior + yield from run_inference_api( client, model, sampling_params_dict, @@ -388,6 +433,7 @@ def build_interface( video_file, use_audio_in_video, output_modalities, + stream, ) css = """ @@ -458,8 +504,15 @@ def build_interface( with gr.Row(): output_modalities = gr.Textbox( label="Output Modalities", + value=None, placeholder="For example: text, image, video. Use comma to separate multiple modalities.", lines=1, + scale=2, + ) + stream_checkbox = gr.Checkbox( + label="Stream output", + value=False, + info="Enable streaming to see output as it's generated.", scale=1, ) @@ -477,7 +530,15 @@ def build_interface( generate_btn.click( fn=run_inference, - inputs=[input_box, audio_input, image_input, video_input, use_audio_in_video_checkbox, output_modalities], + inputs=[ + input_box, + audio_input, + image_input, + video_input, + use_audio_in_video_checkbox, + output_modalities, + stream_checkbox, + ], outputs=[text_output, audio_output], ) demo.queue() diff --git a/examples/online_serving/qwen3_omni/openai_chat_completion_client_for_multimodal_generation.py b/examples/online_serving/qwen3_omni/openai_chat_completion_client_for_multimodal_generation.py index ca361ef15..0d5020635 100644 --- a/examples/online_serving/qwen3_omni/openai_chat_completion_client_for_multimodal_generation.py +++ b/examples/online_serving/qwen3_omni/openai_chat_completion_client_for_multimodal_generation.py @@ -395,19 +395,43 @@ def run_multimodal_generation(args) -> None: model=model_name, modalities=output_modalities, extra_body=extra_body, + stream=args.stream, ) count = 0 - for choice in chat_completion.choices: - if choice.message.audio: - audio_data = base64.b64decode(choice.message.audio.data) - audio_file_path = f"audio_{count}.wav" - with open(audio_file_path, "wb") as f: - f.write(audio_data) - print(f"Audio saved to {audio_file_path}") - count += 1 - elif choice.message.content: - print("Chat completion output from text:", choice.message.content) + if not args.stream: + for choice in chat_completion.choices: + if choice.message.audio: + audio_data = base64.b64decode(choice.message.audio.data) + audio_file_path = f"audio_{count}.wav" + with open(audio_file_path, "wb") as f: + f.write(audio_data) + print(f"Audio saved to {audio_file_path}") + count += 1 + elif choice.message.content: + print("Chat completion output from text:", choice.message.content) + else: + printed_content = False + for chunk in chat_completion: + for choice in chunk.choices: + if hasattr(choice, "delta"): + content = getattr(choice.delta, "content", None) + else: + content = None + + if getattr(chunk, "modality", None) == "audio" and content: + audio_data = base64.b64decode(content) + audio_file_path = f"audio_{count}.wav" + with open(audio_file_path, "wb") as f: + f.write(audio_data) + print(f"\nAudio saved to {audio_file_path}") + count += 1 + + elif getattr(chunk, "modality", None) == "text": + if not printed_content: + printed_content = True + print("\ncontent:", end="", flush=True) + print(content, end="", flush=True) def parse_args(): @@ -461,6 +485,11 @@ def parse_args(): default=None, help="Output modalities to use for the prompts.", ) + parser.add_argument( + "--stream", + action="store_true", + help="Stream the response.", + ) return parser.parse_args() diff --git a/mkdocs.yml b/mkdocs.yml index 4d4fbbb60..dea2f2566 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -113,7 +113,8 @@ plugins: - https://docs.aiohttp.org/en/stable/objects.inv - https://pillow.readthedocs.io/en/stable/objects.inv - https://numpy.org/doc/stable/objects.inv - - https://pytorch.org/docs/stable/objects.inv + # Temporarily disabled due to decompression errors + # - https://pytorch.org/docs/stable/objects.inv - https://psutil.readthedocs.io/en/stable/objects.inv markdown_extensions: diff --git a/tests/e2e/offline_inference/conftest.py b/tests/e2e/offline_inference/conftest.py index e53e0ae3b..f513e55cc 100644 --- a/tests/e2e/offline_inference/conftest.py +++ b/tests/e2e/offline_inference/conftest.py @@ -4,6 +4,7 @@ Pytest configuration and fixtures for vllm-omni tests. """ +from collections.abc import Generator from typing import Any import pytest @@ -189,7 +190,7 @@ class OmniRunner: self, prompts: list[dict[str, Any]], sampling_params_list: list[SamplingParams] | None = None, - ) -> list[Any]: + ) -> Generator[Any, None, None]: """ Generate outputs for the given prompts. @@ -205,7 +206,7 @@ class OmniRunner: if sampling_params_list is None: sampling_params_list = self.get_default_sampling_params_list() - return self.omni.generate(prompts, sampling_params_list) + yield from self.omni.generate(prompts, sampling_params_list) def generate_multimodal( self, @@ -217,7 +218,7 @@ class OmniRunner: videos: PromptVideoInput = None, mm_processor_kwargs: dict[str, Any] | None = None, modalities: list[str] | None = None, - ) -> list[Any]: + ) -> Generator[Any, None, None]: """ Convenience method to generate with multimodal inputs. @@ -242,7 +243,7 @@ class OmniRunner: mm_processor_kwargs=mm_processor_kwargs, modalities=modalities, ) - return self.generate(omni_inputs, sampling_params_list) + yield from self.generate(omni_inputs, sampling_params_list) def generate_audio( self, @@ -251,7 +252,7 @@ class OmniRunner: system_prompt: str | None = None, audios: PromptAudioInput = None, mm_processor_kwargs: dict[str, Any] | None = None, - ) -> list[Any]: + ) -> Generator[Any, None, None]: """ Convenience method to generate with multimodal inputs. Args: @@ -269,7 +270,7 @@ class OmniRunner: audios=audios, mm_processor_kwargs=mm_processor_kwargs, ) - return self.generate(omni_inputs, sampling_params_list) + yield from self.generate(omni_inputs, sampling_params_list) def generate_video( self, @@ -278,7 +279,7 @@ class OmniRunner: system_prompt: str | None = None, videos: PromptVideoInput = None, mm_processor_kwargs: dict[str, Any] | None = None, - ) -> list[Any]: + ) -> Generator[Any, None, None]: """ Convenience method to generate with multimodal inputs. Args: @@ -296,7 +297,7 @@ class OmniRunner: videos=videos, mm_processor_kwargs=mm_processor_kwargs, ) - return self.generate(omni_inputs, sampling_params_list) + yield from self.generate(omni_inputs, sampling_params_list) def generate_image( self, @@ -305,7 +306,7 @@ class OmniRunner: system_prompt: str | None = None, images: PromptImageInput = None, mm_processor_kwargs: dict[str, Any] | None = None, - ) -> list[Any]: + ) -> Generator[Any, None, None]: """ Convenience method to generate with multimodal inputs. Args: @@ -323,7 +324,7 @@ class OmniRunner: images=images, mm_processor_kwargs=mm_processor_kwargs, ) - return self.generate(omni_inputs, sampling_params_list) + yield from self.generate(omni_inputs, sampling_params_list) def __enter__(self): """Context manager entry.""" diff --git a/tests/e2e/offline_inference/test_cache_dit.py b/tests/e2e/offline_inference/test_cache_dit.py index 01b41342e..4bd8ba0d7 100644 --- a/tests/e2e/offline_inference/test_cache_dit.py +++ b/tests/e2e/offline_inference/test_cache_dit.py @@ -21,6 +21,7 @@ if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from vllm_omni import Omni +from vllm_omni.outputs import OmniRequestOutput os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "1" @@ -61,16 +62,16 @@ def test_cache_dit(model_name: str): num_outputs_per_prompt=1, # Single output for speed ) # Extract images from request_output[0]['images'] - first_output = outputs[0] + first_output = list(outputs)[0] assert first_output.final_output_type == "image" if not hasattr(first_output, "request_output") or not first_output.request_output: raise ValueError("No request_output found in OmniRequestOutput") req_out = first_output.request_output[0] - if not isinstance(req_out, dict) or "images" not in req_out: + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out["images"] + images = req_out.images # Verify generation succeeded assert images is not None diff --git a/tests/e2e/offline_inference/test_qwen2_5_omni.py b/tests/e2e/offline_inference/test_qwen2_5_omni.py index 63eea1ba2..9bcc48827 100644 --- a/tests/e2e/offline_inference/test_qwen2_5_omni.py +++ b/tests/e2e/offline_inference/test_qwen2_5_omni.py @@ -58,15 +58,15 @@ def test_mixed_modalities_to_audio(omni_runner: type[OmniRunner], test_config: t videos=video, ) - # Verify we got outputs from multiple stages - assert len(outputs) > 0 - # Find and verify text output (thinker stage) text_output = None + output_count = 0 for stage_output in outputs: if stage_output.final_output_type == "text": text_output = stage_output + output_count += 1 break + assert output_count > 0 assert text_output is not None assert len(text_output.request_output) > 0 @@ -76,10 +76,13 @@ def test_mixed_modalities_to_audio(omni_runner: type[OmniRunner], test_config: t # Find and verify audio output (code2wav stage) audio_output = None + output_count = 0 for stage_output in outputs: if stage_output.final_output_type == "audio": audio_output = stage_output + output_count += 1 break + assert output_count > 0 assert audio_output is not None assert len(audio_output.request_output) > 0 @@ -113,18 +116,16 @@ def test_mixed_modalities_to_text_only(omni_runner: type[OmniRunner], test_confi modalities=modalities, ) - # Verify we got outputs from multiple stages - assert len(outputs) > 0 - - for stage_output in outputs: - assert stage_output.final_output_type != "audio" - # Find and verify text output (thinker stage) text_output = None + output_count = 0 for stage_output in outputs: + assert stage_output.final_output_type != "audio" if stage_output.final_output_type == "text": text_output = stage_output + output_count += 1 break + assert output_count > 0 assert text_output is not None assert len(text_output.request_output) > 0 diff --git a/tests/e2e/offline_inference/test_qwen3_omni.py b/tests/e2e/offline_inference/test_qwen3_omni.py index b6f6a7590..9dc3d614a 100644 --- a/tests/e2e/offline_inference/test_qwen3_omni.py +++ b/tests/e2e/offline_inference/test_qwen3_omni.py @@ -37,16 +37,16 @@ def test_video_to_audio(omni_runner: type[OmniRunner], test_config) -> None: videos=video, ) - # Verify we got outputs from multiple stages - assert len(outputs) > 0 - # Find and verify text output (thinker stage) text_output = None + output_count = 0 for stage_output in outputs: if stage_output.final_output_type == "text": text_output = stage_output + output_count += 1 break + assert output_count > 0 assert text_output is not None assert len(text_output.request_output) > 0 text_content = text_output.request_output[0].outputs[0].text @@ -55,11 +55,14 @@ def test_video_to_audio(omni_runner: type[OmniRunner], test_config) -> None: # Find and verify audio output (code2wav stage) audio_output = None + output_count = 0 for stage_output in outputs: if stage_output.final_output_type == "audio": audio_output = stage_output + output_count += 1 break + assert output_count > 0 assert audio_output is not None assert len(audio_output.request_output) > 0 diff --git a/tests/e2e/offline_inference/test_sequence_parallel.py b/tests/e2e/offline_inference/test_sequence_parallel.py index 93a802394..9101f98f2 100644 --- a/tests/e2e/offline_inference/test_sequence_parallel.py +++ b/tests/e2e/offline_inference/test_sequence_parallel.py @@ -87,7 +87,7 @@ def test_sequence_parallel(model_name: str, ulysses_degree: int, ring_degree: in generator=torch.Generator(get_device_name()).manual_seed(seed), num_outputs_per_prompt=1, ) - baseline_images = outputs[0].request_output[0]["images"] + baseline_images = list(outputs)[0].request_output[0].images finally: baseline.close() @@ -113,7 +113,7 @@ def test_sequence_parallel(model_name: str, ulysses_degree: int, ring_degree: in generator=torch.Generator(get_device_name()).manual_seed(seed), num_outputs_per_prompt=1, ) - sp_images = outputs[0].request_output[0]["images"] + sp_images = list(outputs)[0].request_output[0].images finally: sp.close() diff --git a/tests/e2e/offline_inference/test_t2i_model.py b/tests/e2e/offline_inference/test_t2i_model.py index ffdd7a463..57164038d 100644 --- a/tests/e2e/offline_inference/test_t2i_model.py +++ b/tests/e2e/offline_inference/test_t2i_model.py @@ -5,6 +5,7 @@ from pathlib import Path import pytest import torch +from vllm_omni.outputs import OmniRequestOutput from vllm_omni.utils.platform_utils import is_npu, is_rocm # ruff: noqa: E402 @@ -47,16 +48,16 @@ def test_diffusion_model(model_name: str): num_outputs_per_prompt=2, ) # Extract images from request_output[0]['images'] - first_output = outputs[0] + first_output = list(outputs)[0] assert first_output.final_output_type == "image" if not hasattr(first_output, "request_output") or not first_output.request_output: raise ValueError("No request_output found in OmniRequestOutput") req_out = first_output.request_output[0] - if not isinstance(req_out, dict) or "images" not in req_out: + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out["images"] + images = req_out.images assert len(images) == 2 # check image size diff --git a/tests/e2e/offline_inference/test_teacache.py b/tests/e2e/offline_inference/test_teacache.py index 2b737911a..d7df2870b 100644 --- a/tests/e2e/offline_inference/test_teacache.py +++ b/tests/e2e/offline_inference/test_teacache.py @@ -21,6 +21,7 @@ if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from vllm_omni import Omni +from vllm_omni.outputs import OmniRequestOutput os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "1" @@ -57,16 +58,16 @@ def test_teacache(model_name: str): num_outputs_per_prompt=1, # Single output for speed ) # Extract images from request_output[0]['images'] - first_output = outputs[0] + first_output = list(outputs)[0] assert first_output.final_output_type == "image" if not hasattr(first_output, "request_output") or not first_output.request_output: raise ValueError("No request_output found in OmniRequestOutput") req_out = first_output.request_output[0] - if not isinstance(req_out, dict) or "images" not in req_out: + if not isinstance(req_out, OmniRequestOutput) or not hasattr(req_out, "images"): raise ValueError("Invalid request_output structure or missing 'images' key") - images = req_out["images"] + images = req_out.images # Verify generation succeeded assert images is not None diff --git a/tests/e2e/offline_inference/utils.py b/tests/e2e/offline_inference/utils.py index c491c10b9..3113599a3 100644 --- a/tests/e2e/offline_inference/utils.py +++ b/tests/e2e/offline_inference/utils.py @@ -110,10 +110,10 @@ def fork_new_process_for_each_test(func: Callable[_P, None]) -> Callable[_P, Non ): exc_info = cloudpickle.load(f) - if (original_exception := exc_info.get("pickled_exception")) is not None: + original_exception = exc_info.get("pickled_exception") + if original_exception is not None and isinstance(original_exception, Exception): # Re-raise the actual exception object if it was # successfully pickled. - assert isinstance(original_exception, Exception) raise original_exception if (original_tb := exc_info.get("traceback")) is not None: diff --git a/tests/e2e/online_serving/test_qwen3_omni.py b/tests/e2e/online_serving/test_qwen3_omni.py index ed2e05e2b..5d54c00c0 100644 --- a/tests/e2e/online_serving/test_qwen3_omni.py +++ b/tests/e2e/online_serving/test_qwen3_omni.py @@ -4,6 +4,7 @@ E2E Online tests for Qwen3-Omni model with video input and audio output. """ +import base64 import concurrent.futures import ctypes import os @@ -146,8 +147,6 @@ def client(omni_server): @pytest.fixture(scope="session") def base64_encoded_video() -> str: """Base64 encoded video for testing.""" - import base64 - video = VideoAsset(name="baby_reading", num_frames=4) with open(video.video_path, "rb") as f: content = f.read() @@ -240,3 +239,40 @@ def test_video_to_audio_concurrent( if hasattr(audio_message, "audio") and audio_message.audio: assert audio_message.audio.data is not None assert len(audio_message.audio.data) > 0 + + # Test streaming completion + chat_completion = client.chat.completions.create( + model=omni_server.model, + messages=messages, + stream=True, + ) + + # Collect text and audio data from stream + text_content = "" + audio_data = None + + for chunk in chat_completion: + for choice in chunk.choices: + if hasattr(choice, "delta"): + content = getattr(choice.delta, "content", None) + else: + content = None + + modality = getattr(chunk, "modality", None) + + if modality == "audio" and content: + # Audio chunk - decode base64 content + if audio_data is None: + audio_data = base64.b64decode(content) + else: + audio_data += base64.b64decode(content) + elif modality == "text" and content: + # Text chunk - accumulate text content + text_content += content if content else "" + + # Verify text output + assert text_content is not None and len(text_content) >= 2 + + # Verify audio output + assert audio_data is not None + assert len(audio_data) > 0 diff --git a/tests/entrypoints/test_omni_llm.py b/tests/entrypoints/test_omni_llm.py index 381c53284..e9c20f3c6 100644 --- a/tests/entrypoints/test_omni_llm.py +++ b/tests/entrypoints/test_omni_llm.py @@ -553,7 +553,7 @@ def test_generate_raises_on_length_mismatch(monkeypatch, fake_stage_config): omni = Omni(model="any", init_timeout=1) with pytest.raises(ValueError): - omni.generate(prompts=["hi"], sampling_params_list=[]) + list(omni.generate(prompts=["hi"], sampling_params_list=[])) def test_generate_pipeline_and_final_outputs(monkeypatch, fake_stage_config): @@ -640,7 +640,7 @@ def test_generate_pipeline_and_final_outputs(monkeypatch, fake_stage_config): # Use dicts instead of object() for serializable sampling params sampling_params_list = [{"temperature": 0.7}, {"temperature": 0.8}] prompts = ["hi"] - outputs = omni.generate(prompts=prompts, sampling_params_list=sampling_params_list) + outputs = list(omni.generate(prompts=prompts, sampling_params_list=sampling_params_list)) # Both stages have final_output=True, so should aggregate two OmniRequestOutput assert len(outputs) == 2 @@ -723,7 +723,7 @@ def test_generate_no_final_output_returns_empty(monkeypatch, fake_stage_config): ) # Use dicts instead of object() for serializable sampling params - outputs = omni.generate(prompts=["p"], sampling_params_list=[{"temperature": 0.7}, {"temperature": 0.8}]) + outputs = list(omni.generate(prompts=["p"], sampling_params_list=[{"temperature": 0.7}, {"temperature": 0.8}])) assert outputs == [] @@ -796,7 +796,7 @@ def test_generate_sampling_params_none_use_default(monkeypatch, fake_stage_confi } ) # Use the default sampling params - omni.generate(prompts=["p"], sampling_params_list=None) + list(omni.generate(prompts=["p"], sampling_params_list=None)) def test_wait_for_stages_ready_timeout(monkeypatch, fake_stage_config): @@ -924,7 +924,7 @@ def test_generate_handles_error_messages(monkeypatch, fake_stage_config): # Generate should handle error gracefully (log but continue) # Use dict instead of object() for serializable sampling params sampling_params_list = [{"temperature": 0.7}] - outputs = omni.generate(prompts=["hi"], sampling_params_list=sampling_params_list) + outputs = list(omni.generate(prompts=["hi"], sampling_params_list=sampling_params_list)) # Should return final output (error was logged but didn't stop processing) assert isinstance(outputs, list) # Since final_output=True, should have one output diff --git a/vllm_omni/distributed/omni_connectors/utils/serialization.py b/vllm_omni/distributed/omni_connectors/utils/serialization.py index c1055e5eb..ee7ddf45f 100644 --- a/vllm_omni/distributed/omni_connectors/utils/serialization.py +++ b/vllm_omni/distributed/omni_connectors/utils/serialization.py @@ -22,6 +22,11 @@ _REQUEST_OUTPUT_KEYS = frozenset({"request_id", "prompt", "prompt_token_ids", "o # Keys that identify a CompletionOutput dict (for reconstruction) _COMPLETION_OUTPUT_KEYS = frozenset({"index", "text", "token_ids", "finish_reason"}) +# Keys that identify an OmniRequestOutput dict (for reconstruction) +# OmniRequestOutput has 'final_output_type' which is unique, or can be identified by +# having 'finished' and ('images' or 'final_output_type') +_OMNI_REQUEST_OUTPUT_KEYS = frozenset({"finished", "final_output_type"}) + class OmniMsgpackEncoder: """ @@ -173,7 +178,7 @@ class OmniMsgpackDecoder: return self._post_process(result) def _post_process(self, obj: Any) -> Any: - """Recursively restore tensor/ndarray/image/RequestOutput from their dict representations.""" + """Recursively restore tensor/ndarray/image/RequestOutput/OmniRequestOutput from their dict representations.""" if isinstance(obj, dict): # Check for type markers first if obj.get(_TENSOR_MARKER): @@ -186,6 +191,11 @@ class OmniMsgpackDecoder: # Process values recursively first processed = {k: self._post_process(v) for k, v in obj.items()} + # Check if this looks like an OmniRequestOutput (check before RequestOutput + # since OmniRequestOutput may also have some RequestOutput-like fields) + if self._is_omni_request_output(processed): + return self._decode_omni_request_output(processed) + # Check if this looks like a RequestOutput if _REQUEST_OUTPUT_KEYS.issubset(processed.keys()): return self._decode_request_output(processed) @@ -204,6 +214,48 @@ class OmniMsgpackDecoder: return obj + def _is_omni_request_output(self, obj: dict[str, Any]) -> bool: + """Check if a dict looks like an OmniRequestOutput. + + OmniRequestOutput can be identified by: + - Having 'finished' and 'final_output_type' fields (unique to OmniRequestOutput) + - OR having 'finished' and 'images' fields (diffusion mode) + """ + # Must have 'finished' field + if "finished" not in obj: + return False + + # Check for unique identifier: 'final_output_type' + if "final_output_type" in obj: + return True + + # Alternative: check for 'images' field (diffusion mode) + if "images" in obj: + return True + + return False + + def _decode_omni_request_output(self, obj: dict[str, Any]) -> Any: + """Decode dict to OmniRequestOutput. + + OmniRequestOutput is a dataclass, so we can use msgspec.convert + or construct it directly. + """ + from vllm_omni.outputs import OmniRequestOutput + + try: + # Use msgspec.convert for dataclass reconstruction + return msgspec.convert(obj, OmniRequestOutput) + except Exception: + try: + # Fallback: construct directly if msgspec.convert fails + # (e.g., if some fields are missing or have wrong types) + return OmniRequestOutput(**obj) + except Exception: + # If both attempts fail, return dict as-is (defensive fallback) + # This should rarely happen if _is_omni_request_output is correct + return obj + def _decode_tensor(self, obj: dict[str, Any]) -> torch.Tensor: """Decode dict to torch.Tensor.""" dtype_str = obj["dtype"] diff --git a/vllm_omni/entrypoints/async_omni.py b/vllm_omni/entrypoints/async_omni.py index d133c9b1b..e2e8b83d4 100644 --- a/vllm_omni/entrypoints/async_omni.py +++ b/vllm_omni/entrypoints/async_omni.py @@ -341,78 +341,89 @@ class AsyncOmni(OmniBase): logger.debug(f"[{self._name}] Entering scheduling loop: stages={num_stages}") for stage_id, stage in enumerate(self.stage_list[: final_stage_id_for_e2e + 1]): - result = await req_state.queue.get() - assert stage_id == req_state.stage_id + finished = False + while not finished: + result = await req_state.queue.get() + assert stage_id == req_state.stage_id - req_id = result.get("request_id") - if "error" in result: - logger.error( - f"[{self._name}] Stage {stage_id} error on request {req_id}: {result['error']}", - ) - raise RuntimeError(result) # Request Finished due to error + req_id = result.get("request_id") + if "error" in result: + logger.error( + f"[{self._name}] Stage {stage_id} error on request {req_id}: {result['error']}", + ) + raise RuntimeError(result) # Request Finished due to error + req_id = result.get("request_id") + if "error" in result: + logger.error( + f"[{self._name}] Stage {stage_id} error on request {req_id}: {result['error']}", + ) + raise RuntimeError(result) # Request Finished due to error - engine_outputs = _load(result, obj_key="engine_outputs", shm_key="engine_outputs_shm") - # Mark last output time for this stage whenever we receive outputs - metrics.stage_last_ts[stage_id] = max(metrics.stage_last_ts[stage_id] or 0.0, time.time()) - try: - _m = asdict(result.get("metrics")) - if _m is not None: - metrics.on_stage_metrics(stage_id, req_id, _m) - except Exception as e: - logger.exception( - f"[{self._name}] Failed to process metrics for stage {stage_id}, req {req_id}: {e}", - ) - logger.debug( - f"[{self._name}] Stage-{stage_id} completed request {req_id}; forwarding or finalizing", - ) - stage.set_engine_outputs(engine_outputs) - - if getattr(stage, "final_output", False): - logger.debug( - f"[{self._name}] Request {req_id} finalized at stage-{stage_id}", - ) - - # End-to-end timing and time-per-token for final output - # (only once per request at the designated final stage) + engine_outputs = _load(result, obj_key="engine_outputs", shm_key="engine_outputs_shm") + # Mark last output time for this stage whenever we receive outputs + metrics.stage_last_ts[stage_id] = max(metrics.stage_last_ts[stage_id] or 0.0, time.time()) try: - rid_key = str(req_id) - if stage_id == final_stage_id_for_e2e and rid_key not in metrics.e2e_done: - metrics.on_finalize_request( - stage_id, - req_id, - engine_outputs, - _req_start_ts.get(req_id, _wall_start_ts), - ) + _m = asdict(result.get("metrics")) + if _m is not None: + metrics.on_stage_metrics(stage_id, req_id, _m) except Exception as e: logger.exception( - f"[{self._name}] Finalize request handling error for req {req_id} at stage {stage_id}: {e}", + f"[{self._name}] Failed to process metrics for stage {stage_id}, req {req_id}: {e}", ) + logger.debug( + f"[{self._name}] Stage-{stage_id} completed request {req_id}; forwarding or finalizing", + ) + stage.set_engine_outputs(engine_outputs) if isinstance(engine_outputs, list): engine_outputs = engine_outputs[0] - # Handle diffusion outputs that already contain images - if stage.final_output_type == "image": - images = [] - if isinstance(engine_outputs, OmniRequestOutput) and engine_outputs.images: - images = engine_outputs.images - elif hasattr(engine_outputs, "images") and engine_outputs.images: - images = engine_outputs.images - yield OmniRequestOutput( - stage_id=stage_id, - final_output_type=stage.final_output_type, - request_output=engine_outputs, - images=images, - ) - else: - yield OmniRequestOutput( - stage_id=stage_id, - final_output_type=stage.final_output_type, - request_output=engine_outputs, + finished = engine_outputs.finished + + if getattr(stage, "final_output", False): + logger.debug( + f"[{self._name}] Request {req_id} finalized at stage-{stage_id}", ) + # End-to-end timing and time-per-token for final output + # (only once per request at the designated final stage) + try: + rid_key = str(req_id) + if stage_id == final_stage_id_for_e2e and rid_key not in metrics.e2e_done: + metrics.on_finalize_request( + stage_id, + req_id, + [engine_outputs], + _req_start_ts.get(req_id, _wall_start_ts), + ) + except Exception as e: + logger.exception( + f"[{self._name}] Finalize request handling error for req " + f"{req_id} at stage {stage_id}: {e}", + ) + + # Handle diffusion outputs that already contain images + if stage.final_output_type == "image": + images = [] + if isinstance(engine_outputs, OmniRequestOutput) and engine_outputs.images: + images = engine_outputs.images + elif hasattr(engine_outputs, "images") and engine_outputs.images: + images = engine_outputs.images + yield OmniRequestOutput( + stage_id=stage_id, + final_output_type=stage.final_output_type, + request_output=engine_outputs, + images=images, + ) + else: + yield OmniRequestOutput( + stage_id=stage_id, + final_output_type=stage.final_output_type, + request_output=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: + if next_stage_id <= final_stage_id_for_e2e and finished: 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] diff --git a/vllm_omni/entrypoints/omni.py b/vllm_omni/entrypoints/omni.py index 378caf378..b5100ee5b 100644 --- a/vllm_omni/entrypoints/omni.py +++ b/vllm_omni/entrypoints/omni.py @@ -6,7 +6,7 @@ import os import time import uuid import weakref -from collections.abc import Callable, Sequence +from collections.abc import Callable, Generator, Sequence from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import asdict from pprint import pformat @@ -388,7 +388,7 @@ class Omni(OmniBase): self._ray_pg, ) - def generate(self, *args: Any, **kwargs: dict[str, Any]) -> list[OmniRequestOutput]: + def generate(self, *args: Any, **kwargs: dict[str, Any]) -> Generator[OmniRequestOutput, None, None]: """Generate outputs for the given prompts. Orchestrates the multi-stage pipeline based on YAML configuration. @@ -439,7 +439,7 @@ class Omni(OmniBase): sampling_params_list = per_stage_params try: - return self._run_generation(prompts, sampling_params_list) + yield from self._run_generation(prompts, sampling_params_list) except Exception as e: logger.exception("[Orchestrator] Failed to run generation: %s", e) raise e @@ -451,7 +451,7 @@ class Omni(OmniBase): prompts: PromptType | Sequence[PromptType] | OmniDiffusionRequest | Sequence[OmniDiffusionRequest], sampling_params_list: Any | Sequence[Any] | None = None, use_tqdm: bool | Callable[..., tqdm] = True, - ) -> list[OmniRequestOutput]: + ) -> Generator[OmniRequestOutput, None, None]: """Run generation through all stages in the pipeline.""" logger.debug(f"[{self._name}] generate() called") if sampling_params_list is None: @@ -472,8 +472,6 @@ class Omni(OmniBase): else: request_prompts = list(prompts) - final_outputs: list[OmniRequestOutput] = [] - # Orchestrator keeps stage objects for input derivation num_stages = len(self.stage_list) @@ -600,13 +598,6 @@ class Omni(OmniBase): stage.set_engine_outputs(engine_outputs) if getattr(stage, "final_output", False): - final_outputs.append( - OmniRequestOutput( - stage_id=stage_id, - final_output_type=stage.final_output_type, # type: ignore[attr-defined] - request_output=engine_outputs, - ) - ) logger.debug( f"[{self._name}] Request {req_id} finalized at stage-{stage_id}", ) @@ -626,6 +617,11 @@ class Omni(OmniBase): logger.exception( f"[{self._name}] Finalize request handling error for req {req_id} at stage {stage_id}: {e}", ) + yield OmniRequestOutput( + stage_id=stage_id, + final_output_type=stage.final_output_type, # type: ignore[attr-defined] + request_output=engine_outputs, + ) next_stage_id = stage_id + 1 if next_stage_id <= final_stage_id_to_prompt[req_id]: @@ -690,8 +686,6 @@ class Omni(OmniBase): except Exception as e: logger.exception(f"[{self._name}] Failed to build/log summary: {e}") - return final_outputs - @property def _name(self) -> str: return "Orchestrator" diff --git a/vllm_omni/entrypoints/omni_stage.py b/vllm_omni/entrypoints/omni_stage.py index 205deff58..698c7a87e 100644 --- a/vllm_omni/entrypoints/omni_stage.py +++ b/vllm_omni/entrypoints/omni_stage.py @@ -1149,16 +1149,18 @@ async def _stage_worker_async( diffusion_kwargs = prepare_sampling_params(sampling_params, "diffusion") # AsyncOmniDiffusion.generate returns a single result, not an async generator gen_output = await stage_engine.generate(prompt=prompt, request_id=rid, **diffusion_kwargs) + _gen_t1 = _time.time() + _gen_ms = (_gen_t1 - _gen_t0) * 1000.0 + await generation_out_q.put((rid, gen_output, _gen_ms)) else: # LLM stages: ensure using SamplingParams llm_sampling_params = prepare_sampling_params(sampling_params, "llm") gen_output = None async for res in stage_engine.generate(ein, llm_sampling_params, rid): gen_output = res - - _gen_t1 = _time.time() - _gen_ms = (_gen_t1 - _gen_t0) * 1000.0 - await generation_out_q.put((rid, gen_output, _gen_ms)) + _gen_t1 = _time.time() + _gen_ms = (_gen_t1 - _gen_t0) * 1000.0 + await generation_out_q.put((rid, gen_output, _gen_ms)) except Exception as e: logger.exception("Failed on request %s: %s", rid, e) out_q.put( diff --git a/vllm_omni/entrypoints/openai/protocol/__init__.py b/vllm_omni/entrypoints/openai/protocol/__init__.py index b17b648eb..da65e1817 100644 --- a/vllm_omni/entrypoints/openai/protocol/__init__.py +++ b/vllm_omni/entrypoints/openai/protocol/__init__.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +from vllm_omni.entrypoints.openai.protocol.chat_completion import OmniChatCompletionStreamResponse from vllm_omni.entrypoints.openai.protocol.images import ( ImageData, ImageGenerationRequest, @@ -13,4 +14,5 @@ __all__ = [ "ImageGenerationRequest", "ImageGenerationResponse", "ResponseFormat", + "OmniChatCompletionStreamResponse", ] diff --git a/vllm_omni/entrypoints/openai/protocol/chat_completion.py b/vllm_omni/entrypoints/openai/protocol/chat_completion.py new file mode 100644 index 000000000..9f6076249 --- /dev/null +++ b/vllm_omni/entrypoints/openai/protocol/chat_completion.py @@ -0,0 +1,5 @@ +from vllm.entrypoints.openai.protocol import ChatCompletionStreamResponse + + +class OmniChatCompletionStreamResponse(ChatCompletionStreamResponse): + modality: str | None = "text" diff --git a/vllm_omni/entrypoints/openai/serving_chat.py b/vllm_omni/entrypoints/openai/serving_chat.py index 1f225e5b9..6fb9750cc 100644 --- a/vllm_omni/entrypoints/openai/serving_chat.py +++ b/vllm_omni/entrypoints/openai/serving_chat.py @@ -6,7 +6,7 @@ import uuid from collections.abc import AsyncGenerator, AsyncIterator, Callable, Sequence from datetime import datetime, timedelta, timezone from io import BytesIO -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any, Final, Optional import jinja2 from fastapi import Request @@ -29,13 +29,18 @@ from vllm.entrypoints.chat_utils import ( make_tool_call_id, resolve_chat_template_content_format, ) -from vllm.entrypoints.harmony_utils import parse_chat_output +from vllm.entrypoints.harmony_utils import get_streamable_parser_for_assistant, parse_chat_output from vllm.entrypoints.openai.protocol import ( ChatCompletionNamedToolChoiceParam, ChatCompletionRequest, ChatCompletionResponse, ChatCompletionResponseChoice, + ChatCompletionResponseStreamChoice, + ChatCompletionStreamResponse, ChatMessage, + DeltaFunctionCall, + DeltaMessage, + DeltaToolCall, ErrorResponse, FunctionCall, FunctionDefinition, @@ -56,6 +61,8 @@ from vllm.entrypoints.openai.serving_engine import ( ) from vllm.entrypoints.openai.tool_parsers import ToolParser from vllm.entrypoints.openai.tool_parsers.mistral_tool_parser import MistralToolCall +from vllm.entrypoints.openai.utils import maybe_filter_parallel_tool_calls +from vllm.entrypoints.utils import should_include_usage from vllm.inputs.data import PromptType from vllm.logger import init_logger from vllm.lora.request import LoRARequest @@ -68,10 +75,12 @@ from vllm.tokenizers.mistral import ( truncate_tool_call_ids, validate_request_params, ) +from vllm.transformers_utils.tokenizer import AnyTokenizer from vllm.utils.collection_utils import as_list from vllm_omni.entrypoints.chat_utils import parse_chat_messages_futures from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin +from vllm_omni.entrypoints.openai.protocol import OmniChatCompletionStreamResponse from vllm_omni.entrypoints.openai.protocol.audio import AudioResponse, CreateAudio from vllm_omni.outputs import OmniRequestOutput @@ -226,6 +235,9 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): raw_request.state.request_metadata = request_metadata output_modalities = getattr(request, "modalities", self.engine_client.output_modalities) + request.modalities = ( + output_modalities if output_modalities is not None else self.engine_client.output_modalities + ) # Schedule the request and get the result generator. generators: list[AsyncGenerator[RequestOutput, None]] = [] @@ -266,7 +278,15 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): # Streaming response if request.stream: - raise RuntimeError("Not support streaming output now.") + return self.chat_completion_stream_generator( + request, + result_generator, + request_id, + model_name, + conversation, + tokenizer, + request_metadata, + ) try: return await self.chat_completion_full_generator( @@ -518,6 +538,732 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): lora_request, ) + async def chat_completion_stream_generator( + self, + request: ChatCompletionRequest, + result_generator: AsyncIterator[RequestOutput], + request_id: str, + model_name: str, + conversation: list[ConversationMessage], + tokenizer: AnyTokenizer, + request_metadata: RequestResponseMetadata, + ): + created_time = int(time.time()) + chunk_object_type: Final = "chat.completion.chunk" + first_iteration_dict = {} + assert hasattr(request, "modalities") and request.modalities is not None, ( + "Streaming request must specify output modalities" + ) + for modality in request.modalities: + first_iteration_dict[modality] = True + + # Send response for each token for each request.n (index) + num_choices = 1 if request.n is None else request.n + previous_num_tokens = [0] * num_choices + finish_reason_sent = [False] * num_choices + num_prompt_tokens = 0 + num_cached_tokens = None + if self.use_harmony: + harmony_parsers = [get_streamable_parser_for_assistant() for _ in range(num_choices)] + harmony_tools_streamed = [False] * num_choices + tools_streamed = [False] * num_choices + + if isinstance(request.tool_choice, ChatCompletionNamedToolChoiceParam): + tool_choice_function_name = request.tool_choice.function.name + else: + tool_choice_function_name = None + + # Determine whether tools are in use with "auto" tool choice + tool_choice_auto = not tool_choice_function_name and self._should_stream_with_auto_tool_parsing(request) + + all_previous_token_ids: list[list[int]] | None + function_name_returned = [False] * num_choices + if self.tool_call_id_type == "kimi_k2": + history_tool_call_cnt = get_history_tool_calls_cnt(conversation) + else: + history_tool_call_cnt = 0 + + # Always track previous_texts for comprehensive output logging + previous_texts = [""] * num_choices + + # Only one of these will be used, thus previous_texts and + # all_previous_token_ids will not be used twice in the same iteration. + if tool_choice_auto or self.reasoning_parser: + # These are only required in "auto" tool choice case + all_previous_token_ids = [[]] * num_choices + # For reasoning parser and tool call all enabled + added_content_delta_arr = [False] * num_choices + reasoning_end_arr = [False] * num_choices + else: + all_previous_token_ids = None + + try: + if self.reasoning_parser: + reasoning_parser = self.reasoning_parser( + tokenizer, + chat_template_kwargs=request.chat_template_kwargs, # type: ignore + ) + except RuntimeError as e: + logger.exception("Error in reasoning parser creation.") + data = self.create_streaming_error_response(str(e)) + yield f"data: {data}\n\n" + yield "data: [DONE]\n\n" + return + # Prepare the tool parser if it's needed + try: + if tool_choice_auto and self.tool_parser: + tool_parsers: list[ToolParser | None] = [self.tool_parser(tokenizer)] * num_choices + else: + tool_parsers = [None] * num_choices + except Exception as e: + logger.exception("Error in tool parser creation.") + data = self.create_streaming_error_response(str(e)) + yield f"data: {data}\n\n" + yield "data: [DONE]\n\n" + return + + stream_options = request.stream_options + include_usage, include_continuous_usage = should_include_usage(stream_options, self.enable_force_include_usage) + + try: + async for omni_res in result_generator: + final_output_type = omni_res.final_output_type + res = omni_res.request_output + if final_output_type not in first_iteration_dict: + logger.warning(f"final output type: {final_output_type} is not needed by the request") + continue + + if res.prompt_token_ids is not None: + num_prompt_tokens = len(res.prompt_token_ids) + if res.encoder_prompt_token_ids is not None: + num_prompt_tokens += len(res.encoder_prompt_token_ids) + + # We need to do it here, because if there are exceptions in + # the result_generator, it needs to be sent as the FIRST + # response (by the try...catch). + if first_iteration_dict[final_output_type] and final_output_type == "text": + num_cached_tokens = res.num_cached_tokens + # Send first response for each request.n (index) with + # the role + role = self.get_chat_request_role(request) + + # NOTE num_choices defaults to 1 so this usually executes + # once per request + for i in range(num_choices): + choice_data = ChatCompletionResponseStreamChoice( + index=i, + delta=DeltaMessage( + role=role, + content="", + ), + logprobs=None, + finish_reason=None, + ) + + # return prompt_token_ids at the first chunk ever + chunk = OmniChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=[choice_data], + model=model_name, + prompt_token_ids=(res.prompt_token_ids if request.return_token_ids else None), + modality=final_output_type, + ) + + # if continuous usage stats are requested, add it + if include_continuous_usage: + chunk.usage = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=0, + total_tokens=num_prompt_tokens, + ) + + data = chunk.model_dump_json(exclude_unset=True) + yield f"data: {data}\n\n" + + # Send response to echo the input portion of the + # last message + if request.echo: + last_msg_content: str | list[dict[str, str]] = "" + if conversation and "content" in conversation[-1] and conversation[-1].get("role") == role: + last_msg_content = conversation[-1]["content"] or "" + + if last_msg_content: + for i in range(num_choices): + choice_data = ChatCompletionResponseStreamChoice( + index=i, + delta=DeltaMessage(content=last_msg_content), + logprobs=None, + finish_reason=None, + ) + chunk = OmniChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=[choice_data], + model=model_name, + modality=final_output_type, + ) + if include_continuous_usage: + chunk.usage = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=0, + total_tokens=num_prompt_tokens, + ) + + data = chunk.model_dump_json(exclude_unset=True) + yield f"data: {data}\n\n" + first_iteration_dict[final_output_type] = False + + if final_output_type == "text": + for output in res.outputs: + i = output.index + tool_parser = tool_parsers[i] + + if finish_reason_sent[i]: + continue + + if request.logprobs and request.top_logprobs is not None: + assert output.logprobs is not None, "Did not output logprobs" + logprobs = self._create_chat_logprobs( + token_ids=output.token_ids, + top_logprobs=output.logprobs, + tokenizer=tokenizer, + num_output_top_logprobs=request.top_logprobs, + return_as_token_id=request.return_tokens_as_token_ids, + ) + else: + logprobs = None + + if self.use_harmony: + harmony_parser = harmony_parsers[i] + prev_recipient = harmony_parser.current_recipient + delta_text = "" + for token_id in output.token_ids: + harmony_parser.process(token_id) + delta_text += harmony_parser.last_content_delta or "" + cur_channel = harmony_parser.current_channel + cur_recipient = harmony_parser.current_recipient + else: + # output.text is cumulative, extract only the delta portion + previous_text = previous_texts[i] if previous_texts else "" + if output.text is not None: + delta_text = output.text[len(previous_text) :] + else: + delta_text = "" + + if not delta_text and not output.token_ids and not previous_num_tokens[i]: + # Chunked prefill case, don't return empty chunks + continue + + delta_message: DeltaMessage | None + + # just update previous_texts and previous_token_ids + if tool_choice_auto or self.reasoning_parser: + assert previous_texts is not None + assert all_previous_token_ids is not None + previous_text = previous_texts[i] + previous_token_ids = all_previous_token_ids[i] + current_text = previous_text + delta_text + # avoid the None + list error. + if previous_token_ids: + current_token_ids = previous_token_ids + as_list(output.token_ids) + else: + current_token_ids = as_list(output.token_ids) + + if self.use_harmony: + if cur_channel == "final": + delta_message = DeltaMessage(content=delta_text) + elif cur_channel == "analysis": + if request.include_reasoning: + delta_message = DeltaMessage(reasoning=delta_text) + else: + delta_message = None + elif ( + cur_channel == "commentary" and cur_recipient and cur_recipient.startswith("functions.") + ): + # Count completed tool calls to determine index + base_index = 0 + for msg in harmony_parser.messages: + if ( + msg.channel == "commentary" + and msg.recipient + and msg.recipient.startswith("functions.") + ): + base_index += 1 + + if prev_recipient != cur_recipient: + tool_name = cur_recipient.split("functions.", 1)[1] + delta_message = DeltaMessage( + tool_calls=[ + DeltaToolCall( + id=make_tool_call_id(), + type="function", + function=DeltaFunctionCall( + name=tool_name, + arguments="", + ), + index=base_index, + ) + ] + ) + elif delta_text: + delta_message = DeltaMessage( + tool_calls=[ + DeltaToolCall( + index=base_index, + function=DeltaFunctionCall(arguments=delta_text), + ) + ] + ) + else: + delta_message = None + + if delta_message is not None: + harmony_tools_streamed[i] = True + else: + delta_message = None + # handle streaming deltas for tools with named tool_choice + elif tool_choice_function_name: + if ( + self.reasoning_parser + and not reasoning_end_arr[i] + and not reasoning_parser.is_reasoning_end(previous_token_ids) + ): + assert reasoning_parser is not None + delta_message = reasoning_parser.extract_reasoning_streaming( + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + output.token_ids, + ) + # When encountering think end id in delta_token_ids + # or think end id in prompt_token_ids + # i.e {"enable_thinking": False}, + # set reasoning status to end. + # Only keep 'content', remove 'reasoning'. + if reasoning_parser.is_reasoning_end(as_list(output.token_ids)) or ( + res.prompt_token_ids and reasoning_parser.is_reasoning_end(res.prompt_token_ids) + ): + reasoning_end_arr[i] = True + if delta_message and delta_message.content: + # This need to be added to next `delta_text` + current_text = delta_message.content + delta_message.content = None + else: + current_text = "" + else: + # Just to add remaining `content` + if self.reasoning_parser: + delta_text = previous_text + delta_text + current_text = "" + + if function_name_returned[i]: + delta_tool_call = DeltaToolCall( + function=DeltaFunctionCall(arguments=delta_text), + index=i, + ) + else: + delta_tool_call = DeltaToolCall( + id=make_tool_call_id(), + type="function", + function=DeltaFunctionCall( + name=tool_choice_function_name, + arguments=delta_text, + ), + index=i, + ) + function_name_returned[i] = True + + delta_message = DeltaMessage( + tool_calls=[ + delta_tool_call, + ] + ) + tools_streamed[i] = True + + elif request.tool_choice == "required": + assert previous_texts is not None + previous_text = previous_texts[i] + current_text = previous_text + delta_text + fn_name_returned = function_name_returned[i] + output_token_ids = as_list(output.token_ids) + + if ( + self.reasoning_parser is not None + and not reasoning_end_arr[i] + and res.prompt_token_ids + and reasoning_parser.is_reasoning_end(res.prompt_token_ids) + ): + reasoning_end_arr[i] = True + + if self.reasoning_parser and not reasoning_end_arr[i]: + delta_message = reasoning_parser.extract_reasoning_streaming( + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + output_token_ids, + ) + if reasoning_parser.is_reasoning_end(output_token_ids): + reasoning_end_arr[i] = True + if delta_message and delta_message.content: + current_text = delta_message.content + delta_message.content = None + else: + # reasoning ended + current_text = "" + + else: + # either finished reasoning or no reasoning at all + content = current_text + + delta_message, function_name_returned[i] = self.extract_tool_call_required_streaming( + previous_text=previous_text, + current_text=content, + delta_text=delta_text, + function_name_returned=fn_name_returned, + tool_call_idx=history_tool_call_cnt, + ) + if ( + delta_message + and delta_message.tool_calls + and delta_message.tool_calls[0].id is not None + ): + history_tool_call_cnt += 1 + tools_streamed[i] = True + + # handle streaming deltas for tools with "auto" tool choice + # and reasoning parser + elif tool_choice_auto and self.reasoning_parser: + assert tool_parser is not None + assert reasoning_parser is not None + assert added_content_delta_arr is not None + assert reasoning_end_arr is not None + output_token_ids = as_list(output.token_ids) + if not reasoning_end_arr[i]: + delta_message = reasoning_parser.extract_reasoning_streaming( + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + output_token_ids, + ) + # When encountering think end id in prompt_token_ids + # i.e {"enable_thinking": False}, + # set reasoning status to end. + # Remove the text and token ids related + # to 'reasoning'. + if res.prompt_token_ids and reasoning_parser.is_reasoning_end(res.prompt_token_ids): + reasoning_end_arr[i] = True + current_token_ids = output_token_ids + if delta_message and delta_message.content: + current_text = delta_message.content + delta_message.content = None + else: + current_text = "" + # When encountering think end id in delta_token_ids, + # set reasoning status to end. + # Remove the text and token ids related + # to 'reasoning'. + if reasoning_parser.is_reasoning_end(output_token_ids): + reasoning_end_arr[i] = True + current_token_ids = reasoning_parser.extract_content_ids(output_token_ids) + if delta_message and delta_message.content: + current_text = delta_message.content + delta_message.content = None + else: + current_text = "" + + # handle tool calls only after reasoning is done, + else: + delta_token_ids = output_token_ids + # First time to tool call, + # add the remaining text and token ids + # to delta from previous + if not added_content_delta_arr[i]: + added_content_delta_arr[i] = True + previous_text = "" + previous_token_ids = [] + delta_text = current_text + delta_token_ids = current_token_ids + + delta_message = tool_parser.extract_tool_calls_streaming( + previous_text=previous_text, + current_text=current_text, + delta_text=delta_text, + previous_token_ids=previous_token_ids, + current_token_ids=current_token_ids, + delta_token_ids=delta_token_ids, + request=request, + ) + if delta_message and delta_message.tool_calls: + tools_streamed[i] = True + # when only tool calls + elif tool_choice_auto: + assert tool_parser is not None + delta_message = tool_parser.extract_tool_calls_streaming( + previous_text=previous_text, + current_text=current_text, + delta_text=delta_text, + previous_token_ids=previous_token_ids, + current_token_ids=current_token_ids, + delta_token_ids=output.token_ids, + request=request, + ) + if delta_message and delta_message.tool_calls: + tools_streamed[i] = True + + # when only reasoning + elif self.reasoning_parser: + delta_message = reasoning_parser.extract_reasoning_streaming( + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + output.token_ids, + ) + # handle streaming just a content delta + else: + delta_message = DeltaMessage(content=delta_text) + + # update the previous values for the next iteration + if (tool_choice_auto or self.reasoning_parser) and not self.use_harmony: + assert previous_texts is not None + assert all_previous_token_ids is not None + previous_texts[i] = current_text + all_previous_token_ids[i] = current_token_ids + else: + # Update for comprehensive logging even in simple case + assert previous_texts is not None + previous_texts[i] += delta_text + + # set the previous values for the next iteration + previous_num_tokens[i] += len(output.token_ids) + + # if the message delta is None (e.g. because it was a + # "control token" for tool calls or the parser otherwise + # wasn't ready to send a token, then + # get the next token without streaming a chunk + if delta_message is None: + if output.finish_reason is None: + continue + else: + delta_message = DeltaMessage() + + # Log streaming delta if output logging is enabled + if self.enable_log_outputs and self.request_logger: + delta_content = "" + if delta_message.content: + delta_content = delta_message.content + elif delta_message.tool_calls: + delta_content = "".join( + tc.function.arguments + for tc in delta_message.tool_calls + if tc.function and tc.function.arguments + ) + + if delta_content: + self.request_logger.log_outputs( + request_id=request_id, + outputs=delta_content, + output_token_ids=as_list(output.token_ids), + finish_reason=output.finish_reason, + is_streaming=True, + delta=True, + ) + + if output.finish_reason is None: + # Send token-by-token response for each request.n + choice_data = ChatCompletionResponseStreamChoice( + index=i, + delta=delta_message, + logprobs=logprobs, + finish_reason=None, + token_ids=(as_list(output.token_ids) if request.return_token_ids else None), + ) + + # if the model is finished generating + else: + # check to make sure we haven't "forgotten" to stream + # any tokens that were generated but previously + # matched by partial json parsing + # only happens if we are NOT using structured outputs + auto_tools_called = False + if tool_parser: + auto_tools_called = len(tool_parser.prev_tool_call_arr) > 0 + index = len(tool_parser.prev_tool_call_arr) - 1 if auto_tools_called else 0 + else: + index = 0 + + if self._should_check_for_unstreamed_tool_arg_tokens(delta_message, output) and tool_parser: + latest_delta_len = 0 + if ( + isinstance( + delta_message.tool_calls[0].function, + DeltaFunctionCall, + ) + ) and isinstance(delta_message.tool_calls[0].function.arguments, str): + latest_delta_len = len(delta_message.tool_calls[0].function.arguments) + + # get the expected call based on partial JSON + # parsing which "autocompletes" the JSON + expected_call = json.dumps( + tool_parser.prev_tool_call_arr[index].get("arguments", {}), + ensure_ascii=False, + ) + + # get what we've streamed so far for arguments + # for the current tool + actual_call = tool_parser.streamed_args_for_tool[index] + if latest_delta_len > 0: + actual_call = actual_call[:-latest_delta_len] + + # check to see if there's anything left to stream + remaining_call = expected_call.replace(actual_call, "", 1) + # set that as a delta message + delta_message = DeltaMessage( + tool_calls=[ + DeltaToolCall( + index=index, + function=DeltaFunctionCall(arguments=remaining_call).model_dump( + exclude_none=True + ), + ) + ] + ) + + # Send the finish response for each request.n only once + # In OpenAI's API, when a tool is called, the + # finish_reason is: + # "tool_calls" for "auto" or "required" tool calls, + # and "stop" for named tool calls. + if ( + auto_tools_called + or (tools_streamed[i] and not tool_choice_function_name) + or (self.use_harmony and harmony_tools_streamed[i]) + ): + finish_reason_ = "tool_calls" + else: + finish_reason_ = output.finish_reason if output.finish_reason else "stop" + choice_data = ChatCompletionResponseStreamChoice( + index=i, + delta=delta_message, + logprobs=logprobs, + finish_reason=finish_reason_, + stop_reason=output.stop_reason, + token_ids=(as_list(output.token_ids) if request.return_token_ids else None), + ) + + finish_reason_sent[i] = True + + choice_data = maybe_filter_parallel_tool_calls(choice_data, request) + chunk = OmniChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=[choice_data], + model=model_name, + modality=final_output_type, + ) + + # handle usage stats if requested & if continuous + if include_continuous_usage: + completion_tokens = previous_num_tokens[i] + chunk.usage = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=num_prompt_tokens + completion_tokens, + ) + + data = chunk.model_dump_json(exclude_unset=True) + yield f"data: {data}\n\n" + + elif final_output_type == "audio": + choices_data = self._create_audio_choice(omni_res, role, request, stream=True) + chunk = OmniChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=choices_data, + model=model_name, + modality=final_output_type, + ) + chunk.usage = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=0, + total_tokens=num_prompt_tokens, + ) + data = chunk.model_dump_json(exclude_unset=True) + yield f"data: {data}\n\n" + + else: + logger.warning(f"Unsupported streaming final output type: {final_output_type}") + continue + + # once the final token is handled, if stream_options.include_usage + # is sent, send the usage + if include_usage: + completion_tokens = sum(previous_num_tokens) + final_usage = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=num_prompt_tokens + completion_tokens, + ) + if self.enable_prompt_tokens_details and num_cached_tokens: + final_usage.prompt_tokens_details = PromptTokenUsageInfo(cached_tokens=num_cached_tokens) + + final_usage_chunk = ChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=[], + model=model_name, + usage=final_usage, + ) + final_usage_data = final_usage_chunk.model_dump_json(exclude_unset=True, exclude_none=True) + yield f"data: {final_usage_data}\n\n" + + # report to FastAPI middleware aggregate usage across all choices + num_completion_tokens = sum(previous_num_tokens) + request_metadata.final_usage_info = UsageInfo( + prompt_tokens=num_prompt_tokens, + completion_tokens=num_completion_tokens, + total_tokens=num_prompt_tokens + num_completion_tokens, + ) + + # Log complete streaming response if output logging is enabled + if self.enable_log_outputs and self.request_logger: + # Log the complete response for each choice + for i in range(num_choices): + full_text = ( + previous_texts[i] + if previous_texts and i < len(previous_texts) + else f"" + ) + self.request_logger.log_outputs( + request_id=request_id, + outputs=full_text, + output_token_ids=None, # Consider also logging all token IDs + finish_reason="streaming_complete", + is_streaming=True, + delta=False, + ) + + except Exception as e: + # TODO: Use a vllm-specific Validation Error + logger.exception("Error in chat completion stream generator.") + data = self.create_streaming_error_response(str(e)) + yield f"data: {data}\n\n" + # Send the final done message after all response.n are finished + yield "data: [DONE]\n\n" + async def chat_completion_full_generator( self, request: ChatCompletionRequest, @@ -553,6 +1299,9 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): for omni_outputs in final_outputs: choices_data = [] + if omni_outputs.request_output is not None and not getattr(omni_outputs.request_output, "finished", False): + continue + if omni_outputs.final_output_type == "text": ( choices_data, @@ -562,9 +1311,9 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): kv_transfer_params, ) = self._create_text_choice(request, omni_outputs, tokenizer, conversation, role) elif omni_outputs.final_output_type == "audio": - choices_data = self._create_audio_choice(omni_outputs, role) + choices_data = self._create_audio_choice(omni_outputs, role, request, stream=False) elif omni_outputs.final_output_type == "image": - choices_data = self._create_image_choice(omni_outputs, role) + choices_data = self._create_image_choice(omni_outputs, role, request, stream=False) else: logger.warning(f"Unsupported final output type: {omni_outputs.final_output_type}") continue @@ -857,7 +1606,9 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): return choices, usage, prompt_logprobs, prompt_token_ids, kv_transfer_params - def _create_audio_choice(self, omni_outputs: OmniRequestOutput, role: str): + def _create_audio_choice( + self, omni_outputs: OmniRequestOutput, role: str, request: ChatCompletionRequest, stream: bool = False + ): choices: list[ChatCompletionResponseChoice] = [] final_res = omni_outputs.request_output audio_tensor = final_res.multimodal_output["audio"].float().detach().cpu().numpy() @@ -893,17 +1644,29 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): ) for output in final_res.outputs: - choice_data = ChatCompletionResponseChoice( - index=output.index, - message=ChatMessage(role=role, audio=audio_obj), - logprobs=None, - finish_reason="stop", - stop_reason=None, - ) + if stream: + choice_data = ChatCompletionResponseStreamChoice( + index=output.index, + delta=DeltaMessage(role=role, content=audio_base64), + logprobs=None, + finish_reason="stop", + stop_reason=output.stop_reason, + token_ids=(as_list(output.token_ids) if request.return_token_ids else None), + ) + else: + choice_data = ChatCompletionResponseChoice( + index=output.index, + message=ChatMessage(role=role, audio=audio_obj), + logprobs=None, + finish_reason="stop", + stop_reason=None, + ) choices.append(choice_data) return choices - def _create_image_choice(self, omni_outputs: OmniRequestOutput, role: str): + def _create_image_choice( + self, omni_outputs: OmniRequestOutput, role: str, request: ChatCompletionRequest, stream: bool = False + ): """Create chat completion response choices for image output. Converts image tensor or PIL Image output from diffusion models @@ -1152,9 +1915,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): result = await self._diffusion_engine.generate(**gen_kwargs) # Extract images from result # Handle nested OmniRequestOutput structure where images might be in request_output - images: list[Image.Image] = [] - if result.request_output["images"]: - images = result.request_output["images"] + images = getattr(result.request_output, "images", []) # Convert images to base64 content image_contents: list[dict[str, Any]] = [] diff --git a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni.py b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni.py index 504ca4486..f2da62b4a 100644 --- a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni.py +++ b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni.py @@ -449,10 +449,6 @@ class Qwen3OmniMoeForConditionalGeneration( ) ) - # # Remove EOS token if present - # if code[-1] == TALKER_CODEC_EOS_TOKEN_ID: - # code = code[:-1] - # Generate audio from codec codes audio_tensors = [] for code in codes: