diff --git a/recipes/JD/JoyAI-VL-Interaction.md b/recipes/JD/JoyAI-VL-Interaction.md index e09d5b16c..7efc45d38 100644 --- a/recipes/JD/JoyAI-VL-Interaction.md +++ b/recipes/JD/JoyAI-VL-Interaction.md @@ -219,3 +219,51 @@ the audio-track caveat. 5 chunks, `key_frames=0` = summarize all chunk frames) are aligned to the JoyVL reference adapter so per-tick behavior matches the released model; the framework only supplies the serving structure. + +## Native multi-stage pipeline (experimental) + +An experimental native speech path is available alongside the existing Day-0 serving path: + +```text +one request with image/video and text + -> JoyAI complete action text + -> Qwen3-TTS Talker + -> Code2Wav + -> 24 kHz audio +``` + +It runs behind one `vllm serve --omni` endpoint: + +```bash +vllm serve jdopensource/JoyAI-VL-Interaction-Preview --omni \ + --deploy-config vllm_omni/deploy/joyai_vl_interaction.yaml \ + --port 8091 +``` + +With `modalities: ["text", "audio"]`, its behavior is: + +| JoyAI action | Native output | +|---|---| +| ` ` | action text + speech for `` | +| `` | action text + empty audio output (zero samples) | +| ` ` | action text + speech for `` only | + +Native speech supports the Qwen3-TTS `CustomVoice` task. The request-level `voice` and +`language` fields select the speaker and language; when omitted, they default to +`Vivian` and `Auto`. Values supplied as `tts_speaker` and `tts_language` inside +`additional_information` take priority. + +This experimental native multi-stage pipeline is request-scoped, stateless, and +all-sync. It currently does not include: + +- continuous frame sessions, persistent standing instructions, or session memory; +- audio input / ASR, full-duplex interaction, or barge-in; +- background Agent execution for delegated questions; +- async-chunk support from Talker to Code2Wav; +- integration with the existing WebUI. + +For actions with spoken text, the Talker starts only after the complete JoyAI action is +available, and Code2Wav starts only after the Talker completes. Silence actions skip both +TTS stages and return an empty audio output with zero samples. All three stages are placed +on GPU 0 by the provided +[`deploy config`](../../vllm_omni/deploy/joyai_vl_interaction.yaml). diff --git a/tests/config/test_omni_config.py b/tests/config/test_omni_config.py index aed9c970d..0846a1ec6 100644 --- a/tests/config/test_omni_config.py +++ b/tests/config/test_omni_config.py @@ -532,6 +532,17 @@ def test_from_pipeline_config_dispatches_async_chunk_processors_without_mutating assert pipeline.get_stage(1).custom_process_input_func is None +def test_joyai_code2wav_waits_for_full_payload(): + config = _from_pipeline_key("joyai_vl_interaction") + talker = config.stage_by_id(1) + code2wav = config.stage_by_id(2) + + assert talker.custom_process_next_stage_input_func.endswith("talker2code2wav_full_payload") + assert code2wav.custom_process_input_func.endswith("talker2code2wav_token_only") + assert code2wav.connector_config.async_chunk is False + assert code2wav.model_config.requires_full_payload_input is True + + def test_vllm_omni_stage_config_public_fields_use_typed_stage_realizations(): assert not hasattr(BaseVllmOmniStageConfig, "from_stage_config") assert not hasattr(BaseVllmOmniStageConfig, "to_legacy_stage_config") diff --git a/tests/engine/test_orchestrator_error_handling.py b/tests/engine/test_orchestrator_error_handling.py index 90db55edb..eaa246072 100644 --- a/tests/engine/test_orchestrator_error_handling.py +++ b/tests/engine/test_orchestrator_error_handling.py @@ -16,7 +16,7 @@ from types import SimpleNamespace import janus import pytest -from vllm.sampling_params import SamplingParams +from vllm.sampling_params import RequestOutputKind, SamplingParams from vllm.v1.engine.exceptions import EngineDeadError from vllm.v1.serial_utils import MsgpackEncoder @@ -33,11 +33,16 @@ from vllm_omni.engine.orchestrator import ( cleanup_request_artifact_dirs, ) from vllm_omni.engine.stage_pool import StageUnavailableError +from vllm_omni.errors import OmniClientError from vllm_omni.inputs.data import OmniDiffusionSamplingParams +from vllm_omni.model_executor.stage_input_processors import ( + joyai_vl_interaction as joyai_bridge, +) from vllm_omni.outputs import OmniRequestOutput from .test_orchestrator import ( FakeOutputProcessor, + FakeRunningCounter, FakeStageClient, OrchestratorFixture, _build_harness, @@ -45,6 +50,7 @@ from .test_orchestrator import ( _build_stage_pools, _engine_core_outputs, _enqueue_add_request, + _get_output_message, _wait_for, ) @@ -935,6 +941,139 @@ async def test_diffusion_client_error_output_propagates_status_code(orchestrator orchestrator_fixture.thread.join(timeout=5) +@pytest.mark.asyncio +@pytest.mark.parametrize("event_driven", [False, True], ids=["legacy", "event-driven"]) +@pytest.mark.parametrize("first_speaker", ["Ryan", "not-a-real-speaker"]) +async def test_joyai_tts_request_validation_keeps_orchestrator_running( + orchestrator_factory, + monkeypatch: pytest.MonkeyPatch, + first_speaker: str, + event_driven: bool, +) -> None: + """Check that the same orchestrator can handle the next valid request.""" + monkeypatch.setenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", "1" if event_driven else "0") + + # Use a fixed token count so the test does not need tokenizer files. + class FakeTokenizer: + def encode(self, _text: str, *, add_special_tokens: bool) -> list[int]: + return list(range(12)) + + talker_model_config = SimpleNamespace( + max_model_len=64, + hf_config=SimpleNamespace( + talker_config=SimpleNamespace(spk_id={"vivian": 0, "ryan": 1}, codec_language_id={}, spk_is_dialect={}) + ), + ) + monkeypatch.setattr(joyai_bridge, "cached_tokenizer_from_config", lambda _: FakeTokenizer()) + + # Use the real bridge so speaker validation is part of the test. + class JoyAITalkerStage(FakeStageClient): + def process_engine_inputs(self, source_outputs, prompt=None, streaming_context=None): + self.bridge_inputs = joyai_bridge.joyai_action_to_tts( + source_outputs, prompt, self.requires_multimodal_data, target_model_config=talker_model_config + ) + return self.bridge_inputs + + stage0 = FakeStageClient(stage_type="llm", final_output=False) + stage1 = JoyAITalkerStage(stage_type="llm", final_output=True) + stage0_processor, stage1_processor = FakeOutputProcessor(), FakeOutputProcessor() + fixture = orchestrator_factory( + [stage0, stage1], + output_processors=[stage0_processor, stage1_processor], + stage_vllm_configs=[ + SimpleNamespace(model_config=SimpleNamespace(max_model_len=64)), + SimpleNamespace(model_config=talker_model_config), + ], + ) + counter = FakeRunningCounter() + fixture.orchestrator._running_counter = counter + # With a valid speaker, task_type should be changed from Base to CustomVoice. + # An unknown speaker should cause a request error. + first_info = {"tts_task_type": "Base", "speaker": "Vivian", "tts_speaker": first_speaker} + talker_submissions = 0 + # Send both requests to the same orchestrator. The second uses the default Vivian voice. + for index, (request_id, info) in enumerate((("first", first_info), ("good", {})), start=1): + await _enqueue_add_request( + fixture, + request_id=request_id, + prompt=SimpleNamespace(request_id=request_id, prompt_token_ids=[1, 2]), + original_prompt={"additional_information": info}, + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + ) + await _wait_for(lambda: len(stage0.add_request_calls) == index and counter.value == 1) + stage0_processor.request_outputs = [_build_request_output(request_id, text=" Hello.")] + stage0.push_engine_core_outputs(_engine_core_outputs(f"{request_id}-raw", float(index))) + + if request_id == "first" and first_speaker == "not-a-real-speaker": + # Reject this request before Talker runs, while keeping the engine alive. + error = await _wait_for_error_message(fixture, request_id=request_id) + assert error.status_code == 400 and error.error_type == "BadRequestError" + assert error.fatal is False and error.stage_id == 1 + assert "not-a-real-speaker" in error.error + assert stage1.add_request_calls == [] + else: + talker_submissions += 1 + await _wait_for(lambda: len(stage1.add_request_calls) == talker_submissions or fixture.result_future.done()) + assert not fixture.result_future.done() + assert stage1.add_request_calls[-1][0].request_id == request_id + metadata = stage1.bridge_inputs[0]["additional_information"] + assert metadata["task_type"] == ["CustomVoice"] + assert metadata["speaker"] == [first_speaker if request_id == "first" else "Vivian"] + # Finish the valid request and check its final output. + stage1_processor.request_outputs = [_build_request_output(request_id, text="Completed.")] + stage1.push_engine_core_outputs(_engine_core_outputs(f"{request_id}-talker", float(index))) + output = await _get_output_message(fixture) + assert output.request_id == request_id and output.stage_id == 1 + assert output.finished is True + assert output.engine_outputs.outputs[0].text == "Completed." + + # Wait for full cleanup before sending the next request. + await _wait_for(lambda: request_id not in fixture.orchestrator.request_states and counter.value == 0) + assert all(pool.get_bound_replica_id(request_id) is None for pool in fixture.orchestrator.stage_pools) + assert fixture.thread.is_alive() and not fixture.result_future.done() + + +@pytest.mark.asyncio +async def test_streaming_input_processor_client_error_does_not_forward_terminal_update() -> None: + """Do not send a second update after the first one fails and clears the request.""" + processor_calls = [] + + class FailingStage(FakeStageClient): + def process_engine_inputs(self, *args, **kwargs): + processor_calls.append((args, kwargs)) + raise OmniClientError("Input rejected") + + orchestrator, queues = _build_bare_orchestrator( + _build_stage_pools([[FakeStageClient()], [FailingStage(final_output=True)]]) + ) + state = OrchestratorRequestState( + request_id="bad-stream", + prompt={"prompt_token_ids": [1]}, + sampling_params_list=[SamplingParams(output_kind=RequestOutputKind.CUMULATIVE), _sampling_params()], + final_stage_id=1, + ) + state.streaming.enabled = True + assert state.duplex_identity is None + orchestrator.request_states[state.request_id] = state + try: + await orchestrator._route_output(0, 0, _build_request_output(state.request_id), state, stage_metrics=None) + + assert len(processor_calls) == 1 + error = queues[1].sync_q.get_nowait() + assert isinstance(error, ErrorMessage) + assert error.request_id == state.request_id + assert error.stage_id == 1 + assert error.status_code == 400 + assert error.fatal is False + assert queues[1].sync_q.empty() + assert state.request_id not in orchestrator.request_states + assert all(pool.get_bound_replica_id(state.request_id) is None for pool in orchestrator.stage_pools) + finally: + for q in queues: + q.close() + + @pytest.mark.asyncio async def test_duplex_input_processor_failure_is_request_scoped(orchestrator_factory, monkeypatch) -> None: class FailingStage(FakeStageClient): diff --git a/tests/model_executor/stage_input_processors/test_joyai_vl_interaction.py b/tests/model_executor/stage_input_processors/test_joyai_vl_interaction.py new file mode 100644 index 000000000..0822cfcd2 --- /dev/null +++ b/tests/model_executor/stage_input_processors/test_joyai_vl_interaction.py @@ -0,0 +1,98 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Tests for JoyAI action-to-TTS conversion.""" + +from types import SimpleNamespace + +import pytest + +from vllm_omni.engine.stage_engine_core_client import StageEngineCoreClient +from vllm_omni.model_executor.stage_input_processors import ( + joyai_vl_interaction as joyai_bridge, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def test_joyai_action_to_tts_bridge(monkeypatch: pytest.MonkeyPatch) -> None: + """Convert each JoyAI action through the real stage-client entry point.""" + target_model_config = SimpleNamespace( + hf_config=SimpleNamespace(talker_config=SimpleNamespace(spk_id={"vivian": 1, "ryan": 2})) + ) + prompt_length_calls: list[tuple[dict[str, list[str]], object]] = [] + prompt_lengths = { + "Colored squares are visible.": 12, + "I am checking this request now.": 18, + } + + def compute_prompt_length(metadata, model_config): + prompt_length_calls.append((metadata, model_config)) + return prompt_lengths[metadata["text"][0]] + + monkeypatch.setattr( + joyai_bridge, + "_compute_talker_prompt_length", + compute_prompt_length, + ) + + client = object.__new__(StageEngineCoreClient) + client.custom_process_input_func = joyai_bridge.joyai_action_to_tts + client.requires_multimodal_data = False + client.vllm_config = SimpleNamespace(model_config=target_model_config) + + cases = [ + ( + " Colored squares are visible.", + None, + { + "tts_task_type": "Base", + "language": ["Chinese"], + "speaker": ["Ryan"], + }, + ("Colored squares are visible.", "Chinese", "Ryan"), + ), + ("", None, {}, None), + ( + "", + ( + " I am checking this request now. " + " Why is the purple elephant dancing beside a volcano?" + ), + { + "language": "Chinese", + "speaker": "Ryan", + "tts_language": "English", + "tts_speaker": "Vivian", + }, + ("I am checking this request now.", "English", "Vivian"), + ), + ] + + for text, cumulative_text, additional_information, expected in cases: + completion = SimpleNamespace(text=text) + if cumulative_text is not None: + completion.cumulative_text = cumulative_text + + call_count = len(prompt_length_calls) + talker_inputs = client.process_engine_inputs( + [SimpleNamespace(outputs=[completion])], + prompt={"additional_information": additional_information}, + ) + + if expected is None: + assert talker_inputs == [] + assert len(prompt_length_calls) == call_count + continue + + spoken_text, language, speaker = expected + [talker_input] = talker_inputs + metadata = talker_input["additional_information"] + assert talker_input["prompt_token_ids"] == [0] * prompt_lengths[spoken_text] + assert metadata == { + "task_type": ["CustomVoice"], + "language": [language], + "speaker": [speaker], + "instruct": [""], + "text": [spoken_text], + } + assert prompt_length_calls[-1] == (metadata, target_model_config) diff --git a/vllm_omni/config/pipeline_registry.py b/vllm_omni/config/pipeline_registry.py index 7b14f49b1..5586b956e 100644 --- a/vllm_omni/config/pipeline_registry.py +++ b/vllm_omni/config/pipeline_registry.py @@ -67,6 +67,9 @@ from vllm_omni.model_executor.models.indextts2.pipeline import ( INDEXTTS2_PIPELINE, INDEXTTS25_PIPELINE, ) +from vllm_omni.model_executor.models.joyai_vl_interaction.pipeline import ( + JOYAI_VL_INTERACTION_PIPELINE, +) from vllm_omni.model_executor.models.lance.pipeline import LANCE_PIPELINE from vllm_omni.model_executor.models.lingbot_world.pipeline import LINGBOT_WORLD_PIPELINE from vllm_omni.model_executor.models.mammoth_moda2.pipeline import ( @@ -122,6 +125,7 @@ PipelineResolverFunc: TypeAlias = Callable[[PretrainedConfig | None], PipelineCo # --- Multi-stage omni pipelines (LLM-centric; audio / video I/O) --- OMNI_PIPELINES: dict[str, PipelineConfig | PipelineResolverFunc] = { "aura_omni": AURA_OMNI_PIPELINE, + "joyai_vl_interaction": JOYAI_VL_INTERACTION_PIPELINE, "qwen2_5_omni": QWEN2_5_OMNI_PIPELINE, "qwen2_5_omni_thinker_only": QWEN2_5_OMNI_THINKER_ONLY_PIPELINE, "personaplex": PERSONAPLEX_PIPELINE, diff --git a/vllm_omni/deploy/joyai_vl_interaction.yaml b/vllm_omni/deploy/joyai_vl_interaction.yaml new file mode 100644 index 000000000..59eaa8f6c --- /dev/null +++ b/vllm_omni/deploy/joyai_vl_interaction.yaml @@ -0,0 +1,79 @@ +# Single-GPU JoyAI-VL-Interaction -> Qwen3-TTS all-sync profile. +pipeline: joyai_vl_interaction +async_chunk: false +trust_remote_code: true + +connectors: + connector_of_shared_memory: + name: SharedMemoryConnector + extra: + codec_streaming: true + connector_get_sleep_s: 0.01 + connector_get_max_wait_first_chunk: 3000 + connector_get_max_wait: 300 + codec_chunk_frames: 25 + codec_left_context_frames: 72 + initial_codec_chunk_frames: 1 + # Qwen3-TTS Code2Wav buckets: 25, 72+1, 72+25, 72+72+25, and 300+25 frames. + decode_cudagraph_capture_sizes: [25, 73, 97, 169, 325] + +stages: + # GPU 0: JoyAI action model. + - stage_id: 0 + model: jdopensource/JoyAI-VL-Interaction-Preview + devices: "0" + gpu_memory_utilization: 0.5 + max_num_seqs: 4 + # Some OmniInteract video prompts contain more than 12K input tokens. + max_model_len: 16384 + max_num_batched_tokens: 16384 + limit_mm_per_prompt: {image: 256, video: 1} + enable_prefix_caching: true + default_sampling_params: + temperature: 0.8 + top_p: 0.9 + top_k: 40 + max_tokens: 256 + repetition_penalty: 1.1 + presence_penalty: 1.5 + + # GPU 0: Qwen3-TTS Talker. + - stage_id: 1 + model: Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice + devices: "0" + gpu_memory_utilization: 0.2 + max_num_seqs: 8 + max_model_len: 4096 + max_num_batched_tokens: 512 + enable_prefix_caching: false + output_connectors: + to_stage_2: connector_of_shared_memory + default_sampling_params: + temperature: 0.9 + top_k: 50 + min_tokens: 2 + max_tokens: 4096 + repetition_penalty: 1.05 + subtalker_sampling_params: + do_sample: true + temperature: 0.9 + top_k: 50 + top_p: 1.0 + + # GPU 0: Qwen3-TTS Code2Wav. + - stage_id: 2 + model: Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice + devices: "0" + gpu_memory_utilization: 0.2 + max_num_seqs: 8 + max_model_len: 65536 + max_num_batched_tokens: 65536 + enable_prefix_caching: false + input_connectors: + from_stage_1: connector_of_shared_memory + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 65536 + repetition_penalty: 1.0 diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index e86dc7673..299e1dd07 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -75,7 +75,7 @@ from vllm_omni.engine.messages import ( from vllm_omni.engine.orchestrator_monitor import create_orch_monitor, replica_key from vllm_omni.engine.serialization import serialize_additional_information from vllm_omni.engine.stage_pool import StagePool, StageUnavailableError -from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE +from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE, OmniClientError from vllm_omni.metrics import definitions as metric_defs from vllm_omni.metrics.prometheus import OmniRequestCounter from vllm_omni.metrics.stat_logger import OmniPrometheusStatLogger @@ -1647,9 +1647,11 @@ class Orchestrator: stage_id: int, error: str, *, + status_code: int = HTTPStatus.BAD_REQUEST.value, + error_type: str = DEFAULT_CLIENT_ERROR_TYPE, close_duplex_sessions: bool = False, ) -> None: - """Fail one request with a non-fatal 400 (bad input, engine survives). + """Fail one request with a non-fatal client error (400 by default). The non-fatal counterpart of `_fail_request_dead_stage`: emits a client-error ErrorMessage (default `fatal=False`) so the engine keeps @@ -1658,8 +1660,8 @@ class Orchestrator: await self.output_async_queue.put( ErrorMessage( error=error, - status_code=HTTPStatus.BAD_REQUEST.value, - error_type=DEFAULT_CLIENT_ERROR_TYPE, + status_code=status_code, + error_type=error_type, request_id=req_id, stage_id=stage_id, ) @@ -2073,6 +2075,8 @@ class Orchestrator: ) if ( req_state.streaming.enabled + # A failed forward may already have cleaned up this request. + and self.request_states.get(req_id) is req_state and finished and not final_only_finished and not self._is_duplex_session_request(req_state) @@ -2734,6 +2738,16 @@ class Orchestrator: req_state.prompt, streaming_context=req_state.streaming, ) + except OmniClientError as exc: + await self._fail_request_client_error( + req_id, + next_logical, + str(exc), + status_code=exc.status_code, + error_type=exc.error_type, + close_duplex_sessions=self._is_duplex_session_request(req_state), + ) + return except Exception as exc: logger.exception( "[Orchestrator] req=%s process_engine_inputs FAILED for stage-%s", diff --git a/vllm_omni/engine/stage_engine_core_client.py b/vllm_omni/engine/stage_engine_core_client.py index f96cbd1d3..87baaed5f 100644 --- a/vllm_omni/engine/stage_engine_core_client.py +++ b/vllm_omni/engine/stage_engine_core_client.py @@ -396,36 +396,54 @@ class StageEngineCoreClientBase(StageClientBase): and the original prompt. """ if self.custom_process_input_func is not None: - signature = inspect.signature(self.custom_process_input_func) - extra_kwargs: dict[str, Any] = {} - if "next_stage_hf_config" in signature.parameters: - # Let a processor size the next stage's prompt from that - # stage's model config (e.g. a talker whose engine positions - # must cover a speaker-prompt prefill). - extra_kwargs["next_stage_hf_config"] = self._stage_hf_config - # Match the context parameter by name, including the - # underscore-prefixed spelling some processors use (e.g. - # MiniCPM-o's ``llm2tts(..., _streaming_context)``), so bridge - # state keeps flowing to them. - if "streaming_context" in signature.parameters or "_streaming_context" in signature.parameters: - return self.custom_process_input_func( - source_outputs, - prompt, - self.requires_multimodal_data, - streaming_context, - **extra_kwargs, - ) - return self.custom_process_input_func( - source_outputs, - prompt, - self.requires_multimodal_data, - **extra_kwargs, - ) + return self._call_custom_process_input(source_outputs, prompt, streaming_context) if not self.engine_input_source: raise ValueError(f"engine_input_source empty for stage {self.stage_id}") return _default_process_engine_inputs(source_outputs, prompt, self.requires_multimodal_data) + def _call_custom_process_input( + self, + source_outputs: list[Any], + prompt: Any, + streaming_context: Any | None, + ) -> list[OmniTokensPrompt]: + """Call a stage input processor with its explicitly requested context.""" + processor = self.custom_process_input_func + assert processor is not None + signature = inspect.signature(processor) + extra_kwargs: dict[str, Any] = {} + if "next_stage_hf_config" in signature.parameters: + # Let a processor size the next stage's prompt from that + # stage's model config (e.g. a talker whose engine positions + # must cover a speaker-prompt prefill). + extra_kwargs["next_stage_hf_config"] = self._stage_hf_config + target_model_config = signature.parameters.get("target_model_config") + if target_model_config is not None: + # The JoyAI bridge needs the Talker tokenizer and model config + # to calculate the exact prompt length. + if target_model_config.kind is not inspect.Parameter.KEYWORD_ONLY: + raise TypeError("target_model_config must be a keyword-only parameter") + extra_kwargs["target_model_config"] = self.vllm_config.model_config + # Match the context parameter by name, including the + # underscore-prefixed spelling some processors use (e.g. + # MiniCPM-o's ``llm2tts(..., _streaming_context)``), so bridge + # state keeps flowing to them. + if "streaming_context" in signature.parameters or "_streaming_context" in signature.parameters: + return processor( + source_outputs, + prompt, + self.requires_multimodal_data, + streaming_context, + **extra_kwargs, + ) + return processor( + source_outputs, + prompt, + self.requires_multimodal_data, + **extra_kwargs, + ) + async def collective_rpc_async( self, method: str, diff --git a/vllm_omni/model_executor/models/joyai_vl_interaction/__init__.py b/vllm_omni/model_executor/models/joyai_vl_interaction/__init__.py new file mode 100644 index 000000000..b38884c62 --- /dev/null +++ b/vllm_omni/model_executor/models/joyai_vl_interaction/__init__.py @@ -0,0 +1,3 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""JoyAI-VL-Interaction native pipeline.""" diff --git a/vllm_omni/model_executor/models/joyai_vl_interaction/pipeline.py b/vllm_omni/model_executor/models/joyai_vl_interaction/pipeline.py new file mode 100644 index 000000000..478ce4a11 --- /dev/null +++ b/vllm_omni/model_executor/models/joyai_vl_interaction/pipeline.py @@ -0,0 +1,65 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""JoyAI-VL-Interaction -> Qwen3-TTS native all-sync pipeline.""" + +from vllm_omni.config.stage_config import ( + PipelineConfig, + StageExecutionType, + StagePipelineConfig, +) + +_JOYAI_INPUT_PROCESSOR = "vllm_omni.model_executor.stage_input_processors.joyai_vl_interaction" +_QWEN3_TTS_INPUT_PROCESSOR = "vllm_omni.model_executor.stage_input_processors.qwen3_tts" + + +JOYAI_VL_INTERACTION_PIPELINE = PipelineConfig( + model_type="joyai_vl_interaction", + model_arch="Qwen3VLForConditionalGeneration", + default_deploy_config_name="joyai_vl_interaction.yaml", + stages=( + StagePipelineConfig( + stage_id=0, + model_stage="joyvl", + execution_type=StageExecutionType.LLM_AR, + input_sources=(), + final_output=True, + final_output_type="text", + owns_tokenizer=True, + requires_multimodal_data=True, + engine_output_type="text", + sampling_constraints={ + "detokenize": True, + "skip_special_tokens": False, + }, + ), + StagePipelineConfig( + stage_id=1, + model_stage="qwen3_tts", + execution_type=StageExecutionType.LLM_AR, + input_sources=(0,), + owns_tokenizer=True, + engine_output_type="latent", + model_arch="Qwen3TTSTalkerForConditionalGeneration", + custom_process_input_func=f"{_JOYAI_INPUT_PROCESSOR}.joyai_action_to_tts", + custom_process_next_stage_input_func=f"{_QWEN3_TTS_INPUT_PROCESSOR}.talker2code2wav_full_payload", + sampling_constraints={ + "detokenize": False, + "stop_token_ids": [2150], + }, + ), + StagePipelineConfig( + stage_id=2, + model_stage="code2wav", + execution_type=StageExecutionType.LLM_GENERATION, + input_sources=(1,), + final_output=True, + final_output_type="audio", + engine_output_type="audio", + model_arch="Qwen3TTSCode2Wav", + sync_process_input_func=f"{_QWEN3_TTS_INPUT_PROCESSOR}.talker2code2wav_token_only", + requires_full_payload_input=True, + sampling_constraints={"detokenize": True}, + extras={"tts_args": {"max_instructions_length": 500}}, + ), + ), +) diff --git a/vllm_omni/model_executor/stage_input_processors/joyai_vl_interaction.py b/vllm_omni/model_executor/stage_input_processors/joyai_vl_interaction.py new file mode 100644 index 000000000..77b86691c --- /dev/null +++ b/vllm_omni/model_executor/stage_input_processors/joyai_vl_interaction.py @@ -0,0 +1,129 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Build Qwen3-TTS Talker inputs from completed JoyAI actions.""" + +from typing import Any + +from vllm.outputs import RequestOutput +from vllm.tokenizers import cached_tokenizer_from_config + +from vllm_omni.errors import OmniClientError +from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import ( + parse_action, +) +from vllm_omni.inputs.data import OmniTokensPrompt +from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import ( + Qwen3TTSPromptEmbedsBuilder, + first_value, +) + +# Match the standalone Qwen3-TTS default attributes +_DEFAULT_TTS_TASK_TYPE = "CustomVoice" +_DEFAULT_TTS_LANGUAGE = "Auto" +_DEFAULT_TTS_SPEAKER = "Vivian" + + +def _extract_completed_action_text(joyai_output: RequestOutput) -> str: + """Return the final JoyAI action text, using cumulative text when available.""" + completion = joyai_output.outputs[0] + action_text = getattr(completion, "cumulative_text", None) or completion.text + return action_text if isinstance(action_text, str) else "" + + +def _build_tts_metadata(request_prompt: object, spoken_text: str) -> dict[str, list[str]]: + """Build Qwen3-TTS metadata for the text JoyAI chose to speak.""" + prompt_dict = request_prompt if isinstance(request_prompt, dict) else {} + raw_additional_info = prompt_dict.get("additional_information") + additional_info = raw_additional_info if isinstance(raw_additional_info, dict) else {} + + # The native pipeline loads CustomVoice checkpoints for both TTS stages. + # Keep client metadata from selecting an incompatible Qwen3-TTS task. + task_type = _DEFAULT_TTS_TASK_TYPE + request_language = first_value(additional_info.get("language"), _DEFAULT_TTS_LANGUAGE) + language = first_value(additional_info.get("tts_language"), request_language) + request_speaker = first_value(additional_info.get("speaker"), _DEFAULT_TTS_SPEAKER) + speaker = first_value(additional_info.get("tts_speaker"), request_speaker) + instruction = first_value(additional_info.get("tts_instruct"), "") + + return { + "task_type": [str(task_type)], + "language": [str(language)], + "speaker": [str(speaker)], + "instruct": [str(instruction)], + "text": [spoken_text], + } + + +def _validate_talker_speaker(tts_metadata: dict[str, list[str]], talker_model_config: Any) -> None: + """Validate the selected voice using the destination Talker's speaker table.""" + talker_config = getattr(talker_model_config.hf_config, "talker_config", None) + if talker_config is None: + raise ValueError("The target stage is not a Qwen3-TTS Talker.") + + # Match CustomVoice prompt construction, including case and whitespace. + speaker = str(first_value(tts_metadata.get("speaker"), "")).lower().strip() + speakers = {name.lower() for name in (getattr(talker_config, "spk_id", None) or {})} + if not speaker or speaker not in speakers: + raise OmniClientError(f"Unsupported speaker: {speaker!r}") + + +def _compute_talker_prompt_length( + tts_metadata: dict[str, list[str]], + talker_model_config: Any, +) -> int: + """Compute the exact number of prompt positions used by the Talker.""" + task_type = str(first_value(tts_metadata.get("task_type"), _DEFAULT_TTS_TASK_TYPE)) + if task_type != _DEFAULT_TTS_TASK_TYPE: + raise ValueError("JoyAI native TTS supports only CustomVoice.") + + talker_tokenizer = cached_tokenizer_from_config(talker_model_config) + if talker_tokenizer is None: + raise ValueError("The Qwen3-TTS Talker must have an initialized tokenizer.") + + talker_config = getattr(talker_model_config.hf_config, "talker_config", None) + if talker_config is None: + raise ValueError("The target stage is not a Qwen3-TTS Talker.") + + def tokenize_prompt(text: str) -> list[int]: + return talker_tokenizer.encode(text, add_special_tokens=True) + + return Qwen3TTSPromptEmbedsBuilder.estimate_prompt_len_from_additional_information( + additional_information=tts_metadata, + task_type=task_type, + tokenize_prompt=tokenize_prompt, + codec_language_id=getattr(talker_config, "codec_language_id", None), + spk_is_dialect=getattr(talker_config, "spk_is_dialect", None), + ) + + +def joyai_action_to_tts( + source_outputs: list[RequestOutput], + prompt: object | None = None, + requires_multimodal_data: bool = False, + *, + target_model_config: Any, +) -> list[OmniTokensPrompt]: + """Build Talker inputs for JoyAI actions that contain text to speak.""" + del requires_multimodal_data + + request_prompts = prompt if isinstance(prompt, list) else [prompt] * len(source_outputs) + talker_inputs: list[OmniTokensPrompt] = [] + for request_index, joyai_output in enumerate(source_outputs): + parsed_action = parse_action(_extract_completed_action_text(joyai_output)) + if not parsed_action.spoke or not parsed_action.text: + continue + + request_prompt = request_prompts[request_index] if request_index < len(request_prompts) else None + tts_metadata = _build_tts_metadata(request_prompt, parsed_action.text) + _validate_talker_speaker(tts_metadata, target_model_config) + talker_prompt_length = _compute_talker_prompt_length(tts_metadata, target_model_config) + talker_inputs.append( + OmniTokensPrompt( + # OmniTokensPrompt reserves vllm scheduler positions with zero IDs; + # the Talker builds the actual prompt from tts_metadata. + prompt_token_ids=[0] * talker_prompt_length, + additional_information=tts_metadata, + ) + ) + + return talker_inputs