diff --git a/.env.example b/.env.example index 12ebe03..f1b375c 100644 --- a/.env.example +++ b/.env.example @@ -52,13 +52,13 @@ AI_VOICE_RUNTIME_SERVICE_URL=http://localhost:8018 AI_VOICE_ENABLED=0 AI_VOICE_POLICY_MODE=v2_fast_conversational AI_VOICE_QUEUE_CONFIG_JSON={} -AI_VOICE_ASR_PROVIDER=yandex -AI_VOICE_TTS_PROVIDER=yandex +AI_VOICE_ASR_PROVIDER=elevenlabs +AI_VOICE_TTS_PROVIDER=elevenlabs 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=yandex_speechkit +AI_VOICE_V2_STREAMING_ASR_BACKEND=elevenlabs_realtime AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://127.0.0.1:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS=4 AI_VOICE_V2_STREAMING_ASR_MODEL=base @@ -81,6 +81,10 @@ 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_ELEVENLABS_REALTIME_MODEL_ID=scribe_v2_realtime +AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT=pcm_16000 +AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY=manual +AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS=2.5 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/.env.production.template b/.env.production.template index 61bc0cc..dc9a595 100644 --- a/.env.production.template +++ b/.env.production.template @@ -47,13 +47,13 @@ AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65 AI_VOICE_ENABLED=0 AI_VOICE_POLICY_MODE=v2_fast_conversational AI_VOICE_QUEUE_CONFIG_JSON={"voice_lab_ai":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}} -AI_VOICE_ASR_PROVIDER=openai -AI_VOICE_TTS_PROVIDER=yandex +AI_VOICE_ASR_PROVIDER=elevenlabs +AI_VOICE_TTS_PROVIDER=elevenlabs 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_BACKEND=elevenlabs_realtime AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS=4 AI_VOICE_V2_STREAMING_ASR_MODEL=base @@ -71,6 +71,15 @@ AI_VOICE_AUDIOSOCKET_ENABLED=1 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_ELEVENLABS_REALTIME_MODEL_ID=scribe_v2_realtime +AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT=pcm_16000 +AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY=manual +AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS=2.5 AI_VOICE_TTS_MODEL=gpt-4o-mini-tts AI_VOICE_TTS_VOICE=alloy AI_VOICE_TTS_YANDEX_API_BASE=https://tts.api.ml.yandexcloud.kz diff --git a/deployment/docker-compose.server.yml b/deployment/docker-compose.server.yml index 2dd3ffb..d5a052c 100644 --- a/deployment/docker-compose.server.yml +++ b/deployment/docker-compose.server.yml @@ -22,9 +22,9 @@ x-app-env: &app_env KB_SERVICE_URL: http://kb-service:8000 REPORTING_SERVICE_URL: http://reporting-service:8000 SUPERVISOR_SERVICE_URL: http://supervisor-service:8000 - AI_VOICE_ASR_PROVIDER: ${AI_VOICE_ASR_PROVIDER:-yandex} - AI_VOICE_TTS_PROVIDER: ${AI_VOICE_TTS_PROVIDER:-yandex} - AI_VOICE_V2_STREAMING_ASR_BACKEND: ${AI_VOICE_V2_STREAMING_ASR_BACKEND:-yandex_speechkit} + AI_VOICE_ASR_PROVIDER: ${AI_VOICE_ASR_PROVIDER:-elevenlabs} + AI_VOICE_TTS_PROVIDER: ${AI_VOICE_TTS_PROVIDER:-elevenlabs} + AI_VOICE_V2_STREAMING_ASR_BACKEND: ${AI_VOICE_V2_STREAMING_ASR_BACKEND:-elevenlabs_realtime} AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1" diff --git a/requirements.txt b/requirements.txt index fbe544f..b6d7380 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,3 +14,4 @@ grpcio-tools==1.80.0 protobuf==6.33.6 requests==2.32.5 yandexcloud==0.386.0 +websocket-client==1.8.0 diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index 4a4bb86..3c3caf9 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -164,7 +164,7 @@ def _voice_v2_duplex_enabled() -> bool: def _voice_v2_streaming_asr_backend() -> str: - return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_BACKEND", "yandex_speechkit") or "yandex_speechkit").strip().lower() + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_BACKEND", "elevenlabs_realtime") or "elevenlabs_realtime").strip().lower() def _voice_v2_prebaked_ack_enabled() -> bool: diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 08df5b5..b4d445e 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -3,12 +3,15 @@ from __future__ import annotations import audioop import base64 import io +import json import os import queue import threading import time +import urllib.parse import uuid import wave +from collections.abc import Callable from dataclasses import dataclass, field import httpx @@ -89,6 +92,73 @@ def _elevenlabs_asr_sample_rate_hz() -> int: return value +def _elevenlabs_realtime_model_id() -> str: + return ( + os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_MODEL_ID", "scribe_v2_realtime").strip() + or "scribe_v2_realtime" + ) + + +def _elevenlabs_realtime_audio_format() -> str: + return ( + os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT", "pcm_16000").strip().lower() + or "pcm_16000" + ) + + +def _elevenlabs_realtime_sample_rate_hz(audio_format: str | None = None) -> int: + explicit = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_SAMPLE_RATE_HZ", "").strip() + if explicit: + try: + value = int(explicit) + except ValueError: + value = 16000 + return value if 8000 <= value <= 48000 else 16000 + + normalized = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower() + if normalized.startswith("pcm_"): + try: + value = int(normalized.rsplit("_", 1)[-1]) + except ValueError: + value = 16000 + return value if 8000 <= value <= 48000 else 16000 + return 16000 + + +def _elevenlabs_realtime_commit_strategy() -> str: + raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY", "manual").strip().lower() + return raw if raw in {"manual", "vad"} else "manual" + + +def _elevenlabs_realtime_finalize_timeout_seconds() -> float: + raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS", "").strip() + if not raw: + return max(_streaming_asr_timeout_seconds(), 1.0) + try: + value = float(raw) + except ValueError: + value = _streaming_asr_timeout_seconds() + return max(value, 1.0) + + +def _normalize_elevenlabs_realtime_language(language: str | None) -> str: + raw = str(language or "").strip().lower().replace("_", "-") + if not raw: + raw = _elevenlabs_asr_default_language().strip().lower().replace("_", "-") + mapping = { + "rus": "ru", + "ru-ru": "ru", + "kaz": "kk", + "kz": "kk", + "kk-kz": "kk", + "kz-kz": "kk", + "eng": "en", + "en-us": "en", + "en-gb": "en", + } + return mapping.get(raw, raw or "ru") + + 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"}: @@ -434,6 +504,330 @@ class ElevenLabsASRProvider(ASRProvider): return self.transcribe(audio_bytes, language_hint=language_hint) +@dataclass +class _ElevenLabsRealtimeStreamState: + stream_id: str + session_id: str + language: str + websocket: object + updates_queue: queue.Queue[StreamingASRPartial] + thread: threading.Thread | None = None + latest_partial: StreamingASRPartial | None = None + final_transcription: ASRTranscription | None = None + error: BaseException | None = None + close_requested: bool = False + final_event: threading.Event = field(default_factory=threading.Event) + lock: threading.Lock = field(default_factory=threading.Lock) + send_lock: threading.Lock = field(default_factory=threading.Lock) + + +class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): + name = "elevenlabs-realtime" + supports_streaming = True + + def __init__( + self, + *, + api_base: str | None = None, + api_key: str | None = None, + timeout_seconds: float | None = None, + finalize_timeout_seconds: float | None = None, + model_id: str | None = None, + audio_format: str | None = None, + commit_strategy: str | None = None, + websocket_factory: Callable[..., object] | 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 _streaming_asr_timeout_seconds()), + 0.25, + ) + self._finalize_timeout_seconds = max( + float( + finalize_timeout_seconds + if finalize_timeout_seconds is not None + else _elevenlabs_realtime_finalize_timeout_seconds() + ), + 1.0, + ) + self._model_id = str(model_id or _elevenlabs_realtime_model_id()).strip() or "scribe_v2_realtime" + self._audio_format = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower() or "pcm_16000" + self._sample_rate_hz = _elevenlabs_realtime_sample_rate_hz(self._audio_format) + self._commit_strategy = str(commit_strategy or _elevenlabs_realtime_commit_strategy()).strip().lower() + if self._commit_strategy not in {"manual", "vad"}: + self._commit_strategy = "manual" + self._websocket_factory = websocket_factory + self._streams: dict[str, _ElevenLabsRealtimeStreamState] = {} + self._lock = threading.Lock() + + def _websocket_url(self, *, language: str) -> str: + base = self._api_base + if base.startswith("https://"): + websocket_base = "wss://" + base[len("https://") :] + elif base.startswith("http://"): + websocket_base = "ws://" + base[len("http://") :] + else: + websocket_base = base + params = { + "model_id": self._model_id, + "audio_format": self._audio_format, + "language_code": language, + "commit_strategy": self._commit_strategy, + "include_timestamps": "false", + "include_language_detection": "false", + } + return f"{websocket_base}/v1/speech-to-text/realtime?{urllib.parse.urlencode(params)}" + + def _create_websocket(self, url: str) -> object: + if not self._api_key: + raise StreamingASRUnavailable( + "AI_VOICE_ASR_ELEVENLABS_API_KEY or AI_VOICE_TTS_ELEVENLABS_API_KEY is required for ElevenLabs realtime ASR" + ) + factory = self._websocket_factory + if factory is None: + try: + import websocket # type: ignore[import-not-found] + except ImportError as exc: + raise StreamingASRUnavailable("websocket-client is required for ElevenLabs realtime ASR") from exc + factory = websocket.create_connection + try: + return factory( + url, + header=[f"xi-api-key: {self._api_key}"], + timeout=self._timeout_seconds, + ) + except BaseException as exc: + raise StreamingASRUnavailable(str(exc)[:500] or "ElevenLabs realtime ASR is unavailable") from exc + + @staticmethod + def _message_type(payload: object) -> str: + if not isinstance(payload, dict): + return "" + return str(payload.get("message_type") or payload.get("type") or "").strip() + + @staticmethod + def _payload_text(payload: object) -> str: + if not isinstance(payload, dict): + return "" + return str(payload.get("text") or payload.get("transcript") or "").strip() + + @staticmethod + def _payload_language(payload: object, fallback: str) -> str: + if not isinstance(payload, dict): + return fallback + return str(payload.get("language_code") or payload.get("language") or fallback).strip() or fallback + + @staticmethod + def _payload_error_message(payload: object) -> str: + if not isinstance(payload, dict): + return "ElevenLabs realtime ASR error" + for key in ("message", "error", "reason", "detail"): + value = payload.get(key) + if value: + return str(value)[:500] + return json.dumps(payload, ensure_ascii=True)[:500] + + def _send_json(self, state: _ElevenLabsRealtimeStreamState, payload: dict[str, object]) -> None: + with state.lock: + if state.error is not None: + raise StreamingASRUnavailable(str(state.error)[:500]) + if state.close_requested: + raise StreamingASRUnavailable("ElevenLabs realtime ASR stream is closed") + try: + with state.send_lock: + state.websocket.send(json.dumps(payload, separators=(",", ":"))) # type: ignore[attr-defined] + except BaseException as exc: + with state.lock: + state.error = exc + state.final_event.set() + raise StreamingASRUnavailable(str(exc)[:500] or "ElevenLabs realtime ASR send failed") from exc + + def _drain_updates(self, state: _ElevenLabsRealtimeStreamState) -> None: + while True: + try: + update = state.updates_queue.get_nowait() + except queue.Empty: + return + with state.lock: + state.latest_partial = update + if update.is_final: + state.final_transcription = ASRTranscription( + text=update.text, + language=update.language or state.language, + confidence=update.confidence, + ) + state.final_event.set() + + def _run_reader(self, state: _ElevenLabsRealtimeStreamState) -> None: + try: + while True: + with state.lock: + if state.close_requested: + return + raw_message = state.websocket.recv() # type: ignore[attr-defined] + if raw_message is None: + continue + if isinstance(raw_message, bytes): + raw_message = raw_message.decode("utf-8", errors="replace") + try: + payload = json.loads(str(raw_message)) + except json.JSONDecodeError: + continue + message_type = self._message_type(payload) + if message_type == "partial_transcript": + text = self._payload_text(payload) + if text: + state.updates_queue.put( + StreamingASRPartial( + text=text, + language=self._payload_language(payload, state.language), + confidence=None, + is_final=False, + is_stable=False, + ) + ) + continue + if message_type in {"committed_transcript", "committed_transcript_with_timestamps"}: + text = self._payload_text(payload) + language = self._payload_language(payload, state.language) + if text: + state.updates_queue.put( + StreamingASRPartial( + text=text, + language=language, + confidence=None, + is_final=True, + is_stable=True, + ) + ) + else: + with state.lock: + state.final_event.set() + continue + if message_type.startswith("scribe") and "error" in message_type.lower(): + raise StreamingASRUnavailable(self._payload_error_message(payload)) + except BaseException as exc: + with state.lock: + if not state.close_requested: + state.error = exc + state.final_event.set() + + def _state(self, stream_id: str) -> _ElevenLabsRealtimeStreamState: + with self._lock: + state = self._streams.get(stream_id) + if state is None: + raise StreamingASRUnavailable("ElevenLabs realtime ASR stream is not active") + return state + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + stream_id = f"elrt_{uuid.uuid4().hex}" + language = _normalize_elevenlabs_realtime_language(language_hint) + websocket = self._create_websocket(self._websocket_url(language=language)) + state = _ElevenLabsRealtimeStreamState( + stream_id=stream_id, + session_id=str(session_id or "").strip(), + language=language, + websocket=websocket, + updates_queue=queue.Queue(), + ) + thread = threading.Thread( + target=self._run_reader, + args=(state,), + name=f"elevenlabs-realtime-asr-{stream_id[:16]}", + daemon=True, + ) + state.thread = thread + with self._lock: + self._streams[stream_id] = state + thread.start() + return stream_id + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + if not pcm_8k_chunk: + return + state = self._state(stream_id) + pcm_bytes = pcm_8k_chunk + if self._sample_rate_hz != 8000: + pcm_bytes = resample_pcm16le( + pcm_8k_chunk, + input_rate_hz=8000, + output_rate_hz=self._sample_rate_hz, + ) + self._send_json( + state, + { + "message_type": "input_audio_chunk", + "audio_base_64": base64.b64encode(pcm_bytes).decode("ascii"), + "commit": False, + "sample_rate": self._sample_rate_hz, + }, + ) + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + state = self._state(stream_id) + self._drain_updates(state) + with state.lock: + if state.error is not None and state.latest_partial is None: + raise StreamingASRUnavailable(str(state.error)[:500]) + return state.latest_partial + + def finalize(self, stream_id: str) -> ASRTranscription: + state = self._state(stream_id) + silence = b"\x00\x00" * int(self._sample_rate_hz * 0.1) + self._send_json( + state, + { + "message_type": "input_audio_chunk", + "audio_base_64": base64.b64encode(silence).decode("ascii"), + "commit": True, + "sample_rate": self._sample_rate_hz, + }, + ) + deadline = time.monotonic() + self._finalize_timeout_seconds + while True: + self._drain_updates(state) + with state.lock: + if state.final_transcription is not None: + return state.final_transcription + latest_partial = state.latest_partial + error = state.error + event_is_set = state.final_event.is_set() + if event_is_set or time.monotonic() >= deadline: + if latest_partial is not None: + return ASRTranscription( + text=latest_partial.text, + language=latest_partial.language or state.language, + confidence=latest_partial.confidence, + ) + if error is not None: + raise StreamingASRUnavailable(str(error)[:500]) + if time.monotonic() >= deadline: + raise StreamingASRUnavailable("ElevenLabs realtime ASR finalize timed out") + return ASRTranscription(text="", language=state.language, confidence=None) + state.final_event.wait(timeout=0.05) + + def close_stream(self, stream_id: str) -> None: + try: + state = self._state(stream_id) + except StreamingASRUnavailable: + return + with state.lock: + state.close_requested = True + state.final_event.set() + close = getattr(state.websocket, "close", None) + if callable(close): + try: + close() + except BaseException: + pass + thread = state.thread + if thread is not None and thread.is_alive(): + thread.join(timeout=0.5) + with self._lock: + self._streams.pop(stream_id, None) + + class YandexSpeechKitASRProvider(ASRProvider): name = "yandex" @@ -1144,6 +1538,15 @@ def build_asr_provider(name: str) -> ASRProvider: def build_streaming_asr_provider(name: str) -> StreamingASRProvider: normalized = str(name or "disabled").strip().lower() + if normalized in { + "elevenlabs", + "elevenlabs_realtime", + "elevenlabs-realtime", + "scribe_realtime", + "scribe-v2-realtime", + "scribe_v2_realtime", + }: + return ElevenLabsRealtimeStreamingASRProvider() if normalized in {"local_sidecar", "local-sidecar", "sidecar"}: return LocalSidecarStreamingASRProvider() if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit", "yandex_grpc"}: diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index 7783a71..7834313 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -1,5 +1,7 @@ from __future__ import annotations +import json +import queue import time import httpx @@ -301,6 +303,102 @@ def test_yandex_buffered_streaming_provider_buffers_pcm_until_finalize(): assert calls == [{"audio_bytes": b"\x10\x00\x20\x00", "language_hint": "ru"}] +def test_elevenlabs_realtime_streaming_provider_returns_partial_and_final(): + calls: list[dict] = [] + + class _FakeRealtimeWebSocket: + def __init__(self) -> None: + self.sent_payloads: list[dict] = [] + self.incoming: queue.Queue[str | None] = queue.Queue() + self.incoming.put( + json.dumps( + { + "message_type": "session_started", + "session_id": "session-test", + "config": {"sample_rate": 16000, "audio_format": "pcm_16000"}, + } + ) + ) + + def send(self, raw_payload: str) -> None: + payload = json.loads(raw_payload) + self.sent_payloads.append(payload) + if payload.get("commit"): + self.incoming.put( + json.dumps( + { + "message_type": "committed_transcript", + "text": "нужно узнать график работы", + "language_code": "ru", + } + ) + ) + else: + self.incoming.put( + json.dumps( + { + "message_type": "partial_transcript", + "text": "нужно узнать график", + "language_code": "ru", + } + ) + ) + + def recv(self) -> str: + item = self.incoming.get(timeout=1) + if item is None: + raise RuntimeError("closed") + return item + + def close(self) -> None: + self.incoming.put(None) + + websocket = _FakeRealtimeWebSocket() + + def _factory(url: str, *, header: list[str], timeout: float): + calls.append({"url": url, "header": header, "timeout": timeout}) + return websocket + + provider = asr_module.ElevenLabsRealtimeStreamingASRProvider( + api_base="https://api.elevenlabs.example", + api_key="asr-key", + timeout_seconds=3, + finalize_timeout_seconds=1, + websocket_factory=_factory, + ) + stream_id = provider.open_stream("session-1", language_hint="ru") + provider.push_pcm(stream_id, b"\x01\x00" * 160) + + partial = None + for _ in range(20): + partial = provider.poll_partial(stream_id) + if partial is not None: + break + time.sleep(0.02) + + assert partial is not None + assert partial.text == "нужно узнать график" + assert partial.is_final is False + + final = provider.finalize(stream_id) + provider.close_stream(stream_id) + + assert final.text == "нужно узнать график работы" + assert final.language == "ru" + assert calls[0]["url"].startswith("wss://api.elevenlabs.example/v1/speech-to-text/realtime?") + assert "model_id=scribe_v2_realtime" in calls[0]["url"] + assert "audio_format=pcm_16000" in calls[0]["url"] + assert "language_code=ru" in calls[0]["url"] + assert calls[0]["header"] == ["xi-api-key: asr-key"] + assert calls[0]["timeout"] == 3.0 + assert len(websocket.sent_payloads) == 2 + assert websocket.sent_payloads[0]["message_type"] == "input_audio_chunk" + assert websocket.sent_payloads[0]["commit"] is False + assert websocket.sent_payloads[0]["sample_rate"] == 16000 + assert websocket.sent_payloads[0]["audio_base_64"] + assert websocket.sent_payloads[1]["commit"] is True + + def test_yandex_grpc_streaming_provider_returns_partial_and_final(): calls: list[dict] = [] @@ -424,6 +522,10 @@ 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( + asr_module.build_streaming_asr_provider("elevenlabs_realtime"), + asr_module.ElevenLabsRealtimeStreamingASRProvider, + ) assert isinstance( asr_module.build_streaming_asr_provider("yandex_speechkit"), asr_module.YandexSpeechKitGrpcStreamingASRProvider,