Module livekit.plugins.speechify

Speechify plugin for LiveKit Agents

Provides Speechify text-to-speech for LiveKit voice agents. See https://docs.speechify.ai for API details.

Classes

class ChunkedStream (*,
tts: TTS,
input_text: str,
conn_options: APIConnectOptions)

Expand source code

class ChunkedStream(tts.ChunkedStream):
    def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
        super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        try:
            response = await self._tts._client.audio.speech(
                **_request_kwargs(self._input_text, self._opts),
                request_options={"timeout_in_seconds": int(self._conn_options.timeout)},
            )
            output_emitter.initialize(
                request_id=utils.shortuuid(),
                sample_rate=SAMPLE_RATE,
                num_channels=NUM_CHANNELS,
                mime_type=MIME_TYPE,
            )
            timed = _timed_transcript(response.speech_marks, 0.0)
            if timed:
                output_emitter.push_timed_transcript(timed)
            output_emitter.push(base64.b64decode(response.audio_data))
            output_emitter.flush()
        except Exception as e:
            _raise_from(e)

Used by the non-streamed synthesize API, some providers support chunked http responses

Ancestors

• livekit.agents.tts.tts.ChunkedStream
• abc.ABC

class SynthesizeStream (*,
tts: TTS,
conn_options: APIConnectOptions)

Expand source code

