diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 0a1fc32..6d5a2fa 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -607,8 +607,12 @@ def _voice_reply_with_name(language: str, reply_text: str, name: str | None) -> short_name = _voice_short_name(name) if not short_name: return reply_text + reply_text = _voice_strip_reply_greeting_prefix(reply_text, name=short_name) prefix = _voice_disclosure_prefix(language) normalized_short = _voice_text_key(short_name) + first_words = _voice_text_key(" ".join(str(reply_text or "").split()[:8])).split() + if normalized_short in first_words: + return reply_text if reply_text.startswith(prefix): rest = reply_text[len(prefix) :].lstrip() if _voice_text_key(rest).startswith(normalized_short): @@ -1344,6 +1348,28 @@ def _voice_is_midcall_greeting_reply(text: str | None) -> bool: ) +def _voice_strip_reply_greeting_prefix(reply_text: str | None, *, name: str | None = None) -> str: + text = str(reply_text or "").strip() + if not text: + return text + greeting_pattern = ( + r"^\s*(?:здравствуйте|здравствуй|привет(?:ствую)?|" + r"добрый\s+(?:день|вечер|утро)|сәлеметсіз\s+бе|сәлем)\b[\s,!.:-]*" + ) + stripped = re.sub(greeting_pattern, "", text, count=1, flags=re.IGNORECASE).strip() + if name and stripped != text: + short_name = _voice_short_name(name) + if short_name: + stripped = re.sub( + rf"^\s*{re.escape(short_name)}\b[\s,!.:-]*", + "", + stripped, + count=1, + flags=re.IGNORECASE, + ).strip() + return stripped or text + + def _voice_contains_false_lookup_promise(text: str | None) -> bool: normalized = _voice_text_key(text) if not normalized: @@ -1388,15 +1414,16 @@ def _voice_postprocess_reply_text( prior_caller_texts = _voice_caller_context_before_current(caller_texts, transcript_text) active_topic_texts = _voice_service_context_texts(prior_caller_texts or caller_texts) active_topic_prompt = _voice_topic_prompt(language, active_topic_texts) if active_topic_texts else None + has_assistant_context = any(segment.speaker == "assistant" for segment in transcript_window) if _voice_is_hearing_check_caller_text(transcript_text): return _voice_hearing_check_reply(language, prior_caller_texts or caller_texts) if _voice_contains_false_lookup_promise(normalized_reply) and not kb_results: return active_topic_prompt or _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language) - if _voice_is_midcall_greeting_reply(normalized_reply) and any( - segment.speaker == "assistant" for segment in transcript_window - ): + if _voice_is_midcall_greeting_reply(normalized_reply) and has_assistant_context: context_texts = prior_caller_texts if _voice_is_low_signal_caller_text(transcript_text) else caller_texts return _voice_topic_prompt(language, context_texts) or _voice_generic_prompt(language) + if has_assistant_context or _voice_has_service_topic(transcript_text): + normalized_reply = _voice_strip_reply_greeting_prefix(normalized_reply) if ( active_topic_prompt and not _voice_has_service_topic(transcript_text) diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 82ee255..4b8ebb9 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -40,6 +40,16 @@ from services.shared.models import VoiceAITurnDecisionOut logger = logging.getLogger("uvicorn.error") +def _env_float(name: str, default: float) -> float: + raw = str(os.getenv(name, "")).strip() + if not raw: + return default + try: + return float(raw) + except ValueError: + return default + + @dataclass(slots=True) class MediaRegistration: voice_session_id: str @@ -204,6 +214,10 @@ class AudioSocketMediaRuntime: 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._streaming_finalize_race_grace_seconds = max( + 0.0, + _env_float("AI_VOICE_V2_STREAMING_FINALIZE_RACE_GRACE_SECONDS", 0.35), + ) self._thinking_continuation_grace_seconds = 0.45 self._thinking_continuation_max_bytes = int(1800 * 16) self._partial_first_final_enabled = ( @@ -1056,6 +1070,68 @@ class AudioSocketMediaRuntime: finally: await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id) + def _observe_late_streaming_finalize(self, actor: MediaActor, task: asyncio.Task) -> None: + def _done(done_task: asyncio.Task) -> None: + try: + done_task.result() + except asyncio.CancelledError: + return + except StreamingASRUnavailable as exc: + logger.warning( + "audiosocket.streaming_asr_late_finalize_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + self._mark_streaming_asr_backoff(actor) + except Exception as exc: + logger.warning( + "audiosocket.streaming_asr_late_finalize_error session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + + task.add_done_callback(_done) + + async def _batch_transcribe_turn( + self, + actor: MediaActor, + *, + pcm_bytes: bytes, + ) -> ASRTranscription: + wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) + return await asyncio.to_thread( + self._asr_provider.transcribe, + wav_bytes, + language_hint=actor.registration.language, + ) + + async def _finalize_transcription_after_streaming_failure( + self, + actor: MediaActor, + *, + pcm_bytes: bytes, + partial_transcript: str, + error: Exception, + ) -> tuple[ASRTranscription, str]: + logger.warning( + "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(error)[:500], + ) + self._mark_streaming_asr_backoff(actor) + partial_first_text = str(partial_transcript or actor.stable_partial_transcript or actor.partial_transcript or "").strip() + if self._partial_first_final_enabled and partial_first_text and not self._is_low_signal_transcript(partial_first_text): + return ( + ASRTranscription( + text=partial_first_text, + language=actor.registration.language, + confidence=None, + ), + "streaming_partial_after_finalize_failure", + ) + transcription = await self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes) + return transcription, "batch_fallback_after_streaming_failure" + async def _finalize_turn_transcription( self, actor: MediaActor, @@ -1066,45 +1142,91 @@ class AudioSocketMediaRuntime: ) -> tuple[ASRTranscription, str]: if detached_stream is not None and detached_stream[0]: stream_id, push_task, push_queue, streaming_failed = detached_stream - try: - transcription = await self._finalize_detached_streaming_transcription( + streaming_task = asyncio.create_task( + self._finalize_detached_streaming_transcription( actor, stream_id=stream_id, push_task=push_task, push_queue=push_queue, streaming_failed=streaming_failed, ) - return transcription, "streaming_final" - except StreamingASRUnavailable as exc: - logger.warning( - "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", - actor.registration.voice_session_id, - str(exc)[:500], + ) + if self._streaming_finalize_race_grace_seconds > 0: + done, _pending = await asyncio.wait( + {streaming_task}, + timeout=self._streaming_finalize_race_grace_seconds, ) - self._mark_streaming_asr_backoff(actor) - partial_first_text = str(partial_transcript or actor.stable_partial_transcript or actor.partial_transcript or "").strip() - if self._partial_first_final_enabled and partial_first_text and not self._is_low_signal_transcript(partial_first_text): - return ( - ASRTranscription( - text=partial_first_text, - language=actor.registration.language, - confidence=None, - ), - "streaming_partial_after_finalize_failure", + if streaming_task in done: + try: + return streaming_task.result(), "streaming_final" + except StreamingASRUnavailable as exc: + return await self._finalize_transcription_after_streaming_failure( + actor, + pcm_bytes=pcm_bytes, + partial_transcript=partial_transcript, + error=exc, + ) + partial_first_text = str(partial_transcript or actor.stable_partial_transcript or actor.partial_transcript or "").strip() + if self._partial_first_final_enabled and partial_first_text and not self._is_low_signal_transcript(partial_first_text): + self._observe_late_streaming_finalize(actor, streaming_task) + return ( + ASRTranscription( + text=partial_first_text, + language=actor.registration.language, + confidence=None, + ), + "streaming_partial_before_slow_finalize", + ) + batch_task = asyncio.create_task( + self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes), + name=f"voice-batch-asr-race-{actor.registration.voice_session_id}", + ) + done, _pending = await asyncio.wait( + {streaming_task, batch_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + if streaming_task in done: + try: + return streaming_task.result(), "streaming_final" + except StreamingASRUnavailable as exc: + logger.warning( + "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], ) - 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, - ) - return transcription, "batch_fallback_after_streaming_failure" - 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, - ) + self._mark_streaming_asr_backoff(actor) + partial_first_text = str( + partial_transcript or actor.stable_partial_transcript or actor.partial_transcript or "" + ).strip() + if ( + self._partial_first_final_enabled + and partial_first_text + and not self._is_low_signal_transcript(partial_first_text) + ): + return ( + ASRTranscription( + text=partial_first_text, + language=actor.registration.language, + confidence=None, + ), + "streaming_partial_after_finalize_failure", + ) + return await batch_task, "batch_fallback_after_streaming_failure" + try: + transcription = batch_task.result() + except Exception: + try: + return await streaming_task, "streaming_final" + except StreamingASRUnavailable as exc: + return await self._finalize_transcription_after_streaming_failure( + actor, + pcm_bytes=pcm_bytes, + partial_transcript=partial_transcript, + error=exc, + ) + self._observe_late_streaming_finalize(actor, streaming_task) + return transcription, "batch_race_before_streaming_finalize" + transcription = await self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes) return transcription, "batch" async def _record_reply_status( diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index 5011133..a650869 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1546,6 +1546,19 @@ def test_voice_postprocess_reply_rewrites_midcall_greeting_with_followup_questio assert "о чем именно" not in normalized +def test_voice_reply_with_name_strips_midcall_greeting_and_does_not_double_prefix(): + reply_text = voice_module._voice_reply_with_name( + "ru", + "Здравствуйте, Ернур! Чтобы уточнить график работы, назовите город или филиал.", + "Ернур", + ) + + normalized = reply_text.lower() + assert reply_text.startswith("Ернур, Чтобы уточнить") + assert normalized.count("ернур") == 1 + assert "здравствуйте" not in normalized + + def test_voice_postprocess_reply_reuses_active_topic_after_frustration_turn(): reply_text = voice_module._voice_postprocess_reply_text( language="ru", diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index fc7ac2c..a2938a1 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -2240,6 +2240,127 @@ def test_media_runtime_voice_v2_uses_partial_as_final_when_streaming_finalize_fa assert final_turns[0][1]["transcript_source"] == "streaming_partial_after_finalize_failure" +def test_media_runtime_races_batch_asr_when_streaming_finalize_is_slow(): + class _CountingASRProvider(ASRProvider): + name = "counting-asr" + + def __init__(self) -> None: + self.transcribe_count = 0 + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + self.transcribe_count += 1 + return ASRTranscription(text="batch schedule", language=language_hint or "ru", confidence=0.9) + + class _SlowStreamingProvider(StreamingASRProvider): + name = "slow-streaming" + supports_streaming = True + + def __init__(self) -> None: + self.finalize_started = False + self.finalize_done = False + + def finalize(self, stream_id: str) -> ASRTranscription: + assert stream_id == "stream-1" + self.finalize_started = True + time.sleep(0.35) + self.finalize_done = True + return ASRTranscription(text="streaming schedule", language="ru", confidence=0.9) + + def close_stream(self, stream_id: str) -> None: + assert stream_id == "stream-1" + + asr_provider = _CountingASRProvider() + 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=asr_provider, + 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="schedule", + 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, + ) + runtime._streaming_finalize_race_grace_seconds = 0.01 + + async def _scenario() -> tuple[ASRTranscription, str, float]: + actor = MediaActor( + registration=MediaRegistration( + voice_session_id="avs_media_runtime_asr_race", + call_id="call_media_runtime_asr_race", + interaction_id="int_media_runtime_asr_race", + ai_session_id="ais_media_runtime_asr_race", + 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", + ), + 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, + ) + pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 + started = time.monotonic() + result = await runtime._finalize_turn_transcription( + actor, + pcm_bytes=pcm_frame * 40, + partial_transcript="", + detached_stream=("stream-1", None, None, False), + ) + elapsed = time.monotonic() - started + for _ in range(50): + if streaming_provider.finalize_done: + break + await asyncio.sleep(0.01) + return result[0], result[1], elapsed + + transcription, source, elapsed = asyncio.run(_scenario()) + + assert streaming_provider.finalize_started is True + assert asr_provider.transcribe_count == 1 + assert transcription.text == "batch schedule" + assert source == "batch_race_before_streaming_finalize" + assert elapsed < 0.2 + + def test_media_runtime_merges_thinking_continuation_into_current_utterance(): runtime = AudioSocketMediaRuntime( enabled=True,