Module livekit.plugins.playai

PlayAI plugin for LiveKit Agents

See https://docs.livekit.io/agents/integrations/tts/playai/ for more information.

Classes

class TTS (*,
api_key: NotGivenOr[str] = NOT_GIVEN,
user_id: NotGivenOr[str] = NOT_GIVEN,
voice: str = 's3://voice-cloning-zero-shot/baf1ef41-36b6-428c-9bdf-50ba54682bd8/original/manifest.json',
language: str = 'english',
sample_rate: int = 24000,
model: TTSModel | str = 'PlayDialog',
word_tokenizer: tokenize.WordTokenizer | None = None,
**kwargs: dict[str, Any])
Expand source code 
class TTS(tts.TTS):
    def __init__(
        self,
        *,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        user_id: NotGivenOr[str] = NOT_GIVEN,
        voice: str = "s3://voice-cloning-zero-shot/baf1ef41-36b6-428c-9bdf-50ba54682bd8/original/manifest.json",
        language: str = "english",
        sample_rate: int = 24000,
        model: TTSModel | str = "PlayDialog",
        word_tokenizer: tokenize.WordTokenizer | None = None,
        **kwargs: dict[str, Any],
    ) -> None:
        """
        Initialize the PlayAI TTS engine.

        Args:
            api_key (str): PlayAI API key.
            user_id (str): PlayAI user ID.
            voice (str): Voice manifest URL.
            model (TTSModel): TTS model, defaults to "Play3.0-mini".
            language (str): language, defaults to "english".
            sample_rate (int): sample rate (Hz), A number greater than or equal to 8000, and must be less than or equal to 48000
            word_tokenizer (tokenize.WordTokenizer): Tokenizer for processing text. Defaults to basic WordTokenizer.
            **kwargs: Additional options.
        """  # noqa: E501

        super().__init__(
            capabilities=tts.TTSCapabilities(
                streaming=True,
            ),
            sample_rate=sample_rate,
            num_channels=1,
        )

        if not word_tokenizer:
            word_tokenizer = tokenize.basic.WordTokenizer(ignore_punctuation=False)

        pyht_api_key = api_key if is_given(api_key) else os.environ.get("PLAYHT_API_KEY")
        pyht_user_id = user_id if is_given(user_id) else os.environ.get("PLAYHT_USER_ID")

        if not pyht_api_key or not pyht_user_id:
            raise ValueError(
                "PlayHT API key and user ID are required. Set environment variables PLAYHT_API_KEY and PLAYHT_USER_ID or pass them explicitly."  # noqa: E501
            )

        _validate_kwargs(kwargs)

        self._opts = _TTSOptions(
            model=model,
            play_options=PlayOptions(
                voice=voice,
                format=Format.FORMAT_OGG,
                sample_rate=sample_rate,
                language=Language(language),
                **kwargs,
            ),
            word_tokenizer=word_tokenizer,
        )

        self._client = PlayAsyncClient(
            user_id=pyht_user_id,
            api_key=pyht_api_key,
        )
        self._streams = weakref.WeakSet[SynthesizeStream]()

    def update_options(
        self,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[TTSModel | str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        **kwargs: dict[str, Any],
    ) -> None:
        """
        Update the TTS options.
        """
        updates: dict[str, Any] = {}
        if is_given(voice):
            updates["voice"] = voice
        if is_given(language):
            updates["language"] = Language(language)
        updates.update(kwargs)

        _validate_kwargs(updates)

        for key, value in updates.items():
            if value is not None:
                setattr(self._opts.play_options, key, value)

        if is_given(model):
            self._opts.model = model

    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> ChunkedStream:
        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    def stream(
        self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> SynthesizeStream:
        stream = SynthesizeStream(tts=self, conn_options=conn_options)
        self._streams.add(stream)
        return stream

    async def aclose(self) -> None:
        await self._client.close()

Helper class that provides a standard way to create an ABC using inheritance.

Initialize the PlayAI TTS engine.

Args

api_key : str
PlayAI API key.
user_id : str
PlayAI user ID.
voice : str
Voice manifest URL.
model : TTSModel
TTS model, defaults to "Play3.0-mini".
language : str
language, defaults to "english".
sample_rate : int
sample rate (Hz), A number greater than or equal to 8000, and must be less than or equal to 48000
word_tokenizer : tokenize.WordTokenizer
Tokenizer for processing text. Defaults to basic WordTokenizer.
**kwargs
Additional options.

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:
    await self._client.close()
def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.playai.tts.SynthesizeStream
Expand source code 
def stream(
    self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
) -> SynthesizeStream:
    stream = SynthesizeStream(tts=self, conn_options=conn_options)
    self._streams.add(stream)
    return stream
def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.playai.tts.ChunkedStream
Expand source code 
def synthesize(
    self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
) -> ChunkedStream:
    return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)
def update_options(self,
*,
voice: NotGivenOr[str] = NOT_GIVEN,
model: NotGivenOr[TTSModel | str] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
**kwargs: dict[str, Any]) ‑> None
Expand source code 
def update_options(
    self,
    *,
    voice: NotGivenOr[str] = NOT_GIVEN,
    model: NotGivenOr[TTSModel | str] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    **kwargs: dict[str, Any],
) -> None:
    """
    Update the TTS options.
    """
    updates: dict[str, Any] = {}
    if is_given(voice):
        updates["voice"] = voice
    if is_given(language):
        updates["language"] = Language(language)
    updates.update(kwargs)

    _validate_kwargs(updates)

    for key, value in updates.items():
        if value is not None:
            setattr(self._opts.play_options, key, value)

    if is_given(model):
        self._opts.model = model

Update the TTS options.

Inherited members