From c8b3a17bd3be56ebef509987498907bcbfc0b660 Mon Sep 17 00:00:00 2001 From: Linze Shi Date: Wed, 16 Sep 2026 17:29:57 -0700 Subject: [PATCH] [Frontend] Allow explicit turn-based serving (#7675) Signed-off-by: Linze-Shi --- docs/configuration/stage_configs.md | 2 +- docs/design/fullduplex.md | 19 ++- docs/serving/full_duplex_api.md | 30 +++- docs/serving/realtime_duplex_api.md | 47 +++---- tests/e2e/online_serving/test_minicpmo_4_5.py | 23 +++- .../openai_api/test_duplex_api_server.py | 130 ++++++++++++++---- vllm_omni/config/stage_config.py | 6 +- vllm_omni/deploy/minicpmo_4_5.yaml | 3 +- vllm_omni/deploy/minicpmo_4_5_turn.yaml | 3 + vllm_omni/entrypoints/openai/api_server.py | 38 +++-- 10 files changed, 205 insertions(+), 96 deletions(-) create mode 100644 vllm_omni/deploy/minicpmo_4_5_turn.yaml diff --git a/docs/configuration/stage_configs.md b/docs/configuration/stage_configs.md index aa81d42c6..199241cf6 100644 --- a/docs/configuration/stage_configs.md +++ b/docs/configuration/stage_configs.md @@ -53,7 +53,7 @@ The new deploy schema lives under `vllm_omni/deploy/` and is paired with a froze | ------- | ------ | ---------- | --------- | ------------- | | `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. Duplex models (pipelines with a `duplex_plugin`, e.g. MiniCPM-o 4.5) set `"duplex"`; `vllm-omni serve` then serves it over `/v1/realtime?duplex=1` and `/v1/chat/completions`, both on duplex sessions. | +| `session_mode` | str | optional | `"turn"` | Session behavior. For pipelines with a `duplex_plugin`, explicitly select `"duplex"` for the duplex engine or `"turn"` for the ordinary online serving stack. The shipped MiniCPM-o 4.5 default remains `"duplex"`. See [Full Duplex](../serving/full_duplex_api.md#enable-full-duplex). | | `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). | diff --git a/docs/design/fullduplex.md b/docs/design/fullduplex.md index 4646b85cd..41a050a47 100644 --- a/docs/design/fullduplex.md +++ b/docs/design/fullduplex.md @@ -108,18 +108,17 @@ plugin and the session runtime config to `DuplexOrchestrator` directly. `AsyncOmniEngine`, `OrchestratorBase` and `Orchestrator` carry only template seams; `tests/engine/test_duplex_import_boundary.py` checks that importing the turn-based stack loads no duplex module. -4. **Every surface a duplex server exposes is backed by a session.** +4. **Deployment selects the serving stack.** `vllm-omni serve` constructs `DuplexOmni` when the pipeline declares - `duplex_plugin`; the server exposes `/v1/realtime?duplex=1` (alias + `duplex_plugin` and the deploy configuration sets `session_mode: duplex`. + An explicit `session_mode: turn` selects `AsyncOmni` instead, without + changing the pipeline's duplex capability. In duplex mode the server exposes `/v1/realtime?duplex=1` (alias `/v1/duplex`), `POST /v1/chat/completions`, `/v1/models` and `/health`, and - every other turn-based route reports "not available". The chat route is - served by `DuplexChatCompletionsAdapter`, a Realtime client that runs one - short-lived session per request: speech becomes a committed turn, text is - seeded as the session's opening turn because a model-native model takes a - turn only when it hears speech. Whether a model can be reached with text is - `DuplexCapabilities.supports_chat_completions`; one that cannot is refused - at once rather than left to idle out. Turn-based use of the same model stays - available offline through `Omni` / `AsyncOmni`. + every other turn-based route reports "not available". The chat route uses + the ordinary chat service on the duplex engine, without a session, when + the plugin declares `DuplexCapabilities.supports_chat_completions`. + Turn-based deployments use the ordinary API initialization and retain + model capability checks and endpoint restrictions. 5. **Serving is transport only.** The websocket handler does socket I/O, wire-envelope validation, command translation, event rendering and the attachment/resume/replay bookkeeping; it holds no session state. diff --git a/docs/serving/full_duplex_api.md b/docs/serving/full_duplex_api.md index 92511e84f..9b3eceb58 100644 --- a/docs/serving/full_duplex_api.md +++ b/docs/serving/full_duplex_api.md @@ -22,14 +22,14 @@ capability gates, see the [Realtime Duplex API guide](realtime_duplex_api.md). ## Enable Full Duplex A model is served full duplex when its registered pipeline declares a -`duplex_plugin` (the model's `DuplexModelPlugin`). That declaration alone -decides it: `vllm-omni serve` constructs `DuplexOmni` instead of `AsyncOmni`, -and every surface the server exposes is backed by a duplex session. It serves +`duplex_plugin` (the model's `DuplexModelPlugin`) and its deploy configuration +sets `session_mode: duplex`. `vllm serve --omni` constructs `DuplexOmni` +instead of `AsyncOmni`. It serves `/v1/realtime?duplex=1` (and its alias `/v1/duplex`), `POST /v1/chat/completions`, `/v1/models` and `/health`; every other turn-based HTTP route (speech, batch, embeddings, video, ...) answers "not -available". Turn-based use of such a model stays available offline through the -Python API (`Omni` / `AsyncOmni`). +available". Set `session_mode: turn` to select the ordinary online serving +stack instead. The model's supported tasks and endpoint restrictions still apply. `/v1/chat/completions` is the ordinary chat service running on the duplex engine: a request is a turn-based generation on the same stages, served @@ -45,8 +45,24 @@ The deploy configuration of such a model must agree: session_mode: duplex ``` -A duplex model started with a deploy configuration that does not set it fails -at startup rather than falling back to turn-based serving. +A duplex-capable model must explicitly set `session_mode` to `duplex` or `turn` +in its deploy configuration, possibly through `base_config` inheritance. +Missing or invalid values fail at startup. The default MiniCPM-o deployment +continues to use duplex mode. + +To run MiniCPM-o 4.5 with the turn-based engine: + +```bash +vllm serve openbmb/MiniCPM-o-4_5 --omni \ + --deploy-config vllm_omni/deploy/minicpmo_4_5_turn.yaml \ + --trust-remote-code \ + --port 8091 +``` + +This profile inherits the default model and stage settings and overrides only +`session_mode`. It supports ordinary HTTP requests without creating a duplex +session handler. Mode selection happens at startup; a WebSocket query parameter +does not switch engines. The Python `Omni` / `AsyncOmni` APIs are unchanged. !!! warning diff --git a/docs/serving/realtime_duplex_api.md b/docs/serving/realtime_duplex_api.md index cc10d99c9..c18b416c7 100644 --- a/docs/serving/realtime_duplex_api.md +++ b/docs/serving/realtime_duplex_api.md @@ -32,33 +32,19 @@ vllm-omni serve openbmb/MiniCPM-o-4_5 \ mounts `ws://:8099/v1/realtime?duplex=1` (this page; `ws://:8099/v1/duplex` is an alias of the same route), `POST /v1/chat/completions`, `/v1/models` and `/health`, and nothing else. The remaining turn-based HTTP routes (speech, -batch, embeddings, video, ...) are not served by a duplex model. +batch, embeddings, video, ...) are not served in duplex mode. -`/v1/chat/completions` is not a turn-based path on a duplex server: each -request opens a short-lived duplex session and reads the answer off it. Audio -content becomes an ordinary committed turn; text is seeded as the session's -opening turn, because a model-native model takes a turn only when it hears -speech. A model that cannot be seeded that way declares so, and a text request -to it is refused with 400. Two consequences are worth sizing for: +`/v1/chat/completions` uses the ordinary chat service on the same engine +when the plugin declares `supports_chat_completions`. HTTP chat requests do +not create duplex sessions or consume session admission slots. Their options +follow the [Chat Completions API](chat_completions_api.md) and the model's +capabilities and endpoint restrictions. -- **Every request holds an admission slot** for its lifetime, so - `duplex_session.max_sessions` caps HTTP concurrency as well as websocket - sessions. Beyond it a request is refused with HTTP 503. -- **Latency follows the model's own clock.** A model-native session generates - per audio unit, so even a short answer costs what that many units cost -- - noticeably more than the same model would take turn-based. - -`n > 1`, `logprobs`, `tools`, and image or video content parts have no -representation in one duplex turn and are refused with HTTP 400 -- a Realtime -conversation item carries text and audio only. `temperature`, -`max_completion_tokens`, `modalities` and -`chat_template_kwargs.use_tts_template` apply; the other -`chat_template_kwargs` keys are logged as ignored. `extra_body` is passed to -the session unchanged, exactly as a websocket client's session `extra_body` -is, which is how a request asking for `modalities: ["text", "audio"]` supplies -the `ref_audio` MiniCPM-o requires for audio output. A model that should not -serve the route lists it in the deploy config's `endpoint_restrictions`, -exactly as a turn-based model does. +For a turn-based online deployment, use `minicpmo_4_5_turn.yaml` or explicitly +set `session_mode: turn` in a deploy configuration derived from the model's +profile. This selects `AsyncOmni` and the ordinary API initialization instead +of the duplex handler. See [Full Duplex](full_duplex_api.md#enable-full-duplex) +for the command and configuration requirements. ### Run the Example Client @@ -1345,8 +1331,9 @@ them out into typed events before they reach a client. - Session capacity is bounded by `duplex_session.max_sessions` in the deploy configuration; admission beyond it fails with `resource_exhausted` (`rate_limit_error`, retryable). -- A duplex deployment serves the duplex session surfaces only: the websocket - route and `/v1/chat/completions`, the latter on a session per request (see - [Start the Server](#start-the-server)). The other turn-based HTTP routes - report "not available" on a MiniCPM-o 4.5 server; turn-based use of the - model stays available offline through `Omni` / `AsyncOmni`. +- A duplex deployment serves the websocket route and, when supported by the + model, ordinary `/v1/chat/completions` requests on the same engine without a + session per request (see [Start the Server](#start-the-server)). Other + turn-based HTTP routes report "not available" in duplex mode. To use the + ordinary online serving stack, select `session_mode: turn` as described in + [Full Duplex](full_duplex_api.md#enable-full-duplex). diff --git a/tests/e2e/online_serving/test_minicpmo_4_5.py b/tests/e2e/online_serving/test_minicpmo_4_5.py index 5417d2b16..a1d045b72 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5.py @@ -83,6 +83,25 @@ test_params = [ ), ] +turn_test_params = [ + pytest.param( + OmniServerParams( + model=_MODEL, + stage_config_path=modify_stage_config(_CI_DEPLOY, updates={"session_mode": "turn"}), + use_stage_cli=False, + server_args=[ + "--trust-remote-code", + "--async-chunk", + "--chat-template", + str(_NATIVE_CHAT_TEMPLATE), + "--chat-template-content-format", + "openai", + ], + ), + id="turn_async_chunk", + ), +] + def get_system_prompt(): return { @@ -138,7 +157,7 @@ def _read_prompt_selection_log(offset: int) -> str: @pytest.mark.advanced_model @pytest.mark.omni @hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1) -@pytest.mark.parametrize("omni_server", test_params, indirect=True) +@pytest.mark.parametrize("omni_server", test_params + turn_test_params, indirect=True) def test_text_to_text_001(omni_server, openai_client) -> None: """ Test text-only input generating text output via OpenAI API. @@ -163,7 +182,7 @@ def test_text_to_text_001(omni_server, openai_client) -> None: @pytest.mark.full_model @pytest.mark.omni @hardware_test(res={"cuda": ["H100", "B200"], "npu": "A3"}, num_cards=1) -@pytest.mark.parametrize("omni_server", test_params, indirect=True) +@pytest.mark.parametrize("omni_server", test_params + turn_test_params, indirect=True) def test_text_to_audio_001(omni_server, openai_client) -> None: """ Test text-only input generating text + audio output via OpenAI API. diff --git a/tests/entrypoints/openai_api/test_duplex_api_server.py b/tests/entrypoints/openai_api/test_duplex_api_server.py index b528cbc2a..9cf58b114 100644 --- a/tests/entrypoints/openai_api/test_duplex_api_server.py +++ b/tests/entrypoints/openai_api/test_duplex_api_server.py @@ -1,28 +1,12 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project -"""Guards for the duplex-only API server (unified full-duplex framework). - -A model whose pipeline declares ``duplex_plugin`` is served through -``DuplexOmni`` and nothing else: sessions run over ``/v1/realtime?duplex=1`` -(alias ``/v1/duplex``) and every turn-based route reports "not available". -These tests lock that shape, which ``test_api_server_guards.py`` covers for the -diffusion and multi-stage servers: - -* the pipeline probe that decides duplex vs turn-based at startup, -* the ``app.state`` snapshot of a duplex server (what is wired, what stays - ``None``), -* both duplex websocket routes reaching the session handler, and - ``/v1/realtime`` without the flag still reporting the turn-based Realtime API - as unavailable, -* the startup-warmup gate that holds real clients while the warmup session runs. - -CPU-only: no engine is started; ``DuplexOmni`` is subclassed with just the -accessors the app-state wiring reads. -""" +"""CPU tests for serving-mode selection, duplex API wiring and warmup.""" from __future__ import annotations import asyncio +from argparse import Namespace +from dataclasses import replace from types import SimpleNamespace import pytest @@ -31,12 +15,14 @@ from fastapi.testclient import TestClient from starlette.datastructures import State from starlette.websockets import WebSocketDisconnect +from vllm_omni.config import stage_config from vllm_omni.config.config_factory import StageConfigFactory -from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig +from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig, PipelineConfig, StagePipelineConfig from vllm_omni.engine.duplex.config import DuplexCapabilities from vllm_omni.entrypoints.duplex.serving import OmniDuplexSessionHandler from vllm_omni.entrypoints.duplex_omni import DuplexOmni from vllm_omni.entrypoints.openai import api_server +from vllm_omni.utils.tracking_parser import TrackingNamespace pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -202,9 +188,10 @@ def _minimal_args(**overrides) -> SimpleNamespace: (None, False), ], ) -def test_duplex_model_probe_follows_the_pipeline_plugin(monkeypatch, pipeline_config, expected: bool) -> None: - """``duplex_plugin`` alone decides whether ``serve`` builds ``DuplexOmni``.""" +def test_duplex_serving_requires_a_pipeline_plugin(monkeypatch, tmp_path, pipeline_config, expected: bool) -> None: seen: dict[str, object] = {} + deploy_path = tmp_path / "deploy.yaml" + deploy_path.write_text("session_mode: duplex\n") def fake_get_pipeline_config(*, model, trust_remote_code, deploy_config_path=None, **_kwargs): seen.update(model=model, trust_remote_code=trust_remote_code, deploy_config_path=deploy_config_path) @@ -212,12 +199,12 @@ def test_duplex_model_probe_follows_the_pipeline_plugin(monkeypatch, pipeline_co monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", fake_get_pipeline_config) - kwargs = {"trust_remote_code": True, "deploy_config": "deploy.yaml"} - assert api_server._is_duplex_model("demo-duplex-model", kwargs) is expected + kwargs = {"trust_remote_code": True, "deploy_config": str(deploy_path)} + assert api_server._should_serve_duplex("demo-duplex-model", kwargs) is expected assert seen == { "model": "demo-duplex-model", "trust_remote_code": True, - "deploy_config_path": "deploy.yaml", + "deploy_config_path": str(deploy_path), } @@ -230,7 +217,98 @@ def test_duplex_model_probe_propagates_resolution_errors(monkeypatch) -> None: monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", fake_get_pipeline_config) with pytest.raises(ValueError, match="trust_remote_code"): - api_server._is_duplex_model("demo-duplex-model", {}) + api_server._should_serve_duplex("demo-duplex-model", {}) + + +@pytest.fixture +def duplex_pipeline(): + return PipelineConfig( + model_type="demo", + stages=(StagePipelineConfig(stage_id=0, model_stage="demo", final_output=True),), + duplex_plugin="pkg.mod.Plugin", + ) + + +@pytest.mark.parametrize("session_mode", ["turn", "duplex"]) +@pytest.mark.parametrize("config_source", ["explicit", "default", "basename", "inherited"]) +@pytest.mark.asyncio +async def test_serving_mode_selects_engine_and_preserves_stage_config( + monkeypatch, tmp_path, duplex_pipeline, session_mode, config_source +): + base_path = tmp_path / "base.yaml" + base_path.write_text("session_mode: duplex\nstages:\n - stage_id: 0\n max_num_seqs: 4\n") + deploy_path = tmp_path / "deploy.yaml" + deploy_path.write_text(f"base_config: base.yaml\nsession_mode: {session_mode}\n") + if config_source == "inherited": + deploy_path = tmp_path / "overlay.yaml" + deploy_path.write_text("base_config: deploy.yaml\n") + pipeline = replace(duplex_pipeline, default_deploy_config_name=deploy_path.name) + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **_: pipeline) + monkeypatch.setattr(stage_config, "_DEPLOY_DIR", tmp_path) + kwargs = {} + if config_source != "default": + kwargs["deploy_config"] = deploy_path.name if config_source == "basename" else str(deploy_path) + + created = [] + + class TurnEngine: + def __init__(self, model, **engine_kwargs): + self.model = model + self.kwargs = engine_kwargs + self.closed = False + created.append(self) + + def shutdown(self): + self.closed = True + + class DuplexEngine(TurnEngine): + pass + + monkeypatch.setattr(api_server, "AsyncOmni", TurnEngine) + monkeypatch.setattr(api_server, "DuplexOmni", DuplexEngine) + args = TrackingNamespace( + Namespace(model="demo-model", disable_log_stats=True, **kwargs), frozenset({"model", *kwargs}) + ) + async with api_server.build_async_omni_from_stage_config(args) as engine: + assert type(engine) is (DuplexEngine if session_mode == "duplex" else TurnEngine) + assert engine.model == "demo-model" + assert engine.kwargs == {**kwargs, "log_stats": False} + assert not engine.closed + assert len(created) == 1 + assert engine.closed + deploy = stage_config.load_deploy_config(deploy_path) + assert deploy.session_mode == session_mode + assert deploy.stages[0].max_num_seqs == 4 + + +@pytest.mark.parametrize("contents", ["{}\n", "session_mode: null\n", "session_mode: typo\n"]) +def test_duplex_serving_rejects_missing_or_invalid_mode(monkeypatch, tmp_path, duplex_pipeline, contents): + deploy_path = tmp_path / "deploy.yaml" + deploy_path.write_text(contents) + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **_: duplex_pipeline) + with pytest.raises(ValueError, match="session_mode: turn or duplex"): + api_server._should_serve_duplex("demo-model", {"deploy_config": str(deploy_path)}) + + +def test_duplex_serving_requires_a_deploy_config(monkeypatch, duplex_pipeline): + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **_: duplex_pipeline) + with pytest.raises(ValueError, match="requires a deploy config"): + api_server._should_serve_duplex("demo-model", {}) + + +def test_duplex_serving_propagates_deploy_load_errors(monkeypatch, tmp_path, duplex_pipeline): + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **_: duplex_pipeline) + with pytest.raises(FileNotFoundError): + api_server._should_serve_duplex("demo-model", {"deploy_config": str(tmp_path / "missing.yaml")}) + + +@pytest.mark.parametrize("deploy_name,expected", [("minicpmo_4_5.yaml", True), ("minicpmo_4_5_turn.yaml", False)]) +def test_minicpmo_serving_profiles(monkeypatch, deploy_name, expected): + from vllm_omni.model_executor.models.minicpmo_4_5.pipeline import MINICPMO_4_5_PIPELINE + + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **_: MINICPMO_4_5_PIPELINE) + assert api_server._should_serve_duplex("openbmb/MiniCPM-o-4_5", {"deploy_config": deploy_name}) is expected + assert api_server._should_serve_duplex("openbmb/MiniCPM-o-4_5", {}) is True # --------------------------------------------------------------------------- # diff --git a/vllm_omni/config/stage_config.py b/vllm_omni/config/stage_config.py index 6af65b8ed..24453ae88 100644 --- a/vllm_omni/config/stage_config.py +++ b/vllm_omni/config/stage_config.py @@ -309,10 +309,8 @@ class PipelineConfig: diffusers_class_name: str | None = None diffusers_class_aliases: tuple[str, ...] = () endpoint_restrictions: tuple[EndpointRestriction, ...] = () - # Dotted path of the model's ``DuplexModelPlugin``. A pipeline is a duplex - # model iff this is set: ``vllm-omni serve`` then always runs it through - # ``DuplexOmni`` (every served surface runs on a duplex session) and the engine hosts one - # ``DuplexOrchestrator`` with the plugin loaded. + # Dotted path of the model's ``DuplexModelPlugin``. Online serving uses + # DuplexOmni only when the deploy configuration selects session_mode: duplex. duplex_plugin: str | None = None # Legacy duplex wiring of the models that are not ported to the plugin # framework yet (PersonaPlex, Nemotron VoiceChat). Nothing reads them: a diff --git a/vllm_omni/deploy/minicpmo_4_5.yaml b/vllm_omni/deploy/minicpmo_4_5.yaml index 5c26fc2c6..a04cff400 100644 --- a/vllm_omni/deploy/minicpmo_4_5.yaml +++ b/vllm_omni/deploy/minicpmo_4_5.yaml @@ -1,8 +1,7 @@ # MiniCPM-o 4.5 three-stage deploy on one large-memory GPU. pipeline: minicpmo_4_5 async_chunk: true -# session_mode: duplex makes vllm-omni serve run this model duplex-only -# (/v1/realtime?duplex=1); every other duplex_session knob keeps its runtime default. +# Use minicpmo_4_5_turn.yaml to select the ordinary turn-based serving engine. session_mode: duplex # Keep the active window aligned with the session and stage capacities below. # A smaller window serializes first-packet generation for concurrent chat turns. diff --git a/vllm_omni/deploy/minicpmo_4_5_turn.yaml b/vllm_omni/deploy/minicpmo_4_5_turn.yaml new file mode 100644 index 000000000..8dada1257 --- /dev/null +++ b/vllm_omni/deploy/minicpmo_4_5_turn.yaml @@ -0,0 +1,3 @@ +# MiniCPM-o 4.5 using the turn-based engine and ordinary HTTP serving. +base_config: minicpmo_4_5.yaml +session_mode: turn diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 376a9ccd2..20706396c 100644 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/vllm_omni/entrypoints/openai/api_server.py @@ -19,6 +19,7 @@ from argparse import Namespace from collections.abc import AsyncIterator from contextlib import asynccontextmanager from http import HTTPStatus +from pathlib import Path from typing import Annotated, Any, Literal import vllm.envs as envs @@ -440,21 +441,34 @@ async def build_async_omni( yield async_omni -def _is_duplex_model(model: str, kwargs: dict[str, Any]) -> bool: - """Whether ``model``'s pipeline declares a ``duplex_plugin`` (served by ``DuplexOmni``). - - Resolution errors propagate: a duplex model whose pipeline or deploy - config cannot be resolved must fail startup rather than silently start a - turn-based server. - """ +def _should_serve_duplex(model: str, kwargs: dict[str, Any]) -> bool: + """Select the serving engine without changing the model's duplex capability.""" from vllm_omni.config.config_factory import StageConfigFactory + from vllm_omni.config.stage_config import _DEPLOY_DIR, resolve_deploy_yaml pipeline_config = StageConfigFactory.get_pipeline_config( model=model, trust_remote_code=bool(kwargs.get("trust_remote_code")), deploy_config_path=kwargs.get("deploy_config"), ) - return bool(pipeline_config is not None and getattr(pipeline_config, "duplex_plugin", None)) + if pipeline_config is None or not getattr(pipeline_config, "duplex_plugin", None): + return False + + deploy_path = kwargs.get("deploy_config") + if deploy_path is None: + if pipeline_config.default_deploy_config_name is None: + raise ValueError("A duplex-capable model requires a deploy config with session_mode: turn or duplex") + deploy_path = _DEPLOY_DIR / pipeline_config.default_deploy_config_name + else: + deploy_path = Path(deploy_path) + if not deploy_path.exists() and deploy_path.parent == Path("."): + deploy_path = _DEPLOY_DIR / deploy_path + + # Resolve base_config too, so the API and stage workers use the same mode. + session_mode = resolve_deploy_yaml(deploy_path).get("session_mode") + if session_mode not in ("turn", "duplex"): + raise ValueError("A duplex-capable model requires session_mode: turn or duplex in its deploy config") + return session_mode == "duplex" @asynccontextmanager @@ -518,9 +532,7 @@ async def build_async_omni_from_stage_config( kwargs.pop("robot_openpi_idle_timeout", None) model = kwargs.pop("model", None) or args.model kwargs.setdefault("log_stats", not args.disable_log_stats) - if _is_duplex_model(model, kwargs): - # A duplex model is always served in duplex mode: sessions over - # /v1/realtime?duplex=1, no turn-based HTTP routes. + if _should_serve_duplex(model, kwargs): async_omni = DuplexOmni(model=model, **kwargs) else: async_omni = AsyncOmni(model=model, **kwargs) @@ -712,9 +724,7 @@ async def omni_init_app_state( state.stage_configs = engine_client.stage_configs if hasattr(engine_client, "stage_configs") else None model_name = served_model_names[0] if served_model_names else args.model - # Duplex mode: a duplex model is served through DuplexOmni only. Sessions - # run over /v1/realtime?duplex=1; every turn-based HTTP route reports - # "not available". + # Initialize the API surface for the selected engine, not just the model. if isinstance(engine_client, DuplexOmni): await _init_duplex_app_state(engine_client, state, args, base_model_paths, vllm_config, request_logger) return