Module `livekit.plugins.mistralai`

LiveKit plugin for Mistral AI models. Supports Chat, STT, and TTS models

Sub-modules

livekit.plugins.mistralai.tools

Classes

class ApiMode (*args, **kwds)

Expand source code

class ApiMode(str, Enum):
    """Deprecated: the LLM always uses the Conversations API."""

    CONVERSATIONS = "conversations"
    CHAT_COMPLETIONS = "chat_completions"

Deprecated: the LLM always uses the Conversations API.

Ancestors

builtins.str
enum.Enum

Class variables

var CHAT_COMPLETIONS
var CONVERSATIONS

class LLM (*, client: Mistral | None = None, api_key: NotGivenOr[str] = NOT_GIVEN, model: NotGivenOr[ChatModels | str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, presence_penalty: NotGivenOr[float] = NOT_GIVEN, frequency_penalty: NotGivenOr[float] = NOT_GIVEN, random_seed: NotGivenOr[int] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, max_completion_tokens: NotGivenOr[int] = NOT_GIVEN, store: NotGivenOr[bool] = NOT_GIVEN, api_mode: NotGivenOr[ApiMode | str] = NOT_GIVEN)

Expand source code

class LLM(llm.LLM):
    @deprecate_params({"api_mode": _API_MODE_DEPRECATION})
    def __init__(
        self,
        *,
        client: Mistral | None = None,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[ChatModels | str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        presence_penalty: NotGivenOr[float] = NOT_GIVEN,
        frequency_penalty: NotGivenOr[float] = NOT_GIVEN,
        random_seed: NotGivenOr[int] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
        store: NotGivenOr[bool] = NOT_GIVEN,
        api_mode: NotGivenOr[ApiMode | str] = NOT_GIVEN,
    ) -> None:
        """Create a new instance of MistralAI LLM.

        Args:
            client: Optional pre-configured MistralAI client instance.
            api_key: Your Mistral AI API key.  If not provided, will use
                the ``MISTRAL_API_KEY`` environment variable.
            model: The Mistral AI model to use, default is
                ``"mistral-small-latest"``.
            temperature: The temperature to use the LLM with.
            top_p: Nucleus sampling parameter.
            presence_penalty: Penalize new tokens based on their presence
                in the text so far.
            frequency_penalty: Penalize new tokens based on their
                frequency in the text so far.
            random_seed: Random seed for reproducibility.
            tool_choice: Default tool choice strategy
                (``"auto"``, ``"required"``, ``"none"``).
            max_completion_tokens: The max. number of tokens the LLM can
                output.
            store: Whether Mistral stores each conversation on its servers. The
                Conversations API stores them by default; pass ``False`` to opt out.
                The plugin sends the full chat context with every request, so it
                never relies on a stored conversation.
            api_mode: Deprecated and ignored, the LLM always uses the Conversations API.
        """
        super().__init__()
        self._opts = _LLMOptions(
            model=model if is_given(model) else DEFAULT_MODEL,
            temperature=temperature if is_given(temperature) else None,
            top_p=top_p if is_given(top_p) else None,
            presence_penalty=presence_penalty if is_given(presence_penalty) else None,
            frequency_penalty=frequency_penalty if is_given(frequency_penalty) else None,
            random_seed=random_seed if is_given(random_seed) else None,
            tool_choice=tool_choice if is_given(tool_choice) else None,
            max_completion_tokens=max_completion_tokens
            if is_given(max_completion_tokens)
            else None,
            store=store if is_given(store) else None,
        )

        mistral_api_key = api_key if is_given(api_key) else os.environ.get("MISTRAL_API_KEY")
        if not client and not mistral_api_key:
            raise ValueError("Mistral AI API key is required. Set MISTRAL_API_KEY or pass api_key")
        self._client = client or Mistral(api_key=mistral_api_key)

    async def _prewarm_impl(self) -> None:
        await asyncio.to_thread(lambda: self._client.beta.conversations)
        await self._client.models.list_async()

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

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

    @deprecate_params({"api_mode": _API_MODE_DEPRECATION})
    def update_options(
        self,
        *,
        model: NotGivenOr[ChatModels | str] = NOT_GIVEN,
        max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        presence_penalty: NotGivenOr[float] = NOT_GIVEN,
        frequency_penalty: NotGivenOr[float] = NOT_GIVEN,
        random_seed: NotGivenOr[int] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        store: NotGivenOr[bool] = NOT_GIVEN,
        api_mode: NotGivenOr[ApiMode | str] = NOT_GIVEN,
    ) -> None:
        if is_given(model):
            self._opts.model = model
        if is_given(max_completion_tokens):
            self._opts.max_completion_tokens = max_completion_tokens
        if is_given(temperature):
            self._opts.temperature = temperature
        if is_given(top_p):
            self._opts.top_p = top_p
        if is_given(presence_penalty):
            self._opts.presence_penalty = presence_penalty
        if is_given(frequency_penalty):
            self._opts.frequency_penalty = frequency_penalty
        if is_given(random_seed):
            self._opts.random_seed = random_seed
        if is_given(tool_choice):
            self._opts.tool_choice = tool_choice
        if is_given(store):
            self._opts.store = store

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[llm.Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        extra: dict[str, Any] = {}
        if self._opts.store is not None:
            extra["store"] = self._opts.store
        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        completion_args: dict[str, Any] = {}
        if self._opts.max_completion_tokens is not None:
            completion_args["max_tokens"] = self._opts.max_completion_tokens
        if self._opts.temperature is not None:
            completion_args["temperature"] = self._opts.temperature
        if self._opts.top_p is not None:
            completion_args["top_p"] = self._opts.top_p
        if self._opts.presence_penalty is not None:
            completion_args["presence_penalty"] = self._opts.presence_penalty
        if self._opts.frequency_penalty is not None:
            completion_args["frequency_penalty"] = self._opts.frequency_penalty
        if self._opts.random_seed is not None:
            completion_args["random_seed"] = self._opts.random_seed

        resolved_tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
        if resolved_tool_choice is not None:
            has_provider_tools = any(isinstance(t, MistralTool) for t in (tools or []))
            if isinstance(resolved_tool_choice, dict) or resolved_tool_choice == "required":
                completion_args["tool_choice"] = "auto" if has_provider_tools else "required"
            elif resolved_tool_choice in ("auto", "none"):
                completion_args["tool_choice"] = resolved_tool_choice
        if completion_args:
            extra["completion_args"] = CompletionArgs(**completion_args)

        return LLMStream(
            self,
            model=self._opts.model,
            client=self._client,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            extra_kwargs=extra,
        )

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

Create a new instance of MistralAI LLM.

Args

client: Optional pre-configured MistralAI client instance.
api_key: Your Mistral AI API key. If not provided, will use the MISTRAL_API_KEY environment variable.
model: The Mistral AI model to use, default is "mistral-small-latest".
temperature: The temperature to use the LLM with.
top_p: Nucleus sampling parameter.
presence_penalty: Penalize new tokens based on their presence in the text so far.
frequency_penalty: Penalize new tokens based on their frequency in the text so far.
random_seed: Random seed for reproducibility.
tool_choice: Default tool choice strategy ("auto", "required", "none").
max_completion_tokens: The max. number of tokens the LLM can output.
store: Whether Mistral stores each conversation on its servers. The Conversations API stores them by default; pass False to opt out. The plugin sends the full chat context with every request, so it never relies on a stored conversation.
api_mode: Deprecated and ignored, the LLM always uses the Conversations API.

Ancestors

livekit.agents.llm.llm.LLM
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 LLM 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 "MistralAI"

Get the provider name/identifier for this LLM instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

def chat(self, *, chat_ctx: ChatContext, tools: list[llm.Tool] | None = None, conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0), parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> livekit.plugins.mistralai.llm.LLMStream

Expand source code

def chat(
    self,
    *,
    chat_ctx: ChatContext,
    tools: list[llm.Tool] | None = None,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> LLMStream:
    extra: dict[str, Any] = {}
    if self._opts.store is not None:
        extra["store"] = self._opts.store
    if is_given(extra_kwargs):
        extra.update(extra_kwargs)

    completion_args: dict[str, Any] = {}
    if self._opts.max_completion_tokens is not None:
        completion_args["max_tokens"] = self._opts.max_completion_tokens
    if self._opts.temperature is not None:
        completion_args["temperature"] = self._opts.temperature
    if self._opts.top_p is not None:
        completion_args["top_p"] = self._opts.top_p
    if self._opts.presence_penalty is not None:
        completion_args["presence_penalty"] = self._opts.presence_penalty
    if self._opts.frequency_penalty is not None:
        completion_args["frequency_penalty"] = self._opts.frequency_penalty
    if self._opts.random_seed is not None:
        completion_args["random_seed"] = self._opts.random_seed

    resolved_tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
    if resolved_tool_choice is not None:
        has_provider_tools = any(isinstance(t, MistralTool) for t in (tools or []))
        if isinstance(resolved_tool_choice, dict) or resolved_tool_choice == "required":
            completion_args["tool_choice"] = "auto" if has_provider_tools else "required"
        elif resolved_tool_choice in ("auto", "none"):
            completion_args["tool_choice"] = resolved_tool_choice
    if completion_args:
        extra["completion_args"] = CompletionArgs(**completion_args)

    return LLMStream(
        self,
        model=self._opts.model,
        client=self._client,
        chat_ctx=chat_ctx,
        tools=tools or [],
        conn_options=conn_options,
        extra_kwargs=extra,
    )

def update_options(self, *, model: NotGivenOr[ChatModels | str] = NOT_GIVEN, max_completion_tokens: NotGivenOr[int] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, presence_penalty: NotGivenOr[float] = NOT_GIVEN, frequency_penalty: NotGivenOr[float] = NOT_GIVEN, random_seed: NotGivenOr[int] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, store: NotGivenOr[bool] = NOT_GIVEN, api_mode: NotGivenOr[ApiMode | str] = NOT_GIVEN) ‑> None

Expand source code

@deprecate_params({"api_mode": _API_MODE_DEPRECATION})
def update_options(
    self,
    *,
    model: NotGivenOr[ChatModels | str] = NOT_GIVEN,
    max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
    presence_penalty: NotGivenOr[float] = NOT_GIVEN,
    frequency_penalty: NotGivenOr[float] = NOT_GIVEN,
    random_seed: NotGivenOr[int] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    store: NotGivenOr[bool] = NOT_GIVEN,
    api_mode: NotGivenOr[ApiMode | str] = NOT_GIVEN,
) -> None:
    if is_given(model):
        self._opts.model = model
    if is_given(max_completion_tokens):
        self._opts.max_completion_tokens = max_completion_tokens
    if is_given(temperature):
        self._opts.temperature = temperature
    if is_given(top_p):
        self._opts.top_p = top_p
    if is_given(presence_penalty):
        self._opts.presence_penalty = presence_penalty
    if is_given(frequency_penalty):
        self._opts.frequency_penalty = frequency_penalty
    if is_given(random_seed):
        self._opts.random_seed = random_seed
    if is_given(tool_choice):
        self._opts.tool_choice = tool_choice
    if is_given(store):
        self._opts.store = store

Inherited members

EventEmitter:
- emit
- off
- on
- once

class STT (client: Mistral | None = None, api_key: NotGivenOr[str] = NOT_GIVEN, model: NotGivenOr[str] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, context_bias: NotGivenOr[list[str]] = NOT_GIVEN, target_streaming_delay_ms: NotGivenOr[int] = NOT_GIVEN, diarize: NotGivenOr[bool] = NOT_GIVEN, turn_detection: NotGivenOr[TurnDetection | None] = NOT_GIVEN, vad: vad.VAD | None = None)

Expand source code

class STT(stt.STT):
    def __init__(
        self,
        client: Mistral | None = None,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        context_bias: NotGivenOr[list[str]] = NOT_GIVEN,
        target_streaming_delay_ms: NotGivenOr[int] = NOT_GIVEN,
        diarize: NotGivenOr[bool] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetection | None] = NOT_GIVEN,
        vad: vad.VAD | None = None,
    ):
        """
        Create a new instance of MistralAI STT.

        Args:
            client: Optional pre-configured MistralAI client instance.
            api_key: Your Mistral AI API key. If not provided, will use the MISTRAL_API_KEY environment variable.
            model: The Mistral AI model to use for transcription, default is the realtime "voxtral-mini-transcribe-realtime-3".
            language: The optional language code to use for better transcription accuracy if language is already known (e.g., "fr" for French).
                Used with batch models and the Voxtral 3 realtime model.
            context_bias: Up to 100 words or phrases to guide the model toward good spelling or names or domain-specific vocabulary.
                Used with batch models and the Voxtral 3 realtime model.
            target_streaming_delay_ms: Target streaming delay in milliseconds for realtime mode. Only used with realtime models.
            diarize: Label each piece of text with its speaker (``speaker_id``). Supported by the batch models and
                the Voxtral 3 realtime model, but not by earlier realtime models.
                Voxtral 3 doesn't support diarization together with its server-side turn detection, so enabling
                ``diarize`` disables ``turn_detection`` by default, and passing both raises a ``ValueError``.
                The plugin then emits no start and end of speech events, and the voice pipeline can't use
                ``turn_detection="stt"``, unless a ``vad`` is given to provide them.
                The realtime model numbers the speakers within one transcript only: the numbering restarts after
                each flush, so the speakers of different transcripts can't be compared.
            turn_detection: Server-side turn detection of the Voxtral 3 realtime model, which provides the start and
                end of speech events and finalizes the transcript of each turn. Enabled with the server defaults
                when not provided, unless ``diarize`` is enabled or a ``vad`` is given, which it can't be used with.
                Pass a :class:`TurnDetection` dict to tune it, or ``None`` to disable it. Only supported by the
                Voxtral 3 realtime model.
            vad: Optional Voice Activity Detector run by the plugin on the realtime models: its end of speech
                flushes the audio to get the final transcript, and its start and end of speech are emitted as speech
                events, which the voice pipeline needs for ``turn_detection="stt"`` when the server doesn't provide
                them (earlier realtime models, or Voxtral 3 with ``diarize``). It takes precedence over the
                server-side turn detection of Voxtral 3, which is disabled by default, and can't be used with an
                explicit ``turn_detection``. Only used with realtime models.

        Note:
            Realtime models without server-side turn detection or a ``vad`` (earlier models, or Voxtral 3 with
            ``turn_detection=None`` or ``diarize``) have no endpointing. The voice pipeline (`AgentSession`)
            requests a flush from its own VAD and turn detection to get the final transcripts. A stream used on its
            own needs to be flushed with `stream.flush()`.
        """
        resolved_model = model if is_given(model) else DEFAULT_MODEL
        is_realtime = _is_realtime(resolved_model)

        diarize_enabled = is_given(diarize) and diarize
        if diarize_enabled and is_realtime and not _is_voxtral_3(resolved_model):
            raise ValueError("diarization is not supported by this realtime model")

        server_turn_detection: TurnDetection | None = None
        if is_realtime and _is_voxtral_3(resolved_model):
            if is_given(turn_detection):
                server_turn_detection = (
                    turn_detection.copy() if turn_detection is not None else None
                )
            elif not diarize_enabled and vad is None:
                server_turn_detection = {}
        elif is_given(turn_detection) and turn_detection is not None:
            raise ValueError("turn detection is only supported by the latest realtime model")

        if diarize_enabled and server_turn_detection is not None:
            raise ValueError("diarization and turn detection can't be used together")
        if vad is not None and server_turn_detection is not None:
            raise ValueError("a vad and turn detection can't be used together")

        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=is_realtime,
                interim_results=is_realtime,
                diarization=diarize_enabled,
                aligned_transcript=False,
                offline_recognize=not is_realtime,
                manual_flush=is_realtime and server_turn_detection is None and vad is None,
            )
        )
        self._opts = _STTOptions(
            model=resolved_model,
            language=LanguageCode(language) if is_given(language) else None,
            context_bias=list(context_bias) if is_given(context_bias) and context_bias else None,
            target_streaming_delay_ms=target_streaming_delay_ms
            if is_given(target_streaming_delay_ms)
            else None,
            diarize=diarize_enabled,
            turn_detection=server_turn_detection,
        )

        mistral_api_key = api_key if is_given(api_key) else os.environ.get("MISTRAL_API_KEY")
        if not client and not mistral_api_key:
            raise ValueError("Mistral AI API key is required. Set MISTRAL_API_KEY or pass api_key")
        self._client = client or Mistral(api_key=mistral_api_key)
        self._vad = vad if is_realtime else None
        self._streams: weakref.WeakSet[SpeechStream] = weakref.WeakSet()
        self._pool = utils.ConnectionPool[RealtimeConnection](
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
        )

    async def _connect_ws(self, timeout: float) -> RealtimeConnection:
        rt = RealtimeTranscription(self._client.sdk_configuration)
        http_headers = None
        cfg = self._client.sdk_configuration
        client_headers = getattr(cfg.async_client, "headers", None) or getattr(
            cfg.client, "headers", None
        )
        if client_headers:
            http_headers = dict(client_headers)

        async def _connect() -> RealtimeConnection:
            connection = await rt.connect(model=self._opts.model, http_headers=http_headers)
            if session := self._session_fields():
                try:
                    await connection._websocket.send(
                        json.dumps({"type": "session.update", "session": session})
                    )
                except BaseException:
                    await connection.close()
                    raise
            return connection

        return await asyncio.wait_for(_connect(), timeout=timeout)

    def _session_fields(self) -> dict[str, Any]:
        """The realtime session settings, sent with a ``session.update`` message."""
        fields: dict[str, Any] = {}
        if self._opts.target_streaming_delay_ms is not None:
            fields["target_streaming_delay_ms"] = self._opts.target_streaming_delay_ms

        if _is_realtime(self._opts.model) and _is_voxtral_3(self._opts.model):
            if self._opts.language:
                fields["language"] = str(self._opts.language)
            if self._opts.context_bias:
                fields["context_bias"] = self._opts.context_bias
            if self._opts.diarize:
                fields["diarize"] = True
            if self._opts.turn_detection is not None:
                fields["turn_detection"] = dict(self._opts.turn_detection)
        return fields

    async def _close_ws(self, conn: RealtimeConnection) -> None:
        await conn.close()
        ws = conn._websocket
        if ws.keepalive_task is not None:
            ws.keepalive_task.cancel()

    def prewarm(self) -> None:
        if _is_realtime(self._opts.model):
            self._pool.prewarm()

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

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

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

    def update_options(
        self,
        *,
        model: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        context_bias: NotGivenOr[list[str]] = NOT_GIVEN,
        target_streaming_delay_ms: NotGivenOr[int] = NOT_GIVEN,
        diarize: NotGivenOr[bool] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetection | None] = NOT_GIVEN,
    ) -> None:
        """
        Update the STT options.

        Options that don't apply to the model are accepted and ignored, like in the constructor.
        Batch models apply the new options to the next `recognize()` call.
        The realtime session can't be reconfigured once audio has started, so changing a setting of
        the session reconnects the active streams: a transcript that was not finalized yet is lost.

        Args:
            model: The Mistral AI model to use for transcription. It must be of the same kind
                (batch or realtime) as the current one, since the kind is fixed at construction.
            language: The optional language code to use for better transcription accuracy if language is already known (e.g., "fr" for French).
                Used with batch models and the Voxtral 3 realtime model.
            context_bias: Up to 100 words or phrases to guide the model toward good spelling or names or domain-specific vocabulary.
                Used with batch models and the Voxtral 3 realtime model.
            target_streaming_delay_ms: Target streaming delay in milliseconds for realtime mode. Only used with realtime models.
            diarize: Label each piece of text with its speaker. Not supported by earlier realtime models.
                Voxtral 3 doesn't support diarization together with its server-side turn detection: changing
                ``diarize`` on the Voxtral 3 realtime model also resets the turn detection to its default, which is
                disabled with ``diarize`` (or a ``vad``) and enabled otherwise, unless ``turn_detection`` is given.
                Without turn detection, the plugin emits no start and end of speech events, so the voice pipeline
                can't use ``turn_detection="stt"`` unless the STT was created with a ``vad``.
            turn_detection: Server-side turn detection of the Voxtral 3 realtime model. Only supported by that
                model, and not together with ``diarize`` or a ``vad``.

        Raises:
            ValueError: If the model is not of the same kind as the current one, or if the options
                are not supported by the model or can't be used together.
        """
        resolved_model = model if is_given(model) else self._opts.model
        resolved_diarize = diarize if is_given(diarize) else self._opts.diarize
        was_voxtral_3_realtime = _is_realtime(self._opts.model) and _is_voxtral_3(self._opts.model)
        is_voxtral_3_realtime = _is_realtime(resolved_model) and _is_voxtral_3(resolved_model)

        new_turn_detection: TurnDetection | None = None
        if is_given(turn_detection):
            new_turn_detection = turn_detection.copy() if turn_detection is not None else None
        elif is_voxtral_3_realtime:
            if was_voxtral_3_realtime and resolved_diarize == self._opts.diarize:
                new_turn_detection = self._opts.turn_detection
            elif not resolved_diarize and self._vad is None:
                new_turn_detection = {}

        if _is_realtime(resolved_model) != self.capabilities.streaming:
            raise ValueError(
                f"can't switch from {self._opts.model} to {resolved_model}: this STT was created "
                f"with a {'realtime' if self.capabilities.streaming else 'batch'} model, "
                "create a new STT to use the other kind"
            )
        if new_turn_detection is not None and not is_voxtral_3_realtime:
            raise ValueError("turn detection is only supported by latest realtime model")
        if resolved_diarize and _is_realtime(resolved_model) and not is_voxtral_3_realtime:
            raise ValueError("diarization is not supported by this realtime model")
        if resolved_diarize and new_turn_detection is not None:
            raise ValueError("diarization and turn detection can't be used together")
        if self._vad is not None and new_turn_detection is not None:
            raise ValueError("a vad and turn detection can't be used together")

        new_context_bias = (
            (list(context_bias) if context_bias else None)
            if is_given(context_bias)
            else self._opts.context_bias
        )

        session_changed = (
            resolved_model != self._opts.model
            or new_turn_detection != self._opts.turn_detection
            or resolved_diarize != self._opts.diarize
            or (
                is_given(target_streaming_delay_ms)
                and target_streaming_delay_ms != self._opts.target_streaming_delay_ms
            )
            or (
                is_voxtral_3_realtime
                and (
                    (
                        is_given(language)
                        and (LanguageCode(language) if language else None) != self._opts.language
                    )
                    or new_context_bias != self._opts.context_bias
                )
            )
        )

        self._opts.model = resolved_model
        self._opts.diarize = resolved_diarize
        self._opts.turn_detection = new_turn_detection
        if is_given(language):
            self._opts.language = LanguageCode(language) if language else None
        self._opts.context_bias = new_context_bias
        if is_given(target_streaming_delay_ms):
            self._opts.target_streaming_delay_ms = target_streaming_delay_ms
        self._capabilities.diarization = resolved_diarize
        self._capabilities.manual_flush = (
            self.capabilities.streaming and new_turn_detection is None and self._vad is None
        )

        if session_changed and self.capabilities.streaming:
            self._pool.invalidate()
            for stream in self._streams:
                stream._reconnect_event.set()

    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        try:
            # a per-call language applies to this request only
            language_code = LanguageCode(language) if is_given(language) else self._opts.language
            data = rtc.combine_audio_frames(buffer).to_wav_bytes()

            resp = await self._client.audio.transcriptions.complete_async(
                model=self._opts.model,
                file={"content": data, "file_name": "audio.wav"},
                language=language_code if language_code else None,
                context_bias=self._opts.context_bias if self._opts.context_bias else None,
                diarize=self._opts.diarize,
                timestamp_granularities=["segment"],
            )

            return stt.SpeechEvent(
                type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                alternatives=[
                    stt.SpeechData(
                        text=resp.text,
                        language=LanguageCode(resp.language)
                        if resp.language
                        else language_code or LanguageCode(""),
                        start_time=resp.segments[0].start if resp.segments else 0,
                        end_time=resp.segments[-1].end if resp.segments else 0,
                        speaker_id=next(
                            (seg.speaker_id for seg in resp.segments or [] if seg.speaker_id), None
                        ),
                        words=[
                            TimedString(
                                text=segment.text,
                                start_time=segment.start,
                                end_time=segment.end,
                                speaker_id=segment.speaker_id or None,
                            )
                            for segment in resp.segments
                        ]
                        if resp.segments
                        else None,
                    ),
                ],
            )

        except (asyncio.TimeoutError, httpx2.TimeoutException) as e:
            raise APITimeoutError() from e
        except (SDKError, HTTPValidationError) as e:
            raise APIStatusError(
                e.message,
                status_code=e.status_code,
                request_id=e.headers.get("x-request-id"),
                body=e.body,
                retryable=e.status_code != 429,
            ) from e
        except Exception as e:
            raise APIConnectionError() from e

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        stream = SpeechStream(
            stt=self,
            pool=self._pool,
            conn_options=conn_options,
            language=language,
        )
        self._streams.add(stream)
        return stream

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

