digiops/call-center
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(voice): add elevenlabs stt and transcript truth fixes

Browse Source
This commit is contained in:
Yera All
2026-04-18 18:25:07 +05:00
parent ac2269b6a4
commit 34b807e460
9 changed files with 807 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+5
View File
@@ -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
services/ai_orchestrator_service/voice.py
+58 -5
View File
@@ -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,
services/ai_voice_runtime_service/app.py
+136 -3
View File
@@ -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,
services/ai_voice_runtime_service/media_runtime.py
+24 -1
View File
@@ -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
services/ai_voice_runtime_service/providers/asr.py
+150
View File
@@ -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"}:
services/asterisk_bridge_service/voice_ai.py
+136 -5
View File
@@ -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(
tests/test_ai_orchestrator_service.py
+32 -2
View File
@@ -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"),
[
tests/test_ai_voice_asr_provider.py
+97
View File
@@ -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(
tests/test_asterisk_bridge_service.py
+169 -2
View File
@@ -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"}')