Module livekit.plugins.google.beta

Classes

class GeminiSTT (*,
model: str = 'gemini-3.5-transcribe-live',
language: LanguageCode | str | None = 'en-US',
language_codes: list[str] | None = None,
custom_vocabulary: list[str] | None = None,
sample_rate: int = 16000,
api_key: NotGivenOr[str] = NOT_GIVEN,
vertexai: NotGivenOr[bool] = NOT_GIVEN,
credentials: Any | None = None,
credentials_path: str | None = None,
project: NotGivenOr[str] = NOT_GIVEN,
location: NotGivenOr[str] = NOT_GIVEN,
http_options: Any | None = None)

Expand source code

class STT(stt.STT):
    def __init__(
        self,
        *,
        model: str = DEFAULT_MODEL,
        language: LanguageCode | str | None = "en-US",
        language_codes: list[str] | None = None,
        custom_vocabulary: list[str] | None = None,
        sample_rate: int = DEFAULT_SAMPLE_RATE,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        vertexai: NotGivenOr[bool] = NOT_GIVEN,
        credentials: Any | None = None,
        credentials_path: str | None = None,
        project: NotGivenOr[str] = NOT_GIVEN,
        location: NotGivenOr[str] = NOT_GIVEN,
        http_options: Any | None = None,
    ) -> None:
        """Create a new instance of Gemini STT.

        Args:
            model: Live-capable Gemini model identifier. Defaults to
                "gemini-3.5-transcribe-live". Plain chat models are not accepted by the
                Live endpoint.
            language: Target language BCP-47 code or LanguageCode. Defaults to "en-US".
            language_codes: BCP-47 codes for the languages in the audio. Omit, or pass an
                empty list, to let the model detect the language.
            custom_vocabulary: Up to 1000 terms that bias recognition -- names, acronyms
                and jargon the model would otherwise mishear.
            sample_rate: Sample rate in Hz. Defaults to 16000.
            api_key: Optional Gemini API key. If not set, uses environment variables.
            vertexai: Whether to use Vertex AI backend.
            credentials: Service account credentials object or JSON string.
            credentials_path: Path to service account credentials JSON file.
            project: Google Cloud project ID (required for Vertex AI).
            location: Google Cloud region (e.g. "us-central1").
            http_options: Optional HTTP options for Client.
        """
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                offline_recognize=False,
            )
        )

        lang_code = LanguageCode(language) if language is not None else None

        self._opts = _STTOptions(
            model=model,
            language=lang_code,
            language_codes=language_codes,
            custom_vocabulary=custom_vocabulary,
            sample_rate=sample_rate,
            vertexai=vertexai if is_given(vertexai) else None,
            project=project if is_given(project) else None,
            location=location if is_given(location) else None,
        )

        self._api_key = api_key if is_given(api_key) else None
        self._credentials = credentials
        self._credentials_path = credentials_path
        self._http_options = http_options

    @property
    def model(self) -> str:
        return self._opts.model

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

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> RecognizeStream:
        opts = self._opts
        if is_given(language):
            opts = _STTOptions(
                model=self._opts.model,
                language=LanguageCode(language),
                language_codes=self._opts.language_codes,
                custom_vocabulary=self._opts.custom_vocabulary,
                sample_rate=self._opts.sample_rate,
                vertexai=self._opts.vertexai,
                project=self._opts.project,
                location=self._opts.location,
            )

        return RecognizeStream(
            stt=self,
            opts=opts,
            conn_options=conn_options,
            api_key=self._api_key,
            credentials=self._credentials,
            credentials_path=self._credentials_path,
            http_options=self._http_options,
        )

    async def _recognize_impl(
        self,
        buffer: utils.AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.SpeechEvent:
        raise NotImplementedError("Gemini STT only supports streaming recognition")

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

Create a new instance of Gemini STT.

Args

model

Live-capable Gemini model identifier. Defaults to "gemini-3.5-transcribe-live". Plain chat models are not accepted by the Live endpoint.

language

Target language BCP-47 code or LanguageCode. Defaults to "en-US".

language_codes

BCP-47 codes for the languages in the audio. Omit, or pass an empty list, to let the model detect the language.

custom_vocabulary

Up to 1000 terms that bias recognition – names, acronyms and jargon the model would otherwise mishear.

sample_rate

Sample rate in Hz. Defaults to 16000.

api_key

Optional Gemini API key. If not set, uses environment variables.

vertexai

Whether to use Vertex AI backend.

credentials

Service account credentials object or JSON string.

credentials_path

Path to service account credentials JSON file.

project

Google Cloud project ID (required for Vertex AI).

location

Google Cloud region (e.g. "us-central1").

http_options

Optional HTTP options for Client.

Ancestors

• livekit.agents.stt.stt.STT
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this STT 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 "google"

Get the provider name/identifier for this STT instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

def stream(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.google.beta.gemini_stt.RecognizeStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> RecognizeStream:
    opts = self._opts
    if is_given(language):
        opts = _STTOptions(
            model=self._opts.model,
            language=LanguageCode(language),
            language_codes=self._opts.language_codes,
            custom_vocabulary=self._opts.custom_vocabulary,
            sample_rate=self._opts.sample_rate,
            vertexai=self._opts.vertexai,
            project=self._opts.project,
            location=self._opts.location,
        )

    return RecognizeStream(
        stt=self,
        opts=opts,
        conn_options=conn_options,
        api_key=self._api_key,
        credentials=self._credentials,
        credentials_path=self._credentials_path,
        http_options=self._http_options,
    )

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class GeminiTTS (*,
model: GEMINI_TTS_MODELS | str = 'gemini-3.1-flash-tts-preview',
voice_name: GEMINI_VOICES | str = 'Kore',
api_key: NotGivenOr[str] = NOT_GIVEN,
vertexai: NotGivenOr[bool] = NOT_GIVEN,
project: NotGivenOr[str] = NOT_GIVEN,
location: NotGivenOr[str] = NOT_GIVEN,
instructions: NotGivenOr[str | None] = NOT_GIVEN,
speakers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
speaker: NotGivenOr[str] = NOT_GIVEN)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        *,
        model: GEMINI_TTS_MODELS | str = DEFAULT_MODEL,
        voice_name: GEMINI_VOICES | str = DEFAULT_VOICE,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        vertexai: NotGivenOr[bool] = NOT_GIVEN,
        project: NotGivenOr[str] = NOT_GIVEN,
        location: NotGivenOr[str] = NOT_GIVEN,
        instructions: NotGivenOr[str | None] = NOT_GIVEN,
        speakers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
        speaker: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        """
        Create a new instance of Gemini TTS.

        Environment Requirements:
        - For VertexAI: Set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to the path of the service account key file.
        - For Google Gemini API: Set the `api_key` argument or the `GOOGLE_API_KEY` environment variable.

        Args:
            model (str, optional): The Gemini TTS model to use. Defaults to "gemini-3.1-flash-tts-preview".
            voice_name (str, optional): The voice to use for synthesis. Defaults to "Kore".
            api_key (str, optional): The API key for Google Gemini. If not provided, it attempts to read from the `GOOGLE_API_KEY` environment variable.
            vertexai (bool, optional): Whether to use VertexAI. Defaults to False.
            project (str, optional): The Google Cloud project to use (only for VertexAI).
            location (str, optional): The location to use for VertexAI API requests. Defaults to "us-central1".
            instructions (str, optional): Control the style, tone, accent, and pace using prompts. See https://ai.google.dev/gemini-api/docs/speech-generation#controllable
            speakers (dict[str, str], optional): Speaker name -> voice, for a multi-speaker voice config. Replaces `voice_name`.
            speaker (str, optional): Which of `speakers` this instance voices. Required with `speakers`, and switchable via `update_options`.
        """  # noqa: E501
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=DEFAULT_SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
        )

        gcp_project: str | None = (
            project if is_given(project) else os.environ.get("GOOGLE_CLOUD_PROJECT")
        )
        gcp_location: str | None = (
            location
            if is_given(location)
            else os.environ.get("GOOGLE_CLOUD_LOCATION") or "us-central1"
        )
        use_vertexai = (
            vertexai
            if is_given(vertexai)
            else os.environ.get("GOOGLE_GENAI_USE_VERTEXAI", "0").lower() in ["true", "1"]
        )
        gemini_api_key = api_key if is_given(api_key) else os.environ.get("GOOGLE_API_KEY")

        if use_vertexai:
            if not gcp_project:
                from google.auth._default_async import default_async

                _, gcp_project = default_async(  # type: ignore
                    scopes=["https://www.googleapis.com/auth/cloud-platform"]
                )
            gemini_api_key = None  # VertexAI does not require an API key
        else:
            gcp_project = None
            gcp_location = None
            if not gemini_api_key:
                raise ValueError(
                    "API key is required for Google API either via api_key or GOOGLE_API_KEY environment variable"  # noqa: E501
                )

        speaker_map = dict(speakers) if is_given(speakers) else None
        current_speaker = speaker if is_given(speaker) else None
        if speaker_map is not None:
            # not "up to two": the API rejects any other count outright, with
            # "the number of speaker_voice_configs must equal 2"
            if len(speaker_map) != 2:
                raise ValueError(f"`speakers` must name exactly 2 speakers, got {len(speaker_map)}")
            # the API rejects a multi-speaker turn whose speech_metadata names no speaker,
            # so there is no useful default to fall back to here
            if current_speaker is None:
                raise ValueError("`speaker` is required when `speakers` is set")
            if current_speaker not in speaker_map:
                raise ValueError(
                    f"speaker {current_speaker!r} is not one of the configured speakers: "
                    f"{', '.join(sorted(speaker_map))}"
                )
            if not _styles_per_part(model):
                raise ValueError(
                    f"multi-speaker needs a model that takes per-part speech_metadata; "
                    f"{model!r} does not"
                )

        self._opts = _TTSOptions(
            speakers=speaker_map,
            speaker=current_speaker,
            model=model,
            voice_name=voice_name,
            vertexai=use_vertexai,
            project=gcp_project,
            location=gcp_location,
            # this family speaks anything left in a part's text, so the preamble would
            # be read aloud (measured: it roughly triples a short line). _styled_part
            # puts instructions in speech_metadata.style instead.
            instructions=instructions
            if is_given(instructions)
            else (None if _styles_per_part(model) else DEFAULT_INSTRUCTIONS),
        )

        self._client = Client(
            api_key=gemini_api_key,
            vertexai=use_vertexai,
            project=gcp_project,
            location=gcp_location,
        )

    class Markup(tts.TTS.Markup):
        # only a model that can carry a style out of band declares the dialect; on the
        # others a marker would have nowhere to go and would be read aloud
        def _provider_key(self) -> str:
            assert isinstance(self._tts, TTS)
            return "gemini" if _styles_per_part(self._tts._opts.model) else ""

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        if self._client.vertexai:
            return "Vertex AI"
        else:
            return "Gemini"

    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_name: NotGivenOr[str] = NOT_GIVEN,
        speaker: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        """
        Update the TTS options.

        Args:
            voice_name (str, optional): The voice to use for synthesis.
            speaker (str, optional): Which configured speaker to voice from now on.
        """
        if is_given(voice_name):
            self._opts.voice_name = voice_name
        if is_given(speaker):
            if not self._opts.speakers:
                raise ValueError("`speaker` needs a TTS constructed with `speakers`")
            if speaker not in self._opts.speakers:
                raise ValueError(
                    f"speaker {speaker!r} is not one of the configured speakers: "
                    f"{', '.join(sorted(self._opts.speakers))}"
                )
            self._opts.speaker = speaker

    async def aclose(self) -> None:
        """Close the TTS and release its GenAI HTTP clients."""
        try:
            await self._client.aio.aclose()
        except Exception:
            logger.warning("failed to close the genai client", exc_info=True)

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

Create a new instance of Gemini TTS.

Environment Requirements: - For VertexAI: Set the GOOGLE_APPLICATION_CREDENTIALS environment variable to the path of the service account key file. - For Google Gemini API: Set the api_key argument or the GOOGLE_API_KEY environment variable.

Args

model : str, optional

The Gemini TTS model to use. Defaults to "gemini-3.1-flash-tts-preview".

voice_name : str, optional

The voice to use for synthesis. Defaults to "Kore".

api_key : str, optional

The API key for Google Gemini. If not provided, it attempts to read from the GOOGLE_API_KEY environment variable.

vertexai : bool, optional

Whether to use VertexAI. Defaults to False.

project : str, optional

The Google Cloud project to use (only for VertexAI).

location : str, optional

The location to use for VertexAI API requests. Defaults to "us-central1".

instructions : str, optional

Control the style, tone, accent, and pace using prompts. See https://ai.google.dev/gemini-api/docs/speech-generation#controllable

speakers : dict[str, str], optional

Speaker name -> voice, for a multi-speaker voice config. Replaces voice_name.

speaker : str, optional

Which of speakers this instance voices. Required with speakers, and switchable via update_options.

Ancestors

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

Class variables

var Markup

Declares TTS markup capabilities for the expressive pipeline.

Plugins override this inner class to declare which markup dialect the TTS speaks: what the LLM is taught to write, and how those markers are normalized and lowered to the provider's native syntax before synthesis. Stripping markup back out is not here — the transcript sinks do it provider-agnostically (see _provider_format.split_all_markup).

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

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:
    if self._client.vertexai:
        return "Vertex AI"
    else:
        return "Gemini"

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:
    """Close the TTS and release its GenAI HTTP clients."""
    try:
        await self._client.aio.aclose()
    except Exception:
        logger.warning("failed to close the genai client", exc_info=True)

Close the TTS and release its GenAI HTTP clients.

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.google.beta.gemini_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_name: NotGivenOr[str] = NOT_GIVEN,
speaker: NotGivenOr[str] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    voice_name: NotGivenOr[str] = NOT_GIVEN,
    speaker: NotGivenOr[str] = NOT_GIVEN,
) -> None:
    """
    Update the TTS options.

    Args:
        voice_name (str, optional): The voice to use for synthesis.
        speaker (str, optional): Which configured speaker to voice from now on.
    """
    if is_given(voice_name):
        self._opts.voice_name = voice_name
    if is_given(speaker):
        if not self._opts.speakers:
            raise ValueError("`speaker` needs a TTS constructed with `speakers`")
        if speaker not in self._opts.speakers:
            raise ValueError(
                f"speaker {speaker!r} is not one of the configured speakers: "
                f"{', '.join(sorted(self._opts.speakers))}"
            )
        self._opts.speaker = speaker

Update the TTS options.

Args

voice_name : str, optional

The voice to use for synthesis.

speaker : str, optional

Which configured speaker to voice from now on.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once