From d31a2a3ba8e33991665e3f1a652a0a7277f91e17 Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Fri, 25 Sep 2026 10:25:15 -0700 Subject: [PATCH] [Bugfix] Fix Nemotron VoiceChat duplex nightly handshake (#7884) Signed-off-by: Sy03 <1370724210@qq.com> --- docs/design/fullduplex.md | 16 +- docs/serving/full_duplex_api.md | 17 +- docs/serving/realtime_duplex_api.md | 11 +- tests/clients/test_model_session_configs.py | 17 +- .../nemotron_voicechat_realtime_duplex.py | 170 +++--- .../test_nemotron_voicechat_duplex.py | 21 +- .../test_session_runner_nemotron_voicechat.py | 167 ++++++ .../nemotron_voicechat/duplex/conftest.py | 20 - .../duplex/test_data_plane.py | 3 +- .../nemotron_voicechat/duplex/test_input.py | 9 + .../duplex/test_runtime_contract.py | 115 ---- vllm_omni/clients/nemotron_voicechat.py | 29 + vllm_omni/engine/duplex/intermediate.py | 53 +- vllm_omni/engine/duplex/plugin.py | 88 +++ .../engine/duplex/session/model_channel.py | 6 + .../models/minicpmo_4_5/duplex/plugin.py | 50 +- .../models/minicpmo_4_5/duplex/session.py | 56 +- .../nemotron_voicechat/duplex/__init__.py | 14 +- .../nemotron_voicechat/duplex/capabilities.py | 55 ++ .../nemotron_voicechat/duplex/data_plane.py | 43 +- .../models/nemotron_voicechat/duplex/input.py | 35 +- .../nemotron_voicechat/duplex/plugin.py | 555 ++++++++++++++++++ .../nemotron_voicechat/duplex/runtime.py | 167 ------ .../duplex/serving_adapter.py | 361 ------------ .../nemotron_voicechat_thinker.py | 30 +- .../models/nemotron_voicechat/pipeline.py | 12 +- 26 files changed, 1186 insertions(+), 934 deletions(-) create mode 100644 tests/engine/duplex/test_session_runner_nemotron_voicechat.py delete mode 100644 tests/model_executor/models/nemotron_voicechat/duplex/conftest.py delete mode 100644 tests/model_executor/models/nemotron_voicechat/duplex/test_runtime_contract.py create mode 100644 vllm_omni/clients/nemotron_voicechat.py create mode 100644 vllm_omni/model_executor/models/nemotron_voicechat/duplex/capabilities.py create mode 100644 vllm_omni/model_executor/models/nemotron_voicechat/duplex/plugin.py delete mode 100644 vllm_omni/model_executor/models/nemotron_voicechat/duplex/runtime.py delete mode 100644 vllm_omni/model_executor/models/nemotron_voicechat/duplex/serving_adapter.py diff --git a/docs/design/fullduplex.md b/docs/design/fullduplex.md index 2733ecf08..da0584843 100644 --- a/docs/design/fullduplex.md +++ b/docs/design/fullduplex.md @@ -242,10 +242,12 @@ vllm_omni/ │ └── overlap_policy.py / commit_policy.py / playback_ledger.py ├── config/stage_config.py PipelineConfig.duplex_plugin; DuplexSessionRuntimeConfig ├── model_executor/models/minicpmo_4_5/duplex/plugin.py MiniCPMO45DuplexPlugin (+ data_plane, input, policy, ...) +├── model_executor/models/nemotron_voicechat/duplex/plugin.py NemotronVoiceChatDuplexPlugin (+ data_plane, input, capabilities) └── clients/ ├── duplex.py DuplexClientBase (ABC), DuplexClient (websocket), client-side events ├── inline_duplex.py InlineDuplexClient (in-process, over DuplexOmni) ├── minicpmo_4_5.py MiniCPM-o 4.5 session preset + ├── nemotron_voicechat.py Nemotron VoiceChat session preset └── personaplex.py PersonaPlex session preset ``` @@ -388,7 +390,7 @@ policy that used to be two separately configured objects: | Half | Members | | --- | --- | | engine policy | `configure_sampling_params(runtime_config, defaults)`, `plan_append(...) -> DuplexAppendPlan` (the resumable Stage0 prompt for one unit), `decide_output(...) -> DuplexOutputDecision \| None` (e.g. the listen decision on a finished Stage0 segment) | -| session policy | `capabilities(max_sessions)`, `validate_client_extra_body`, `prepare_runtime_config(config, model_config)` (server-owned runtime keys, reference audio resolution), `runtime_config_for_update`, `runtime_config_for_function_output`, `create_session_state() -> DuplexModelSessionState`, `data_plane: DuplexDataPlane` (projects raw stage outputs into internal events), `data_plane_context(...)` | +| session policy | `capabilities(max_sessions)`, `validate_client_extra_body`, `prepare_runtime_config(config, model_config)` (server-owned runtime keys, reference audio resolution), `runtime_config_for_update`, `runtime_config_for_function_output`, `runtime_config_after_model_output` (consumption acknowledgement), `create_session_state() -> DuplexModelSessionState`, `data_plane: DuplexDataPlane` (projects raw stage outputs into internal events), `data_plane_context(...)` | `DuplexOmniEngine._validate_deployment` loads the plugin before any stage starts; `DuplexSessionManager.__init__` validates it against the stage @@ -396,12 +398,12 @@ sampling defaults once the stage pools exist. Plugin hooks that may block (`prepare_runtime_config` fetching `ref_audio`) are awaited in `open()` and offloaded from the loop. -The MiniCPM-o 4.5 plugin (`model_executor/models/minicpmo_4_5/duplex/plugin.py`) -is the only integration in this framework version. PersonaPlex and Nemotron -VoiceChat still carry their pre-framework duplex code (runtime extension plus -serving adapter) and are therefore not served over this framework yet: their -pipelines declare no `duplex_plugin`, so they run turn-based until the -follow-up PRs port them (RFC vllm-omni#7181, PR 2/3). +Frame-based plugins use `engine/duplex/intermediate.py::build_duplex_append_prompt` +for the shared request identity, sequencing and config snapshots. Token budgets, +PCM framing and model-specific worker fields remain in the plugins. + +See [supported models and deployments](../serving/full_duplex_api.md#enable-full-duplex) +for the current plugin integrations and deployment configurations. ## Serving diff --git a/docs/serving/full_duplex_api.md b/docs/serving/full_duplex_api.md index 9b3eceb58..a54fa6f66 100644 --- a/docs/serving/full_duplex_api.md +++ b/docs/serving/full_duplex_api.md @@ -72,12 +72,21 @@ does not switch engines. The Python `Omni` / `AsyncOmni` APIs are unchanged. `session.created.session.capabilities` is present before treating the connection as full duplex. -**MiniCPM-o 4.5** (`vllm_omni/deploy/minicpmo_4_5.yaml`) is the only model -served over this endpoint today. PersonaPlex and Nemotron VoiceChat still carry -their pre-framework duplex code: their pipelines declare no `duplex_plugin`, so -they run turn-based until the follow-up PRs port them to the plugin contract +**MiniCPM-o 4.5** (`vllm_omni/deploy/minicpmo_4_5.yaml`), **Qwen3-Omni** +(`vllm_omni/deploy/qwen3_omni_duplex.yaml`), and **Nemotron VoiceChat** +(`vllm_omni/deploy/nemotron_labs_voicechat_duplex.yaml`) are served +over this endpoint today. PersonaPlex still carries its pre-framework duplex +code: its pipeline declares no `duplex_plugin`, so it runs turn-based until the +follow-up PR ports it to the plugin contract (RFC [vllm-omni#7181](https://github.com/vllm-project/vllm-omni/issues/7181)). +For Nemotron VoiceChat Python clients, use +`vllm_omni.clients.nemotron_voicechat.create_duplex_session_config()` with +`DuplexClient`. The preset selects 16 kHz float32 input, 22.05 kHz PCM16 output, +and automatic responses. Append 1,280-sample (80 ms) input frames; when using +`stream_pcm`, set `chunk_ms=80`. Pass `instructions` and optional `tools` to +the preset before opening the session. + JoyVL is a separate HTTP interaction orchestrator and does not use these WebSocket endpoints. See [Standalone Experimental Servers](standalone_servers.md). diff --git a/docs/serving/realtime_duplex_api.md b/docs/serving/realtime_duplex_api.md index cd8832c82..f2c26d70e 100644 --- a/docs/serving/realtime_duplex_api.md +++ b/docs/serving/realtime_duplex_api.md @@ -8,11 +8,9 @@ history stays honest, and a dropped connection can resume the same session. This page covers how to run a duplex deployment, how to drive it from Python with `vllm_omni.clients.duplex.DuplexClient`, and the complete wire contract. -The endpoint is served for models that ship a duplex plugin — including -MiniCPM-o 4.5 and Qwen3-Omni — and just for deploy configurations that declare -`session_mode: duplex`. PersonaPlex and Nemotron VoiceChat are ported to the -plugin contract in follow-up PRs and are not served over this endpoint yet. The runtime architecture -is described in [Full-Duplex Runtime (MiniCPM-o 4.5)](../design/fullduplex.md). +The endpoint requires a duplex plugin and a deployment with `session_mode: duplex`. +See [supported models and deployments](full_duplex_api.md#enable-full-duplex) +and the [runtime architecture](../design/fullduplex.md). ## Qwen3-Omni conversation history @@ -1340,9 +1338,6 @@ them out into typed events before they reach a client. ## Known Limitations -- Only MiniCPM-o 4.5 is served over this endpoint today; PersonaPlex and - Nemotron VoiceChat arrive with the follow-up PRs that port them to the - plugin contract. - Several surfaces are capability-gated per model (see *Capability negotiation by model* above): PersonaPlex does not support session resume, barge-in, or audio truncation; Nemotron VoiceChat does not support barge-in diff --git a/tests/clients/test_model_session_configs.py b/tests/clients/test_model_session_configs.py index 080fea879..2423b8b00 100644 --- a/tests/clients/test_model_session_configs.py +++ b/tests/clients/test_model_session_configs.py @@ -11,6 +11,9 @@ from vllm_omni.clients.duplex import AudioFormat from vllm_omni.clients.minicpmo_4_5 import ( create_duplex_session_config as create_minicpmo45_session_config, ) +from vllm_omni.clients.nemotron_voicechat import ( + create_duplex_session_config as create_nemotron_voicechat_session_config, +) from vllm_omni.clients.personaplex import ( create_duplex_session_config as create_personaplex_session_config, ) @@ -38,12 +41,22 @@ def test_personaplex_session_config_matches_deployment(): assert payload["instructions"] == "You are calm." +def test_nemotron_voicechat_session_config_matches_deployment(): + tools = [{"type": "function", "name": "lookup", "parameters": {"type": "object"}}] + config = create_nemotron_voicechat_session_config(instructions="hi", tools=tools) + assert config.input_audio == AudioFormat("pcm_f32le", 16_000) + assert config.output_audio == AudioFormat("pcm16", 22_050) + payload = config.to_session_payload(model="nemotron") + assert payload["model"] == "nemotron" and payload["instructions"] == "hi" + assert payload["extra_body"] == {"auto_response": True, "realtime_tools": tools} + + @pytest.mark.parametrize( "duplex_package", [ - # PersonaPlex and Nemotron VoiceChat rejoin this list with the follow-up - # PRs that port them to the duplex plugin framework. + # PersonaPlex rejoins with its port to the duplex plugin framework. "vllm_omni.model_executor.models.minicpmo_4_5.duplex", + "vllm_omni.model_executor.models.nemotron_voicechat.duplex", ], ) def test_model_duplex_packages_stay_clear_of_client_library(duplex_package: str): diff --git a/tests/e2e/online_serving/nemotron_voicechat_realtime_duplex.py b/tests/e2e/online_serving/nemotron_voicechat_realtime_duplex.py index 6aba7cfe7..002d4e46c 100644 --- a/tests/e2e/online_serving/nemotron_voicechat_realtime_duplex.py +++ b/tests/e2e/online_serving/nemotron_voicechat_realtime_duplex.py @@ -6,17 +6,16 @@ import asyncio import base64 import json import math -import uuid import wave from collections.abc import Sequence from contextlib import asynccontextmanager, suppress from pathlib import Path -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import numpy as np from scipy.signal import resample_poly -from vllm_omni.experimental.fullduplex.client import RealtimeDuplexClient, wait_for, write_pcm16_wav +from vllm_omni.clients.duplex import DuplexClient, EventCollector, wait_for_condition, write_pcm16_wav +from vllm_omni.clients.nemotron_voicechat import create_duplex_session_config INPUT_SAMPLE_RATE_HZ = 16_000 OUTPUT_SAMPLE_RATE_HZ = 22_050 @@ -45,13 +44,6 @@ DEFAULT_FUNCTION_INSTRUCTIONS = ( ) -def _url(base_url: str, model: str, session_id: str) -> str: - parts = urlsplit(base_url) - query = dict(parse_qsl(parts.query, keep_blank_values=True)) - query.update(duplex="1", model=model, autostart="0", session_id=session_id) - return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) - - def _read_wav(path: Path, *, input_channel: int = 0) -> np.ndarray: with wave.open(str(path), "rb") as wav_file: channels = wav_file.getnchannels() @@ -73,50 +65,42 @@ def _read_wav(path: Path, *, input_channel: int = 0) -> np.ndarray: return np.ascontiguousarray(pcm, dtype=" int: +async def _stream(client: DuplexClient, pcm: np.ndarray, *, max_frames: int | None, realtime: bool) -> int: count = math.ceil(pcm.size / FRAME_SAMPLES) if max_frames is not None: count = min(count, max_frames) for seq in range(count): frame = pcm[seq * FRAME_SAMPLES : (seq + 1) * FRAME_SAMPLES] frame = np.pad(frame, (0, FRAME_SAMPLES - frame.size)).astype(" list[dict[str, object]]: - return [event for event in client.events.events if event.get("type") == event_type] +def _events(collector: EventCollector, event_type: str) -> list[dict[str, object]]: + return [event for event in collector.events if event.get("type") == event_type] async def _return_function_output_when_ready( - client: RealtimeDuplexClient, + client: DuplexClient, + collector: EventCollector, *, output: str, timeout_s: float, ) -> tuple[str, int]: - await wait_for( - lambda: bool(client.events.errors()) or client.events.count("response.function_call_arguments.done") > 0, + await wait_for_condition( + lambda: bool(collector.errors()) or collector.count("response.function_call_arguments.done") > 0, timeout_s=timeout_s, label="function call to execute", ) - if client.events.errors(): - raise AssertionError(f"function call failed before tool execution: {client.events.errors()}") - function_done = _events(client, "response.function_call_arguments.done")[-1] + if collector.errors(): + raise AssertionError(f"function call failed before tool execution: {collector.errors()}") + function_done = _events(collector, "response.function_call_arguments.done")[-1] call_id = function_done.get("call_id") if not isinstance(call_id, str) or not call_id: raise AssertionError(f"completed function call has no call_id: {function_done}") - event_count_before_output = len(client.events.events) + event_count_before_output = len(collector.events) await client.send( { "type": "conversation.item.create", @@ -130,22 +114,26 @@ async def _return_function_output_when_ready( return call_id, event_count_before_output -def _write_events(path: Path, client: RealtimeDuplexClient) -> None: +def _write_events(path: Path, collector: EventCollector) -> None: path.write_text( - "".join(json.dumps(event, ensure_ascii=False) + "\n" for event in client.events.events), + "".join(json.dumps(event, ensure_ascii=False) + "\n" for event in collector.events), encoding="utf-8", ) @asynccontextmanager -async def _managed_client(client: RealtimeDuplexClient, *, timeout_s: float): - async with client: - try: +async def _managed_client(client: DuplexClient, collector: EventCollector): + consume_task = asyncio.create_task(collector.consume(client)) + # Subscribe before the handshake so the collector sees session.created. + await asyncio.sleep(0) + try: + async with client: yield - finally: - if client.events.count("session.created") and not client.events.count("session.closed"): - with suppress(Exception): - await client.close_session(timeout_s=min(timeout_s, 30.0)) + await consume_task + finally: + consume_task.cancel() + with suppress(asyncio.CancelledError): + await consume_task async def run(args: argparse.Namespace) -> dict[str, object]: @@ -156,33 +144,22 @@ async def run(args: argparse.Namespace) -> dict[str, object]: else: instructions = args.instructions tools = DEFAULT_FUNCTION_TOOLS if args.expect_function_call else None - session_id = f"nemotron-voicechat-{uuid.uuid4().hex}" - client = RealtimeDuplexClient(_url(args.url, args.model, session_id)) - async with _managed_client(client, timeout_s=args.timeout_s): - session_payload: dict[str, object] = { - "session_id": session_id, - "model": args.model, - "modalities": ["audio", "text"], - "input_audio_format": "pcm_f32le", - "output_audio_format": "pcm16", - "instructions": instructions, - "idle_timeout_s": args.timeout_s, - "turn_detection": None, - "extra_body": {"auto_response": True}, - } - if tools is not None: - session_payload["tools"] = tools - await client.send({"type": "session.update", "session": session_payload}) - await wait_for( - lambda: client.events.count("session.created") > 0 or bool(client.events.errors()), - timeout_s=args.timeout_s, - label="session.created", - ) - if client.events.errors(): - raise AssertionError(f"session setup failed: {client.events.errors()}") - created = _events(client, "session.created")[-1] - session = created.get("session") - capabilities = session.get("capabilities") if isinstance(session, dict) else None + collector = EventCollector() + client = DuplexClient( + args.url, + model=args.model, + config=create_duplex_session_config( + instructions=instructions, + tools=tools, + idle_timeout_s=args.timeout_s, + ), + handshake_timeout_s=args.timeout_s, + reconnect=None, + heartbeat_interval_s=None, + ) + async with _managed_client(client, collector): + session_id = client.session_id + capabilities = client.session_info.get("capabilities") expected = dict( implementation_level="model_native_duplex", chunk_period_ms=80, @@ -197,6 +174,7 @@ async def run(args: argparse.Namespace) -> dict[str, object]: asyncio.create_task( _return_function_output_when_ready( client, + collector, output=args.function_output, timeout_s=args.timeout_s, ) @@ -206,19 +184,19 @@ async def run(args: argparse.Namespace) -> dict[str, object]: ) pcm = _read_wav(Path(args.input_wav), input_channel=args.input_channel) frame_count = await _stream(client, pcm, max_frames=args.max_frames, realtime=not args.no_realtime) - completed_responses_at_commit = client.events.count("response.done") - await client.send({"type": "input_audio_buffer.commit", "final": True}) - await wait_for( + completed_responses_at_commit = collector.count("response.done") + await client.commit() + await wait_for_condition( lambda: ( - bool(client.events.errors()) + bool(collector.errors()) or ( - client.events.count("response.function_call_arguments.done") > 0 + collector.count("response.function_call_arguments.done") > 0 if args.expect_function_call else ( - client.events.count("response.output_audio.delta") >= args.minimum_audio_chunks + collector.count("response.output_audio.delta") >= args.minimum_audio_chunks and ( args.allow_incomplete_response - or client.events.count("response.done") > completed_responses_at_commit + or collector.count("response.done") > completed_responses_at_commit ) ) ) @@ -227,9 +205,9 @@ async def run(args: argparse.Namespace) -> dict[str, object]: label="model output", ) await asyncio.sleep(args.drain_s) - if client.events.errors(): - raise AssertionError(f"Realtime session emitted errors: {client.events.errors()}") - done_events = _events(client, "response.done") + if collector.errors(): + raise AssertionError(f"Realtime session emitted errors: {collector.errors()}") + done_events = _events(collector, "response.done") if not args.expect_function_call and not args.allow_incomplete_response: response = done_events[-1].get("response") if done_events else None status = response.get("status") if isinstance(response, dict) else None @@ -237,24 +215,22 @@ async def run(args: argparse.Namespace) -> dict[str, object]: raise AssertionError(f"response did not complete successfully: {done_events[-1:]}") function_events = [ - event for event in client.events.events if str(event.get("type", "")).startswith("response.function_call") + event for event in collector.events if str(event.get("type", "")).startswith("response.function_call") ] function_items = [ - event - for event in _events(client, "response.output_item.done") - if isinstance(event.get("item"), dict) and event["item"].get("type") == "function_call" + item + for event in _events(collector, "response.output_item.done") + if isinstance((item := event.get("item")), dict) and item.get("type") == "function_call" ] if args.expect_function_call and not any( event.get("type") == "response.function_call_arguments.done" for event in function_events ): raise AssertionError(f"no completed function call: {function_events}") if args.expect_function_call: - matching_items = [ - event for event in function_items if event["item"].get("name") == args.expected_function_name - ] + matching_items = [item for item in function_items if item.get("name") == args.expected_function_name] if not matching_items: raise AssertionError(f"expected {args.expected_function_name!r}, got {function_items}") - function_item = matching_items[-1]["item"] + function_item: dict[str, object] = matching_items[-1] try: function_arguments = json.loads(str(function_item.get("arguments", ""))) except json.JSONDecodeError as exc: @@ -275,7 +251,7 @@ async def run(args: argparse.Namespace) -> dict[str, object]: assert returned_call_id == call_id def tool_result_completed() -> bool: - later = client.events.events[event_count_before_output:] + later = collector.events[event_count_before_output:] transcript = "".join( str(event.get("delta", "")) for event in later @@ -287,23 +263,23 @@ async def run(args: argparse.Namespace) -> dict[str, object]: and (args.expected_post_tool_text is None or args.expected_post_tool_text.lower() in transcript) ) - await wait_for( - lambda: bool(client.events.errors()) or tool_result_completed(), + await wait_for_condition( + lambda: bool(collector.errors()) or tool_result_completed(), timeout_s=args.timeout_s, label="completed response after function output", ) - if client.events.errors(): - raise AssertionError(f"function output failed: {client.events.errors()}") + if collector.errors(): + raise AssertionError(f"function output failed: {collector.errors()}") await asyncio.sleep(args.drain_s) - audio = client.events.audio_bytes() - audio_events = _events(client, "response.output_audio.delta") + audio = collector.audio_bytes() + audio_events = _events(collector, "response.output_audio.delta") rates = {event.get("sample_rate_hz") for event in audio_events} if not args.expect_function_call and audio and rates != {OUTPUT_SAMPLE_RATE_HZ}: raise AssertionError(f"unexpected output sample rates: {rates}") if not args.expect_function_call and args.minimum_audio_chunks and not audio: raise AssertionError("model produced no audio") - expected_bytes = 2 * OUTPUT_SAMPLE_RATE_HZ * expected["chunk_period_ms"] // 1000 + expected_bytes = 2 * OUTPUT_SAMPLE_RATE_HZ * int(str(expected["chunk_period_ms"])) // 1000 packet_sizes = [len(base64.b64decode(str(event.get("delta", "")), validate=True)) for event in audio_events] if not args.expect_function_call and any(size != expected_bytes for size in packet_sizes): raise AssertionError(f"audio deltas are not fixed 80 ms PCM16 packets: {packet_sizes}") @@ -314,17 +290,17 @@ async def run(args: argparse.Namespace) -> dict[str, object]: f"model output RMS {audio_rms:.6f} is below {args.minimum_audio_rms:.6f}; " "received packets contain only silence" ) - _write_events(output_dir / "events.jsonl", client) + _write_events(output_dir / "events.jsonl", collector) if audio: write_pcm16_wav(output_dir / "output.wav", audio, sample_rate_hz=OUTPUT_SAMPLE_RATE_HZ) - result = { + result: dict[str, object] = { "ok": True, - "session_id": session_id, + "session_id": session_id if isinstance(session_id, str) else None, "input_frames": frame_count, "capabilities": capabilities, "event_counts": { - event_type: client.events.count(event_type) - for event_type in sorted({str(event.get("type")) for event in client.events.events}) + event_type: collector.count(event_type) + for event_type in sorted({str(event.get("type")) for event in collector.events}) }, "audio_bytes": len(audio), "audio_rms": audio_rms, diff --git a/tests/e2e/online_serving/test_nemotron_voicechat_duplex.py b/tests/e2e/online_serving/test_nemotron_voicechat_duplex.py index dfb3044e8..1fd70f572 100644 --- a/tests/e2e/online_serving/test_nemotron_voicechat_duplex.py +++ b/tests/e2e/online_serving/test_nemotron_voicechat_duplex.py @@ -29,16 +29,7 @@ _SERVER_ENV: dict[str, str] = {"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S": str(SERVER_INPU if TOKENIZER: _SERVER_ENV["NEMOTRON_VOICECHAT_LLM_PATH"] = TOKENIZER -pytestmark = [ - pytest.mark.full_model, - pytest.mark.omni, - # Since #7413 a duplex-capable model is served by the duplex engine only when - # its pipeline declares a duplex_plugin. The Nemotron VoiceChat pipeline does - # not, so the server starts the turn engine and the /v1/realtime websocket - # falls through to vllm's speech-to-text realtime handler, which rejects the - # native duplex client's session.update ("Missing required field: model"). - pytest.mark.skip(reason="Nemotron VoiceChat is not served by the duplex engine yet (no duplex_plugin)"), -] +pytestmark = [pytest.mark.full_model, pytest.mark.omni] @hardware_test(res={"cuda": ["H100", "B200"]}, num_cards=1) @@ -71,6 +62,10 @@ def test_native_duplex_turn_taking_streams_model_audio(omni_server, tmp_path: Pa ) result = asyncio.run(run(args)) assert result["ok"] is True - assert result["input_frames"] == 190 - assert result["event_counts"]["response.speak"] > 0 - assert result["audio_bytes"] >= result["input_frames"] * 1764 * 2 // 4 + input_frames = result["input_frames"] + event_counts = result["event_counts"] + audio_bytes = result["audio_bytes"] + assert isinstance(input_frames, int) and input_frames == 190 + assert isinstance(event_counts, dict) and int(str(event_counts.get("response.speak", 0))) > 0 + assert isinstance(audio_bytes, int) + assert audio_bytes >= input_frames * 1764 * 2 // 4 diff --git a/tests/engine/duplex/test_session_runner_nemotron_voicechat.py b/tests/engine/duplex/test_session_runner_nemotron_voicechat.py new file mode 100644 index 000000000..254434fa5 --- /dev/null +++ b/tests/engine/duplex/test_session_runner_nemotron_voicechat.py @@ -0,0 +1,167 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Essential Nemotron regressions through the real session runner and plugin.""" + +import asyncio +from types import SimpleNamespace + +import pytest +import pytest_asyncio +import torch + +from tests.engine.duplex.test_session_runner import ( + SESSION_ID, + Harness, + RecordingStagePort, + _fake_encode_audio, + append_audio, + types, +) +from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig +from vllm_omni.engine.duplex import commands +from vllm_omni.engine.duplex.config import DuplexSessionConfig +from vllm_omni.engine.duplex.messages import OpenDuplexSessionMessage +from vllm_omni.engine.duplex.plugin import load_duplex_plugin +from vllm_omni.engine.duplex.session.manager import DuplexSessionManager +from vllm_omni.model_executor.models.nemotron_voicechat.duplex import plugin as plugin_module +from vllm_omni.model_executor.models.nemotron_voicechat.nemotron_voicechat_thinker import ( + NemotronVoiceChatThinkerForConditionalGeneration as Thinker, +) +from vllm_omni.model_executor.models.nemotron_voicechat.pipeline import NEMOTRON_VOICECHAT_PIPELINE +from vllm_omni.protocol.realtime.items import normalize_conversation_item +from vllm_omni.utils.mm_outputs import build_mm_cpu +from vllm_omni.worker.output.payload_build import build_omni_mm_payload + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +@pytest_asyncio.fixture +async def harness(monkeypatch): + monkeypatch.setenv("NEMOTRON_VOICECHAT_LLM_PATH", "test-tokenizer") + special_ids = dict( + nvc_text_bos_id=0, + nvc_text_eos_id=1, + nvc_text_pad_id=12, + nvc_function_sotc_id=20, + nvc_function_eotc_id=21, + nvc_function_eotr_id=22, + nvc_tokenizer_ref="test-tokenizer", + ) + tokenizer = SimpleNamespace( + encode=lambda text, **kwargs: [42], + decode=lambda ids, **kwargs: '[{"name":"lookup","arguments":{"x":1}}]', + ) + monkeypatch.setattr(plugin_module, "_load_tokenizer_runtime", lambda _: (special_ids, tokenizer)) + plugin = load_duplex_plugin(NEMOTRON_VOICECHAT_PIPELINE.duplex_plugin, _fake_encode_audio) + port = RecordingStagePort(stage_count=3) + output: asyncio.Queue = asyncio.Queue() + results: asyncio.Queue = asyncio.Queue() + manager = DuplexSessionManager( + plugin=plugin, + stage_port=port, + output_sink=output, + result_sink=results, + runtime_config=DuplexSessionRuntimeConfig(), + model_config=SimpleNamespace(hf_config=SimpleNamespace(stt_cfg={}), max_model_len=8192), + ) + try: + config = DuplexSessionConfig(model="nemotron", instructions="hi", extra_body={"auto_response": True}) + await manager.handle(OpenDuplexSessionMessage(control_id="open", session_id=SESSION_ID, session_config=config)) + result = await asyncio.wait_for(results.get(), timeout=2) + assert result.ok, result + h = Harness(manager, port, output, results, manager.runners[SESSION_ID]) + await h.settle() + yield h + finally: + await manager.shutdown() + + +async def deliver(h, token_id=12, **metadata): + output = SimpleNamespace( + request_id=h.stage0_request_id(), + finished=False, + multimodal_output={}, + outputs=[SimpleNamespace(text="", token_ids=[token_id], multimodal_output={})], + ) + return await h.deliver_and_settle( + output, + stage_id=0, + segment_finished=True, + segment_token_ids=[token_id], + segment_output_metadata=metadata, + ) + + +@pytest.mark.asyncio +async def test_frame_buffering_prefill_and_resumable_append(harness): + h = harness + await h.run(append_audio(samples=1000)) + assert h.port.submissions == [] + await h.run(append_audio(samples=280)) + await h.run(append_audio(samples=1280)) + first, second = h.port.submissions + assert first.prompt["prompt_token_ids"] == [0, 42, 1, 12] + assert second.prompt["prompt_token_ids"] == [12] + assert first.prompt["model_intermediate_buffer"]["duplex"]["source_input_seq"] == 1 + assert second.prompt["model_intermediate_buffer"]["duplex"]["source_input_seq"] == 2 + assert not first.already_submitted and second.already_submitted + assert first.context.request_id == second.context.request_id == h.stage0_request_id() + assert "error" not in types(h.events) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("request_ids", [("active",), ("other", "active"), ("active", "other")]) +async def test_tool_batches_reach_worker_and_drain_before_ninth_result(harness, request_ids): + h = harness + await h.run(append_audio(samples=1280)) + worker_state = {"function_response_generation": 0, "forced_function_tokens": []} + thinker = Thinker.__new__(Thinker) + torch.nn.Module.__init__(thinker) + thinker._sessions = {"active": worker_state, "other": {"function_response_generation": 2}} + thinker._use_function_head, thinker._dtype = True, torch.float32 + thinker.function_head = torch.nn.Identity() + key = "nvc_function_response_consumed_generation" + assert key in thinker.omni_client_multimodal_output_keys + infos = [{"request_id": rid} for rid in request_ids] + hidden = torch.zeros((len(infos) + 1, 2)) + for generation in range(1, 10): + for token_id in (20, 7, 21): + events = await deliver(h, nvc_function_token=[token_id]) + done = next(event for event in events if event.type == "response.function_call_arguments.done") + item = {"type": "function_call_output", "call_id": done.call_id, "output": "20"} + events = await h.run(commands.CreateItem(item=normalize_conversation_item(item))) + assert "error" not in types(events) + runtime = h.session.runtime_config + assert runtime["nvc_function_response_generation"] == generation + assert runtime["nvc_function_response_batches"][-1] == {"generation": generation, "token_ids": [42]} + if generation < 8: + continue + if generation == 8: + with pytest.raises(plugin_module.NemotronVoiceChatClientRuntimeConfigError, match="faster than"): + h.runner.plugin.runtime_config_for_function_output(h.session.config, runtime, item) + # Copy into model state exactly once, even if the same runtime snapshot is replayed. + Thinker._sync_forced_function_response(worker_state, runtime) + Thinker._sync_forced_function_response(worker_state, runtime) + assert worker_state["forced_function_tokens"] == [42] * generation + assert worker_state["function_response_generation"] == generation + output = thinker.make_omni_output(hidden, model_intermediate_buffer=infos) + for idx, rid in enumerate(request_ids): + payload = build_omni_mm_payload( + combined_multimodal_outputs=None, + mm_cpu=build_mm_cpu(output.multimodal_outputs), + rid=rid, + idx=idx, + start=0 if idx == 0 else 2, + end=2 if idx == 0 else 3, + audio_sparse_output=False, + sparse_mm_index={}, + hidden_seq_len=len(hidden), + scheduled_seq_len=len(hidden), + ) + assert int(payload[key]) == (generation if rid == "active" else 2) + if rid == "active": + await deliver(h, **payload) + assert h.session.runtime_config["nvc_function_response_batches"] == [] + assert h.session.runtime_config["nvc_function_response_generation"] == generation + assert "error" not in types(h.events) diff --git a/tests/model_executor/models/nemotron_voicechat/duplex/conftest.py b/tests/model_executor/models/nemotron_voicechat/duplex/conftest.py deleted file mode 100644 index 134731bf6..000000000 --- a/tests/model_executor/models/nemotron_voicechat/duplex/conftest.py +++ /dev/null @@ -1,20 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project - -"""Nemotron VoiceChat duplex is not on the duplex plugin framework yet. - -RFC vllm-omni#7181 splits the port: this PR lands the unified framework with -MiniCPM-o 4.5, and the follow-up PR moves Nemotron VoiceChat onto -``DuplexModelPlugin``. Until then the modules under test still import the -pre-framework duplex runtime / serving adapter that this PR removed, so -collecting them would fail at import time and abort the whole pytest session. -The tests themselves are unchanged and come back with the port. - -Listed per file rather than globbed, so the coverage loss is visible and shrinks -file by file as the port lands. ``test_input.py`` imports nothing that was -removed and keeps running. -""" - -#: Both import ``DuplexFence`` from ``engine.duplex.messages``, which no longer -#: exports it; ``test_data_plane`` also imports ``engine.duplex.runtime``. -collect_ignore = ["test_data_plane.py", "test_runtime_contract.py"] diff --git a/tests/model_executor/models/nemotron_voicechat/duplex/test_data_plane.py b/tests/model_executor/models/nemotron_voicechat/duplex/test_data_plane.py index 04e57c397..0328fd71b 100644 --- a/tests/model_executor/models/nemotron_voicechat/duplex/test_data_plane.py +++ b/tests/model_executor/models/nemotron_voicechat/duplex/test_data_plane.py @@ -7,9 +7,8 @@ from types import SimpleNamespace import numpy as np import pytest -from vllm_omni.engine.duplex.runtime import duplex_resource_request_id -from vllm_omni.engine.duplex.messages import DuplexFence +from vllm_omni.engine.duplex.contracts import DuplexFence, duplex_resource_request_id from vllm_omni.model_executor.models.nemotron_voicechat.duplex.data_plane import ( NemotronVoiceChatDataPlaneContext, NemotronVoiceChatDataPlaneSession, diff --git a/tests/model_executor/models/nemotron_voicechat/duplex/test_input.py b/tests/model_executor/models/nemotron_voicechat/duplex/test_input.py index 0e8ff504c..751b391f2 100644 --- a/tests/model_executor/models/nemotron_voicechat/duplex/test_input.py +++ b/tests/model_executor/models/nemotron_voicechat/duplex/test_input.py @@ -45,6 +45,15 @@ def test_irregular_browser_packets_emit_exact_ordered_80ms_frame() -> None: np.testing.assert_array_equal(decoded, np.arange(1280, dtype=np.float32)) assert buffer.pending_byte_count == 0 + later = _append(buffer, np.ones(1280), "later") + later.commit() + later.rollback() + reservation.rollback() + assert not buffer.has_reserved() + retry = _append(buffer, [], "retry") + assert retry.payload == reservation.payload + assert not buffer.has_pending() + def test_append_rejects_invalid_frame_contract() -> None: cases = ( diff --git a/tests/model_executor/models/nemotron_voicechat/duplex/test_runtime_contract.py b/tests/model_executor/models/nemotron_voicechat/duplex/test_runtime_contract.py deleted file mode 100644 index 6d2fbaada..000000000 --- a/tests/model_executor/models/nemotron_voicechat/duplex/test_runtime_contract.py +++ /dev/null @@ -1,115 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project - -from __future__ import annotations - -import base64 -from types import SimpleNamespace - -import numpy as np -import pytest -from vllm.sampling_params import SamplingParams - -from vllm_omni.engine.duplex.contracts import DuplexInputMode -from vllm_omni.engine.duplex.messages import DuplexFence -from vllm_omni.model_executor.models.nemotron_voicechat.duplex.runtime import ( - NemotronVoiceChatDuplexRuntimeExtension, -) -from vllm_omni.model_executor.models.nemotron_voicechat.duplex.serving_adapter import ( - NemotronVoiceChatServingRuntimeAdapter, - _render_tool_response, -) -from vllm_omni.model_executor.models.nemotron_voicechat.nemotron_voicechat_thinker import ( - NemotronVoiceChatThinkerForConditionalGeneration, -) - -pytestmark = [pytest.mark.core_model, pytest.mark.cpu] - - -def _frame() -> dict[str, object]: - raw = np.zeros(1280, dtype=np.float32).tobytes() - return { - "type": "audio", - "audio": base64.b64encode(raw).decode("ascii"), - "format": "pcm_f32le", - "sample_rate_hz": 16000, - } - - -def _plan(extension, *, input_seq: int): - return extension.plan_append( - request_id="req", - fence=DuplexFence("sid", incarnation=2, epoch=3), - session_config={}, - runtime_config={ - "nvc_prompt_token_ids": [0, 42, 1], - "nvc_text_pad_id": 12, - "nvc_max_model_len": 8192, - }, - seq=input_seq, - turn_seq=input_seq, - mode=DuplexInputMode.APPEND_AUDIO_CHUNK, - payload=_frame(), - final=False, - sampling_params=SamplingParams(), - ) - - -def test_first_append_prefills_prompt_then_each_append_consumes_one_frame() -> None: - extension = NemotronVoiceChatDuplexRuntimeExtension() - - params = extension.configure_sampling_params( - runtime_config={}, defaults=(SamplingParams(temperature=0.7, max_tokens=4),) - )[0] - assert params.temperature == 0.0 and params.max_tokens == 1 and params.top_p == 1.0 and params.top_k == 0 - - first = _plan(extension, input_seq=1) - later = _plan(extension, input_seq=2) - - assert first.prompt["prompt_token_ids"] == [0, 42, 1, 12] - assert later.prompt["prompt_token_ids"] == [12] - assert first.prompt["model_intermediate_buffer"]["duplex"]["source_input_seq"] == 1 - assert later.prompt["model_intermediate_buffer"]["duplex"]["source_input_seq"] == 2 - - -def test_append_rejects_stage0_context_overflow() -> None: - with pytest.raises(ValueError, match="max_model_len"): - _plan(NemotronVoiceChatDuplexRuntimeExtension(), input_seq=8190) - - -def test_function_output_becomes_versioned_nvidia_channel_tokens() -> None: - encoded: list[str] = [] - adapter = NemotronVoiceChatServingRuntimeAdapter(lambda *_: None) - adapter._tokenizer = SimpleNamespace( - encode=lambda text, **_kwargs: encoded.append(text) or [31, 32, 33], - ) - - first = adapter.runtime_config_for_function_output( - {"type": "function_call_output", "call_id": "call-1", "output": '{"result":20}'}, - {}, - ) - second = adapter.runtime_config_for_function_output( - {"type": "function_call_output", "call_id": "call-2", "output": "plain text"}, - first, - ) - - assert _render_tool_response('{"result":20}') == '[{"result":20}]' - assert encoded == [ - '[{"result":20}]', - '["plain text"]', - ] - assert first["nvc_function_response_generation"] == 1 - assert second["nvc_function_response_generation"] == 2 - assert second["nvc_function_response_token_ids"] == [31, 32, 33] - assert second["nvc_function_response_call_id"] == "call-2" - assert second["nvc_function_response_batches"] == [ - {"generation": 1, "call_id": "call-1", "token_ids": [31, 32, 33]}, - {"generation": 2, "call_id": "call-2", "token_ids": [31, 32, 33]}, - ] - - session = {"function_response_generation": 0, "forced_function_tokens": []} - NemotronVoiceChatThinkerForConditionalGeneration._sync_forced_function_response(session, second) - - assert session["function_response_generation"] == 2 - assert session["forced_function_token"] == 31 - assert session["forced_function_tokens"] == [31, 32, 33, 31, 32, 33] diff --git a/vllm_omni/clients/nemotron_voicechat.py b/vllm_omni/clients/nemotron_voicechat.py new file mode 100644 index 000000000..671b093e8 --- /dev/null +++ b/vllm_omni/clients/nemotron_voicechat.py @@ -0,0 +1,29 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Client-side duplex session preset for Nemotron VoiceChat.""" + +from __future__ import annotations + +from vllm_omni.clients.duplex import AudioFormat, SessionConfig + +__all__ = ["create_duplex_session_config"] + + +def create_duplex_session_config( + *, + instructions: str | None = None, + tools: list[dict[str, object]] | None = None, + idle_timeout_s: float | None = None, + extra_body: dict[str, object] | None = None, +) -> SessionConfig: + """Use 16 kHz float32 input and 22.05 kHz PCM16 output; append 80 ms frames.""" + session_extra_body = dict(extra_body or {}) + if tools is not None: + session_extra_body["realtime_tools"] = tools + return SessionConfig( + input_audio=AudioFormat("pcm_f32le", 16_000), + output_audio=AudioFormat("pcm16", 22_050), + instructions=instructions, + idle_timeout_s=idle_timeout_s, + extra_body=session_extra_body, + ) diff --git a/vllm_omni/engine/duplex/intermediate.py b/vllm_omni/engine/duplex/intermediate.py index 3beb73099..175259dd9 100644 --- a/vllm_omni/engine/duplex/intermediate.py +++ b/vllm_omni/engine/duplex/intermediate.py @@ -3,7 +3,11 @@ from __future__ import annotations -from typing import TypedDict +from collections.abc import Mapping +from typing import TYPE_CHECKING, TypedDict + +if TYPE_CHECKING: + from vllm_omni.engine.duplex.contracts import DuplexFence class DuplexIntermediateBuffer(TypedDict, total=False): @@ -32,6 +36,49 @@ class DuplexIntermediateBuffer(TypedDict, total=False): mel_spec: object +def build_duplex_append_prompt( + *, + request_id: str, + fence: DuplexFence, + session_config: Mapping[str, object], + runtime_config: Mapping[str, object], + seq: int, + turn_seq: int, + payload: object, + final: bool, + prompt_token_ids: list[int], + model_fields: Mapping[str, object], +) -> dict[str, object]: + """Wrap model-selected tokens and payload in the shared append envelope. + + Models own token selection and extra worker fields; the framework owns + request identity, sequencing and shallow config snapshots. + """ + return { + "prompt_token_ids": prompt_token_ids, + "model_intermediate_buffer": { + "request_id": request_id, + "global_request_id": [fence.session_id], + "duplex": { + **model_fields, + "data_plane": True, + "fence": fence, + "session_id": fence.session_id, + "epoch": fence.epoch, + "seq": seq, + "turn_id": fence.turn_id, + "turn_seq": turn_seq, + "mode": "append_audio_chunk", + "payload": payload, + "final": final, + "session_config": dict(session_config), + "runtime_config": dict(runtime_config), + "scheduler_token_budget": len(prompt_token_ids), + }, + }, + } + + def build_duplex_intermediate_buffer( *, request_id: str, @@ -62,12 +109,12 @@ def build_duplex_intermediate_buffer( return buffer -def set_ref_audio(buffer: dict[str, object], waveform: object, sample_rate_hz: int) -> None: +def set_ref_audio(buffer: DuplexIntermediateBuffer, waveform: object, sample_rate_hz: int) -> None: buffer.setdefault("codes", {})["ref"] = waveform buffer.setdefault("meta", {})["ref_audio_sr"] = int(sample_rate_hz) -def set_tts_handoff(buffer: dict[str, object], token_ids: object | None, hidden_states: object | None) -> None: +def set_tts_handoff(buffer: DuplexIntermediateBuffer, token_ids: object | None, hidden_states: object | None) -> None: """Store the AR-to-TTS handoff used by the full-duplex stage bridge.""" if token_ids is not None: buffer.setdefault("ids", {})["tts"] = token_ids diff --git a/vllm_omni/engine/duplex/plugin.py b/vllm_omni/engine/duplex/plugin.py index fee1c20ad..4fb23977b 100644 --- a/vllm_omni/engine/duplex/plugin.py +++ b/vllm_omni/engine/duplex/plugin.py @@ -38,6 +38,21 @@ class DuplexRuntimeConfigError(ValueError): self.code = code +def reject_private_runtime_keys( + extra_body: object, + private_runtime_config_keys: frozenset[str], + *, + message: str, + error_cls: type[DuplexRuntimeConfigError] = DuplexRuntimeConfigError, +) -> None: + """Reject client overrides while preserving each plugin's error contract.""" + if not isinstance(extra_body, dict): + return + private_keys = sorted(private_runtime_config_keys.intersection(extra_body)) + if private_keys: + raise error_cls(message + ", ".join(private_keys)) + + def reject_changed_runtime_value( new_value: object, current_value: object, @@ -146,6 +161,64 @@ class DuplexModelSessionState(ABC): def clear_continuation(self) -> None: ... +@dataclass(slots=True) +class DefaultDuplexModelSessionState(DuplexModelSessionState): + """Framework-owned flag anatomy, implemented once. + + The flag set above is the session runner's contract with the model (commit + retention, deferred response/creates, silence-continuation bookkeeping); it + is identical for every lockstep or frame-locked model, so a plugin only + supplies its input packetizer as ``audio_buffer``. + """ + + audio_buffer: PcmAppendBuffer + input_since_commit: bool = False + speech_since_commit: bool = False + context_locked: bool = False + committed_audio_payload: dict[str, object] | None = None + committed_audio_operation_id: str | None = None + committed_audio_reserved_bytes: int = 0 + deferred_response_create: bool = False + deferred_precreate_response: bool = False + continuation_owner_id: str | None = None + continuation_units: int = 0 + pending_silence_task: asyncio.Task[bool] | None = None + pending_silence_owner_id: str | None = None + # Deadline-aligned silence continuation state: the monotonic submission + # time of the most recent native input unit and the next silence + # continuation deadline. A real (non-silence) input resets the chain. + last_native_submit_monotonic: float | None = None + silence_deadline_monotonic: float | None = None + + def retain_committed_audio( + self, + payload: dict[str, object], + *, + operation_id: str | None, + reserved_bytes: int = 0, + ) -> None: + self.committed_audio_payload = payload + self.committed_audio_operation_id = operation_id + self.committed_audio_reserved_bytes += max(0, int(reserved_bytes)) + + def clear_committed_audio(self) -> int: + reserved_bytes = self.committed_audio_reserved_bytes + self.committed_audio_payload = None + self.committed_audio_operation_id = None + self.committed_audio_reserved_bytes = 0 + self.deferred_response_create = False + self.deferred_precreate_response = False + return reserved_bytes + + def clear_continuation(self) -> None: + self.continuation_owner_id = None + self.continuation_units = 0 + self.pending_silence_task = None + self.pending_silence_owner_id = None + self.last_native_submit_monotonic = None + self.silence_deadline_monotonic = None + + class DuplexDataPlane(ABC): """Projects raw stage outputs of one model into internal duplex events.""" @@ -397,6 +470,20 @@ class DuplexModelPlugin(ABC): del config, current, item return None + def runtime_config_after_model_output( + self, + current: Mapping[str, object], + output_metadata: Mapping[str, object], + ) -> dict[str, object] | None: + """Return a runtime-config patch after a model output is observed. + + Plugins may use model-owned output metadata to retire server-side + runtime state that has been consumed by the worker. The default keeps + the framework unaware of model-specific metadata. + """ + del current, output_metadata + return None + def load_duplex_plugin(path: str, encode_audio: EncodeAudio) -> DuplexModelPlugin: module_name, separator, attribute_name = path.rpartition(".") @@ -448,6 +535,7 @@ def coerce_int(value: object) -> int | None: __all__ = [ + "DefaultDuplexModelSessionState", "DuplexDataPlane", "DuplexModelPlugin", "DuplexModelSessionState", diff --git a/vllm_omni/engine/duplex/session/model_channel.py b/vllm_omni/engine/duplex/session/model_channel.py index ef3655995..1c2d67c58 100644 --- a/vllm_omni/engine/duplex/session/model_channel.py +++ b/vllm_omni/engine/duplex/session/model_channel.py @@ -569,6 +569,12 @@ class ModelChannel: ): return engine_output = self._build_stage_output(item) + runtime_config = self._ctx.plugin.runtime_config_after_model_output( + dict(session.runtime_config), + item.context.segment_output_metadata, + ) + if runtime_config is not None: + session.replace_runtime_config(runtime_config) drain_result = {"data_plane_outputs": [engine_output]} close_reason, emitted_response = await self._send_model_output_events( drain_result, expected_epoch=expected_epoch diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py index 4a6ac7606..16203ecd5 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py @@ -32,11 +32,13 @@ from vllm_omni.engine.duplex.contracts import ( DuplexOutputAction, DuplexOutputDecision, ) +from vllm_omni.engine.duplex.intermediate import build_duplex_append_prompt from vllm_omni.engine.duplex.plugin import ( DuplexModelPlugin, DuplexRuntimeConfigError, EncodeAudio, reject_changed_runtime_value, + reject_private_runtime_keys, ) from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import ( minicpmo45_native_capabilities, @@ -269,29 +271,18 @@ def build_duplex_data_plane_prompt( and payload.get("force_listen") is not True ): payload = {**payload, "force_listen": True} - return { - "prompt_token_ids": [token_id] * token_budget, - "model_intermediate_buffer": { - "request_id": request_id, - "global_request_id": [fence.session_id], - "duplex": { - "fence": fence, - "session_id": fence.session_id, - "epoch": fence.epoch, - "seq": seq, - "turn_id": fence.turn_id, - "turn_seq": turn_seq, - "mode": "append_audio_chunk", - "payload": payload, - "final": final, - "data_plane": True, - "session_config": dict(session_config), - "runtime_config": dict(runtime_config), - "scheduler_token_budget": token_budget, - "scheduler_token_id": token_id, - }, - }, - } + return build_duplex_append_prompt( + request_id=request_id, + fence=fence, + session_config=session_config, + runtime_config=runtime_config, + seq=seq, + turn_seq=turn_seq, + payload=payload, + final=final, + prompt_token_ids=[token_id] * token_budget, + model_fields={"scheduler_token_id": token_id}, + ) # ---- engine policy helpers: listen decision ---- @@ -725,13 +716,12 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin): return minicpmo45_native_capabilities(max_sessions=max_sessions) def validate_client_extra_body(self, extra_body: object) -> None: - if not isinstance(extra_body, dict): - return - private_keys = sorted(PRIVATE_RUNTIME_CONFIG_KEYS.intersection(extra_body)) - if private_keys: - raise MiniCPMO45ClientRuntimeConfigError( - "duplex runtime configuration is server-owned: " + ", ".join(private_keys) - ) + reject_private_runtime_keys( + extra_body, + self.private_runtime_config_keys, + message="duplex runtime configuration is server-owned: ", + error_cls=MiniCPMO45ClientRuntimeConfigError, + ) async def prepare_runtime_config( self, config: DuplexSessionConfig, *, model_config: ModelConfig | None diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py index fe53ec653..f36bddc67 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py @@ -3,62 +3,20 @@ from __future__ import annotations -import asyncio from dataclasses import dataclass, field -from vllm_omni.engine.duplex.plugin import DuplexModelSessionState +from vllm_omni.engine.duplex.plugin import DefaultDuplexModelSessionState from vllm_omni.model_executor.models.minicpmo_4_5.duplex.input import ( MiniCPMO45PcmAppendBuffer, ) @dataclass(slots=True) -class MiniCPMO45ServingSessionState(DuplexModelSessionState): - """Mutable model-owned state of one MiniCPM duplex session (owned by the session runner).""" +class MiniCPMO45ServingSessionState(DefaultDuplexModelSessionState): + """Mutable model-owned state of one MiniCPM duplex session (owned by the session runner). + + The runner-facing flag contract lives on ``DefaultDuplexModelSessionState``; + MiniCPM-o only binds its ~1 s PCM unit buffer. + """ audio_buffer: MiniCPMO45PcmAppendBuffer = field(default_factory=MiniCPMO45PcmAppendBuffer) - input_since_commit: bool = False - speech_since_commit: bool = False - context_locked: bool = False - committed_audio_payload: dict[str, object] | None = None - committed_audio_operation_id: str | None = None - committed_audio_reserved_bytes: int = 0 - deferred_response_create: bool = False - deferred_precreate_response: bool = False - continuation_owner_id: str | None = None - continuation_units: int = 0 - pending_silence_task: asyncio.Task[bool] | None = None - pending_silence_owner_id: str | None = None - # Deadline-aligned silence continuation state: the monotonic submission - # time of the most recent native input unit and the next silence - # continuation deadline. A real (non-silence) input resets the chain. - last_native_submit_monotonic: float | None = None - silence_deadline_monotonic: float | None = None - - def retain_committed_audio( - self, - payload: dict[str, object], - *, - operation_id: str | None, - reserved_bytes: int = 0, - ) -> None: - self.committed_audio_payload = payload - self.committed_audio_operation_id = operation_id - self.committed_audio_reserved_bytes += max(0, int(reserved_bytes)) - - def clear_committed_audio(self) -> int: - reserved_bytes = self.committed_audio_reserved_bytes - self.committed_audio_payload = None - self.committed_audio_operation_id = None - self.committed_audio_reserved_bytes = 0 - self.deferred_response_create = False - self.deferred_precreate_response = False - return reserved_bytes - - def clear_continuation(self) -> None: - self.continuation_owner_id = None - self.continuation_units = 0 - self.pending_silence_task = None - self.pending_silence_owner_id = None - self.last_native_submit_monotonic = None - self.silence_deadline_monotonic = None diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/__init__.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/__init__.py index ab79ab4d2..1531388bd 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/__init__.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/__init__.py @@ -1,4 +1,16 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project -"""Native Full-Duplex integration for NVIDIA NemotronLabs VoiceChat.""" +"""Native Full-Duplex integration for NVIDIA NemotronLabs VoiceChat. + +The single ``duplex_plugin`` seam (RFC vllm-omni#7181): +``NemotronVoiceChatDuplexPlugin`` carries the engine and session policies, +``data_plane.py`` projects stage outputs and ``input.py`` packetizes 80 ms PCM +frames into the framework's per-session model state. +""" + +from vllm_omni.model_executor.models.nemotron_voicechat.duplex.plugin import ( + NemotronVoiceChatDuplexPlugin, +) + +__all__ = ["NemotronVoiceChatDuplexPlugin"] diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/capabilities.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/capabilities.py new file mode 100644 index 000000000..dd93e243d --- /dev/null +++ b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/capabilities.py @@ -0,0 +1,55 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + +from vllm_omni.engine.duplex.config import DuplexCapabilities + + +def nemotron_voicechat_capabilities(*, max_sessions: int = 1) -> DuplexCapabilities: + """The Nemotron VoiceChat native-duplex preset of the generic capability payload. + + One 80 ms acoustic frame is appended to one resumable Stage-0 request, so + the scheduler owns the frame clock (``requires_model_runner_kv``) and the + model decides by itself when to speak (``supports_model_native_turn_policy``). + Barge-in and mid-incarnation context edits are not supported: the fused + prompt/audio KV cannot be revised in place. + """ + supports_multi_session = max_sessions > 1 + return DuplexCapabilities( + supports_model_native_turn_policy=True, + supports_external_turn_signal=False, + supports_client_commit=True, + supports_barge_in=False, + supports_playback_ack=True, + supports_input_append=True, + supports_replace_latest_chunk=False, + supports_reencode_context=False, + supports_rollback_to_checkpoint=False, + supports_turn_commit_only=False, + supports_kv_lease=False, + supports_core_kv_lease=False, + supports_model_internal_state=True, + supports_stage_resumption=True, + supports_scheduler_native_append=False, + supports_core_resumable_request=True, + supports_stage_connector_handoff=True, + supports_independent_io_streams=True, + supports_realtime_endpoint=True, + supports_multi_session=supports_multi_session, + supports_multi_session_same_replica=False, + supports_session_lease=True, + supports_session_resume=True, + session_admission_mode="engine_managed", + supports_audio_truncate=False, + requires_model_runner_kv=True, + requires_native_stage_role=True, + adapter_patterns=["scheduler_data_plane"], + signal_sources=["model_native", "client_event"], + stage_handoff_transport="scheduler_data_plane", + chunk_period_ms=80, + target_barge_in_latency_ms=None, + ) + + +__all__ = ["nemotron_voicechat_capabilities"] diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/data_plane.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/data_plane.py index 0fb3f93c7..156fddd8a 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/data_plane.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/data_plane.py @@ -6,7 +6,7 @@ from __future__ import annotations import json -from collections.abc import Callable, Iterator, Mapping +from collections.abc import Iterator, Mapping from dataclasses import dataclass, field from typing import Any from uuid import uuid4 @@ -18,14 +18,13 @@ from vllm.logger import init_logger from vllm_omni.engine.duplex.contracts import ( duplex_resource_request_belongs_to_session, ) +from vllm_omni.engine.duplex.plugin import DuplexDataPlane, EncodeAudio from vllm_omni.outputs.duplex import ( get_duplex_output_decision, ) logger = init_logger(__name__) -EncodeAudio = Callable[[object, int, str, float | None], str | None] - @dataclass(frozen=True, slots=True) class NemotronVoiceChatDataPlaneContext: @@ -137,7 +136,7 @@ def _speech_end_event(request_id: str) -> dict[str, object]: } -class NemotronVoiceChatDataPlaneSession: +class NemotronVoiceChatDataPlaneSession(DuplexDataPlane): """Join frame-locked text/function outputs with Stage-2 audio.""" def __init__(self, encode_audio: EncodeAudio) -> None: @@ -148,7 +147,14 @@ class NemotronVoiceChatDataPlaneSession: self._special_ids: dict[str, int] | None = None def configure_runtime(self, runtime_config: Mapping[str, object], *, tokenizer: Any | None = None) -> None: - """Install the serving-resolved tokenizer contract as the sole truth.""" + """Install the serving-resolved tokenizer contract as the sole truth. + + The data plane instance is engine-scoped (one per loaded plugin) while + :meth:`configure_runtime` runs once per session open. The contract is a + function of the checkpoint, so reconfiguration must be idempotent — a + second session that resolves a *different* contract indicates a server + bug or a mixed deployment and is rejected in the open path. + """ try: special_ids = { "bos": int(runtime_config["nvc_text_bos_id"]), @@ -162,9 +168,15 @@ class NemotronVoiceChatDataPlaneSession: tokenizer_ref = runtime_config.get("nvc_tokenizer_ref") if not isinstance(tokenizer_ref, str) or not tokenizer_ref: raise ValueError("Nemotron VoiceChat data plane requires nvc_tokenizer_ref") + if self._special_ids is not None and (self._special_ids != special_ids or self._tokenizer_ref != tokenizer_ref): + raise ValueError( + "Nemotron VoiceChat data plane is engine-scoped: a later session resolved a " + f"different token contract ({self._tokenizer_ref!r} -> {tokenizer_ref!r})" + ) self._special_ids = special_ids self._tokenizer_ref = tokenizer_ref - self._tokenizer = tokenizer + if tokenizer is not None: + self._tokenizer = tokenizer def begin_request(self, request_id: str) -> None: self._requests.setdefault(request_id, _RequestState()).terminal = False @@ -185,24 +197,23 @@ class NemotronVoiceChatDataPlaneSession: if duplex_resource_request_belongs_to_session(request_id, session_id): self._requests.pop(request_id, None) - def _load_tokenizer(self): - if self._tokenizer is None: - from transformers import AutoTokenizer - - if self._tokenizer_ref is None: - raise RuntimeError("Nemotron VoiceChat data plane was used before runtime configuration") - self._tokenizer = AutoTokenizer.from_pretrained(self._tokenizer_ref, trust_remote_code=False) - return self._tokenizer - def _ids(self) -> dict[str, int]: if self._special_ids is None: raise RuntimeError("Nemotron VoiceChat data plane was used before runtime configuration") return self._special_ids def _decode(self, token_ids: list[int]) -> str: + """Decode token ids on the projection path; the tokenizer must already be installed. + + ``configure_runtime`` (called from the plugin's ``prepare_runtime_config`` during + session open, off the event loop) is the only place a tokenizer is loaded. + """ if not token_ids: return "" - return str(self._load_tokenizer().decode(token_ids, skip_special_tokens=True)) + tokenizer = self._tokenizer + if tokenizer is None: + raise RuntimeError("Nemotron VoiceChat data plane was used before runtime configuration") + return str(tokenizer.decode(token_ids, skip_special_tokens=True)) def project( self, diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/input.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/input.py index 860671cb2..dbd474ede 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/input.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/input.py @@ -10,6 +10,12 @@ import binascii import numpy as np import pybase64 as base64 +from vllm_omni.engine.duplex.pcm_reservation import ( + commit_ordered_reservation, + rollback_ordered_reservation, +) +from vllm_omni.engine.duplex.plugin import PcmAppendBuffer, PcmAppendReservation + NEMOTRON_VOICECHAT_SAMPLE_RATE = 16000 NEMOTRON_VOICECHAT_FRAME_SAMPLES = 1280 _SAMPLE_BYTES = 4 @@ -60,7 +66,7 @@ def _frame_payload(raw: bytes, *, final: bool) -> dict[str, object]: } -class NemotronVoiceChatPcmAppendReservation: +class NemotronVoiceChatPcmAppendReservation(PcmAppendReservation): def __init__( self, owner: NemotronVoiceChatPcmAppendBuffer, @@ -72,7 +78,7 @@ class NemotronVoiceChatPcmAppendReservation: self._owner = owner self.operation_id = operation_id self.payload = payload - self.raw = raw + self._raw = raw self._active = True @property @@ -81,33 +87,16 @@ class NemotronVoiceChatPcmAppendReservation: @property def byte_count(self) -> int: - return len(self.raw) + return len(self._raw) def commit(self) -> None: - if not self._active: - return - self._active = False - if self in self._owner._reservations: - self._owner._reservations.remove(self) + commit_ordered_reservation(self._owner._reservations, self, head_only=False) def rollback(self) -> None: - if not self._active: - return - try: - index = self._owner._reservations.index(self) - except ValueError: - self._active = False - return - restore = bytearray() - for reservation in self._owner._reservations[index:]: - if reservation._active: - restore.extend(reservation.raw) - reservation._active = False - del self._owner._reservations[index:] - self._owner._buffer[:0] = restore + rollback_ordered_reservation(self._owner._reservations, self, self._owner._buffer, active_only=True) -class NemotronVoiceChatPcmAppendBuffer: +class NemotronVoiceChatPcmAppendBuffer(PcmAppendBuffer): """Adapt Realtime packets to the current full-duplex PCM buffer contract. The model consumes exactly one 1280-sample frame per scheduler append. diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/plugin.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/plugin.py new file mode 100644 index 000000000..b9d49c2b2 --- /dev/null +++ b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/plugin.py @@ -0,0 +1,555 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Nemotron VoiceChat full-duplex model plugin: engine policy and session policy in one class. + +Engine policy (frame-locked resumable Stage-0 prompt, greedy 1-token sampling, +Stage-2 direct audio output) and session policy (tool rendering, function-channel +injection, per-session PCM state) both run engine-side, so one class owns them. + +Timeline contract (see ``nemotron_voicechat_thinker.py``): the Stage-0 prompt is +``[bos] + instructions + [eos]`` with the PAD text token fused over acoustic +frame 0; every scheduler append afterwards carries exactly one 80 ms acoustic +frame and decodes exactly one token. ``ignore_eos`` keeps the request alive; +the tokenizer's EOS *is* the model's "end of speech" signal, and the PAD token +(````) its "keep listening" signal. +""" + +from __future__ import annotations + +import asyncio +import json +import os +from collections.abc import Mapping +from copy import deepcopy +from typing import TYPE_CHECKING, Any + +from vllm.sampling_params import RequestOutputKind, SamplingParams + +from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig +from vllm_omni.engine.duplex.contracts import ( + DuplexAppendPlan, + DuplexFence, + DuplexOutputAction, + DuplexOutputDecision, +) +from vllm_omni.engine.duplex.intermediate import build_duplex_append_prompt +from vllm_omni.engine.duplex.plugin import ( + DefaultDuplexModelSessionState, + DuplexModelPlugin, + DuplexRuntimeConfigError, + EncodeAudio, + reject_private_runtime_keys, +) +from vllm_omni.model_executor.models.nemotron_voicechat.duplex.capabilities import ( + nemotron_voicechat_capabilities, +) +from vllm_omni.model_executor.models.nemotron_voicechat.duplex.data_plane import ( + NemotronVoiceChatDataPlaneContext, + NemotronVoiceChatDataPlaneSession, +) +from vllm_omni.model_executor.models.nemotron_voicechat.duplex.input import ( + NEMOTRON_VOICECHAT_FRAME_SAMPLES, + NemotronVoiceChatPcmAppendBuffer, + decode_pcm_f32le, +) + +if TYPE_CHECKING: + from transformers import PreTrainedTokenizerBase + from vllm.config import ModelConfig + +_DEFAULT_SYSTEM_PROMPT = ( + "You are an AI voice assistant developed by NVIDIA. " + "Your name is NVIDIA Voice Chat. " + "Answer in a spoken, conversational style rather than a written one. " + "Do not repeat the same sentence over and over again. " + "Start the conversation by greeting the user." +) + +#: Server-owned runtime_config keys a client must not override through extra_body. +PRIVATE_RUNTIME_CONFIG_KEYS = frozenset( + { + "nvc_prompt_token_ids", + "nvc_text_bos_id", + "nvc_text_eos_id", + "nvc_text_pad_id", + "nvc_function_sotc_id", + "nvc_function_eotc_id", + "nvc_function_eotr_id", + "nvc_max_model_len", + "nvc_tokenizer_ref", + "nvc_tools_signature", + "nvc_function_response_generation", + "nvc_function_response_batches", + } +) + + +class NemotronVoiceChatClientRuntimeConfigError(DuplexRuntimeConfigError): + pass + + +#: Bound batches awaiting worker ownership so per-append runtime snapshots +#: stay bounded. Once acknowledged, their tokens remain in the model-owned +#: queue and drain at one forced token per 80 ms frame. +_MAX_PENDING_FUNCTION_RESPONSE_BATCHES = 8 + + +def _stt_config(model_config: Any) -> dict[str, Any]: + hf_config = getattr(model_config, "hf_config", None) + stt_cfg = getattr(hf_config, "stt_cfg", None) + if not isinstance(stt_cfg, dict): + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat checkpoint STT configuration is unavailable" + ) + return stt_cfg + + +def _normalized_tools(config: DuplexSessionConfig) -> tuple[list[dict[str, object]], str]: + raw_tools = config.extra_body.get("realtime_tools") + if raw_tools is None: + return [], "[]" + if not isinstance(raw_tools, list): + raise NemotronVoiceChatClientRuntimeConfigError("Nemotron VoiceChat tools must be a list") + if len(raw_tools) > 5: + raise NemotronVoiceChatClientRuntimeConfigError("Nemotron VoiceChat supports at most 5 tools per session") + + normalized: list[dict[str, object]] = [] + for index, tool in enumerate(raw_tools): + if not isinstance(tool, dict): + raise NemotronVoiceChatClientRuntimeConfigError(f"Nemotron VoiceChat tool {index} must be an object") + function = tool.get("function", tool) + if not isinstance(function, dict): + raise NemotronVoiceChatClientRuntimeConfigError( + f"Nemotron VoiceChat tool {index} has no function definition" + ) + definition = {key: value for key, value in function.items() if key != "type"} + if not isinstance(definition.get("name"), str) or not str(definition["name"]).strip(): + raise NemotronVoiceChatClientRuntimeConfigError(f"Nemotron VoiceChat tool {index} requires a name") + normalized.append(definition) + signature = json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return normalized, signature + + +def _render_tool_prompt(instructions: str, tools: list[dict[str, object]]) -> str: + if not tools: + return instructions + available = json.dumps(tools, ensure_ascii=False, sort_keys=True) + return ( + f"{instructions}\n\n" + "You can use the following tools to assist the user if required:" + f"\n{available}\n\n" + "If you decide to call any tool(s), use the following format:\n" + '[{"name": "tool_name1", "arguments": "tool_args1"}, ' + '{"name": "tool_name2", "arguments": "tool_args2"}]\n\n' + "The user will execute tool-calls and return responses from tool(s) in this format:\n" + '[{"tool_response1"}, {"tool_response2"}]\n\n' + "Based on the tool responses, you can call additional tools if needed, correct tool calls if any " + "errors are found, or just respond to the user." + ) + + +def _render_tool_response(output: str) -> str: + """Serialize one Realtime function result using NVIDIA's function channel.""" + try: + value = json.loads(output) + except json.JSONDecodeError: + value = output + payload = json.dumps([value], ensure_ascii=True, separators=(",", ":")) + return f"{payload}" + + +def _require_native_full_duplex(config: DuplexSessionConfig) -> None: + extra_body = config.extra_body + enabled = extra_body.get("auto_response") is True or extra_body.get("full_duplex") is True + if not enabled: + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat currently supports model-native full-duplex streaming only; " + "set extra_body.auto_response=true", + code="unsupported_nemotron_duplex_mode", + ) + + +def _load_tokenizer_runtime(model_config: Any) -> tuple[dict[str, object], Any]: + """Load the session tokenizer and resolve the checkpoint's special-token ids.""" + from transformers import AutoTokenizer + + stt_cfg = _stt_config(model_config) + tokenizer_ref = os.environ.get("NEMOTRON_VOICECHAT_LLM_PATH") or stt_cfg.get( + "pretrained_llm", "nvidia/NVIDIA-Nemotron-Nano-9B-v2" + ) + tokenizer = AutoTokenizer.from_pretrained(tokenizer_ref, trust_remote_code=False) + + def token(name: str, default: str) -> int: + value = tokenizer.convert_tokens_to_ids(stt_cfg.get(name, default)) + if value is None: + raise NemotronVoiceChatClientRuntimeConfigError(f"Nemotron VoiceChat tokenizer does not define {name}") + return int(value) + + runtime = { + "nvc_text_bos_id": token("bos_token", ""), + "nvc_text_eos_id": token("eos_token", ""), + "nvc_text_pad_id": token("pad_token", ""), + "nvc_function_sotc_id": int(tokenizer.convert_tokens_to_ids("")), + "nvc_function_eotc_id": int(tokenizer.convert_tokens_to_ids("")), + "nvc_function_eotr_id": int(tokenizer.convert_tokens_to_ids("")), + "nvc_tokenizer_ref": str(tokenizer_ref), + } + return runtime, tokenizer + + +def _render_prompt_ids( + tokenizer: PreTrainedTokenizerBase, + rendered_prompt: str, + bos_id: int, + eos_id: int, +) -> list[int]: + return [bos_id] + list(tokenizer.encode(rendered_prompt, add_special_tokens=False)) + [eos_id] + + +def _plain_token_ids(value: object, *, name: str) -> list[int]: + if not isinstance(value, list | tuple) or not value: + raise ValueError(f"Nemotron VoiceChat runtime requires non-empty {name}") + try: + return [int(token_id) for token_id in value] + except (TypeError, ValueError) as exc: + raise ValueError(f"Nemotron VoiceChat {name} must contain token ids") from exc + + +def _positive_int(value: object, *, name: str) -> int: + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"Nemotron VoiceChat requires a positive {name}") from exc + if parsed <= 0: + raise ValueError(f"Nemotron VoiceChat requires a positive {name}") + return parsed + + +class NemotronVoiceChatDuplexPlugin(DuplexModelPlugin): + """Nemotron-owned sampling policy, append planning, session state and output projection.""" + + plugin_id = "nemotron_voicechat" + private_runtime_config_keys = PRIVATE_RUNTIME_CONFIG_KEYS + #: One silence continuation unit is one 80 ms acoustic frame at 16 kHz. + silence_continuation_samples = NEMOTRON_VOICECHAT_FRAME_SAMPLES + + def __init__(self, encode_audio: EncodeAudio) -> None: + super().__init__(encode_audio) + self.data_plane = NemotronVoiceChatDataPlaneSession(encode_audio) + # The tokenizer only resolves the special-token ids and renders tool + # prompts; it is the same for every session of a checkpoint, so load it + # once per tokenizer ref, off the orchestrator loop. + self._tokenizers: dict[str, Any] = {} + self._tokenizer_lock = asyncio.Lock() + + # ---- engine policy (the frame-locked resumable Stage-0 request) ---- + + def configure_sampling_params( + self, + *, + runtime_config: dict[str, object], + defaults: tuple[object, ...], + ) -> tuple[object, ...]: + del runtime_config + if not defaults: + return defaults + # Every scheduler segment is one transport wake of a long-lived + # request. CUMULATIVE output makes the multimodal output processor + # concatenate all earlier Stage-2 waveforms and resend them on every + # wake (80, 160, 240, ... ms), causing O(n^2) audio replay. Duplex + # consumers require per-wake deltas at every stage. + configured = [params.clone() if isinstance(params, SamplingParams) else params for params in defaults] + for params in configured: + if isinstance(params, SamplingParams): + params.output_kind = RequestOutputKind.DELTA + stage0 = configured[0] + if isinstance(stage0, SamplingParams): + stage0.temperature = 0.0 + stage0.top_p = 1.0 + stage0.top_k = 0 + stage0.max_tokens = 1 + stage0.ignore_eos = True + return tuple(configured) + + def plan_append( + self, + *, + request_id: str, + fence: DuplexFence, + session_config: dict[str, object], + runtime_config: dict[str, object], + seq: int, + turn_seq: int, + payload: object, + final: bool, + sampling_params: object, + ) -> DuplexAppendPlan: + del sampling_params + decode_pcm_f32le(payload, exact_frame=True) + normalized_payload = dict(payload) + prompt_ids = _plain_token_ids( + runtime_config.get("nvc_prompt_token_ids"), + name="nvc_prompt_token_ids", + ) + max_model_len = _positive_int( + runtime_config.get("nvc_max_model_len"), + name="nvc_max_model_len", + ) + # ``seq`` is the per-epoch frame counter; the worker maps it 1:1 onto + # timeline positions of the resumable Stage-0 request. + required_model_len = len(prompt_ids) + seq + 1 + if required_model_len > max_model_len: + raise ValueError( + "Nemotron VoiceChat native duplex session exceeds the Stage-0 " + f"max_model_len: prompt_tokens={len(prompt_ids)} + input_frames={seq} + " + f"sampled_token=1 gives {required_model_len} > {max_model_len}; " + "start a new session or raise Stage 0 max_model_len" + ) + try: + pad_id = int(runtime_config["nvc_text_pad_id"]) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("Nemotron VoiceChat runtime requires nvc_text_pad_id") from exc + scheduler_prompt = prompt_ids + [pad_id] if seq <= 1 else [pad_id] + return DuplexAppendPlan( + prompt=build_duplex_append_prompt( + request_id=request_id, + fence=fence, + session_config=session_config, + runtime_config=runtime_config, + seq=seq, + turn_seq=turn_seq, + payload=normalized_payload, + final=final, + prompt_token_ids=scheduler_prompt, + model_fields={"source_input_seq": seq}, + ) + ) + + def decide_output( + self, + *, + stage_id: int, + final_stage_id: int, + segment_finished: bool, + segment_token_ids: tuple[int, ...], + segment_output_metadata: dict[str, object], + output: object, + ) -> DuplexOutputDecision | None: + del final_stage_id, segment_finished, output + if stage_id != 0: + return None + metadata = dict(segment_output_metadata) + metadata["nvc_text_token_ids"] = list(segment_token_ids) + # Stage 0 is a client-visible side channel even though Stage 2 is the + # configured audio response stage. Mark it explicitly so the shared + # collector does not discard it while waiting for the final stage. + metadata["duplex_direct_response"] = True + return DuplexOutputDecision( + action=DuplexOutputAction.DIRECT_RESPONSE, + metadata=metadata, + final_output_type="text", + ) + + # ---- session policy ---- + + def create_session_state(self) -> DefaultDuplexModelSessionState: + return DefaultDuplexModelSessionState(audio_buffer=NemotronVoiceChatPcmAppendBuffer()) + + def capabilities(self, *, max_sessions: int) -> DuplexCapabilities: + return nemotron_voicechat_capabilities(max_sessions=max_sessions) + + def validate_client_extra_body(self, extra_body: object) -> None: + reject_private_runtime_keys( + extra_body, + self.private_runtime_config_keys, + message="Nemotron VoiceChat runtime configuration is server-owned: ", + error_cls=NemotronVoiceChatClientRuntimeConfigError, + ) + + async def prepare_runtime_config( + self, + config: DuplexSessionConfig, + *, + model_config: ModelConfig | None, + ) -> dict[str, object]: + self.validate_client_extra_body(config.extra_body) + _require_native_full_duplex(config) + instructions = str(config.instructions or _DEFAULT_SYSTEM_PROMPT) + tools, tools_signature = _normalized_tools(config) + rendered_prompt = _render_tool_prompt(instructions, tools) + special, tokenizer = await self._tokenizer_runtime(model_config) + prompt_ids = await asyncio.to_thread( + _render_prompt_ids, + tokenizer, + rendered_prompt, + int(special["nvc_text_bos_id"]), + int(special["nvc_text_eos_id"]), + ) + max_model_len = getattr(model_config, "max_model_len", None) + if not isinstance(max_model_len, int) or max_model_len <= 0: + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat requires a positive Stage-0 max_model_len" + ) + runtime: dict[str, object] = dict(special) + runtime["nvc_prompt_token_ids"] = prompt_ids + runtime["nvc_max_model_len"] = max_model_len + # Keep raw session values for immutable-in-epoch update checks; only + # the rendered prompt token ids enter Stage-0 KV. + runtime["instructions"] = instructions + runtime["nvc_tools_signature"] = tools_signature + self.data_plane.configure_runtime(runtime, tokenizer=tokenizer) + return runtime + + async def _tokenizer_runtime(self, model_config: ModelConfig | None) -> tuple[dict[str, object], Any]: + """Return the (special ids, tokenizer) pair for this checkpoint, cached per tokenizer ref.""" + stt_cfg = _stt_config(model_config) + tokenizer_ref = str( + os.environ.get("NEMOTRON_VOICECHAT_LLM_PATH") + or stt_cfg.get("pretrained_llm", "nvidia/NVIDIA-Nemotron-Nano-9B-v2") + ) + cached = self._tokenizers.get(tokenizer_ref) + if cached is not None: + return cached + async with self._tokenizer_lock: + cached = self._tokenizers.get(tokenizer_ref) + if cached is None: + cached = await asyncio.to_thread(_load_tokenizer_runtime, model_config) + self._tokenizers[tokenizer_ref] = cached + return cached + + def runtime_config_for_update( + self, + config: DuplexSessionConfig, + current: Mapping[str, object], + ) -> dict[str, object]: + self.validate_client_extra_body(config.extra_body) + _require_native_full_duplex(config) + runtime = deepcopy(dict(current)) + instructions = str(config.instructions or _DEFAULT_SYSTEM_PROMPT) + _, tools_signature = _normalized_tools(config) + # The system/tool prompt is already resident in Stage-0 KV. Updating it + # without a new epoch would make the advertised config disagree with + # model state, so reject such updates explicitly. + if runtime and instructions != runtime.get("instructions"): + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat instructions cannot change inside an active duplex session" + ) + if runtime and tools_signature != runtime.get("nvc_tools_signature", "[]"): + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat tools cannot change inside an active duplex session" + ) + return runtime + + def runtime_config_for_function_output( + self, + config: DuplexSessionConfig, + current: Mapping[str, object], + item: Mapping[str, object], + ) -> dict[str, object]: + """Queue a client tool result for frame-locked function-channel injection.""" + del config + call_id = item.get("call_id") + output = item.get("output") + if not isinstance(call_id, str) or not call_id: + raise NemotronVoiceChatClientRuntimeConfigError( + "function_call_output requires call_id", + code="invalid_function_call_output", + ) + if not isinstance(output, str): + raise NemotronVoiceChatClientRuntimeConfigError( + "function_call_output requires a string output", + code="invalid_function_call_output", + ) + tokenizer_ref = current.get("nvc_tokenizer_ref") + cached = self._tokenizers.get(str(tokenizer_ref)) if tokenizer_ref else None + tokenizer = cached[1] if cached is not None else None + if tokenizer is None: + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat tokenizer is unavailable for function output", + code="invalid_function_call_output", + ) + response_token_ids = list( + tokenizer.encode( + _render_tool_response(output), + add_special_tokens=False, + ) + ) + if not response_token_ids: + raise NemotronVoiceChatClientRuntimeConfigError( + "function_call_output tokenized to an empty response", + code="invalid_function_call_output", + ) + runtime = deepcopy(dict(current)) + generation = int(runtime.get("nvc_function_response_generation", 0)) + 1 + token_ids = [int(token_id) for token_id in response_token_ids] + batches = runtime.get("nvc_function_response_batches") + if not isinstance(batches, list): + batches = [] + if len(batches) >= _MAX_PENDING_FUNCTION_RESPONSE_BATCHES: + raise NemotronVoiceChatClientRuntimeConfigError( + "Nemotron VoiceChat received function results faster than the frame-locked " + f"channel drains them (>{_MAX_PENDING_FUNCTION_RESPONSE_BATCHES} pending batches); " + "return tool results after the model has consumed the previous one", + code="function_response_backlog", + ) + batches = [*batches, {"generation": generation, "token_ids": token_ids}] + runtime["nvc_function_response_generation"] = generation + runtime["nvc_function_response_batches"] = batches + return runtime + + def runtime_config_after_model_output( + self, + current: Mapping[str, object], + output_metadata: Mapping[str, object], + ) -> dict[str, object] | None: + """Retire batches after the thinker takes ownership of their tokens.""" + consumed = output_metadata.get("nvc_function_response_consumed_generation") + try: + consumed_generation = int(consumed) + except (TypeError, ValueError): + return None + if consumed_generation <= 0: + return None + batches = current.get("nvc_function_response_batches") + if not isinstance(batches, list): + return None + remaining = [ + batch + for batch in batches + if isinstance(batch, dict) and int(batch.get("generation", 0)) > consumed_generation + ] + if len(remaining) == len(batches): + return None + runtime = deepcopy(dict(current)) + runtime["nvc_function_response_batches"] = remaining + return runtime + + def data_plane_context( + self, + *, + epoch: int, + turn_id: int, + active_response_turn_id: int | None, + active_response_id: str | None, + auto_responds: bool, + response_format: str, + speed: float | None, + modalities: tuple[str, ...], + ) -> NemotronVoiceChatDataPlaneContext: + del active_response_turn_id, active_response_id + return NemotronVoiceChatDataPlaneContext( + epoch=epoch, + turn_id=turn_id, + auto_responds=auto_responds, + response_format=response_format, + speed=speed, + modalities=modalities, + ) + + +__all__ = [ + "PRIVATE_RUNTIME_CONFIG_KEYS", + "NemotronVoiceChatClientRuntimeConfigError", + "NemotronVoiceChatDuplexPlugin", +] diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/runtime.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/runtime.py deleted file mode 100644 index b017246bd..000000000 --- a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/runtime.py +++ /dev/null @@ -1,167 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project - -"""Engine policy for Nemotron VoiceChat's frame-locked duplex timeline.""" - -from __future__ import annotations - -from typing import Any - -from vllm.sampling_params import RequestOutputKind, SamplingParams - -from vllm_omni.engine.duplex.contracts import ( - DuplexAppendPlan, - DuplexInputMode, - DuplexOutputAction, - DuplexOutputDecision, -) -from vllm_omni.engine.duplex.messages import DuplexFence -from vllm_omni.model_executor.models.nemotron_voicechat.duplex.input import ( - decode_pcm_f32le, -) - - -def _plain_token_ids(value: object, *, name: str) -> list[int]: - if not isinstance(value, list | tuple) or not value: - raise ValueError(f"Nemotron VoiceChat runtime requires non-empty {name}") - try: - return [int(token_id) for token_id in value] - except (TypeError, ValueError) as exc: - raise ValueError(f"Nemotron VoiceChat {name} must contain token ids") from exc - - -def _positive_int(value: object, *, name: str) -> int: - try: - parsed = int(value) - except (TypeError, ValueError) as exc: - raise ValueError(f"Nemotron VoiceChat requires a positive {name}") from exc - if parsed <= 0: - raise ValueError(f"Nemotron VoiceChat requires a positive {name}") - return parsed - - -class NemotronVoiceChatDuplexRuntimeExtension: - """Append one 80 ms acoustic frame to one resumable thinker request.""" - - def configure_sampling_params( - self, - *, - runtime_config: dict[str, Any], - defaults: tuple[object, ...], - ) -> tuple[object, ...]: - del runtime_config - if not defaults: - return defaults - # Every scheduler segment is one transport wake of a long-lived - # request. CUMULATIVE output makes the multimodal output processor - # concatenate all earlier Stage-2 waveforms and resend them on every - # wake (80, 160, 240, ... ms), causing O(n^2) audio replay. Duplex - # consumers require per-wake deltas at every stage. - configured = [params.clone() if isinstance(params, SamplingParams) else params for params in defaults] - for params in configured: - if isinstance(params, SamplingParams): - params.output_kind = RequestOutputKind.DELTA - stage0 = defaults[0] - if isinstance(stage0, SamplingParams): - stage0 = stage0.clone() - stage0.temperature = 0.0 - stage0.top_p = 1.0 - stage0.top_k = 0 - stage0.max_tokens = 1 - stage0.ignore_eos = True - stage0.output_kind = RequestOutputKind.DELTA - configured[0] = stage0 - return tuple(configured) - - def plan_append( - self, - *, - request_id: str, - fence: DuplexFence, - session_config: dict[str, Any], - runtime_config: dict[str, Any], - seq: int, - turn_seq: int, - mode: DuplexInputMode, - payload: object, - final: bool, - sampling_params: object, - ) -> DuplexAppendPlan: - del sampling_params, turn_seq - if mode is not DuplexInputMode.APPEND_AUDIO_CHUNK: - raise ValueError(f"Nemotron VoiceChat does not support duplex input mode {mode.value!r}") - decode_pcm_f32le(payload, exact_frame=True) - normalized_payload = dict(payload) - prompt_ids = _plain_token_ids( - runtime_config.get("nvc_prompt_token_ids"), - name="nvc_prompt_token_ids", - ) - max_model_len = _positive_int( - runtime_config.get("nvc_max_model_len"), - name="nvc_max_model_len", - ) - required_model_len = len(prompt_ids) + seq + 1 - if required_model_len > max_model_len: - raise ValueError( - "Nemotron VoiceChat native duplex session exceeds the Stage-0 " - f"max_model_len: prompt_tokens={len(prompt_ids)} + input_frames={seq} + " - f"sampled_token=1 gives {required_model_len} > {max_model_len}; " - "start a new session or raise Stage 0 max_model_len" - ) - try: - pad_id = int(runtime_config["nvc_text_pad_id"]) - except (KeyError, TypeError, ValueError) as exc: - raise ValueError("Nemotron VoiceChat runtime requires nvc_text_pad_id") from exc - scheduler_prompt = prompt_ids + [pad_id] if seq <= 1 else [pad_id] - return DuplexAppendPlan( - prompt={ - "prompt_token_ids": scheduler_prompt, - "model_intermediate_buffer": { - "request_id": request_id, - "global_request_id": [fence.session_id], - "duplex": { - "data_plane": True, - "fence": fence, - "session_id": fence.session_id, - "incarnation": fence.incarnation, - "epoch": fence.epoch, - "source_input_seq": seq, - "seq": seq, - "mode": mode.value, - "payload": normalized_payload, - "final": final, - "session_config": dict(session_config), - "runtime_config": dict(runtime_config), - "scheduler_token_budget": len(scheduler_prompt), - }, - }, - } - ) - - def decide_output( - self, - *, - stage_id: int, - final_stage_id: int, - segment_finished: bool, - segment_token_ids: tuple[int, ...], - segment_output_metadata: dict[str, Any], - output: object, - ) -> DuplexOutputDecision | None: - del final_stage_id, segment_finished, output - if stage_id != 0: - return None - metadata = dict(segment_output_metadata) - metadata["nvc_text_token_ids"] = list(segment_token_ids) - # Stage 0 is a client-visible side channel even though Stage 2 is the - # configured audio response stage. Mark it explicitly so the shared - # collector does not discard it while waiting for the final stage. - metadata["duplex_direct_response"] = True - return DuplexOutputDecision( - action=DuplexOutputAction.DIRECT_RESPONSE, - metadata=metadata, - final_output_type="text", - ) - - -__all__ = ["NemotronVoiceChatDuplexRuntimeExtension"] diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/serving_adapter.py b/vllm_omni/model_executor/models/nemotron_voicechat/duplex/serving_adapter.py deleted file mode 100644 index 1b2107287..000000000 --- a/vllm_omni/model_executor/models/nemotron_voicechat/duplex/serving_adapter.py +++ /dev/null @@ -1,361 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project - -"""Realtime serving adapter for Nemotron VoiceChat native duplex.""" - -from __future__ import annotations - -import asyncio -import json -import os -from collections.abc import Callable, Mapping -from copy import deepcopy -from typing import Any - -from vllm_omni.entrypoints.duplex.protocol import DuplexCapabilities -from vllm_omni.entrypoints.duplex.runtime_adapter import ServingRuntimeConfigError -from vllm_omni.model_executor.models.minicpmo_4_5.duplex.session import ( - MiniCPMO45ServingSessionState, -) -from vllm_omni.model_executor.models.nemotron_voicechat.duplex.data_plane import ( - NemotronVoiceChatDataPlaneContext, - NemotronVoiceChatDataPlaneSession, -) -from vllm_omni.model_executor.models.nemotron_voicechat.duplex.input import ( - NemotronVoiceChatPcmAppendBuffer, -) - -EncodeAudio = Callable[[object, int, str, float | None], str | None] - -_DEFAULT_SYSTEM_PROMPT = ( - "You are an AI voice assistant developed by NVIDIA. " - "Your name is NVIDIA Voice Chat. " - "Answer in a spoken, conversational style rather than a written one. " - "Do not repeat the same sentence over and over again. " - "Start the conversation by greeting the user." -) -_PRIVATE_KEYS = frozenset( - { - "nvc_prompt_token_ids", - "nvc_text_bos_id", - "nvc_text_eos_id", - "nvc_text_pad_id", - "nvc_function_sotc_id", - "nvc_function_eotc_id", - "nvc_function_eotr_id", - "nvc_max_model_len", - "nvc_tokenizer_ref", - "nvc_tools_signature", - "nvc_function_response_generation", - "nvc_function_response_token_ids", - "nvc_function_response_call_id", - "nvc_function_response_batches", - } -) - - -def _stt_config(model_config: Any) -> dict[str, Any]: - hf_config = getattr(model_config, "hf_config", None) - stt_cfg = getattr(hf_config, "stt_cfg", None) - if not isinstance(stt_cfg, dict): - raise ServingRuntimeConfigError("Nemotron VoiceChat checkpoint STT configuration is unavailable") - return stt_cfg - - -def _normalized_tools(config: object) -> tuple[list[dict[str, object]], str]: - extra_body = getattr(config, "extra_body", None) - raw_tools = extra_body.get("realtime_tools") if isinstance(extra_body, dict) else None - if raw_tools is None: - return [], "[]" - if not isinstance(raw_tools, list): - raise ServingRuntimeConfigError("Nemotron VoiceChat tools must be a list") - if len(raw_tools) > 5: - raise ServingRuntimeConfigError("Nemotron VoiceChat supports at most 5 tools per session") - - normalized: list[dict[str, object]] = [] - for index, tool in enumerate(raw_tools): - if not isinstance(tool, dict): - raise ServingRuntimeConfigError(f"Nemotron VoiceChat tool {index} must be an object") - function = tool.get("function", tool) - if not isinstance(function, dict): - raise ServingRuntimeConfigError(f"Nemotron VoiceChat tool {index} has no function definition") - definition = {key: value for key, value in function.items() if key != "type"} - if not isinstance(definition.get("name"), str) or not str(definition["name"]).strip(): - raise ServingRuntimeConfigError(f"Nemotron VoiceChat tool {index} requires a name") - normalized.append(definition) - signature = json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":")) - return normalized, signature - - -def _render_tool_prompt(instructions: str, tools: list[dict[str, object]]) -> str: - if not tools: - return instructions - available = json.dumps(tools, ensure_ascii=False, sort_keys=True) - return ( - f"{instructions}\n\n" - "You can use the following tools to assist the user if required:" - f"\n{available}\n\n" - "If you decide to call any tool(s), use the following format:\n" - '[{"name": "tool_name1", "arguments": "tool_args1"}, ' - '{"name": "tool_name2", "arguments": "tool_args2"}]\n\n' - "The user will execute tool-calls and return responses from tool(s) in this format:\n" - '[{"tool_response1"}, {"tool_response2"}]\n\n' - "Based on the tool responses, you can call additional tools if needed, correct tool calls if any " - "errors are found, or just respond to the user." - ) - - -def _render_tool_response(output: str) -> str: - """Serialize one Realtime function result using NVIDIA's function channel.""" - try: - value = json.loads(output) - except json.JSONDecodeError: - value = output - payload = json.dumps([value], ensure_ascii=True, separators=(",", ":")) - return f"{payload}" - - -def _require_native_full_duplex(config: object) -> None: - extra_body = getattr(config, "extra_body", None) - enabled = isinstance(extra_body, dict) and extra_body.get("auto_response") is True - if not enabled: - raise ServingRuntimeConfigError( - "Nemotron VoiceChat currently supports model-native full-duplex streaming only; " - "set extra_body.auto_response=true", - code="unsupported_nemotron_duplex_mode", - ) - - -def _tokenize_runtime(model_config: Any, instructions: str) -> tuple[dict[str, object], Any]: - from transformers import AutoTokenizer - - stt_cfg = _stt_config(model_config) - tokenizer_ref = os.environ.get("NEMOTRON_VOICECHAT_LLM_PATH") or stt_cfg.get( - "pretrained_llm", "nvidia/NVIDIA-Nemotron-Nano-9B-v2" - ) - tokenizer = AutoTokenizer.from_pretrained(tokenizer_ref, trust_remote_code=False) - - def token(name: str, default: str) -> int: - value = tokenizer.convert_tokens_to_ids(stt_cfg.get(name, default)) - if value is None: - raise ServingRuntimeConfigError(f"Nemotron VoiceChat tokenizer does not define {name}") - return int(value) - - bos_id = token("bos_token", "") - eos_id = token("eos_token", "") - pad_id = token("pad_token", "") - prompt_ids = [bos_id] + list(tokenizer.encode(instructions, add_special_tokens=False)) + [eos_id] - runtime = { - "instructions": instructions, - "nvc_prompt_token_ids": prompt_ids, - "nvc_text_bos_id": bos_id, - "nvc_text_eos_id": eos_id, - "nvc_text_pad_id": pad_id, - "nvc_function_sotc_id": int(tokenizer.convert_tokens_to_ids("")), - "nvc_function_eotc_id": int(tokenizer.convert_tokens_to_ids("")), - "nvc_function_eotr_id": int(tokenizer.convert_tokens_to_ids("")), - "nvc_tokenizer_ref": str(tokenizer_ref), - } - return runtime, tokenizer - - -class NemotronVoiceChatServingRuntimeAdapter: - adapter_id = "nemotron_voicechat" - silence_continuation_samples = 1280 - # The session-owned persistent drain is the sole data-plane output - # consumer. Collecting on append would race that drain and can consume a - # Stage2 audio output before the realtime projector observes it. - collect_outputs_on_append = False - clean_response_done_prefix = "" - interrupted_tts_prefix = "" - private_runtime_config_keys = _PRIVATE_KEYS - - def __init__(self, encode_audio: EncodeAudio) -> None: - self.session_states: dict[str, MiniCPMO45ServingSessionState] = {} - self.data_plane = NemotronVoiceChatDataPlaneSession(encode_audio) - self._tokenizer: Any | None = None - - def create_session_state(self) -> MiniCPMO45ServingSessionState: - return MiniCPMO45ServingSessionState( - audio_buffer=NemotronVoiceChatPcmAppendBuffer(), - ) - - def session_state(self, session_id: str) -> MiniCPMO45ServingSessionState: - return self.session_states.setdefault(session_id, self.create_session_state()) - - def remove_session_state(self, session_id: str) -> None: - self.session_states.pop(session_id, None) - - @staticmethod - def is_enabled(config: object) -> bool: - del config - return True - - @staticmethod - def capabilities(*, max_sessions: int) -> DuplexCapabilities: - supports_multi_session = max_sessions > 1 - return DuplexCapabilities( - supports_model_native_turn_policy=True, - supports_external_turn_signal=False, - supports_client_commit=True, - supports_barge_in=False, - supports_playback_ack=True, - supports_input_append=True, - supports_replace_latest_chunk=False, - supports_reencode_context=False, - supports_rollback_to_checkpoint=False, - supports_turn_commit_only=False, - supports_kv_lease=False, - supports_core_kv_lease=False, - supports_model_internal_state=True, - supports_stage_resumption=True, - supports_scheduler_native_append=False, - supports_core_resumable_request=True, - supports_stage_connector_handoff=True, - supports_independent_io_streams=True, - supports_realtime_endpoint=True, - supports_multi_session=supports_multi_session, - supports_multi_session_same_replica=False, - supports_session_lease=True, - supports_session_resume=True, - session_admission_mode="engine_managed", - supports_audio_truncate=False, - requires_model_runner_kv=True, - requires_native_stage_role=True, - implementation_level="model_native_duplex", - adapter_patterns=["scheduler_data_plane"], - input_modes=["append_audio_chunk"], - signal_sources=["model_native", "client_event"], - stage_handoff_transport="scheduler_data_plane", - chunk_period_ms=80, - target_barge_in_latency_ms=None, - ) - - @staticmethod - def validate_client_extra_body(extra_body: object) -> None: - if not isinstance(extra_body, dict): - return - private = sorted(_PRIVATE_KEYS.intersection(extra_body)) - if private: - raise ServingRuntimeConfigError( - "Nemotron VoiceChat runtime configuration is server-owned: " + ", ".join(private) - ) - - async def prepare_runtime_config( - self, - config: object, - *, - model_config: Any, - ) -> dict[str, object]: - self.validate_client_extra_body(getattr(config, "extra_body", None)) - _require_native_full_duplex(config) - instructions = str(getattr(config, "instructions", None) or _DEFAULT_SYSTEM_PROMPT) - tools, tools_signature = _normalized_tools(config) - rendered_prompt = _render_tool_prompt(instructions, tools) - runtime, tokenizer = await asyncio.to_thread(_tokenize_runtime, model_config, rendered_prompt) - max_model_len = getattr(model_config, "max_model_len", None) - if not isinstance(max_model_len, int) or max_model_len <= 0: - raise ServingRuntimeConfigError("Nemotron VoiceChat requires a positive Stage-0 max_model_len") - runtime["nvc_max_model_len"] = max_model_len - # Keep raw session values for immutable-in-incarnation update checks; - # only the rendered prompt token ids enter Stage-0 KV. - runtime["instructions"] = instructions - runtime["nvc_tools_signature"] = tools_signature - self.data_plane.configure_runtime(runtime, tokenizer=tokenizer) - self._tokenizer = tokenizer - return runtime - - def runtime_config_for_function_output( - self, - item: Mapping[str, object], - current: Mapping[str, object], - ) -> dict[str, object]: - """Queue a client tool result for frame-locked function-channel injection.""" - call_id = item.get("call_id") - output = item.get("output") - if not isinstance(call_id, str) or not call_id: - raise ServingRuntimeConfigError( - "function_call_output requires call_id", - code="invalid_function_call_output", - ) - if not isinstance(output, str): - raise ServingRuntimeConfigError( - "function_call_output requires a string output", - code="invalid_function_call_output", - ) - if self._tokenizer is None: - raise ServingRuntimeConfigError( - "Nemotron VoiceChat tokenizer is unavailable for function output", - code="invalid_function_call_output", - ) - response_token_ids = list( - self._tokenizer.encode( - _render_tool_response(output), - add_special_tokens=False, - ) - ) - if not response_token_ids: - raise ServingRuntimeConfigError( - "function_call_output tokenized to an empty response", - code="invalid_function_call_output", - ) - runtime = deepcopy(dict(current)) - generation = int(runtime.get("nvc_function_response_generation", 0)) + 1 - token_ids = [int(token_id) for token_id in response_token_ids] - batches = runtime.get("nvc_function_response_batches") - if not isinstance(batches, list): - batches = [] - batches = [*batches, {"generation": generation, "call_id": call_id, "token_ids": token_ids}] - runtime["nvc_function_response_generation"] = generation - runtime["nvc_function_response_token_ids"] = token_ids - runtime["nvc_function_response_call_id"] = call_id - runtime["nvc_function_response_batches"] = batches - return runtime - - @staticmethod - def runtime_config_for_update( - config: object, - current: Mapping[str, object], - ) -> dict[str, object]: - NemotronVoiceChatServingRuntimeAdapter.validate_client_extra_body(getattr(config, "extra_body", None)) - _require_native_full_duplex(config) - runtime = deepcopy(dict(current)) - instructions = str(getattr(config, "instructions", None) or _DEFAULT_SYSTEM_PROMPT) - _, tools_signature = _normalized_tools(config) - # The system/tool prompt is already resident in Stage-0 KV. Updating it - # without a new incarnation would make the advertised config disagree - # with model state, so reject such updates explicitly. - if runtime and instructions != runtime.get("instructions"): - raise ServingRuntimeConfigError( - "Nemotron VoiceChat instructions cannot change inside an active duplex incarnation" - ) - if runtime and tools_signature != runtime.get("nvc_tools_signature", "[]"): - raise ServingRuntimeConfigError( - "Nemotron VoiceChat tools cannot change inside an active duplex incarnation" - ) - return runtime - - @staticmethod - def data_plane_context( - *, - epoch: int, - turn_id: int, - active_response_turn_id: int | None, - active_response_id: str | None, - auto_responds: bool, - response_format: str, - speed: float | None, - modalities: tuple[str, ...], - ) -> NemotronVoiceChatDataPlaneContext: - del active_response_turn_id, active_response_id - return NemotronVoiceChatDataPlaneContext( - epoch=epoch, - turn_id=turn_id, - auto_responds=auto_responds, - response_format=response_format, - speed=speed, - modalities=modalities, - ) - - -__all__ = ["NemotronVoiceChatServingRuntimeAdapter"] diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_thinker.py b/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_thinker.py index 0701a8b7b..07c6d6cb8 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_thinker.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_thinker.py @@ -221,6 +221,11 @@ class NemotronVoiceChatThinkerForConditionalGeneration(nn.Module, HasInnerState, checkpoint config. """ + omni_client_multimodal_output_keys = ( + "nvc_function_token", + "nvc_function_response_consumed_generation", + ) + @classmethod def get_mamba_state_dtype_from_config(cls, vllm_config: VllmConfig) -> tuple[torch.dtype, torch.dtype]: from vllm.model_executor.models.nemotron_h import NemotronHForCausalLM @@ -323,7 +328,6 @@ class NemotronVoiceChatThinkerForConditionalGeneration(nn.Module, HasInnerState, self.have_multimodal_outputs = True self.has_preprocess = True self.prefer_model_sampler = True - self.omni_client_multimodal_output_keys = ("nvc_function_token",) self.has_postprocess = True # per-step greedy function-channel token self.requires_full_prefix_cached_hidden_states = False # The per-step function token is a scalar; no GPU-resident buffers needed. @@ -399,8 +403,23 @@ class NemotronVoiceChatThinkerForConditionalGeneration(nn.Module, HasInnerState, def make_omni_output(self, model_outputs: torch.Tensor | OmniOutput, **kwargs: Any) -> OmniOutput: if isinstance(model_outputs, OmniOutput): return model_outputs - multimodal_outputs: dict[str, torch.Tensor] = {} + multimodal_outputs: dict[str, torch.Tensor | list[torch.Tensor]] = {} info_dicts = kwargs.get("model_intermediate_buffer") or kwargs.get("runtime_additional_information") + if model_outputs.numel() and isinstance(info_dicts, list): + # Preprocess has copied these batches into the model-owned queue. + # Acknowledge that ownership through the wire payload, not the + # worker-local postprocess buffer. Lists preserve request alignment + # even when a mixed prefill/decode batch has unequal token spans. + generations = [ + int(self._sessions.get(info.get("request_id"), {}).get("function_response_generation", 0)) + if isinstance(info, dict) + else 0 + for info in info_dicts + ] + if any(generations): + multimodal_outputs["nvc_function_response_consumed_generation"] = [ + torch.tensor([generation], dtype=torch.int64) for generation in generations + ] single_request = isinstance(info_dicts, list) and len(info_dicts) == 1 if self._use_function_head and model_outputs.numel() and single_request: # Only a single-request batch can attribute the batch's last row @@ -735,13 +754,6 @@ class NemotronVoiceChatThinkerForConditionalGeneration(nn.Module, HasInnerState, if not isinstance(raw_tokens, list | tuple) or not raw_tokens: raise ValueError("Nemotron VoiceChat function response requires token ids") unseen.append((batch_generation, [int(token_id) for token_id in raw_tokens])) - if not unseen: - # Backward-compatible single-response payload used by older - # serving adapters and serialized test fixtures. - raw_tokens = runtime_config.get("nvc_function_response_token_ids") - if not isinstance(raw_tokens, list | tuple) or not raw_tokens: - raise ValueError("Nemotron VoiceChat function response requires token ids") - unseen = [(generation, [int(token_id) for token_id in raw_tokens])] expected = seen + 1 queue = session.setdefault("forced_function_tokens", []) if not isinstance(queue, list): diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/pipeline.py b/vllm_omni/model_executor/models/nemotron_voicechat/pipeline.py index 3cbe12742..e3eb3290b 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/pipeline.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/pipeline.py @@ -31,13 +31,11 @@ _PROC = "vllm_omni.model_executor.stage_input_processors.nemotron_voicechat" NEMOTRON_VOICECHAT_PIPELINE = PipelineConfig( model_type="nemotron_voicechat", model_arch="NemotronVoiceChatThinkerForConditionalGeneration", - duplex_runtime_extension=( - "vllm_omni.model_executor.models.nemotron_voicechat.duplex.runtime.NemotronVoiceChatDuplexRuntimeExtension" - ), - duplex_serving_adapter=( - "vllm_omni.model_executor.models.nemotron_voicechat.duplex.serving_adapter.NemotronVoiceChatServingRuntimeAdapter" - ), - duplex_control_enabled=True, + duplex_plugin=("vllm_omni.model_executor.models.nemotron_voicechat.duplex.plugin.NemotronVoiceChatDuplexPlugin"), + # Duplex is opt-in per deployment (``nemotron_labs_voicechat_duplex.yaml`` + # declares ``session_mode: duplex``); the bare/streaming deploys stay + # turn-based. + default_session_mode="turn", # Named after the alias key so the deploy-yaml stem substring-matches the # checkpoint directory name (NVIDIA-NemotronLabs-VoiceChat-11B) during # bare-path auto-detection.