From 34b807e4609b566e759d4208a19e1519e0b5f18a Mon Sep 17 00:00:00 2001 From: Yera All Date: Sat, 18 Apr 2026 18:25:07 +0500 Subject: [PATCH] feat(voice): add elevenlabs stt and transcript truth fixes --- .env.example | 5 + services/ai_orchestrator_service/voice.py | 63 ++++++- services/ai_voice_runtime_service/app.py | 139 +++++++++++++- .../ai_voice_runtime_service/media_runtime.py | 25 ++- .../ai_voice_runtime_service/providers/asr.py | 150 +++++++++++++++ services/asterisk_bridge_service/voice_ai.py | 141 ++++++++++++++- tests/test_ai_orchestrator_service.py | 34 +++- tests/test_ai_voice_asr_provider.py | 97 ++++++++++ tests/test_asterisk_bridge_service.py | 171 +++++++++++++++++- 9 files changed, 807 insertions(+), 18 deletions(-) diff --git a/.env.example b/.env.example index 3355d3d..12ebe03 100644 --- a/.env.example +++ b/.env.example @@ -76,6 +76,11 @@ AI_VOICE_AUDIOSOCKET_ENABLED=0 AI_VOICE_AUDIOSOCKET_HOST=0.0.0.0 AI_VOICE_AUDIOSOCKET_PORT=9019 AI_VOICE_ASR_MODEL=gpt-4o-mini-transcribe +AI_VOICE_ASR_ELEVENLABS_API_BASE=https://api.elevenlabs.io +AI_VOICE_ASR_ELEVENLABS_API_KEY= +AI_VOICE_ASR_ELEVENLABS_MODEL_ID=scribe_v1 +AI_VOICE_ASR_ELEVENLABS_LANGUAGE=ru +AI_VOICE_ASR_ELEVENLABS_SAMPLE_RATE_HZ=16000 AI_VOICE_ASR_YANDEX_API_BASE=https://stt.api.cloud.yandex.net AI_VOICE_ASR_YANDEX_OPERATIONS_BASE=https://operation.api.cloud.yandex.net AI_VOICE_ASR_YANDEX_API_VERSION=auto diff --git a/services/ai_orchestrator_service/voice.py b/services/ai_orchestrator_service/voice.py index daf617c..74224d3 100644 --- a/services/ai_orchestrator_service/voice.py +++ b/services/ai_orchestrator_service/voice.py @@ -1,5 +1,6 @@ from __future__ import annotations +import difflib import importlib import json import re @@ -268,6 +269,7 @@ def _normalize_name_candidate(text: str | None) -> str | None: return None invalid_tokens = { "\u0434\u0430", + "\u0445\u0430\u0447\u0443", "\u043d\u0435\u0442", "\u0430\u043b\u043b\u043e", "\u043f\u0440\u0438\u0432\u0435\u0442", @@ -286,6 +288,16 @@ def _normalize_name_candidate(text: str | None) -> str | None: } if any(_voice_text_key(word) in invalid_tokens for word in filtered): return None + if len(filtered) > 2: + return None + if len(filtered) == 2: + similarity = difflib.SequenceMatcher( + None, + _voice_text_key(filtered[0]), + _voice_text_key(filtered[1]), + ).ratio() + if similarity >= 0.72: + return None return _canonical_name(" ".join(filtered)) @@ -316,6 +328,7 @@ def _extract_name_candidate(text: str | None, language: str) -> tuple[str | None if not raw: return None, False normalized = _voice_text_key(raw) + service_topic = _voice_has_service_topic(raw) explicit_patterns = [ r"(?:\u043c\u0435\u043d\u044f\s+\u0437\u043e\u0432\u0443\u0442|my name is|i am|this is)\s+(.+)", r"(?:\u044f|it's me)\s+(.+)", @@ -331,7 +344,15 @@ def _extract_name_candidate(text: str | None, language: str) -> tuple[str | None cutoff_tokens = { "\u0445\u043e\u0442\u0435\u043b", "\u0445\u043e\u0447\u0443", + "\u043c\u043d\u0435", + "\u043d\u0430\u0434\u043e", "\u043d\u0443\u0436\u043d\u043e", + "\u0443\u0437\u043d\u0430\u0442\u044c", + "\u0433\u0440\u0430\u0444\u0438\u043a", + "\u0440\u0430\u0431\u043e\u0442\u044b", + "\u0430\u0434\u0440\u0435\u0441", + "\u0444\u0438\u043b\u0438\u0430\u043b", + "\u0433\u043e\u0440\u043e\u0434", "\u0442\u0430\u0440\u0438\u0444", "\u0441\u0442\u0430\u0442\u0443\u0441", "\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440", @@ -349,10 +370,12 @@ def _extract_name_candidate(text: str | None, language: str) -> tuple[str | None break candidate = _normalize_name_candidate(" ".join(candidate_words)) or _normalize_name_candidate(tail) if candidate: - return candidate, _name_followup_needed(raw) + return candidate, False candidate = _normalize_name_candidate(raw) if candidate: + if service_topic: + return None, False lower_cand = candidate.lower() stopwords = { "здравствуйте", "привет", "алло", "да", "нет", "добрый", "день", @@ -482,6 +505,26 @@ def _voice_explicit_name_candidate(text: str | None) -> str | None: if not raw: return None normalized = _voice_text_key(raw) + cutoff_tokens = { + "\u043c\u043d\u0435", + "\u043d\u0430\u0434\u043e", + "\u043d\u0443\u0436\u043d\u043e", + "\u0445\u043e\u0447\u0443", + "\u0445\u043e\u0442\u0435\u043b", + "\u0443\u0437\u043d\u0430\u0442\u044c", + "\u0433\u0440\u0430\u0444\u0438\u043a", + "\u0440\u0430\u0431\u043e\u0442\u044b", + "\u0430\u0434\u0440\u0435\u0441", + "\u0444\u0438\u043b\u0438\u0430\u043b", + "\u0433\u043e\u0440\u043e\u0434", + "\u0442\u0430\u0440\u0438\u0444", + "\u0441\u0442\u0430\u0442\u0443\u0441", + "\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440", + "\u0432\u043e\u043f\u0440\u043e\u0441", + "\u043a\u0435\u0440\u0435\u043a", + "\u0441\u04b1\u0440\u0430\u0493", + "\u043a\u04e9\u043c\u0435\u043a", + } patterns = ( r"(?:меня зовут|мое имя|это|my name is|i am|this is)\s+(.+)", r"(?:менің атым|аты[мң]?|mening atym)\s+(.+)", @@ -491,7 +534,15 @@ def _voice_explicit_name_candidate(text: str | None) -> str | None: if not match: continue tail = match.group(1).strip() - candidate = _normalize_name_candidate(tail) + tail_words = re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", tail, flags=re.UNICODE) + candidate_words: list[str] = [] + for word in tail_words: + if _voice_text_key(word) in cutoff_tokens: + break + candidate_words.append(word) + if len(candidate_words) >= 3: + break + candidate = _normalize_name_candidate(" ".join(candidate_words)) or _normalize_name_candidate(tail) if candidate: return candidate return None @@ -663,6 +714,7 @@ def _voice_downstream_name_update( source = str(current_source or "none").strip() or "none" resolved_at = str(current_resolved_at or "").strip() or None action = "ignored" + service_topic = _voice_has_service_topic(transcript_text) explicit_candidate = _voice_explicit_name_candidate(transcript_text) candidate, needs_followup = _extract_name_candidate(transcript_text, language) @@ -684,9 +736,9 @@ def _voice_downstream_name_update( elif name_value and _voice_is_name_confirmation(transcript_text, name_value): action = "confirm" candidate = name_value - elif candidate and name_value and candidate_key and candidate_key != current_key: + elif candidate and name_value and candidate_key and candidate_key != current_key and (explicit_candidate or not service_topic): action = "correct" - elif candidate: + elif candidate and (explicit_candidate or not service_topic): action = "provide" if not name_value else "confirm" candidate = candidate or name_value elif status == "name_obtained": @@ -694,7 +746,7 @@ def _voice_downstream_name_update( else: if explicit_candidate: action = "provide" - elif candidate and not needs_followup: + elif candidate and not needs_followup and not service_topic: action = "provide" finalizable = False @@ -729,6 +781,7 @@ def _voice_downstream_name_update( and action == "ignored" and not _voice_name_followup_asked(transcript_window, language, config) and not _voice_is_low_signal_caller_text(transcript_text) + and not service_topic ) return { "status": status, diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index aa28722..4a4bb86 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -333,6 +333,107 @@ def _record_segment( ) +def _normalize_voice_text_key(text: str | None) -> str: + compact = str(text or "").strip().lower() + for char in ",.!?;:\"'()[]{}": + compact = compact.replace(char, " ") + return " ".join(compact.split()) + + +def _is_low_signal_voice_text(text: str | None) -> bool: + normalized = _normalize_voice_text_key(text) + if not normalized: + return True + return normalized in { + "\u0430\u043b\u043b\u043e", + "\u0430\u0433\u0430", + "\u0434\u0430", + "\u0434\u043e\u0431\u0440\u044b\u0439 \u0434\u0435\u043d\u044c", + "\u0437\u0434\u0440\u0430\u0432\u0441\u0442\u0432\u0443\u0439\u0442\u0435", + "\u043b\u0430\u0434\u043d\u043e", + "\u043d\u0435\u0442", + "\u043d\u0435\u0430", + "\u043e\u0439", + "\u043e\u043a", + "\u043f\u0440\u0438\u0432\u0435\u0442", + "\u0441\u043b\u044b\u0448\u043d\u043e", + "\u0441\u043b\u044b\u0448\u0443", + "\u0443\u0433\u0443", + "\u0445\u043e\u0440\u043e\u0448\u043e", + "\u044f\u0441\u043d\u043e", + } + + +def _is_intent_bearing_turn(transcript_text: str | None, intent: str | None) -> bool: + normalized_intent = str(intent or "").strip().lower() + if normalized_intent in { + "schedule", + "address", + "price", + "status", + "problem", + "operator_request", + "handoff_request", + "kb_answer", + }: + return True + normalized_text = _normalize_voice_text_key(transcript_text) + if not normalized_text or _is_low_signal_voice_text(normalized_text): + return False + markers = ( + "\u0433\u0440\u0430\u0444\u0438\u043a", + "\u0432\u0440\u0435\u043c\u044f \u0440\u0430\u0431\u043e\u0442\u044b", + "\u0440\u0435\u0436\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u044b", + "\u0430\u0434\u0440\u0435\u0441", + "\u0444\u0438\u043b\u0438\u0430\u043b", + "\u0433\u043e\u0440\u043e\u0434", + "\u0437\u0430\u044f\u0432\u043a", + "\u0441\u0442\u0430\u0442\u0443\u0441", + "\u0442\u0430\u0440\u0438\u0444", + "\u0446\u0435\u043d\u0430", + "\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c", + "\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440", + "\u043c\u0435\u043d\u0435\u0434\u0436\u0435\u0440", + "\u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442", + "\u043e\u0448\u0438\u0431\u043a", + "\u043f\u0440\u043e\u0431\u043b\u0435\u043c", + ) + return any(marker in normalized_text for marker in markers) + + +def _annotate_latest_caller_segment( + session, + *, + session_id: str, + provider_name: str, + transcript_text: str, + intent: str | None, + metadata: dict[str, Any] | None, +) -> None: + row = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == session_id) + .where(VoiceTranscriptSegmentRow.speaker == "caller") + .order_by(VoiceTranscriptSegmentRow.id.desc()) + .limit(1) + ).scalar_one_or_none() + if row is None: + return + try: + payload = json.loads(row.payload_json or "{}") + except Exception: + payload = {} + payload["provider"] = str(provider_name or "").strip() or payload.get("provider") + payload["intent_bearing"] = _is_intent_bearing_turn(transcript_text, intent) + if intent: + payload["decision_intent"] = str(intent).strip() + if isinstance(metadata, dict) and metadata: + merged_metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {} + merged_metadata.update(metadata) + payload["metadata"] = merged_metadata + row.payload_json = json.dumps(payload, ensure_ascii=False) + + def _load_greeting_segment(session, session_id: str) -> VoiceTranscriptSegmentRow | None: rows = session.execute( select(VoiceTranscriptSegmentRow) @@ -379,7 +480,12 @@ def _ensure_greeting_segment(session, voice_session: VoiceAISessionRow, greeting source_type="tts", text=greeting_text, sequence_no=_next_sequence(session, voice_session.session_id), - payload={"kind": "greeting", "delivery_status": "planned"}, + payload={ + "kind": "greeting", + "provider": _current_tts_provider_name(), + "delivery_status": "planned", + "delivery_state": "planned", + }, is_final=False, ) @@ -399,6 +505,8 @@ def _upsert_greeting_segment(session, voice_session: VoiceAISessionRow, greeting payload = {} payload["kind"] = "greeting" payload.setdefault("delivery_status", "planned") + payload.setdefault("delivery_state", str(payload.get("delivery_status") or "planned")) + payload.setdefault("provider", _current_tts_provider_name()) row.payload_json = json.dumps(payload, ensure_ascii=False) row.is_final = False @@ -414,7 +522,12 @@ def _queue_new_greeting_segment(session, voice_session: VoiceAISessionRow, greet source_type="tts", text=normalized, sequence_no=_next_sequence(session, voice_session.session_id), - payload={"kind": "greeting", "delivery_status": "planned"}, + payload={ + "kind": "greeting", + "provider": _current_tts_provider_name(), + "delivery_status": "planned", + "delivery_state": "planned", + }, is_final=False, ) @@ -492,6 +605,7 @@ def _mark_last_assistant_segment_interrupted(session, session_id: str) -> None: except Exception: payload = {} payload["delivery_status"] = "interrupted" + payload["delivery_state"] = "interrupted" row.payload_json = json.dumps(payload, ensure_ascii=False) @@ -718,6 +832,8 @@ def _update_reply_delivery_status( continue matched_row = row payload["delivery_status"] = status + payload["delivery_state"] = status + payload.setdefault("provider", _current_tts_provider_name()) row.payload_json = json.dumps(payload, ensure_ascii=False) break if matched_row is not None: @@ -844,7 +960,9 @@ def _record_runtime_reply_planned( text=normalized_text, sequence_no=_next_sequence(session, voice_session.session_id), payload={ + "provider": _current_tts_provider_name(), "delivery_status": "planned", + "delivery_state": "planned", "kind": kind, "metadata": metadata or {}, }, @@ -1065,7 +1183,12 @@ 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={ + "provider": _ASR_PROVIDER.name, + "metadata": payload_metadata, + "barge_in": payload.barge_in, + "intent_bearing": False, + }, ) session.commit() @@ -1080,6 +1203,14 @@ def _process_voice_ai_turn_sync( voice_session.language = decision.language or voice_session.language voice_session.handoff_reason = decision.handoff_reason _apply_voice_start_metadata(voice_session, decision.metadata) + _annotate_latest_caller_segment( + session, + session_id=voice_session.session_id, + provider_name=_ASR_PROVIDER.name, + transcript_text=payload.transcript_text, + intent=decision.intent, + metadata=payload_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")): @@ -1094,7 +1225,9 @@ def _process_voice_ai_turn_sync( "intent": decision.intent, "kb_refs": decision.kb_refs, "model": decision.model, + "provider": _current_tts_provider_name(), "delivery_status": "planned", + "delivery_state": "planned", "metadata": decision.metadata, }, is_final=False, diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 6f54478..968ebc7 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -254,6 +254,19 @@ class AudioSocketMediaRuntime: return "problem" return "unknown" + @staticmethod + def _asr_failure_category(exc: Exception) -> str: + text = str(exc or "").lower() + if "unauthorized" in text or "unknown api key" in text or "invalid api key" in text: + return "auth" + if "timeout" in text: + return "timeout" + if "not found" in text: + return "provider_not_found" + if "service unavailable" in text or "temporarily unavailable" in text: + return "provider_unavailable" + return "provider_error" + @staticmethod def _ack_kind_for_intent(intent: str) -> str: if intent == "operator_request": @@ -1223,10 +1236,12 @@ class AudioSocketMediaRuntime: "queue_code": actor.registration.queue_code, "reply_phase": actor.current_reply_phase or "error_handoff", "error_class": exc.__class__.__name__, + "error_category": self._asr_failure_category(exc), } logger.warning( - "audiosocket.turn_failed session_id=%s error=%s", + "audiosocket.turn_failed session_id=%s category=%s error=%s", actor.registration.voice_session_id, + metadata["error_category"], error_text, ) await self._set_actor_state(actor, "handoff_requested", error_text) @@ -1326,6 +1341,14 @@ class AudioSocketMediaRuntime: language_hint=actor.registration.language, ) transcript_text = str(transcription.text or "").strip() or partial_transcript + logger.info( + "audiosocket.asr_turn_ready session_id=%s provider=%s utterance_ms=%s text_len=%s empty=%s", + actor.registration.voice_session_id, + getattr(self._asr_provider, "name", "unknown"), + int(len(pcm_bytes) / 16), + len(str(transcript_text or "").strip()), + not bool(str(transcript_text or "").strip()), + ) if not transcript_text: await self._set_actor_state(actor, "listening") return diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 7183e85..08df5b5 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 audioop import base64 import io import os @@ -12,6 +13,7 @@ from dataclasses import dataclass, field import httpx +from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes, resample_pcm16le from services.shared.security import issue_app_token @@ -36,6 +38,67 @@ def _openai_asr_model() -> str: return os.getenv("AI_VOICE_ASR_MODEL", "gpt-4o-mini-transcribe").strip() or "gpt-4o-mini-transcribe" +def _elevenlabs_asr_api_base() -> str: + return ( + os.getenv("AI_VOICE_ASR_ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() + or "https://api.elevenlabs.io" + ).rstrip("/") + + +def _elevenlabs_asr_api_key() -> str: + return ( + os.getenv("AI_VOICE_ASR_ELEVENLABS_API_KEY", "").strip() + or os.getenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "").strip() + ) + + +def _elevenlabs_asr_model_id() -> str: + return os.getenv("AI_VOICE_ASR_ELEVENLABS_MODEL_ID", "scribe_v1").strip() or "scribe_v1" + + +def _elevenlabs_asr_default_language() -> str: + return os.getenv("AI_VOICE_ASR_ELEVENLABS_LANGUAGE", "ru").strip() or "ru" + + +def _normalize_elevenlabs_asr_language(language: str | None) -> str: + raw = str(language or "").strip().lower().replace("_", "-") + if not raw: + raw = _elevenlabs_asr_default_language().strip().lower().replace("_", "-") + mapping = { + "ru": "rus", + "ru-ru": "rus", + "kk": "kaz", + "kz": "kaz", + "kk-kz": "kaz", + "kz-kz": "kaz", + "en": "eng", + "en-us": "eng", + "en-gb": "eng", + } + return mapping.get(raw, raw or "rus") + + +def _elevenlabs_asr_sample_rate_hz() -> int: + raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_SAMPLE_RATE_HZ", "16000").strip() + try: + value = int(raw) + except ValueError: + value = 16000 + if value < 8000 or value > 48000: + return 16000 + return value + + +def _resolve_elevenlabs_asr_language(language_code: str | None, language_hint: str | None) -> str | None: + normalized = str(language_code or "").strip().lower() + if normalized in {"rus", "ru", "ru-ru"}: + return "ru" + if normalized in {"kaz", "kk", "kz", "kk-kz", "kz-kz"}: + return "kz" + resolved = str(language_code or language_hint or "").strip() + return resolved or language_hint + + def _yandex_asr_api_base() -> str: explicit = os.getenv("AI_VOICE_ASR_YANDEX_API_BASE", "").strip() if explicit: @@ -286,6 +349,91 @@ class OpenAIASRProvider(ASRProvider): return self.transcribe(audio_bytes, language_hint=language_hint) +class ElevenLabsASRProvider(ASRProvider): + name = "elevenlabs" + + def __init__( + self, + *, + api_base: str | None = None, + api_key: str | None = None, + timeout_seconds: float | None = None, + model_id: str | None = None, + sample_rate_hz: int | None = None, + ) -> None: + self._api_base = str(api_base or _elevenlabs_asr_api_base()).strip().rstrip("/") + self._api_key = str(api_key if api_key is not None else _elevenlabs_asr_api_key()).strip() + self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 3.0) + self._model_id = str(model_id or _elevenlabs_asr_model_id()).strip() or _elevenlabs_asr_model_id() + self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _elevenlabs_asr_sample_rate_hz()) + + @staticmethod + def _pcm_from_audio_bytes(audio_bytes: bytes, *, default_sample_rate_hz: int) -> tuple[bytes, int]: + if not audio_bytes: + return b"", default_sample_rate_hz + try: + with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file: + sample_width = wav_file.getsampwidth() + channels = wav_file.getnchannels() + sample_rate = wav_file.getframerate() + pcm_bytes = wav_file.readframes(wav_file.getnframes()) + if sample_width != 2: + return audio_bytes, default_sample_rate_hz + if channels == 2: + pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5) # type: ignore[name-defined] + elif channels != 1: + return audio_bytes, default_sample_rate_hz + return pcm_bytes, int(sample_rate or default_sample_rate_hz) + except (wave.Error, EOFError): + return audio_bytes, default_sample_rate_hz + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + if not audio_bytes: + return ASRTranscription(text="", language=language_hint, confidence=None) + if not self._api_key: + raise RuntimeError( + "AI_VOICE_ASR_ELEVENLABS_API_KEY or AI_VOICE_TTS_ELEVENLABS_API_KEY is required for ElevenLabs ASR" + ) + + pcm_bytes, input_sample_rate_hz = self._pcm_from_audio_bytes( + audio_bytes, + default_sample_rate_hz=self._sample_rate_hz, + ) + if not pcm_bytes: + return ASRTranscription(text="", language=language_hint, confidence=None) + normalized_sample_rate_hz = self._sample_rate_hz + if input_sample_rate_hz != normalized_sample_rate_hz: + pcm_bytes = resample_pcm16le( + pcm_bytes, + input_rate_hz=input_sample_rate_hz, + output_rate_hz=normalized_sample_rate_hz, + ) + wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=normalized_sample_rate_hz) + language_code = _normalize_elevenlabs_asr_language(language_hint) + with httpx.Client(timeout=self._timeout_seconds) as client: + response = client.post( + f"{self._api_base}/v1/speech-to-text", + headers={"xi-api-key": self._api_key}, + data={ + "model_id": self._model_id, + "language_code": language_code, + }, + files={"file": ("turn.wav", wav_bytes, "audio/wav")}, + ) + response.raise_for_status() + payload = response.json() + transcript_text = str(payload.get("text") or payload.get("transcript") or "").strip() + resolved_language = _resolve_elevenlabs_asr_language(payload.get("language_code"), language_hint) + return ASRTranscription( + text=transcript_text, + language=resolved_language, + 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 YandexSpeechKitASRProvider(ASRProvider): name = "yandex" @@ -985,6 +1133,8 @@ class YandexSpeechKitBufferedStreamingASRProvider(StreamingASRProvider): def build_asr_provider(name: str) -> ASRProvider: normalized = str(name or "stub").strip().lower() + if normalized == "elevenlabs": + return ElevenLabsASRProvider() if normalized == "openai": return OpenAIASRProvider() if normalized in {"yandex", "yandex_speechkit", "speechkit"}: diff --git a/services/asterisk_bridge_service/voice_ai.py b/services/asterisk_bridge_service/voice_ai.py index b2965c8..0b5fda3 100644 --- a/services/asterisk_bridge_service/voice_ai.py +++ b/services/asterisk_bridge_service/voice_ai.py @@ -615,6 +615,117 @@ def _next_transcript_sequence(session, session_id: str) -> int: return next_transcript_sequence(session, session_id) +def _segment_payload(row: VoiceTranscriptSegmentRow) -> dict[str, Any]: + try: + payload = json.loads(row.payload_json or "{}") + except Exception: + payload = {} + return payload if isinstance(payload, dict) else {} + + +def _segment_delivery_state(row: VoiceTranscriptSegmentRow) -> str | None: + payload = _segment_payload(row) + raw = payload.get("delivery_state") or payload.get("delivery_status") + if raw: + return str(raw).strip().lower() or None + if row.speaker == "assistant" and row.is_final and not row.barge_in_interrupted: + return "delivered" + return None + + +def _caller_low_signal_text(text: str | None) -> bool: + normalized = " ".join(str(text or "").strip().lower().replace(",", " ").replace(".", " ").split()) + if not normalized: + return True + return normalized in { + "\u0430\u0433\u0430", + "\u0430\u043b\u043b\u043e", + "\u0434\u0430", + "\u0434\u043e\u0431\u0440\u044b\u0439 \u0434\u0435\u043d\u044c", + "\u0437\u0434\u0440\u0430\u0432\u0441\u0442\u0432\u0443\u0439\u0442\u0435", + "\u043b\u0430\u0434\u043d\u043e", + "\u043d\u0435\u0442", + "\u043d\u0435\u0430", + "\u043e\u0439", + "\u043e\u043a", + "\u043f\u0440\u0438\u0432\u0435\u0442", + "\u0441\u043b\u044b\u0448\u043d\u043e", + "\u0441\u043b\u044b\u0448\u0443", + "\u0443\u0433\u0443", + "\u0445\u043e\u0440\u043e\u0448\u043e", + "\u044f\u0441\u043d\u043e", + } + + +def _caller_segment_intent_bearing(row: VoiceTranscriptSegmentRow) -> bool: + payload = _segment_payload(row) + marker = payload.get("intent_bearing") + if isinstance(marker, bool): + return marker + text = str(row.text or "").strip().lower() + if _caller_low_signal_text(text): + return False + markers = ( + "\u0433\u0440\u0430\u0444\u0438\u043a", + "\u0432\u0440\u0435\u043c\u044f \u0440\u0430\u0431\u043e\u0442\u044b", + "\u0440\u0435\u0436\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u044b", + "\u0430\u0434\u0440\u0435\u0441", + "\u0444\u0438\u043b\u0438\u0430\u043b", + "\u0433\u043e\u0440\u043e\u0434", + "\u0437\u0430\u044f\u0432\u043a", + "\u0441\u0442\u0430\u0442\u0443\u0441", + "\u0442\u0430\u0440\u0438\u0444", + "\u0446\u0435\u043d\u0430", + "\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c", + "\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440", + "\u043c\u0435\u043d\u0435\u0434\u0436\u0435\u0440", + "\u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442", + "\u043e\u0448\u0438\u0431\u043a", + "\u043f\u0440\u043e\u0431\u043b\u0435\u043c", + "\u0445\u043e\u0447\u0443 \u0443\u0437\u043d\u0430\u0442\u044c", + "\u043c\u043d\u0435 \u043d\u0430\u0434\u043e", + "\u043c\u043d\u0435 \u043d\u0443\u0436\u043d\u043e", + ) + return any(marker in text for marker in markers) + + +def _latest_customer_request_text(session, *, voice_session_id: str) -> str | None: + rows = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == voice_session_id) + .where(VoiceTranscriptSegmentRow.speaker == "caller") + .where(VoiceTranscriptSegmentRow.is_final.is_(True)) + .order_by(VoiceTranscriptSegmentRow.sequence_no.desc(), VoiceTranscriptSegmentRow.id.desc()) + ).scalars().all() + if not rows: + return None + for row in rows: + text = _truncate(row.text, 4000) + if text and _caller_segment_intent_bearing(row): + return text + for row in rows: + text = _truncate(row.text, 4000) + if text and not _caller_low_signal_text(text): + return text + return _truncate(rows[0].text, 4000) + + +def _latest_delivered_assistant_text(session, *, voice_session_id: str) -> str | None: + rows = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == voice_session_id) + .where(VoiceTranscriptSegmentRow.speaker == "assistant") + .order_by(VoiceTranscriptSegmentRow.sequence_no.desc(), VoiceTranscriptSegmentRow.id.desc()) + ).scalars().all() + for row in rows: + if _segment_delivery_state(row) != "delivered": + continue + text = _truncate(row.text, 4000) + if text: + return text + return None + + def _latest_segment_text( session, *, @@ -622,6 +733,23 @@ def _latest_segment_text( speaker: str, final_only: bool = False, ) -> str | None: + normalized_speaker = str(speaker or "").strip().lower() + if normalized_speaker == "caller": + return _latest_customer_request_text(session, voice_session_id=voice_session_id) + if normalized_speaker == "assistant": + if final_only: + return _latest_delivered_assistant_text(session, voice_session_id=voice_session_id) + rows = session.execute( + select(VoiceTranscriptSegmentRow) + .where(VoiceTranscriptSegmentRow.session_id == voice_session_id) + .where(VoiceTranscriptSegmentRow.speaker == "assistant") + .order_by(VoiceTranscriptSegmentRow.id.desc()) + ).scalars().all() + for row in rows: + text = _truncate(row.text, 4000) + if text: + return text + return None query = ( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session_id) @@ -629,9 +757,7 @@ def _latest_segment_text( ) if final_only: query = query.where(VoiceTranscriptSegmentRow.is_final.is_(True)) - row = session.execute( - query.order_by(VoiceTranscriptSegmentRow.id.desc()).limit(1) - ).scalar_one_or_none() + row = session.execute(query.order_by(VoiceTranscriptSegmentRow.id.desc()).limit(1)).scalar_one_or_none() return _truncate(row.text if row else None, 4000) @@ -643,7 +769,6 @@ def _summary_transcript_segments( rows = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session_id) - .where(VoiceTranscriptSegmentRow.is_final.is_(True)) .order_by(VoiceTranscriptSegmentRow.sequence_no.asc(), VoiceTranscriptSegmentRow.id.asc()) ).scalars().all() result: list[VoiceAISummaryTranscriptSegmentOut] = [] @@ -652,7 +777,13 @@ def _summary_transcript_segments( if not text: continue speaker = str(row.speaker or "").strip().lower() - if speaker not in {"caller", "assistant"}: + if speaker == "caller": + if not row.is_final: + continue + elif speaker == "assistant": + if _segment_delivery_state(row) != "delivered": + continue + else: continue result.append( VoiceAISummaryTranscriptSegmentOut( diff --git a/tests/test_ai_orchestrator_service.py b/tests/test_ai_orchestrator_service.py index b2128eb..2e39da8 100644 --- a/tests/test_ai_orchestrator_service.py +++ b/tests/test_ai_orchestrator_service.py @@ -3104,7 +3104,8 @@ def test_custom_voice_name_texts_are_used_in_start_and_followup(): metadata={"voice_start_language": "ru"}, ), ) - assert "Как к вам обращаться сейчас?" in decision.reply_text + assert "как к вам обращаться сейчас" not in decision.reply_text.lower() + assert "город" in decision.reply_text.lower() def test_downstream_voice_start_personalizes_greeting_and_finalizes_confirmed_name(): @@ -3176,7 +3177,8 @@ def test_downstream_voice_turn_adds_inline_name_followup_then_finalizes_provided ) assert first_turn.metadata["customer_name_status"] == "name_not_obtained" - assert "как мне к вам обращаться" in first_turn.reply_text.lower() + assert "как мне к вам обращаться" not in first_turn.reply_text.lower() + assert "город" in first_turn.reply_text.lower() second_turn = voice_module.turn_voice_session( seeded["session_id"], @@ -3212,6 +3214,34 @@ def test_downstream_voice_turn_adds_inline_name_followup_then_finalizes_provided session.close() +def test_downstream_voice_turn_extracts_explicit_name_without_restarting_name_flow(): + seeded = seed_voice_downstream_session( + marker=f"voice_explicit_name_intent_{new_id('seed')}", + name_status="name_not_obtained", + customer_display_name="+77010009999", + ) + + turn = 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_start_language": "ru"}, + ), + ) + + assert turn.metadata["customer_name_status"] == "name_obtained" + assert turn.metadata["customer_name_value"] == "Ания" + assert "город" in turn.reply_text.lower() + assert "филиал" in turn.reply_text.lower() + assert "какой вопрос по работе" not in turn.reply_text.lower() + assert "как мне к вам обращаться" not in turn.reply_text.lower() + + @pytest.mark.parametrize( ("transcript_text", "expected_name"), [ diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index 9f24585..7783a71 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -2,6 +2,9 @@ from __future__ import annotations import time +import httpx +import pytest + from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes from services.ai_voice_runtime_service.providers import asr as asr_module @@ -78,6 +81,99 @@ def test_yandex_asr_provider_posts_lpcm_with_tts_credential_fallback(monkeypatch assert calls[0]["timeout"] == 5.0 +def test_elevenlabs_asr_provider_posts_wav_with_tts_credential_fallback(monkeypatch): + calls: list[dict] = [] + pcm = b"\x01\x00" * 160 + wav_bytes = pcm16le_to_wav_bytes(pcm, sample_rate_hz=8000) + + class _DummyClient: + def __init__(self, *, timeout: float) -> None: + self.timeout = timeout + + def __enter__(self) -> "_DummyClient": + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def post( + self, + url: str, + *, + headers: dict[str, str], + data: dict[str, str], + files: dict[str, tuple[str, bytes, str]], + ) -> _DummyResponse: + calls.append( + { + "url": url, + "headers": headers, + "data": data, + "files": files, + "timeout": self.timeout, + } + ) + return _DummyResponse({"text": "schedule", "language_code": "rus"}) + + monkeypatch.delenv("AI_VOICE_ASR_ELEVENLABS_API_KEY", raising=False) + monkeypatch.setenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "tts-elevenlabs-key") + monkeypatch.setenv("AI_VOICE_ASR_ELEVENLABS_API_BASE", "https://api.elevenlabs.example") + monkeypatch.setenv("AI_VOICE_ASR_ELEVENLABS_MODEL_ID", "scribe_v1") + monkeypatch.setattr(asr_module.httpx, "Client", _DummyClient) + + provider = asr_module.ElevenLabsASRProvider() + result = provider.transcribe(wav_bytes, language_hint="ru") + + assert result.text == "schedule" + assert result.language == "ru" + assert len(calls) == 1 + assert calls[0]["url"] == "https://api.elevenlabs.example/v1/speech-to-text" + assert calls[0]["headers"]["xi-api-key"] == "tts-elevenlabs-key" + assert calls[0]["data"]["model_id"] == "scribe_v1" + assert calls[0]["data"]["language_code"] == "rus" + assert calls[0]["files"]["file"][0] == "turn.wav" + assert calls[0]["files"]["file"][2] == "audio/wav" + assert calls[0]["timeout"] == 20.0 + + +def test_elevenlabs_asr_provider_surfaces_http_auth_errors(monkeypatch): + request = httpx.Request("POST", "https://api.elevenlabs.example/v1/speech-to-text") + response = httpx.Response(401, request=request) + + class _DummyClient: + def __init__(self, *, timeout: float) -> None: + self.timeout = timeout + + def __enter__(self) -> "_DummyClient": + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def post( + self, + url: str, + *, + headers: dict[str, str], + data: dict[str, str], + files: dict[str, tuple[str, bytes, str]], + ) -> _DummyResponse: + del url, headers, data, files + class _FailingResponse(_DummyResponse): + def raise_for_status(self_nonlocal) -> None: + raise httpx.HTTPStatusError("401 Unauthorized", request=request, response=response) + + return _FailingResponse() + + monkeypatch.setenv("AI_VOICE_ASR_ELEVENLABS_API_KEY", "asr-elevenlabs-key") + monkeypatch.setattr(asr_module.httpx, "Client", _DummyClient) + + provider = asr_module.ElevenLabsASRProvider() + + with pytest.raises(httpx.HTTPStatusError): + provider.transcribe(b"\x01\x00" * 160, language_hint="ru") + + def test_yandex_asr_provider_uses_iam_token(monkeypatch): calls: list[dict] = [] @@ -325,6 +421,7 @@ def test_yandex_grpc_streaming_provider_returns_partial_and_final(): def test_yandex_asr_builders(): + assert isinstance(asr_module.build_asr_provider("elevenlabs"), asr_module.ElevenLabsASRProvider) assert isinstance(asr_module.build_asr_provider("yandex"), asr_module.YandexSpeechKitASRProvider) assert isinstance(asr_module.build_asr_provider("speechkit"), asr_module.YandexSpeechKitASRProvider) assert isinstance( diff --git a/tests/test_asterisk_bridge_service.py b/tests/test_asterisk_bridge_service.py index d846b38..069168c 100644 --- a/tests/test_asterisk_bridge_service.py +++ b/tests/test_asterisk_bridge_service.py @@ -1082,7 +1082,7 @@ def test_voice_ai_summary_exposes_customer_name_state_for_operator(tmp_path): confidence=0.96, is_final=True, barge_in_interrupted=False, - payload_json="{}", + payload_json='{"intent_bearing": true, "provider": "elevenlabs"}', created_at=now, ), VoiceTranscriptSegmentRow( @@ -1097,7 +1097,7 @@ def test_voice_ai_summary_exposes_customer_name_state_for_operator(tmp_path): confidence=None, is_final=True, barge_in_interrupted=False, - payload_json="{}", + payload_json='{"delivery_state": "delivered", "provider": "elevenlabs"}', created_at=now, ), ] @@ -1113,6 +1113,173 @@ def test_voice_ai_summary_exposes_customer_name_state_for_operator(tmp_path): assert summary.customer_name_value == "Айдос" assert summary.customer_name_source == "voice_start" assert summary.voice_start_language == "ru" + assert summary.customer_request_text == "Соедините с оператором" + assert summary.ai_outcome_text == "Сейчас переведу вас на оператора." + assert [segment.text for segment in summary.transcript_segments] == [ + "Соедините с оператором", + "Сейчас переведу вас на оператора.", + ] + + +def test_voice_ai_summary_uses_intent_bearing_caller_turns_and_delivered_assistant_segments(tmp_path): + now = utc_now_iso() + call_id = f"call_voice_truth_{tmp_path.name}" + session_id = f"avs_voice_truth_{tmp_path.name}" + ai_session_id = f"ais_voice_truth_{tmp_path.name}" + interaction_id = f"int_voice_truth_{tmp_path.name}" + + session = get_session() + try: + session.add( + AsteriskCallLinkRow( + call_id=call_id, + linked_id=f"linked_voice_truth_{tmp_path.name}", + queue_code="voice_lab_ai", + queue_id="que_voice_lab", + interaction_id=interaction_id, + caller_number="+77010002233", + caller_name="Truth Caller", + status="active", + telephony_status="connected", + claimed_by_user=None, + claimed_at=None, + operator_extension=None, + channel_name="PJSIP/1001-000111", + started_at=now, + connected_at=now, + ended_at=None, + updated_at=now, + voice_session_id=session_id, + ai_session_id=ai_session_id, + ai_state="active", + ai_handoff_reason=None, + ai_last_model_at=now, + voice_start_language="ru", + ) + ) + session.add( + AISessionRow( + session_id=ai_session_id, + channel="voice", + call_id=call_id, + thread_id=None, + interaction_id=interaction_id, + customer_id="cus_voice_truth", + agent_profile="voice_support", + language="ru", + status="active", + summary_text="", + last_user_message_id=None, + last_ai_message_id=None, + handoff_reason=None, + created_at=now, + updated_at=now, + closed_at=None, + ) + ) + session.add( + VoiceAISessionRow( + session_id=session_id, + call_id=call_id, + linked_id=f"linked_voice_truth_{tmp_path.name}", + interaction_id=interaction_id, + customer_id="cus_voice_truth", + queue_id="que_voice_lab", + ai_session_id=ai_session_id, + agent_profile="voice_support", + language="ru", + asr_provider="elevenlabs", + tts_provider="elevenlabs", + status="active", + handoff_reason=None, + handoff_target_queue_id="que_voice_lab", + disclosure_played_at=now, + last_user_utterance_at=now, + last_ai_reply_at=now, + started_at=now, + updated_at=now, + ended_at=None, + voice_start_language="ru", + ) + ) + session.add_all( + [ + VoiceTranscriptSegmentRow( + segment_id=f"{session_id}_seg_1", + session_id=session_id, + call_id=call_id, + interaction_id=interaction_id, + sequence_no=1, + speaker="caller", + source_type="voice_asr", + text="Мне надо узнать график работы.", + confidence=0.96, + is_final=True, + barge_in_interrupted=False, + payload_json='{"intent_bearing": true, "provider": "elevenlabs"}', + created_at=now, + ), + VoiceTranscriptSegmentRow( + segment_id=f"{session_id}_seg_2", + session_id=session_id, + call_id=call_id, + interaction_id=interaction_id, + sequence_no=2, + speaker="assistant", + source_type="voice_policy", + text="Подскажите, пожалуйста, город или филиал.", + confidence=None, + is_final=True, + barge_in_interrupted=False, + payload_json='{"delivery_state": "delivered", "provider": "elevenlabs"}', + created_at=now, + ), + VoiceTranscriptSegmentRow( + segment_id=f"{session_id}_seg_3", + session_id=session_id, + call_id=call_id, + interaction_id=interaction_id, + sequence_no=3, + speaker="caller", + source_type="voice_asr", + text="Алло, приветствую.", + confidence=0.8, + is_final=True, + barge_in_interrupted=False, + payload_json='{"intent_bearing": false, "provider": "elevenlabs"}', + created_at=now, + ), + VoiceTranscriptSegmentRow( + segment_id=f"{session_id}_seg_4", + session_id=session_id, + call_id=call_id, + interaction_id=interaction_id, + sequence_no=4, + speaker="assistant", + source_type="voice_policy", + text="Это прерванный ответ.", + confidence=None, + is_final=True, + barge_in_interrupted=True, + payload_json='{"delivery_state": "interrupted", "provider": "elevenlabs"}', + created_at=now, + ), + ] + ) + session.commit() + finally: + session.close() + + summary = bridge_module._voice_ai_summary_for_call(call_id) + + assert summary is not None + assert summary.customer_request_text == "Мне надо узнать график работы." + assert summary.ai_outcome_text == "Подскажите, пожалуйста, город или филиал." + assert [(segment.speaker, segment.text) for segment in summary.transcript_segments] == [ + ("caller", "Мне надо узнать график работы."), + ("assistant", "Подскажите, пожалуйста, город или филиал."), + ("caller", "Алло, приветствую."), + ] def test_call_started_re_raises_original_create_error_when_all_fallbacks_fail(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')