class SynthesizeStream(tts.SynthesizeStream):
    def __init__(self, *, tts: TTS, conn_options: APIConnectOptions) -> None:
        super().__init__(tts=tts, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        request_id = utils.shortuuid()
        output_emitter.initialize(
            request_id=request_id,
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            mime_type=MIME_TYPE,
            stream=True,
        )
        output_emitter.start_segment(segment_id=request_id)

        sent_stream = self._tts._tokenizer.stream()

        async def _forward_input() -> None:
            async for data in self._input_ch:
                if isinstance(data, self._FlushSentinel):
                    sent_stream.flush()
                    continue
                sent_stream.push_text(data)
            sent_stream.end_input()

        async def _synthesize() -> None:
            offset = 0.0
            async for ev in sent_stream:
                if not (text := ev.token.strip()):
                    continue
                self._mark_started()
                sentence_bytes = 0

                if _supports_streaming_marks(self._opts.model):
                    stream = self._tts._client.audio.stream_with_timestamps(
                        **_stream_request_kwargs(text, self._opts),
                        request_options={"timeout_in_seconds": int(self._conn_options.timeout)},
                    )
                    async for event in stream:
                        if isinstance(event, SpeechStreamEvent_SpeechChunk):
                            if event.audio:
                                audio = base64.b64decode(event.audio)
                                output_emitter.push(audio)
                                sentence_bytes += len(audio)
                            if event.speech_marks:
                                timed = _marks_to_timed(event.speech_marks, offset)
                                if timed:
                                    output_emitter.push_timed_transcript(timed)
                        elif isinstance(event, SpeechStreamEvent_SpeechError):
                            raise APIStatusError(
                                message=event.error.message,
                                status_code=-1,
                                request_id=event.request_id,
                                body=None,
                            )
                        elif isinstance(event, SpeechStreamEvent_SpeechDone):
                            break
                else:
                    response = await self._tts._client.audio.speech(
                        **_request_kwargs(text, self._opts),
                        request_options={"timeout_in_seconds": int(self._conn_options.timeout)},
                    )
                    audio = base64.b64decode(response.audio_data)
                    timed = _timed_transcript(response.speech_marks, offset)
                    if timed:
                        output_emitter.push_timed_transcript(timed)
                    output_emitter.push(audio)
                    sentence_bytes = len(audio)

                output_emitter.flush()
                offset += sentence_bytes / (2 * SAMPLE_RATE * NUM_CHANNELS)

            output_emitter.end_segment()

        tasks = [
            asyncio.create_task(_forward_input()),
            asyncio.create_task(_synthesize()),
        ]
        try:
            await asyncio.gather(*tasks)
        except Exception as e:
            _raise_from(e)
        finally:
            await sent_stream.aclose()
            await utils.aio.cancel_and_wait(*tasks)

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

Ancestors

• livekit.agents.tts.tts.SynthesizeStream
• abc.ABC

class TTS (*,
voice_id: str = 'dominic_32',
model: TTSModels = 'simba-3.2',
language: NotGivenOr[str] = NOT_GIVEN,
loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
text_normalization: NotGivenOr[bool] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
client: AsyncSpeechify | None = None,
**kwargs: Any)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        *,
        voice_id: str = DEFAULT_VOICE_ID,
        model: TTSModels = DEFAULT_MODEL,
        language: NotGivenOr[str] = NOT_GIVEN,
        loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
        text_normalization: NotGivenOr[bool] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
        client: AsyncSpeechify | None = None,
        **kwargs: Any,
    ) -> None:
        """Create a new instance of Speechify TTS.

        ``synthesize()`` uses the Speechify ``/v1/audio/speech`` endpoint, which
        returns raw PCM (24 kHz mono) together with word-level speech marks in a
        single non-streamed response.

        ``stream()`` uses the ``/v1/audio/stream/with-timestamps`` endpoint,
        which streams audio chunks together with word-level speech marks so
        audio and aligned timestamps are emitted as they become final. Input is
        split into sentences and one streaming request is issued per sentence.
        Speech marks are produced by the streaming-native models (``simba-3.0``
        and ``simba-3.2``); the legacy ``simba-english`` and
        ``simba-multilingual`` models do not serve the streaming route and fall
        back to one ``/v1/audio/speech`` request per sentence.

        Args:
            voice_id: Id of the voice to synthesize with. The voice must support
                the chosen ``model`` (see the ``/v1/voices`` endpoint). Defaults
                to ``dominic_32``. Voices whose id carries a ``_32`` suffix are
                part of the curated simba-3.2 roster; a warning is logged when
                they are paired with any other model.
            model: Synthesis model. One of ``simba-english``,
                ``simba-multilingual``, ``simba-3.0`` or ``simba-3.2``. Defaults
                to ``simba-3.2``.
            language: BCP-47 language code of the input (e.g. ``en-US``).
            loudness_normalization: Normalize output loudness to a standard
                level. Increases latency slightly when enabled.
            text_normalization: Expand numbers, dates, etc. into words before
                synthesis. Increases latency slightly when enabled.
            api_key: Speechify API key. Falls back to the ``SPEECHIFY_API_KEY``
                environment variable.
            base_url: Override the Speechify API base URL.
            tokenizer: Sentence tokenizer used to chunk input in ``stream()``.
            client: A preconfigured ``AsyncSpeechify`` client. When provided,
                ``api_key`` and ``base_url`` are ignored.
            **kwargs: Catches deprecated parameters. A warning is logged for
                any recognised deprecated name.
        """
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=True, aligned_transcript=True),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
        )

        self._owns_client = client is None
        if client is not None:
            self._client = client
        else:
            resolved_key = api_key if is_given(api_key) else os.environ.get("SPEECHIFY_API_KEY")
            if not resolved_key:
                raise ValueError(
                    "Speechify API key is required, either as the api_key argument "
                    "or via the SPEECHIFY_API_KEY environment variable"
                )
            # Fixed httpx.AsyncClient default header so every request the SDK
            # issues is attributed to this integration, regardless of call site.
            # Timeout/limits mirror the openai plugin's owned-client defaults —
            # httpx's own 5s default is too short for longer synthesis requests.
            self._httpx_client = httpx.AsyncClient(
                headers={CALLER_HEADER: "livekit"},
                timeout=httpx.Timeout(connect=15.0, read=30.0, write=30.0, pool=5.0),
                limits=httpx.Limits(
                    max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
                ),
            )
            self._client = AsyncSpeechify(
                token=resolved_key,
                base_url=base_url if is_given(base_url) else None,
                httpx_client=self._httpx_client,
            )

        self._tokenizer = tokenizer if is_given(tokenizer) else tokenize.basic.SentenceTokenizer()
        self._opts = _TTSOptions(
            voice_id=voice_id,
            model=model,
            language=language,
            loudness_normalization=loudness_normalization,
            text_normalization=text_normalization,
        )

        _check_deprecated_args(kwargs)
        _warn_voice_model_compat(self._opts.voice_id, self._opts.model)

    @property
    def model(self) -> str:
        return self._opts.model if is_given(self._opts.model) else "unknown"

    @property
    def provider(self) -> str:
        return "Speechify"

    async def aclose(self) -> None:
        if self._owns_client:
            await self._httpx_client.aclose()

    async def list_voices(self) -> list[Voice]:
        """List the voices available for the configured Speechify account."""
        sdk_voices: list[GetVoice] = await self._client.voices.list()
        return [_voice_from_sdk(v) for v in sdk_voices]

    def update_options(
        self,
        *,
        voice_id: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[TTSModels] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
        text_normalization: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        if is_given(voice_id):
            self._opts.voice_id = voice_id
        if is_given(model):
            self._opts.model = model
        if is_given(language):
            self._opts.language = language
        if is_given(loudness_normalization):
            self._opts.loudness_normalization = loudness_normalization
        if is_given(text_normalization):
            self._opts.text_normalization = text_normalization
        _warn_voice_model_compat(self._opts.voice_id, self._opts.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:
        return SynthesizeStream(tts=self, conn_options=conn_options)

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

Create a new instance of Speechify TTS.

synthesize() uses the Speechify /v1/audio/speech endpoint, which returns raw PCM (24 kHz mono) together with word-level speech marks in a single non-streamed response.

stream() uses the /v1/audio/stream/with-timestamps endpoint, which streams audio chunks together with word-level speech marks so audio and aligned timestamps are emitted as they become final. Input is split into sentences and one streaming request is issued per sentence. Speech marks are produced by the streaming-native models (simba-3.0 and simba-3.2); the legacy simba-english and simba-multilingual models do not serve the streaming route and fall back to one /v1/audio/speech request per sentence.

Args

voice_id

Id of the voice to synthesize with. The voice must support the chosen model (see the /v1/voices endpoint). Defaults to dominic_32. Voices whose id carries a _32 suffix are part of the curated simba-3.2 roster; a warning is logged when they are paired with any other model.

model

Synthesis model. One of simba-english, simba-multilingual, simba-3.0 or simba-3.2. Defaults to simba-3.2.

language

BCP-47 language code of the input (e.g. en-US).

loudness_normalization

Normalize output loudness to a standard level. Increases latency slightly when enabled.

text_normalization

Expand numbers, dates, etc. into words before synthesis. Increases latency slightly when enabled.

api_key

Speechify API key. Falls back to the SPEECHIFY_API_KEY environment variable.

base_url

Override the Speechify API base URL.

tokenizer

Sentence tokenizer used to chunk input in stream().

client

A preconfigured AsyncSpeechify client. When provided, api_key and base_url are ignored.

**kwargs

Catches deprecated parameters. A warning is logged for any recognised deprecated name.

Ancestors

• livekit.agents.tts.tts.TTS
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model if is_given(self._opts.model) else "unknown"

Get the model name/identifier for this TTS instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "Speechify"

Get the provider name/identifier for this TTS instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    if self._owns_client:
        await self._httpx_client.aclose()

async def list_voices(self) ‑> list[livekit.plugins.speechify.tts.Voice]

Expand source code

async def list_voices(self) -> list[Voice]:
    """List the voices available for the configured Speechify account."""
    sdk_voices: list[GetVoice] = await self._client.voices.list()
    return [_voice_from_sdk(v) for v in sdk_voices]

List the voices available for the configured Speechify account.

def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.speechify.tts.SynthesizeStream

Expand source code

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

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.speechify.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_id: NotGivenOr[str] = NOT_GIVEN,
model: NotGivenOr[TTSModels] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
text_normalization: NotGivenOr[bool] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    voice_id: NotGivenOr[str] = NOT_GIVEN,
    model: NotGivenOr[TTSModels] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
    text_normalization: NotGivenOr[bool] = NOT_GIVEN,
) -> None:
    if is_given(voice_id):
        self._opts.voice_id = voice_id
    if is_given(model):
        self._opts.model = model
    if is_given(language):
        self._opts.language = language
    if is_given(loudness_normalization):
        self._opts.loudness_normalization = loudness_normalization
    if is_given(text_normalization):
        self._opts.text_normalization = text_normalization
    _warn_voice_model_compat(self._opts.voice_id, self._opts.model)

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class Voice (id: str,
type: VoiceType,
display_name: str,
gender: Gender,
avatar_image: str | None,
models: list[TTSModels],
locale: str)

Expand source code

@dataclass
class Voice:
    id: str
    type: VoiceType
    display_name: str
    gender: Gender
    avatar_image: str | None
    models: list[TTSModels]
    locale: str

Voice(id: 'str', type: 'VoiceType', display_name: 'str', gender: 'Gender', avatar_image: 'str | None', models: 'list[TTSModels]', locale: 'str')

Instance variables

var avatar_image : str | None

var display_name : str

var gender : Literal['male', 'female', 'neutral']

var id : str

var locale : str

var models : list[typing.Literal['simba-english', 'simba-multilingual', 'simba-3.0', 'simba-3.2']]

var type : Literal['shared', 'personal']