diff --git a/.env.example b/.env.example index 1e99052..73880fe 100644 --- a/.env.example +++ b/.env.example @@ -50,10 +50,15 @@ AI_WHATSAPP_MAX_KB_RESULTS=3 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_VOICE_RUNTIME_SERVICE_URL=http://localhost:8018 AI_VOICE_ENABLED=0 -AI_VOICE_POLICY_MODE=llm_guarded +AI_VOICE_POLICY_MODE=v2_fast_conversational AI_VOICE_QUEUE_CONFIG_JSON={} AI_VOICE_ASR_PROVIDER=openai -AI_VOICE_TTS_PROVIDER=openai +AI_VOICE_TTS_PROVIDER=yandex +AI_VOICE_V2_ENABLED=1 +AI_VOICE_V2_QUEUE_CODES=voice_lab_ai +AI_VOICE_V2_ACK_MODE=immediate_short +AI_VOICE_V2_STREAMING_TTS=1 +AI_VOICE_V2_PARTIAL_ASR=1 AI_VOICE_AUDIOSOCKET_ENABLED=0 AI_VOICE_AUDIOSOCKET_HOST=0.0.0.0 AI_VOICE_AUDIOSOCKET_PORT=9019 @@ -73,7 +78,7 @@ AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ=8000 AI_VOICE_TTS_CACHE_ENABLED=1 AI_VOICE_TTS_CACHE_DIR=.data/ai_voice_tts_cache AI_VOICE_VAD_MIN_SPEECH_MS=300 -AI_VOICE_VAD_TRAILING_SILENCE_MS=700 +AI_VOICE_VAD_TRAILING_SILENCE_MS=400 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 AI_VOICE_MAX_CONTEXT_SEGMENTS=8 diff --git a/.env.production.template b/.env.production.template index 7f32934..64642aa 100644 --- a/.env.production.template +++ b/.env.production.template @@ -45,10 +45,15 @@ AI_WHATSAPP_MAX_CONTEXT_MESSAGES=20 AI_WHATSAPP_MAX_KB_RESULTS=3 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_VOICE_ENABLED=0 -AI_VOICE_POLICY_MODE=llm_guarded +AI_VOICE_POLICY_MODE=v2_fast_conversational AI_VOICE_QUEUE_CONFIG_JSON={"voice_lab_ai":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}} AI_VOICE_ASR_PROVIDER=openai -AI_VOICE_TTS_PROVIDER=openai +AI_VOICE_TTS_PROVIDER=yandex +AI_VOICE_V2_ENABLED=1 +AI_VOICE_V2_QUEUE_CODES=voice_lab_ai +AI_VOICE_V2_ACK_MODE=immediate_short +AI_VOICE_V2_STREAMING_TTS=1 +AI_VOICE_V2_PARTIAL_ASR=1 AI_VOICE_AUDIOSOCKET_ENABLED=1 AI_VOICE_AUDIOSOCKET_HOST=0.0.0.0 AI_VOICE_AUDIOSOCKET_PORT=9019 @@ -69,7 +74,7 @@ AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ=8000 AI_VOICE_TTS_CACHE_ENABLED=1 AI_VOICE_TTS_CACHE_DIR=/app/.data_local/ai_voice_tts_cache AI_VOICE_VAD_MIN_SPEECH_MS=300 -AI_VOICE_VAD_TRAILING_SILENCE_MS=700 +AI_VOICE_VAD_TRAILING_SILENCE_MS=400 AI_VOICE_TURN_MAX_MS=10000 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15 AI_VOICE_MAX_CONTEXT_SEGMENTS=8 diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index a8efc53..373c751 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -18,8 +18,9 @@ x-app-env: &app_env AI_VOICE_RUNTIME_SERVICE_URL: http://ai-voice-runtime-service:8000 AI_VOICE_ENABLED: "0" AI_VOICE_QUEUE_CONFIG_JSON: "{}" + AI_VOICE_POLICY_MODE: v2_fast_conversational AI_VOICE_ASR_PROVIDER: openai - AI_VOICE_TTS_PROVIDER: openai + AI_VOICE_TTS_PROVIDER: yandex AI_VOICE_AUDIOSOCKET_ENABLED: "1" AI_VOICE_AUDIOSOCKET_HOST: 0.0.0.0 AI_VOICE_AUDIOSOCKET_PORT: "9019" @@ -29,7 +30,12 @@ x-app-env: &app_env AI_VOICE_TTS_CACHE_ENABLED: "1" AI_VOICE_TTS_CACHE_DIR: /app/.data/ai_voice_tts_cache AI_VOICE_VAD_MIN_SPEECH_MS: "300" - AI_VOICE_VAD_TRAILING_SILENCE_MS: "700" + AI_VOICE_VAD_TRAILING_SILENCE_MS: "400" + AI_VOICE_V2_ENABLED: "1" + AI_VOICE_V2_QUEUE_CODES: voice_lab_ai + AI_VOICE_V2_ACK_MODE: immediate_short + AI_VOICE_V2_STREAMING_TTS: "1" + AI_VOICE_V2_PARTIAL_ASR: "1" AI_VOICE_TURN_MAX_MS: "10000" AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS: "15" AI_VOICE_MAX_CONTEXT_SEGMENTS: "8" diff --git a/scripts/local_stack.py b/scripts/local_stack.py index 680fa3e..de0f8e8 100644 --- a/scripts/local_stack.py +++ b/scripts/local_stack.py @@ -309,9 +309,16 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ env["AI_VOICE_ENABLED"] = env.get("AI_VOICE_ENABLED", "0") env["AI_VOICE_QUEUE_CONFIG_JSON"] = env.get("AI_VOICE_QUEUE_CONFIG_JSON", "{}") env["AI_VOICE_ASR_PROVIDER"] = env.get("AI_VOICE_ASR_PROVIDER", "openai") - env["AI_VOICE_TTS_PROVIDER"] = env.get("AI_VOICE_TTS_PROVIDER", "openai") + env["AI_VOICE_TTS_PROVIDER"] = env.get("AI_VOICE_TTS_PROVIDER", "yandex") + env["AI_VOICE_POLICY_MODE"] = env.get("AI_VOICE_POLICY_MODE", "v2_fast_conversational") + env["AI_VOICE_V2_ENABLED"] = env.get("AI_VOICE_V2_ENABLED", "1") + env["AI_VOICE_V2_QUEUE_CODES"] = env.get("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai") + env["AI_VOICE_V2_ACK_MODE"] = env.get("AI_VOICE_V2_ACK_MODE", "immediate_short") + env["AI_VOICE_V2_STREAMING_TTS"] = env.get("AI_VOICE_V2_STREAMING_TTS", "1") + env["AI_VOICE_V2_PARTIAL_ASR"] = env.get("AI_VOICE_V2_PARTIAL_ASR", "1") env["AI_VOICE_MAX_CONTEXT_SEGMENTS"] = env.get("AI_VOICE_MAX_CONTEXT_SEGMENTS", "8") env["AI_VOICE_HANDOFF_TIMEOUT_SECONDS"] = env.get("AI_VOICE_HANDOFF_TIMEOUT_SECONDS", "8") + env["AI_VOICE_VAD_TRAILING_SILENCE_MS"] = env.get("AI_VOICE_VAD_TRAILING_SILENCE_MS", "400") env["AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS"] = env.get( "AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS", "svc:ai-voice-runtime", diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 04ac415..2bed35f 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -1060,6 +1060,78 @@ def _voice_policy_mode() -> str: return persona.voice_policy_mode() +def _voice_v2_enabled(metadata: dict[str, Any] | None = None) -> bool: + payload = metadata if isinstance(metadata, dict) else {} + if bool(payload.get("voice_v2_enabled")): + return True + return _voice_policy_mode() == "v2_fast_conversational" + + +def _voice_early_intent_bucket(text: str) -> str: + normalized = " ".join(str(text or "").strip().lower().split()) + if not normalized: + return "unknown" + if any(token in normalized for token in ("оператор", "оператором", "человек", "менеджер", "сотрудник")): + return "operator_request" + if any(token in normalized for token in ("график", "распис", "жұмыс")): + return "schedule" + if any(token in normalized for token in ("адрес", "филиал", "офис", "мекен", "қайда")): + return "address" + if any(token in normalized for token in ("тариф", "цена", "стоимость", "баға", "сколько стоит")): + return "price" + if any(token in normalized for token in ("статус", "заявк", "заказ", "өтінім")): + return "status" + if any(token in normalized for token in ("не работает", "ошибка", "проблем", "істемей")): + return "problem" + return "unknown" + + +def _voice_ack_kind_for_intent(intent: str) -> str: + if intent == "operator_request": + return "handoff" + if intent in {"schedule", "address", "price", "status", "problem"}: + return "understanding" + return "generic" + + +def _voice_compact_reply_text(text: str, *, language: str) -> str: + normalized = " ".join(str(text or "").strip().split()) + if not normalized: + return normalized + if len(normalized) <= 180: + return normalized + parts = [segment.strip() for segment in re.split(r"(?<=[.!?])\s+", normalized) if segment.strip()] + if parts: + compact = " ".join(parts[:2]).strip() + if len(compact) <= 180: + return compact + shortened = normalized[:177].rsplit(" ", 1)[0].strip() + if not shortened: + shortened = normalized[:177].strip() + ending = "…" if language == "kz" else "..." + return f"{shortened}{ending}" + + +def _voice_v2_metadata( + transcript_text: str, + request_metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + if not _voice_v2_enabled(request_metadata): + return {} + payload = request_metadata if isinstance(request_metadata, dict) else {} + early_intent = _voice_early_intent_bucket(transcript_text) + metadata: dict[str, Any] = { + "voice_v2_enabled": True, + "early_intent": early_intent, + "ack_kind": _voice_ack_kind_for_intent(early_intent), + } + for key in ("response_plan_id", "playback_generation", "partial_transcript"): + value = payload.get(key) + if value not in {None, ""}: + metadata[key] = value + return metadata + + def _voice_llm_prompt_messages( *, language: str, @@ -1155,7 +1227,7 @@ def _voice_llm_decision( name_status: str | None, ) -> dict[str, Any] | None: app = _app() - if _voice_policy_mode() != "llm_guarded": + if _voice_policy_mode() not in {"llm_guarded", "v2_fast_conversational"}: return None if app._ai_provider() != "openai_compatible": return None @@ -1385,12 +1457,14 @@ def _voice_decision( disclosure_required: bool, customer_name_value: str | None = None, customer_name_status: str | None = None, + request_metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: app = _app() normalized = str(transcript_text or "").strip() lower_text = normalized.lower() caller_texts = _voice_recent_caller_texts(transcript_window) model = app._ai_model() + v2_metadata = _voice_v2_metadata(transcript_text, request_metadata) if app._looks_like_human_request(lower_text) or app._is_sensitive_request(lower_text): return { @@ -1441,12 +1515,21 @@ def _voice_decision( name_status=customer_name_status, ) if llm_decision is not None: + if v2_metadata: + llm_decision["reply_text"] = _voice_compact_reply_text( + str(llm_decision.get("reply_text") or ""), + language=language, + ) + llm_decision["metadata"] = { + **(llm_decision.get("metadata") or {}), + **v2_metadata, + } return llm_decision if kb_results: article = kb_results[0] snippet = app._article_snippet(article, limit=220) - return { + decision = { "language": language, "intent": "kb_answer", "reply_text": ( @@ -1463,11 +1546,15 @@ def _voice_decision( "model": "voice_policy_fallback", "latency_ms": 1, } + if v2_metadata: + decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language) + decision["metadata"] = v2_metadata + return decision reply_text = _voice_confusion_prompt(language, caller_texts) if caller_confused else ( _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language) ) - return { + decision = { "language": language, "intent": "clarification", "reply_text": reply_text, @@ -1480,6 +1567,10 @@ def _voice_decision( "model": "voice_policy_fallback", "latency_ms": 1, } + if v2_metadata: + decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language) + decision["metadata"] = v2_metadata + return decision def start_voice_session(session_id: str, payload: VoiceAIStartIn) -> VoiceAIStartOut: @@ -1971,6 +2062,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe disclosure_required=disclosure_required, customer_name_value=name_update["value"], customer_name_status=name_update["status"], + request_metadata=payload.metadata if isinstance(payload.metadata, dict) else {}, ) if decision.get("extracted_name"): @@ -2008,6 +2100,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe source=name_update["source"], resolved_at=name_update["resolved_at"], ) + decision_metadata.update(decision.get("metadata") or {}) if name_update["status"] == "name_obtained" and name_update["value"]: decision["reply_text"] = _voice_reply_with_name( decision["language"], @@ -2021,6 +2114,11 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe if str(decision["reply_text"] or "").strip() else inline_followup ) + if decision_metadata.get("voice_v2_enabled"): + decision["reply_text"] = _voice_compact_reply_text( + decision["reply_text"], + language=decision["language"], + ) decision["metadata"] = decision_metadata _record_voice_ai_turn( session, diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index 942dbea..594c184 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -33,7 +33,7 @@ from services.shared.models import ( ) from services.shared.security import issue_app_token, require_roles from services.shared.sql_init import init_sql_schema -from services.shared.sql_models import VoiceAISessionRow, VoiceTranscriptSegmentRow +from services.shared.sql_models import AsteriskCallLinkRow, VoiceAISessionRow, VoiceTranscriptSegmentRow from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence init_sql_schema() @@ -86,7 +86,7 @@ def _asr_provider_name() -> str: def _tts_provider_name() -> str: - return os.getenv("AI_VOICE_TTS_PROVIDER", "openai").strip() or "openai" + return os.getenv("AI_VOICE_TTS_PROVIDER", "yandex").strip() or "yandex" def _handoff_timeout_seconds() -> float: @@ -114,7 +114,7 @@ def _vad_min_speech_ms() -> int: def _vad_trailing_silence_ms() -> int: - return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 500), _vad_frame_ms()) + return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 400), _vad_frame_ms()) def _turn_max_ms() -> int: @@ -132,6 +132,27 @@ def _media_registration_wait_timeout_seconds() -> float: return max(_float_env("AI_VOICE_MEDIA_REGISTRATION_WAIT_TIMEOUT_SECONDS", 12.0), 0.0) +def _voice_v2_enabled() -> bool: + return _bool_env("AI_VOICE_V2_ENABLED", False) + + +def _voice_v2_queue_codes() -> set[str]: + raw = str(os.getenv("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai") or "voice_lab_ai").strip() + return {item.strip() for item in raw.split(",") if item.strip()} + + +def _voice_v2_ack_mode() -> str: + return str(os.getenv("AI_VOICE_V2_ACK_MODE", "immediate_short") or "immediate_short").strip().lower() + + +def _voice_v2_streaming_tts_enabled() -> bool: + return _bool_env("AI_VOICE_V2_STREAMING_TTS", True) + + +def _voice_v2_partial_asr_enabled() -> bool: + return _bool_env("AI_VOICE_V2_PARTIAL_ASR", True) + + def _service_headers() -> dict[str, str]: token = issue_app_token( subject="svc:ai-voice-runtime", @@ -457,7 +478,11 @@ def _push_bridge_call_state( ) -def _media_registration_from_row(row: VoiceAISessionRow) -> MediaRegistration: +def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | None = None) -> MediaRegistration: + normalized_queue_code = str(queue_code or "").strip() or None + voice_v2_for_session = bool( + _voice_v2_enabled() and normalized_queue_code and normalized_queue_code in _voice_v2_queue_codes() + ) return MediaRegistration( voice_session_id=row.session_id, call_id=row.call_id, @@ -465,6 +490,13 @@ def _media_registration_from_row(row: VoiceAISessionRow) -> MediaRegistration: ai_session_id=str(row.ai_session_id or "").strip() or None, language=str(row.language or "").strip() or None, media_uuid=str(row.media_uuid or "").strip() or None, + queue_code=normalized_queue_code, + queue_id=str(row.queue_id or "").strip() or None, + agent_profile=str(row.agent_profile or "").strip() or None, + voice_v2_enabled=voice_v2_for_session, + voice_v2_ack_mode=_voice_v2_ack_mode() if voice_v2_for_session else "disabled", + voice_v2_streaming_tts=bool(voice_v2_for_session and _voice_v2_streaming_tts_enabled()), + voice_v2_partial_asr=bool(voice_v2_for_session and _voice_v2_partial_asr_enabled()), ) @@ -480,7 +512,14 @@ def _load_media_registration_by_uuid(media_uuid: str) -> MediaRegistration | Non ).scalar_one_or_none() if row is None: return None - return _media_registration_from_row(row) + call_link = session.execute( + select(AsteriskCallLinkRow) + .where(AsteriskCallLinkRow.call_id == row.call_id) + .order_by(AsteriskCallLinkRow.id.desc()) + .limit(1) + ).scalar_one_or_none() + queue_code = str(call_link.queue_code or "").strip() or None if call_link is not None else None + return _media_registration_from_row(row, queue_code=queue_code) finally: session.close() @@ -635,6 +674,42 @@ def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool) -> None _retry_db_write(_write) +def _record_runtime_reply_planned( + session_id: str, + text: str, + metadata: dict[str, Any] | None = None, + kind: str = "reply", +) -> None: + normalized_text = str(text or "").strip() + if not normalized_text: + return + + def _write() -> None: + session = get_session() + try: + voice_session = _load_voice_session(session, session_id) + _record_segment( + session, + voice_session=voice_session, + speaker="assistant", + source_type="tts", + text=normalized_text, + sequence_no=_next_sequence(session, voice_session.session_id), + payload={ + "delivery_status": "planned", + "kind": kind, + "metadata": metadata or {}, + }, + is_final=False, + ) + voice_session.updated_at = utc_now_iso() + session.commit() + finally: + session.close() + + _retry_db_write(_write) + + def _set_voice_session_state( session_id: str, ai_state: str, @@ -866,6 +941,7 @@ def _process_voice_ai_turn_sync( ) -> VoiceAITurnDecisionOut: session = get_session() voice_session = None + payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {} try: voice_session = _load_voice_session(session, session_id) if voice_session.status in {"human_owned", "completed", "error"}: @@ -883,7 +959,7 @@ def _process_voice_ai_turn_sync( source_type="asr", text=payload.transcript_text, sequence_no=max(payload.sequence_no, _next_sequence(session, voice_session.session_id)), - payload={"metadata": payload.metadata, "barge_in": payload.barge_in}, + payload={"metadata": payload_metadata, "barge_in": payload.barge_in}, ) session.commit() @@ -899,7 +975,7 @@ def _process_voice_ai_turn_sync( _apply_voice_start_metadata(voice_session, decision.metadata) voice_session.status = decision.status or ("handoff_requested" if decision.needs_handoff else "active") voice_session.updated_at = utc_now_iso() - if decision.reply_text: + if decision.reply_text and not bool(payload_metadata.get("runtime_defer_reply_planned")): _record_segment( session, voice_session=voice_session, @@ -1057,6 +1133,7 @@ _MEDIA_RUNTIME = AudioSocketMediaRuntime( set_state=_set_voice_session_state, get_pending_greeting=_load_pending_greeting_text, mark_reply_delivered=_mark_reply_delivered, + plan_reply=_record_runtime_reply_planned, process_turn=_media_process_turn, request_handoff=_request_runtime_handoff, handle_media_error=_handle_media_error, diff --git a/services/ai_voice_runtime_service/audiosocket.py b/services/ai_voice_runtime_service/audiosocket.py index 6c125b9..3beb919 100644 --- a/services/ai_voice_runtime_service/audiosocket.py +++ b/services/ai_voice_runtime_service/audiosocket.py @@ -158,3 +158,12 @@ class EnergyVAD: samples_per_ms = 8 bytes_per_sample = 2 return self._max_turn_ms * samples_per_ms * bytes_per_sample + + @property + def is_active(self) -> bool: + return self._active + + def snapshot_utterance_pcm(self) -> bytes: + if not self._active or not self._utterance_pcm: + return b"" + return bytes(self._utterance_pcm) diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index d56b96d..8ca9611 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio import contextlib import logging +import threading import time +import uuid from dataclasses import dataclass, field from typing import Any, Callable @@ -36,6 +38,13 @@ class MediaRegistration: ai_session_id: str | None = None language: str | None = None media_uuid: str | None = None + queue_code: str | None = None + queue_id: str | None = None + agent_profile: str | None = None + voice_v2_enabled: bool = False + voice_v2_ack_mode: str = "disabled" + voice_v2_streaming_tts: bool = False + voice_v2_partial_asr: bool = False @dataclass(slots=True) @@ -59,6 +68,15 @@ class MediaActor: last_outbound_audio_monotonic: float = 0.0 first_pcm_logged: bool = False keepalive_loop_logged: bool = False + early_ack_started: bool = False + partial_transcript: str | None = None + partial_intent: str | None = None + response_plan_id: str | None = None + playback_generation: int = 0 + utterance_generation: int = 0 + finalized_utterance_generation: int = 0 + partial_asr_task: asyncio.Task | None = None + partial_asr_attempted: bool = False class AudioSocketMediaRuntime: @@ -83,6 +101,7 @@ class AudioSocketMediaRuntime: set_state: Callable[[str, str, str | None, dict[str, Any] | None], None], get_pending_greeting: Callable[[str], str | None], mark_reply_delivered: Callable[[str, str, bool], None], + plan_reply: Callable[[str, str, dict[str, Any] | None, str], None], process_turn: Callable[[str, str, str | None, bool, dict[str, Any] | None], VoiceAITurnDecisionOut], request_handoff: Callable[[str, str, VoiceAITurnDecisionOut], None], handle_media_error: Callable[[str, str, dict[str, Any] | None], None], @@ -107,12 +126,165 @@ class AudioSocketMediaRuntime: self._set_state = set_state self._get_pending_greeting = get_pending_greeting self._mark_reply_delivered = mark_reply_delivered + self._plan_reply = plan_reply self._process_turn = process_turn self._request_handoff = request_handoff self._handle_media_error = handle_media_error self._server: asyncio.base_events.Server | None = None self._loop: asyncio.AbstractEventLoop | None = None self._actors: dict[str, MediaActor] = {} + self._v2_ack_wait_seconds = 0.18 + self._partial_asr_min_ms = 650 + + @staticmethod + def _normalize_intent_text(text: str) -> str: + return " ".join(str(text or "").strip().lower().split()) + + def _detect_early_intent(self, text: str) -> str: + normalized = self._normalize_intent_text(text) + if not normalized: + return "unknown" + if any(token in normalized for token in ("оператор", "оператором", "человеком", "менеджер", "сотрудник")): + return "operator_request" + if any(token in normalized for token in ("график", "распис", "время работы", "work schedule", "жұмыс")): + return "schedule" + if any(token in normalized for token in ("адрес", "филиал", "офис", "где вы", "мекен", "қайда")): + return "address" + if any(token in normalized for token in ("тариф", "цена", "стоимость", "сколько стоит", "баға")): + return "price" + if any(token in normalized for token in ("статус", "заявк", "заказ", "обращени", "өтінім")): + return "status" + if any(token in normalized for token in ("не работает", "ошибка", "проблем", "сломал", "істемей")): + return "problem" + return "unknown" + + @staticmethod + def _ack_kind_for_intent(intent: str) -> str: + if intent == "operator_request": + return "handoff" + if intent in {"schedule", "address", "price", "status", "problem"}: + return "understanding" + return "generic" + + @staticmethod + def _ack_text(language: str | None, ack_kind: str) -> str: + normalized = str(language or "").strip().lower() + if normalized == "kz": + if ack_kind == "handoff": + return "Бір сәт." + if ack_kind == "understanding": + return "Қазір айтып шығамын." + return "Қазір айтайын." + if ack_kind == "handoff": + return "Секунду." + if ack_kind == "understanding": + return "Сейчас сориентирую." + return "Сейчас подскажу." + + def _should_use_voice_v2(self, registration: MediaRegistration) -> bool: + return bool(registration.voice_v2_enabled and str(registration.voice_v2_ack_mode or "").strip() == "immediate_short") + + @property + def _partial_asr_min_bytes(self) -> int: + return self._partial_asr_min_ms * 16 + + @staticmethod + def _reset_live_turn_state(actor: MediaActor) -> None: + actor.utterance_generation += 1 + actor.finalized_utterance_generation = 0 + actor.early_ack_started = False + actor.partial_transcript = None + actor.partial_intent = None + actor.response_plan_id = None + actor.partial_asr_attempted = False + partial_task = actor.partial_asr_task + actor.partial_asr_task = None + if partial_task is not None and not partial_task.done(): + partial_task.cancel() + + async def _run_partial_asr_probe( + self, + actor: MediaActor, + *, + utterance_generation: int, + pcm_bytes: bytes, + ) -> None: + try: + wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) + transcription = await asyncio.to_thread( + lambda: self._asr_provider.transcribe_partial( + wav_bytes, + language_hint=actor.registration.language, + ) + ) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "audiosocket.partial_asr_failed session_id=%s generation=%s error=%s", + actor.registration.voice_session_id, + utterance_generation, + str(exc)[:500], + ) + return + + if actor.closed: + return + if utterance_generation != actor.utterance_generation: + return + if actor.finalized_utterance_generation >= utterance_generation: + return + + transcript_text = str(transcription.text or "").strip() + if not transcript_text: + return + actor.partial_transcript = transcript_text + actor.partial_intent = self._detect_early_intent(transcript_text) + logger.warning( + "audiosocket.partial_asr_ready session_id=%s generation=%s text_len=%s intent=%s", + actor.registration.voice_session_id, + utterance_generation, + len(transcript_text), + actor.partial_intent, + ) + + def _maybe_schedule_partial_asr(self, actor: MediaActor) -> None: + if actor.closed or not actor.registration.voice_v2_partial_asr: + return + if actor.partial_asr_attempted: + return + if not actor.vad.is_active: + return + snapshot = actor.vad.snapshot_utterance_pcm() + if len(snapshot) < self._partial_asr_min_bytes: + return + actor.partial_asr_attempted = True + utterance_generation = actor.utterance_generation + actor.partial_asr_task = asyncio.create_task( + self._run_partial_asr_probe( + actor, + utterance_generation=utterance_generation, + pcm_bytes=snapshot, + ) + ) + + async def _plan_reply_segment( + self, + actor: MediaActor, + text: str, + *, + kind: str, + metadata: dict[str, Any] | None = None, + ) -> None: + if actor.closed or not str(text or "").strip(): + return + await asyncio.to_thread( + self._plan_reply, + actor.registration.voice_session_id, + text, + metadata or {}, + kind, + ) @property def address(self) -> str: @@ -296,9 +468,12 @@ class AudioSocketMediaRuntime: return vad_result = actor.vad.feed(pcm_frame) + if vad_result.speech_started: + self._reset_live_turn_state(actor) if actor.state == "speaking" and vad_result.speech_started: actor.playback_interrupt.set() actor.barge_in_pending = True + self._maybe_schedule_partial_asr(actor) if vad_result.utterance_pcm: await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending)) actor.barge_in_pending = False @@ -327,34 +502,123 @@ class AudioSocketMediaRuntime: async def _process_utterance(self, actor: MediaActor, pcm_bytes: bytes, barge_in: bool) -> None: await self._set_actor_state(actor, "thinking") wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000) - transcription = await asyncio.to_thread( - lambda: self._asr_provider.transcribe( - wav_bytes, - language_hint=actor.registration.language, + actor.playback_generation += 1 + actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" + utterance_generation = actor.utterance_generation + partial_transcript = str(actor.partial_transcript or "").strip() + partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript) + full_asr_task = asyncio.create_task( + asyncio.to_thread( + lambda: self._asr_provider.transcribe( + wav_bytes, + language_hint=actor.registration.language, + ) ) ) - transcript_text = str(transcription.text or "").strip() + if self._should_use_voice_v2(actor.registration) and partial_transcript and not actor.early_ack_started: + ack_kind = self._ack_kind_for_intent(partial_intent or "unknown") + ack_text = self._ack_text(actor.registration.language, ack_kind) + actor.early_ack_started = True + base_metadata = { + "turn_duration_ms": int(len(pcm_bytes) / 16), + "media_uuid": actor.registration.media_uuid, + "queue_code": actor.registration.queue_code, + "voice_v2_enabled": actor.registration.voice_v2_enabled, + "response_plan_id": actor.response_plan_id, + "playback_generation": actor.playback_generation, + "partial_transcript": partial_transcript, + "early_intent": partial_intent, + } + await self._plan_reply_segment( + actor, + ack_text, + kind="ack", + metadata={ + **base_metadata, + "ack_kind": ack_kind, + "phase": "ack", + "partial_ack_source": "precomputed_partial_asr", + }, + ) + await self._speak_text(actor, ack_text, is_greeting=False) + if not actor.closed: + await self._set_actor_state(actor, "thinking") + + transcription = await full_asr_task + transcript_text = str(transcription.text or "").strip() or partial_transcript if not transcript_text: await self._set_actor_state(actor, "listening") return + actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation) + actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None + actor.partial_intent = self._detect_early_intent(transcript_text) metadata = { "turn_duration_ms": int(len(pcm_bytes) / 16), "media_uuid": actor.registration.media_uuid, + "queue_code": actor.registration.queue_code, + "voice_v2_enabled": actor.registration.voice_v2_enabled, + "response_plan_id": actor.response_plan_id, + "playback_generation": actor.playback_generation, + "partial_transcript": actor.partial_transcript, + "early_intent": actor.partial_intent, } - decision = await asyncio.to_thread( - self._process_turn, - actor.registration.voice_session_id, - transcript_text, - transcription.language or actor.registration.language, - barge_in, - metadata, + decision_task = asyncio.create_task( + asyncio.to_thread( + self._process_turn, + actor.registration.voice_session_id, + transcript_text, + transcription.language or actor.registration.language, + barge_in, + { + **metadata, + "runtime_defer_reply_planned": self._should_use_voice_v2(actor.registration), + }, + ) ) + 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: + ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown") + ack_text = self._ack_text(transcription.language or actor.registration.language, ack_kind) + actor.early_ack_started = True + await self._plan_reply_segment( + actor, + ack_text, + kind="ack", + metadata={ + **metadata, + "partial_transcript": actor.partial_transcript, + "early_intent": actor.partial_intent, + "ack_kind": ack_kind, + "phase": "ack", + }, + ) + await self._speak_text(actor, ack_text, is_greeting=False) + if not actor.closed: + await self._set_actor_state(actor, "thinking") + decision = await decision_task + else: + decision = await decision_task handoff_task: asyncio.Task | None = None if decision.needs_handoff: await self._set_actor_state(actor, "handoff_requested", decision.handoff_reason) handoff_task = self._start_handoff_request(actor, transcript_text, decision) if decision.reply_text: + if self._should_use_voice_v2(actor.registration): + await self._plan_reply_segment( + actor, + decision.reply_text, + kind="reply", + metadata={ + **metadata, + **(decision.metadata if isinstance(decision.metadata, dict) else {}), + "phase": "main", + "early_ack_started": actor.early_ack_started, + }, + ) await self._speak_text(actor, decision.reply_text, is_greeting=False) if actor.closed: return @@ -393,46 +657,86 @@ class AudioSocketMediaRuntime: actor.handoff_task = asyncio.create_task(_run()) return actor.handoff_task + async def _stream_tts_chunks(self, actor: MediaActor, text: str): + if not actor.registration.voice_v2_streaming_tts: + synthesis = await asyncio.to_thread( + lambda: self._tts_provider.synthesize( + text, + language=actor.registration.language, + ) + ) + if synthesis.audio_bytes: + yield synthesis + return + + loop = asyncio.get_running_loop() + queue: asyncio.Queue[Any] = asyncio.Queue() + done = object() + + def _producer() -> None: + try: + for synthesis in self._tts_provider.synthesize_chunks(text, language=actor.registration.language): + asyncio.run_coroutine_threadsafe(queue.put(synthesis), loop).result() + except Exception as exc: # pragma: no cover - defensive bridge from thread to loop + asyncio.run_coroutine_threadsafe(queue.put(exc), loop).result() + finally: + asyncio.run_coroutine_threadsafe(queue.put(done), loop).result() + + threading.Thread( + target=_producer, + name=f"tts-stream-{actor.registration.voice_session_id}", + daemon=True, + ).start() + while True: + item = await queue.get() + if item is done: + break + if isinstance(item, Exception): + raise item + yield item + async def _speak_text(self, actor: MediaActor, text: str, *, is_greeting: bool) -> None: if actor.closed or not text: return await self._set_actor_state(actor, "speaking") synth_started_at = time.monotonic() - synthesis = await asyncio.to_thread( - lambda: self._tts_provider.synthesize( - text, - language=actor.registration.language, - ) - ) - if not synthesis.audio_bytes: - raise RuntimeError("TTS provider returned empty audio") - logger.warning( - "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s", - actor.registration.voice_session_id, - is_greeting, - int((time.monotonic() - synth_started_at) * 1000), - len(synthesis.audio_bytes), - ) - - pcm_8k = resample_pcm16le( - synthesis.audio_bytes, - input_rate_hz=synthesis.sample_rate_hz, - output_rate_hz=8000, - ) first_frame_sent = False - for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes): - if actor.closed or actor.playback_interrupt.is_set(): - break - await self._write_audio_packet(actor, frame) + total_audio_bytes = 0 + async for synthesis in self._stream_tts_chunks(actor, text): + if not synthesis.audio_bytes: + continue + total_audio_bytes += len(synthesis.audio_bytes) if not first_frame_sent: - first_frame_sent = True logger.warning( - "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s", + "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s", actor.registration.voice_session_id, is_greeting, - len(frame), + int((time.monotonic() - synth_started_at) * 1000), + total_audio_bytes, ) - await asyncio.sleep(actor.frame_ms / 1000.0) + pcm_8k = resample_pcm16le( + synthesis.audio_bytes, + input_rate_hz=synthesis.sample_rate_hz, + output_rate_hz=8000, + ) + for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes): + if actor.closed or actor.playback_interrupt.is_set(): + break + await self._write_audio_packet(actor, frame) + if not first_frame_sent: + first_frame_sent = True + logger.warning( + "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s", + actor.registration.voice_session_id, + is_greeting, + len(frame), + ) + await asyncio.sleep(actor.frame_ms / 1000.0) + if actor.closed or actor.playback_interrupt.is_set(): + break + + if total_audio_bytes <= 0: + raise RuntimeError("TTS provider returned empty audio") interrupted = actor.playback_interrupt.is_set() actor.playback_interrupt.clear() @@ -529,6 +833,10 @@ class AudioSocketMediaRuntime: actor.handoff_task.cancel() with contextlib.suppress(asyncio.CancelledError, Exception): await actor.handoff_task + if actor.partial_asr_task is not None: + actor.partial_asr_task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await actor.partial_asr_task if actor.worker_task is not None: actor.worker_task.cancel() with contextlib.suppress(asyncio.CancelledError, Exception): diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 5094782..4351ab7 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -41,6 +41,9 @@ class ASRProvider: del audio_bytes return ASRTranscription(text="", language=language_hint, confidence=None) + def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + return self.transcribe(audio_bytes, language_hint=language_hint) + class OpenAIASRProvider(ASRProvider): name = "openai" @@ -77,6 +80,9 @@ class OpenAIASRProvider(ASRProvider): confidence=None, ) + def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + return self.transcribe(audio_bytes, language_hint=language_hint) + def build_asr_provider(name: str) -> ASRProvider: normalized = str(name or "stub").strip().lower() diff --git a/services/ai_voice_runtime_service/providers/tts.py b/services/ai_voice_runtime_service/providers/tts.py index ba47123..e3c16e1 100644 --- a/services/ai_voice_runtime_service/providers/tts.py +++ b/services/ai_voice_runtime_service/providers/tts.py @@ -154,6 +154,11 @@ class TTSProvider: del language return TTSSynthesis(text=text, audio_bytes=b"", sample_rate_hz=8000) + def synthesize_chunks(self, text: str, *, language: str | None = None): + synthesis = self.synthesize(text, language=language) + if synthesis.audio_bytes: + yield synthesis + class OpenAITTSProvider(TTSProvider): name = "openai" diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index 972343e..569a9fe 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -1300,6 +1300,62 @@ def test_voice_llm_guarded_decision_uses_operator_style_without_ai_or_kb(monkeyp assert "Do not say or imply that you are an AI" in system_prompt +def test_voice_v2_fast_conversational_adds_ack_metadata_and_compacts_reply(monkeypatch): + monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_fast_conversational") + monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible") + + def _fake_structured(messages): + del messages + return { + "language": "ru", + "intent": "kb_answer", + "reply_text": ( + "Сейчас сориентирую по графику работы филиала в Алматы. " + "Он работает с понедельника по пятницу с 9:00 до 18:00. " + "Если нужно, подскажу и по субботе." + ), + "confidence": 0.91, + "needs_handoff": False, + "handoff_reason": None, + "case_action": "keep_open", + "kb_refs": ["kba_voice_v2"], + "_model": "gpt-test", + "_latency_ms": 35, + "_finish_reason": "stop", + } + + monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured) + + decision = voice_module._voice_decision( + language="ru", + customer=SimpleNamespace(customer_id="cus_voice_v2", display_name="Ернор"), + interaction=SimpleNamespace(interaction_id="int_voice_v2", status="new", queue_id="que_voice", subject="hours"), + transcript_text="Хочу узнать график работы филиала в Алматы", + transcript_window=[ + SimpleNamespace( + speaker="caller", + text="Хочу узнать график работы филиала в Алматы", + sequence_no=1, + source_type="voice_asr", + barge_in_interrupted=False, + created_at=utc_now_iso(), + ) + ], + kb_results=[SimpleNamespace(article_id="kba_voice_v2", title="График", body="Будни 9:00-18:00")], + disclosure_required=False, + customer_name_value="Ернор", + customer_name_status="name_obtained", + request_metadata={"voice_v2_enabled": True, "response_plan_id": "rsp_test"}, + ) + + assert decision["intent"] == "kb_answer" + assert decision["metadata"]["voice_v2_enabled"] is True + assert decision["metadata"]["early_intent"] == "schedule" + assert decision["metadata"]["ack_kind"] == "understanding" + assert decision["metadata"]["response_plan_id"] == "rsp_test" + assert len(decision["reply_text"]) <= 180 + + 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 f8c32d6..ce9e7c3 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -107,6 +107,7 @@ def test_media_runtime_streams_greeting_and_turn(): set_state=lambda session_id, state, handoff_reason, metadata: states.append((session_id, state)), get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime" else None, mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)), + plan_reply=lambda session_id, text, metadata, kind: None, process_turn=lambda session_id, transcript_text, language, barge_in, metadata: ( turns.append((session_id, transcript_text, barge_in)) or VoiceAITurnDecisionOut( @@ -201,6 +202,7 @@ def test_media_runtime_waits_for_late_registration(): set_state=lambda session_id, state, handoff_reason, metadata: None, get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_late" else None, mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)), + 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="answer", @@ -303,6 +305,7 @@ def test_media_runtime_sends_keepalive_while_tts_is_slow(): set_state=lambda session_id, state, handoff_reason, metadata: None, get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_keepalive" else 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="answer", @@ -369,6 +372,7 @@ def test_media_runtime_sends_keepalive_before_registration_is_ready(): set_state=lambda session_id, state, handoff_reason, metadata: None, get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_prereg" else None, mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)), + 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="answer", @@ -453,6 +457,7 @@ def test_media_runtime_starts_handoff_before_handoff_tts_finishes(): 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="handoff_request", @@ -512,3 +517,208 @@ def test_media_runtime_starts_handoff_before_handoff_tts_finishes(): assert handoff_started.is_set() assert events.index("handoff") < events.index("speak_end") + + +def test_media_runtime_voice_v2_plays_short_ack_before_main_reply(): + planned: list[tuple[str, str, str, dict | None]] = [] + delivered: list[tuple[str, str, bool]] = [] + + class _ScheduleASRProvider(ASRProvider): + name = "schedule-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + del audio_bytes + return ASRTranscription(text="Хочу узнать график работы", 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=400, + asr_provider=_ScheduleASRProvider(), + 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: delivered.append((session_id, text, is_greeting)), + plan_reply=lambda session_id, text, metadata, kind: planned.append((session_id, text, kind, metadata)), + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: ( + time.sleep(0.25) + or 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", + metadata={"early_intent": "schedule", "ack_kind": "understanding"}, + ) + ), + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: None, + ) + + async def _fake_write_audio_packet(current_actor, pcm_frame: bytes) -> None: + del current_actor, pcm_frame + + runtime._write_audio_packet = _fake_write_audio_packet # 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", + call_id="call_media_runtime_v2", + interaction_id="int_media_runtime_v2", + ai_session_id="ais_media_runtime_v2", + 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=400), + frame_ms=20, + frame_bytes=320, + ) + await runtime._process_utterance(actor, pcm_frame, False) + + asyncio.run(_scenario()) + + assert [item[2] for item in planned] == ["ack", "reply"] + assert planned[0][1] == "Сейчас сориентирую." + assert planned[1][1].startswith("Подскажите, пожалуйста") + assert delivered == [ + ("avs_media_runtime_v2", "Сейчас сориентирую.", False), + ("avs_media_runtime_v2", "Подскажите, пожалуйста, какой город вас интересует?", False), + ] + + +def test_media_runtime_voice_v2_uses_partial_asr_to_start_ack_before_full_asr(): + timings: dict[str, float] = {} + speak_events: list[tuple[str, float]] = [] + + class _PartialAwareASRProvider(ASRProvider): + name = "partial-aware-asr" + + def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + timings["partial_ready"] = time.monotonic() + return ASRTranscription(text="work schedule", language=language_hint or "ru", confidence=0.8) + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + timings["full_started"] = time.monotonic() + time.sleep(0.35) + timings["full_finished"] = time.monotonic() + return ASRTranscription(text="work schedule", 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=_PartialAwareASRProvider(), + 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) -> None: + del current_actor, is_greeting + speak_events.append((text, time.monotonic())) + await asyncio.sleep(0) + + runtime._speak_text = _fake_speak_text # type: ignore[method-assign] + + async def _scenario() -> None: + actor = MediaActor( + registration=MediaRegistration( + voice_session_id="avs_media_runtime_v2_partial", + call_id="call_media_runtime_v2_partial", + interaction_id="int_media_runtime_v2_partial", + ai_session_id="ais_media_runtime_v2_partial", + 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, + ), + 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, + state="listening", + ) + speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 + silence_frame = b"\x00\x00" * 160 + for _ in range(35): + await runtime._handle_pcm(actor, speech_frame) + for _ in range(20): + if actor.partial_transcript: + break + await asyncio.sleep(0.01) + assert actor.partial_transcript == "work schedule" + for _ in range(2): + await runtime._handle_pcm(actor, silence_frame) + pcm_bytes, barge_in = await asyncio.wait_for(actor.turn_queue.get(), timeout=0.2) + await runtime._process_utterance(actor, pcm_bytes, barge_in) + + asyncio.run(_scenario()) + + assert timings["partial_ready"] < timings["full_finished"] + assert speak_events + assert speak_events[0][0] == runtime._ack_text("ru", "understanding") + assert speak_events[0][1] < timings["full_finished"]