diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml index 2d3be8bc4..0aed5f492 100644 --- a/.buildkite/cuda/test-merge.yml +++ b/.buildkite/cuda/test-merge.yml @@ -169,7 +169,6 @@ steps: - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py - vllm_omni/deploy/minicpmo_4_5.yaml - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py - - vllm_omni/deploy/minicpmo_4_5_duplex.yaml - vllm_omni/experimental/fullduplex/ - vllm_omni/model_executor/models/minicpmo_4_5/ - vllm_omni/model_executor/models/step_audio2/ diff --git a/.buildkite/cuda/test-ready.yml b/.buildkite/cuda/test-ready.yml index 1b9afdd0f..7b8dd77ec 100644 --- a/.buildkite/cuda/test-ready.yml +++ b/.buildkite/cuda/test-ready.yml @@ -208,7 +208,6 @@ steps: - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py - vllm_omni/deploy/minicpmo_4_5.yaml - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py - - vllm_omni/deploy/minicpmo_4_5_duplex.yaml - vllm_omni/experimental/fullduplex/ - vllm_omni/model_executor/models/minicpmo_4_5/ - vllm_omni/model_executor/models/step_audio2/ diff --git a/docs/configuration/stage_configs.md b/docs/configuration/stage_configs.md index 1afa16d54..1deb7946c 100644 --- a/docs/configuration/stage_configs.md +++ b/docs/configuration/stage_configs.md @@ -14,7 +14,7 @@ describe fixed model topology and are not accepted in deploy YAMLs. Common `PipelineConfig` fields include: | Field | Description | -|-------|-------------| +| ------- | ------------- | | `model_type` | Pipeline identifier used during model and config resolution. | | `default_deploy_config_name` | Bundled deploy YAML loaded when the user does not pass `deploy_config`. | | `model_arch` | Default Hugging Face architecture for the pipeline. | @@ -26,7 +26,7 @@ Common `PipelineConfig` fields include: Common `StagePipelineConfig` fields include: | Field | Description | -|-------|-------------| +| ------- | ------------- | | `stage_id` | Stable stage identifier. | | `model_stage` | Logical stage name used by runtime and strategy resolution. | | `execution_type` | `LLM_AR`, `LLM_GENERATION`, or `DIFFUSION`. | @@ -49,10 +49,10 @@ deployment overrides. The new deploy schema lives under `vllm_omni/deploy/` and is paired with a frozen `PipelineConfig` registered by the model's `pipeline.py`. Each deploy YAML has these top-level fields: | Field | Type | Required | Default | Description | -|-------|------|----------|---------|-------------| +| ------- | ------ | ---------- | --------- | ------------- | | `base_config` | str (path) | optional | — | Overlay parent (relative or absolute). `stages:` / `platforms:` deep-merged by stage_id; other scalars overlay-wins. Intended for user-authored overlays; prod yamls stay flat. | | `async_chunk` | bool | optional | `true` | Enable chunked streaming between stages. Pin to `false` if the pipeline runs end-to-end. | -| `session_mode` | str | optional | `"turn"` | Session behavior. Use `"duplex"` only with a pipeline that enables duplex control. | +| `session_mode` | str | optional | `"turn"` | Session behavior. MiniCPM-o 4.5 deploy YAMLs set `"duplex"` so `/v1/realtime` and chat share the same profile. | | `active_stream_window` | int | optional | `0` | Number of active downstream stream slots; `0` preserves all-stream cycling. | | `duplex_session` | dict | optional | runtime defaults | Full-duplex session lifecycle, buffering, replay, and capacity limits. | | `connectors` | dict | optional | `null` | Named connector specs (`{name, extra}`). Referenced by each stage's `input_connectors` / `output_connectors`. See [Connector schema](#connector-schema). | @@ -84,7 +84,7 @@ subprocess on one GPU. `ray` / `external_launcher` are not fully supported yet. Each entry under `stages:` accepts any `StageDeployConfig` field directly (no nested `engine_args:`). Only fields whose value legitimately varies across stages live here; pipeline-wide settings (trust_remote_code, distributed_executor_backend, dtype, quantization, prefix/chunked prefill, DP/PP sizes) are declared at the top level and applied to every stage. Unknown keys fall through to `engine_extras:` and are forwarded to the engine. Frequently used fields are listed below; the source-of-truth schema is `StageDeployConfig` in `vllm_omni/config/stage_config.py`. | Field | Type | Required | Default | Description | -|-------|------|----------|---------|-------------| +| ------- | ------ | ---------- | --------- | ------------- | | `stage_id` | int | required | — | Stage identity; matched against `PipelineConfig.stages[*].stage_id`. | | `max_num_seqs` | int \| null | optional | `null` | Max concurrent sequences per stage. | | `gpu_memory_utilization` | float \| null | optional | `null` | Per-stage total memory target; used for automatic KV-cache sizing. | @@ -113,9 +113,9 @@ connectors: # Additional connector-specific options ``` -| Connector class | Use case | `extra` keys | -|-----------------|----------|--------------| -| `SharedMemoryConnector` | Same-host KV transfer between stages (default for bundled YAMLs). | None. All payloads use shared memory. | +| Connector class | Use case | `extra` keys | +|--------------------------|-----------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------| +| `SharedMemoryConnector` | Same-host KV transfer between stages (default for bundled YAMLs). | None. All payloads use shared memory. | | `MooncakeStoreConnector` | Cross-host KV transfer over TCP. Required for multi-node deployments. | `host`, `metadata_server`, `master`, `segment` (int bytes), `localbuf` (int bytes), `proto` (`"tcp"` / `"rdma"`). | A stage references a connector by name in its `input_connectors` / `output_connectors`: @@ -135,7 +135,7 @@ stages: ### CLI flags | Flag | Description | -|------|-------------| +| ------ | ------------- | | `--deploy-config PATH` | Load a deploy YAML. **Optional** — when omitted, the bundled `vllm_omni/deploy/.yaml` is auto-loaded by the model registry. | | `--stage-overrides JSON` | Per-stage JSON overrides, e.g. `'{"0":{"gpu_memory_utilization":0.5}}'`. Per-stage values always win over global flags. | | `--async-chunk` / `--no-async-chunk` | Flip the deploy YAML's `async_chunk:` bool. Unset (default) leaves the YAML value in force. | @@ -249,7 +249,7 @@ CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct --omni \ Effective config per stage after the merge: | Stage | Field | Final value | Source | -|-------|-------|-------------|--------| +| ------- | ------- | ------------- | -------- | | 0 | `gpu_memory_utilization` | `0.9` | base YAML (overlay didn't touch stage 0) | | 0 | `max_num_seqs` | `8` | per-stage CLI (`--stage-overrides`) — wins over base `32` | | 0 | `max_model_len` | `16384` | global CLI | @@ -273,12 +273,14 @@ when loading a deploy configuration. Examples: For offline inference (assuming the necessary dependencies have been imported): + ```python model_name = "Qwen/Qwen2.5-Omni-7B" omni = Omni(model=model_name, deploy_config="/path/to/deploy_config.yaml") ``` For online serving: + ```bash vllm serve Qwen/Qwen2.5-Omni-7B --omni --port 8091 --deploy-config /path/to/deploy_config.yaml ``` diff --git a/examples/online_serving/minicpmo/README.md b/examples/online_serving/minicpmo/README.md index 8db5630ba..1212f0ff7 100644 --- a/examples/online_serving/minicpmo/README.md +++ b/examples/online_serving/minicpmo/README.md @@ -32,12 +32,14 @@ GPU 0 (90%) and colocates the Talker (55%) and Code2Wav (35%) on GPU 1. That profile admits at most four concurrent sequences per stage. | deploy config | GPUs | Notes | -|---|---|---| +| --- | --- | --- | | `minicpmo_4_5.yaml` (default) | 1 | Memory-constrained compatibility layout. | | `minicpmo_4_5_2gpu.yaml` | 2 | Recommended continuous-batching layout; Talker and Code2Wav share GPU 1. | | `minicpmo_4_5_3gpu.yaml` | 3 | One GPU per stage. | | `minicpmo_4_5_8x4090.yaml` | 8 | Full 8x4090 layout. | -| `minicpmo_4_5_duplex.yaml` | 1 | Experimental native full-duplex overlay. | + +Every profile sets `session_mode: duplex`, so native full-duplex is served +from the same process as `/v1/chat/completions`. The split pipeline preserves native-duplex epoch/turn identity, segment text, turn completion, reference voice, and terminal-audio metadata through @@ -55,8 +57,8 @@ vllm-omni serve openbmb/MiniCPM-o-4_5 \ ``` For local ModelScope checkpoints, replace `openbmb/MiniCPM-o-4_5` with the -checkpoint path. To start the experimental native duplex backend, use -`vllm_omni/deploy/minicpmo_4_5_duplex.yaml`. +checkpoint path. For native full-duplex, connect `/v1/realtime?duplex=1` on +this same server. ### Per-stage overrides @@ -145,7 +147,7 @@ assistant template, so it is not an apples-to-apples accuracy run. ## Run the Realtime duplex CLI demo -After the duplex backend is running, stream one WAV through the Realtime +After the server is running, stream one WAV through the Realtime WebSocket endpoint: ```bash diff --git a/tests/config/test_config_factory.py b/tests/config/test_config_factory.py index e5b6d607a..5f7359ddc 100644 --- a/tests/config/test_config_factory.py +++ b/tests/config/test_config_factory.py @@ -1109,8 +1109,35 @@ class TestDeployConfigLoading: with pytest.raises(ValueError, match=r"stage_args.*PipelineConfig.*stages"): load_deploy_config(deploy_path) - def test_load_minicpmo_duplex_deploy_config(self): - deploy_path = Path(get_deploy_config_path("minicpmo_4_5_duplex.yaml")) + @pytest.mark.parametrize( + ("filename", "max_sessions"), + [ + ("minicpmo_4_5.yaml", 4), + ("minicpmo_4_5_2gpu.yaml", 4), + ("minicpmo_4_5_3gpu.yaml", 4), + ("minicpmo_4_5_8x4090.yaml", 1), + ("minicpmo_4_5_3gpu_stage1_replicas.yaml", 4), + ("minicpmo_4_5_4gpu_stage1_replicas.yaml", 4), + ("minicpmo_4_5_8x4090_stage1_replicas.yaml", 1), + ], + ) + def test_minicpmo_deploy_configs_enable_duplex(self, filename: str, max_sessions: int): + """Every MiniCPM-o profile serves /v1/realtime, so the retired + minicpmo_4_5_duplex.yaml overlay has no replacement to fall back on. + """ + deploy = load_deploy_config(Path(get_deploy_config_path(filename))) + pipeline = resolve_pipeline_config("minicpmo_4_5") + assert isinstance(pipeline, PipelineConfig) + stages = merge_pipeline_deploy(pipeline, deploy) + + assert deploy.session_mode == "duplex" + assert deploy.active_stream_window == 1 + assert deploy.duplex_session.max_sessions == max_sessions + assert [stage.session_mode for stage in stages] == ["duplex", "duplex", "duplex"] + assert [stage.to_omegaconf().session_mode for stage in stages] == ["duplex", "duplex", "duplex"] + + def test_load_minicpmo_default_deploy_config(self): + deploy_path = Path(get_deploy_config_path("minicpmo_4_5.yaml")) deploy = load_deploy_config(deploy_path) pipeline = resolve_pipeline_config("minicpmo_4_5") @@ -1118,17 +1145,15 @@ class TestDeployConfigLoading: stages = merge_pipeline_deploy(pipeline, deploy) - assert deploy.session_mode == "duplex" assert deploy.async_chunk is True - assert deploy.active_stream_window == 1 - assert [stage.session_mode for stage in stages] == ["duplex", "duplex", "duplex"] - assert [stage.to_omegaconf().session_mode for stage in stages] == ["duplex", "duplex", "duplex"] - assert [stage.yaml_engine_args["async_scheduling"] for stage in stages] == [False, False, False] - assert all("Async" not in (stage.scheduler_cls or "") for stage in stages) + assert stages[0].yaml_engine_args["async_scheduling"] is True + assert stages[1].yaml_engine_args["async_scheduling"] is True + assert "Async" in (stages[0].scheduler_cls or "") + assert "Async" in (stages[1].scheduler_cls or "") assert [stage.devices for stage in deploy.stages] == ["0", "0", "0"] assert deploy.stages[1].enforce_eager is False assert stages[1].yaml_extras["default_sampling_params"]["max_tokens"] == 4096 - assert stages[1].yaml_extras["default_sampling_params"]["min_tokens"] == 0 + assert stages[1].yaml_extras["default_sampling_params"]["min_tokens"] == 50 assert stages[1].yaml_extras["default_sampling_params"]["temperature"] == 0.8 assert stages[1].yaml_extras["default_sampling_params"]["stop_token_ids"] == [6561] assert "codec_sampling_params" not in stages[1].yaml_engine_args @@ -2116,7 +2141,8 @@ class TestBaseConfigInheritance: def test_minicpmo_overlays_inherit_talker_sampling_params(self): """Overlays must keep the base Talker codec Sampler knobs.""" for filename in ( - "minicpmo_4_5_duplex.yaml", + "minicpmo_4_5_2gpu.yaml", + "minicpmo_4_5_3gpu.yaml", "minicpmo_4_5_3gpu_stage1_replicas.yaml", "minicpmo_4_5_4gpu_stage1_replicas.yaml", "minicpmo_4_5_8x4090_stage1_replicas.yaml", @@ -2128,9 +2154,6 @@ class TestBaseConfigInheritance: sampling = deploy.stages[1].default_sampling_params assert sampling is not None, f"{filename} stage 1 lost default_sampling_params" assert sampling["temperature"] == 0.8, filename - # Upstream utils.TTSSamplingParams. min_tokens is not checked here: - # the duplex overlay drops it to 0 because the model budgets each - # generate_chunk itself. assert sampling["top_k"] == 25, filename assert sampling["top_p"] == 0.85, filename assert sampling["repetition_penalty"] == 1.05, filename diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py index ba0b75dca..fb4188ced 100644 --- a/tests/core/sched/test_omni_ar_scheduler_streaming.py +++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py @@ -252,6 +252,37 @@ def test_running_decode_step_without_inter_stage_payload_does_not_raise() -> Non sched.chunk_transfer_adapter.save_async.assert_not_called() +def test_queued_streaming_update_on_async_stop_fences_in_flight_once() -> None: + """Native duplex + async: the next unit is usually already queued when the + current one stops, so _handle_stopped_request applies the update and the + stop site below it fences the same in-flight decode a second time. An + accumulated residue outlives the drain and swallows the next unit's + listen/speak, which the client sees as silence until its timeout. + """ + session = _make_request() + session.status = RequestStatus.RUNNING + session.resumable = True + session.append_output_token_ids([7]) + session.num_computed_tokens = 4 + session.num_output_placeholders = 1 + # This step's frame (settled to 0 by update_from_output) plus one extra + # async decode still in flight. + session.num_in_flight_tokens = 2 + + sched = _make_scheduler(stage_id=0) + sched._enqueue_waiting_request = MagicMock() + + def handle_stopped(request: Request) -> bool: + sched._update_request_as_session(request, _make_update([10, 20])) + return False + + _run_resumable_segment_stop(session, handle_stopped=handle_stopped) + + # Exactly the one unreported decode, so the drain reaches zero before the + # new segment's first frame arrives. + assert session.num_stale_output_tokens == session.num_in_flight_tokens == 1 + + def test_stage0_streaming_update_discards_outstanding_async_placeholder_token() -> None: sched = _make_scheduler(stage_id=0) session = _make_request() diff --git a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json index 2038e45e0..073c0c2da 100644 --- a/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json +++ b/tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json @@ -18,7 +18,7 @@ "model": "openbmb/MiniCPM-o-4_5", "extra_cli_args": [ "--deploy-config", - "vllm_omni/deploy/minicpmo_4_5_duplex.yaml", + "vllm_omni/deploy/minicpmo_4_5.yaml", "--trust-remote-code" ] }, diff --git a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py index 506b39c82..988f9c582 100644 --- a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py +++ b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py @@ -15,20 +15,10 @@ import pytest from huggingface_hub import snapshot_download from tests.helpers.runtime import OmniServerParams, get_model_prefix -from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config +from tests.helpers.stage_config import get_deploy_config_path MODEL = "openbmb/MiniCPM-o-4_5" -DEPLOY_CONFIG = modify_stage_config( - get_deploy_config_path("minicpmo_4_5_duplex.yaml"), - updates={ - "base_config": get_deploy_config_path("minicpmo_4_5.yaml"), - # Talker context is 4096 (tts_config.max_position_embeddings); KV sizing - # is left automatic so duplex matches the simplex deploy profiles. - "stages": { - 1: {"max_model_len": 4096}, - }, - }, -) +DEPLOY_CONFIG = get_deploy_config_path("minicpmo_4_5.yaml") ASSET_DIR = Path(__file__).resolve().parents[3] / "assets" / "minicpmo_4_5" RESPONSE_REQUIRED_WAV = ASSET_DIR / "response_required_16k.wav" RESPONSE_REQUIRED_SHA256 = "2e5fd4eb3ee434ce107ee3a0591fa624a33f7683c7462f45fe651c443c9af941" diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index a37d9d2b1..d45cac20c 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + from __future__ import annotations import asyncio @@ -105,16 +108,16 @@ class FakeEngineClient: self.close_result = close_result self.opened: list[str] = [] self.opened_fences: list[DuplexFence | None] = [] - self.appended: list[tuple[str, str, object, bool]] = [] + self.appended: list[tuple[str, str, Any, bool]] = [] self.append_operation_ids: list[str | None] = [] self.appended_fences: list[DuplexFence | None] = [] - self.opened_configs: list[dict[str, object]] = [] - self.opened_runtime_configs: list[dict[str, object]] = [] + self.opened_configs: list[dict[str, Any]] = [] + self.opened_runtime_configs: list[dict[str, Any]] = [] self.signals: list[tuple[str, str]] = [] self.signal_fences: list[DuplexFence | None] = [] self.signal_next_fences: list[DuplexFence | None] = [] - self.signal_session_configs: list[dict[str, object] | None] = [] - self.signal_runtime_configs: list[dict[str, object] | None] = [] + self.signal_session_configs: list[dict[str, Any] | None] = [] + self.signal_runtime_configs: list[dict[str, Any] | None] = [] self.closed: list[tuple[str, str]] = [] self.touched: list[tuple[str, DuplexLeaseActivity]] = [] self.touch_fences: list[DuplexFence] = [] @@ -136,7 +139,7 @@ class FakeEngineClient: runtime_config: dict[str, object] | None = None, timeout: float | None = None, fence: DuplexFence | None = None, - ) -> None: + ) -> Any: del timeout if self.fail_open: raise RuntimeError("open failed") @@ -157,7 +160,7 @@ class FakeEngineClient: timeout: float | None = None, collect_outputs: bool = True, fence: DuplexFence | None = None, - ) -> None: + ) -> Any: del timeout, collect_outputs self.appended.append((session_id, mode, payload, final)) self.append_operation_ids.append(operation_id) @@ -191,7 +194,7 @@ class FakeEngineClient: next_fence: DuplexFence | None = None, session_config: dict[str, object] | None = None, runtime_config: dict[str, object] | None = None, - ) -> None: + ) -> Any: del timeout if self.fail_signal or event in self.fail_signal_events: raise RuntimeError("signal failed") @@ -209,7 +212,7 @@ class FakeEngineClient: reason: str = "client_close", timeout: float | None = None, fence: DuplexFence | None = None, - ) -> None: + ) -> Any: del timeout if self.fail_close: raise RuntimeError("close failed") @@ -344,7 +347,7 @@ def _project_data_plane( result: object, *, session: DuplexSession | None = None, -) -> list[dict[str, object]]: +) -> list[dict[str, Any]]: return list(data_plane.project(result, context=_data_plane_context(session))) @@ -4194,6 +4197,7 @@ async def test_realtime_resume_preserves_append_tail_order_across_connections(): await asyncio.sleep(0.01) else: pytest.fail("resumable session did not open") + assert created is not None session = handler._registry.get("sid-resume-append-order") assert session is not None native_state = handler._minicpmo_session_state(session) @@ -4323,6 +4327,7 @@ async def test_realtime_takeover_does_not_clear_new_attachment_pending_turns(): await asyncio.sleep(0.001) else: pytest.fail("first attachment did not create the session") + assert created is not None second = TimedWebSocket(receive_timeout_s=1.0) second.put( @@ -4386,6 +4391,7 @@ async def test_realtime_takeover_drops_input_queued_by_replaced_attachment(): await asyncio.sleep(0.001) else: pytest.fail("first attachment did not create the session") + assert created is not None second = TimedWebSocket(receive_timeout_s=1.0) second.put( @@ -5002,11 +5008,12 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(): *, mode: str, payload: object, + operation_id: str | None = None, final: bool = False, timeout: float | None = None, collect_outputs: bool = True, fence: DuplexFence | None = None, - ): + ) -> Any: self.append_started.set() try: await self.release_append.wait() @@ -5017,6 +5024,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(): session_id, mode=mode, payload=payload, + operation_id=operation_id, final=final, timeout=timeout, collect_outputs=collect_outputs, @@ -5128,7 +5136,9 @@ async def test_duplex_handler_session_update_control_failure_rolls_back_config() error = next(m for m in ws.sent if m.get("type") == "error") assert error["code"] == "runtime_signal_failed" assert error["runtime_control"]["error_count"] == 1 - assert engine.signal_session_configs[-1]["instructions"] == "must roll back" + signaled_config = engine.signal_session_configs[-1] + assert signaled_config is not None + assert signaled_config["instructions"] == "must roll back" @pytest.mark.asyncio @@ -5429,7 +5439,7 @@ async def test_minicpmo_native_duplex_open_unsupported_fails_session_create(): @pytest.mark.asyncio async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract(): class LegacyEngineClient(FakeEngineClient): - async def open_duplex_session_async( + async def open_duplex_session_async( # type: ignore[override] self, session_id: str, *, @@ -5437,7 +5447,7 @@ async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract(): capabilities: dict[str, object] | None = None, session_config: dict[str, object] | None = None, timeout: float | None = None, - ) -> None: + ) -> Any: return await super().open_duplex_session_async( session_id, session_mode=session_mode, @@ -5534,6 +5544,9 @@ async def test_minicpmo_native_duplex_separates_public_and_runtime_config(monkey assert opened_runtime_config["duplex_stage_max_tokens"] == {"0": 20, "1": 8192} assert opened_runtime_config["duplex_stage_sampling_params"]["0"]["top_k"] == 20 assert opened_runtime_config["duplex_stage_sampling_params"]["0"]["top_p"] == 0.8 + # The Talker YAML carries upstream's min_new_token=50 for chat TTS; a duplex + # chunk is 26 codec samples, so check_stop would never release the request. + assert opened_runtime_config["duplex_stage_sampling_params"]["1"] == {"min_tokens": 0} assert base64.b64decode(opened_runtime_config["ref_audio_data"]) == struct.pack("<1600f", *([0.25] * 1600)) diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py index 4a76c8bc5..64bd94916 100644 --- a/vllm_omni/core/sched/omni_ar_scheduler.py +++ b/vllm_omni/core/sched/omni_ar_scheduler.py @@ -295,7 +295,9 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): # them. Upstream vllm raises RuntimeError on this status; omni allows # async abort (e.g. client disconnect during TTS streaming) to leave # requests in the waiting/running queues temporarily. - for queue in (self.waiting, self.running): + waiting = getattr(self, "waiting") + running = getattr(self, "running") + for queue in (waiting, running): for req in list(queue): if getattr(req, "status", None) == RequestStatus.FINISHED_ABORTED: queue.remove(req) @@ -303,7 +305,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): original_waiting = None if self._should_defer_waiting_admission(): - original_waiting = self.waiting + original_waiting = waiting self.waiting = create_request_queue(self.policy) try: @@ -579,8 +581,12 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): # valid new-segment frames or underflowed the drain # assert whenever placeholder counts diverged from # scheduled counts (spec drafts, in-flight prefill). + # Assign, never accumulate: _handle_stopped_request above + # may already have fenced the same in-flight tokens through + # a queued streaming update, and adding twice swallows the + # next duplex unit's listen/speak under async scheduling. if request.num_in_flight_tokens > 0: - request.num_stale_output_tokens += request.num_in_flight_tokens + request.num_stale_output_tokens = request.num_in_flight_tokens if outstanding_async_tokens > 0: # Discard only outputs that are already in flight and # roll back their optimistic computed-token accounting. @@ -716,10 +722,12 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): # pre-replacement frame will drain, so the counter reaches exactly # zero and the new segment's frames are never swallowed. This also # covers an in-flight prefill chunk, which carries no placeholders - # but must still have its late output dropped. + # but must still have its late output dropped. Assign, never + # accumulate: update_from_output fences the same in-flight tokens when + # a resumable stop applies a queued update through this helper. in_flight_tokens = int(getattr(session, "num_in_flight_tokens", 0) or 0) if in_flight_tokens > 0: - session.num_stale_output_tokens += in_flight_tokens + session.num_stale_output_tokens = in_flight_tokens if outstanding_async_tokens > 0: # Async scheduling may already have sampled the previous # segment's next token. Drop that late token instead of diff --git a/vllm_omni/deploy/minicpmo_4_5.yaml b/vllm_omni/deploy/minicpmo_4_5.yaml index 973c4291e..9e5a2dec9 100644 --- a/vllm_omni/deploy/minicpmo_4_5.yaml +++ b/vllm_omni/deploy/minicpmo_4_5.yaml @@ -1,6 +1,13 @@ # MiniCPM-o 4.5 three-stage deploy on one large-memory GPU. pipeline: minicpmo_4_5 async_chunk: true +# Native full-duplex shares this profile with /v1/chat/completions. Every other +# duplex_session knob keeps its runtime default. +session_mode: duplex +active_stream_window: 1 +duplex_session: + # Matches Stage 0/1 max_num_seqs below. + max_sessions: 4 trust_remote_code: true enable_prefix_caching: false diff --git a/vllm_omni/deploy/minicpmo_4_5_2gpu.yaml b/vllm_omni/deploy/minicpmo_4_5_2gpu.yaml index d40715dbe..9cc6fd302 100644 --- a/vllm_omni/deploy/minicpmo_4_5_2gpu.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_2gpu.yaml @@ -4,6 +4,13 @@ # isolated from codec/vocoder contention while preserving stage-2 batching. pipeline: minicpmo_4_5 async_chunk: true +# Native full-duplex shares this profile with /v1/chat/completions. Every other +# duplex_session knob keeps its runtime default. +session_mode: duplex +active_stream_window: 1 +duplex_session: + # Matches Stage 0/1 max_num_seqs below. + max_sessions: 4 trust_remote_code: true enable_prefix_caching: false diff --git a/vllm_omni/deploy/minicpmo_4_5_3gpu.yaml b/vllm_omni/deploy/minicpmo_4_5_3gpu.yaml index 26c2aeaa8..c179824a2 100644 --- a/vllm_omni/deploy/minicpmo_4_5_3gpu.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_3gpu.yaml @@ -4,6 +4,13 @@ # Code2Wav run independently on GPUs 0, 1, and 2. pipeline: minicpmo_4_5 async_chunk: true +# Native full-duplex shares this profile with /v1/chat/completions. Every other +# duplex_session knob keeps its runtime default. +session_mode: duplex +active_stream_window: 1 +duplex_session: + # Matches Stage 0/1 max_num_seqs below. + max_sessions: 4 trust_remote_code: true enable_prefix_caching: false diff --git a/vllm_omni/deploy/minicpmo_4_5_3gpu_stage1_replicas.yaml b/vllm_omni/deploy/minicpmo_4_5_3gpu_stage1_replicas.yaml index 7a0513167..806f3330e 100644 --- a/vllm_omni/deploy/minicpmo_4_5_3gpu_stage1_replicas.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_3gpu_stage1_replicas.yaml @@ -5,7 +5,6 @@ # Stage 1: replicas on GPU 1 and 2 base_config: minicpmo_4_5.yaml pipeline: minicpmo_4_5 -session_mode: duplex stages: - stage_id: 0 diff --git a/vllm_omni/deploy/minicpmo_4_5_4gpu_stage1_replicas.yaml b/vllm_omni/deploy/minicpmo_4_5_4gpu_stage1_replicas.yaml index da28456fe..16d9f79da 100644 --- a/vllm_omni/deploy/minicpmo_4_5_4gpu_stage1_replicas.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_4gpu_stage1_replicas.yaml @@ -5,7 +5,6 @@ # Stage 1: replicas on GPU 1, 2, and 3 base_config: minicpmo_4_5.yaml pipeline: minicpmo_4_5 -session_mode: duplex stages: - stage_id: 0 diff --git a/vllm_omni/deploy/minicpmo_4_5_8x4090.yaml b/vllm_omni/deploy/minicpmo_4_5_8x4090.yaml index b7a7bc3f5..eacc069d1 100644 --- a/vllm_omni/deploy/minicpmo_4_5_8x4090.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_8x4090.yaml @@ -15,6 +15,10 @@ # so 4096 is the empirically-verified safe ceiling. pipeline: minicpmo_4_5 async_chunk: true +# Native full-duplex shares this profile with /v1/chat/completions. The default +# duplex_session caps sessions at 1, matching Stage 0/1 max_num_seqs below. +session_mode: duplex +active_stream_window: 1 trust_remote_code: true enable_prefix_caching: false diff --git a/vllm_omni/deploy/minicpmo_4_5_8x4090_stage1_replicas.yaml b/vllm_omni/deploy/minicpmo_4_5_8x4090_stage1_replicas.yaml index f4ad2c6b4..f377ac040 100644 --- a/vllm_omni/deploy/minicpmo_4_5_8x4090_stage1_replicas.yaml +++ b/vllm_omni/deploy/minicpmo_4_5_8x4090_stage1_replicas.yaml @@ -5,7 +5,6 @@ # Stage 1: replicas on GPU 4, 5, 6, and 7 base_config: minicpmo_4_5_8x4090.yaml pipeline: minicpmo_4_5 -session_mode: duplex stages: - stage_id: 0 diff --git a/vllm_omni/deploy/minicpmo_4_5_duplex.yaml b/vllm_omni/deploy/minicpmo_4_5_duplex.yaml deleted file mode 100644 index b51c46a9d..000000000 --- a/vllm_omni/deploy/minicpmo_4_5_duplex.yaml +++ /dev/null @@ -1,42 +0,0 @@ -# MiniCPM-o 4.5 native duplex deployment. -# -# The registered minicpmo_4_5 pipeline owns the fixed -# Thinker -> Talker -> Code2Wav topology. This overlay changes session -# lifetime and bounds all three stages to two live duplex sessions. -base_config: minicpmo_4_5.yaml -pipeline: minicpmo_4_5 -session_mode: duplex -active_stream_window: 1 -duplex_session: - idle_ttl_s: 300 - disconnect_grace_s: 30 - reaper_interval_s: 5 - resume_replay_ttl_s: 60 - resume_replay_max_bytes_per_session: 8388608 - max_pending_input_bytes_per_session: 16777216 - max_pending_turns_per_session: 4 - # Matches the configured Stage0/Stage1 sequence capacity below. - max_sessions: 2 - completed_append_cache_size: 256 - -stages: - - stage_id: 0 - max_num_seqs: 2 - async_scheduling: false - - - stage_id: 1 - max_num_seqs: 2 - async_scheduling: false - # Talker context is 4096 (tts_config.max_position_embeddings). - max_model_len: 4096 - # Overlay only the knobs duplex changes. Codec temperature / top_k stay - # on the base YAML. Native duplex still ends each Talker request after - # one generate_chunk (26 samples, last is codec EOS) inside the model; - # YAML max_tokens is only the hard ceiling. - default_sampling_params: - max_tokens: 4096 - min_tokens: 0 - - - stage_id: 2 - max_num_seqs: 2 - async_scheduling: false diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py b/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py index 83ef48b1e..d3d81f31d 100644 --- a/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py +++ b/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + from __future__ import annotations import base64 @@ -97,10 +100,14 @@ class MiniCPMO45NativeDuplexServingAdapter: cls._apply_default_scheduler_policy(runtime_config, config=config, model_config=model_config) ref_audio = config.ref_audio - if ref_audio is None and isinstance(extra_body.get("ref_audio"), str): - ref_audio = extra_body.pop("ref_audio") - if ref_audio is None and isinstance(extra_body.get("tts_ref_audio"), str): - ref_audio = extra_body.pop("tts_ref_audio") + extra_ref_audio = extra_body.get("ref_audio") + if ref_audio is None and isinstance(extra_ref_audio, str): + extra_body.pop("ref_audio") + ref_audio = extra_ref_audio + extra_tts_ref_audio = extra_body.get("tts_ref_audio") + if ref_audio is None and isinstance(extra_tts_ref_audio, str): + extra_body.pop("tts_ref_audio") + ref_audio = extra_tts_ref_audio if ref_audio is None: if any(str(modality).lower() == "audio" for modality in config.modalities): @@ -161,7 +168,11 @@ class MiniCPMO45NativeDuplexServingAdapter: stop_token_ids = cls._native_stage0_stop_token_ids(model_config) if stop_token_ids: stage0_params["stop_token_ids"] = stop_token_ids - runtime_config["duplex_stage_sampling_params"] = {"0": stage0_params} + # Stage 1 keeps the deploy YAML's codec knobs, minus upstream's + # min_new_token=50: that floor is for whole-utterance chat TTS, while a + # duplex chunk is 26 codec samples, so check_stop would never release + # the Talker and Thinker would stall on the next model turn. + runtime_config["duplex_stage_sampling_params"] = {"0": stage0_params, "1": {"min_tokens": 0}} scheduler_token_id = cls._native_scheduler_token_id(model_config) if scheduler_token_id is not None: runtime_config["duplex_scheduler_token_id"] = scheduler_token_id @@ -236,6 +247,8 @@ class MiniCPMO45NativeDuplexServingAdapter: if token_id is not None: return token_id eos_id = getattr(tokenizer, "eos_token_id", None) + if eos_id is None: + return None try: return int(eos_id) except (TypeError, ValueError): @@ -265,10 +278,13 @@ class MiniCPMO45NativeDuplexServingAdapter: value = convert(token) if isinstance(value, list): value = value[0] if len(value) == 1 else None - try: - token_id = int(value) - except (TypeError, ValueError): + if value is None: token_id = -1 + else: + try: + token_id = int(value) + except (TypeError, ValueError): + token_id = -1 unk_token_id = getattr(tokenizer, "unk_token_id", None) if token_id >= 0 and token_id != unk_token_id: return token_id