Create a new instance of MistralAI STT.

Args

client: Optional pre-configured MistralAI client instance.
api_key: Your Mistral AI API key. If not provided, will use the MISTRAL_API_KEY environment variable.
model: The Mistral AI model to use for transcription, default is the realtime "voxtral-mini-transcribe-realtime-3".
language: The optional language code to use for better transcription accuracy if language is already known (e.g., "fr" for French). Used with batch models and the Voxtral 3 realtime model.
context_bias: Up to 100 words or phrases to guide the model toward good spelling or names or domain-specific vocabulary. Used with batch models and the Voxtral 3 realtime model.
target_streaming_delay_ms: Target streaming delay in milliseconds for realtime mode. Only used with realtime models.
diarize: Label each piece of text with its speaker (speaker_id). Supported by the batch models and the Voxtral 3 realtime model, but not by earlier realtime models. Voxtral 3 doesn't support diarization together with its server-side turn detection, so enabling diarize disables turn_detection by default, and passing both raises a ValueError. The plugin then emits no start and end of speech events, and the voice pipeline can't use turn_detection="stt", unless a vad is given to provide them. The realtime model numbers the speakers within one transcript only: the numbering restarts after each flush, so the speakers of different transcripts can't be compared.
turn_detection: Server-side turn detection of the Voxtral 3 realtime model, which provides the start and end of speech events and finalizes the transcript of each turn. Enabled with the server defaults when not provided, unless diarize is enabled or a vad is given, which it can't be used with. Pass a :class:TurnDetection dict to tune it, or None to disable it. Only supported by the Voxtral 3 realtime model.
vad: Optional Voice Activity Detector run by the plugin on the realtime models: its end of speech flushes the audio to get the final transcript, and its start and end of speech are emitted as speech events, which the voice pipeline needs for turn_detection="stt" when the server doesn't provide them (earlier realtime models, or Voxtral 3 with diarize). It takes precedence over the server-side turn detection of Voxtral 3, which is disabled by default, and can't be used with an explicit turn_detection. Only used with realtime models.

