From 989bf9a0c167d2eed0240a3423f0a1a250630cc1 Mon Sep 17 00:00:00 2001 From: Yera All Date: Sun, 12 Apr 2026 21:59:32 +0500 Subject: [PATCH] fix(voice): stabilize v2 streaming topic handling --- services/ai_orchestrator_service/voice.py | 30 +++++- .../ai_voice_runtime_service/media_runtime.py | 34 ++++++- services/streaming_asr_sidecar_service/app.py | 17 +++- tests/test_ai_orchestrator_service.py | 39 ++++++++ tests/test_ai_voice_media_runtime.py | 99 +++++++++++++++++++ tests/test_streaming_asr_sidecar_service.py | 43 ++++++++ 6 files changed, 254 insertions(+), 8 deletions(-) diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index d12a8a4..30005ee 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -787,7 +787,7 @@ def _voice_text_key(text: str | None) -> str: def _voice_recent_caller_texts( transcript_window: list[VoiceTranscriptSegmentRow], *, - limit: int = 4, + limit: int = 8, ) -> list[str]: caller_texts = [str(segment.text or "").strip() for segment in transcript_window if segment.speaker == "caller"] return caller_texts[-max(limit, 1) :] @@ -1256,12 +1256,24 @@ def _voice_caller_context_before_current(caller_texts: list[str], transcript_tex return caller_texts +def _voice_service_context_texts(caller_texts: list[str]) -> list[str]: + return [text for text in caller_texts if _voice_has_service_topic(text)] + + def _voice_is_midcall_greeting_reply(text: str | None) -> bool: normalized = _voice_text_key(text) if not normalized: return False greeting_markers = ("здравствуйте", "сәлеметсіз", "сәлем") - followup_markers = ("чем помочь", "как я могу помочь", "коротко расскажите", "қалай көмектесе") + followup_markers = ( + "чем помочь", + "как я могу помочь", + "коротко расскажите", + "о чем именно", + "что именно", + "что вас интересует", + "қалай көмектесе", + ) return any(marker in normalized for marker in greeting_markers) and any( marker in normalized for marker in followup_markers ) @@ -1308,15 +1320,27 @@ def _voice_postprocess_reply_text( return normalized_reply caller_texts = _voice_recent_caller_texts(transcript_window) 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 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 _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language) + 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 ): 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 ( + active_topic_prompt + and not _voice_has_service_topic(transcript_text) + and ( + "тариф" in _voice_text_key(normalized_reply) + or "услуг" in _voice_text_key(normalized_reply) + or "как я могу помочь" in _voice_text_key(normalized_reply) + ) + ): + return active_topic_prompt return normalized_reply diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 8682ddf..ba93f5d 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -107,6 +107,7 @@ class MediaActor: last_ack_completed_monotonic: float = 0.0 last_ack_variant: str | None = None current_reply_phase: str | None = None + finalized_caller_turn_count: int = 0 class AudioSocketMediaRuntime: @@ -266,6 +267,20 @@ class AudioSocketMediaRuntime: normalized = str(language or registration.language or "").strip().lower() return normalized.startswith("ru") + def _should_emit_partial_ack(self, actor: MediaActor, partial_transcript: str, partial_intent: str) -> bool: + transcript_text = str(partial_transcript or "").strip() + if not transcript_text or self._is_low_signal_transcript(transcript_text): + return False + normalized_intent = str(partial_intent or "").strip() or "unknown" + if actor.finalized_caller_turn_count > 0: + return True + return normalized_intent != "unknown" + + def _should_emit_blind_ack(self, actor: MediaActor, pcm_bytes: bytes) -> bool: + if len(pcm_bytes) < self._immediate_ack_min_bytes: + return False + return actor.finalized_caller_turn_count > 0 + @staticmethod def _base_ack_variants(language: str | None, ack_kind: str) -> tuple[str, ...]: normalized = str(language or "").strip().lower() @@ -929,7 +944,6 @@ class AudioSocketMediaRuntime: actor.asr_stream_id, pcm_frame, ) - await self._poll_streaming_partial(actor) except StreamingASRUnavailable as exc: logger.warning( "audiosocket.streaming_asr_push_failed session_id=%s error=%s", @@ -937,6 +951,19 @@ class AudioSocketMediaRuntime: str(exc)[:500], ) 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], + ) elif actor.registration.voice_v2_partial_asr: self._maybe_schedule_partial_asr(actor) if vad_result.utterance_pcm: @@ -1002,14 +1029,14 @@ class AudioSocketMediaRuntime: 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: + if self._should_emit_partial_ack(actor, partial_transcript, partial_intent): await self._emit_early_ack( actor, language=actor.registration.language, metadata=base_metadata, ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr", ) - elif len(pcm_bytes) >= self._immediate_ack_min_bytes: + elif self._should_emit_blind_ack(actor, pcm_bytes): await self._emit_early_ack( actor, language=actor.registration.language, @@ -1041,6 +1068,7 @@ class AudioSocketMediaRuntime: return actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation) + actor.finalized_caller_turn_count += 1 actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None actor.partial_intent = self._detect_early_intent(transcript_text) actor.stable_partial_intent = actor.partial_intent diff --git a/services/streaming_asr_sidecar_service/app.py b/services/streaming_asr_sidecar_service/app.py index 00d2c12..e6fe84f 100644 --- a/services/streaming_asr_sidecar_service/app.py +++ b/services/streaming_asr_sidecar_service/app.py @@ -51,13 +51,17 @@ def _partial_min_audio_ms() -> int: def _partial_recompute_interval_ms() -> int: - return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_RECOMPUTE_INTERVAL_MS", 200), 80) + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_RECOMPUTE_INTERVAL_MS", 450), 80) def _partial_stability_hold_ms() -> int: return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_STABILITY_HOLD_MS", 400), 120) +def _partial_max_audio_ms() -> int: + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_MAX_AUDIO_MS", 1800), 400) + + def _model_name() -> str: return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_MODEL", "base") or "base").strip() or "base" @@ -430,6 +434,15 @@ def _copy_stream_snapshot(stream: StreamState) -> tuple[bytes, int, int, str]: ) +def _trim_partial_snapshot(pcm_bytes: bytes, *, sample_rate_hz: int) -> bytes: + if not pcm_bytes or sample_rate_hz <= 0: + return pcm_bytes + max_bytes = int((_partial_max_audio_ms() / 1000.0) * sample_rate_hz * 2) + if max_bytes <= 0 or len(pcm_bytes) <= max_bytes: + return pcm_bytes + return pcm_bytes[-max_bytes:] + + def _transcribe_stream_snapshot(pcm_bytes: bytes, *, sample_rate_hz: int, language_hint: str) -> SidecarTranscript: return _get_engine().transcribe_pcm( pcm_bytes, @@ -540,7 +553,7 @@ def poll_partial(stream_id: str, _: None = Depends(_require_internal_actor)) -> pcm_bytes, buffer_version, sample_rate_hz, language_hint = _copy_stream_snapshot(stream) transcript = _transcribe_stream_snapshot( - pcm_bytes, + _trim_partial_snapshot(pcm_bytes, sample_rate_hz=sample_rate_hz), sample_rate_hz=sample_rate_hz, language_hint=language_hint, ) diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index 68c0735..506b179 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1497,6 +1497,45 @@ def test_voice_postprocess_reply_rewrites_false_lookup_promise(): assert "филиал" in reply_text.lower() or "адрес" in reply_text.lower() +def test_voice_postprocess_reply_rewrites_midcall_greeting_with_followup_question(): + reply_text = voice_module._voice_postprocess_reply_text( + language="ru", + transcript_text="Здравствуйте, мне надо узнать...", + transcript_window=[ + SimpleNamespace(speaker="assistant", text=voice_module._voice_greeting("ru"), sequence_no=1, source_type="voice_policy", barge_in_interrupted=False, created_at=utc_now_iso()), + SimpleNamespace(speaker="caller", text="Здравствуйте, мне надо узнать...", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + ], + reply_text="Здравствуйте, Ернур! О чем именно вы хотите узнать?", + kb_results=[], + needs_handoff=False, + ) + + normalized = reply_text.lower() + assert "здравствуйте" not in normalized + 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", + transcript_text="Сколько раз повторять тебе?", + transcript_window=[ + SimpleNamespace(speaker="caller", text="Мне надо узнать график работы.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + SimpleNamespace(speaker="caller", text="В городе Алма-Ата.", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + SimpleNamespace(speaker="caller", text="Меня интересует филиал в Ауэзовском районе.", sequence_no=3, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + SimpleNamespace(speaker="caller", text="Сколько раз повторять тебе?", sequence_no=4, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + ], + reply_text="Пожалуйста, уточните, какая услуга вас интересует, чтобы я мог подсказать тариф.", + kb_results=[], + needs_handoff=False, + ) + + normalized = reply_text.lower() + assert "тариф" not in normalized + assert "услуг" not in normalized + assert "филиал" in normalized or "адрес" in normalized + + def test_turn_voice_session_early_plan_does_not_persist_partial_turns(monkeypatch): monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex") diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index e5c750d..bae2715 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -624,6 +624,7 @@ def test_media_runtime_voice_v2_plays_short_ack_before_main_reply(): frame_ms=20, frame_bytes=320, ) + actor.finalized_caller_turn_count = 1 await runtime._process_utterance(actor, pcm_frame, False) asyncio.run(_scenario()) @@ -842,6 +843,7 @@ def test_media_runtime_voice_v2_emits_generic_ack_before_full_asr_without_partia frame_ms=20, frame_bytes=320, ) + actor.finalized_caller_turn_count = 1 await runtime._process_utterance(actor, pcm_frame * 45, False) asyncio.run(_scenario()) @@ -940,6 +942,7 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre frame_ms=20, frame_bytes=320, ) + actor.finalized_caller_turn_count = 1 await runtime._process_utterance(actor, pcm_frame * 40, False) asyncio.run(_scenario()) @@ -949,6 +952,100 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre assert speak_events[0][1] < timings["full_finished"] +def test_media_runtime_voice_v2_skips_blind_ack_on_first_turn_without_partial_signal(): + speak_events: list[tuple[str, float]] = [] + + class _FastASRProvider(ASRProvider): + name = "fast-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + time.sleep(0.08) + 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=_FastASRProvider(), + 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, + style_hints: dict[str, object] | None = None, + ) -> None: + del current_actor, is_greeting, style_hints + 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_first_turn", + call_id="call_media_runtime_v2_first_turn", + interaction_id="int_media_runtime_v2_first_turn", + ai_session_id="ais_media_runtime_v2_first_turn", + 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 * 40, False) + + asyncio.run(_scenario()) + + assert len(speak_events) == 1 + assert speak_events[0][0] == "Подскажите подробнее, пожалуйста." + + def test_media_runtime_ignores_low_signal_utterance_without_ack_or_turn(): planned: list[tuple[str, str, str, dict | None]] = [] delivered: list[tuple[str, str, bool]] = [] @@ -1120,6 +1217,7 @@ def test_media_runtime_voice_v2_inserts_small_gap_between_ack_and_main_reply(): frame_ms=20, frame_bytes=320, ) + actor.finalized_caller_turn_count = 1 await runtime._process_utterance(actor, pcm_frame * 40, False) asyncio.run(_scenario()) @@ -1226,6 +1324,7 @@ def test_media_runtime_voice_v2_emotive_ack_uses_ru_variants_and_style_hints_onl frame_ms=20, frame_bytes=320, ) + actor.finalized_caller_turn_count = 1 pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 await runtime._process_utterance(actor, pcm_frame, False) diff --git a/tests/test_streaming_asr_sidecar_service.py b/tests/test_streaming_asr_sidecar_service.py index 2b330d3..7b9bf6e 100644 --- a/tests/test_streaming_asr_sidecar_service.py +++ b/tests/test_streaming_asr_sidecar_service.py @@ -133,6 +133,49 @@ def test_streaming_asr_sidecar_stream_lifecycle(monkeypatch): assert len(engine.calls) >= 3 +def test_streaming_asr_sidecar_partial_uses_recent_audio_tail(monkeypatch): + engine = _reset_sidecar(monkeypatch) + monkeypatch.setattr(sidecar_module, "_partial_max_audio_ms", lambda: 1800) + client = TestClient(sidecar_module.app) + + open_response = client.post( + "/internal/asr/streams", + headers=_admin_headers(), + json={ + "session_id": "avs_sidecar_tail", + "language_hint": "ru", + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + assert open_response.status_code == 200 + stream_id = open_response.json()["stream_id"] + + long_chunk = _pcm_chunk(3200) + push_response = client.post( + f"/internal/asr/streams/{stream_id}/chunks", + headers=_admin_headers(), + json={ + "pcm_b64": base64.b64encode(long_chunk).decode("ascii"), + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + assert push_response.status_code == 200 + + partial_response = client.get( + f"/internal/asr/streams/{stream_id}/partial", + headers=_admin_headers(), + ) + assert partial_response.status_code == 200 + assert engine.calls + partial_len, sample_rate_hz, language_hint = engine.calls[-1] + assert sample_rate_hz == 8000 + assert language_hint == "ru" + assert partial_len < len(long_chunk) + assert partial_len <= int(1.8 * 8000 * 2) + + def test_streaming_asr_sidecar_rejects_unsupported_language(monkeypatch): _reset_sidecar(monkeypatch, supported_languages={"ru"}) client = TestClient(sidecar_module.app)