diff --git a/.env.example b/.env.example index 3673b1b..50b94cd 100644 --- a/.env.example +++ b/.env.example @@ -100,6 +100,7 @@ AI_VOICE_TTS_CACHE_ENABLED=1 AI_VOICE_TTS_CACHE_DIR=.data/ai_voice_tts_cache AI_VOICE_VAD_MIN_SPEECH_MS=300 AI_VOICE_VAD_TRAILING_SILENCE_MS=400 +AI_VOICE_VAD_RMS_THRESHOLD=250 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 AI_VOICE_MAX_CONTEXT_SEGMENTS=12 diff --git a/.env.production.template b/.env.production.template index ad4be9e..61bc0cc 100644 --- a/.env.production.template +++ b/.env.production.template @@ -96,6 +96,7 @@ AI_VOICE_TTS_CACHE_ENABLED=1 AI_VOICE_TTS_CACHE_DIR=/app/.data_local/ai_voice_tts_cache AI_VOICE_VAD_MIN_SPEECH_MS=300 AI_VOICE_VAD_TRAILING_SILENCE_MS=400 +AI_VOICE_VAD_RMS_THRESHOLD=250 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 AI_VOICE_MAX_CONTEXT_SEGMENTS=12 diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index 97dc547..ed2eecf 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -123,6 +123,10 @@ def _turn_max_ms() -> int: return max(_int_env("AI_VOICE_TURN_MAX_MS", 10000), _vad_frame_ms()) +def _vad_rms_threshold() -> int: + return max(_int_env("AI_VOICE_VAD_RMS_THRESHOLD", 250), 1) + + def _media_idle_timeout_seconds() -> float: return max(_float_env("AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS", 15.0), 5.0) @@ -1227,6 +1231,7 @@ _MEDIA_RUNTIME = AudioSocketMediaRuntime( min_speech_ms=_vad_min_speech_ms(), trailing_silence_ms=_vad_trailing_silence_ms(), max_turn_ms=_turn_max_ms(), + vad_rms_threshold=_vad_rms_threshold(), asr_provider=_ASR_PROVIDER, streaming_asr_provider=_STREAMING_ASR_PROVIDER, tts_provider=_TTS_PROVIDER, diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 9bf79b5..6934062 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -95,9 +95,12 @@ class MediaActor: utterance_generation: int = 0 finalized_utterance_generation: int = 0 partial_asr_task: asyncio.Task | None = None + streaming_asr_push_task: asyncio.Task | None = None + streaming_asr_push_queue: asyncio.Queue[bytes | None] | None = None partial_asr_attempted: bool = False asr_stream_id: str | None = None asr_streaming_enabled: bool = False + asr_streaming_failed: bool = False asr_poll_due_monotonic: float = 0.0 streaming_asr_backoff_until_monotonic: float = 0.0 barge_in_speech_ms: int = 0 @@ -126,6 +129,7 @@ class AudioSocketMediaRuntime: min_speech_ms: int, trailing_silence_ms: int, max_turn_ms: int, + vad_rms_threshold: int = 250, asr_provider: ASRProvider, streaming_asr_provider: StreamingASRProvider | None = None, tts_provider: TTSProvider, @@ -155,6 +159,7 @@ class AudioSocketMediaRuntime: self._min_speech_ms = max(min_speech_ms, self._frame_ms) self._trailing_silence_ms = max(trailing_silence_ms, self._frame_ms) self._max_turn_ms = max(max_turn_ms, self._frame_ms) + self._vad_rms_threshold = max(int(vad_rms_threshold), 1) self._asr_provider = asr_provider self._streaming_asr_provider = streaming_asr_provider or StreamingASRProvider() self._tts_provider = tts_provider @@ -184,6 +189,9 @@ class AudioSocketMediaRuntime: self._stable_partial_hold_seconds = 0.40 self._barge_in_trigger_ms = 220 self._streaming_asr_reopen_backoff_seconds = 2.0 + self._streaming_asr_push_queue_max_frames = 250 + self._streaming_asr_push_batch_max_bytes = self._frame_bytes * 8 + self._streaming_asr_push_drain_timeout_seconds = 1.5 self._thinking_continuation_grace_seconds = 0.45 self._thinking_continuation_max_bytes = int(1800 * 16) @@ -377,6 +385,7 @@ class AudioSocketMediaRuntime: actor.stable_partial_intent = None actor.response_plan_id = None actor.partial_asr_attempted = False + actor.asr_streaming_failed = False actor.asr_poll_due_monotonic = 0.0 actor.speech_started_monotonic = time.monotonic() actor.speech_ended_monotonic = 0.0 @@ -423,21 +432,134 @@ class AudioSocketMediaRuntime: return actor.asr_stream_id = stream_id actor.asr_streaming_enabled = True + actor.asr_streaming_failed = False actor.asr_poll_due_monotonic = 0.0 + actor.streaming_asr_push_queue = asyncio.Queue(maxsize=self._streaming_asr_push_queue_max_frames) + actor.streaming_asr_push_task = asyncio.create_task( + self._streaming_asr_push_loop(actor, stream_id), + name=f"streaming-asr-push-{actor.registration.voice_session_id}", + ) def _mark_streaming_asr_backoff(self, actor: MediaActor) -> None: actor.streaming_asr_backoff_until_monotonic = ( time.monotonic() + self._streaming_asr_reopen_backoff_seconds ) - async def _close_streaming_asr(self, actor: MediaActor) -> None: + async def _close_streaming_asr(self, actor: MediaActor, *, drain: bool = True) -> None: stream_id = actor.asr_stream_id actor.asr_stream_id = None actor.asr_streaming_enabled = False + await self._stop_streaming_asr_push_loop(actor, drain=drain) if not stream_id: return await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id) + async def _streaming_asr_push_loop(self, actor: MediaActor, stream_id: str) -> None: + queue = actor.streaming_asr_push_queue + if queue is None: + return + while True: + item = await queue.get() + if item is None: + queue.task_done() + return + + batch = bytearray(item) + task_done_count = 1 + stop_after_batch = False + while len(batch) < self._streaming_asr_push_batch_max_bytes: + try: + extra = queue.get_nowait() + except asyncio.QueueEmpty: + break + if extra is None: + stop_after_batch = True + queue.task_done() + break + batch.extend(extra) + task_done_count += 1 + + try: + await asyncio.to_thread( + self._streaming_asr_provider.push_pcm, + stream_id, + bytes(batch), + ) + with contextlib.suppress(StreamingASRUnavailable): + await self._poll_streaming_partial(actor) + except StreamingASRUnavailable as exc: + actor.asr_streaming_failed = True + actor.asr_streaming_enabled = False + self._mark_streaming_asr_backoff(actor) + logger.warning( + "audiosocket.streaming_asr_push_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + return + finally: + for _ in range(task_done_count): + queue.task_done() + if stop_after_batch: + return + + async def _stop_streaming_asr_push_loop(self, actor: MediaActor, *, drain: bool) -> None: + task = actor.streaming_asr_push_task + queue = actor.streaming_asr_push_queue + actor.streaming_asr_push_task = None + actor.streaming_asr_push_queue = None + if task is None: + return + if task.done(): + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + return + if queue is not None: + if drain: + try: + await asyncio.wait_for( + queue.join(), + timeout=self._streaming_asr_push_drain_timeout_seconds, + ) + except asyncio.TimeoutError: + actor.asr_streaming_failed = True + actor.asr_streaming_enabled = False + self._mark_streaming_asr_backoff(actor) + logger.warning( + "audiosocket.streaming_asr_push_drain_timeout session_id=%s queued_frames=%s", + actor.registration.voice_session_id, + queue.qsize(), + ) + sentinel_enqueued = False + with contextlib.suppress(asyncio.QueueFull): + queue.put_nowait(None) + sentinel_enqueued = True + if not sentinel_enqueued: + task.cancel() + try: + await asyncio.wait_for(task, timeout=0.5) + except asyncio.TimeoutError: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + + def _queue_streaming_asr_pcm(self, actor: MediaActor, pcm_frame: bytes) -> None: + queue = actor.streaming_asr_push_queue + if queue is None or actor.asr_streaming_failed: + return + try: + queue.put_nowait(pcm_frame) + except asyncio.QueueFull: + actor.asr_streaming_failed = True + actor.asr_streaming_enabled = False + self._mark_streaming_asr_backoff(actor) + logger.warning( + "audiosocket.streaming_asr_push_queue_full session_id=%s max_frames=%s", + actor.registration.voice_session_id, + self._streaming_asr_push_queue_max_frames, + ) + asyncio.create_task(self._close_streaming_asr(actor, drain=False)) + def _buffer_thinking_continuation(self, actor: MediaActor, pcm_frame: bytes, *, is_speech: bool, now: float) -> None: if not is_speech and not actor.thinking_continuation_pcm: return @@ -500,6 +622,9 @@ class AudioSocketMediaRuntime: if not actor.asr_streaming_enabled or not actor.asr_stream_id: raise StreamingASRUnavailable("Streaming ASR stream is not active") stream_id = actor.asr_stream_id + await self._stop_streaming_asr_push_loop(actor, drain=True) + if actor.asr_streaming_failed: + raise StreamingASRUnavailable("Streaming ASR push failed before finalize") try: return await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id) finally: @@ -891,6 +1016,7 @@ class AudioSocketMediaRuntime: min_speech_ms=self._min_speech_ms, trailing_silence_ms=self._trailing_silence_ms, max_turn_ms=self._max_turn_ms, + rms_threshold=self._vad_rms_threshold, ), frame_ms=self._frame_ms, frame_bytes=self._frame_bytes, @@ -965,7 +1091,7 @@ class AudioSocketMediaRuntime: if actor.state not in {"listening", "speaking", "thinking"}: return - is_speech = audioop.rms(pcm_frame, 2) >= 250 + is_speech = audioop.rms(pcm_frame, 2) >= self._vad_rms_threshold if actor.state == "thinking": self._buffer_thinking_continuation(actor, pcm_frame, is_speech=is_speech, now=now) return @@ -988,43 +1114,25 @@ class AudioSocketMediaRuntime: vad_result = actor.vad.feed(pcm_frame) if vad_result.speech_started: self._reset_live_turn_state(actor) + logger.info( + "audiosocket.speech_started session_id=%s", + actor.registration.voice_session_id, + ) await self._ensure_streaming_asr(actor) stream_input_active = actor.input_active or actor.barge_in_pending if actor.asr_streaming_enabled and actor.asr_stream_id and stream_input_active: - try: - await asyncio.to_thread( - self._streaming_asr_provider.push_pcm, - actor.asr_stream_id, - pcm_frame, - ) - except StreamingASRUnavailable as exc: - logger.warning( - "audiosocket.streaming_asr_push_failed session_id=%s error=%s", - actor.registration.voice_session_id, - str(exc)[:500], - ) - self._mark_streaming_asr_backoff(actor) - await self._close_streaming_asr(actor) - else: - try: - await self._poll_streaming_partial(actor) - except StreamingASRUnavailable as exc: - actor.asr_poll_due_monotonic = time.monotonic() + max( - self._partial_poll_interval_seconds, - 0.60, - ) - logger.warning( - "audiosocket.streaming_asr_partial_poll_failed session_id=%s error=%s", - actor.registration.voice_session_id, - str(exc)[:500], - ) - self._mark_streaming_asr_backoff(actor) - await self._close_streaming_asr(actor) + self._queue_streaming_asr_pcm(actor, pcm_frame) elif actor.registration.voice_v2_partial_asr: self._maybe_schedule_partial_asr(actor) if vad_result.utterance_pcm: actor.input_active = False actor.speech_ended_monotonic = time.monotonic() + logger.info( + "audiosocket.utterance_finalized session_id=%s duration_ms=%s barge_in=%s", + actor.registration.voice_session_id, + int(len(vad_result.utterance_pcm) / 16), + actor.barge_in_pending, + ) await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending)) actor.barge_in_pending = False @@ -1101,8 +1209,23 @@ class AudioSocketMediaRuntime: ack_source="immediate_turn_close", ) - if actor.asr_streaming_enabled: - transcription = await self._finalize_streaming_transcription(actor) + if actor.asr_streaming_enabled and not actor.asr_streaming_failed: + try: + transcription = await self._finalize_streaming_transcription(actor) + except StreamingASRUnavailable as exc: + logger.warning( + "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + self._mark_streaming_asr_backoff(actor) + await self._close_streaming_asr(actor, drain=False) + wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) + transcription = await asyncio.to_thread( + self._asr_provider.transcribe, + wav_bytes, + language_hint=actor.registration.language, + ) else: wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) transcription = await asyncio.to_thread( @@ -1455,7 +1578,7 @@ class AudioSocketMediaRuntime: with contextlib.suppress(asyncio.CancelledError, Exception): await actor.partial_asr_task with contextlib.suppress(Exception): - await self._close_streaming_asr(actor) + await self._close_streaming_asr(actor, drain=False) if actor.worker_task is not None: actor.worker_task.cancel() with contextlib.suppress(asyncio.CancelledError, Exception): diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index 0c8029b..b9cf100 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -1504,6 +1504,10 @@ def test_media_runtime_streaming_timeout_enters_backoff_before_reopen(): speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 await runtime._handle_pcm(actor, speech_frame) await runtime._handle_pcm(actor, speech_frame) + for _ in range(20): + if not actor.asr_streaming_enabled: + break + await asyncio.sleep(0.01) assert actor.asr_streaming_enabled is False assert actor.streaming_asr_backoff_until_monotonic > time.monotonic() await runtime._ensure_streaming_asr(actor) @@ -1512,6 +1516,109 @@ def test_media_runtime_streaming_timeout_enters_backoff_before_reopen(): asyncio.run(_scenario()) +def test_media_runtime_streaming_sidecar_push_does_not_block_vad_finalization(): + class _SlowStreamingProvider(StreamingASRProvider): + name = "slow-streaming" + supports_streaming = True + + def __init__(self) -> None: + self.push_count = 0 + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + del session_id, language_hint + return "stream-1" + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + del stream_id, pcm_8k_chunk + self.push_count += 1 + time.sleep(0.25) + + registration = MediaRegistration( + voice_session_id="avs_media_runtime_nonblocking_asr", + call_id="call_media_runtime_nonblocking_asr", + interaction_id="int_media_runtime_nonblocking_asr", + ai_session_id="ais_media_runtime_nonblocking_asr", + language="ru", + media_uuid=str(uuid.uuid4()), + queue_code="voice_lab_ai", + queue_id="que_voice_lab_ai", + agent_profile="voice_support", + voice_v2_enabled=True, + voice_v2_ack_mode="immediate_short", + voice_v2_streaming_tts=True, + voice_v2_partial_asr=True, + voice_v2_duplex=True, + voice_v2_streaming_asr_backend="local_sidecar", + ) + streaming_provider = _SlowStreamingProvider() + runtime = AudioSocketMediaRuntime( + enabled=True, + host="127.0.0.1", + port=0, + frame_ms=20, + idle_timeout_seconds=2.0, + registration_wait_timeout_seconds=0.5, + min_speech_ms=40, + trailing_silence_ms=40, + max_turn_ms=2000, + asr_provider=_StubASRProvider(), + streaming_asr_provider=streaming_provider, + tts_provider=_StubTTSProvider(), + load_registration_by_media_uuid=lambda value: None, + mark_media_connected=lambda session_id, value: None, + mark_media_ended=lambda session_id, reason: None, + touch_media_frame=lambda session_id: None, + set_state=lambda session_id, state, handoff_reason, metadata: None, + get_pending_greeting=lambda session_id: None, + mark_reply_delivered=lambda session_id, text, is_greeting: None, + plan_reply=lambda session_id, text, metadata, kind: None, + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut( + language=language or "ru", + intent="clarification", + reply_text="reply", + confidence=0.8, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="reply ready", + model="stub-voice", + latency_ms=1, + status="active", + ), + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: None, + ) + + async def _scenario() -> float: + actor = MediaActor( + registration=registration, + reader=asyncio.StreamReader(), + writer=None, # type: ignore[arg-type] + vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000), + frame_ms=20, + frame_bytes=320, + state="listening", + ) + speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 + silence_frame = b"\x00\x00" * 160 + started = time.monotonic() + await runtime._handle_pcm(actor, speech_frame) + await runtime._handle_pcm(actor, speech_frame) + await runtime._handle_pcm(actor, silence_frame) + await runtime._handle_pcm(actor, silence_frame) + elapsed = time.monotonic() - started + pcm_bytes, _ = await asyncio.wait_for(actor.turn_queue.get(), timeout=0.1) + assert pcm_bytes + await runtime._close_streaming_asr(actor, drain=False) + return elapsed + + elapsed = asyncio.run(_scenario()) + + assert elapsed < 0.15 + assert streaming_provider.push_count >= 1 + + def test_media_runtime_merges_thinking_continuation_into_current_utterance(): runtime = AudioSocketMediaRuntime( enabled=True,