Note

Realtime models without server-side turn detection or a vad (earlier models, or Voxtral 3 with turn_detection=None or diarize) have no endpointing. The voice pipeline (AgentSession) requests a flush from its own VAD and turn detection to get the final transcripts. A stream used on its own needs to be flushed with stream.flush().

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 "MistralAI"

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

async def aclose(self) ‑> None

Expand source code

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

Close the STT, and every stream/requests associated with it

def prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    if _is_realtime(self._opts.model):
        self._pool.prewarm()

Pre-warm connection to the STT service

def stream(self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0)) ‑> livekit.plugins.mistralai.stt.SpeechStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SpeechStream:
    stream = SpeechStream(
        stt=self,
        pool=self._pool,
        conn_options=conn_options,
        language=language,
    )
    self._streams.add(stream)
    return stream

def update_options(self, *, model: NotGivenOr[str] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, context_bias: NotGivenOr[list[str]] = NOT_GIVEN, target_streaming_delay_ms: NotGivenOr[int] = NOT_GIVEN, diarize: NotGivenOr[bool] = NOT_GIVEN, turn_detection: NotGivenOr[TurnDetection | None] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[str] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    context_bias: NotGivenOr[list[str]] = NOT_GIVEN,
    target_streaming_delay_ms: NotGivenOr[int] = NOT_GIVEN,
    diarize: NotGivenOr[bool] = NOT_GIVEN,
    turn_detection: NotGivenOr[TurnDetection | None] = NOT_GIVEN,
) -> None:
    """
    Update the STT options.

    Options that don't apply to the model are accepted and ignored, like in the constructor.
    Batch models apply the new options to the next `recognize()` call.
    The realtime session can't be reconfigured once audio has started, so changing a setting of
    the session reconnects the active streams: a transcript that was not finalized yet is lost.

    Args:
        model: The Mistral AI model to use for transcription. It must be of the same kind
            (batch or realtime) as the current one, since the kind is fixed at construction.
        language: The optional language code to use for better transcription accuracy if language is already known (e.g., "fr" for French).
            Used with batch models and the Voxtral 3 realtime model.
        context_bias: Up to 100 words or phrases to guide the model toward good spelling or names or domain-specific vocabulary.
            Used with batch models and the Voxtral 3 realtime model.
        target_streaming_delay_ms: Target streaming delay in milliseconds for realtime mode. Only used with realtime models.
        diarize: Label each piece of text with its speaker. Not supported by earlier realtime models.
            Voxtral 3 doesn't support diarization together with its server-side turn detection: changing
            ``diarize`` on the Voxtral 3 realtime model also resets the turn detection to its default, which is
            disabled with ``diarize`` (or a ``vad``) and enabled otherwise, unless ``turn_detection`` is given.
            Without turn detection, the plugin emits no start and end of speech events, so the voice pipeline
            can't use ``turn_detection="stt"`` unless the STT was created with a ``vad``.
        turn_detection: Server-side turn detection of the Voxtral 3 realtime model. Only supported by that
            model, and not together with ``diarize`` or a ``vad``.

    Raises:
        ValueError: If the model is not of the same kind as the current one, or if the options
            are not supported by the model or can't be used together.
    """
    resolved_model = model if is_given(model) else self._opts.model
    resolved_diarize = diarize if is_given(diarize) else self._opts.diarize
    was_voxtral_3_realtime = _is_realtime(self._opts.model) and _is_voxtral_3(self._opts.model)
    is_voxtral_3_realtime = _is_realtime(resolved_model) and _is_voxtral_3(resolved_model)

    new_turn_detection: TurnDetection | None = None
    if is_given(turn_detection):
        new_turn_detection = turn_detection.copy() if turn_detection is not None else None
    elif is_voxtral_3_realtime:
        if was_voxtral_3_realtime and resolved_diarize == self._opts.diarize:
            new_turn_detection = self._opts.turn_detection
        elif not resolved_diarize and self._vad is None:
            new_turn_detection = {}

    if _is_realtime(resolved_model) != self.capabilities.streaming:
        raise ValueError(
            f"can't switch from {self._opts.model} to {resolved_model}: this STT was created "
            f"with a {'realtime' if self.capabilities.streaming else 'batch'} model, "
            "create a new STT to use the other kind"
        )
    if new_turn_detection is not None and not is_voxtral_3_realtime:
        raise ValueError("turn detection is only supported by latest realtime model")
    if resolved_diarize and _is_realtime(resolved_model) and not is_voxtral_3_realtime:
        raise ValueError("diarization is not supported by this realtime model")
    if resolved_diarize and new_turn_detection is not None:
        raise ValueError("diarization and turn detection can't be used together")
    if self._vad is not None and new_turn_detection is not None:
        raise ValueError("a vad and turn detection can't be used together")

    new_context_bias = (
        (list(context_bias) if context_bias else None)
        if is_given(context_bias)
        else self._opts.context_bias
    )

    session_changed = (
        resolved_model != self._opts.model
        or new_turn_detection != self._opts.turn_detection
        or resolved_diarize != self._opts.diarize
        or (
            is_given(target_streaming_delay_ms)
            and target_streaming_delay_ms != self._opts.target_streaming_delay_ms
        )
        or (
            is_voxtral_3_realtime
            and (
                (
                    is_given(language)
                    and (LanguageCode(language) if language else None) != self._opts.language
                )
                or new_context_bias != self._opts.context_bias
            )
        )
    )

    self._opts.model = resolved_model
    self._opts.diarize = resolved_diarize
    self._opts.turn_detection = new_turn_detection
    if is_given(language):
        self._opts.language = LanguageCode(language) if language else None
    self._opts.context_bias = new_context_bias
    if is_given(target_streaming_delay_ms):
        self._opts.target_streaming_delay_ms = target_streaming_delay_ms
    self._capabilities.diarization = resolved_diarize
    self._capabilities.manual_flush = (
        self.capabilities.streaming and new_turn_detection is None and self._vad is None
    )

    if session_changed and self.capabilities.streaming:
        self._pool.invalidate()
        for stream in self._streams:
            stream._reconnect_event.set()

