diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 8ca9611..b911f16 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -135,6 +135,7 @@ class AudioSocketMediaRuntime: self._actors: dict[str, MediaActor] = {} self._v2_ack_wait_seconds = 0.18 self._partial_asr_min_ms = 650 + self._immediate_ack_min_ms = 450 @staticmethod def _normalize_intent_text(text: str) -> str: @@ -188,6 +189,10 @@ class AudioSocketMediaRuntime: def _partial_asr_min_bytes(self) -> int: return self._partial_asr_min_ms * 16 + @property + def _immediate_ack_min_bytes(self) -> int: + return self._immediate_ack_min_ms * 16 + @staticmethod def _reset_live_turn_state(actor: MediaActor) -> None: actor.utterance_generation += 1 @@ -268,6 +273,36 @@ class AudioSocketMediaRuntime: ) ) + async def _emit_early_ack( + self, + actor: MediaActor, + *, + language: str | None, + metadata: dict[str, Any], + ack_source: str, + ) -> None: + if actor.closed or actor.early_ack_started: + return + ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown") + ack_text = self._ack_text(language or actor.registration.language, ack_kind) + actor.early_ack_started = True + await self._plan_reply_segment( + actor, + ack_text, + kind="ack", + metadata={ + **metadata, + "partial_transcript": actor.partial_transcript, + "early_intent": actor.partial_intent, + "ack_kind": ack_kind, + "phase": "ack", + "partial_ack_source": ack_source, + }, + ) + await self._speak_text(actor, ack_text, is_greeting=False) + if not actor.closed: + await self._set_actor_state(actor, "thinking") + async def _plan_reply_segment( self, actor: MediaActor, @@ -515,34 +550,39 @@ class AudioSocketMediaRuntime: ) ) ) - if self._should_use_voice_v2(actor.registration) and partial_transcript and not actor.early_ack_started: - ack_kind = self._ack_kind_for_intent(partial_intent or "unknown") - ack_text = self._ack_text(actor.registration.language, ack_kind) - actor.early_ack_started = True - base_metadata = { - "turn_duration_ms": int(len(pcm_bytes) / 16), - "media_uuid": actor.registration.media_uuid, - "queue_code": actor.registration.queue_code, - "voice_v2_enabled": actor.registration.voice_v2_enabled, - "response_plan_id": actor.response_plan_id, - "playback_generation": actor.playback_generation, - "partial_transcript": partial_transcript, - "early_intent": partial_intent, - } - await self._plan_reply_segment( - actor, - ack_text, - kind="ack", - metadata={ - **base_metadata, - "ack_kind": ack_kind, - "phase": "ack", - "partial_ack_source": "precomputed_partial_asr", - }, - ) - await self._speak_text(actor, ack_text, is_greeting=False) - if not actor.closed: - await self._set_actor_state(actor, "thinking") + base_metadata = { + "turn_duration_ms": int(len(pcm_bytes) / 16), + "media_uuid": actor.registration.media_uuid, + "queue_code": actor.registration.queue_code, + "voice_v2_enabled": actor.registration.voice_v2_enabled, + "response_plan_id": actor.response_plan_id, + "playback_generation": actor.playback_generation, + "partial_transcript": partial_transcript, + "early_intent": partial_intent, + } + if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started: + partial_task = actor.partial_asr_task + if partial_task is not None and not partial_task.done(): + with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError): + await asyncio.wait_for(asyncio.shield(partial_task), timeout=0.08) + partial_transcript = str(actor.partial_transcript or "").strip() + partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript) + base_metadata["partial_transcript"] = partial_transcript + base_metadata["early_intent"] = partial_intent + if partial_transcript: + await self._emit_early_ack( + actor, + language=actor.registration.language, + metadata=base_metadata, + ack_source="precomputed_partial_asr", + ) + elif len(pcm_bytes) >= self._immediate_ack_min_bytes: + await self._emit_early_ack( + actor, + language=actor.registration.language, + metadata=base_metadata, + ack_source="immediate_turn_close", + ) transcription = await full_asr_task transcript_text = str(transcription.text or "").strip() or partial_transcript @@ -554,12 +594,7 @@ class AudioSocketMediaRuntime: actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None actor.partial_intent = self._detect_early_intent(transcript_text) metadata = { - "turn_duration_ms": int(len(pcm_bytes) / 16), - "media_uuid": actor.registration.media_uuid, - "queue_code": actor.registration.queue_code, - "voice_v2_enabled": actor.registration.voice_v2_enabled, - "response_plan_id": actor.response_plan_id, - "playback_generation": actor.playback_generation, + **base_metadata, "partial_transcript": actor.partial_transcript, "early_intent": actor.partial_intent, } @@ -581,24 +616,12 @@ class AudioSocketMediaRuntime: decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds) except asyncio.TimeoutError: if not actor.early_ack_started: - ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown") - ack_text = self._ack_text(transcription.language or actor.registration.language, ack_kind) - actor.early_ack_started = True - await self._plan_reply_segment( + await self._emit_early_ack( actor, - ack_text, - kind="ack", - metadata={ - **metadata, - "partial_transcript": actor.partial_transcript, - "early_intent": actor.partial_intent, - "ack_kind": ack_kind, - "phase": "ack", - }, + language=transcription.language or actor.registration.language, + metadata=metadata, + ack_source="decision_timeout", ) - await self._speak_text(actor, ack_text, is_greeting=False) - if not actor.closed: - await self._set_actor_state(actor, "thinking") decision = await decision_task else: decision = await decision_task diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index ce9e7c3..4a828c9 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -722,3 +722,95 @@ def test_media_runtime_voice_v2_uses_partial_asr_to_start_ack_before_full_asr(): assert speak_events assert speak_events[0][0] == runtime._ack_text("ru", "understanding") assert speak_events[0][1] < timings["full_finished"] + + +def test_media_runtime_voice_v2_emits_generic_ack_before_full_asr_without_partial_signal(): + timings: dict[str, float] = {} + speak_events: list[tuple[str, float]] = [] + + class _SlowOnlyASRProvider(ASRProvider): + name = "slow-only-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + timings["full_started"] = time.monotonic() + time.sleep(0.35) + timings["full_finished"] = time.monotonic() + return ASRTranscription(text="hello there", language=language_hint or "ru", confidence=0.9) + + 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=_SlowOnlyASRProvider(), + 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="Подскажите подробнее, пожалуйста.", + confidence=0.9, + 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 _fake_speak_text(current_actor, text: str, *, is_greeting: bool) -> None: + del current_actor, is_greeting + speak_events.append((text, time.monotonic())) + await asyncio.sleep(0) + + runtime._speak_text = _fake_speak_text # type: ignore[method-assign] + pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 + + async def _scenario() -> None: + actor = MediaActor( + registration=MediaRegistration( + voice_session_id="avs_media_runtime_v2_generic_ack", + call_id="call_media_runtime_v2_generic_ack", + interaction_id="int_media_runtime_v2_generic_ack", + ai_session_id="ais_media_runtime_v2_generic_ack", + 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=False, + ), + 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, + ) + await runtime._process_utterance(actor, pcm_frame * 30, False) + + asyncio.run(_scenario()) + + assert speak_events + assert speak_events[0][0] == runtime._ack_text("ru", "generic") + assert speak_events[0][1] < timings["full_finished"]