From 224973d84027c308d9e29602b2966b0c768ac81a Mon Sep 17 00:00:00 2001 From: Yera All Date: Sun, 19 Apr 2026 01:30:26 +0500 Subject: [PATCH] Revert "fix(voice): race slow streaming asr finalize" This reverts commit 77df41bce068ac5367bd6e9d571cbd7db11da90f. --- services/ai_orchestrator_service/voice.py | 33 +--- .../ai_voice_runtime_service/media_runtime.py | 184 +++--------------- tests/test_ai_orchestrator_service.py | 13 -- tests/test_ai_voice_media_runtime.py | 121 ------------ 4 files changed, 34 insertions(+), 317 deletions(-) diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 6d5a2fa..0a1fc32 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -607,12 +607,8 @@ 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): @@ -1348,28 +1344,6 @@ 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: @@ -1414,16 +1388,15 @@ 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 has_assistant_context: + if _voice_is_midcall_greeting_reply(normalized_reply) and any( + segment.speaker == "assistant" for segment in transcript_window + ): 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 4b8ebb9..82ee255 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -40,16 +40,6 @@ 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 @@ -214,10 +204,6 @@ 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 = ( @@ -1070,68 +1056,6 @@ 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, @@ -1142,91 +1066,45 @@ 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 - streaming_task = asyncio.create_task( - self._finalize_detached_streaming_transcription( + try: + transcription = await self._finalize_detached_streaming_transcription( actor, stream_id=stream_id, push_task=push_task, push_queue=push_queue, streaming_failed=streaming_failed, ) - ) - if self._streaming_finalize_race_grace_seconds > 0: - done, _pending = await asyncio.wait( - {streaming_task}, - timeout=self._streaming_finalize_race_grace_seconds, + 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 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", + 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", + ) + 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, ) - 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], - ) - 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_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, + ) 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 a650869..5011133 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1546,19 +1546,6 @@ 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 a2938a1..fc7ac2c 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -2240,127 +2240,6 @@ 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,