Update the STT options.

Options that don't apply to the model are accepted and ignored, like in the constructor. Batch models apply the new options to the next recognize() call. The realtime session can't be reconfigured once audio has started, so changing a setting of the session reconnects the active streams: a transcript that was not finalized yet is lost.

Args

model: The Mistral AI model to use for transcription. It must be of the same kind (batch or realtime) as the current one, since the kind is fixed at construction.
language: The optional language code to use for better transcription accuracy if language is already known (e.g., "fr" for French). Used with batch models and the Voxtral 3 realtime model.
context_bias: Up to 100 words or phrases to guide the model toward good spelling or names or domain-specific vocabulary. Used with batch models and the Voxtral 3 realtime model.
target_streaming_delay_ms: Target streaming delay in milliseconds for realtime mode. Only used with realtime models.
diarize: Label each piece of text with its speaker. Not supported by earlier realtime models. Voxtral 3 doesn't support diarization together with its server-side turn detection: changing diarize on the Voxtral 3 realtime model also resets the turn detection to its default, which is disabled with diarize (or a vad) and enabled otherwise, unless turn_detection is given. Without turn detection, the plugin emits no start and end of speech events, so the voice pipeline can't use turn_detection="stt" unless the STT was created with a vad.
turn_detection: Server-side turn detection of the Voxtral 3 realtime model. Only supported by that model, and not together with diarize or a vad.

