From 009b992c5b5d36acc55759dba7ed5acfee90d324 Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Fri, 28 Aug 2026 00:46:41 -0700 Subject: [PATCH] [Bugfix][MiniCPM-o] Isolate duplex handoff failures and barge-in (#6170) Signed-off-by: Sy03 <1370724210@qq.com> --- .buildkite/cuda/test-merge.yml | 1 + .buildkite/cuda/test-ready.yml | 1 + .../minicpmo/realtime_duplex_demo.py | 2 +- pyproject.toml | 5 + ...run_minicpmo_realtime_duplex_server_vad.py | 85 ++++++++ .../test_minicpmo_4_5_duplex_expansion.py | 20 ++ .../test_orchestrator_error_handling.py | 30 +++ .../openai/test_duplex_protocol.py | 2 +- .../openai_api/test_duplex_handler.py | 149 ++++++++++--- .../test_nemotron_voicechat.py | 6 +- vllm_omni/engine/orchestrator.py | 19 +- vllm_omni/experimental/fullduplex/DESIGN.md | 15 +- vllm_omni/experimental/fullduplex/client.py | 9 +- .../fullduplex/openai/protocol.py | 5 +- .../fullduplex/openai/realtime_input.py | 196 ++++++++++++++++-- .../fullduplex/openai/realtime_output.py | 13 +- .../fullduplex/openai/realtime_session.py | 1 + .../fullduplex/openai/realtime_state.py | 9 + .../experimental/fullduplex/openai/serving.py | 52 ++++- .../fullduplex/openai/session_runner.py | 78 +++++-- .../experimental/fullduplex/openai/vad.py | 178 ++++++++++++++++ .../nemotron_voicechat_talker.py | 9 +- .../minicpmo_4_5_omni.py | 30 ++- 23 files changed, 817 insertions(+), 98 deletions(-) create mode 100644 tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py create mode 100644 vllm_omni/experimental/fullduplex/openai/vad.py diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml index aba70c286..dde5a17e7 100644 --- a/.buildkite/cuda/test-merge.yml +++ b/.buildkite/cuda/test-merge.yml @@ -166,6 +166,7 @@ steps: - tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py - tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py - tests/e2e/online_serving/run_minicpmo_realtime_duplex_multi_session.py + - tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py - vllm_omni/deploy/minicpmo_4_5.yaml - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py diff --git a/.buildkite/cuda/test-ready.yml b/.buildkite/cuda/test-ready.yml index 418d8f996..f4b6b0188 100644 --- a/.buildkite/cuda/test-ready.yml +++ b/.buildkite/cuda/test-ready.yml @@ -205,6 +205,7 @@ steps: - tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py - tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py - tests/e2e/online_serving/run_minicpmo_realtime_duplex_multi_session.py + - tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py - vllm_omni/deploy/minicpmo_4_5.yaml - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py index 23d65eb35..f726eec92 100644 --- a/examples/online_serving/minicpmo/realtime_duplex_demo.py +++ b/examples/online_serving/minicpmo/realtime_duplex_demo.py @@ -372,7 +372,7 @@ def parse_args() -> argparse.Namespace: "--temperature", type=float, default=None, - help="Stage0 sampling temperature; response-required soft-interrupt uses 0.0.", + help="Stage 0 sampling temperature; omit to preserve the model default.", ) parser.add_argument("--no-realtime-pacing", action="store_true") parser.add_argument("--require-audio", action="store_true") diff --git a/pyproject.toml b/pyproject.toml index 2d43edfee..a484227cd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,6 +75,7 @@ dev = [ # Provides top-level cosyvoice2; do not co-install stepaudio2-minicpmo, # whose top-level stepaudio2 package conflicts with this distribution. "step-audio2==1.0.0", # Step-Audio2 / MiniCPM-o 4.5 Token2Wav (flashcosyvoice / hyperpyyaml) + "silero-vad==6.1", # Realtime server-VAD duplex E2E ] demo = [ @@ -84,6 +85,10 @@ demo = [ "cn2an>=0.5.22", # Chinese number normalization for TTS (e.g. "1次" -> "一次") ] +server-vad = [ + "silero-vad==6.1", +] + forced-aligner = [ "qwen-asr>=0.0.6", # faithful Japanese/Korean word segmentation for the TTS forced aligner ] diff --git a/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py b/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py new file mode 100644 index 000000000..4047a0901 --- /dev/null +++ b/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py @@ -0,0 +1,85 @@ +import base64 +from pathlib import Path + +from vllm_omni.experimental.fullduplex.client import ( + PCM16_BYTES_PER_SAMPLE, + PCM16_SAMPLE_RATE, + RealtimeDuplexClient, + RealtimeEventCollector, + build_realtime_url, + read_pcm16_wav, + wait_for, +) + + +def _events(events: list[dict[str, object]], response_id: str, kind: str) -> list[dict[str, object]]: + return [e for e in events if e.get("type") == kind and RealtimeEventCollector.response_id(e) == response_id] + + +async def run_server_vad_interrupt(args) -> dict[str, object]: + initial = read_pcm16_wav(Path(args.input_wav)) + interrupt = read_pcm16_wav(Path(args.interrupt_wav))[: PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * 47 // 10] + session_id = f"server-vad-hard-interrupt-{id(args)}" + client = RealtimeDuplexClient(build_realtime_url(args.url, args.model, autostart=False, session_id=session_id)) + + async def until(predicate, label: str) -> None: + await wait_for(predicate, timeout_s=args.timeout_s, label=label) + + try: + await client.__aenter__() + await client.configure( + args.model, + ref_audio="data:audio/wav;base64," + base64.b64encode(Path(args.ref_audio).read_bytes()).decode(), + session_id=session_id, + turn_detection={"type": "server_vad", "interrupt_response": True}, + timeout_s=args.timeout_s, + ) + await client.stream_pcm16(initial + bytes(16_000 * 2 * 1600 // 1000), chunk_ms=200, realtime=True) + await client.commit() + await until(lambda: any(e.get("type") == "response.created" for e in client.events.events), "response.created") + target_id = RealtimeEventCollector.response_id( + next(e for e in client.events.events if e.get("type") == "response.created") + ) + assert target_id + await until(lambda: _events(client.events.events, target_id, "response.audio.delta"), "response.audio.delta") + cursor = len(client.events.events) + await client.stream_pcm16(interrupt + bytes(16_000 * 2 * 800 // 1000), chunk_ms=args.chunk_ms, realtime=True) + await until(lambda: _events(client.events.events, target_id, "response.done"), "cancelled response.done") + await client.commit() + await until( + lambda: any( + e.get("type") == "response.done" and RealtimeEventCollector.response_id(e) != target_id + for e in client.events.events[cursor:] + ), + "follow-up response.done", + ) + events = client.events.events + finally: + try: + await client.close_session(timeout_s=args.timeout_s) + finally: + await client.__aexit__(None, None, None) + + done = _events(events, target_id, "response.done") + terminal = done[0] if len(done) == 1 else None + trailing = events[events.index(terminal) + 1 :] if terminal is not None else events + stale = any( + e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) == target_id for e in trailing + ) + followup_audio = any( + e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) != target_id + for e in events[cursor:] + ) + ok = ( + terminal is not None + and isinstance(terminal.get("response"), dict) + and terminal["response"].get("status") == "cancelled" + and terminal["response"].get("status_details", {}).get("reason") == "turn_detected" + and not stale + and sum(e.get("type") == "input_audio_buffer.speech_started" for e in events[cursor:]) == 1 + and sum(e.get("type") == "input_audio_buffer.speech_stopped" for e in events[cursor:]) == 1 + and any(e.get("type") == "input_audio_buffer.committed" for e in events[cursor:]) + and followup_audio + and not any(e.get("type") == "error" for e in events) + ) + return {"ok": ok} diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py index 1adf5b614..5b35c4b53 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py @@ -25,6 +25,7 @@ from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import ( from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import ( run_lifecycle_probes, ) +from tests.e2e.online_serving.run_minicpmo_realtime_duplex_server_vad import run_server_vad_interrupt from tests.e2e.online_serving.run_minicpmo_realtime_duplex_soft_interrupt import ( run_soft_interrupt, ) @@ -92,3 +93,22 @@ def test_duplex_soft_interrupt(omni_server, tmp_path: Path) -> None: assert result["response_lifecycle_ok"] is True assert result["response_audio_contract_ok"] is True assert result["followup_response_transcript_ok"] is True + + +@hardware_test(res={"cuda": "H100"}, num_cards=1) +@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True) +def test_duplex_server_vad_hard_interrupt(omni_server) -> None: + result = asyncio.run( + run_server_vad_interrupt( + SimpleNamespace( + url=realtime_url(omni_server), + model=omni_server.model, + input_wav=str(validated_input_wav()), + interrupt_wav=str(validated_soft_interrupt_wav()), + ref_audio=str(resolve_ref_audio()), + chunk_ms=200, + timeout_s=180.0, + ) + ) + ) + assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2) diff --git a/tests/engine/test_orchestrator_error_handling.py b/tests/engine/test_orchestrator_error_handling.py index e1ec355ce..90db55edb 100644 --- a/tests/engine/test_orchestrator_error_handling.py +++ b/tests/engine/test_orchestrator_error_handling.py @@ -935,6 +935,36 @@ async def test_diffusion_client_error_output_propagates_status_code(orchestrator orchestrator_fixture.thread.join(timeout=5) +@pytest.mark.asyncio +async def test_duplex_input_processor_failure_is_request_scoped(orchestrator_factory, monkeypatch) -> None: + class FailingStage(FakeStageClient): + def process_engine_inputs(self, *_args, **_kwargs): + raise ValueError("No latent or hidden_states found in thinker output") + + fixture = orchestrator_factory( + [FakeStageClient(stage_type="llm"), FailingStage(stage_type="llm", final_output=True)] + ) + state = OrchestratorRequestState( + request_id="bad", + prompt=SimpleNamespace(request_id="bad", prompt_token_ids=[1]), + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + duplex_identity=SimpleNamespace(), + ) + + async def no_cleanup(*_args, **_kwargs): + pass + + monkeypatch.setattr(fixture.orchestrator, "_cleanup_request_ids", no_cleanup) + try: + await fixture.orchestrator._forward_to_next_stage_unguarded("bad", 0, _build_request_output("raw"), state) + error = await _wait_for_error_message(fixture, request_id="bad") + assert error.fatal is False + finally: + fixture.request_sync_q.put_nowait(ShutdownRequestMessage()) + fixture.thread.join(timeout=5) + + @pytest.mark.asyncio @pytest.mark.parametrize( ("status_code", "error_type"), diff --git a/tests/entrypoints/openai/test_duplex_protocol.py b/tests/entrypoints/openai/test_duplex_protocol.py index 3c3666026..89ae03f6f 100644 --- a/tests/entrypoints/openai/test_duplex_protocol.py +++ b/tests/entrypoints/openai/test_duplex_protocol.py @@ -452,7 +452,7 @@ def test_minicpmo_native_capabilities_separate_model_state_from_core_kv_lease(): assert caps["supports_core_resumable_request"] is True assert caps["supports_stage_connector_handoff"] is True assert caps["supports_audio_truncate"] is True - assert caps["supports_barge_in"] is False + assert caps["supports_barge_in"] is True assert caps["target_barge_in_latency_ms"] is None assert caps["supports_multi_session"] is True assert caps["supports_multi_session_same_replica"] is True diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index 7baced741..467f928ce 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -33,6 +33,7 @@ from vllm_omni.experimental.fullduplex.minicpmo45.runtime import ( from vllm_omni.experimental.fullduplex.minicpmo45.session import ( MiniCPMO45ServingSessionState, ) +from vllm_omni.experimental.fullduplex.openai import vad as realtime_vad from vllm_omni.experimental.fullduplex.openai.protocol import ( DuplexCapabilities, DuplexOverlapPolicy, @@ -706,7 +707,7 @@ async def test_native_realtime_protocol_preserves_input_turn_policy_hints(): @pytest.mark.asyncio -async def test_native_realtime_protocol_rejects_unimplemented_server_vad(): +async def test_native_realtime_protocol_rejects_server_vad_without_interrupt(): ws = TimedWebSocket() protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] protocol.bind_sender(ws.send_json) @@ -729,7 +730,11 @@ async def test_native_realtime_protocol_rejects_unimplemented_server_vad(): error = ws.sent[-1] assert error["type"] == "error" assert error["error"]["code"] == "unsupported_turn_detection" - assert "server_vad" in error["error"]["message"] + assert "interrupt_response=false" in error["error"]["message"] + threshold_error = protocol._validate_realtime_turn_detection( + {"turn_detection": {"type": "server_vad", "threshold": 0.15}} + ) + assert threshold_error == "turn_detection.threshold must be greater than 0.15 and at most 1" @pytest.mark.asyncio @@ -753,7 +758,7 @@ async def test_native_realtime_protocol_accepts_disabled_turn_detection(): @pytest.mark.asyncio -async def test_native_realtime_protocol_rejects_nested_vad_even_when_top_level_is_disabled(): +async def test_native_realtime_protocol_prefers_nested_vad_over_top_level(): ws = TimedWebSocket() protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] protocol.bind_sender(ws.send_json) @@ -773,8 +778,55 @@ async def test_native_realtime_protocol_rejects_nested_vad_even_when_top_level_i } ) - assert translated is None - assert ws.sent[-1]["error"]["code"] == "unsupported_turn_detection" + assert translated["session"]["extra_body"]["realtime_session_payload"]["overlap_policy"] == "barge_in_on_speech" + + +@pytest.mark.asyncio +async def test_realtime_vad_edge_order_and_tiny_chunk_resampling(monkeypatch): + ws = TimedWebSocket() + protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] + protocol.bind_sender(ws.send_json) + protocol._server_vad = realtime_vad.SileroStreamingVAD(realtime_vad.SileroVADConfig()) + result = realtime_vad.StreamingVADResult(True, True, True, True) + monkeypatch.setattr(protocol._server_vad, "process_base64", lambda *args, **kwargs: result) + protocol._input_speech_started = True + protocol._hold_realtime_output_until_session_created = False + await protocol._to_duplex_event( + {"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(1), "format": "pcm_f32le"} + ) + assert [event["type"] for event in ws.sent] == [ + "input_audio_buffer.speech_stopped", + "input_audio_buffer.speech_started", + ] + ws.sent.clear() + protocol._input_speech_started = False + result = realtime_vad.StreamingVADResult(True, False, True, True) + await protocol._to_duplex_event({"type": "input_audio_buffer.append", "audio": "AAAA", "format": "pcm_f32le"}) + assert [event["type"] for event in ws.sent] == [ + "input_audio_buffer.speech_started", + "input_audio_buffer.speech_stopped", + ] + detector = realtime_vad.SileroStreamingVAD(realtime_vad.SileroVADConfig(), frame_scorer=lambda _: 0.0) + for _ in range(1024): + detector.process_base64(_pcm_f32_b64(1), fmt="pcm_f32le", sample_rate_hz=48_000) + assert detector._processed_samples + detector._pending.size == 341 + + +@pytest.mark.asyncio +async def test_realtime_server_vad_runtime_failures_are_terminal(monkeypatch): + handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient())) + ws = TimedWebSocket() + ws.put(_server_vad_update("generic-vad", model="test-model", session_id="generic-vad")) + await handler.handle_session(ws, realtime_protocol=NativeRealtimeSessionProtocol({})) + assert ws.sent[-1]["error"]["code"] == "server_vad_requires_native_duplex" + monkeypatch.setattr(realtime_vad.SileroStreamingVAD, "process_base64", lambda *args, **kwargs: 1 / 0) + ws = TimedWebSocket() + create = _native_realtime_session_update("vad-inference-failure") + create["session"]["turn_detection"] = {"type": "server_vad"} + ws.put(create) + ws.put({"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(1), "format": "pcm_f32le"}) + await asyncio.wait_for(handler.handle_session(ws, realtime_protocol=NativeRealtimeSessionProtocol({})), timeout=2) + assert any(e.get("error", {}).get("code") == "realtime_input_failed" for e in ws.sent) def test_native_duplex_handler_has_no_fixed_session_admission_cap(): @@ -1551,6 +1603,14 @@ def _native_realtime_session_update( } +def _server_vad_update(event_id: str, **session: object) -> dict[str, object]: + return { + "type": "session.update", + "event_id": event_id, + "session": {"turn_detection": {"type": "server_vad"}, **session}, + } + + def test_native_stage0_request_id_matches_engine_after_same_session_id_reopen(): handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient())) first = handler._registry.create(session_id="sid-reused") @@ -5311,7 +5371,7 @@ async def test_audio_clear_preserves_later_wire_order_append(): @pytest.mark.asyncio -async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(): +async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(mocker): class SlowAppendEngine(FakeEngineClient): def __init__(self) -> None: super().__init__() @@ -5348,6 +5408,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(): fence=fence, ) + rollback = mocker.spy(MiniCPMO45PcmAppendBuffer, "_rollback_reservation") engine = SlowAppendEngine() handler = OmniDuplexSessionHandler( chat_service=FakeChatService(engine), @@ -5378,6 +5439,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(): await asyncio.wait_for(handler_task, timeout=1) assert engine.append_cancelled.is_set() + rollback.assert_called_once() assert engine.signal_fences[-1] == DuplexFence("sid-cancel-slow-append") assert engine.signal_next_fences[-1] == DuplexFence("sid-cancel-slow-append", epoch=1) @@ -5473,33 +5535,59 @@ async def test_native_session_update_commits_config_only_after_runtime_ack(): return await super().signal_duplex_turn_async(session_id, **kwargs) engine = BlockingUpdateEngine() - handler = OmniDuplexSessionHandler( - chat_service=FakeChatService(engine), - config_timeout_s=0.1, - idle_timeout_s=1, - ) + handler = OmniDuplexSessionHandler(chat_service=FakeChatService(engine)) ws = TimedWebSocket() - ws.put(_session_create("sid-update-two-phase")) - ws.put( - { - "type": "turn.signal", - "event": "session.update", - "payload": {"instructions": "candidate instructions"}, - } - ) + protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] + ws.put(_native_realtime_session_update("sid-update-two-phase")) + ws.put(_server_vad_update("update-two-phase")) - handler_task = asyncio.create_task(handler.handle_session(ws)) + handler_task = asyncio.create_task(handler.handle_session(ws, realtime_protocol=protocol)) await asyncio.wait_for(engine.update_started.wait(), timeout=1) live_session = handler._registry.get("sid-update-two-phase") assert live_session is not None - assert live_session.config.instructions != "candidate instructions" + assert live_session.config.overlap_policy == DuplexOverlapPolicy.LISTEN_ONLY.value + assert protocol._pending_turn_detection_update is not None + assert protocol._server_vad is None + + protocol.encode_outbound_event({"type": "error", "code": "unrelated", "error": "unrelated"}) + assert protocol._pending_turn_detection_update is not None engine.release_update.set() ws.put({"type": "session.close"}) await asyncio.wait_for(handler_task, timeout=2) - updated = next(message for message in ws.sent if message.get("type") == "session.updated") - assert updated["session"]["instructions"] == "candidate instructions" + updated = next(message for message in reversed(ws.sent) if message.get("type") == "session.updated") + assert updated["session"]["overlap_policy"] == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value + assert isinstance(protocol._server_vad, realtime_vad.SileroStreamingVAD) + + +@pytest.mark.parametrize( + ("append_fails", "expected_code"), + [(True, "session_update_aborted"), (False, "instructions_update_unsupported")], +) +@pytest.mark.asyncio +async def test_realtime_vad_update_rejection_does_not_stall(append_fails, expected_code): + append_result = {"operation": "append", "ok": False, "error_count": 1} if append_fails else None + handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient(append_result=append_result))) + ws = TimedWebSocket() + protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] + create = _native_realtime_session_update("sid-rejected-vad-update") + create["session"]["extra_body"]["auto_response"] = append_fails + ws.put(create) + if append_fails: + ws.put({"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(16_000), "format": "pcm_f32le"}) + update = _server_vad_update("rejected-vad-update") + if not append_fails: + update["session"]["instructions"] = "You are now a pirate." + ws.put(update) + ws.put({"type": "session.close"}) + await asyncio.wait_for(handler.handle_session(ws, realtime_protocol=protocol), timeout=2) + + errors = [message["error"] for message in ws.sent if isinstance(message.get("error"), dict)] + error = next(error for error in errors if error.get("code") == expected_code) + assert error["event_id"] == "rejected-vad-update" + assert protocol._pending_turn_detection_update is None + assert protocol._server_vad is None @pytest.mark.asyncio @@ -5999,7 +6087,7 @@ async def test_minicpmo_native_duplex_rejects_ref_audio_path(): assert engine.opened == [] -def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen(): +def test_minicpmo_native_duplex_explicit_barge_in_request_interrupts(): engine = FakeEngineClient() handler = OmniDuplexSessionHandler( chat_service=FakeChatService(engine), @@ -6026,12 +6114,10 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen( ) assert decision == { - "action": "listen", - "reason": "barge_in_unsupported", + "action": "barge_in", + "reason": "client_force_barge_in", "duration_ms": 1000, - "overlap_speech_ms": 1000, "buffer_audio": True, - "defer_runtime_append": True, } @@ -6043,7 +6129,7 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen( {"type": "turn.signal", "event": "barge_in"}, ], ) -async def test_minicpmo_native_duplex_rejects_unadvertised_barge_in_control( +async def test_minicpmo_native_duplex_accepts_explicit_barge_in_control( barge_in_event: dict[str, object], ): engine = FakeEngineClient() @@ -6059,9 +6145,8 @@ async def test_minicpmo_native_duplex_rejects_unadvertised_barge_in_control( await handler.handle_session(ws) - error = next(message for message in ws.sent if message.get("code") == "barge_in_unsupported") - assert error["session_id"] == "sid-native-barge-control-disabled" - assert ("sid-native-barge-control-disabled", "barge_in") not in engine.signals + assert not any(message.get("code") == "barge_in_unsupported" for message in ws.sent) + assert ("sid-native-barge-control-disabled", "barge_in") in engine.signals @pytest.mark.asyncio diff --git a/tests/model_executor/stage_input_processors/test_nemotron_voicechat.py b/tests/model_executor/stage_input_processors/test_nemotron_voicechat.py index aba69ad14..172faf9fb 100644 --- a/tests/model_executor/stage_input_processors/test_nemotron_voicechat.py +++ b/tests/model_executor/stage_input_processors/test_nemotron_voicechat.py @@ -414,8 +414,10 @@ def test_talker_drains_all_received_positions_per_wake() -> None: session = talker._sessions["req-0"] assert session["step"] == 1 and not session["sync_mode"] - # First decode wake: drains ALL prompt-region steps (t=1..P-1) at once. - _, embeds, update = talker.preprocess(ids, None, _omni_is_prefill=False, **info) + # First resumed wake: drains ALL prompt-region steps (t=1..P-1) at once. + # vLLM 0.28 reports each resumed one-token segment as prefill; the model + # session must still advance instead of restarting its EAR-TTS warmup. + _, embeds, update = talker.preprocess(ids, None, _omni_is_prefill=True, **info) assert session["step"] == prompt_len # t advanced 1 -> P in one wake assert update["codes"]["audio"].shape == (prompt_len - 1, 31) assert float(embeds[0, 0]) == 0.0 # offline stream waits for upstream terminal diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index dc09c2968..e1b078084 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -2638,13 +2638,28 @@ class Orchestrator: req_state.prompt, streaming_context=req_state.streaming, ) - except Exception: + except Exception as exc: logger.exception( "[Orchestrator] req=%s process_engine_inputs FAILED for stage-%s", req_id, next_logical, ) - raise + if not self._is_duplex_session_request(req_state): + raise + await self.output_async_queue.put( + ErrorMessage( + request_id=req_id, + stage_id=next_logical, + error=f"Stage-{next_logical} input processor failed: {type(exc).__name__}: {exc}", + error_type=type(exc).__name__, + ) + ) + await self._cleanup_request_ids( + [req_id, *self._cfg_tracker.cleanup_parent(req_id)], + abort=True, + close_duplex_sessions=True, + ) + return finally: req_state.streaming.source_token_decoder = previous_decoder diff --git a/vllm_omni/experimental/fullduplex/DESIGN.md b/vllm_omni/experimental/fullduplex/DESIGN.md index 256214299..916a2a096 100644 --- a/vllm_omni/experimental/fullduplex/DESIGN.md +++ b/vllm_omni/experimental/fullduplex/DESIGN.md @@ -58,13 +58,13 @@ The checkpoint keeps these verified contracts: - scheduler data-plane append over a resumable request; - stale epoch/turn/response fencing; - `/v1/duplex` and OpenAI Realtime projection; +- optional OpenAI `server_vad` turn detection backed by per-session Silero VAD; - existing JoyVL behavior. The checkpoint does not claim: - scheduler-native KV append; -- deterministic VAD-triggered interruption (the browser intentionally does not - run VAD; MiniCPM owns listen/speak decisions at model-unit boundaries); +- an acoustic-onset-to-cancel latency target across arbitrary client chunk sizes; - production multi-session admission, fairness, capacity, or failure recovery; - bounded long-session KV; - video input or audio/video synchronization. @@ -640,9 +640,8 @@ multi-turn runtime evidence. During auto-response overlap, `preserve_realtime_input` distinguishes "do not append this silent chunk to the native buffer" from "clear the open Realtime -item". Silent overlap no longer discards earlier user PCM. This is an input -ownership correction. It does not add a VAD policy: overlap is admitted to -Stage0 and the model decides whether to listen or speak. +item". Silent overlap no longer discards earlier user PCM. A separate explicit +`server_vad` policy is documented in the Realtime contract above. The first chunk of one overlapping input item also reserves its target model turn. A later Realtime commit uses that reserved identity even if response EOS @@ -1024,6 +1023,6 @@ Passing this checkpoint supports the statement: > Single-session, model-owned MiniCPM-o 4.5 native duplex is reviewable on the > validated H20 configuration. -It does not support claims for deterministic VAD-triggered interruption, -multi-session production concurrency, bounded long-session KV, -scheduler-native append, or video input. +It does not support claims for an acoustic-onset barge-in latency target across +arbitrary client chunk sizes, multi-session production concurrency, bounded +long-session KV, scheduler-native append, or video input. diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py index 77855bf6a..69ff91f99 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/vllm_omni/experimental/fullduplex/client.py @@ -497,6 +497,7 @@ class RealtimeDuplexClient: auto_response: bool = True, temperature: float | None = None, extra_body: dict[str, object] | None = None, + turn_detection: dict[str, object] | None = None, session_id: str | None = None, idle_timeout_s: float | None = None, timeout_s: float = 20.0, @@ -517,8 +518,12 @@ class RealtimeDuplexClient: "modalities": ["audio", "text"], "input_audio_format": "pcm16", "output_audio_format": output_audio_format, - "turn_detection": None, - "overlap_policy": "listen_only", + "turn_detection": dict(turn_detection) if turn_detection is not None else None, + "overlap_policy": ( + "barge_in_on_speech" + if turn_detection is not None and turn_detection.get("interrupt_response", True) is True + else "listen_only" + ), "playback_commit_policy": "ack_only", "extra_body": session_extra_body, } diff --git a/vllm_omni/experimental/fullduplex/openai/protocol.py b/vllm_omni/experimental/fullduplex/openai/protocol.py index 0a87d55b4..d5a9cdcf8 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/vllm_omni/experimental/fullduplex/openai/protocol.py @@ -9,7 +9,6 @@ from uuid import uuid4 class DuplexOverlapPolicy(str, Enum): - AUTO = "auto" LISTEN_ONLY = "listen_only" BARGE_IN_ON_SPEECH = "barge_in_on_speech" @@ -101,7 +100,7 @@ class DuplexCapabilities: supports_multi_session = max_sessions > 1 return cls( supports_model_native_turn_policy=True, - supports_barge_in=False, + supports_barge_in=True, supports_input_append=True, supports_replace_latest_chunk=False, supports_reencode_context=False, @@ -129,6 +128,8 @@ class DuplexCapabilities: signal_sources=["model_native", "client_event", "server_policy"], stage_handoff_transport="scheduler_data_plane", chunk_period_ms=1000, + # Barge-in latency depends on client chunking and lacks hardware E2E + # measurement, so do not advertise an invented target. target_barge_in_latency_ms=None, ) diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_input.py b/vllm_omni/experimental/fullduplex/openai/realtime_input.py index e9799c1b7..bd8e8270b 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_input.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_input.py @@ -18,6 +18,13 @@ from vllm_omni.experimental.fullduplex.openai.realtime_state import ( REALTIME_INPUT_AUDIO_FORMATS, REALTIME_OUTPUT_AUDIO_FORMATS, ) +from vllm_omni.experimental.fullduplex.openai.vad import ( + SILERO_VAD_MIN_THRESHOLD, + ServerVADUnavailableError, + SileroStreamingVAD, + SileroVADConfig, + StreamingVADResult, +) class RealtimeInputTranslator: @@ -92,6 +99,7 @@ class RealtimeInputTranslator: ) ) return None + self._prepare_realtime_turn_detection_update(session_payload) self._apply_realtime_session_defaults(session_payload) session_payload.update(self._realtime_overlap_fields(session_payload)) if not self._opened: @@ -102,6 +110,7 @@ class RealtimeInputTranslator: "type": "turn.signal", "event": "session.update", "payload": session_payload, + "realtime_event_id": event.get("event_id"), } if event_type == "conversation.item.create": item = event.get("item") @@ -278,15 +287,56 @@ class RealtimeInputTranslator: ) ) return None - looks_like_speech = self._input_looks_like_speech(event, audio=audio, fmt=fmt) + detector = self._server_vad + vad_result: StreamingVADResult | None = None + if isinstance(detector, SileroStreamingVAD): + try: + vad_result = detector.process_base64(audio, fmt=fmt, sample_rate_hz=sample_rate_hz) + except (ServerVADUnavailableError, ValueError) as exc: + unavailable = isinstance(exc, ServerVADUnavailableError) + await self._send_realtime_payload( + self._realtime_error_payload( + "server_vad_unavailable" if unavailable else "bad_audio", + str(exc), + event_id=event.get("event_id"), + param="turn_detection" if unavailable else "audio", + ) + ) + return None + looks_like_speech = ( + vad_result.is_speech + if vad_result is not None + else self._input_looks_like_speech(event, audio=audio, fmt=fmt) + ) self._input_audio_buffer_has_audio = self._input_audio_buffer_has_audio or ( looks_like_speech and isinstance(audio, str) and bool(audio) ) self._input_audio_buffer_had_non_speech = self._input_audio_buffer_had_non_speech or ( not looks_like_speech and isinstance(audio, str) and bool(audio) ) - if looks_like_speech: + stopped_event: dict[str, object] | None = None + if vad_result is not None and vad_result.speech_stopped: + stopped_event = dict(event) + if vad_result.speech_end_ms is not None: + stopped_event["audio_end_ms"] = vad_result.speech_end_ms + if stopped_event is not None and vad_result is not None and vad_result.speech_active: + await self._emit_input_speech_stopped( + stopped_event, + item_id=self._active_input_item_id or f"item_{uuid4().hex}", + ) + if vad_result is not None and vad_result.speech_started: + started_event = dict(event) + if vad_result.speech_start_ms is not None: + started_event["audio_start_ms"] = vad_result.speech_start_ms + await self._emit_input_speech_started(started_event) + elif vad_result is None and looks_like_speech: await self._emit_input_speech_started(event) + if stopped_event is not None and vad_result is not None and not vad_result.speech_active: + await self._emit_input_speech_stopped( + stopped_event, + item_id=self._active_input_item_id or f"item_{uuid4().hex}", + ) + if looks_like_speech: self._remember_input_transcript_hint(event) payload = { "type": "input_audio_buffer.append", @@ -309,14 +359,25 @@ class RealtimeInputTranslator: return None payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame] self._copy_realtime_input_hints(event, payload) - if not looks_like_speech: - payload["is_speech"] = False + payload["is_speech"] = looks_like_speech + if vad_result is not None: + payload["vad"] = { + "backend": "silero", + "is_speech": vad_result.is_speech, + "speech_active": vad_result.speech_active, + "speech_started": vad_result.speech_started, + "speech_stopped": vad_result.speech_stopped, + "speech_probability": vad_result.speech_probability, + } + if vad_result.speech_active: + payload["force_listen"] = True return payload if event_type == "input_audio_buffer.commit": if not self._input_audio_buffer_has_audio: if self._input_audio_buffer_had_non_speech: self._input_audio_buffer_had_non_speech = False self._active_input_item_id = None + self._reset_server_vad() return { "type": "input_audio_buffer.commit", "final": event.get("final", True), @@ -338,6 +399,7 @@ class RealtimeInputTranslator: self._active_input_item_id = None self._input_audio_buffer_has_audio = False self._input_audio_buffer_had_non_speech = False + self._reset_server_vad() payload = { "type": "input_audio_buffer.commit", "final": event.get("final", True), @@ -353,6 +415,7 @@ class RealtimeInputTranslator: self._input_audio_buffer_has_audio = False self._input_audio_buffer_had_non_speech = False self._input_audio_buffer_transcript_parts.clear() + self._reset_server_vad() return {"type": "input_audio_buffer.clear", "reason": event_type} if event_type == "output_audio_buffer.clear": payload = {"type": "output_audio_buffer.clear", "reason": event_type} @@ -995,25 +1058,124 @@ class RealtimeInputTranslator: return None @staticmethod - def _validate_realtime_turn_detection(session_payload: dict[str, object]) -> str | None: - configured_values: list[tuple[str, object]] = [] - if "turn_detection" in session_payload: - configured_values.append(("turn_detection", session_payload["turn_detection"])) + def _configured_realtime_turn_detection(session_payload: dict[str, object]) -> tuple[str | None, object]: + field = "turn_detection" if "turn_detection" in session_payload else None + selected = session_payload.get("turn_detection") audio_config = session_payload.get("audio") if isinstance(audio_config, dict): audio_input = audio_config.get("input") if isinstance(audio_input, dict) and "turn_detection" in audio_input: - configured_values.append(("audio.input.turn_detection", audio_input["turn_detection"])) - for field_path, turn_detection in configured_values: - if turn_detection is None: - continue - turn_detection_type = turn_detection.get("type") if isinstance(turn_detection, dict) else turn_detection - return ( - f"{field_path}={turn_detection_type!r} is not implemented by the duplex Realtime adapter; " - "set turn_detection to null and commit input explicitly, or use the model-owned duplex policy" - ) + field, selected = "audio.input.turn_detection", audio_input["turn_detection"] + return field, selected + + @classmethod + def _validate_realtime_turn_detection(cls, session_payload: dict[str, object]) -> str | None: + field, turn_detection = cls._configured_realtime_turn_detection(session_payload) + if field is None: + if session_payload.get("overlap_policy") == "barge_in_on_speech": + return ( + "overlap_policy='barge_in_on_speech' requires " + "turn_detection.type='server_vad' on the Realtime endpoint" + ) + return None + if turn_detection is not None and not isinstance(turn_detection, dict): + return f"{field} must be null or an object with type='server_vad'" + if isinstance(turn_detection, dict): + if turn_detection.get("type") != "server_vad": + return f"{field}.type must be 'server_vad'" + threshold = turn_detection.get("threshold", 0.5) + if not cls._valid_vad_number(threshold, minimum=SILERO_VAD_MIN_THRESHOLD, strict=True, maximum=1): + return f"{field}.threshold must be greater than {SILERO_VAD_MIN_THRESHOLD} and at most 1" + for name in ("prefix_padding_ms", "silence_duration_ms", "min_speech_duration_ms"): + value = turn_detection.get(name) + if value is not None and not cls._valid_vad_number(value, minimum=0): + return f"{field}.{name} must be a non-negative number" + if turn_detection.get("interrupt_response", True) is not True: + return ( + f"{field}.interrupt_response=false is unsupported; " + "use turn_detection=null for model-owned listen/speak" + ) + desired_policy = "barge_in_on_speech" if turn_detection is not None else "listen_only" + overlap_policy = session_payload.get("overlap_policy") + if isinstance(overlap_policy, str) and overlap_policy != desired_policy: + return f"overlap_policy={overlap_policy!r} conflicts with turn_detection; expected {desired_policy!r}" return None + @staticmethod + def _valid_vad_number(value: object, *, minimum: float, maximum: float | None = None, strict: bool = False) -> bool: + return ( + isinstance(value, int | float) + and not isinstance(value, bool) + and np.isfinite(float(value)) + and (float(value) > minimum if strict else float(value) >= minimum) + and (maximum is None or float(value) <= maximum) + ) + + def _prepare_realtime_turn_detection_update(self, session_payload: dict[str, object]) -> None: + field, configured = self._configured_realtime_turn_detection(session_payload) + if field is None: + return + if self._pending_turn_detection_update is not None: + raise RuntimeError("A Realtime turn-detection update is already pending") + + turn_detection: dict[str, object] | None = None + detector: SileroStreamingVAD | None = None + if configured is None: + session_payload["overlap_policy"] = "listen_only" + else: + assert isinstance(configured, dict) + turn_detection = { + "type": "server_vad", + "interrupt_response": True, + "threshold": 0.5, + "prefix_padding_ms": 300, + "silence_duration_ms": 500, + "min_speech_duration_ms": 96, + **configured, + } + detector = SileroStreamingVAD( + SileroVADConfig( + threshold=float(turn_detection["threshold"]), + prefix_padding_ms=int(turn_detection["prefix_padding_ms"]), + silence_duration_ms=int(turn_detection["silence_duration_ms"]), + min_speech_duration_ms=max(32, int(turn_detection["min_speech_duration_ms"])), + ) + ) + session_payload["turn_detection"] = dict(turn_detection) + session_payload["overlap_policy"] = "barge_in_on_speech" + + self._pending_turn_detection_update = (turn_detection, detector, asyncio.Event()) + + async def wait_for_realtime_turn_detection_update(self) -> None: + candidate = self._pending_turn_detection_update + if candidate is not None: + await candidate[2].wait() + + def commit_realtime_turn_detection_update(self) -> None: + self._resolve_realtime_turn_detection_update(commit=True) + + def reject_realtime_turn_detection_update(self) -> None: + self._resolve_realtime_turn_detection_update(commit=False) + + def _resolve_realtime_turn_detection_update(self, *, commit: bool) -> None: + candidate = self._pending_turn_detection_update + if candidate is None: + return + turn_detection, detector, resolved = candidate + discarded = self._server_vad if commit else detector + if commit: + self._turn_detection, self._server_vad = turn_detection, detector + self._pending_turn_detection_update = None + resolved.set() + reset = getattr(discarded, "reset", None) + if callable(reset): + reset() + + def _reset_server_vad(self) -> None: + reset = getattr(self._server_vad, "reset", None) + if callable(reset): + reset() + @staticmethod def _input_audio_transcription_config(session_payload: dict[str, object]) -> dict[str, object] | None: transcription = session_payload.get("input_audio_transcription") diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_output.py b/vllm_omni/experimental/fullduplex/openai/realtime_output.py index 7729092e3..2a68f79ad 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_output.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_output.py @@ -262,15 +262,8 @@ class RealtimeOutputProjector: content_index=0, audio_end_ms=committed_audio_ms, ) - payloads.append( - { - "type": "conversation.item.truncated", - "item_id": item_id, - "content_index": 0, - "audio_end_ms": committed_audio_ms, - "event": event, - } - ) + # Align server history with acknowledged playback; only explicit + # client truncation emits ``conversation.item.truncated``. payloads.extend(self._realtime_audio_done_events(event, response_id)) payloads.extend( self._realtime_response_terminal_events( @@ -413,7 +406,7 @@ class RealtimeOutputProjector: return [event] message = str(raw_error or event.get("message") or "Duplex runtime error") code = str(event.get("code") or "duplex_error") - return [self._realtime_error_payload(code, message)] + return [self._realtime_error_payload(code, message, event_id=event.get("realtime_event_id"))] if event_type == "session.closed": return [{"type": "session.closed", "event": event}] return [{"type": f"duplex.{event_type}", "event": event}] diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_session.py b/vllm_omni/experimental/fullduplex/openai/realtime_session.py index 519c071cf..a85ab1354 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_session.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_session.py @@ -58,6 +58,7 @@ class NativeRealtimeSessionProtocol( while True: if not self._pending_outbound.empty(): return json.dumps(await self._pending_outbound.get()) + await self.wait_for_realtime_turn_detection_update() raw = await websocket.receive_text() try: event = json.loads(raw) diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_state.py b/vllm_omni/experimental/fullduplex/openai/realtime_state.py index ac3b2a8de..de05cc71c 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_state.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_state.py @@ -57,6 +57,7 @@ REALTIME_ERROR_TYPES_BY_CODE = { "native_text_append_unsupported": "invalid_request_error", "runtime_native_stage_role_required": "server_error", "runtime_native_runner_kv_required": "server_error", + "server_vad_unavailable": "server_error", } @@ -105,6 +106,9 @@ class RealtimeSessionState: _input_sample_rate_hz: int = 16000 _output_audio_format: str = "pcm16" _overlap_silence_rms: float = 0.003 + _turn_detection: dict[str, object] | None = None + _server_vad: object | None = None + _pending_turn_detection_update: tuple[dict[str, object] | None, object | None, asyncio.Event] | None = None _send_realtime_json: Any = None _initial_session_update: bool = False _input_speech_started: bool = False @@ -180,6 +184,11 @@ class RealtimeStateOwner: _input_sample_rate_hz: int = _RealtimeStateField() _output_audio_format: str = _RealtimeStateField() _overlap_silence_rms: float = _RealtimeStateField() + _turn_detection: dict[str, object] | None = _RealtimeStateField() + _server_vad: object | None = _RealtimeStateField() + _pending_turn_detection_update: tuple[dict[str, object] | None, object | None, asyncio.Event] | None = ( + _RealtimeStateField() + ) _send_realtime_json: Any = _RealtimeStateField() _initial_session_update: bool = _RealtimeStateField() _input_speech_started: bool = _RealtimeStateField() diff --git a/vllm_omni/experimental/fullduplex/openai/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index 516d9708f..d0ddccb27 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/vllm_omni/experimental/fullduplex/openai/serving.py @@ -494,6 +494,22 @@ class OmniDuplexSessionHandler( # playback only controls history ACKs, not model admission. if is_speech: session.accumulate_overlap_speech(duration_ms) + vad_speech_started = self._vad_speech_started(event, payload) + if ( + self._uses_native_input_append(session) + and session.capabilities.supports_barge_in + and is_speech + and session.config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value + and vad_speech_started is not False + ): + return { + "action": "barge_in", + "reason": ("server_vad_speech_started" if vad_speech_started is True else "barge_in_on_speech"), + "cancel_reason": "turn_detected" if vad_speech_started is True else "barge_in", + "duration_ms": duration_ms, + "overlap_speech_ms": session.overlap_speech_ms, + "buffer_audio": True, + } return { "action": "listen", "reason": "auto_response_continuous", @@ -501,7 +517,7 @@ class OmniDuplexSessionHandler( "overlap_speech_ms": session.overlap_speech_ms, "buffer_audio": True, "defer_runtime_append": False, - "force_listen": False, + "force_listen": event.get("force_listen") is True or payload.get("force_listen") is True, "preserve_realtime_input": True, } if bool(event.get("force_listen", False)): @@ -553,9 +569,22 @@ class OmniDuplexSessionHandler( session.accumulate_overlap_speech(duration_ms) if policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value: + vad_speech_started = self._vad_speech_started(event, payload) + if vad_speech_started is False: + return { + "action": "listen", + "reason": "server_vad_utterance_active", + "duration_ms": duration_ms, + "overlap_speech_ms": session.overlap_speech_ms, + "buffer_audio": True, + "defer_runtime_append": False, + "force_listen": True, + "preserve_realtime_input": True, + } return { "action": "barge_in", - "reason": "policy_barge_in_on_speech", + "reason": ("server_vad_speech_started" if vad_speech_started is True else "policy_barge_in_on_speech"), + "cancel_reason": "turn_detected" if vad_speech_started is True else "barge_in", "duration_ms": duration_ms, "overlap_speech_ms": session.overlap_speech_ms, "buffer_audio": True, @@ -599,6 +628,17 @@ class OmniDuplexSessionHandler( "defer_runtime_append": True, } + @staticmethod + def _vad_speech_started( + event: Mapping[str, object], + payload: Mapping[str, object], + ) -> bool | None: + for source in (event, payload): + vad = source.get("vad") + if isinstance(vad, Mapping) and isinstance(vad.get("speech_started"), bool): + return bool(vad["speech_started"]) + return None + @staticmethod def _event_requests_barge_in(event: Mapping[str, object]) -> bool: if event.get("force_barge_in") is True: @@ -1155,7 +1195,13 @@ class OmniDuplexSessionHandler( candidate_config: DuplexSessionConfig, ) -> dict[str, object] | None: if not self._uses_native_input_append(session): - return None + if candidate_config.overlap_policy != DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value: + return None + return { + "type": "error", + "code": "server_vad_requires_native_duplex", + "error": "Realtime server_vad requires a model-native duplex runtime", + } if ( candidate_config.instructions != session.config.instructions and self._runtime_session_state(session).native_context_locked diff --git a/vllm_omni/experimental/fullduplex/openai/session_runner.py b/vllm_omni/experimental/fullduplex/openai/session_runner.py index 23aee61b6..b7694c6a0 100644 --- a/vllm_omni/experimental/fullduplex/openai/session_runner.py +++ b/vllm_omni/experimental/fullduplex/openai/session_runner.py @@ -18,6 +18,7 @@ from vllm_omni.experimental.fullduplex.openai.commit_policy import ( decide_commit_action, ) from vllm_omni.experimental.fullduplex.openai.protocol import ( + DuplexOverlapPolicy, DuplexPlaybackCommitPolicy, DuplexSession, DuplexSessionState, @@ -248,6 +249,9 @@ class DuplexSessionRunnerMixin: await actor.enqueue_event(event) except WebSocketDisconnect: await actor.enqueue_event({"type": "__disconnect__"}) + except Exception as exc: + await emit_event({"type": "error", "error": str(exc), "code": "realtime_input_failed"}) + await actor.enqueue_event({"type": "__disconnect__"}) async def next_actor_event() -> dict[str, object]: nonlocal pending_turn_reservations, transport_detached @@ -411,6 +415,8 @@ class DuplexSessionRunnerMixin: ) return append_ok except asyncio.CancelledError: + if pcm_reservation is not None: + pcm_reservation.rollback() raise except Exception as exc: if pcm_reservation is not None: @@ -670,6 +676,9 @@ class DuplexSessionRunnerMixin: if handshake is None: return session = handshake.session + if initial_update_error := self._runtime_session_candidate_update_error(session, session.config): + await emit_event(initial_update_error) + return if handshake.resumed: native = self._serving_runtime_adapter.session_states[session.session_id] actor.tasks = self._session_tasks[session.session_id] @@ -721,6 +730,8 @@ class DuplexSessionRunnerMixin: ) if isinstance(open_result, dict): created_payload["runtime_control"] = self._redact_runtime_control_result(open_result) + if realtime_protocol is not None: + realtime_protocol.commit_realtime_turn_detection_update() await emit_event(created_payload) resume_credential_delivered = session.capabilities.supports_session_resume reader_task = asyncio.create_task(read_event_loop(), name="duplex-session-reader") @@ -903,7 +914,11 @@ class DuplexSessionRunnerMixin: if event_type in {"input.cancel", "response.cancel", "barge_in", "output_audio_buffer.clear"}: cancel_reason = ( - "output_audio_buffer_clear" if event_type == "output_audio_buffer.clear" else "barge_in" + "output_audio_buffer_clear" + if event_type == "output_audio_buffer.clear" + else "client_cancelled" + if event_type == "response.cancel" + else "barge_in" ) cancelled_fence = DuplexFence( session.session_id, @@ -1068,9 +1083,20 @@ class DuplexSessionRunnerMixin: await emit_event(self._barge_in_unsupported_error(session)) continue if turn_event == "session.update": + realtime_event_id = event.get("realtime_event_id") + + async def emit_update_event(update_event: dict[str, object]) -> None: + if isinstance(realtime_event_id, str) and realtime_event_id: + update_event = {**update_event, "realtime_event_id": realtime_event_id} + await emit_event(update_event) + + def reject_update() -> None: + if realtime_protocol is not None: + realtime_protocol.reject_realtime_turn_detection_update() + payload = event.get("payload") if not isinstance(payload, dict): - await emit_event( + await emit_update_event( { "type": "error", "session_id": session.session_id, @@ -1078,12 +1104,22 @@ class DuplexSessionRunnerMixin: "error": "session.update requires a session payload", } ) + reject_update() continue if not await wait_for_native_append_tail(): + await emit_update_event( + { + "type": "error", + "code": "session_update_aborted", + "error": "session.update was not applied because the preceding append failed", + } + ) + reject_update() continue runtime_update_error = self._runtime_session_update_error(session, payload) if runtime_update_error is not None: - await emit_event(runtime_update_error) + await emit_update_event(runtime_update_error) + reject_update() continue previous_config = session.config candidate_config = deepcopy(previous_config) @@ -1093,14 +1129,16 @@ class DuplexSessionRunnerMixin: finally: session.replace_config(previous_config) if update_error is not None: - await emit_event(update_error) + await emit_update_event(update_error) + reject_update() continue runtime_update_error = self._runtime_session_candidate_update_error( session, candidate_config, ) if runtime_update_error is not None: - await emit_event(runtime_update_error) + await emit_update_event(runtime_update_error) + reject_update() continue try: candidate_runtime_config = self._runtime_config_for_session_update( @@ -1108,7 +1146,7 @@ class DuplexSessionRunnerMixin: candidate_config, ) except ServingRuntimeConfigError as exc: - await emit_event( + await emit_update_event( { "type": "error", "session_id": session.session_id, @@ -1116,18 +1154,22 @@ class DuplexSessionRunnerMixin: "error": str(exc), } ) + reject_update() continue if not await self._signal_runtime_session( session, turn_event, - emit_event, + emit_update_event, session_config=candidate_config.as_dict(), runtime_config=candidate_runtime_config, ): + reject_update() continue session.replace_config(candidate_config) session.replace_runtime_config(candidate_runtime_config) - await emit_event( + if realtime_protocol is not None: + realtime_protocol.commit_realtime_turn_detection_update() + await emit_update_event( { "type": "session.updated", "session": session.as_public_dict(), @@ -1337,7 +1379,16 @@ class DuplexSessionRunnerMixin: buffer_overlap_audio = True if self._uses_native_input_append(session): mark_pending_silence_superseded() - overlap_active = not self._session_auto_responds(session) and native_response_in_progress() + overlap_active = native_response_in_progress() and ( + not self._session_auto_responds(session) + or ( + session.capabilities.supports_barge_in + and ( + session.config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value + or self._event_requests_barge_in(event) + ) + ) + ) if overlap_active: decision = self._overlap_decision(session, event, payload) await self._emit_overlap_decision(emit_event, session, decision) @@ -1378,11 +1429,12 @@ class DuplexSessionRunnerMixin: native.speech_since_commit = False await actor.cancel_append_tasks() had_native_stream = native.data_plane_task is not None + cancel_reason = str(decision.get("cancel_reason") or "barge_in") cancelled = await self._cancel_active_response( session, actor.active_response_task, emit_event, - reason="barge_in", + reason=cancel_reason, ) had_native_stream = ( await self._cancel_native_data_plane_stream(session) or had_native_stream @@ -1398,7 +1450,7 @@ class DuplexSessionRunnerMixin: "type": "audio.cancelled", "session_id": session.session_id, "response_id": old_response_id, - "reason": "barge_in", + "reason": cancel_reason, "cancelled_epoch": old_epoch, "epoch": new_epoch, "committed_ms": committed_ms, @@ -1418,7 +1470,7 @@ class DuplexSessionRunnerMixin: "type": "audio.cancelled", "session_id": session.session_id, "response_id": session.last_response_id, - "reason": "barge_in", + "reason": cancel_reason, "cancelled_epoch": old_epoch, "epoch": new_epoch, "committed_ms": committed_ms, @@ -1950,6 +2002,8 @@ class DuplexSessionRunnerMixin: with suppress(Exception): await emit_event({"type": "error", "error": str(exc), "code": "internal_error"}) finally: + if realtime_protocol is not None: + realtime_protocol.reject_realtime_turn_detection_update() if session is not None: if reader_task is not None and not reader_task.done(): reader_task.cancel() diff --git a/vllm_omni/experimental/fullduplex/openai/vad.py b/vllm_omni/experimental/fullduplex/openai/vad.py new file mode 100644 index 000000000..f81559613 --- /dev/null +++ b/vllm_omni/experimental/fullduplex/openai/vad.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import binascii +from collections.abc import Callable +from dataclasses import dataclass + +import numpy as np +import pybase64 as base64 + +SILERO_VAD_MIN_THRESHOLD = 0.15 + + +class ServerVADUnavailableError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class SileroVADConfig: + threshold: float = 0.5 + prefix_padding_ms: int = 300 + silence_duration_ms: int = 500 + min_speech_duration_ms: int = 96 + + +@dataclass(frozen=True, slots=True) +class StreamingVADResult: + is_speech: bool + speech_active: bool + speech_started: bool = False + speech_stopped: bool = False + speech_probability: float = 0.0 + speech_start_ms: int | None = None + speech_end_ms: int | None = None + + +class SileroStreamingVAD: + _SAMPLE_RATE_HZ = 16_000 + _WINDOW_SAMPLES = 512 + + def __init__( + self, + config: SileroVADConfig, + *, + frame_scorer: Callable[[np.ndarray], float] | None = None, + ) -> None: + self.config = config + self._frame_scorer = frame_scorer + self._model: object | None = None + self._clear_stream_state() + + def _clear_stream_state(self) -> None: + self._pending = np.empty(0, dtype=np.float32) + self._speech_active = False + self._candidate_samples = 0 + self._candidate_start_sample: int | None = None + self._silence_samples = 0 + self._processed_samples = 0 + self._resample_rate_hz: int | None = None + self._resample_remainder = 0 + + def reset(self) -> None: + self._clear_stream_state() + reset_states = getattr(self._model, "reset_states", None) + if callable(reset_states): + reset_states() + + def process_base64( + self, + audio: object, + *, + fmt: object, + sample_rate_hz: object, + ) -> StreamingVADResult: + if fmt != "pcm_f32le" or not isinstance(audio, str): + raise ValueError("Silero server VAD requires decoded pcm_f32le audio") + try: + raw = base64.b64decode(audio, validate=True) + except (binascii.Error, ValueError) as exc: + raise ValueError("Silero server VAD received invalid base64 audio") from exc + if len(raw) % 4: + raise ValueError("Silero server VAD received an incomplete pcm_f32le frame") + samples = np.frombuffer(raw, dtype=" StreamingVADResult: + samples = np.asarray(samples, dtype=np.float32).reshape(-1) + if samples.size: + self._pending = np.concatenate((self._pending, samples)) + + started = False + stopped = False + start_ms: int | None = None + end_ms: int | None = None + max_probability = 0.0 + contained_speech = self._speech_active + min_speech_samples = self.config.min_speech_duration_ms * 16 + min_silence_samples = self.config.silence_duration_ms * 16 + negative_threshold = max(0.0, self.config.threshold - SILERO_VAD_MIN_THRESHOLD) + + while self._pending.size >= self._WINDOW_SAMPLES: + frame = self._pending[: self._WINDOW_SAMPLES] + self._pending = self._pending[self._WINDOW_SAMPLES :] + frame_start = self._processed_samples + self._processed_samples += self._WINDOW_SAMPLES + probability = min(1.0, max(0.0, float(self._score_frame(frame)))) + max_probability = max(max_probability, probability) + + if self._speech_active: + contained_speech = True + if probability < negative_threshold: + self._silence_samples += self._WINDOW_SAMPLES + if self._silence_samples >= min_silence_samples: + speech_end_sample = self._processed_samples - self._silence_samples + self._speech_active = False + self._silence_samples = 0 + stopped = True + end_ms = max(0, int(round(speech_end_sample / 16))) + else: + self._silence_samples = 0 + continue + + if probability >= self.config.threshold: + if self._candidate_samples == 0: + self._candidate_start_sample = frame_start + self._candidate_samples += self._WINDOW_SAMPLES + if self._candidate_samples >= min_speech_samples: + candidate_start = self._candidate_start_sample or 0 + prefix_samples = self.config.prefix_padding_ms * 16 + self._speech_active = True + self._candidate_samples = self._silence_samples = 0 + self._candidate_start_sample = None + contained_speech = True + started = True + start_ms = max(0, int(round((candidate_start - prefix_samples) / 16))) + else: + self._candidate_samples = 0 + self._candidate_start_sample = None + + return StreamingVADResult( + contained_speech, self._speech_active, started, stopped, max_probability, start_ms, end_ms + ) + + def _score_frame(self, frame: np.ndarray) -> float: + if self._frame_scorer is not None: + return float(self._frame_scorer(frame)) + if self._model is None: + try: + import torch + from silero_vad import load_silero_vad + except ImportError as exc: + raise ServerVADUnavailableError( + "server_vad requires the optional 'silero-vad' package; install vllm-omni[server-vad]" + ) from exc + self._model = load_silero_vad(onnx=False) + self._frame_scorer = lambda value: float( + self._model(torch.from_numpy(np.ascontiguousarray(value)), self._SAMPLE_RATE_HZ).item() + ) + return float(self._frame_scorer(frame)) diff --git a/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_talker.py b/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_talker.py index 5764afa98..515b29c6c 100644 --- a/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_talker.py +++ b/vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_talker.py @@ -392,14 +392,13 @@ class NemotronVoiceChatTalkerForConditionalGeneration(nn.Module): info = merge_runtime_info(info_dict) device = input_ids.device span = int(input_ids.shape[0]) - is_prefill_raw = info.get("_omni_is_prefill") - is_prefill = bool(is_prefill_raw) if isinstance(is_prefill_raw, bool) else span > 1 request_id = require_request_id(info, "talker") embeds = torch.zeros((span, self._hidden), device=device, dtype=self._dtype) - if is_prefill or request_id not in self._sessions: - # The 1-token vLLM prefill is the session boundary: run the vendored - # warmup here so the first decode step is a pure NeMo t=1 step. + if request_id not in self._sessions: + # The request id, not vLLM's per-segment prefill flag, is the + # session boundary. Resumable updates are prefill again in v0.28 + # and must preserve the existing EAR-TTS cache and step. self._init_session(request_id, info, device) return input_ids, embeds, {} diff --git a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py index 5bbbfbc4a..9f094dada 100644 --- a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py +++ b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py @@ -550,6 +550,29 @@ def _native_tts_boundary_token_ids(special_token_ids): } +def _native_duplex_output_is_control_only( + output_ids: Sequence[int], + special_token_ids: Mapping[str, int], +) -> bool: + if not output_ids: + return True + control_ids = { + token_id + for token_id in ( + special_token_ids.get("tts_eos_token_id"), + special_token_ids.get("tts_pad_token_id"), + special_token_ids.get("listen_token_id"), + special_token_ids.get("chunk_eos_token_id"), + special_token_ids.get("chunk_tts_eos_token_id"), + special_token_ids.get("turn_eos_token_id"), + special_token_ids.get("unit_token_id"), + special_token_ids.get("unit_end_token_id"), + ) + if token_id is not None + } + return bool(control_ids) and all(token_id in control_ids for token_id in output_ids) + + def _decode_native_duplex_token_ids( token_ids: list[int], streaming_context, @@ -757,10 +780,16 @@ def llm2tts( ) prompt_token_ids_len = len(prompt_token_ids) + is_native_duplex_handoff = _has_native_duplex_prompt_metadata(mm_output) + latent = mm_output.get("latent", None) if latent is None: latent = output.hidden_states if hasattr(output, "hidden_states") else None if latent is None: + if is_native_duplex_handoff and _native_duplex_output_is_control_only( + llm_output_ids, special_token_ids + ): + continue raise ValueError("No latent or hidden_states found in thinker output") thinker_hidden_states = latent.detach() @@ -771,7 +800,6 @@ def llm2tts( full_token_ids = prompt_token_ids + llm_output_ids tts_bos_id = special_token_ids.get("tts_bos_token_id") - is_native_duplex_handoff = _has_native_duplex_prompt_metadata(mm_output) if is_native_duplex_handoff: _require_native_tts_boundary_metadata(special_token_ids) tts_end_ids = _native_tts_boundary_token_ids(special_token_ids)