diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml index 51d752534..2d3be8bc4 100644 --- a/.buildkite/cuda/test-merge.yml +++ b/.buildkite/cuda/test-merge.yml @@ -166,7 +166,6 @@ 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 830e6aa63..ff55409b3 100644 --- a/.buildkite/cuda/test-ready.yml +++ b/.buildkite/cuda/test-ready.yml @@ -229,7 +229,6 @@ 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/pyproject.toml b/pyproject.toml index a02825dce..84a950719 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,7 +74,6 @@ dev = [ "kernels==0.14.1", # for accuracy tests comparison with diffusers output (0.15.0 incompatible with diffusers 0.38.0) "s3tokenizer==0.3.0", # for tts test "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,10 +83,6 @@ 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/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py b/tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py index 0f315475a..d9fbc452b 100644 --- a/tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py +++ b/tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py @@ -59,63 +59,6 @@ def _make_scheduler(policy: SchedulingPolicy, *, waiting=(), skipped_waiting=(), return scheduler -@pytest.mark.parametrize("policy", _POLICIES) -def test_sweep_drops_aborted_request_from_skipped_waiting(policy: SchedulingPolicy) -> None: - """The sweep must cover ``skipped_waiting``. - - Upstream ``schedule()`` admits from ``skipped_waiting`` as readily as from - ``waiting``, so an aborted request left there is re-selected and raises. - """ - aborted = _StubRequest("req-aborted", RequestStatus.FINISHED_ABORTED) - scheduler = _make_scheduler(policy, skipped_waiting=[aborted]) - - scheduler._drop_aborted_queued_requests() - - assert list(scheduler.skipped_waiting) == [] - - -@pytest.mark.parametrize("policy", _POLICIES) -def test_sweep_drops_aborted_requests_from_waiting_and_running(policy: SchedulingPolicy) -> None: - """``waiting`` and ``running`` stay covered alongside the new queue.""" - aborted_waiting = _StubRequest("req-waiting", RequestStatus.FINISHED_ABORTED) - aborted_running = _StubRequest("req-running", RequestStatus.FINISHED_ABORTED) - scheduler = _make_scheduler( - policy, - waiting=[aborted_waiting], - running=[aborted_running], - ) - - scheduler._drop_aborted_queued_requests() - - assert list(scheduler.waiting) == [] - assert scheduler.running == [] - - -@pytest.mark.parametrize("policy", _POLICIES) -def test_sweep_leaves_live_requests_intact(policy: SchedulingPolicy) -> None: - """The sweep must not over-sweep. - - A request parked in ``skipped_waiting`` on a blocked waiting status is one - upstream still intends to schedule; only the aborted entry may go. - """ - blocked = _StubRequest("req-blocked", RequestStatus.WAITING_FOR_STREAMING_REQ, arrival_time=0.0) - aborted = _StubRequest("req-aborted", RequestStatus.FINISHED_ABORTED, arrival_time=1.0) - live_waiting = _StubRequest("req-live-waiting", RequestStatus.WAITING) - live_running = _StubRequest("req-live-running", RequestStatus.RUNNING) - scheduler = _make_scheduler( - policy, - waiting=[live_waiting], - skipped_waiting=[blocked, aborted], - running=[live_running], - ) - - scheduler._drop_aborted_queued_requests() - - assert list(scheduler.skipped_waiting) == [blocked] - assert list(scheduler.waiting) == [live_waiting] - assert scheduler.running == [live_running] - - @pytest.mark.parametrize("policy", _POLICIES) def test_schedule_sweeps_skipped_waiting_before_upstream_selection( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/core/sched/test_omni_scheduler_streaming_input_counter.py b/tests/core/sched/test_omni_scheduler_streaming_input_counter.py index 34529e836..2fbb320eb 100644 --- a/tests/core/sched/test_omni_scheduler_streaming_input_counter.py +++ b/tests/core/sched/test_omni_scheduler_streaming_input_counter.py @@ -60,12 +60,6 @@ def _make_scheduler(scheduler_cls, *, requests, running, waiting, counter, skipp return scheduler -def _upstream_num_unfinished(scheduler) -> int: - """Mirror of ``Scheduler.get_num_unfinished_requests`` arithmetic.""" - num_waiting = len(scheduler.waiting) + len(scheduler.skipped_waiting) - scheduler.num_waiting_for_streaming_input - return num_waiting + len(scheduler.running) - - @pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS) def test_finish_requests_clears_counter_left_by_a_stomped_status( monkeypatch: pytest.MonkeyPatch, @@ -99,73 +93,6 @@ def test_finish_requests_clears_counter_left_by_a_stomped_status( assert scheduler.num_waiting_for_streaming_input == 0 -@pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS) -def test_next_request_is_visible_as_work_after_the_abort( - monkeypatch: pytest.MonkeyPatch, - scheduler_cls, - scheduler_mod, -) -> None: - """The consequence, stated as the engine sees it. - - A fresh request queued behind the leak must read as unfinished work. - Without the resync this is ``(1 + 0 - 1) + 0 == 0`` and the engine parks - in ``input_queue.get()`` with a live request in ``waiting``. - """ - aborted = _StubRequest("req-aborted", RequestStatus.RUNNING) - incoming = _StubRequest("req-next-session", RequestStatus.WAITING) - scheduler = _make_scheduler( - scheduler_cls, - requests={incoming.request_id: incoming}, - running=[], - waiting=[incoming], - counter=1, - ) - - monkeypatch.setattr( - scheduler_mod.VLLMScheduler, - "finish_requests", - lambda self, request_ids, finished_status: [aborted], - ) - - scheduler_cls.finish_requests(scheduler, [aborted.request_id], RequestStatus.FINISHED_ABORTED) - - assert _upstream_num_unfinished(scheduler) == 1 - - -@pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS) -def test_finish_requests_keeps_counting_a_genuinely_parked_request( - monkeypatch: pytest.MonkeyPatch, - scheduler_cls, - scheduler_mod, -) -> None: - """The resync must not over-correct. - - A request still parked as ``WAITING_FOR_STREAMING_REQ`` in ``waiting`` is - exactly what the counter is for -- it must stay counted, so the engine - does not spin on a request that has no input to process. - """ - parked = _StubRequest("req-parked", RequestStatus.WAITING_FOR_STREAMING_REQ) - leaving = _StubRequest("req-leaving", RequestStatus.FINISHED_STOPPED) - scheduler = _make_scheduler( - scheduler_cls, - requests={parked.request_id: parked}, - running=[], - waiting=[parked], - counter=1, - ) - - monkeypatch.setattr( - scheduler_mod.VLLMScheduler, - "finish_requests", - lambda self, request_ids, finished_status: [leaving], - ) - - scheduler_cls.finish_requests(scheduler, [leaving.request_id], RequestStatus.FINISHED_STOPPED) - - assert scheduler.num_waiting_for_streaming_input == 1 - assert _upstream_num_unfinished(scheduler) == 0 - - def test_resync_counts_parked_requests_in_skipped_waiting() -> None: """``get_num_unfinished_requests`` sums ``waiting`` and ``skipped_waiting``, so the resync has to look at both or it under-counts and reintroduces the @@ -185,36 +112,6 @@ def test_resync_counts_parked_requests_in_skipped_waiting() -> None: assert scheduler.num_waiting_for_streaming_input == 1 -def test_resync_is_a_no_op_when_upstream_accounting_is_correct() -> None: - """In the healthy case the derived value equals upstream's incremental one, - so the helper must leave it alone rather than fight it.""" - parked = _StubRequest("req-parked", RequestStatus.WAITING_FOR_STREAMING_REQ) - running = _StubRequest("req-running", RequestStatus.RUNNING) - scheduler = _make_scheduler( - OmniARScheduler, - requests={r.request_id: r for r in (parked, running)}, - running=[running], - waiting=[parked], - counter=1, - ) - - scheduler._resync_streaming_input_counter() - - assert scheduler.num_waiting_for_streaming_input == 1 - - -def test_resync_tolerates_a_scheduler_without_the_counter() -> None: - """Guard for vLLM versions that predate ``num_waiting_for_streaming_input``: - the helper must no-op rather than invent the attribute.""" - scheduler = OmniARScheduler.__new__(OmniARScheduler) - scheduler.waiting = [] - scheduler.skipped_waiting = [] - - scheduler._resync_streaming_input_counter() - - assert not hasattr(scheduler, "num_waiting_for_streaming_input") - - def test_ar_schedule_resyncs_before_delegating_upstream(monkeypatch: pytest.MonkeyPatch) -> None: """Pins placement in ``schedule``: the chunk-transfer adapter rewrites status during ``process_pending_chunks``, so the resync has to run after 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 deleted file mode 100644 index 1bd3db703..000000000 --- a/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py +++ /dev/null @@ -1,107 +0,0 @@ -"""MiniCPM-o server-VAD hard-interruption acceptance driver.""" - -from __future__ import annotations - -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 [ - event - for event in events - if event.get("type") == kind and RealtimeEventCollector.response_id(event) == 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)) - interrupt = interrupt[: PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * 4700 // 1000] - session_id = f"server-vad-hard-interrupt-{id(args)}" - client = RealtimeDuplexClient(build_realtime_url(args.url, args.model, autostart=False, session_id=session_id)) - 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 wait_for( - lambda: any(e.get("type") == "response.created" for e in client.events.events), - timeout_s=args.timeout_s, - label="response.created", - ) - target = next(e for e in client.events.events if e.get("type") == "response.created") - target_id = RealtimeEventCollector.response_id(target) - assert target_id - await wait_for( - lambda: _events(client.events.events, target_id, "response.audio.delta"), - timeout_s=args.timeout_s, - label="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 wait_for( - lambda: _events(client.events.events, target_id, "response.done"), - timeout_s=args.timeout_s, - label="cancelled response.done", - ) - await client.commit() - await wait_for( - lambda: any( - e.get("type") == "response.done" and RealtimeEventCollector.response_id(e) != target_id - for e in client.events.events[cursor:] - ), - timeout_s=args.timeout_s, - label="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") - done_index = events.index(done[0]) if len(done) == 1 else len(events) - stale = ( - any( - e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) == target_id - for e in events[done_index + 1 :] - ) - if len(done) == 1 - else False - ) - followup_audio = any( - e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) != target_id - for e in events[cursor:] - ) - ok = ( - len(done) == 1 - and isinstance(done[0].get("response"), dict) - and done[0]["response"].get("status") == "cancelled" - and done[0]["response"].get("status_details", {}).get("reason") == "turn_detected" - and not stale - 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 aa5320b8c..68204fe28 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 @@ -21,7 +21,6 @@ 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, ) @@ -89,22 +88,3 @@ def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) - 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, model_prefix: str) -> 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(model_prefix)), - chunk_ms=200, - timeout_s=180.0, - ) - ) - ) - assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2) diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index 070bcc1f9..3f0d98b35 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -652,7 +652,7 @@ async def test_native_realtime_protocol_preserves_input_turn_policy_hints(): @pytest.mark.asyncio -async def test_native_realtime_protocol_rejects_server_vad_without_interrupt(): +async def test_native_realtime_protocol_rejects_unimplemented_server_vad(): ws = TimedWebSocket() protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] protocol.bind_sender(ws.send_json) @@ -675,10 +675,7 @@ async def test_native_realtime_protocol_rejects_server_vad_without_interrupt(): error = ws.sent[-1] assert error["type"] == "error" assert error["error"]["code"] == "unsupported_turn_detection" - assert "interrupt_response=false" in error["error"]["message"] - assert "greater than 0.15" in str( - protocol._validate_realtime_turn_detection({"turn_detection": {"type": "server_vad", "threshold": 0.15}}) - ) + assert "server_vad" in error["error"]["message"] @pytest.mark.asyncio @@ -702,7 +699,7 @@ async def test_native_realtime_protocol_accepts_disabled_turn_detection(): @pytest.mark.asyncio -async def test_native_realtime_protocol_prefers_nested_vad_over_top_level(): +async def test_native_realtime_protocol_rejects_nested_vad_even_when_top_level_is_disabled(): ws = TimedWebSocket() protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] protocol.bind_sender(ws.send_json) @@ -722,18 +719,8 @@ async def test_native_realtime_protocol_prefers_nested_vad_over_top_level(): } ) - assert translated["session"]["extra_body"]["realtime_session_payload"]["overlap_policy"] == "barge_in_on_speech" - - -@pytest.mark.asyncio -async def test_realtime_vad_candidate_waits_for_session_ack(): - protocol = NativeRealtimeSessionProtocol(TimedWebSocket()) # type: ignore[arg-type] - event = {"type": "session.update", "model": "test-model", "turn_detection": {"type": "server_vad"}} - - for outbound, enabled in (({"type": "error"}, False), ({"type": "session.updated", "session": {}}, True)): - await protocol._to_duplex_event(event) - protocol.encode_outbound_event(outbound) - assert (protocol._server_vad is not None) is enabled + assert translated is None + assert ws.sent[-1]["error"]["code"] == "unsupported_turn_detection" def test_native_duplex_handler_has_no_fixed_session_admission_cap(): @@ -5439,33 +5426,6 @@ async def test_minicpmo_native_duplex_open_unsupported_fails_session_create(): assert ws.sent[0]["runtime_control"]["unsupported_count"] == 1 -@pytest.mark.asyncio -async def test_native_runtime_rejects_server_vad_without_barge_in(monkeypatch: pytest.MonkeyPatch): - engine = FakeEngineClient() - handler = OmniDuplexSessionHandler( - chat_service=FakeChatService(engine), - config_timeout_s=0.1, - idle_timeout_s=1, - ) - capabilities = DuplexCapabilities.minicpmo45_native() - capabilities.supports_barge_in = False - monkeypatch.setattr( - handler._serving_runtime_adapter, - "capabilities", - lambda *, max_sessions: capabilities, - ) - event = _native_session_create("sid-native-no-barge-in") - event["session"]["overlap_policy"] = "barge_in_on_speech" - ws = TimedWebSocket() - ws.put(event) - - await handler.handle_session(ws) - - assert ws.sent_types() == ["error"] - assert ws.sent[0]["code"] == "unsupported_turn_detection" - assert engine.opened == [] - - @pytest.mark.asyncio async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract(): class LegacyEngineClient(FakeEngineClient): diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_input.py b/vllm_omni/experimental/fullduplex/openai/realtime_input.py index 3e55d44e1..e3fe7f3ec 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_input.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_input.py @@ -15,13 +15,6 @@ 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: @@ -96,7 +89,6 @@ 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: @@ -283,62 +275,16 @@ class RealtimeInputTranslator: ) ) return None - vad_result: StreamingVADResult | None = None - if isinstance(self._server_vad, SileroStreamingVAD): - try: - vad_result = self._server_vad.process_base64( - audio, - fmt=fmt, - sample_rate_hz=sample_rate_hz, - ) - except ServerVADUnavailableError as exc: - await self._send_realtime_payload( - self._realtime_error_payload( - "server_vad_unavailable", - str(exc), - event_id=event.get("event_id"), - param="turn_detection", - ) - ) - return None - except ValueError as exc: - await self._send_realtime_payload( - self._realtime_error_payload( - "bad_audio", - str(exc), - event_id=event.get("event_id"), - param="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) - ) + looks_like_speech = 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 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 looks_like_speech: + await self._emit_input_speech_started(event) self._remember_input_transcript_hint(event) - 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 - await self._emit_input_speech_stopped( - stopped_event, - item_id=self._active_input_item_id or f"item_{uuid4().hex}", - ) payload = { "type": "input_audio_buffer.append", "audio": audio, @@ -360,20 +306,14 @@ 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) - payload["is_speech"] = looks_like_speech - if vad_result is not None: - payload["vad"] = vad_result.as_hint() - if self._server_vad_interrupts_response() and (vad_result.speech_active or vad_result.speech_started): - # Hold the MiniCPM stream in listen mode for the remainder - # of this utterance after the one-shot onset cancellation. - payload["force_listen"] = True + if not looks_like_speech: + payload["is_speech"] = False 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), @@ -395,7 +335,6 @@ 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), @@ -411,7 +350,6 @@ 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} @@ -1053,8 +991,8 @@ class RealtimeInputTranslator: return "video_frames entries must be JPEG or PNG images" return None - @classmethod - def _validate_realtime_turn_detection(cls, session_payload: dict[str, object]) -> str | 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"])) @@ -1066,152 +1004,13 @@ class RealtimeInputTranslator: for field_path, turn_detection in configured_values: if turn_detection is None: continue - if not isinstance(turn_detection, dict): - return f"{field_path} must be null or an object with type='server_vad'" - turn_detection_type = turn_detection.get("type") - if turn_detection_type != "server_vad": - return ( - f"{field_path}={turn_detection_type!r} is unsupported; " - "use null for model-owned listen/speak or type='server_vad' for hard barge-in" - ) - threshold = turn_detection.get("threshold", 0.5) - if ( - not isinstance(threshold, int | float) - or isinstance(threshold, bool) - or not np.isfinite(float(threshold)) - or not SILERO_VAD_MIN_THRESHOLD < float(threshold) <= 1.0 - ): - return f"{field_path}.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 isinstance(value, int | float) - or isinstance(value, bool) - or not np.isfinite(float(value)) - or float(value) < 0 - ): - return f"{field_path}.{name} must be a non-negative number" - interrupt_response = turn_detection.get("interrupt_response", True) - if not isinstance(interrupt_response, bool): - return f"{field_path}.interrupt_response must be a boolean" - if interrupt_response is not True: - return ( - f"{field_path}=server_vad with interrupt_response=false is unsupported; " - "use turn_detection=null for model-owned listen/speak" - ) - - present, turn_detection = cls._configured_realtime_turn_detection(session_payload) - if not present: - 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 - desired_policy = ( - "barge_in_on_speech" - if isinstance(turn_detection, dict) and turn_detection.get("interrupt_response", True) is True - 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}" + 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" + ) return None - @staticmethod - def _configured_realtime_turn_detection( - session_payload: dict[str, object], - ) -> tuple[bool, dict[str, object] | None]: - present = "turn_detection" in session_payload - 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: - present = True - selected = audio_input.get("turn_detection") - return present, dict(selected) if isinstance(selected, dict) else None - - def _prepare_realtime_turn_detection_update(self, session_payload: dict[str, object]) -> None: - """Stage a VAD replacement until the Session/runtime update commits.""" - present, turn_detection = self._configured_realtime_turn_detection(session_payload) - if not present: - return - if self._turn_detection_update_pending: - raise RuntimeError("A Realtime turn-detection update is already pending") - - pending_turn_detection: dict[str, object] | None = None - pending_server_vad: SileroStreamingVAD | None = None - if turn_detection is None: - session_payload["overlap_policy"] = "listen_only" - else: - normalized = dict(turn_detection) - normalized.setdefault("type", "server_vad") - normalized.setdefault("interrupt_response", True) - normalized.setdefault("threshold", 0.5) - normalized.setdefault("prefix_padding_ms", 300) - normalized.setdefault("silence_duration_ms", 500) - normalized.setdefault("min_speech_duration_ms", 96) - pending_turn_detection = normalized - pending_server_vad = SileroStreamingVAD(SileroVADConfig.from_turn_detection(normalized)) - session_payload["turn_detection"] = dict(normalized) - session_payload["overlap_policy"] = "barge_in_on_speech" - - self._pending_turn_detection = pending_turn_detection - self._pending_server_vad = pending_server_vad - self._turn_detection_update_pending = True - self._turn_detection_update_resolved = asyncio.Event() - - async def wait_for_realtime_turn_detection_update(self) -> None: - """Preserve wire order while the runner accepts or rejects a candidate.""" - resolved = self._turn_detection_update_resolved - if self._turn_detection_update_pending and resolved is not None: - await resolved.wait() - - def commit_realtime_turn_detection_update(self) -> None: - """Publish the staged VAD only after the Session/runtime transaction ACKs.""" - if not self._turn_detection_update_pending: - return - previous_server_vad = self._server_vad - self._turn_detection = ( - dict(self._pending_turn_detection) if isinstance(self._pending_turn_detection, dict) else None - ) - self._server_vad = self._pending_server_vad - self._clear_pending_realtime_turn_detection_update() - reset = getattr(previous_server_vad, "reset", None) - if callable(reset): - reset() - - def reject_realtime_turn_detection_update(self) -> None: - """Discard a staged VAD after any rejected Session/runtime update.""" - if not self._turn_detection_update_pending: - return - reset = getattr(self._pending_server_vad, "reset", None) - if callable(reset): - reset() - self._clear_pending_realtime_turn_detection_update() - - def _clear_pending_realtime_turn_detection_update(self) -> None: - resolved = self._turn_detection_update_resolved - self._turn_detection_update_pending = False - self._pending_turn_detection = None - self._pending_server_vad = None - self._turn_detection_update_resolved = None - if resolved is not None: - resolved.set() - - def _server_vad_interrupts_response(self) -> bool: - return isinstance(self._turn_detection, dict) and self._turn_detection.get("interrupt_response", True) is True - - 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_session.py b/vllm_omni/experimental/fullduplex/openai/realtime_session.py index 5b64b9fff..519c071cf 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_session.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_session.py @@ -58,7 +58,6 @@ 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) @@ -98,22 +97,11 @@ class NativeRealtimeSessionProtocol( return json.dumps(translated) def encode_outbound_event(self, data: dict[str, Any]) -> list[dict[str, object]]: - self._resolve_realtime_turn_detection_update(data) payloads = self._from_duplex_event(data) for payload in payloads: self._attach_event_id(payload) return payloads - def _resolve_realtime_turn_detection_update(self, data: dict[str, Any]) -> None: - """Publish or discard a staged VAD with the Realtime session ACK.""" - if not self._turn_detection_update_pending: - return - event_type = data.get("type") - if event_type in {"session.created", "session.updated"}: - self.commit_realtime_turn_detection_update() - elif event_type == "error": - self.reject_realtime_turn_detection_update() - @staticmethod def _attach_event_id(payload: dict[str, object]) -> None: payload.setdefault("event_id", f"event_{uuid4().hex}") diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_state.py b/vllm_omni/experimental/fullduplex/openai/realtime_state.py index 880a67be6..ac3b2a8de 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_state.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_state.py @@ -57,7 +57,6 @@ 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", } @@ -106,12 +105,6 @@ 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 - _turn_detection_update_pending: bool = False - _pending_turn_detection: dict[str, object] | None = None - _pending_server_vad: object | None = None - _turn_detection_update_resolved: asyncio.Event | None = None _send_realtime_json: Any = None _initial_session_update: bool = False _input_speech_started: bool = False @@ -187,12 +180,6 @@ 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() - _turn_detection_update_pending: bool = _RealtimeStateField() - _pending_turn_detection: dict[str, object] | None = _RealtimeStateField() - _pending_server_vad: object | None = _RealtimeStateField() - _turn_detection_update_resolved: 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 6b65829dd..15a804f86 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/vllm_omni/experimental/fullduplex/openai/serving.py @@ -494,26 +494,6 @@ 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 - ): - # The configured server overlap detector classified this chunk - # as speech while the assistant is still generating/playing. - # Fence the active response so the same interrupting chunk is - # retained as the start of the next turn. - 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", @@ -573,22 +553,9 @@ 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": ("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", + "reason": "policy_barge_in_on_speech", "duration_ms": duration_ms, "overlap_speech_ms": session.overlap_speech_ms, "buffer_audio": True, @@ -632,18 +599,6 @@ class OmniDuplexSessionHandler( "defer_runtime_append": True, } - @staticmethod - def _vad_speech_started( - event: Mapping[str, object], - payload: Mapping[str, object], - ) -> bool | None: - """Return an onset edge, or None for legacy non-server-VAD input.""" - 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: @@ -961,20 +916,6 @@ class OmniDuplexSessionHandler( max_sessions=self._duplex_session_config.max_sessions, ) ) - if ( - config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value - and not session.capabilities.supports_barge_in - ): - if realtime_protocol is not None: - realtime_protocol.reject_realtime_turn_detection_update() - await send_json( - { - "type": "error", - "error": ("server_vad with interrupt_response=true requires a runtime that supports barge-in"), - "code": "unsupported_turn_detection", - } - ) - return None session.replace_runtime_config(runtime_config) return _DuplexSessionHandshake(session=session) diff --git a/vllm_omni/experimental/fullduplex/openai/vad.py b/vllm_omni/experimental/fullduplex/openai/vad.py deleted file mode 100644 index 76b6ecef3..000000000 --- a/vllm_omni/experimental/fullduplex/openai/vad.py +++ /dev/null @@ -1,218 +0,0 @@ -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): - """Raised when the optional server VAD runtime is not installed.""" - - -@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 - - @classmethod - def from_turn_detection(cls, turn_detection: dict[str, object]) -> SileroVADConfig: - defaults = cls() - - def number(name: str, default: int | float) -> int | float: - value = turn_detection.get(name, default) - return value if isinstance(value, int | float) and not isinstance(value, bool) else default - - return cls( - threshold=float(number("threshold", defaults.threshold)), - prefix_padding_ms=max(0, int(number("prefix_padding_ms", defaults.prefix_padding_ms))), - silence_duration_ms=max(0, int(number("silence_duration_ms", defaults.silence_duration_ms))), - min_speech_duration_ms=max( - 32, - int(number("min_speech_duration_ms", defaults.min_speech_duration_ms)), - ), - ) - - -@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 - - def as_hint(self) -> dict[str, object]: - hint: dict[str, object] = { - "backend": "silero", - "is_speech": self.is_speech, - "speech_active": self.speech_active, - "speech_started": self.speech_started, - "speech_stopped": self.speech_stopped, - "speech_probability": self.speech_probability, - } - if self.speech_start_ms is not None: - hint["speech_start_ms"] = self.speech_start_ms - if self.speech_end_ms is not None: - hint["speech_end_ms"] = self.speech_end_ms - return hint - - -class SileroStreamingVAD: - """Per-session streaming Silero VAD with onset debounce and utterance latch. - - Silero's model is stateful, so an instance must never be shared across - Realtime sessions. Audio is normalized to 16 kHz and scored in the model's - 512-sample (32 ms) windows. Loading is lazy so deployments which use the - default model-owned mode do not need the optional ``silero-vad`` package. - """ - - _SAMPLE_RATE_HZ = 16_000 - _WINDOW_SAMPLES = 512 - _NEGATIVE_THRESHOLD_GAP = SILERO_VAD_MIN_THRESHOLD - - 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._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 - - def reset(self) -> None: - self._pending = np.empty(0, dtype=np.float32) - self._speech_active = False - self._candidate_samples = 0 - self._candidate_start_sample = None - self._silence_samples = 0 - self._processed_samples = 0 - 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=" 1: - target_size = max(1, int(round(samples.size * self._SAMPLE_RATE_HZ / rate))) - source_x = np.linspace(0.0, 1.0, num=samples.size, endpoint=True) - target_x = np.linspace(0.0, 1.0, num=target_size, endpoint=True) - samples = np.interp(target_x, source_x, samples).astype(np.float32) - return self.process(samples) - - def process(self, samples: np.ndarray) -> 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 = int(self.config.min_speech_duration_ms * self._SAMPLE_RATE_HZ / 1000) - min_silence_samples = int(self.config.silence_duration_ms * self._SAMPLE_RATE_HZ / 1000) - negative_threshold = max(0.0, self.config.threshold - self._NEGATIVE_THRESHOLD_GAP) - - 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 * 1000 / self._SAMPLE_RATE_HZ))) - 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 = int(self.config.prefix_padding_ms * self._SAMPLE_RATE_HZ / 1000) - self._speech_active = True - self._candidate_samples = 0 - self._candidate_start_sample = None - self._silence_samples = 0 - contained_speech = True - started = True - start_ms = max( - 0, - int(round((candidate_start - prefix_samples) * 1000 / self._SAMPLE_RATE_HZ)), - ) - else: - self._candidate_samples = 0 - self._candidate_start_sample = None - - return StreamingVADResult( - is_speech=contained_speech, - speech_active=self._speech_active, - speech_started=started, - speech_stopped=stopped, - speech_probability=max_probability, - speech_start_ms=start_ms, - speech_end_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))