Raises

ValueError: If the model is not of the same kind as the current one, or if the options are not supported by the model or can't be used together.

Inherited members

EventEmitter:
- emit
- off
- on
- once

class TTS (client: Mistral | None = None, api_key: NotGivenOr[str] = NOT_GIVEN, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN, ref_audio: NotGivenOr[str] = NOT_GIVEN, response_format: NotGivenOr[RESPONSE_FORMAT] = NOT_GIVEN, expressiveness: NotGivenOr[float] = NOT_GIVEN, phoneme_hints: NotGivenOr[dict[str, list[models.IPAPhoneme]]] = NOT_GIVEN, use_realtime: bool = False)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        client: Mistral | None = None,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
        voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
        ref_audio: NotGivenOr[str] = NOT_GIVEN,
        response_format: NotGivenOr[RESPONSE_FORMAT] = NOT_GIVEN,
        expressiveness: NotGivenOr[float] = NOT_GIVEN,
        phoneme_hints: NotGivenOr[dict[str, list[models.IPAPhoneme]]] = NOT_GIVEN,
        use_realtime: bool = False,
    ) -> None:
        """
        Create a new instance of MistralAI TTS.

        Args:
            client: Optional pre-configured MistralAI client instance.
            api_key: Your Mistral AI API key. If not provided, will use the MISTRAL_API_KEY environment variable.
            model: The Mistral AI model to use for text-to-speech, default is "voxtral-mini-tts-3".
            voice: The voice ID to use for synthesis. Mutually exclusive with ``ref_audio``. Defaults to ``en_paul_neutral`` when neither is given.
                The ``voice_<n>`` voices are only supported by latest model, which also accept the older voices.
            ref_audio: Base64-encoded audio sample (3–25 s) for zero-shot voice cloning. Mutually exclusive with ``voice``.
            response_format: The audio format of synthesized speech, between ``mp3``, ``wav``, ``pcm``, ``opus`` or ``flac``. Defaults to ``mp3``.
                Ignored with ``use_realtime``: the speech streamed over the websocket is PCM.
            expressiveness: How expressive the speech is, between 1 and 5. Only supported by latest model.
            phoneme_hints: Pronunciation hints mapping a (case-sensitive) word to its ordered list of IPA phones, e.g.
                ``{"tomato": ["t", "ə", "m", "ɑ", "t", "oʊ"]}``. Only supported by latest model, and not with ``use_realtime``.
            use_realtime: Whether to stream the speech over a realtime websocket while the text is generated,
                instead of synthesizing it sentence by sentence over http. It doesn't support
                ``phoneme_hints``. Only supported by latest model. Defaults to False.
        """
        if is_given(voice) and is_given(ref_audio):
            raise ValueError("Only one of 'voice' or 'ref_audio' may be provided, not both")

        resolved_model = model if is_given(model) else DEFAULT_MODEL
        resolved_hints = (phoneme_hints if is_given(phoneme_hints) else None) or None
        if is_given(expressiveness) and not _is_voxtral_3(resolved_model):
            raise ValueError("expressiveness is only supported by latest speech model")
        _check_voice(resolved_model, voice if is_given(voice) else None)
        _check_realtime(resolved_model, use_realtime, resolved_hints)

        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=use_realtime),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
        )
        self._opts = _TTSOptions(
            model=resolved_model,
            voice=voice if is_given(voice) else (None if is_given(ref_audio) else DEFAULT_VOICE),
            ref_audio=ref_audio if is_given(ref_audio) else None,
            response_format=response_format
            if is_given(response_format)
            else DEFAULT_RESPONSE_FORMAT,
            expressiveness=expressiveness if is_given(expressiveness) else None,
            phoneme_hints=resolved_hints,
            use_realtime=use_realtime,
        )

        mistral_api_key = api_key if is_given(api_key) else os.environ.get("MISTRAL_API_KEY")
        if not client and not mistral_api_key:
            raise ValueError("Mistral AI API key is required. Set MISTRAL_API_KEY or pass api_key")
        self._client = client or Mistral(api_key=mistral_api_key)
        self._pool = utils.ConnectionPool[ClientConnection](
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
        )

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

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

    def update_options(
        self,
        *,
        model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
        voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
        ref_audio: NotGivenOr[str] = NOT_GIVEN,
        response_format: NotGivenOr[RESPONSE_FORMAT] = NOT_GIVEN,
        expressiveness: NotGivenOr[float] = NOT_GIVEN,
        phoneme_hints: NotGivenOr[dict[str, list[models.IPAPhoneme]]] = NOT_GIVEN,
        use_realtime: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        """
        Update the TTS options.

        Args:
            model: The model to use for text-to-speech. Clears ``ref_audio``.
            voice: The voice ID to use for synthesis. The ``voice_<n>`` voices are only supported by the Voxtral 3 models.
            ref_audio: Base64-encoded audio sample for zero-shot voice cloning. Clears ``voice``.
            response_format: The audio format of synthesized speech, between ``mp3``, ``wav``, ``pcm``, ``opus`` or ``flac``. Defaults to ``mp3``.
            expressiveness: How expressive the speech is, between 1 and 5. Only supported by the Voxtral 3 models.
            phoneme_hints: Pronunciation hints mapping a word to its IPA phones. Only supported by the Voxtral 3 models,
                and not with ``use_realtime``.
            use_realtime: Whether to stream the speech over the realtime websocket api. Only supported by the Voxtral 3 models.
        """
        if is_given(voice) and is_given(ref_audio):
            raise ValueError("Only one of 'voice' or 'ref_audio' may be provided, not both")

        resolved_model = model if is_given(model) else self._opts.model
        resolved_expressiveness = (
            expressiveness if is_given(expressiveness) else self._opts.expressiveness
        )
        if resolved_expressiveness is not None and not _is_voxtral_3(resolved_model):
            raise ValueError("expressiveness is only supported by latest speech model")
        resolved_voice: str | None
        if is_given(voice):
            resolved_voice = voice
        else:
            resolved_voice = None if is_given(ref_audio) else self._opts.voice
        _check_voice(resolved_model, resolved_voice)
        resolved_hints = (
            phoneme_hints if is_given(phoneme_hints) else self._opts.phoneme_hints
        ) or None
        resolved_realtime = use_realtime if is_given(use_realtime) else self._opts.use_realtime
        _check_realtime(resolved_model, resolved_realtime, resolved_hints)

        session = (self._opts.model, self._session())

        if is_given(model):
            self._opts.model = model
        if is_given(voice):
            self._opts.voice = voice
            self._opts.ref_audio = None
        if is_given(ref_audio):
            self._opts.ref_audio = ref_audio
            self._opts.voice = None
        if is_given(response_format):
            self._opts.response_format = response_format
        self._opts.expressiveness = resolved_expressiveness
        self._opts.phoneme_hints = resolved_hints
        self._opts.use_realtime = resolved_realtime

        self._capabilities.streaming = resolved_realtime
        if session != (self._opts.model, self._session()):
            self._pool.invalidate()

    def prewarm(self) -> None:
        if self._capabilities.streaming:
            self._pool.prewarm()

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

    def stream(
        self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.SynthesizeStream:
        if not self._capabilities.streaming:
            return super().stream(conn_options=conn_options)
        return SynthesizeStream(tts=self, conn_options=conn_options)

    async def aclose(self) -> None:
        await self._pool.aclose()
        await super().aclose()

    async def _acquire(self, timeout: float) -> ClientConnection:
        while True:
            ws = await self._pool.get(timeout=timeout)
            if ws.state is State.OPEN:
                return ws
            self._pool.remove(ws)

    async def _connect_ws(self, timeout: float) -> ClientConnection:
        config = self._client.sdk_configuration
        base_url, _ = config.get_server_details()
        url = generate_url(base_url, "/v1/audio/speech/realtime", None)
        url = url.replace("https://", "wss://", 1).replace("http://", "ws://", 1)

        query = {"model": self._opts.model}
        headers: dict[str, str] = {}
        security = config.security
        if callable(security):
            security = security()
        resolved_security = get_security_from_env(security, models.Security)
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            query |= {key: values[-1] for key, values in security_query.items() if values}

        async def _connect() -> ClientConnection:
            ws = await connect(
                f"{url}?{urlencode(query)}",
                additional_headers=headers,
                user_agent_header=config.user_agent,
                open_timeout=timeout,
            )
            try:
                await _expect(ws, "session.created")
                # configure the whole session with a single update before any text is sent
                await ws.send(json.dumps({"type": "session.update", "session": self._session()}))
                await _expect(ws, "session.updated")
            except BaseException:
                await ws.close()
                raise
            return ws

        return await asyncio.wait_for(_connect(), timeout=timeout)

    def _session(self) -> dict[str, Any]:
        """The settings of the realtime session, sent with a ``session.update`` message."""
        session: dict[str, Any] = {"streaming_mode": "responsive", "response_format": "pcm"}
        if self._opts.ref_audio is not None:
            session["ref_audio"] = self._opts.ref_audio
        elif self._opts.voice is not None:
            session["voice"] = self._opts.voice
        if self._opts.expressiveness is not None:
            session["expressiveness"] = self._opts.expressiveness
        return session

    async def _close_ws(self, ws: ClientConnection) -> None:
        await ws.close()

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

Create a new instance of MistralAI TTS.

Args

client: Optional pre-configured MistralAI client instance.
api_key: Your Mistral AI API key. If not provided, will use the MISTRAL_API_KEY environment variable.
model: The Mistral AI model to use for text-to-speech, default is "voxtral-mini-tts-3".
voice: The voice ID to use for synthesis. Mutually exclusive with ref_audio. Defaults to en_paul_neutral when neither is given. The voice_<n> voices are only supported by latest model, which also accept the older voices.
ref_audio: Base64-encoded audio sample (3–25 s) for zero-shot voice cloning. Mutually exclusive with voice.
response_format: The audio format of synthesized speech, between mp3, wav, pcm, opus or flac. Defaults to mp3. Ignored with use_realtime: the speech streamed over the websocket is PCM.
expressiveness: How expressive the speech is, between 1 and 5. Only supported by latest model.
phoneme_hints: Pronunciation hints mapping a (case-sensitive) word to its ordered list of IPA phones, e.g. {"tomato": ["t", "ə", "m", "ɑ", "t", "oʊ"]}. Only supported by latest model, and not with use_realtime.
use_realtime: Whether to stream the speech over a realtime websocket while the text is generated, instead of synthesizing it sentence by sentence over http. It doesn't support phoneme_hints. Only supported by latest model. Defaults to False.

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

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 "MistralAI"

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:
    await self._pool.aclose()
    await super().aclose()

def prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    if self._capabilities.streaming:
        self._pool.prewarm()

Pre-warm connection to the TTS service

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

Expand source code

def stream(
    self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
) -> tts.SynthesizeStream:
    if not self._capabilities.streaming:
        return super().stream(conn_options=conn_options)
    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, liveness_timeout=6.0)) ‑> livekit.agents.tts.tts.ChunkedStream

