From 39096e114fd1ec411dfb4a4183cba617d69069e1 Mon Sep 17 00:00:00 2001 From: Yera All Date: Fri, 17 Apr 2026 00:30:41 +0500 Subject: [PATCH] fix(voice): harden noisy streaming turns --- services/ai_orchestrator_service/voice.py | 33 +++++ .../ai_voice_runtime_service/media_runtime.py | 66 +++++++--- services/shared/ai_context_summary.py | 1 + tests/test_ai_context_summary.py | 24 ++++ tests/test_ai_orchestrator_service.py | 38 ++++++ tests/test_ai_voice_media_runtime.py | 120 ++++++++++++++++++ 6 files changed, 267 insertions(+), 15 deletions(-) diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 0d8b8e1..daf617c 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -930,6 +930,19 @@ def _voice_topic_prompt(language: str, caller_texts: list[str]) -> str | None: return None +def _voice_summary_slot_prompt(language: str, context_summary: str | dict[str, Any] | None) -> str | None: + summary = load_context_summary(context_summary) + intent = str(summary.get("active_intent") or "").strip() + facts = summary.get("confirmed_facts") if isinstance(summary.get("confirmed_facts"), dict) else {} + city = str(facts.get("city") or "").strip() + branch_hint = str(facts.get("branch_hint") or "").strip() + if intent == "schedule" and city and not branch_hint: + if language == "kz": + return f"{city} qalasy boiynsha qaysy filial nemese mekenjai qyzyqtyratynyn aitnyz." + return f"По городу {city} уточните, пожалуйста, филиал или адрес." + return None + + def _voice_generic_prompt(language: str) -> str: if language == "kz": return "Jyldam komektesu ushin eki ush sozben ne kerek ekenin aitnyz: jumys uaqyty, otinish statusy, tarif nemese operator." @@ -1850,6 +1863,26 @@ def _voice_decision( decision["metadata"] = v2_metadata return decision + summary_slot_prompt = _voice_summary_slot_prompt(language, context_summary) + if not kb_results and summary_slot_prompt: + decision = { + "language": language, + "intent": "clarification", + "reply_text": summary_slot_prompt, + "confidence": 0.72, + "needs_handoff": False, + "handoff_reason": None, + "case_action": "keep_open", + "kb_refs": [], + "summary_text": "AI продолжил активный сценарий из conversation summary и уточнил недостающий слот.", + "model": "voice_policy_context_summary", + "latency_ms": 1, + } + if v2_metadata: + decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language) + decision["metadata"] = {**v2_metadata, "reply_phase": "final"} + return decision + llm_decision = _voice_llm_decision( language=language, customer=customer, diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index d58acf8..d7cbaf8 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -5,6 +5,7 @@ import audioop import contextlib import hashlib import logging +import os import threading import time import uuid @@ -186,6 +187,10 @@ class AudioSocketMediaRuntime: self._immediate_ack_min_ms = 700 self._v2_ack_post_gap_seconds = 0.10 self._partial_poll_interval_seconds = 0.20 + self._streaming_asr_partial_poll_enabled = ( + str(os.getenv("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "0")).strip().lower() + in {"1", "true", "yes", "on"} + ) self._stable_partial_hold_seconds = 0.40 self._barge_in_trigger_ms = 220 self._streaming_asr_reopen_backoff_seconds = 2.0 @@ -197,7 +202,10 @@ class AudioSocketMediaRuntime: @staticmethod def _normalize_intent_text(text: str) -> str: - return " ".join(str(text or "").strip().lower().split()) + compact = str(text or "").strip().lower() + for char in ",.!?;:…\"'()[]{}": + compact = compact.replace(char, " ") + return " ".join(compact.split()) @classmethod def _is_low_signal_transcript(cls, text: str | None) -> bool: @@ -220,9 +228,12 @@ class AudioSocketMediaRuntime: "привет", "слышу", "слышно", + "твой", "угу", "хорошо", "ясно", + "давай", + "поргай что это", } def _detect_early_intent(self, text: str) -> str: @@ -270,7 +281,7 @@ class AudioSocketMediaRuntime: return "Сейчас сориентирую." if ack_kind == "clarify": return "Сейчас уточню." - return "Сейчас подскажу." + return "Секунду." @staticmethod def _should_use_emotive_ack(registration: MediaRegistration, language: str | None) -> bool: @@ -323,11 +334,10 @@ class AudioSocketMediaRuntime: "Ага, сейчас сориентирую.", ) return ( - "Угу, сейчас подскажу.", "Мхм, секунду.", - "Ага, сейчас подскажу.", + "Секунду.", "Хм, секунду.", - "Хорошо, сейчас подскажу.", + "Хорошо, секунду.", ) def _select_ack_payload( @@ -623,6 +633,36 @@ class AudioSocketMediaRuntime: intent = self._detect_early_intent(transcript_text) self._update_stable_partial_intent(actor, intent) + async def _run_streaming_partial_poll(self, actor: MediaActor, utterance_generation: int) -> None: + try: + if utterance_generation != actor.utterance_generation: + return + await self._poll_streaming_partial(actor) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "audiosocket.streaming_asr_partial_failed session_id=%s generation=%s error=%s", + actor.registration.voice_session_id, + utterance_generation, + str(exc)[:500], + ) + + def _maybe_schedule_streaming_partial_poll(self, actor: MediaActor) -> None: + if not self._streaming_asr_partial_poll_enabled: + return + if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id: + return + if time.monotonic() < actor.asr_poll_due_monotonic: + return + task = actor.partial_asr_task + if task is not None and not task.done(): + return + actor.partial_asr_task = asyncio.create_task( + self._run_streaming_partial_poll(actor, actor.utterance_generation), + name=f"streaming-asr-partial-{actor.registration.voice_session_id}", + ) + async def _finalize_streaming_transcription(self, actor: MediaActor) -> ASRTranscription: if not actor.asr_streaming_enabled or not actor.asr_stream_id: raise StreamingASRUnavailable("Streaming ASR stream is not active") @@ -756,10 +796,11 @@ class AudioSocketMediaRuntime: language: str | None, metadata: dict[str, Any], ack_source: str, + ack_kind: str | None = None, ) -> None: if actor.closed or actor.early_ack_started: return - ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown") + ack_kind = ack_kind or self._ack_kind_for_intent(actor.partial_intent or "unknown") ack_text, style_hints, ack_variant = self._select_ack_payload( actor, language=language, @@ -774,7 +815,7 @@ class AudioSocketMediaRuntime: metadata={ **metadata, "partial_transcript": actor.partial_transcript, - "early_intent": actor.partial_intent, + "early_intent": metadata.get("early_intent") or actor.partial_intent, "ack_kind": ack_kind, "ack_variant": ack_variant, "voice_style": "emotive_ack" if style_hints else "neutral_ack", @@ -1127,6 +1168,7 @@ class AudioSocketMediaRuntime: 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: self._queue_streaming_asr_pcm(actor, pcm_frame) + self._maybe_schedule_streaming_partial_poll(actor) elif actor.registration.voice_v2_partial_asr: self._maybe_schedule_partial_asr(actor) if vad_result.utterance_pcm: @@ -1183,14 +1225,7 @@ class AudioSocketMediaRuntime: "early_intent": partial_intent, } if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started: - if actor.asr_streaming_enabled: - with contextlib.suppress(StreamingASRUnavailable): - await self._poll_streaming_partial(actor) - partial_transcript = str(actor.partial_transcript or "").strip() - partial_intent = str(actor.stable_partial_intent or actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript) - base_metadata["partial_transcript"] = partial_transcript - base_metadata["early_intent"] = partial_intent - else: + if not actor.asr_streaming_enabled: partial_task = actor.partial_asr_task if partial_task is not None and not partial_task.done(): with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError): @@ -1212,6 +1247,7 @@ class AudioSocketMediaRuntime: language=actor.registration.language, metadata=base_metadata, ack_source="immediate_turn_close", + ack_kind="unknown", ) if actor.asr_streaming_enabled and not actor.asr_streaming_failed: diff --git a/services/shared/ai_context_summary.py b/services/shared/ai_context_summary.py index 5c5232d..9992bfd 100644 --- a/services/shared/ai_context_summary.py +++ b/services/shared/ai_context_summary.py @@ -33,6 +33,7 @@ _LOW_SIGNAL_TEXTS = { _CITY_ALIASES = { "алма ата": "Алма-Ата", "алмат": "Алмата", + "матта": "Алмата", "астан": "Астана", "актау": "Актау", "актоб": "Актобе", diff --git a/tests/test_ai_context_summary.py b/tests/test_ai_context_summary.py index 8c195dd..eb09cc3 100644 --- a/tests/test_ai_context_summary.py +++ b/tests/test_ai_context_summary.py @@ -24,6 +24,30 @@ def test_context_summary_tracks_city_and_branch_slot_for_schedule(): assert summary["open_slots"] == ["branch_or_address"] +def test_context_summary_maps_common_asr_city_mishearing_to_almaty(): + summary = update_context_summary_from_user_turn( + None, + channel="voice", + language="ru", + customer_name="Ернур", + text="Хочу узнать график работы.", + now="2026-04-12T00:00:00+00:00", + ) + + updated = update_context_summary_from_user_turn( + json.dumps(summary, ensure_ascii=False), + channel="voice", + language="ru", + customer_name="Ернур", + text="Матта.", + now="2026-04-12T00:00:02+00:00", + ) + + assert updated["active_intent"] == "schedule" + assert updated["confirmed_facts"]["city"] == "Алмата" + assert updated["open_slots"] == ["branch_or_address"] + + def test_context_summary_does_not_overwrite_meaningful_request_with_low_signal(): summary = update_context_summary_from_user_turn( None, diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index c5c5c2b..b2128eb 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1655,6 +1655,44 @@ def test_voice_postprocess_reply_uses_summary_context_when_raw_window_lost_topic assert "филиал" in normalized or "адрес" in normalized +def test_voice_decision_uses_summary_city_slot_instead_of_asking_city_again(monkeypatch): + def _unexpected_llm(messages): + raise AssertionError(f"LLM should not be called when summary slot prompt is enough: {messages!r}") + + monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm) + + decision = voice_module._voice_decision( + language="ru", + customer=SimpleNamespace(display_name="Ернур"), + interaction=SimpleNamespace(interaction_id="int_voice_summary_slot", customer_id=None, status="open", queue_id=None, subject=None), + transcript_text="Матта.", + transcript_window=[ + SimpleNamespace(speaker="caller", text="Матта.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()), + ], + context_summary=json.dumps( + { + "customer_name": "Ернур", + "active_intent": "schedule", + "active_request_text": "Хочу узнать график работы", + "confirmed_facts": {"city": "Алмата", "branch_hint": None, "service_hint": "график работы", "request_number": None}, + "open_slots": ["branch_or_address"], + }, + ensure_ascii=False, + ), + kb_results=[], + disclosure_required=False, + customer_name_value="Ернур", + customer_name_status="name_obtained", + request_metadata={"voice_v2_enabled": True}, + ) + + assert decision["intent"] == "clarification" + assert decision["needs_handoff"] is False + assert "Алмата" in decision["reply_text"] + assert "город" in decision["reply_text"].lower() + assert "филиал" in decision["reply_text"].lower() or "адрес" in decision["reply_text"].lower() + + def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch): monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1") monkeypatch.setenv("AI_PROVIDER", "stub") diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index 4b8145d..e554e80 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -1047,6 +1047,12 @@ def test_media_runtime_voice_v2_emits_blind_ack_on_first_turn_without_partial_si assert speak_events[1][0] == "Подскажите подробнее, пожалуйста." +def test_media_runtime_low_signal_filter_catches_short_asr_noise(): + assert AudioSocketMediaRuntime._is_low_signal_transcript("Давай") + assert AudioSocketMediaRuntime._is_low_signal_transcript("твой") + assert AudioSocketMediaRuntime._is_low_signal_transcript("Поргай, что это") + + 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]] = [] @@ -1624,6 +1630,120 @@ def test_media_runtime_streaming_sidecar_push_does_not_block_vad_finalization(): assert streaming_provider.push_count >= 1 +def test_media_runtime_streaming_partial_poll_does_not_block_turn_close_ack(): + events: list[str] = [] + + class _StreamingProvider(StreamingASRProvider): + name = "streaming-sidecar" + supports_streaming = True + + def __init__(self) -> None: + self.poll_count = 0 + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + del stream_id + self.poll_count += 1 + time.sleep(0.25) + return StreamingASRPartial(text="need schedule", language="ru", confidence=0.8, is_stable=True) + + def finalize(self, stream_id: str) -> ASRTranscription: + assert stream_id == "stream-1" + events.append("finalize") + return ASRTranscription(text="need schedule", language="ru", confidence=0.9) + + def close_stream(self, stream_id: str) -> None: + assert stream_id == "stream-1" + + streaming_provider = _StreamingProvider() + 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="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, + ) + + async def _fake_speak_reply( + actor: MediaActor, + text: str, + *, + is_greeting: bool, + style_hints: dict[str, object] | None = None, + reply_phase: str | None = "main", + ) -> None: + del actor, text, is_greeting, style_hints + events.append(str(reply_phase)) + + runtime._speak_reply = _fake_speak_reply # type: ignore[method-assign] + + async def _scenario() -> None: + actor = MediaActor( + registration=MediaRegistration( + voice_session_id="avs_media_runtime_turn_close_no_poll", + call_id="call_media_runtime_turn_close_no_poll", + interaction_id="int_media_runtime_turn_close_no_poll", + ai_session_id="ais_media_runtime_turn_close_no_poll", + 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, + ) + actor.asr_streaming_enabled = True + actor.asr_stream_id = "stream-1" + pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 + await runtime._process_utterance(actor, pcm_frame * 40, False) + + asyncio.run(_scenario()) + + assert streaming_provider.poll_count == 0 + assert events[:2] == ["ack", "finalize"] + assert "main" in events + + def test_media_runtime_merges_thinking_continuation_into_current_utterance(): runtime = AudioSocketMediaRuntime( enabled=True,