Module livekit.plugins.upliftai
Uplift AI plugin for LiveKit Agents
See https://docs.livekit.io/agents/integrations/tts/upliftai/ for more information.
Classes
class TTS (*,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
voice_id: str = 'v_meklc281',
output_format: OutputFormat = 'MP3_22050_32',
num_channels: int = 1,
phrase_replacement_config_id: NotGivenOr[str] = NOT_GIVEN,
word_tokenizer: NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer] = NOT_GIVEN)-
Expand source code
class TTS(tts.TTS): """Uplift TTS implementation for LiveKit""" def __init__( self, *, base_url: NotGivenOr[str] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, voice_id: str = DEFAULT_VOICE_ID, output_format: OutputFormat = DEFAULT_OUTPUT_FORMAT, num_channels: int = DEFAULT_NUM_CHANNELS, phrase_replacement_config_id: NotGivenOr[str] = NOT_GIVEN, word_tokenizer: NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer] = NOT_GIVEN, ) -> None: """ Create a new instance of Uplift TTS. Args: base_url: Base URL for TTS service. Defaults to wss://api.upliftai.org api_key: API key for authentication voice_id: Voice ID to use. Defaults to "17" output_format: Audio output format. Options: - 'PCM_22050_16': PCM format, 22.05kHz, 16-bit - 'WAV_22050_16': WAV format, 22.05kHz, 16-bit - 'WAV_22050_32': WAV format, 22.05kHz, 32-bit - 'MP3_22050_32': MP3 format, 22.05kHz, 32kbps (default) - 'MP3_22050_64': MP3 format, 22.05kHz, 64kbps - 'MP3_22050_128': MP3 format, 22.05kHz, 128kbps - 'OGG_22050_16': OGG format, 22.05kHz, 16-bit - 'ULAW_8000_8': μ-law format, 8kHz, 8-bit sample_rate: Sample rate for audio output. Defaults to 22050 num_channels: Number of audio channels. Defaults to 1 (mono) phrase_replacement_config_id: Optional ID for phrase replacement configuration word_tokenizer: Tokenizer for processing text. Defaults to `livekit.agents.tokenize.basic.WordTokenizer`. """ super().__init__( capabilities=tts.TTSCapabilities( streaming=True, aligned_transcript=False, ), sample_rate=DEFAULT_SAMPLE_RATE, num_channels=num_channels, ) # Get configuration from environment if not provided resolved_base_url: str = ( base_url if is_given(base_url) else os.environ.get("UPLIFTAI_BASE_URL", DEFAULT_BASE_URL) ) resolved_api_key: str | None = ( api_key if is_given(api_key) else os.environ.get("UPLIFTAI_API_KEY") ) if not resolved_api_key: raise ValueError( "API key is required, either as argument or set UPLIFTAI_API_KEY environment variable" ) # Use provided tokenizer or create default resolved_word_tokenizer: tokenize.WordTokenizer | tokenize.SentenceTokenizer if is_given(word_tokenizer): resolved_word_tokenizer = word_tokenizer else: resolved_word_tokenizer = tokenize.basic.WordTokenizer(ignore_punctuation=False) self._opts = _TTSOptions( base_url=resolved_base_url, api_key=resolved_api_key, voice_settings=VoiceSettings(voice_id=voice_id, output_format=output_format), word_tokenizer=resolved_word_tokenizer, sample_rate=DEFAULT_SAMPLE_RATE, num_channels=num_channels, phrase_replacement_config_id=phrase_replacement_config_id if is_given(phrase_replacement_config_id) else None, ) self._client: WebSocketClient | None = None self._streams = weakref.WeakSet[SynthesizeStream]() def update_options( self, *, voice_id: NotGivenOr[str] = NOT_GIVEN, output_format: NotGivenOr[OutputFormat] = NOT_GIVEN, ) -> None: """ Update TTS configuration options. Args: voice_id: New voice ID output_format: New output format (see __init__ for options) """ if is_given(voice_id): self._opts.voice_settings.voice_id = voice_id if is_given(output_format): self._opts.voice_settings.output_format = output_format def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: """Synthesize text to speech using chunked stream.""" return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> SynthesizeStream: """Create a streaming synthesis session.""" stream = SynthesizeStream(tts=self, conn_options=conn_options) self._streams.add(stream) return stream async def aclose(self) -> None: """Clean up resources""" for stream in list(self._streams): await stream.aclose() self._streams.clear() if self._client: await self._client.disconnect() self._client = NoneUplift TTS implementation for LiveKit
Create a new instance of Uplift TTS.
Args
base_url- Base URL for TTS service. Defaults to wss://api.upliftai.org
api_key- API key for authentication
voice_id- Voice ID to use. Defaults to "17"
output_format- Audio output format. Options: - 'PCM_22050_16': PCM format, 22.05kHz, 16-bit - 'WAV_22050_16': WAV format, 22.05kHz, 16-bit - 'WAV_22050_32': WAV format, 22.05kHz, 32-bit - 'MP3_22050_32': MP3 format, 22.05kHz, 32kbps (default) - 'MP3_22050_64': MP3 format, 22.05kHz, 64kbps - 'MP3_22050_128': MP3 format, 22.05kHz, 128kbps - 'OGG_22050_16': OGG format, 22.05kHz, 16-bit - 'ULAW_8000_8': μ-law format, 8kHz, 8-bit
sample_rate- Sample rate for audio output. Defaults to 22050
num_channels- Number of audio channels. Defaults to 1 (mono)
phrase_replacement_config_id- Optional ID for phrase replacement configuration
word_tokenizer- Tokenizer for processing text. Defaults to
WordTokenizer.
Ancestors
- livekit.agents.tts.tts.TTS
- abc.ABC
- EventEmitter
- typing.Generic
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: """Clean up resources""" for stream in list(self._streams): await stream.aclose() self._streams.clear() if self._client: await self._client.disconnect() self._client = NoneClean up resources
def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0)) ‑> livekit.plugins.upliftai.tts.SynthesizeStream-
Expand source code
def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> SynthesizeStream: """Create a streaming synthesis session.""" stream = SynthesizeStream(tts=self, conn_options=conn_options) self._streams.add(stream) return streamCreate a streaming synthesis session.
def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0)) ‑> livekit.plugins.upliftai.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: """Synthesize text to speech using chunked stream.""" return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)Synthesize text to speech using chunked stream.
def update_options(self,
*,
voice_id: NotGivenOr[str] = NOT_GIVEN,
output_format: NotGivenOr[OutputFormat] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, voice_id: NotGivenOr[str] = NOT_GIVEN, output_format: NotGivenOr[OutputFormat] = NOT_GIVEN, ) -> None: """ Update TTS configuration options. Args: voice_id: New voice ID output_format: New output format (see __init__ for options) """ if is_given(voice_id): self._opts.voice_settings.voice_id = voice_id if is_given(output_format): self._opts.voice_settings.output_format = output_formatUpdate TTS configuration options.
Args
voice_id- New voice ID
output_format- New output format (see init for options)
Inherited members