Expand source code

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

def update_options(self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN, ref_audio: NotGivenOr[str] = NOT_GIVEN, response_format: NotGivenOr[RESPONSE_FORMAT] = NOT_GIVEN, expressiveness: NotGivenOr[float] = NOT_GIVEN, phoneme_hints: NotGivenOr[dict[str, list[models.IPAPhoneme]]] = NOT_GIVEN, use_realtime: NotGivenOr[bool] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
    voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
    ref_audio: NotGivenOr[str] = NOT_GIVEN,
    response_format: NotGivenOr[RESPONSE_FORMAT] = NOT_GIVEN,
    expressiveness: NotGivenOr[float] = NOT_GIVEN,
    phoneme_hints: NotGivenOr[dict[str, list[models.IPAPhoneme]]] = NOT_GIVEN,
    use_realtime: NotGivenOr[bool] = NOT_GIVEN,
) -> None:
    """
    Update the TTS options.

    Args:
        model: The model to use for text-to-speech. Clears ``ref_audio``.
        voice: The voice ID to use for synthesis. The ``voice_<n>`` voices are only supported by the Voxtral 3 models.
        ref_audio: Base64-encoded audio sample for zero-shot voice cloning. Clears ``voice``.
        response_format: The audio format of synthesized speech, between ``mp3``, ``wav``, ``pcm``, ``opus`` or ``flac``. Defaults to ``mp3``.
        expressiveness: How expressive the speech is, between 1 and 5. Only supported by the Voxtral 3 models.
        phoneme_hints: Pronunciation hints mapping a word to its IPA phones. Only supported by the Voxtral 3 models,
            and not with ``use_realtime``.
        use_realtime: Whether to stream the speech over the realtime websocket api. Only supported by the Voxtral 3 models.
    """
    if is_given(voice) and is_given(ref_audio):
        raise ValueError("Only one of 'voice' or 'ref_audio' may be provided, not both")

    resolved_model = model if is_given(model) else self._opts.model
    resolved_expressiveness = (
        expressiveness if is_given(expressiveness) else self._opts.expressiveness
    )
    if resolved_expressiveness is not None and not _is_voxtral_3(resolved_model):
        raise ValueError("expressiveness is only supported by latest speech model")
    resolved_voice: str | None
    if is_given(voice):
        resolved_voice = voice
    else:
        resolved_voice = None if is_given(ref_audio) else self._opts.voice
    _check_voice(resolved_model, resolved_voice)
    resolved_hints = (
        phoneme_hints if is_given(phoneme_hints) else self._opts.phoneme_hints
    ) or None
    resolved_realtime = use_realtime if is_given(use_realtime) else self._opts.use_realtime
    _check_realtime(resolved_model, resolved_realtime, resolved_hints)

    session = (self._opts.model, self._session())

    if is_given(model):
        self._opts.model = model
    if is_given(voice):
        self._opts.voice = voice
        self._opts.ref_audio = None
    if is_given(ref_audio):
        self._opts.ref_audio = ref_audio
        self._opts.voice = None
    if is_given(response_format):
        self._opts.response_format = response_format
    self._opts.expressiveness = resolved_expressiveness
    self._opts.phoneme_hints = resolved_hints
    self._opts.use_realtime = resolved_realtime

    self._capabilities.streaming = resolved_realtime
    if session != (self._opts.model, self._session()):
        self._pool.invalidate()

