Module livekit.agents.tts.tts

Classes

class ChunkedStream

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

Expand source code

class ChunkedStream(ABC):
    """Used by the non-streamed synthesize API, some providers support chunked http responses"""

    def __init__(self):
        self._event_ch = aio.Chan[SynthesizedAudio]()
        self._task = asyncio.create_task(self._main_task())
        self._task.add_done_callback(lambda _: self._event_ch.close())

    async def collect(self) -> rtc.AudioFrame:
        """Utility method to collect every frame in a single call"""
        frames = []
        async for ev in self:
            frames.append(ev.frame)
        return audio.merge_frames(frames)

    @abstractmethod
    async def _main_task(self) -> None: ...

    async def aclose(self) -> None:
        """Close is automatically called if the stream is completely collected"""
        await aio.gracefully_cancel(self._task)
        self._event_ch.close()

    async def __anext__(self) -> SynthesizedAudio:
        return await self._event_ch.__anext__()

    def __aiter__(self) -> AsyncIterator[SynthesizedAudio]:
        return self

Ancestors

• abc.ABC

Subclasses

• livekit.plugins.azure.tts.ChunkedStream
• livekit.plugins.cartesia.tts.ChunkedStream
• livekit.plugins.elevenlabs.tts.ChunkedStream
• livekit.plugins.google.tts.ChunkedStream
• livekit.plugins.openai.tts.ChunkedStream

Methods

async def aclose(self) ‑> None

Close is automatically called if the stream is completely collected

async def collect(self) ‑> AudioFrame

Utility method to collect every frame in a single call

class SynthesizeStream

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

Expand source code

class SynthesizeStream(ABC):
    class _FlushSentinel:
        pass

    def __init__(self):
        self._input_ch = aio.Chan[Union[str, SynthesizeStream._FlushSentinel]]()
        self._event_ch = aio.Chan[SynthesizedAudio]()
        self._task = asyncio.create_task(self._main_task(), name="TTS._main_task")
        self._task.add_done_callback(lambda _: self._event_ch.close())

    @abstractmethod
    async def _main_task(self) -> None: ...

    def push_text(self, token: str) -> None:
        """Push some text to be synthesized"""
        self._check_input_not_ended()
        self._check_not_closed()
        self._input_ch.send_nowait(token)

    def flush(self) -> None:
        """Mark the end of the current segment"""
        self._check_input_not_ended()
        self._check_not_closed()
        self._input_ch.send_nowait(self._FlushSentinel())

    def end_input(self) -> None:
        """Mark the end of input, no more text will be pushed"""
        self.flush()
        self._input_ch.close()

    async def aclose(self) -> None:
        """Close ths stream immediately"""
        self._input_ch.close()
        await aio.gracefully_cancel(self._task)
        self._event_ch.close()

    def _check_not_closed(self) -> None:
        if self._event_ch.closed:
            cls = type(self)
            raise RuntimeError(f"{cls.__module__}.{cls.__name__} is closed")

    def _check_input_not_ended(self) -> None:
        if self._input_ch.closed:
            cls = type(self)
            raise RuntimeError(f"{cls.__module__}.{cls.__name__} input ended")

    async def __anext__(self) -> SynthesizedAudio:
        return await self._event_ch.__anext__()

    def __aiter__(self) -> AsyncIterator[SynthesizedAudio]:
        return self

Ancestors

• abc.ABC

Subclasses

• StreamAdapterWrapper
• livekit.plugins.cartesia.tts.SynthesizeStream
• livekit.plugins.elevenlabs.tts.SynthesizeStream

Methods

async def aclose(self) ‑> None

Close ths stream immediately

def end_input(self) ‑> None

Mark the end of input, no more text will be pushed

def flush(self) ‑> None

Mark the end of the current segment

def push_text(self, token: str) ‑> None

Push some text to be synthesized

class SynthesizedAudio (request_id: str, segment_id: str, frame: rtc.AudioFrame, delta_text: str = '')

SynthesizedAudio(request_id: 'str', segment_id: 'str', frame: 'rtc.AudioFrame', delta_text: 'str' = '')

Expand source code

@dataclass
class SynthesizedAudio:
    request_id: str
    """Request ID (one segment could be made up of multiple requests)"""
    segment_id: str
    """Segment ID, each segment is separated by a flush"""
    frame: rtc.AudioFrame
    """Synthesized audio frame"""
    delta_text: str = ""
    """Current segment of the synthesized audio"""

Class variables

var delta_text : str

Current segment of the synthesized audio

var frame : AudioFrame

Synthesized audio frame

var request_id : str

Request ID (one segment could be made up of multiple requests)

var segment_id : str

Segment ID, each segment is separated by a flush

class TTS (*, capabilities: TTSCapabilities, sample_rate: int, num_channels: int)

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

Expand source code

class TTS(ABC):
    def __init__(
        self, *, capabilities: TTSCapabilities, sample_rate: int, num_channels: int
    ) -> None:
        self._capabilities = capabilities
        self._sample_rate = sample_rate
        self._num_channels = num_channels

    @property
    def capabilities(self) -> TTSCapabilities:
        return self._capabilities

    @property
    def sample_rate(self) -> int:
        return self._sample_rate

    @property
    def num_channels(self) -> int:
        return self._num_channels

    @abstractmethod
    def synthesize(self, text: str) -> ChunkedStream: ...

    def stream(self) -> SynthesizeStream:
        raise NotImplementedError(
            "streaming is not supported by this TTS, please use a different TTS or use a StreamAdapter"
        )

    async def aclose(self) -> None: ...

Ancestors

• abc.ABC

Subclasses

• StreamAdapter
• livekit.plugins.azure.tts.TTS
• livekit.plugins.cartesia.tts.TTS
• livekit.plugins.elevenlabs.tts.TTS
• livekit.plugins.google.tts.TTS
• livekit.plugins.openai.tts.TTS

Instance variables

prop capabilities : TTSCapabilities

Expand source code

@property
def capabilities(self) -> TTSCapabilities:
    return self._capabilities

prop num_channels : int

Expand source code

@property
def num_channels(self) -> int:
    return self._num_channels

prop sample_rate : int

Expand source code

@property
def sample_rate(self) -> int:
    return self._sample_rate

Methods

async def aclose(self) ‑> None

def stream(self) ‑> SynthesizeStream

def synthesize(self, text: str) ‑> ChunkedStream

class TTSCapabilities (streaming: bool)

TTSCapabilities(streaming: 'bool')

Expand source code

@dataclass
class TTSCapabilities:
    streaming: bool

Class variables

var streaming : bool