diff --git a/.env.example b/.env.example index 3b104b5..cf39859 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,11 @@ 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_DUPLEX_ENABLED=1 +AI_VOICE_V2_STREAMING_ASR_BACKEND=local_sidecar +AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://127.0.0.1:8021 +AI_VOICE_V2_PREBAKED_ACK_ENABLED=1 +AI_VOICE_V2_PREBAKED_ACK_DIR=.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS=1 AI_VOICE_V2_PARTIAL_ASR=1 AI_VOICE_V2_EMOTIVE_ACK_ENABLED=1 diff --git a/.env.production.template b/.env.production.template index 0b09249..159ac5b 100644 --- a/.env.production.template +++ b/.env.production.template @@ -52,6 +52,11 @@ 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_DUPLEX_ENABLED=1 +AI_VOICE_V2_STREAMING_ASR_BACKEND=local_sidecar +AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://127.0.0.1:8021 +AI_VOICE_V2_PREBAKED_ACK_ENABLED=1 +AI_VOICE_V2_PREBAKED_ACK_DIR=/app/.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS=1 AI_VOICE_V2_PARTIAL_ASR=1 AI_VOICE_V2_EMOTIVE_ACK_ENABLED=1 diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index 2b49d3a..b83d410 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -34,6 +34,11 @@ x-app-env: &app_env AI_VOICE_V2_ENABLED: "1" AI_VOICE_V2_QUEUE_CODES: voice_lab_ai AI_VOICE_V2_ACK_MODE: immediate_short + AI_VOICE_V2_DUPLEX_ENABLED: "1" + AI_VOICE_V2_STREAMING_ASR_BACKEND: local_sidecar + AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://127.0.0.1:8021 + AI_VOICE_V2_PREBAKED_ACK_ENABLED: "1" + AI_VOICE_V2_PREBAKED_ACK_DIR: /app/.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS: "1" AI_VOICE_V2_PARTIAL_ASR: "1" AI_VOICE_V2_EMOTIVE_ACK_ENABLED: "1" diff --git a/scripts/local_stack.py b/scripts/local_stack.py index c8271b5..948aa31 100644 --- a/scripts/local_stack.py +++ b/scripts/local_stack.py @@ -314,6 +314,11 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ 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_DUPLEX_ENABLED"] = env.get("AI_VOICE_V2_DUPLEX_ENABLED", "1") + env["AI_VOICE_V2_STREAMING_ASR_BACKEND"] = env.get("AI_VOICE_V2_STREAMING_ASR_BACKEND", "local_sidecar") + env["AI_VOICE_V2_STREAMING_ASR_BASE_URL"] = env.get("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021") + env["AI_VOICE_V2_PREBAKED_ACK_ENABLED"] = env.get("AI_VOICE_V2_PREBAKED_ACK_ENABLED", "1") + env["AI_VOICE_V2_PREBAKED_ACK_DIR"] = env.get("AI_VOICE_V2_PREBAKED_ACK_DIR", str(DATA_DIR / "voice_v2_ack_bank")) 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_V2_EMOTIVE_ACK_ENABLED"] = env.get("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", "1") diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index 8df1e71..7c31475 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -952,6 +952,10 @@ def _voice_is_off_domain_request(text: str | None) -> bool: broad_markers = ( "ядерн", "реактор", + "кондиционер", + "компрессор", + "испарител", + "конденсатор", "космос", "планет", "математ", @@ -1151,11 +1155,16 @@ def _voice_policy_mode() -> str: return persona.voice_policy_mode() +def _voice_reply_phase(metadata: dict[str, Any] | None = None) -> str: + payload = metadata if isinstance(metadata, dict) else {} + return str(payload.get("reply_phase") or "final").strip().lower() or "final" + + 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" + return _voice_policy_mode() in {"v2_fast_conversational", "v2_streaming_duplex"} def _voice_early_intent_bucket(text: str) -> str: @@ -1223,6 +1232,70 @@ def _voice_v2_metadata( return metadata +def _voice_early_plan( + *, + language: str, + transcript_text: str, + request_metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + v2_metadata = _voice_v2_metadata(transcript_text, request_metadata) + lower_text = str(transcript_text or "").strip().lower() + if _app()._looks_like_human_request(lower_text) or _app()._is_sensitive_request(lower_text): + return { + "language": language, + "intent": "handoff_request", + "reply_text": _voice_handoff_reply(language), + "confidence": 0.25, + "needs_handoff": True, + "handoff_reason": "Запрос требует участия живого оператора.", + "case_action": "keep_open", + "kb_refs": [], + "summary_text": "AI заранее распознал необходимость подключить оператора.", + "model": "voice_early_plan", + "latency_ms": 1, + "metadata": { + **v2_metadata, + "reply_phase": "early_plan", + }, + } + if _voice_is_off_domain_request(transcript_text): + reply_text, summary_text = _voice_off_domain_reply(language) + return { + "language": language, + "intent": "clarification", + "reply_text": _voice_compact_reply_text(reply_text, language=language), + "confidence": 0.6, + "needs_handoff": False, + "handoff_reason": None, + "case_action": "keep_open", + "kb_refs": [], + "summary_text": summary_text, + "model": "voice_early_plan_off_domain", + "latency_ms": 1, + "metadata": { + **v2_metadata, + "reply_phase": "early_plan", + }, + } + return { + "language": language, + "intent": "clarification", + "reply_text": "", + "confidence": 0.45, + "needs_handoff": False, + "handoff_reason": None, + "case_action": "keep_open", + "kb_refs": [], + "summary_text": "Early plan is prepared.", + "model": "voice_early_plan", + "latency_ms": 1, + "metadata": { + **v2_metadata, + "reply_phase": "early_plan", + }, + } + + def _voice_llm_prompt_messages( *, language: str, @@ -1318,7 +1391,7 @@ def _voice_llm_decision( name_status: str | None, ) -> dict[str, Any] | None: app = _app() - if _voice_policy_mode() not in {"llm_guarded", "v2_fast_conversational"}: + if _voice_policy_mode() not in {"llm_guarded", "v2_fast_conversational", "v2_streaming_duplex"}: return None if app._ai_provider() != "openai_compatible": return None @@ -1556,6 +1629,14 @@ def _voice_decision( caller_texts = _voice_recent_caller_texts(transcript_window) model = app._ai_model() v2_metadata = _voice_v2_metadata(transcript_text, request_metadata) + reply_phase = _voice_reply_phase(request_metadata) + + if reply_phase == "early_plan": + return _voice_early_plan( + language=language, + transcript_text=transcript_text, + request_metadata=request_metadata, + ) if app._looks_like_human_request(lower_text) or app._is_sensitive_request(lower_text): return { @@ -1634,6 +1715,7 @@ def _voice_decision( llm_decision["metadata"] = { **(llm_decision.get("metadata") or {}), **v2_metadata, + "reply_phase": "final", } return llm_decision @@ -1659,7 +1741,7 @@ def _voice_decision( } if v2_metadata: decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language) - decision["metadata"] = v2_metadata + decision["metadata"] = {**v2_metadata, "reply_phase": "final"} return decision reply_text = _voice_confusion_prompt(language, caller_texts) if caller_confused else ( @@ -1680,7 +1762,7 @@ def _voice_decision( } if v2_metadata: decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language) - decision["metadata"] = v2_metadata + decision["metadata"] = {**v2_metadata, "reply_phase": "final"} return decision @@ -2008,42 +2090,47 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe agent_profile=voice_session.agent_profile, ) now = utc_now_iso() + request_metadata = payload.metadata if isinstance(payload.metadata, dict) else {} + early_plan_only = _voice_reply_phase(request_metadata) == "early_plan" voice_session.customer_id = customer_id - voice_session.last_user_utterance_at = now - voice_session.status = "thinking" + if not early_plan_only: + voice_session.last_user_utterance_at = now + voice_session.status = "thinking" voice_session.updated_at = now - ai_session.updated_at = now + if not early_plan_only: + ai_session.updated_at = now ai_session.call_id = payload.call_id ai_session.interaction_id = interaction.interaction_id ai_session.customer_id = customer_id ai_session.language = str(payload.language or voice_session.language or "ru").strip() or "ru" config = load_effective_voice_name_collection_config(session) - _record_voice_ai_turn( - session, - ai_session_id=ai_session.session_id, - interaction_id=interaction.interaction_id, - role="user", - source_type="voice_asr", - text=payload.transcript_text, - payload={ - "voice_session_id": payload.voice_session_id, - "call_id": payload.call_id, - "sequence_no": payload.sequence_no, - "barge_in": payload.barge_in, - "metadata": payload.metadata, - }, - ) - if payload.barge_in: - last_assistant_segment = session.execute( - select(VoiceTranscriptSegmentRow) - .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) - .where(VoiceTranscriptSegmentRow.speaker == "assistant") - .order_by(VoiceTranscriptSegmentRow.id.desc()) - .limit(1) - ).scalar_one_or_none() - if last_assistant_segment: - last_assistant_segment.barge_in_interrupted = True + if not early_plan_only: + _record_voice_ai_turn( + session, + ai_session_id=ai_session.session_id, + interaction_id=interaction.interaction_id, + role="user", + source_type="voice_asr", + text=payload.transcript_text, + payload={ + "voice_session_id": payload.voice_session_id, + "call_id": payload.call_id, + "sequence_no": payload.sequence_no, + "barge_in": payload.barge_in, + "metadata": payload.metadata, + }, + ) + if payload.barge_in: + last_assistant_segment = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) + .where(VoiceTranscriptSegmentRow.speaker == "assistant") + .order_by(VoiceTranscriptSegmentRow.id.desc()) + .limit(1) + ).scalar_one_or_none() + if last_assistant_segment: + last_assistant_segment.barge_in_interrupted = True transcript_window = _voice_recent_segments( session, voice_session_id=voice_session.session_id, @@ -2134,34 +2221,55 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe ) current_name_source = str(voice_session.customer_name_source or "none").strip() or "none" current_name_resolved_at = str(voice_session.customer_name_resolved_at or "").strip() or None - name_update = _voice_downstream_name_update( - language=ai_session.language or "ru", - transcript_text=payload.transcript_text, - transcript_window=transcript_window, - current_status=current_name_status, - current_name=current_name_value, - current_source=current_name_source, - current_resolved_at=current_name_resolved_at, - now=now, - config=config, - ) - if name_update["finalizable"]: - finalized_name = _finalize_customer_name( + effective_name_status = current_name_status + effective_name_value = current_name_value + effective_name_source = current_name_source + effective_name_resolved_at = current_name_resolved_at + inline_name_followup = False + if not early_plan_only: + name_update = _voice_downstream_name_update( + language=ai_session.language or "ru", + transcript_text=payload.transcript_text, + transcript_window=transcript_window, + current_status=current_name_status, + current_name=current_name_value, + current_source=current_name_source, + current_resolved_at=current_name_resolved_at, + now=now, + config=config, + ) + if name_update["finalizable"]: + finalized_name = _finalize_customer_name( + session, + customer=customer, + customer_id=customer_id, + call_id=payload.call_id, + final_name=name_update["value"], + resolved_at=now, + ) + if finalized_name: + name_update["value"] = finalized_name + name_update["resolved_at"] = now + effective_name_status = name_update["status"] + effective_name_value = name_update["value"] + effective_name_source = name_update["source"] + effective_name_resolved_at = name_update["resolved_at"] + inline_name_followup = bool(name_update["inline_followup"]) + _persist_voice_name_state( session, - customer=customer, - customer_id=customer_id, - call_id=payload.call_id, - final_name=name_update["value"], - resolved_at=now, + voice_session=voice_session, + status=effective_name_status, + value=effective_name_value, + source=effective_name_source, + resolved_at=effective_name_resolved_at, + ) + kb_results = [] + if not early_plan_only: + kb_results = app._kb_search( + session, + payload.transcript_text, + language=ai_session.language, ) - if finalized_name: - name_update["value"] = finalized_name - name_update["resolved_at"] = now - kb_results = app._kb_search( - session, - payload.transcript_text, - language=ai_session.language, - ) disclosure_required = voice_session.disclosure_played_at is None decision = _voice_decision( language=ai_session.language or "ru", @@ -2171,54 +2279,52 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe transcript_window=transcript_window, kb_results=kb_results, 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 {}, + customer_name_value=effective_name_value, + customer_name_status=effective_name_status, + request_metadata=request_metadata, ) - if decision.get("extracted_name"): - name_update["status"] = "name_obtained" - name_update["value"] = str(decision["extracted_name"]).strip() - name_update["source"] = "llm_extraction" - name_update["resolved_at"] = now - name_update["inline_followup"] = False - + if decision.get("extracted_name") and not early_plan_only: + effective_name_status = "name_obtained" + effective_name_value = str(decision["extracted_name"]).strip() + effective_name_source = "llm_extraction" + effective_name_resolved_at = now + inline_name_followup = False finalized_name = _finalize_customer_name( session, customer=customer, customer_id=customer_id, call_id=payload.call_id, - final_name=name_update["value"], + final_name=effective_name_value, resolved_at=now, ) if finalized_name: - name_update["value"] = finalized_name - - _persist_voice_name_state( - session, - voice_session=voice_session, - status=name_update["status"], - value=name_update["value"], - source=name_update["source"], - resolved_at=name_update["resolved_at"], - ) + effective_name_value = finalized_name + _persist_voice_name_state( + session, + voice_session=voice_session, + status=effective_name_status, + value=effective_name_value, + source=effective_name_source, + resolved_at=effective_name_resolved_at, + ) decision_metadata = _voice_name_metadata( language=voice_session.voice_start_language or ai_session.language or "ru", customer_id=customer_id, - status=name_update["status"], - value=name_update["value"], - source=name_update["source"], - resolved_at=name_update["resolved_at"], + status=effective_name_status, + value=effective_name_value, + source=effective_name_source, + resolved_at=effective_name_resolved_at, ) decision_metadata.update(decision.get("metadata") or {}) - if name_update["status"] == "name_obtained" and name_update["value"]: + if effective_name_status == "name_obtained" and effective_name_value: decision["reply_text"] = _voice_reply_with_name( decision["language"], decision["reply_text"], - name_update["value"], + effective_name_value, ) - elif name_update["inline_followup"] and not decision["needs_handoff"]: + elif inline_name_followup and not decision["needs_handoff"] and not early_plan_only: inline_followup = _voice_inline_name_followup(decision["language"], config) decision["reply_text"] = ( f"{decision['reply_text']} {inline_followup}".strip() @@ -2231,6 +2337,24 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe language=decision["language"], ) decision["metadata"] = decision_metadata + if early_plan_only: + decision_status = str(voice_session.status or "active").strip() or "active" + session.rollback() + return VoiceAITurnDecisionOut( + language=decision["language"], + intent=decision["intent"], + reply_text=decision["reply_text"], + confidence=decision["confidence"], + needs_handoff=decision["needs_handoff"], + handoff_reason=decision["handoff_reason"], + case_action=decision["case_action"], + kb_refs=decision["kb_refs"], + summary_text=decision["summary_text"], + model=decision["model"], + latency_ms=decision["latency_ms"], + status=decision_status, + metadata=decision_metadata, + ) _record_voice_ai_turn( session, ai_session_id=ai_session.session_id, diff --git a/services/ai_voice_runtime_service/ack_bank.py b/services/ai_voice_runtime_service/ack_bank.py new file mode 100644 index 0000000..3c6d5ce --- /dev/null +++ b/services/ai_voice_runtime_service/ack_bank.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import contextlib +import hashlib +import json +import os +import tempfile +from dataclasses import dataclass +from pathlib import Path +from threading import Lock + +from services.ai_voice_runtime_service.audiosocket import resample_pcm16le +from services.ai_voice_runtime_service.providers.tts import TTSProvider + + +def _ack_bank_dir() -> Path: + explicit = str(os.getenv("AI_VOICE_V2_PREBAKED_ACK_DIR", "") or "").strip() + if explicit: + return Path(explicit).expanduser() + data_dir = str(os.getenv("CC_DATA_DIR", "") or "").strip() + if data_dir: + return Path(data_dir).expanduser() / "voice_ack_bank" + local_data_dir = Path(".data_local") + if local_data_dir.exists(): + return local_data_dir / "voice_ack_bank" + return Path(".data") / "voice_ack_bank" + + +@dataclass(slots=True) +class AckClip: + text: str + pcm_8k_bytes: bytes + source: str + sample_rate_hz: int = 8000 + + +class PrebakedAckBank: + def __init__(self, *, tts_provider: TTSProvider, ack_dir: Path | None = None) -> None: + self._tts_provider = tts_provider + self._ack_dir = ack_dir or _ack_bank_dir() + self._lock = Lock() + + def _cache_key( + self, + *, + text: str, + language: str | None, + style_hints: dict[str, object] | None, + ) -> str: + payload = { + "provider": getattr(self._tts_provider, "name", "tts"), + "language": str(language or "").strip() or None, + "style_hints": style_hints or {}, + "text": text, + } + encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + def _paths(self, cache_key: str) -> tuple[Path, Path]: + prefix = self._ack_dir / cache_key[:2] / cache_key[2:4] + return prefix / f"{cache_key}.pcm", prefix / f"{cache_key}.json" + + def _load(self, cache_key: str, *, text: str) -> AckClip | None: + pcm_path, meta_path = self._paths(cache_key) + if not pcm_path.exists(): + return None + try: + pcm_bytes = pcm_path.read_bytes() + source = "prebaked_cache" + if meta_path.exists(): + metadata = json.loads(meta_path.read_text(encoding="utf-8")) + source = str(metadata.get("source") or source) + except (OSError, ValueError, TypeError, json.JSONDecodeError): + return None + if not pcm_bytes: + return None + return AckClip(text=text, pcm_8k_bytes=pcm_bytes, source=source) + + def _store(self, cache_key: str, *, clip: AckClip, language: str | None, style_hints: dict[str, object] | None) -> None: + pcm_path, meta_path = self._paths(cache_key) + pcm_path.parent.mkdir(parents=True, exist_ok=True) + metadata = { + "provider": getattr(self._tts_provider, "name", "tts"), + "language": str(language or "").strip() or None, + "style_hints": style_hints or {}, + "text": clip.text, + "sample_rate_hz": clip.sample_rate_hz, + "source": clip.source, + } + pcm_tmp: str | None = None + meta_tmp: str | None = None + try: + with tempfile.NamedTemporaryFile(dir=pcm_path.parent, delete=False, suffix=".pcm.tmp") as handle: + handle.write(clip.pcm_8k_bytes) + pcm_tmp = handle.name + with tempfile.NamedTemporaryFile(dir=meta_path.parent, delete=False, suffix=".json.tmp", mode="w", encoding="utf-8") as handle: + json.dump(metadata, handle, ensure_ascii=False, sort_keys=True) + meta_tmp = handle.name + os.replace(pcm_tmp, pcm_path) + os.replace(meta_tmp, meta_path) + finally: + for path in (pcm_tmp, meta_tmp): + if path and os.path.exists(path): + with contextlib.suppress(OSError): + os.unlink(path) + + def get_clip( + self, + *, + text: str, + language: str | None, + style_hints: dict[str, object] | None = None, + ) -> AckClip: + normalized_text = str(text or "").strip() + if not normalized_text: + return AckClip(text="", pcm_8k_bytes=b"", source="empty") + cache_key = self._cache_key(text=normalized_text, language=language, style_hints=style_hints) + with self._lock: + cached = self._load(cache_key, text=normalized_text) + if cached is not None: + return cached + synthesis = self._tts_provider.synthesize( + normalized_text, + language=language, + style_hints=style_hints, + ) + pcm_8k = resample_pcm16le( + synthesis.audio_bytes, + input_rate_hz=synthesis.sample_rate_hz, + output_rate_hz=8000, + ) + clip = AckClip( + text=normalized_text, + pcm_8k_bytes=pcm_8k, + source="prebaked_materialized", + ) + self._store(cache_key, clip=clip, language=language, style_hints=style_hints) + return clip diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index 38cc760..329689e 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -16,7 +16,7 @@ from sqlalchemy.exc import OperationalError from services.ai_voice_runtime_service.audiosocket import normalize_media_uuid from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaRegistration -from services.ai_voice_runtime_service.providers.asr import build_asr_provider +from services.ai_voice_runtime_service.providers.asr import build_asr_provider, build_streaming_asr_provider from services.ai_voice_runtime_service.runtime_tts_provider import RuntimeConfiguredTTSProvider from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import get_session @@ -153,6 +153,18 @@ def _voice_v2_partial_asr_enabled() -> bool: return _bool_env("AI_VOICE_V2_PARTIAL_ASR", True) +def _voice_v2_duplex_enabled() -> bool: + return _bool_env("AI_VOICE_V2_DUPLEX_ENABLED", True) + + +def _voice_v2_streaming_asr_backend() -> str: + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_BACKEND", "local_sidecar") or "local_sidecar").strip().lower() + + +def _voice_v2_prebaked_ack_enabled() -> bool: + return _bool_env("AI_VOICE_V2_PREBAKED_ACK_ENABLED", True) + + def _voice_v2_emotive_ack_enabled() -> bool: return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", True) @@ -249,6 +261,7 @@ def _default_voice_greeting(language: str | None, *, agent_profile: str = "voice _ASR_PROVIDER = build_asr_provider(_asr_provider_name()) +_STREAMING_ASR_PROVIDER = build_streaming_asr_provider(_voice_v2_streaming_asr_backend()) _TTS_PROVIDER = RuntimeConfiguredTTSProvider(default_provider_name=_tts_provider_name()) @@ -495,6 +508,7 @@ def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | No voice_v2_for_session = bool( _voice_v2_enabled() and normalized_queue_code and normalized_queue_code in _voice_v2_queue_codes() ) + streaming_backend = _voice_v2_streaming_asr_backend() if voice_v2_for_session else None return MediaRegistration( voice_session_id=row.session_id, call_id=row.call_id, @@ -509,6 +523,9 @@ def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | No 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()), + voice_v2_duplex=bool(voice_v2_for_session and _voice_v2_duplex_enabled()), + voice_v2_streaming_asr_backend=streaming_backend, + voice_v2_prebaked_ack=bool(voice_v2_for_session and _voice_v2_prebaked_ack_enabled()), voice_v2_emotive_ack=bool(voice_v2_for_session and _voice_v2_emotive_ack_enabled()), voice_v2_emotive_ack_ru_only=bool(_voice_v2_emotive_ack_ru_only()), ) @@ -641,46 +658,139 @@ def _load_pending_greeting_text(session_id: str) -> str | None: session.close() -def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool) -> None: +def _update_reply_delivery_status( + session, + *, + session_id: str, + text: str, + is_greeting: bool, + phase: str | None, + status: str, +) -> VoiceAISessionRow: + voice_session = _load_voice_session(session, session_id) + now = utc_now_iso() + pending_rows = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == session_id) + .where(VoiceTranscriptSegmentRow.speaker == "assistant") + .order_by(VoiceTranscriptSegmentRow.id.desc()) + ).scalars().all() + matched_row = None + normalized_text = str(text or "").strip() + expected_phase = str(phase or "").strip() or None + for row in pending_rows: + try: + payload = json.loads(row.payload_json or "{}") + except Exception: + payload = {} + row_kind = str(payload.get("kind") or "").strip() + row_phase = str(payload.get("metadata", {}).get("phase") or "").strip() or None + if is_greeting: + if row_kind != "greeting": + continue + elif str(row.text or "").strip() != normalized_text: + continue + if expected_phase and row_phase not in {expected_phase, None}: + continue + matched_row = row + payload["delivery_status"] = status + row.payload_json = json.dumps(payload, ensure_ascii=False) + break + if matched_row is not None: + matched_row.is_final = status == "delivered" + if status in {"interrupted", "discarded"}: + matched_row.barge_in_interrupted = status == "interrupted" + if is_greeting and status == "delivered" and not voice_session.disclosure_played_at: + voice_session.disclosure_played_at = now + if status == "delivered": + voice_session.last_ai_reply_at = now + voice_session.updated_at = now + return voice_session + + +def _mark_reply_started(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None: + def _write() -> None: + session = get_session() + try: + _update_reply_delivery_status( + session, + session_id=session_id, + text=text, + is_greeting=is_greeting, + phase=phase, + status="started", + ) + session.commit() + finally: + session.close() + + _retry_db_write(_write) + + +def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None: + def _write() -> None: + session = get_session() + try: + _update_reply_delivery_status( + session, + text=text, + session_id=session_id, + is_greeting=is_greeting, + phase=phase, + status="delivered", + ) + session.commit() + finally: + session.close() + + _retry_db_write(_write) + + +def _record_latency_metric(session_id: str, metric: str, latency_ms: int) -> None: + normalized_metric = str(metric or "").strip() + if not normalized_metric: + return + def _write() -> None: session = get_session() try: voice_session = _load_voice_session(session, session_id) - now = utc_now_iso() - pending_rows = session.execute( - select(VoiceTranscriptSegmentRow) - .where(VoiceTranscriptSegmentRow.session_id == session_id) - .where(VoiceTranscriptSegmentRow.speaker == "assistant") - .where(VoiceTranscriptSegmentRow.is_final.is_(False)) - .order_by(VoiceTranscriptSegmentRow.id.desc()) - ).scalars().all() - matched_row = None - normalized_text = str(text or "").strip() - for row in pending_rows: - try: - payload = json.loads(row.payload_json or "{}") - except Exception: - payload = {} - row_kind = str(payload.get("kind") or "").strip() - if is_greeting: - if row_kind == "greeting": - matched_row = row - payload["delivery_status"] = "delivered" - row.payload_json = json.dumps(payload, ensure_ascii=False) - break - continue - if str(row.text or "").strip() != normalized_text: - continue - matched_row = row - payload["delivery_status"] = "delivered" - row.payload_json = json.dumps(payload, ensure_ascii=False) - break - if matched_row is not None: - matched_row.is_final = True - if is_greeting and not voice_session.disclosure_played_at: - voice_session.disclosure_played_at = now - voice_session.last_ai_reply_at = now - voice_session.updated_at = now + _record_segment( + session, + voice_session=voice_session, + speaker="system", + source_type="runtime", + text=normalized_metric, + sequence_no=_next_sequence(session, voice_session.session_id), + payload={ + "kind": "latency_metric", + "metric": normalized_metric, + "latency_ms": int(latency_ms), + }, + is_final=True, + ) + voice_session.updated_at = utc_now_iso() + session.commit() + finally: + session.close() + + _retry_db_write(_write) + + +def _mark_reply_discarded(session_id: str, text: str, phase: str | None = None, status: str = "discarded") -> None: + normalized_status = str(status or "").strip().lower() or "discarded" + + def _write() -> None: + session = get_session() + try: + _update_reply_delivery_status( + session, + session_id=session_id, + text=text, + is_greeting=False, + phase=phase, + status=normalized_status, + ) session.commit() finally: session.close() @@ -956,26 +1066,29 @@ def _process_voice_ai_turn_sync( session = get_session() voice_session = None payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {} + reply_phase = str(payload_metadata.get("reply_phase") or "").strip().lower() + early_plan_only = reply_phase == "early_plan" try: voice_session = _load_voice_session(session, session_id) - if voice_session.status in {"human_owned", "completed", "error"}: + if not early_plan_only and voice_session.status in {"human_owned", "completed", "error"}: raise HTTPException(status_code=409, detail="Voice AI session is no longer active") now = utc_now_iso() - voice_session.status = "thinking" - voice_session.last_user_utterance_at = now - voice_session.updated_at = now - if payload.barge_in: - _mark_last_assistant_segment_interrupted(session, voice_session.session_id) - _record_segment( - session, - voice_session=voice_session, - speaker="caller", - 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}, - ) - session.commit() + if not early_plan_only: + voice_session.status = "thinking" + voice_session.last_user_utterance_at = now + voice_session.updated_at = now + if payload.barge_in: + _mark_last_assistant_segment_interrupted(session, voice_session.session_id) + _record_segment( + session, + voice_session=voice_session, + speaker="caller", + 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}, + ) + session.commit() decision_payload = _orchestrator_request( "POST", @@ -984,30 +1097,31 @@ def _process_voice_ai_turn_sync( timeout=12.0, ) decision = VoiceAITurnDecisionOut.model_validate(decision_payload) - voice_session.language = decision.language or voice_session.language - voice_session.handoff_reason = decision.handoff_reason - _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 and not bool(payload_metadata.get("runtime_defer_reply_planned")): - _record_segment( - session, - voice_session=voice_session, - speaker="assistant", - source_type="tts", - text=decision.reply_text, - sequence_no=_next_sequence(session, voice_session.session_id), - payload={ - "intent": decision.intent, - "kb_refs": decision.kb_refs, - "model": decision.model, - "delivery_status": "planned", - "metadata": decision.metadata, - }, - is_final=False, - ) - session.commit() - if auto_handoff and decision.needs_handoff: + if not early_plan_only: + voice_session.language = decision.language or voice_session.language + voice_session.handoff_reason = decision.handoff_reason + _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 and not bool(payload_metadata.get("runtime_defer_reply_planned")): + _record_segment( + session, + voice_session=voice_session, + speaker="assistant", + source_type="tts", + text=decision.reply_text, + sequence_no=_next_sequence(session, voice_session.session_id), + payload={ + "intent": decision.intent, + "kb_refs": decision.kb_refs, + "model": decision.model, + "delivery_status": "planned", + "metadata": decision.metadata, + }, + is_final=False, + ) + session.commit() + if auto_handoff and decision.needs_handoff and not early_plan_only: _request_runtime_handoff(session_id, payload.transcript_text, decision) return decision except HTTPException: @@ -1139,6 +1253,7 @@ _MEDIA_RUNTIME = AudioSocketMediaRuntime( trailing_silence_ms=_vad_trailing_silence_ms(), max_turn_ms=_turn_max_ms(), asr_provider=_ASR_PROVIDER, + streaming_asr_provider=_STREAMING_ASR_PROVIDER, tts_provider=_TTS_PROVIDER, load_registration_by_media_uuid=_load_media_registration_by_uuid, mark_media_connected=_mark_media_connected, @@ -1146,8 +1261,11 @@ _MEDIA_RUNTIME = AudioSocketMediaRuntime( touch_media_frame=_touch_media_frame, set_state=_set_voice_session_state, get_pending_greeting=_load_pending_greeting_text, + mark_reply_started=_mark_reply_started, mark_reply_delivered=_mark_reply_delivered, + mark_reply_discarded=_mark_reply_discarded, plan_reply=_record_runtime_reply_planned, + record_latency=_record_latency_metric, process_turn=_media_process_turn, request_handoff=_request_runtime_handoff, handle_media_error=_handle_media_error, diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 5bd0936..777ace8 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import audioop import contextlib import hashlib import logging @@ -10,6 +11,7 @@ import uuid from dataclasses import dataclass, field from typing import Any, Callable +from services.ai_voice_runtime_service.ack_bank import PrebakedAckBank from services.ai_voice_runtime_service.audiosocket import ( AUDIO_SOCKET_PACKET_DTMF, AUDIO_SOCKET_PACKET_HANGUP, @@ -23,7 +25,13 @@ from services.ai_voice_runtime_service.audiosocket import ( read_packet, resample_pcm16le, ) -from services.ai_voice_runtime_service.providers.asr import ASRProvider +from services.ai_voice_runtime_service.providers.asr import ( + ASRTranscription, + ASRProvider, + StreamingASRPartial, + StreamingASRProvider, + StreamingASRUnavailable, +) from services.ai_voice_runtime_service.providers.tts import TTSProvider from services.shared.models import VoiceAITurnDecisionOut @@ -46,6 +54,9 @@ class MediaRegistration: voice_v2_ack_mode: str = "disabled" voice_v2_streaming_tts: bool = False voice_v2_partial_asr: bool = False + voice_v2_duplex: bool = False + voice_v2_streaming_asr_backend: str | None = None + voice_v2_prebaked_ack: bool = False voice_v2_emotive_ack: bool = False voice_v2_emotive_ack_ru_only: bool = True @@ -71,17 +82,31 @@ class MediaActor: last_outbound_audio_monotonic: float = 0.0 first_pcm_logged: bool = False keepalive_loop_logged: bool = False + input_active: bool = False + playback_active: bool = False early_ack_started: bool = False partial_transcript: str | None = None partial_intent: str | None = None + partial_intent_streak: int = 0 + stable_partial_intent: str | None = None response_plan_id: str | None = None playback_generation: int = 0 + tts_generation: int = 0 utterance_generation: int = 0 finalized_utterance_generation: int = 0 partial_asr_task: asyncio.Task | None = None partial_asr_attempted: bool = False + asr_stream_id: str | None = None + asr_streaming_enabled: bool = False + asr_poll_due_monotonic: float = 0.0 + barge_in_speech_ms: int = 0 + barge_in_detected_monotonic: float = 0.0 + speech_started_monotonic: float = 0.0 + speech_ended_monotonic: float = 0.0 last_ack_text: str | None = None last_ack_completed_monotonic: float = 0.0 + last_ack_variant: str | None = None + current_reply_phase: str | None = None class AudioSocketMediaRuntime: @@ -98,6 +123,7 @@ class AudioSocketMediaRuntime: trailing_silence_ms: int, max_turn_ms: int, asr_provider: ASRProvider, + streaming_asr_provider: StreamingASRProvider | None = None, tts_provider: TTSProvider, load_registration_by_media_uuid: Callable[[str], MediaRegistration | None], mark_media_connected: Callable[[str, str], None], @@ -105,8 +131,11 @@ class AudioSocketMediaRuntime: touch_media_frame: Callable[[str], None], 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], + mark_reply_started: Callable[[str, str, bool, str | None], None] | None = None, + mark_reply_delivered: Callable[[str, str, bool, str | None], None], + mark_reply_discarded: Callable[[str, str, str | None, str], None] | None = None, plan_reply: Callable[[str, str, dict[str, Any] | None, str], None], + record_latency: Callable[[str, str, int], None] | None = 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], @@ -123,15 +152,20 @@ class AudioSocketMediaRuntime: self._trailing_silence_ms = max(trailing_silence_ms, self._frame_ms) self._max_turn_ms = max(max_turn_ms, self._frame_ms) self._asr_provider = asr_provider + self._streaming_asr_provider = streaming_asr_provider or StreamingASRProvider() self._tts_provider = tts_provider + self._ack_bank = PrebakedAckBank(tts_provider=tts_provider) self._load_registration_by_media_uuid = load_registration_by_media_uuid self._mark_media_connected = mark_media_connected self._mark_media_ended = mark_media_ended self._touch_media_frame = touch_media_frame self._set_state = set_state self._get_pending_greeting = get_pending_greeting + self._mark_reply_started = mark_reply_started or (lambda session_id, text, is_greeting, phase: None) self._mark_reply_delivered = mark_reply_delivered + self._mark_reply_discarded = mark_reply_discarded or (lambda session_id, text, phase, status: None) self._plan_reply = plan_reply + self._record_latency = record_latency or (lambda session_id, metric, latency_ms: None) self._process_turn = process_turn self._request_handoff = request_handoff self._handle_media_error = handle_media_error @@ -139,9 +173,12 @@ class AudioSocketMediaRuntime: self._loop: asyncio.AbstractEventLoop | None = None self._actors: dict[str, MediaActor] = {} self._v2_ack_wait_seconds = 0.18 - self._partial_asr_min_ms = 650 + self._partial_asr_min_ms = 320 self._immediate_ack_min_ms = 280 - self._v2_ack_post_gap_seconds = 0.12 + self._v2_ack_post_gap_seconds = 0.10 + self._partial_poll_interval_seconds = 0.20 + self._stable_partial_hold_seconds = 0.40 + self._barge_in_trigger_ms = 220 @staticmethod def _normalize_intent_text(text: str) -> str: @@ -171,7 +208,9 @@ class AudioSocketMediaRuntime: return "handoff" if intent in {"schedule", "address", "price", "status", "problem"}: return "understanding" - return "generic" + if intent == "unknown": + return "unknown" + return "clarify" @staticmethod def _ack_text(language: str | None, ack_kind: str) -> str: @@ -181,11 +220,15 @@ class AudioSocketMediaRuntime: return "Бір сәт." if ack_kind == "understanding": return "Қазір айтып шығамын." + if ack_kind == "clarify": + return "Қазір нақтылайын." return "Қазір айтайын." if ack_kind == "handoff": return "Секунду." if ack_kind == "understanding": return "Сейчас сориентирую." + if ack_kind == "clarify": + return "Сейчас уточню." return "Сейчас подскажу." @staticmethod @@ -218,6 +261,13 @@ class AudioSocketMediaRuntime: "Хорошо, сейчас подскажу.", "Понял вас, секунду.", ) + if ack_kind == "clarify": + return ( + "Угу, сейчас уточню.", + "Мхм, одну секунду.", + "Хорошо, сейчас уточню.", + "Ага, сейчас сориентирую.", + ) return ( "Угу, сейчас подскажу.", "Мхм, я в контексте.", @@ -251,10 +301,11 @@ class AudioSocketMediaRuntime: index = (index + 1) % len(variants) ack_text = variants[index] actor.last_ack_text = ack_text + actor.last_ack_variant = f"{ack_kind}:{index}" style_hints: dict[str, object] | None = None if emotive_ack: style_hints = {"role": "good"} - return ack_text, style_hints, f"{ack_kind}:{index}" + return ack_text, style_hints, actor.last_ack_variant 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") @@ -271,16 +322,152 @@ class AudioSocketMediaRuntime: def _reset_live_turn_state(actor: MediaActor) -> None: actor.utterance_generation += 1 actor.finalized_utterance_generation = 0 + actor.input_active = True actor.early_ack_started = False actor.partial_transcript = None actor.partial_intent = None + actor.partial_intent_streak = 0 + actor.stable_partial_intent = None actor.response_plan_id = None actor.partial_asr_attempted = False + actor.asr_poll_due_monotonic = 0.0 + actor.speech_started_monotonic = time.monotonic() + actor.speech_ended_monotonic = 0.0 + actor.barge_in_speech_ms = 0 + actor.barge_in_detected_monotonic = 0.0 + actor.current_reply_phase = None 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() + @staticmethod + def _is_streaming_v2_session(registration: MediaRegistration) -> bool: + return bool( + registration.voice_v2_enabled + and registration.voice_v2_duplex + and registration.voice_v2_partial_asr + and registration.voice_v2_streaming_asr_backend + ) + + async def _ensure_streaming_asr(self, actor: MediaActor) -> None: + if actor.closed or actor.asr_streaming_enabled: + return + if not self._is_streaming_v2_session(actor.registration): + return + try: + stream_id = await asyncio.to_thread( + self._streaming_asr_provider.open_stream, + actor.registration.voice_session_id, + language_hint=actor.registration.language, + ) + except StreamingASRUnavailable as exc: + actor.asr_streaming_enabled = False + actor.registration.voice_v2_duplex = False + actor.registration.voice_v2_partial_asr = False + logger.warning( + "audiosocket.streaming_asr_unavailable session_id=%s backend=%s error=%s", + actor.registration.voice_session_id, + actor.registration.voice_v2_streaming_asr_backend, + str(exc)[:500], + ) + return + actor.asr_stream_id = stream_id + actor.asr_streaming_enabled = True + actor.asr_poll_due_monotonic = 0.0 + + async def _close_streaming_asr(self, actor: MediaActor) -> None: + stream_id = actor.asr_stream_id + actor.asr_stream_id = None + actor.asr_streaming_enabled = False + if not stream_id: + return + await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id) + + def _update_stable_partial_intent(self, actor: MediaActor, intent: str) -> None: + normalized = str(intent or "").strip() or "unknown" + if actor.partial_intent == normalized: + actor.partial_intent_streak += 1 + else: + actor.partial_intent = normalized + actor.partial_intent_streak = 1 + if actor.partial_intent_streak >= 2: + actor.stable_partial_intent = normalized + + 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 + now = time.monotonic() + if now < actor.asr_poll_due_monotonic: + return + actor.asr_poll_due_monotonic = now + self._partial_poll_interval_seconds + partial = await asyncio.to_thread(self._streaming_asr_provider.poll_partial, actor.asr_stream_id) + if partial is None: + return + transcript_text = str(partial.text or "").strip() + if not transcript_text: + return + actor.partial_transcript = transcript_text + intent = self._detect_early_intent(transcript_text) + self._update_stable_partial_intent(actor, intent) + + 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") + stream_id = actor.asr_stream_id + try: + return await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id) + finally: + await self._close_streaming_asr(actor) + + async def _record_reply_status( + self, + actor: MediaActor, + *, + text: str, + is_greeting: bool, + phase: str | None, + status: str, + ) -> None: + if status == "started": + await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_started, actor.registration.voice_session_id, text, is_greeting, phase) + return + if status == "delivered": + await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_delivered, actor.registration.voice_session_id, text, is_greeting, phase) + return + if status in {"interrupted", "discarded"}: + await asyncio.to_thread( + self._invoke_discard_callback, + actor.registration.voice_session_id, + text, + phase, + status, + ) + + async def _record_latency_metric(self, actor: MediaActor, metric: str, start_monotonic: float) -> None: + if start_monotonic <= 0: + return + latency_ms = int(max((time.monotonic() - start_monotonic) * 1000.0, 0.0)) + await asyncio.to_thread( + self._record_latency, + actor.registration.voice_session_id, + metric, + latency_ms, + ) + + @staticmethod + def _invoke_reply_status_callback(callback, session_id: str, text: str, is_greeting: bool, phase: str | None) -> None: + try: + callback(session_id, text, is_greeting, phase) + except TypeError: + callback(session_id, text, is_greeting) + + def _invoke_discard_callback(self, session_id: str, text: str, phase: str | None, status: str) -> None: + try: + self._mark_reply_discarded(session_id, text, phase, status) + except TypeError: + return + async def _run_partial_asr_probe( self, actor: MediaActor, @@ -364,6 +551,7 @@ class AudioSocketMediaRuntime: ack_kind=ack_kind, ) actor.early_ack_started = True + actor.current_reply_phase = "ack" await self._plan_reply_segment( actor, ack_text, @@ -379,7 +567,10 @@ class AudioSocketMediaRuntime: "partial_ack_source": ack_source, }, ) - await self._speak_text(actor, ack_text, is_greeting=False, style_hints=style_hints) + if actor.registration.voice_v2_prebaked_ack and str(language or actor.registration.language or "").strip().lower().startswith("ru"): + await self._play_prebaked_ack(actor, ack_text, style_hints=style_hints) + else: + await self._speak_reply(actor, ack_text, is_greeting=False, style_hints=style_hints, reply_phase="ack") actor.last_ack_completed_monotonic = time.monotonic() if not actor.closed: await self._set_actor_state(actor, "thinking") @@ -402,6 +593,112 @@ class AudioSocketMediaRuntime: kind, ) + async def _play_pcm_payload( + self, + actor: MediaActor, + *, + text: str, + pcm_8k: bytes, + is_greeting: bool, + reply_phase: str | None, + ) -> None: + if actor.closed or not text or not pcm_8k: + return + await self._set_actor_state(actor, "speaking") + actor.current_reply_phase = reply_phase + interrupted = False + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="started", + ) + if reply_phase == "ack": + await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic) + await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic) + elif reply_phase == "main": + await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic) + for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes): + if actor.closed or actor.playback_interrupt.is_set(): + interrupted = True + break + await self._write_audio_packet(actor, frame) + await asyncio.sleep(actor.frame_ms / 1000.0) + if actor.playback_interrupt.is_set(): + interrupted = True + actor.playback_interrupt.clear() + actor.current_reply_phase = None + if interrupted or actor.closed: + await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic) + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="interrupted" if interrupted else "discarded", + ) + return + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="delivered", + ) + logger.warning( + "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s", + actor.registration.voice_session_id, + is_greeting, + reply_phase, + ) + + async def _play_prebaked_ack( + self, + actor: MediaActor, + text: str, + *, + style_hints: dict[str, object] | None = None, + ) -> None: + clip = await asyncio.to_thread( + self._ack_bank.get_clip, + text=text, + language=actor.registration.language, + style_hints=style_hints, + ) + await self._play_pcm_payload( + actor, + text=text, + pcm_8k=clip.pcm_8k_bytes, + is_greeting=False, + reply_phase="ack", + ) + + async def _speak_reply( + self, + actor: MediaActor, + text: str, + *, + is_greeting: bool, + style_hints: dict[str, object] | None = None, + reply_phase: str | None = "main", + ) -> None: + try: + await self._speak_text( + actor, + text, + is_greeting=is_greeting, + style_hints=style_hints, + reply_phase=reply_phase, + ) + except TypeError: + await self._speak_text( + actor, + text, + is_greeting=is_greeting, + style_hints=style_hints, + ) + @property def address(self) -> str: return f"{self._host}:{self._port}" @@ -580,17 +877,49 @@ class AudioSocketMediaRuntime: actor.last_media_touch_monotonic = now await asyncio.to_thread(self._touch_media_frame, actor.registration.voice_session_id) - if actor.state not in {"listening", "speaking"}: + if actor.state not in {"listening", "speaking", "thinking"}: return + is_speech = audioop.rms(pcm_frame, 2) >= 250 + if actor.state == "speaking": + if is_speech: + actor.barge_in_speech_ms += actor.frame_ms + if actor.barge_in_speech_ms >= self._barge_in_trigger_ms and not actor.playback_interrupt.is_set(): + actor.barge_in_detected_monotonic = time.monotonic() + actor.playback_interrupt.set() + actor.barge_in_pending = True + actor.speech_started_monotonic = actor.speech_started_monotonic or time.monotonic() + else: + actor.barge_in_speech_ms = 0 + else: + actor.barge_in_speech_ms = 0 + 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) + await self._ensure_streaming_asr(actor) + if actor.asr_streaming_enabled and actor.asr_stream_id and actor.input_active: + try: + await asyncio.to_thread( + self._streaming_asr_provider.push_pcm, + actor.asr_stream_id, + pcm_frame, + ) + await self._poll_streaming_partial(actor) + except StreamingASRUnavailable as exc: + logger.warning( + "audiosocket.streaming_asr_push_failed session_id=%s error=%s", + actor.registration.voice_session_id, + str(exc)[:500], + ) + await self._close_streaming_asr(actor) + actor.registration.voice_v2_duplex = False + actor.registration.voice_v2_partial_asr = False + elif actor.registration.voice_v2_partial_asr: + self._maybe_schedule_partial_asr(actor) if vad_result.utterance_pcm: + actor.input_active = False + actor.speech_ended_monotonic = time.monotonic() await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending)) actor.barge_in_pending = False @@ -605,7 +934,7 @@ class AudioSocketMediaRuntime: actor.registration.voice_session_id, len(greeting_text), ) - await self._speak_text(actor, greeting_text, is_greeting=True) + await self._speak_reply(actor, greeting_text, is_greeting=True, reply_phase="greeting") if not actor.closed: await self._set_actor_state(actor, "listening") @@ -617,20 +946,12 @@ 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) actor.playback_generation += 1 + actor.tts_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, - ) - ) - ) + 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), "media_uuid": actor.registration.media_uuid, @@ -638,24 +959,33 @@ class AudioSocketMediaRuntime: "voice_v2_enabled": actor.registration.voice_v2_enabled, "response_plan_id": actor.response_plan_id, "playback_generation": actor.playback_generation, + "tts_generation": actor.tts_generation, "partial_transcript": partial_transcript, "early_intent": partial_intent, } if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started: - 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) + 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.partial_intent or "").strip() or self._detect_early_intent(partial_transcript) + 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: + 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 if partial_transcript: await self._emit_early_ack( actor, language=actor.registration.language, metadata=base_metadata, - ack_source="precomputed_partial_asr", + ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr", ) elif len(pcm_bytes) >= self._immediate_ack_min_bytes: await self._emit_early_ack( @@ -665,7 +995,15 @@ class AudioSocketMediaRuntime: ack_source="immediate_turn_close", ) - transcription = await full_asr_task + if actor.asr_streaming_enabled: + transcription = await self._finalize_streaming_transcription(actor) + 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_text = str(transcription.text or "").strip() or partial_transcript if not transcript_text: await self._set_actor_state(actor, "listening") @@ -674,10 +1012,12 @@ class AudioSocketMediaRuntime: 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) + actor.stable_partial_intent = actor.partial_intent metadata = { **base_metadata, "partial_transcript": actor.partial_transcript, "early_intent": actor.partial_intent, + "reply_phase": "final", } decision_task = asyncio.create_task( asyncio.to_thread( @@ -730,7 +1070,7 @@ class AudioSocketMediaRuntime: "early_ack_started": actor.early_ack_started, }, ) - await self._speak_text(actor, decision.reply_text, is_greeting=False) + await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main") if actor.closed: return if decision.needs_handoff: @@ -738,6 +1078,7 @@ class AudioSocketMediaRuntime: with contextlib.suppress(asyncio.CancelledError, Exception): await handoff_task return + actor.input_active = False await self._set_actor_state(actor, "listening") def _start_handoff_request( @@ -824,13 +1165,23 @@ class AudioSocketMediaRuntime: *, is_greeting: bool, style_hints: dict[str, object] | None = None, + reply_phase: str | None = "main", ) -> None: if actor.closed or not text: return await self._set_actor_state(actor, "speaking") + actor.current_reply_phase = reply_phase synth_started_at = time.monotonic() first_frame_sent = False total_audio_bytes = 0 + interrupted = False + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="started", + ) async for synthesis in self._stream_tts_chunks(actor, text, style_hints=style_hints): if not synthesis.audio_bytes: continue @@ -843,6 +1194,11 @@ class AudioSocketMediaRuntime: int((time.monotonic() - synth_started_at) * 1000), total_audio_bytes, ) + if reply_phase == "ack": + await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic) + await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic) + elif reply_phase == "main": + await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic) pcm_8k = resample_pcm16le( synthesis.audio_bytes, input_rate_hz=synthesis.sample_rate_hz, @@ -850,6 +1206,7 @@ class AudioSocketMediaRuntime: ) for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes): if actor.closed or actor.playback_interrupt.is_set(): + interrupted = True break await self._write_audio_packet(actor, frame) if not first_frame_sent: @@ -862,25 +1219,38 @@ class AudioSocketMediaRuntime: ) await asyncio.sleep(actor.frame_ms / 1000.0) if actor.closed or actor.playback_interrupt.is_set(): + interrupted = True break if total_audio_bytes <= 0: raise RuntimeError("TTS provider returned empty audio") - interrupted = actor.playback_interrupt.is_set() + interrupted = interrupted or actor.playback_interrupt.is_set() actor.playback_interrupt.clear() - if not interrupted and not actor.closed: - await asyncio.to_thread( - self._mark_reply_delivered, - actor.registration.voice_session_id, - text, - is_greeting, - ) - logger.warning( - "audiosocket.reply_delivered session_id=%s greeting=%s", - actor.registration.voice_session_id, - is_greeting, + actor.current_reply_phase = None + if interrupted or actor.closed: + await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic) + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="interrupted" if interrupted else "discarded", ) + return + await self._record_reply_status( + actor, + text=text, + is_greeting=is_greeting, + phase=reply_phase, + status="delivered", + ) + logger.warning( + "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s", + actor.registration.voice_session_id, + is_greeting, + reply_phase, + ) async def _write_audio_packet(self, actor: MediaActor, pcm_frame: bytes) -> None: if actor.closed: @@ -928,6 +1298,8 @@ class AudioSocketMediaRuntime: handoff_reason, ) actor.state = state + actor.input_active = state in {"listening", "speaking", "thinking"} + actor.playback_active = state == "speaking" await asyncio.to_thread( self._set_state, actor.registration.voice_session_id, @@ -966,6 +1338,8 @@ class AudioSocketMediaRuntime: actor.partial_asr_task.cancel() with contextlib.suppress(asyncio.CancelledError, Exception): await actor.partial_asr_task + with contextlib.suppress(Exception): + await self._close_streaming_asr(actor) 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 4351ab7..0b5de4b 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -1,5 +1,6 @@ from __future__ import annotations +import base64 import os from dataclasses import dataclass @@ -27,6 +28,22 @@ def _openai_asr_model() -> str: return os.getenv("AI_VOICE_ASR_MODEL", "gpt-4o-mini-transcribe").strip() or "gpt-4o-mini-transcribe" +def _streaming_asr_api_base() -> str: + return ( + os.getenv("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021").strip() + or "http://127.0.0.1:8021" + ).rstrip("/") + + +def _streaming_asr_timeout_seconds() -> float: + raw = os.getenv("AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS", "4").strip() + try: + value = float(raw) + except ValueError: + value = 4.0 + return max(value, 0.25) + + @dataclass(slots=True) class ASRTranscription: text: str @@ -34,6 +51,19 @@ class ASRTranscription: confidence: float | None = None +@dataclass(slots=True) +class StreamingASRPartial: + text: str + language: str | None = None + confidence: float | None = None + is_final: bool = False + is_stable: bool = False + + +class StreamingASRUnavailable(RuntimeError): + pass + + class ASRProvider: name = "stub" @@ -45,6 +75,30 @@ class ASRProvider: return self.transcribe(audio_bytes, language_hint=language_hint) +class StreamingASRProvider: + name = "streaming-stub" + supports_streaming = False + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + del session_id, language_hint + raise StreamingASRUnavailable("Streaming ASR backend is not configured") + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + del stream_id, pcm_8k_chunk + raise StreamingASRUnavailable("Streaming ASR backend is not configured") + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + del stream_id + return None + + def finalize(self, stream_id: str) -> ASRTranscription: + del stream_id + raise StreamingASRUnavailable("Streaming ASR backend is not configured") + + def close_stream(self, stream_id: str) -> None: + del stream_id + + class OpenAIASRProvider(ASRProvider): name = "openai" @@ -84,8 +138,107 @@ class OpenAIASRProvider(ASRProvider): return self.transcribe(audio_bytes, language_hint=language_hint) +class LocalSidecarStreamingASRProvider(StreamingASRProvider): + name = "local-sidecar" + supports_streaming = True + + def __init__( + self, + *, + api_base: str | None = None, + timeout_seconds: float | None = None, + ) -> None: + self._api_base = str(api_base or _streaming_asr_api_base()).strip().rstrip("/") + self._timeout_seconds = max(float(timeout_seconds or _streaming_asr_timeout_seconds()), 0.25) + + def _request( + self, + method: str, + path: str, + *, + payload: dict[str, object] | None = None, + ) -> dict[str, object]: + if not self._api_base: + raise StreamingASRUnavailable("Streaming ASR API base is not configured") + try: + with httpx.Client(timeout=self._timeout_seconds) as client: + response = client.request( + method, + f"{self._api_base}{path}", + json=payload, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + raise StreamingASRUnavailable(str(exc)[:500] or "Streaming ASR sidecar is unavailable") from exc + body = response.json() if response.content else {} + return body if isinstance(body, dict) else {} + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + payload = { + "session_id": str(session_id or "").strip(), + "language_hint": str(language_hint or "").strip() or None, + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + } + body = self._request("POST", "/internal/asr/streams", payload=payload) + stream_id = str(body.get("stream_id") or "").strip() + if not stream_id: + raise StreamingASRUnavailable("Streaming ASR sidecar did not return stream_id") + return stream_id + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + if not pcm_8k_chunk: + return + self._request( + "POST", + f"/internal/asr/streams/{stream_id}/chunks", + payload={ + "pcm_b64": base64.b64encode(pcm_8k_chunk).decode("ascii"), + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + body = self._request("GET", f"/internal/asr/streams/{stream_id}/partial") + text = str(body.get("text") or "").strip() + if not text: + return None + language = str(body.get("language") or "").strip() or None + confidence_raw = body.get("confidence") + confidence = float(confidence_raw) if isinstance(confidence_raw, (int, float)) else None + return StreamingASRPartial( + text=text, + language=language, + confidence=confidence, + is_final=bool(body.get("is_final")), + is_stable=bool(body.get("is_stable")), + ) + + def finalize(self, stream_id: str) -> ASRTranscription: + body = self._request("POST", f"/internal/asr/streams/{stream_id}/finalize") + return ASRTranscription( + text=str(body.get("text") or "").strip(), + language=str(body.get("language") or "").strip() or None, + confidence=float(body["confidence"]) if isinstance(body.get("confidence"), (int, float)) else None, + ) + + def close_stream(self, stream_id: str) -> None: + try: + self._request("DELETE", f"/internal/asr/streams/{stream_id}") + except StreamingASRUnavailable: + return + + def build_asr_provider(name: str) -> ASRProvider: normalized = str(name or "stub").strip().lower() if normalized == "openai": return OpenAIASRProvider() return ASRProvider() + + +def build_streaming_asr_provider(name: str) -> StreamingASRProvider: + normalized = str(name or "disabled").strip().lower() + if normalized in {"local_sidecar", "local-sidecar", "sidecar"}: + return LocalSidecarStreamingASRProvider() + return StreamingASRProvider() diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index ea7ba7d..4611834 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -33,6 +33,7 @@ from services.shared.sql_models import ( VoiceNameCollectionSettingsRow, VoiceTTSSettingsRow, VoiceAISessionRow, + VoiceTranscriptSegmentRow, WhatsAppThreadRow, ) from services.telegram_adapter_service import app as telegram_module @@ -1419,6 +1420,91 @@ def test_voice_v2_off_domain_request_returns_fast_operator_fallback_without_llm( assert decision["metadata"]["response_plan_id"] == "rsp_off_domain" +def test_voice_v2_streaming_duplex_early_plan_returns_fast_safe_reply_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_plan", 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", + }, + ) + + assert decision["model"] == "voice_early_plan_off_domain" + assert decision["metadata"]["reply_phase"] == "early_plan" + assert decision["metadata"]["voice_v2_enabled"] is True + assert decision["metadata"]["response_plan_id"] == "rsp_early" + assert "кондиционер" not in decision["reply_text"].lower() + assert "оператор" in decision["reply_text"].lower() + + +def test_turn_voice_session_early_plan_does_not_persist_partial_turns(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) + + seeded = seed_voice_downstream_session( + marker=f"voice_early_plan_{new_id('seed')}", + name_status="name_not_obtained", + customer_display_name="+77010009999", + ) + + decision = voice_module.turn_voice_session( + seeded["session_id"], + VoiceAITurnIn( + voice_session_id=seeded["session_id"], + call_id=seeded["call_id"], + interaction_id=seeded["interaction_id"], + transcript_text="Расскажи, как устроен кондиционер", + language="ru", + sequence_no=1, + metadata={ + "voice_v2_enabled": True, + "reply_phase": "early_plan", + "response_plan_id": "rsp_early_turn", + }, + ), + ) + + assert decision.metadata["reply_phase"] == "early_plan" + assert decision.metadata["response_plan_id"] == "rsp_early_turn" + assert decision.status == "active" + + session = get_session() + try: + voice_session = session.execute( + select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == seeded["session_id"]) + ).scalar_one() + ai_turns = session.execute( + select(AITurnRow).where(AITurnRow.interaction_id == seeded["interaction_id"]) + ).scalars().all() + transcript_segments = session.execute( + select(VoiceTranscriptSegmentRow).where(VoiceTranscriptSegmentRow.session_id == seeded["session_id"]) + ).scalars().all() + assert voice_session.ai_session_id is None + assert voice_session.status == "active" + assert ai_turns == [] + assert transcript_segments == [] + finally: + session.close() + + def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch): monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1") monkeypatch.setenv("AI_PROVIDER", "stub")