arys
fe8598e09f
RuntimeConfiguredTTSProvider (the live 'dynamic' TTS provider, resolves voice from DB-backed config) never overrode cache_fingerprint(), so it fell back to the base class's empty string. PrebakedAckBank keys its on-disk cache on that fingerprint, so short filler phrases like 'Секунду' kept serving audio baked with the previous ElevenLabs voice even after a voice change, while full LLM replies (cached inside the resolved provider itself, keyed on its own voice id) already used the new voice — explaining why callers heard two different voices in the same call. Fix: delegate cache_fingerprint() to the resolved provider. Also extend the voice delivery_hint so the model transliterates website addresses and English words/abbreviations into spoken Cyrillic instead of leaving raw Latin text for the TTS engine to mangle (egov.kz was coming out as 'эговкз').
128 lines
5.1 KiB
Python
128 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from threading import Lock
|
|
|
|
from services.ai_voice_runtime_service.providers.tts import (
|
|
ElevenLabsTTSProvider,
|
|
OpenAITTSProvider,
|
|
TTSProvider,
|
|
YandexTTSProvider,
|
|
build_tts_provider,
|
|
)
|
|
from services.shared.db import get_session
|
|
from services.shared.voice_tts_config import load_effective_voice_tts_config
|
|
|
|
|
|
class RuntimeConfiguredTTSProvider(TTSProvider):
|
|
name = "dynamic"
|
|
|
|
def __init__(self, *, default_provider_name: str = "yandex") -> None:
|
|
self._default_provider_name = str(default_provider_name or "yandex").strip().lower() or "yandex"
|
|
self._provider_cache: dict[tuple[str, tuple[str | None, ...]], TTSProvider] = {}
|
|
self._lock = Lock()
|
|
|
|
def current_provider_name(self) -> str:
|
|
session = get_session()
|
|
try:
|
|
config = load_effective_voice_tts_config(session)
|
|
normalized = str(config.provider or self._default_provider_name).strip().lower()
|
|
if normalized in {"yandex", "elevenlabs", "openai"}:
|
|
return normalized
|
|
return self._default_provider_name
|
|
except Exception:
|
|
return self._default_provider_name
|
|
finally:
|
|
session.close()
|
|
|
|
def _provider_for_language(self, language: str | None) -> TTSProvider:
|
|
session = get_session()
|
|
try:
|
|
config = load_effective_voice_tts_config(session)
|
|
except Exception:
|
|
return build_tts_provider(self._default_provider_name)
|
|
finally:
|
|
session.close()
|
|
|
|
provider_name = str(config.provider or self._default_provider_name).strip().lower()
|
|
if provider_name not in {"yandex", "elevenlabs", "openai"}:
|
|
provider_name = self._default_provider_name
|
|
|
|
cache_key: tuple[str, tuple[str | None, ...]]
|
|
provider: TTSProvider
|
|
if provider_name == "yandex":
|
|
cache_key = (
|
|
provider_name,
|
|
(
|
|
config.yandex.ru.voice,
|
|
config.yandex.kz.voice,
|
|
),
|
|
)
|
|
with self._lock:
|
|
provider = self._provider_cache.get(cache_key)
|
|
if provider is None:
|
|
provider = YandexTTSProvider(
|
|
ru_voice=config.yandex.ru.voice,
|
|
kz_voice=config.yandex.kz.voice,
|
|
)
|
|
self._provider_cache[cache_key] = provider
|
|
return provider
|
|
|
|
if provider_name == "elevenlabs":
|
|
cache_key = (
|
|
provider_name,
|
|
(
|
|
config.elevenlabs.ru.voice,
|
|
config.elevenlabs.kz.voice,
|
|
config.elevenlabs.ru.model_id,
|
|
config.elevenlabs.kz.model_id,
|
|
config.elevenlabs.ru.language_code,
|
|
config.elevenlabs.kz.language_code,
|
|
),
|
|
)
|
|
with self._lock:
|
|
provider = self._provider_cache.get(cache_key)
|
|
if provider is None:
|
|
provider = ElevenLabsTTSProvider(
|
|
ru_voice=config.elevenlabs.ru.voice,
|
|
kz_voice=config.elevenlabs.kz.voice,
|
|
ru_model=config.elevenlabs.ru.model_id,
|
|
kz_model=config.elevenlabs.kz.model_id,
|
|
ru_language_code=config.elevenlabs.ru.language_code,
|
|
kz_language_code=config.elevenlabs.kz.language_code,
|
|
)
|
|
self._provider_cache[cache_key] = provider
|
|
return provider
|
|
|
|
cache_key = (
|
|
"openai",
|
|
(
|
|
config.openai.ru.voice,
|
|
config.openai.kz.voice,
|
|
config.openai.ru.model_id,
|
|
config.openai.kz.model_id,
|
|
),
|
|
)
|
|
with self._lock:
|
|
provider = self._provider_cache.get(cache_key)
|
|
if provider is None:
|
|
provider = OpenAITTSProvider(
|
|
ru_voice=config.openai.ru.voice,
|
|
kz_voice=config.openai.kz.voice,
|
|
ru_model=config.openai.ru.model_id,
|
|
kz_model=config.openai.kz.model_id,
|
|
)
|
|
self._provider_cache[cache_key] = provider
|
|
return provider
|
|
|
|
def cache_fingerprint(self, language: str | None, *, style_hints: dict[str, object] | None = None) -> str:
|
|
provider = self._provider_for_language(language)
|
|
return f"{provider.name}:{provider.cache_fingerprint(language, style_hints=style_hints)}"
|
|
|
|
def synthesize(self, text: str, *, language: str | None = None, style_hints: dict[str, object] | None = None):
|
|
provider = self._provider_for_language(language)
|
|
return provider.synthesize(text, language=language, style_hints=style_hints)
|
|
|
|
def synthesize_chunks(self, text: str, *, language: str | None = None, style_hints: dict[str, object] | None = None):
|
|
provider = self._provider_for_language(language)
|
|
yield from provider.synthesize_chunks(text, language=language, style_hints=style_hints)
|