Update the TTS options.

Args

model: The model to use for text-to-speech. Clears ref_audio.
voice: The voice ID to use for synthesis. The voice_<n> voices are only supported by the Voxtral 3 models.
ref_audio: Base64-encoded audio sample for zero-shot voice cloning. Clears voice.
response_format: The audio format of synthesized speech, between mp3, wav, pcm, opus or flac. Defaults to mp3.
expressiveness: How expressive the speech is, between 1 and 5. Only supported by the Voxtral 3 models.
phoneme_hints: Pronunciation hints mapping a word to its IPA phones. Only supported by the Voxtral 3 models, and not with use_realtime.
use_realtime: Whether to stream the speech over the realtime websocket api. Only supported by the Voxtral 3 models.

Inherited members

EventEmitter:
- emit
- off
- on
- once

class TurnDetection (*args, **kwargs)

Expand source code

class TurnDetection(TypedDict, total=False):
    """Server-side turn detection settings of the Voxtral 3 realtime models.

    All fields are optional, the server applies its defaults to the ones left out.
    The thresholds must satisfy ``start_threshold >= eager_end_threshold > end_threshold``
    and ``end_timeout_ms`` must be at least ``min_silence_ms``.
    """

    start_threshold: float
    eager_end_threshold: float
    end_threshold: float
    min_silence_ms: int
    end_timeout_ms: int | None

Server-side turn detection settings of the Voxtral 3 realtime models.

All fields are optional, the server applies its defaults to the ones left out. The thresholds must satisfy start_threshold >= eager_end_threshold > end_threshold and end_timeout_ms must be at least min_silence_ms.

Ancestors

builtins.dict

Class variables

var eager_end_threshold : float
var end_threshold : float
var end_timeout_ms : int | None
var min_silence_ms : int
var start_threshold : float