From cde9eb6f0700d3cc2df9c6930e721394943b106c Mon Sep 17 00:00:00 2001 From: Magzhan Zhumabayev Date: Sat, 2 May 2026 01:07:38 +0500 Subject: [PATCH] . --- .env.example | 4 +- main.py | 5 +- providers/tts.py | 231 ++++++++++++++++++++++++++++++++++++++++------- 3 files changed, 206 insertions(+), 34 deletions(-) diff --git a/.env.example b/.env.example index af12e10..c16cabf 100644 --- a/.env.example +++ b/.env.example @@ -43,7 +43,9 @@ AUDIO_DUMP_DIR=debug_audio ELEVENLABS_API_KEY= ELEVENLABS_API_BASE=https://api.elevenlabs.io ELEVENLABS_TTS_VOICE_ID= -ELEVENLABS_TTS_MODEL_ID=eleven_flash_v2_5 +# Use eleven_v3 for Kazakh/Russian. Switch back to v2 anytime with eleven_flash_v2_5 or eleven_multilingual_v2. +ELEVENLABS_TTS_MODEL_ID=eleven_v3 +ELEVENLABS_TTS_TRANSPORT=auto ELEVENLABS_TTS_LANGUAGE_CODE=ru ELEVENLABS_TTS_OUTPUT_FORMAT=pcm_16000 ELEVENLABS_TTS_SPEED=1.2 diff --git a/main.py b/main.py index be89ed8..4d3af2e 100644 --- a/main.py +++ b/main.py @@ -127,7 +127,7 @@ class RealtimeVoiceService: LOGGER.info( "realtime voice service config: sample_rate=%s audiosocket=%s:%s http=%s:%s " "llm_provider=%s llm_model=%s tools_enabled=%s serper_configured=%s stt_provider=%s stt_model=%s stt_realtime_model=%s " - "tts_voice_id=%s tts_model=%s tts_format=%s tts_speed=%s", + "tts_voice_id=%s tts_model=%s tts_transport=%s tts_format=%s tts_speed=%s", self._sample_rate_hz, _audiosocket_host(), _audiosocket_port(), @@ -141,7 +141,8 @@ class RealtimeVoiceService: os.getenv("ELEVENLABS_STT_MODEL_ID", "scribe_v2"), os.getenv("ELEVENLABS_STT_REALTIME_MODEL_ID", "scribe_v2_realtime"), os.getenv("ELEVENLABS_TTS_VOICE_ID", ""), - os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_turbo_v2_5"), + os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_v3"), + os.getenv("ELEVENLABS_TTS_TRANSPORT", "auto"), f"pcm_{self._sample_rate_hz}", os.getenv("ELEVENLABS_TTS_SPEED", "1.0"), ) diff --git a/providers/tts.py b/providers/tts.py index df3bd7c..99ff053 100644 --- a/providers/tts.py +++ b/providers/tts.py @@ -82,6 +82,22 @@ def _preview_text(text: str, *, limit: int = 120) -> str: return f"{normalized[:limit]}..." +def _tts_transport_mode() -> str: + raw = str(os.getenv("ELEVENLABS_TTS_TRANSPORT", "auto")).strip().lower().replace("-", "_") + aliases = { + "auto": "auto", + "ws": "websocket", + "websocket": "websocket", + "http": "http_stream", + "stream": "http_stream", + "http_stream": "http_stream", + } + if raw in aliases: + return aliases[raw] + LOGGER.warning("invalid ELEVENLABS_TTS_TRANSPORT=%r; using auto", raw) + return "auto" + + class ElevenLabsTTS(BaseTTS): def __init__( self, @@ -103,18 +119,16 @@ class ElevenLabsTTS(BaseTTS): self._ws_base = _ws_base(self._api_base) self._voice_id = str(voice_id or os.getenv("ELEVENLABS_TTS_VOICE_ID", "")).strip() self._requested_model_id = ( - str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_multilingual_v2")).strip() - or "eleven_multilingual_v2" - ) - self._websocket_fallback_model_id = ( - str(os.getenv("ELEVENLABS_TTS_WS_FALLBACK_MODEL_ID", "eleven_multilingual_v2")).strip() - or "eleven_multilingual_v2" - ) - self._model_id = self._resolve_websocket_model_id( - requested_model_id=self._requested_model_id, - fallback_model_id=self._websocket_fallback_model_id, + str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_v3")).strip() + or "eleven_v3" ) + self._model_id = self._requested_model_id self._language_code = str(language_code or os.getenv("ELEVENLABS_TTS_LANGUAGE_CODE", "ru")).strip() or None + self._transport_mode = _tts_transport_mode() + self._apply_text_normalization = ( + str(os.getenv("ELEVENLABS_TTS_APPLY_TEXT_NORMALIZATION", "auto")).strip().lower() + or "auto" + ) requested_output_format = ( str(output_format or os.getenv("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000")).strip().lower() or "pcm_16000" @@ -151,17 +165,18 @@ class ElevenLabsTTS(BaseTTS): "use_speaker_boost": self._read_bool_env("ELEVENLABS_TTS_USE_SPEAKER_BOOST", True), } LOGGER.info( - "ElevenLabs TTS config: voice_id=%s requested_model=%s websocket_model=%s " + "ElevenLabs TTS config: voice_id=%s model=%s transport=%s " "provider_output_format=%s provider_sample_rate=%s target_sample_rate=%s " - "language=%s auto_mode=%s chunk_schedule=%s voice_settings=%s", + "language=%s auto_mode=%s text_normalization=%s chunk_schedule=%s voice_settings=%s", self._voice_id, - self._requested_model_id, self._model_id, + self._resolved_transport_mode(), self._output_format, self._provider_sample_rate_hz, self._target_sample_rate_hz, self._language_code, self._auto_mode, + self._apply_text_normalization, self._chunk_length_schedule, self._voice_settings, ) @@ -177,14 +192,34 @@ class ElevenLabsTTS(BaseTTS): if not self._voice_id: raise RuntimeError("ELEVENLABS_TTS_VOICE_ID is required for ElevenLabs TTS") + effective_language_code = (str(language_code).strip() if language_code else "") or self._language_code + if self._resolved_transport_mode() == "http_stream": + async for audio_chunk in self._synthesize_http_stream( + text_stream, + language_code=effective_language_code, + ): + yield audio_chunk + return + + async for audio_chunk in self._synthesize_websocket_stream( + text_stream, + language_code=effective_language_code, + ): + yield audio_chunk + + async def _synthesize_websocket_stream( + self, + text_stream: AsyncIterable[str], + *, + language_code: str | None = None, + ) -> AsyncGenerator[bytes, None]: try: from websockets.exceptions import WebSocketException from websockets.legacy.client import connect as websocket_connect except Exception as exc: # noqa: BLE001 raise RuntimeError("The `websockets` package is required for ElevenLabs TTS WebSocket streaming") from exc - effective_language_code = (str(language_code).strip() if language_code else "") or self._language_code - websocket_url = self._build_websocket_url(language_code=effective_language_code) + websocket_url = self._build_websocket_url(language_code=language_code) started_monotonic = time.perf_counter() audio_chunk_count = 0 audio_byte_count = 0 @@ -199,7 +234,7 @@ class ElevenLabsTTS(BaseTTS): self._output_format, self._provider_sample_rate_hz, self._target_sample_rate_hz, - effective_language_code, + language_code, self._auto_mode, ) try: @@ -296,6 +331,128 @@ class ElevenLabsTTS(BaseTTS): except OSError as exc: raise RuntimeError("ElevenLabs TTS WebSocket connection failed") from exc + async def _synthesize_http_stream( + self, + text_stream: AsyncIterable[str], + *, + language_code: str | None = None, + ) -> AsyncGenerator[bytes, None]: + try: + import aiohttp + except Exception as exc: # noqa: BLE001 + raise RuntimeError("The `aiohttp` package is required for ElevenLabs TTS HTTP streaming") from exc + + timeout = aiohttp.ClientTimeout(total=self._timeout_seconds, sock_read=self._timeout_seconds) + connector = aiohttp.TCPConnector(limit=16, ttl_dns_cache=300) + url = self._build_http_stream_url() + headers = { + "xi-api-key": self._api_key, + "Content-Type": "application/json", + } + request_count = 0 + audio_chunk_count = 0 + audio_byte_count = 0 + yielded_byte_count = 0 + resample_state = None + pcm_remainder = b"" + stream_started_monotonic = time.perf_counter() + LOGGER.info( + "ElevenLabs TTS HTTP streaming: voice_id=%s model=%s output_format=%s " + "provider_sample_rate=%s target_sample_rate=%s language=%s", + self._voice_id, + self._model_id, + self._output_format, + self._provider_sample_rate_hz, + self._target_sample_rate_hz, + language_code, + ) + + try: + async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: + async for raw_chunk in text_stream: + text = str(raw_chunk).strip() + if not text: + continue + request_count += 1 + request_started_monotonic = time.perf_counter() + request_provider_bytes = 0 + request_yielded_bytes = 0 + payload = self._http_stream_payload(text=text, language_code=language_code) + LOGGER.info( + "ElevenLabs TTS HTTP request start: voice_id=%s model=%s index=%s chars=%s preview=%r", + self._voice_id, + self._model_id, + request_count, + len(text), + _preview_text(text), + ) + async with session.post(url, headers=headers, json=payload) as response: + if response.status >= 400: + payload_text = await response.text() + raise RuntimeError( + f"ElevenLabs TTS HTTP {response.status}: {payload_text[:300]}" + ) + async for audio_chunk in response.content.iter_chunked(4096): + if not audio_chunk: + continue + audio_chunk_count += 1 + audio_byte_count += len(audio_chunk) + request_provider_bytes += len(audio_chunk) + audio_chunk, resample_state, pcm_remainder = self._normalize_pcm_chunk( + audio_chunk, + resample_state=resample_state, + pcm_remainder=pcm_remainder, + ) + yielded_byte_count += len(audio_chunk) + request_yielded_bytes += len(audio_chunk) + if audio_chunk_count == 1: + LOGGER.info( + "ElevenLabs TTS first audio: voice_id=%s model=%s transport=http_stream " + "ttfa_ms=%s provider_bytes=%s yielded_bytes=%s", + self._voice_id, + self._model_id, + int((time.perf_counter() - stream_started_monotonic) * 1000.0), + audio_byte_count, + len(audio_chunk), + ) + elif audio_chunk_count % 10 == 0: + LOGGER.info( + "ElevenLabs TTS audio summary: voice_id=%s chunks=%s " + "provider_bytes=%s yielded_bytes=%s", + self._voice_id, + audio_chunk_count, + audio_byte_count, + yielded_byte_count, + ) + if audio_chunk: + yield audio_chunk + LOGGER.info( + "ElevenLabs TTS HTTP request completed: voice_id=%s model=%s index=%s " + "provider_bytes=%s yielded_bytes=%s total_ms=%s", + self._voice_id, + self._model_id, + request_count, + request_provider_bytes, + request_yielded_bytes, + int((time.perf_counter() - request_started_monotonic) * 1000.0), + ) + except asyncio.TimeoutError as exc: + raise RuntimeError("ElevenLabs TTS HTTP stream timed out") from exc + except aiohttp.ClientError as exc: + raise RuntimeError("ElevenLabs TTS HTTP stream failed") from exc + + LOGGER.info( + "ElevenLabs TTS HTTP stream completed: voice_id=%s model=%s requests=%s chunks=%s " + "provider_bytes=%s yielded_bytes=%s total_ms=%s", + self._voice_id, + self._model_id, + request_count, + audio_chunk_count, + audio_byte_count, + yielded_byte_count, + int((time.perf_counter() - stream_started_monotonic) * 1000.0), + ) + def _build_websocket_url(self, *, language_code: str | None = None) -> str: query = { "model_id": self._model_id, @@ -303,7 +460,7 @@ class ElevenLabsTTS(BaseTTS): "inactivity_timeout": self._inactivity_timeout_seconds, "auto_mode": str(self._auto_mode).lower(), "sync_alignment": "false", - "apply_text_normalization": "auto", + "apply_text_normalization": self._apply_text_normalization, } effective_language = (str(language_code).strip() if language_code else "") or self._language_code if effective_language: @@ -311,6 +468,32 @@ class ElevenLabsTTS(BaseTTS): encoded_voice_id = quote(self._voice_id, safe="") return f"{self._ws_base}/v1/text-to-speech/{encoded_voice_id}/stream-input?{urlencode(query)}" + def _build_http_stream_url(self) -> str: + query = { + "output_format": self._output_format, + } + encoded_voice_id = quote(self._voice_id, safe="") + return f"{self._api_base}/v1/text-to-speech/{encoded_voice_id}/stream?{urlencode(query)}" + + def _http_stream_payload(self, *, text: str, language_code: str | None = None) -> dict[str, object]: + payload: dict[str, object] = { + "text": text, + "model_id": self._model_id, + "voice_settings": self._voice_settings, + "apply_text_normalization": self._apply_text_normalization, + } + effective_language = (str(language_code).strip() if language_code else "") or self._language_code + if effective_language: + payload["language_code"] = effective_language + return payload + + def _resolved_transport_mode(self) -> str: + if self._transport_mode != "auto": + return self._transport_mode + if self._model_id == "eleven_v3": + return "http_stream" + return "websocket" + def _initial_payload(self) -> dict[str, object]: payload: dict[str, object] = { "text": " ", @@ -373,20 +556,6 @@ class ElevenLabsTTS(BaseTTS): return text return f"{text} " - @staticmethod - def _resolve_websocket_model_id(*, requested_model_id: str, fallback_model_id: str) -> str: - normalized_requested = requested_model_id.strip() or "eleven_multilingual_v2" - if normalized_requested not in {"eleven_v3", "eleven_ttv_v3"}: - return normalized_requested - - normalized_fallback = fallback_model_id.strip() or "eleven_multilingual_v2" - LOGGER.warning( - "ElevenLabs WebSocket TTS does not support model_id=%s; falling back to model_id=%s", - normalized_requested, - normalized_fallback, - ) - return normalized_fallback - @staticmethod def _resolve_provider_output_format(*, requested_output_format: str, target_sample_rate_hz: int) -> str: normalized = requested_output_format.strip().lower() or "pcm_16000"