diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 74224d3..0a1fc32 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -267,6 +267,10 @@ def _normalize_name_candidate(text: str | None) -> str | None: filtered = [word for word, lowered_word in zip(words, lowered) if lowered_word not in stop_words] if not filtered: return None + if len(filtered) == 2 and len(filtered[1]) == 1: + filtered = [filtered[0]] + elif len(filtered) == 2 and len(filtered[0]) == 1: + filtered = [filtered[1]] invalid_tokens = { "\u0434\u0430", "\u0445\u0430\u0447\u0443", @@ -1430,9 +1434,12 @@ def _voice_early_plan( *, language: str, transcript_text: str, + context_summary: str | dict[str, Any] | None = None, request_metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: v2_metadata = _voice_v2_metadata(transcript_text, request_metadata) + payload = request_metadata if isinstance(request_metadata, dict) else {} + early_intent = str(payload.get("early_intent") or v2_metadata.get("early_intent") or "").strip() lower_text = str(transcript_text or "").strip().lower() if _app()._looks_like_human_request(lower_text) or _app()._is_sensitive_request(lower_text): return { @@ -1471,6 +1478,55 @@ def _voice_early_plan( "reply_phase": "early_plan", }, } + if early_intent in {"schedule", "address", "price", "status", "problem", "operator_request"}: + reply_text = _voice_summary_slot_prompt(language, context_summary) or _voice_topic_prompt(language, [transcript_text]) + if early_intent == "operator_request": + reply_text = _voice_handoff_reply(language) + if not reply_text and language == "kz": + if early_intent == "schedule": + reply_text = "Qai filialdyn, mekenjaidyn nemese qalanyng jumys uaqyty qyzyqtyratynyn aitnyz." + elif early_intent == "address": + reply_text = "Qai filial, mekenjai nemese qala qyzyqtyratynyn aitnyz." + elif early_intent == "price": + reply_text = "Qai qyzmet, tarif nemese bagasy qyzyqtyratynyn naqtylanyz." + elif early_intent == "status": + reply_text = "Otinish nomirin nemese resimdegen telefon nomirin aitnyz." + elif early_intent == "problem": + reply_text = "Bir soilemmen naqty aitynyz: ne istep turmagan?" + if not reply_text: + if early_intent == "schedule": + reply_text = "\u041f\u043e\u0434\u0441\u043a\u0430\u0436\u0438\u0442\u0435, \u0433\u0440\u0430\u0444\u0438\u043a \u0440\u0430\u0431\u043e\u0442\u044b \u043a\u0430\u043a\u043e\u0433\u043e \u0444\u0438\u043b\u0438\u0430\u043b\u0430, \u0430\u0434\u0440\u0435\u0441\u0430 \u0438\u043b\u0438 \u0433\u043e\u0440\u043e\u0434\u0430 \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442?" + elif early_intent == "address": + reply_text = "\u041f\u043e\u0434\u0441\u043a\u0430\u0436\u0438\u0442\u0435, \u043a\u0430\u043a\u043e\u0439 \u0444\u0438\u043b\u0438\u0430\u043b, \u0430\u0434\u0440\u0435\u0441 \u0438\u043b\u0438 \u0433\u043e\u0440\u043e\u0434 \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442." + elif early_intent == "price": + reply_text = "\u0423\u0442\u043e\u0447\u043d\u0438\u0442\u0435, \u043a\u0430\u043a\u0430\u044f \u0443\u0441\u043b\u0443\u0433\u0430, \u0442\u0430\u0440\u0438\u0444 \u0438\u043b\u0438 \u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442." + elif early_intent == "status": + reply_text = "\u041d\u0430\u0437\u043e\u0432\u0438\u0442\u0435, \u043f\u043e\u0436\u0430\u043b\u0443\u0439\u0441\u0442\u0430, \u043d\u043e\u043c\u0435\u0440 \u0437\u0430\u044f\u0432\u043a\u0438 \u0438\u043b\u0438 \u0442\u0435\u043b\u0435\u0444\u043e\u043d, \u043f\u043e \u043a\u043e\u0442\u043e\u0440\u043e\u043c\u0443 \u043e\u043d\u0430 \u043e\u0444\u043e\u0440\u043c\u043b\u044f\u043b\u0430\u0441\u044c." + elif early_intent == "problem": + reply_text = "\u041e\u043f\u0438\u0448\u0438\u0442\u0435, \u043f\u043e\u0436\u0430\u043b\u0443\u0439\u0441\u0442\u0430, \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0443 \u043e\u0434\u043d\u0438\u043c \u043f\u0440\u0435\u0434\u043b\u043e\u0436\u0435\u043d\u0438\u0435\u043c: \u0447\u0442\u043e \u0438\u043c\u0435\u043d\u043d\u043e \u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442?" + if reply_text: + return { + "language": language, + "intent": "handoff_request" if early_intent == "operator_request" else "clarification", + "reply_text": _voice_compact_reply_text(reply_text, language=language), + "confidence": 0.62, + "needs_handoff": early_intent == "operator_request", + "handoff_reason": ( + "Запрос требует участия живого оператора." + if early_intent == "operator_request" + else None + ), + "case_action": "keep_open", + "kb_refs": [], + "summary_text": "Early domain plan is prepared.", + "model": "voice_early_plan_domain", + "latency_ms": 1, + "metadata": { + **v2_metadata, + "reply_phase": "early_plan", + "early_intent": early_intent, + }, + } return { "language": language, "intent": "clarification", @@ -1835,6 +1891,7 @@ def _voice_decision( return _voice_early_plan( language=language, transcript_text=transcript_text, + context_summary=context_summary, request_metadata=request_metadata, ) @@ -2579,7 +2636,8 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe resolved_at=effective_name_resolved_at, ) decision_metadata.update(decision.get("metadata") or {}) - if effective_name_status == "name_obtained" and effective_name_value: + suppress_name_prefix = bool(request_metadata.get("suppress_name_prefix")) if isinstance(request_metadata, dict) else False + if effective_name_status == "name_obtained" and effective_name_value and not early_plan_only and not suppress_name_prefix: decision["reply_text"] = _voice_reply_with_name( decision["language"], decision["reply_text"], diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 968ebc7..fd88dd5 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -87,9 +87,16 @@ class MediaActor: playback_active: bool = False early_ack_started: bool = False partial_transcript: str | None = None + stable_partial_transcript: str | None = None + last_partial_transcript: str | None = None + partial_transcript_streak: int = 0 partial_intent: str | None = None partial_intent_streak: int = 0 stable_partial_intent: str | None = None + early_plan_task: asyncio.Task | None = None + early_plan_transcript: str | None = None + early_plan_intent: str | None = None + early_plan_generation: int = 0 response_plan_id: str | None = None playback_generation: int = 0 tts_generation: int = 0 @@ -199,6 +206,25 @@ class AudioSocketMediaRuntime: self._streaming_asr_push_drain_timeout_seconds = 1.5 self._thinking_continuation_grace_seconds = 0.45 self._thinking_continuation_max_bytes = int(1800 * 16) + self._partial_first_final_enabled = ( + str(os.getenv("AI_VOICE_V2_PARTIAL_FIRST_FINAL", "1")).strip().lower() + in {"1", "true", "yes", "on"} + ) + self._early_plan_enabled = ( + str(os.getenv("AI_VOICE_V2_EARLY_PLAN_ENABLED", "1")).strip().lower() + in {"1", "true", "yes", "on"} + ) + raw_early_plan_intents = str( + os.getenv( + "AI_VOICE_V2_EARLY_PLAN_INTENTS", + "schedule,address,price,status,problem,operator_request", + ) + ) + self._early_plan_intents = { + item.strip().lower() + for item in raw_early_plan_intents.split(",") + if item.strip() + } @staticmethod def _normalize_intent_text(text: str) -> str: @@ -409,10 +435,20 @@ class AudioSocketMediaRuntime: actor.input_active = True actor.early_ack_started = False actor.partial_transcript = None + actor.stable_partial_transcript = None + actor.last_partial_transcript = None + actor.partial_transcript_streak = 0 actor.partial_intent = None actor.partial_intent_streak = 0 actor.stable_partial_intent = None - actor.response_plan_id = None + actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" + actor.early_plan_transcript = None + actor.early_plan_intent = None + actor.early_plan_generation = 0 + early_plan_task = actor.early_plan_task + actor.early_plan_task = None + if early_plan_task is not None and not early_plan_task.done(): + early_plan_task.cancel() actor.partial_asr_attempted = False actor.asr_streaming_failed = False actor.asr_poll_due_monotonic = 0.0 @@ -636,6 +672,134 @@ class AudioSocketMediaRuntime: if actor.partial_intent_streak >= 2: actor.stable_partial_intent = normalized + @staticmethod + def _update_stable_partial_transcript(actor: MediaActor, transcript_text: str, *, provider_stable: bool) -> None: + normalized = " ".join(str(transcript_text or "").strip().lower().split()) + if not normalized: + return + if actor.last_partial_transcript == normalized: + actor.partial_transcript_streak += 1 + else: + actor.last_partial_transcript = normalized + actor.partial_transcript_streak = 1 + if provider_stable or actor.partial_transcript_streak >= 2: + actor.stable_partial_transcript = transcript_text + + def _should_schedule_early_plan(self, actor: MediaActor, transcript_text: str, intent: str) -> bool: + if not self._early_plan_enabled or not self._should_use_voice_v2(actor.registration): + return False + if actor.closed or self._is_low_signal_transcript(transcript_text): + return False + normalized_intent = str(intent or "").strip().lower() or "unknown" + if normalized_intent not in self._early_plan_intents: + return False + task = actor.early_plan_task + if task is not None and task.done() and actor.early_plan_transcript == transcript_text and actor.early_plan_intent == normalized_intent: + return False + return task is None or task.done() + + def _schedule_early_plan( + self, + actor: MediaActor, + *, + transcript_text: str, + language: str | None, + intent: str, + source: str, + ) -> None: + if not self._should_schedule_early_plan(actor, transcript_text, intent): + return + if not actor.response_plan_id: + actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" + actor.early_plan_generation = actor.utterance_generation + actor.early_plan_transcript = transcript_text + actor.early_plan_intent = str(intent or "").strip() or "unknown" + metadata = { + "voice_v2_enabled": actor.registration.voice_v2_enabled, + "reply_phase": "early_plan", + "response_plan_id": actor.response_plan_id, + "playback_generation": actor.playback_generation + 1, + "tts_generation": actor.tts_generation + 1, + "partial_transcript": transcript_text, + "early_intent": actor.early_plan_intent, + "transcript_source": source, + "suppress_name_prefix": True, + } + actor.early_plan_task = asyncio.create_task( + asyncio.to_thread( + self._process_turn, + actor.registration.voice_session_id, + transcript_text, + language or actor.registration.language, + False, + metadata, + ), + name=f"voice-early-plan-{actor.registration.voice_session_id}", + ) + logger.info( + "audiosocket.early_plan_started session_id=%s generation=%s intent=%s source=%s text_len=%s", + actor.registration.voice_session_id, + actor.early_plan_generation, + actor.early_plan_intent, + source, + len(transcript_text), + ) + + async def _take_early_plan_decision( + self, + actor: MediaActor, + *, + timeout_seconds: float, + ) -> VoiceAITurnDecisionOut | None: + task = actor.early_plan_task + if task is None: + return None + if actor.early_plan_generation != actor.utterance_generation: + return None + if not task.done() and timeout_seconds > 0: + with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError): + await asyncio.wait_for(asyncio.shield(task), timeout=timeout_seconds) + if not task.done(): + return None + try: + decision = task.result() + except Exception as exc: + logger.warning( + "audiosocket.early_plan_failed session_id=%s generation=%s error=%s", + actor.registration.voice_session_id, + actor.early_plan_generation, + str(exc)[:500], + ) + return None + if not getattr(decision, "reply_text", None) and not getattr(decision, "needs_handoff", False): + return None + return decision + + async def _reconcile_final_decision_after_early_plan( + self, + actor: MediaActor, + *, + decision_task: asyncio.Task, + early_decision: VoiceAITurnDecisionOut, + transcript_text: str, + ) -> None: + try: + final_decision = await decision_task + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "audiosocket.final_plan_reconcile_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + return + if actor.closed: + return + if final_decision.needs_handoff and not early_decision.needs_handoff: + await self._set_actor_state(actor, "handoff_requested", final_decision.handoff_reason) + self._start_handoff_request(actor, transcript_text, final_decision) + async def _poll_streaming_partial(self, actor: MediaActor) -> None: if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id: return @@ -650,8 +814,19 @@ class AudioSocketMediaRuntime: if not transcript_text: return actor.partial_transcript = transcript_text + self._update_stable_partial_transcript(actor, transcript_text, provider_stable=bool(partial.is_stable or partial.is_final)) intent = self._detect_early_intent(transcript_text) self._update_stable_partial_intent(actor, intent) + stable_text = str(actor.stable_partial_transcript or "").strip() + stable_intent = str(actor.stable_partial_intent or "").strip() + if stable_text and stable_intent: + self._schedule_early_plan( + actor, + transcript_text=stable_text, + language=partial.language or actor.registration.language, + intent=stable_intent, + source="streaming_stable_partial", + ) async def _run_streaming_partial_poll(self, actor: MediaActor, utterance_generation: int) -> None: try: @@ -780,7 +955,16 @@ class AudioSocketMediaRuntime: if not transcript_text: return actor.partial_transcript = transcript_text + self._update_stable_partial_transcript(actor, transcript_text, provider_stable=True) actor.partial_intent = self._detect_early_intent(transcript_text) + actor.stable_partial_intent = actor.partial_intent + self._schedule_early_plan( + actor, + transcript_text=transcript_text, + language=transcription.language or actor.registration.language, + intent=actor.partial_intent, + source="partial_asr_probe", + ) logger.info( "audiosocket.partial_asr_ready session_id=%s generation=%s text_len=%s intent=%s", actor.registration.voice_session_id, @@ -880,6 +1064,9 @@ class AudioSocketMediaRuntime: ) -> None: if actor.closed or not text or not pcm_8k: return + actor.playback_interrupt.clear() + actor.barge_in_detected_monotonic = 0.0 + actor.barge_in_speech_ms = 0 await self._set_actor_state(actor, "speaking") actor.current_reply_phase = reply_phase interrupted = False @@ -1275,9 +1462,9 @@ class AudioSocketMediaRuntime: pcm_bytes = await self._extend_with_thinking_continuation(actor, pcm_bytes) actor.playback_generation += 1 actor.tts_generation += 1 - actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" + actor.response_plan_id = actor.response_plan_id or f"rsp_{uuid.uuid4().hex[:10]}" utterance_generation = actor.utterance_generation - partial_transcript = str(actor.partial_transcript or "").strip() + partial_transcript = str(actor.stable_partial_transcript or 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 = { "turn_duration_ms": int(len(pcm_bytes) / 16), @@ -1291,15 +1478,22 @@ class AudioSocketMediaRuntime: "early_intent": partial_intent, } if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started: - 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): - 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 + 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.stable_partial_transcript or 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 + if partial_transcript: + self._schedule_early_plan( + actor, + transcript_text=partial_transcript, + language=actor.registration.language, + intent=partial_intent, + source="turn_close_partial", + ) if self._should_emit_partial_ack(actor, partial_transcript, partial_intent): await self._emit_early_ack( actor, @@ -1316,9 +1510,11 @@ class AudioSocketMediaRuntime: ack_kind="unknown", ) + transcript_source = "batch" if actor.asr_streaming_enabled and not actor.asr_streaming_failed: try: transcription = await self._finalize_streaming_transcription(actor) + transcript_source = "streaming_final" except StreamingASRUnavailable as exc: logger.warning( "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", @@ -1327,12 +1523,22 @@ class AudioSocketMediaRuntime: ) self._mark_streaming_asr_backoff(actor) await self._close_streaming_asr(actor, drain=False) - 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, - ) + partial_first_text = str(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): + transcription = ASRTranscription( + text=partial_first_text, + language=actor.registration.language, + confidence=None, + ) + transcript_source = "streaming_partial_after_finalize_failure" + else: + 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, + ) + transcript_source = "batch_fallback_after_streaming_failure" else: wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) transcription = await asyncio.to_thread( @@ -1340,6 +1546,7 @@ class AudioSocketMediaRuntime: wav_bytes, language_hint=actor.registration.language, ) + transcript_source = "batch" transcript_text = str(transcription.text or "").strip() or partial_transcript logger.info( "audiosocket.asr_turn_ready session_id=%s provider=%s utterance_ms=%s text_len=%s empty=%s", @@ -1372,6 +1579,7 @@ class AudioSocketMediaRuntime: "partial_transcript": actor.partial_transcript, "early_intent": actor.partial_intent, "reply_phase": "final", + "transcript_source": transcript_source, } decision_task = asyncio.create_task( asyncio.to_thread( @@ -1386,18 +1594,30 @@ class AudioSocketMediaRuntime: }, ) ) + early_plan_decision: VoiceAITurnDecisionOut | None = None if self._should_use_voice_v2(actor.registration): - try: - decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds) - except asyncio.TimeoutError: - if not actor.early_ack_started: - await self._emit_early_ack( - actor, - language=transcription.language or actor.registration.language, - metadata=metadata, - ack_source="decision_timeout", - ) - decision = await decision_task + early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.04) + if early_plan_decision is not None: + decision = early_plan_decision + logger.info( + "audiosocket.early_plan_used session_id=%s generation=%s intent=%s", + actor.registration.voice_session_id, + actor.early_plan_generation, + actor.early_plan_intent, + ) + else: + try: + decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds) + except asyncio.TimeoutError: + if not actor.early_ack_started: + await self._emit_early_ack( + actor, + language=transcription.language or actor.registration.language, + metadata=metadata, + ack_source="decision_timeout", + ) + early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.12) + decision = early_plan_decision if early_plan_decision is not None else await decision_task else: decision = await decision_task handoff_task: asyncio.Task | None = None @@ -1425,6 +1645,24 @@ class AudioSocketMediaRuntime: }, ) await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main") + if early_plan_decision is not None: + if decision_task.done(): + await self._reconcile_final_decision_after_early_plan( + actor, + decision_task=decision_task, + early_decision=early_plan_decision, + transcript_text=transcript_text, + ) + else: + asyncio.create_task( + self._reconcile_final_decision_after_early_plan( + actor, + decision_task=decision_task, + early_decision=early_plan_decision, + transcript_text=transcript_text, + ), + name=f"voice-final-plan-reconcile-{actor.registration.voice_session_id}", + ) if actor.closed: return if decision.needs_handoff: @@ -1523,6 +1761,9 @@ class AudioSocketMediaRuntime: ) -> None: if actor.closed or not text: return + actor.playback_interrupt.clear() + actor.barge_in_detected_monotonic = 0.0 + actor.barge_in_speech_ms = 0 await self._set_actor_state(actor, "speaking") actor.current_reply_phase = reply_phase synth_started_at = time.monotonic() diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index b4d445e..7f57e43 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -131,14 +131,20 @@ def _elevenlabs_realtime_commit_strategy() -> str: def _elevenlabs_realtime_finalize_timeout_seconds() -> float: + raw_ms = os.getenv("AI_VOICE_V2_STREAMING_FINAL_HARD_TIMEOUT_MS", "").strip() + if raw_ms: + try: + return max(float(raw_ms) / 1000.0, 0.25) + except ValueError: + pass raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS", "").strip() if not raw: - return max(_streaming_asr_timeout_seconds(), 1.0) + return 0.7 try: value = float(raw) except ValueError: - value = _streaming_asr_timeout_seconds() - return max(value, 1.0) + value = 0.7 + return max(value, 0.25) def _normalize_elevenlabs_realtime_language(language: str | None) -> str: @@ -513,6 +519,9 @@ class _ElevenLabsRealtimeStreamState: updates_queue: queue.Queue[StreamingASRPartial] thread: threading.Thread | None = None latest_partial: StreamingASRPartial | None = None + stable_partial: StreamingASRPartial | None = None + last_partial_text: str = "" + partial_streak: int = 0 final_transcription: ASRTranscription | None = None error: BaseException | None = None close_requested: bool = False @@ -549,7 +558,7 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): if finalize_timeout_seconds is not None else _elevenlabs_realtime_finalize_timeout_seconds() ), - 1.0, + 0.25, ) self._model_id = str(model_id or _elevenlabs_realtime_model_id()).strip() or "scribe_v2_realtime" self._audio_format = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower() or "pcm_16000" @@ -650,6 +659,21 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): except queue.Empty: return with state.lock: + normalized_text = " ".join(str(update.text or "").strip().lower().split()) + if normalized_text and normalized_text == state.last_partial_text: + state.partial_streak += 1 + else: + state.last_partial_text = normalized_text + state.partial_streak = 1 if normalized_text else 0 + if update.is_final or update.is_stable or state.partial_streak >= 2: + state.stable_partial = StreamingASRPartial( + text=update.text, + language=update.language or state.language, + confidence=update.confidence, + is_final=update.is_final, + is_stable=True, + ) + update = state.stable_partial state.latest_partial = update if update.is_final: state.final_transcription = ASRTranscription( @@ -659,6 +683,25 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): ) state.final_event.set() + @staticmethod + def _transcription_from_partial( + partial: StreamingASRPartial, + *, + fallback_language: str, + ) -> ASRTranscription: + return ASRTranscription( + text=partial.text, + language=partial.language or fallback_language, + confidence=partial.confidence, + ) + + def _best_partial_transcription(self, state: _ElevenLabsRealtimeStreamState) -> ASRTranscription | None: + with state.lock: + partial = state.stable_partial or state.latest_partial + if partial is None: + return None + return self._transcription_from_partial(partial, fallback_language=state.language) + def _run_reader(self, state: _ElevenLabsRealtimeStreamState) -> None: try: while True: @@ -770,36 +813,38 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): with state.lock: if state.error is not None and state.latest_partial is None: raise StreamingASRUnavailable(str(state.error)[:500]) - return state.latest_partial + return state.stable_partial or state.latest_partial def finalize(self, stream_id: str) -> ASRTranscription: state = self._state(stream_id) silence = b"\x00\x00" * int(self._sample_rate_hz * 0.1) - self._send_json( - state, - { - "message_type": "input_audio_chunk", - "audio_base_64": base64.b64encode(silence).decode("ascii"), - "commit": True, - "sample_rate": self._sample_rate_hz, - }, - ) + try: + self._send_json( + state, + { + "message_type": "input_audio_chunk", + "audio_base_64": base64.b64encode(silence).decode("ascii"), + "commit": True, + "sample_rate": self._sample_rate_hz, + }, + ) + except StreamingASRUnavailable: + partial_transcription = self._best_partial_transcription(state) + if partial_transcription is not None and partial_transcription.text.strip(): + return partial_transcription + raise deadline = time.monotonic() + self._finalize_timeout_seconds while True: self._drain_updates(state) with state.lock: if state.final_transcription is not None: return state.final_transcription - latest_partial = state.latest_partial + latest_partial = state.stable_partial or state.latest_partial error = state.error event_is_set = state.final_event.is_set() if event_is_set or time.monotonic() >= deadline: if latest_partial is not None: - return ASRTranscription( - text=latest_partial.text, - language=latest_partial.language or state.language, - confidence=latest_partial.confidence, - ) + return self._transcription_from_partial(latest_partial, fallback_language=state.language) if error is not None: raise StreamingASRUnavailable(str(error)[:500]) if time.monotonic() >= deadline: diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index 2e39da8..5011133 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1451,6 +1451,37 @@ def test_voice_v2_streaming_duplex_early_plan_returns_fast_safe_reply_without_ll assert "оператор" in decision["reply_text"].lower() +def test_voice_v2_streaming_duplex_early_plan_returns_domain_followup_without_llm(monkeypatch): + monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex") + + def _unexpected_llm(messages): + raise AssertionError(f"LLM should not be called for early plan: {messages!r}") + + monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm) + + decision = voice_module._voice_decision( + language="ru", + customer=None, + interaction=SimpleNamespace(interaction_id="int_voice_early_schedule", status="new", queue_id="que_voice", subject="unknown"), + transcript_text="РњРЅРµ надо узнать график работы", + transcript_window=[], + kb_results=[], + disclosure_required=False, + request_metadata={ + "voice_v2_enabled": True, + "reply_phase": "early_plan", + "response_plan_id": "rsp_early_schedule", + "early_intent": "schedule", + }, + ) + + assert decision["model"] == "voice_early_plan_domain" + assert decision["reply_text"] + assert decision["needs_handoff"] is False + assert decision["metadata"]["reply_phase"] == "early_plan" + assert decision["metadata"]["early_intent"] == "schedule" + + def test_voice_decision_hearing_check_keeps_active_topic_without_llm(monkeypatch): def _unexpected_llm(messages): raise AssertionError(f"LLM should not be called for hearing check: {messages!r}") diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index 7834313..60ba1dd 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -399,6 +399,65 @@ def test_elevenlabs_realtime_streaming_provider_returns_partial_and_final(): assert websocket.sent_payloads[1]["commit"] is True +def test_elevenlabs_realtime_streaming_provider_finalizes_from_stable_partial_on_timeout(): + class _FakeRealtimeWebSocket: + def __init__(self) -> None: + self.sent_payloads: list[dict] = [] + self.incoming: queue.Queue[str | None] = queue.Queue() + + def send(self, raw_payload: str) -> None: + payload = json.loads(raw_payload) + self.sent_payloads.append(payload) + if not payload.get("commit"): + partial = json.dumps( + { + "message_type": "partial_transcript", + "text": "need schedule", + "language_code": "ru", + } + ) + self.incoming.put(partial) + self.incoming.put(partial) + + def recv(self) -> str: + item = self.incoming.get(timeout=1) + if item is None: + raise RuntimeError("closed") + return item + + def close(self) -> None: + self.incoming.put(None) + + websocket = _FakeRealtimeWebSocket() + + provider = asr_module.ElevenLabsRealtimeStreamingASRProvider( + api_base="https://api.elevenlabs.example", + api_key="asr-key", + timeout_seconds=1, + finalize_timeout_seconds=0.25, + websocket_factory=lambda url, *, header, timeout: websocket, + ) + stream_id = provider.open_stream("session-1", language_hint="ru") + provider.push_pcm(stream_id, b"\x01\x00" * 160) + + partial = None + for _ in range(20): + partial = provider.poll_partial(stream_id) + if partial is not None and partial.is_stable: + break + time.sleep(0.02) + + assert partial is not None + assert partial.is_stable is True + + final = provider.finalize(stream_id) + provider.close_stream(stream_id) + + assert final.text == "need schedule" + assert final.language == "ru" + assert websocket.sent_payloads[-1]["commit"] is True + + def test_yandex_grpc_streaming_provider_returns_partial_and_final(): calls: list[dict] = [] diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index ec9e85f..09407bf 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -1847,6 +1847,263 @@ def test_media_runtime_streaming_partial_poll_does_not_block_turn_close_ack(): assert "main" in events +def test_media_runtime_voice_v2_uses_early_plan_before_slow_final_decision(): + events: list[str | tuple[str, str]] = [] + + class _StreamingProvider(StreamingASRProvider): + name = "streaming-sidecar" + supports_streaming = True + + def finalize(self, stream_id: str) -> ASRTranscription: + assert stream_id == "stream-1" + return ASRTranscription(text="work schedule", language="ru", confidence=0.9) + + def close_stream(self, stream_id: str) -> None: + assert stream_id == "stream-1" + + def _process_turn(session_id, transcript_text, language, barge_in, metadata): + del session_id, transcript_text, barge_in + if metadata and metadata.get("reply_phase") == "early_plan": + events.append("early_plan") + return VoiceAITurnDecisionOut( + language=language or "ru", + intent="schedule", + reply_text="early reply", + confidence=0.8, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="early ready", + model="early", + latency_ms=1, + status="active", + ) + events.append("final_start") + time.sleep(0.25) + events.append("final_done") + return VoiceAITurnDecisionOut( + language=language or "ru", + intent="schedule", + reply_text="final reply", + confidence=0.9, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="final ready", + model="final", + latency_ms=1, + status="active", + ) + + 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=_StreamingProvider(), + 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=_process_turn, + 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, is_greeting, style_hints + events.append((str(reply_phase), text)) + + runtime._speak_reply = _fake_speak_reply # type: ignore[method-assign] + + async def _scenario() -> None: + actor = MediaActor( + registration=MediaRegistration( + voice_session_id="avs_media_runtime_early_plan", + call_id="call_media_runtime_early_plan", + interaction_id="int_media_runtime_early_plan", + ai_session_id="ais_media_runtime_early_plan", + 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" + actor.partial_transcript = "work schedule" + actor.stable_partial_transcript = "work schedule" + actor.partial_intent = "schedule" + actor.stable_partial_intent = "schedule" + pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 + await runtime._process_utterance(actor, pcm_frame * 40, False) + for _ in range(50): + if "final_done" in events: + break + await asyncio.sleep(0.01) + + asyncio.run(_scenario()) + + assert ("main", "early reply") in events + assert ("main", "final reply") not in events + assert events.index(("main", "early reply")) < events.index("final_done") + + +def test_media_runtime_voice_v2_uses_partial_as_final_when_streaming_finalize_fails(): + captured: list[tuple[str, dict | None]] = [] + + 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: + del audio_bytes + self.transcribe_count += 1 + return ASRTranscription(text="batch text", language=language_hint or "ru", confidence=0.9) + + class _FailingStreamingProvider(StreamingASRProvider): + name = "failing-streaming" + supports_streaming = True + + def finalize(self, stream_id: str) -> ASRTranscription: + assert stream_id == "stream-1" + raise StreamingASRUnavailable("finalize timeout") + + def close_stream(self, stream_id: str) -> None: + assert stream_id == "stream-1" + + asr_provider = _CountingASRProvider() + 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=_FailingStreamingProvider(), + 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: ( + captured.append((transcript_text, metadata)) + or 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, 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_partial_final", + call_id="call_media_runtime_partial_final", + interaction_id="int_media_runtime_partial_final", + ai_session_id="ais_media_runtime_partial_final", + 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" + actor.partial_transcript = "work schedule" + actor.stable_partial_transcript = "work schedule" + actor.partial_intent = "schedule" + actor.stable_partial_intent = "schedule" + pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160 + await runtime._process_utterance(actor, pcm_frame * 40, False) + + asyncio.run(_scenario()) + + assert asr_provider.transcribe_count == 0 + final_turns = [item for item in captured if item[1] and item[1].get("reply_phase") == "final"] + assert final_turns + assert final_turns[0][0] == "work schedule" + assert final_turns[0][1]["transcript_source"] == "streaming_partial_after_finalize_failure" + + def test_media_runtime_merges_thinking_continuation_into_current_utterance(): runtime = AudioSocketMediaRuntime( enabled=True, diff --git a/tests/test_ai_voice_media_runtime_streaming_stable.py b/tests/test_ai_voice_media_runtime_streaming_stable.py index d0d6348..9c74b9d 100644 --- a/tests/test_ai_voice_media_runtime_streaming_stable.py +++ b/tests/test_ai_voice_media_runtime_streaming_stable.py @@ -36,7 +36,7 @@ class _StubTTSProvider(TTSProvider): def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr_stable(): reply_starts: list[tuple[str, str | None, float]] = [] - turns: list[str] = [] + turns: list[tuple[str, str]] = [] class _BatchASRProvider(ASRProvider): name = "batch-asr" @@ -131,7 +131,7 @@ def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr plan_reply=lambda session_id, text, metadata, kind: None, record_latency=lambda session_id, metric, latency_ms: None, process_turn=lambda session_id, transcript_text, language, barge_in, metadata: ( - turns.append(transcript_text) + turns.append((str((metadata or {}).get("reply_phase") or "final"), transcript_text)) or VoiceAITurnDecisionOut( language=language or "ru", intent="schedule", @@ -191,7 +191,7 @@ def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr phases = [phase for _, phase, _ in reply_starts] finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize") - assert turns == ["need work schedule"] + assert [text for phase, text in turns if phase == "final"] == ["need work schedule"] assert "open" in [name for name, _ in streaming_provider.events] assert "close" in [name for name, _ in streaming_provider.events] assert phases[:2] == ["ack", "main"] diff --git a/tests/test_voice_start_policy.py b/tests/test_voice_start_policy.py index 60d880e..b176d50 100644 --- a/tests/test_voice_start_policy.py +++ b/tests/test_voice_start_policy.py @@ -48,6 +48,10 @@ def test_voice_start_name_outcome_rejects_garbage_name_inside_mixed_request(): assert source == "none" +def test_normalize_name_candidate_drops_trailing_asr_initial(): + assert voice_policy._normalize_name_candidate("\u0415\u0440\u043d\u0443\u0440 \u0418") == "\u0415\u0440\u043d\u0443\u0440" + + def test_display_name_looks_trusted_rejects_phone_and_accepts_real_name(): customer = SimpleNamespace(display_name="+77010000001") assert voice_policy._display_name_looks_trusted(customer, "+77010000001", None) is False