Module livekit.plugins.sarvam

Sarvam.ai plugin for LiveKit Agents.

Support for speech-to-text, text-to-speech, and LLM with Sarvam.ai.

Sarvam.ai provides high-quality STT and TTS for Indian languages and OpenAI-compatible LLMs.

For API access, visit https://sarvam.ai/

Classes

class LLM (*,
model: str | SarvamLLMModels = 'sarvam-105b',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
max_tokens: NotGivenOr[int] = NOT_GIVEN,
wiki_grounding: NotGivenOr[bool] = NOT_GIVEN,
stop: NotGivenOr[str | list[str]] = NOT_GIVEN,
n: NotGivenOr[int] = NOT_GIVEN,
seed: NotGivenOr[int] = NOT_GIVEN,
frequency_penalty: NotGivenOr[float] = NOT_GIVEN,
presence_penalty: NotGivenOr[float] = NOT_GIVEN,
extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
timeout: httpx.Timeout | None = None)

Expand source code

class LLM(OpenAILLM):
    """Sarvam LLM service — OpenAI-compatible chat completions.

    Supports four models:

    * ``gemma4`` — vision-capable, ``/v2`` endpoint
    * ``sarvam-105b`` — text-only, ``/v2`` endpoint
    * ``glm5.2`` — text-only, ``/v2`` endpoint
    * ``sarvam-105b-conversations`` — multi-turn optimized, ``/v1`` endpoint

    The endpoint (``/v1`` vs ``/v2``) is resolved automatically from the model.
    An explicit ``base_url`` always overrides automatic resolution.
    """

    def __init__(
        self,
        *,
        model: str | SarvamLLMModels = "sarvam-105b",
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        max_tokens: NotGivenOr[int] = NOT_GIVEN,
        wiki_grounding: NotGivenOr[bool] = NOT_GIVEN,
        stop: NotGivenOr[str | list[str]] = NOT_GIVEN,
        n: NotGivenOr[int] = NOT_GIVEN,
        seed: NotGivenOr[int] = NOT_GIVEN,
        frequency_penalty: NotGivenOr[float] = NOT_GIVEN,
        presence_penalty: NotGivenOr[float] = NOT_GIVEN,
        extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
        extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
    ) -> None:
        """
        Create a new instance of Sarvam LLM.

        ``api_key`` must be set to your Sarvam API key, either using the
        argument or by setting the ``SARVAM_API_KEY`` environment variable.
        """
        validated_model = _validate_model(model)
        sarvam_api_key = _get_api_key(api_key)

        # Resolve base URL: explicit > model-derived
        resolved_base_url = base_url if is_given(base_url) else _resolve_base_url(validated_model)

        # ---- Merge auth / telemetry headers (always enforced) ----
        merged_headers: dict[str, str] = {}
        if is_given(extra_headers):
            merged_headers.update(extra_headers)
        merged_headers["api-subscription-key"] = sarvam_api_key
        merged_headers["User-Agent"] = USER_AGENT

        # ---- Build extra_body with Sarvam-specific fields ----
        merged_body: dict[str, Any] = {}
        if is_given(extra_body):
            merged_body.update(extra_body)
        if is_given(max_tokens):
            merged_body["max_tokens"] = max_tokens
        if is_given(stop):
            merged_body["stop"] = stop
        if is_given(n):
            merged_body["n"] = n
        if is_given(seed):
            merged_body["seed"] = seed
        if is_given(frequency_penalty):
            merged_body["frequency_penalty"] = frequency_penalty
        if is_given(presence_penalty):
            merged_body["presence_penalty"] = presence_penalty

        # wiki_grounding — only for supported models
        if is_given(wiki_grounding):
            if validated_model in _WIKI_GROUNDING_MODELS:
                merged_body["wiki_grounding"] = wiki_grounding
            # silently drop for unsupported models

        filtered_body = _filter_extra_body(merged_body)

        # reasoning_effort — only for supported models
        effective_reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN
        if is_given(reasoning_effort):
            if validated_model in _REASONING_EFFORT_MODELS:
                effective_reasoning_effort = reasoning_effort
            # silently drop for unsupported models

        super().__init__(
            model=validated_model,
            api_key=sarvam_api_key,
            base_url=resolved_base_url,
            client=client,
            user=user,
            temperature=temperature,
            top_p=top_p,
            tool_choice=tool_choice,
            reasoning_effort=effective_reasoning_effort,
            extra_headers=merged_headers,
            extra_body=filtered_body if filtered_body else NOT_GIVEN,
            timeout=timeout,
        )

        # Track the API key and version for runtime model switching
        self._sarvam_api_key = sarvam_api_key
        self._sarvam_api_version = _api_version(validated_model)

    # ------------------------------------------------------------------
    # Public overrides
    # ------------------------------------------------------------------

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

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

    def update_options(
        self,
        *,
        model: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        """Update the model at runtime.

        Validates the new model and recreates the underlying client when the
        API version changes (``/v1`` ↔ ``/v2``).  When both old and new models
        share the same endpoint, the client is **not** recreated.
        """
        if not is_given(model):
            return

        new_model = _validate_model(model)
        new_version = _api_version(new_model)

        if new_version != self._sarvam_api_version:
            # Endpoint changed — recreate the client pointing at the new URL
            new_base_url = _resolve_base_url(new_model)
            self._client = _create_sarvam_client(
                api_key=self._sarvam_api_key,
                base_url=new_base_url,
            )
            self._sarvam_api_version = new_version

        self._opts.model = new_model

    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,
        response_format: NotGivenOr[Any] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> Any:
        """Build a chat-completion stream with Sarvam-specific param stripping.

        * Strips ``stream_options``, ``max_completion_tokens``, ``service_tier``
          (the Sarvam API does not accept these OpenAI SDK fields).
        * Strips ``reasoning_effort`` for models that don't support it.
        * Rejects images sent to non-vision models (client-side ``ValueError``).
        * Rejects ``tool_choice`` without a non-empty ``tools`` array.
        """
        model = self._opts.model

        # --- Image rejection on non-vision models ---
        if model not in _VISION_MODELS and _has_image_content(chat_ctx):
            raise ValueError(
                f"Image input is not supported for model '{model}'. "
                f"Use 'gemma4' for vision capabilities."
            )

        # --- tool_choice without tools ---
        # 'none' and 'auto' are always valid (they don't require tools).
        # 'required' and named-function tool_choice require a non-empty tools array.
        effective_tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
        effective_tools = tools or []
        if is_given(effective_tool_choice) and not effective_tools:
            tc_str = effective_tool_choice if isinstance(effective_tool_choice, str) else "function"
            if tc_str not in ("none", "auto"):
                raise ValueError(
                    "tool_choice requires a non-empty tools array. "
                    "Provide tools or set tool_choice to 'none' or 'auto'."
                )

        # --- Strip unsupported fields from caller-provided extra_kwargs ---
        merged_extra: dict[str, Any] = {}
        if is_given(extra_kwargs):
            merged_extra.update(extra_kwargs)

        for field in _UNSUPPORTED_OAI_FIELDS:
            merged_extra.pop(field, None)

        # Strip reasoning_effort for unsupported models at chat-time too
        if model not in _REASONING_EFFORT_MODELS:
            merged_extra.pop("reasoning_effort", None)

        return super().chat(
            chat_ctx=chat_ctx,
            tools=tools,
            conn_options=conn_options,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            response_format=response_format,
            extra_kwargs=merged_extra if merged_extra else NOT_GIVEN,
        )

Sarvam LLM service — OpenAI-compatible chat completions.

Supports four models:

• gemma4 — vision-capable, /v2 endpoint
• sarvam-105b — text-only, /v2 endpoint
• glm5.2 — text-only, /v2 endpoint
• sarvam-105b-conversations — multi-turn optimized, /v1 endpoint

The endpoint (/v1 vs /v2) is resolved automatically from the model. An explicit base_url always overrides automatic resolution.

Create a new instance of Sarvam LLM.

api_key must be set to your Sarvam API key, either using the argument or by setting the SARVAM_API_KEY environment variable.

Ancestors

• livekit.plugins.openai.llm.LLM
• 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 "Sarvam"

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),
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
response_format: NotGivenOr[Any] = NOT_GIVEN,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> Any

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,
    response_format: NotGivenOr[Any] = NOT_GIVEN,
    extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> Any:
    """Build a chat-completion stream with Sarvam-specific param stripping.

    * Strips ``stream_options``, ``max_completion_tokens``, ``service_tier``
      (the Sarvam API does not accept these OpenAI SDK fields).
    * Strips ``reasoning_effort`` for models that don't support it.
    * Rejects images sent to non-vision models (client-side ``ValueError``).
    * Rejects ``tool_choice`` without a non-empty ``tools`` array.
    """
    model = self._opts.model

    # --- Image rejection on non-vision models ---
    if model not in _VISION_MODELS and _has_image_content(chat_ctx):
        raise ValueError(
            f"Image input is not supported for model '{model}'. "
            f"Use 'gemma4' for vision capabilities."
        )

    # --- tool_choice without tools ---
    # 'none' and 'auto' are always valid (they don't require tools).
    # 'required' and named-function tool_choice require a non-empty tools array.
    effective_tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
    effective_tools = tools or []
    if is_given(effective_tool_choice) and not effective_tools:
        tc_str = effective_tool_choice if isinstance(effective_tool_choice, str) else "function"
        if tc_str not in ("none", "auto"):
            raise ValueError(
                "tool_choice requires a non-empty tools array. "
                "Provide tools or set tool_choice to 'none' or 'auto'."
            )

    # --- Strip unsupported fields from caller-provided extra_kwargs ---
    merged_extra: dict[str, Any] = {}
    if is_given(extra_kwargs):
        merged_extra.update(extra_kwargs)

    for field in _UNSUPPORTED_OAI_FIELDS:
        merged_extra.pop(field, None)

    # Strip reasoning_effort for unsupported models at chat-time too
    if model not in _REASONING_EFFORT_MODELS:
        merged_extra.pop("reasoning_effort", None)

    return super().chat(
        chat_ctx=chat_ctx,
        tools=tools,
        conn_options=conn_options,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        response_format=response_format,
        extra_kwargs=merged_extra if merged_extra else NOT_GIVEN,
    )

Build a chat-completion stream with Sarvam-specific param stripping.

• Strips stream_options, max_completion_tokens, service_tier (the Sarvam API does not accept these OpenAI SDK fields).
• Strips reasoning_effort for models that don't support it.
• Rejects images sent to non-vision models (client-side ValueError).
• Rejects tool_choice without a non-empty tools array.

def update_options(self, *, model: NotGivenOr[str] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[str] = NOT_GIVEN,
) -> None:
    """Update the model at runtime.

    Validates the new model and recreates the underlying client when the
    API version changes (``/v1`` ↔ ``/v2``).  When both old and new models
    share the same endpoint, the client is **not** recreated.
    """
    if not is_given(model):
        return

    new_model = _validate_model(model)
    new_version = _api_version(new_model)

    if new_version != self._sarvam_api_version:
        # Endpoint changed — recreate the client pointing at the new URL
        new_base_url = _resolve_base_url(new_model)
        self._client = _create_sarvam_client(
            api_key=self._sarvam_api_key,
            base_url=new_base_url,
        )
        self._sarvam_api_version = new_version

    self._opts.model = new_model

Update the model at runtime.

Validates the new model and recreates the underlying client when the API version changes (/v1 ↔ /v2). When both old and new models share the same endpoint, the client is not recreated.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class RealtimeSpeechStream (*,
stt: STTRealtime,
opts: RealtimeSTTOptions,
conn_options: APIConnectOptions,
http_session: aiohttp.ClientSession)

Expand source code

class RealtimeSpeechStream(stt.SpeechStream):
    """A single WebSocket session against Sarvam's realtime STT endpoint.

    Audio pushed into the stream is forwarded in the configured wire encoding, and
    the events Sarvam returns are translated into LiveKit speech events. Audio
    duration is reported incrementally while the session runs and reconciled
    against the server's authoritative total when the session ends.
    """

    def __init__(
        self,
        *,
        stt: STTRealtime,
        opts: RealtimeSTTOptions,
        conn_options: APIConnectOptions,
        http_session: aiohttp.ClientSession,
    ) -> None:
        """Create a realtime speech stream.

        Args:
            stt: The parent instance that created this stream.
            opts: Resolved options for this connection.
            conn_options: Connection options for this stream.
            http_session: aiohttp session used to open the WebSocket.
        """
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
        self._opts = opts
        self._session = http_session
        self._ws: aiohttp.ClientWebSocketResponse | None = None
        self._request_id = ""
        self._session_id = ""
        self._resolved_config: dict[str, Any] | None = None
        self._session_ended = False
        self._utterance_idx: int | None = None
        self._utterance_in_progress = False
        self._active_endpointing = opts.endpointing
        self._pending_endpointing: RealtimeEndpointing | str | None = None
        self._endpointing_update_acknowledged = False
        self._endpointing_update_sent = False
        self._pending_config_update: dict[str, Any] | None = None
        self._manual_speech_started = False
        self._flush_observed = False
        self._pending_final_data: dict[str, Any] | None = None
        self._utterance_start_audio_pos = 0.0
        self._utterance_speech_end_audio_pos: float | None = None
        self._utterance_speech_end_wall: float | None = None
        self._final_received_for_utterance = False
        self._eos_emitted_for_utterance = False
        self._stream_started_at = time.time()
        self._audio_position = 0.0
        self._local_audio_duration = 0.0
        self._total_reported_audio_duration = 0.0
        self._server_audio_duration_reported = False
        self._audio_duration_collector = PeriodicCollector(
            callback=self._on_audio_duration_report,
            duration=5.0,
        )
        self._logger = logger

    @property
    def resolved_config(self) -> dict[str, Any] | None:
        """Return the configuration resolved by Sarvam for this connection."""
        return dict(self._resolved_config) if self._resolved_config is not None else None

    def update_options(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
        mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
        endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str | None] = NOT_GIVEN,
        return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
        vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
    ) -> None:
        """Apply an option change to this live connection.

        Only the options explicitly passed here are changed, so per-stream overrides
        such as a ``language`` given to :meth:`STTRealtime.stream` survive an unrelated
        update. Connection-time options are retained at their current values and a
        warning is logged, since changing them would desynchronize the
        already-negotiated session. Every other change is queued as an in-band
        ``config.update`` sent before the next audio frame.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals.
            endpointing: ``vad`` for server-side turn detection, or ``manual``.
            sample_rate: Audio sample rate in Hz; retained on a live stream.
            prompt: Context or terminology hint; ``None`` clears it.
            return_timestamps: Segment-level timestamps; retained on a live stream.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Speech-onset padding; retained on a live stream.

        Raises:
            ValueError: If an option falls outside the values the endpoint accepts.
        """
        previous_opts = self._opts
        requested: dict[str, Any] = {}
        if is_given(language):
            requested["language"] = language
        if is_given(stream_type):
            requested["stream_type"] = stream_type
        if is_given(mode):
            requested["mode"] = mode
        if is_given(endpointing):
            requested["endpointing"] = endpointing
        if is_given(sample_rate):
            requested["sample_rate"] = sample_rate
        if is_given(prompt):
            requested["prompt"] = prompt
        if is_given(return_timestamps):
            requested["return_timestamps"] = return_timestamps
        if is_given(vad_sot_threshold):
            requested["vad_sot_threshold"] = vad_sot_threshold
        if is_given(vad_min_speech_ms):
            requested["vad_min_speech_ms"] = vad_min_speech_ms
        if is_given(vad_min_silence_ms):
            requested["vad_min_silence_ms"] = vad_min_silence_ms
        if is_given(vad_prefix_padding_ms):
            requested["vad_prefix_padding_ms"] = vad_prefix_padding_ms

        if not requested:
            return

        opts = replace(previous_opts, **requested)
        connection_only_options: list[str] = []
        if opts.sample_rate != previous_opts.sample_rate:
            connection_only_options.append("sample_rate")
            opts = replace(opts, sample_rate=previous_opts.sample_rate)
        if opts.return_timestamps != previous_opts.return_timestamps:
            connection_only_options.append("return_timestamps")
            opts = replace(opts, return_timestamps=previous_opts.return_timestamps)
        if opts.vad_prefix_padding_ms != previous_opts.vad_prefix_padding_ms:
            connection_only_options.append("vad_prefix_padding_ms")
            opts = replace(opts, vad_prefix_padding_ms=previous_opts.vad_prefix_padding_ms)
        if connection_only_options:
            self._logger.warning(
                "Sarvam realtime STT connection-only option updates only apply to new streams",
                extra={
                    **self._build_log_context(),
                    "options": connection_only_options,
                },
            )

        self._opts = opts
        if opts.endpointing != previous_opts.endpointing:
            if opts.endpointing == "manual" and not self._flush_observed:
                self._logger.warning(
                    "Sarvam realtime STT switched to manual endpointing without an external VAD; "
                    "turns will not be delimited unless the agent framework flushes the stream. "
                    "Configure a VAD on the AgentSession to receive end-of-turn boundaries.",
                    extra=self._build_log_context(),
                )
            self._pending_endpointing = opts.endpointing
            self._endpointing_update_acknowledged = False
            self._endpointing_update_sent = False

        update = self._config_update_payload(previous_opts, opts)
        if update is not None:
            if self._pending_config_update is None:
                self._pending_config_update = update
            else:
                self._pending_config_update.update(update)

    @staticmethod
    def _config_update_payload(
        previous: RealtimeSTTOptions,
        current: RealtimeSTTOptions,
    ) -> dict[str, Any] | None:
        payload: dict[str, Any] = {"event": "config.update"}
        values = (
            ("language_code", previous.language, current.language),
            ("stream_type", previous.stream_type, current.stream_type),
            ("mode", previous.mode, current.mode),
            ("prompt", previous.prompt, current.prompt),
            ("endpointing", previous.endpointing, current.endpointing),
            ("threshold", previous.vad_sot_threshold, current.vad_sot_threshold),
            (
                "min_speech_duration_ms",
                previous.vad_min_speech_ms,
                current.vad_min_speech_ms,
            ),
            (
                "silence_duration_ms",
                previous.vad_min_silence_ms,
                current.vad_min_silence_ms,
            ),
        )
        for key, old_value, new_value in values:
            if old_value != new_value:
                if key == "prompt" and new_value is None:
                    new_value = ""
                payload[key] = new_value
        return payload if len(payload) > 1 else None

    @staticmethod
    def _ack_lists_endpointing(data: dict[str, Any]) -> bool | None:
        """Whether a ``config.updated`` acknowledges an endpointing change.

        The server echoes each applied key as ``"<key>=<value>"``, optionally
        suffixed when the change is deferred to the next utterance boundary.
        Returns ``None`` when ``applied`` is missing or not a list of strings, so
        the caller can fall back instead of stalling on an unexpected shape.
        """
        applied = data.get("applied")
        if not isinstance(applied, list) or not all(isinstance(e, str) for e in applied):
            return None
        return any(e.split("=", 1)[0].strip() == "endpointing" for e in applied)

    def _handle_config_updated(self, data: dict[str, Any]) -> None:
        if self._pending_endpointing is None:
            return
        if not self._endpointing_update_sent:
            # This acknowledges an earlier update; ours is still queued locally and
            # the server is still in the old mode.
            return
        if self._ack_lists_endpointing(data) is False:
            return

        self._endpointing_update_acknowledged = True
        self._apply_pending_endpointing()

    def _apply_pending_endpointing(self) -> None:
        if (
            self._pending_endpointing is not None
            and self._endpointing_update_acknowledged
            and not self._utterance_in_progress
        ):
            self._active_endpointing = self._pending_endpointing
            self._pending_endpointing = None
            self._endpointing_update_acknowledged = False
            self._endpointing_update_sent = False

    def _complete_utterance(self) -> None:
        self._utterance_in_progress = False
        self._apply_pending_endpointing()

    def _build_log_context(self) -> dict[str, Any]:
        return {
            "request_id": self._request_id,
            "session_id": self._session_id,
            "model": self._opts.model,
            "language": self._opts.language,
            "stream_type": self._opts.stream_type,
            "endpointing": self._opts.endpointing,
            "utterance_idx": self._utterance_idx,
        }

    @staticmethod
    def _extract_request_id(data: dict[str, Any]) -> str | None:
        request_id = data.get("request_id")
        if request_id is None:
            nested = data.get("data")
            if isinstance(nested, dict):
                request_id = nested.get("request_id")
            metadata = data.get("metadata")
            if request_id is None and isinstance(metadata, dict):
                request_id = metadata.get("request_id")
        if isinstance(request_id, str) and request_id:
            return request_id
        return None

    @staticmethod
    def _extract_session_id(data: dict[str, Any]) -> str | None:
        session_id = data.get("session_id")
        if isinstance(session_id, str) and session_id:
            return session_id
        return None

    def _capture_server_ids(self, data: dict[str, Any]) -> None:
        session_id = self._extract_session_id(data)
        if session_id is not None:
            self._session_id = session_id

        if not self._request_id:
            request_id = self._extract_request_id(data)
            if request_id is not None:
                self._request_id = request_id

    async def aclose(self) -> None:
        """Close the connection, reporting any audio duration not yet billed.

        Agents normally end a stream here rather than waiting for ``session.end``, so
        the pending duration is flushed before the base class cancels the tasks that
        deliver usage metrics.
        """
        try:
            if not self._event_ch.closed:
                self._emit_local_usage_fallback()
                # Give the metrics monitor a chance to consume the usage event before
                # super().aclose() cancels it.
                await asyncio.sleep(0)
            if self._ws and not self._ws.closed:
                await self._ws.close()
        finally:
            self._ws = None
            await super().aclose()

    async def _run(self) -> None:
        # A single connection attempt: this endpoint bills per connection, so the
        # stream never reconnects on its own (`stream()` also forces max_retry=0).
        ws: aiohttp.ClientWebSocketResponse | None = None
        try:
            ws = await self._connect_ws()
            self._ws = ws
            tasks = [
                asyncio.create_task(self._process_audio(ws)),
                asyncio.create_task(self._process_messages(ws)),
            ]
            try:
                await asyncio.gather(*tasks)
            finally:
                await utils.aio.gracefully_cancel(*tasks)
        except asyncio.TimeoutError as e:
            raise APITimeoutError("Timed out connecting to Sarvam realtime STT") from e
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(
                message=e.message,
                status_code=e.status,
                request_id=self._request_id or None,
                body=e.message,
            ) from e
        except aiohttp.ClientConnectorError as e:
            raise APIConnectionError("failed to connect to Sarvam realtime STT") from e
        finally:
            if ws is not None:
                await ws.close()
            self._ws = None

    def _reset_utterance_state(self) -> None:
        self._utterance_idx = None
        self._pending_final_data = None
        self._utterance_start_audio_pos = self._audio_position
        self._utterance_speech_end_audio_pos = None
        self._utterance_speech_end_wall = None
        self._final_received_for_utterance = False
        self._eos_emitted_for_utterance = False

    def _begin_manual_utterance(self) -> None:
        """Open a client-delimited turn.

        Sarvam emits no ``vad.speech_start`` under manual endpointing, so the client
        boundary is what starts an utterance. Resetting here keeps the per-utterance
        flags and timings from leaking across turns, including after an ``endpointing``
        switch from ``vad`` to ``manual``.
        """
        self._reset_utterance_state()
        self._utterance_in_progress = True
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.START_OF_SPEECH,
                request_id=self._request_id,
            )
        )

    def _end_manual_utterance(self) -> None:
        """Close a client-delimited turn and anchor its speech-end position."""
        self._utterance_speech_end_audio_pos = self._audio_position
        self._utterance_speech_end_wall = time.time()
        self._emit_end_of_speech()
        self._complete_utterance()

    async def _safe_send_str(
        self,
        ws: Any,
        payload: dict[str, Any],
    ) -> bool:
        """Send a JSON control message, tolerating a peer that already closed.

        Returns:
            Whether the payload reached the socket.
        """
        if ws.closed:
            return False

        try:
            await ws.send_str(json.dumps(payload))
        except (aiohttp.ClientConnectionResetError, ConnectionError):
            self._logger.debug(
                "Sarvam realtime STT WebSocket closed before send completed",
                extra={**self._build_log_context(), "payload": payload},
            )
            return False
        return True

    async def _safe_send_bytes(self, ws: Any, payload: bytes) -> None:
        if ws.closed:
            return

        try:
            await ws.send_bytes(payload)
        except (aiohttp.ClientConnectionResetError, ConnectionError):
            self._logger.debug(
                "Sarvam realtime STT WebSocket closed before audio send completed",
                extra={**self._build_log_context(), "payload_bytes": len(payload)},
            )

    async def _send_pending_config_update(self, ws: Any) -> None:
        payload = self._pending_config_update
        self._pending_config_update = None
        if payload is None:
            return
        if await self._safe_send_str(ws, payload) and "endpointing" in payload:
            # Only now can a config.updated acknowledgement refer to our change.
            self._endpointing_update_sent = True

    async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
        ws_url = _build_realtime_ws_url(self._opts.base_url, self._opts)
        headers = {
            "API-SUBSCRIPTION-KEY": self._opts.api_key,
            "User-Agent": USER_AGENT,
        }
        self._logger.debug(
            "Connecting to Sarvam realtime STT WebSocket", extra=self._build_log_context()
        )
        try:
            ws = await asyncio.wait_for(
                self._session.ws_connect(ws_url, headers=headers, heartbeat=30.0),
                self._conn_options.timeout,
            )
        except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e:
            self._logger.error(
                "Failed to connect to Sarvam realtime STT WebSocket",
                extra={**self._build_log_context(), "error": str(e), "url": ws_url},
                exc_info=True,
            )
            raise
        except aiohttp.ClientResponseError as e:
            self._logger.error(
                "Sarvam realtime STT WebSocket handshake failed",
                extra={
                    **self._build_log_context(),
                    "error": e.message,
                    "status_code": e.status,
                    "url": ws_url,
                },
                exc_info=True,
            )
            raise
        except Exception as e:
            self._logger.error(
                "Unexpected Sarvam realtime STT WebSocket connection error",
                extra={**self._build_log_context(), "error": str(e), "url": ws_url},
                exc_info=True,
            )
            raise APIConnectionError("failed to connect to Sarvam realtime STT") from e
        self._logger.debug(
            "Sarvam realtime STT WebSocket connected", extra=self._build_log_context()
        )
        return ws

    @utils.log_exceptions(logger=logger)
    async def _process_audio(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        samples_per_channel = max(int(self._opts.sample_rate * AUDIO_CHUNK_MS / 1000), 1)
        audio_bstream = utils.audio.AudioByteStream(
            sample_rate=self._opts.sample_rate,
            num_channels=1,
            samples_per_channel=samples_per_channel,
        )

        async for data in self._input_ch:
            # The server is done reading, so stop the pump instead of writing into a
            # socket whose reset would fail the whole stream (max_retry is forced to 0).
            if self._session_ended or ws.closed:
                break

            await self._send_pending_config_update(ws)
            frames: list[rtc.AudioFrame] = []
            if isinstance(data, rtc.AudioFrame):
                frames.extend(audio_bstream.write(data.data.tobytes()))
            if isinstance(data, self._FlushSentinel):
                self._flush_observed = True
                frames.extend(audio_bstream.flush())

            for frame in frames:
                if self._active_endpointing == "manual" and not self._manual_speech_started:
                    await self._safe_send_str(ws, {"event": "speech_start"})
                    self._manual_speech_started = True
                    self._begin_manual_utterance()

                self._audio_duration_collector.push(frame.duration)
                self._audio_position += frame.duration
                await self._safe_send_bytes(
                    ws, _encode_pcm_for_wire(self._opts.encoding, frame.data.tobytes())
                )

            if isinstance(data, self._FlushSentinel):
                self._audio_duration_collector.flush()
                if self._active_endpointing == "manual" and self._manual_speech_started:
                    await self._safe_send_str(ws, {"event": "speech_end"})
                    self._manual_speech_started = False
                    self._end_manual_utterance()

        self._emit_local_usage_fallback()
        if not self._session_ended:
            await self._safe_send_str(ws, {"event": "end"})

    @utils.log_exceptions(logger=logger)
    async def _process_messages(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        while True:
            msg = await ws.receive()
            if msg.type == aiohttp.WSMsgType.TEXT:
                try:
                    await self._handle_message(json.loads(msg.data))
                except json.JSONDecodeError as e:
                    if _looks_like_error_text(msg.data):
                        self._logger.error(
                            "Sarvam realtime STT non-JSON error message",
                            extra={**self._build_log_context(), "raw_message": msg.data},
                        )
                        raise APIStatusError(
                            message=f"Sarvam realtime STT non-JSON error message: {msg.data}",
                            request_id=self._request_id or None,
                            body={"raw_message": msg.data},
                        ) from e
                    self._logger.warning(
                        "Invalid JSON received from Sarvam realtime STT",
                        extra={**self._build_log_context(), "raw_data": msg.data},
                    )
                    continue
                if self._session_ended:
                    break
            elif msg.type == aiohttp.WSMsgType.ERROR:
                self._logger.error(
                    "Sarvam realtime STT WebSocket error",
                    extra={**self._build_log_context(), "raw_message": msg.data},
                )
                raise APIConnectionError(f"Sarvam realtime STT WebSocket error: {msg.data}")
            elif msg.type in (
                aiohttp.WSMsgType.CLOSED,
                aiohttp.WSMsgType.CLOSE,
                aiohttp.WSMsgType.CLOSING,
            ):
                close_code = ws.close_code if ws.close_code is not None else msg.data
                close_reason = msg.extra
                if self._session_ended and close_code in (1000, 1001, None):
                    self._flush_terminal_utterance()
                    self._emit_local_usage_fallback()
                    break
                if close_code in (1000, 1001, None) and not _looks_like_error_text(close_reason):
                    self._flush_terminal_utterance()
                    self._emit_local_usage_fallback()
                    break
                self._logger.error(
                    "Sarvam realtime STT WebSocket closed unexpectedly",
                    extra={
                        **self._build_log_context(),
                        "close_code": close_code,
                        "close_reason": close_reason,
                    },
                )
                raise self._status_error_from_close(close_code, close_reason)
            else:
                self._logger.debug(
                    "Unknown Sarvam realtime STT WebSocket message type",
                    extra={**self._build_log_context(), "message_type": str(msg.type)},
                )

    def _status_error_from_close(self, close_code: object, close_reason: object) -> APIStatusError:
        status_code = int(close_code) if isinstance(close_code, int) else -1
        retryable = close_code == 1013
        message = f"Sarvam realtime STT WebSocket closed unexpectedly: {close_reason}"
        if close_code == 1003:
            message = "Sarvam realtime STT authentication, quota, or rate limit error"
        elif close_code == 1008:
            message = "Sarvam realtime STT session timed out or exceeded the maximum duration"
        elif close_code == 1013:
            message = "Sarvam realtime STT backend temporarily unavailable"
        elif close_code == 4000:
            message = f"Sarvam realtime STT rejected the session: {close_reason}"

        return APIStatusError(
            message=message,
            status_code=status_code,
            request_id=self._request_id or None,
            body={
                "close_code": close_code,
                "close_reason": close_reason,
            },
            retryable=retryable,
        )

    async def _handle_message(self, data: dict[str, Any]) -> None:
        event = data.get("event")
        self._capture_server_ids(data)
        if event == "session.begin":
            config = data.get("config")
            self._resolved_config = dict(config) if isinstance(config, dict) else None
        self._log_stt_event(event, data)
        if event == "session.begin":
            return
        elif event == "vad.speech_start":
            self._reset_utterance_state()
            self._utterance_in_progress = True
            utterance_idx = data.get("utterance_idx")
            self._utterance_idx = utterance_idx if isinstance(utterance_idx, int) else None
            self._event_ch.send_nowait(
                stt.SpeechEvent(
                    type=stt.SpeechEventType.START_OF_SPEECH,
                    request_id=self._request_id,
                )
            )
        elif event == "vad.speech_end":
            self._handle_speech_end()
        elif event == "transcript.partial":
            self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data)
        elif event == "transcript.final":
            if self._active_endpointing == "vad":
                if self._is_valid_transcript(data):
                    self._pending_final_data = data
                    self._final_received_for_utterance = True
                    self._try_commit_utterance()
            elif self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data):
                self._final_received_for_utterance = True
                self._complete_utterance()
        elif event == "session.end":
            self._handle_session_end(data)
        elif event == "config.updated":
            self._handle_config_updated(data)
            return
        elif event == "error":
            self._handle_error_event(data)
        elif event == "pong":
            return
        else:
            self._logger.debug(
                "Unknown Sarvam realtime STT event",
                extra={**self._build_log_context(), "event": event, "data": data},
            )

    def _log_stt_event(self, event: object, data: dict[str, Any]) -> None:
        if event == "pong":
            return

        extra: dict[str, Any] = {
            **self._build_log_context(),
            "event": event,
            "utterance_idx": data.get("utterance_idx"),
        }
        if event in {"transcript.partial", "transcript.final"}:
            # Recognized speech is personal data, so only its length is safe for the
            # INFO record; the text itself stays in the opt-in DEBUG raw payload.
            text = data.get("text")
            if isinstance(text, str):
                extra["text_length"] = len(text)
            extra["language"] = data.get("language") or self._opts.language
            extra["confidence"] = data.get("language_confidence", data.get("confidence"))
        elif event == "vad.speech_start":
            extra["audio_position"] = self._audio_position
        elif event == "vad.speech_end":
            extra["audio_position"] = self._audio_position
        elif event == "session.begin":
            pass
        elif event == "session.end":
            extra["audio_duration_s"] = data.get("audio_duration_s")
        elif event == "config.updated":
            extra["applied"] = data.get("applied")
        elif event == "error":
            extra["error_code"] = data.get("code")
            extra["error_message"] = data.get("message")
            extra["status_code"] = data.get("status_code")
        else:
            return

        if event == "transcript.partial":
            self._logger.debug(
                "Sarvam realtime STT transcript.partial",
                extra={**extra, "raw_data": data},
            )
            return

        self._logger.info(f"Sarvam realtime STT {event}", extra=extra)
        self._logger.debug(
            "Sarvam realtime STT raw event",
            extra={**extra, "raw_data": data},
        )

    def _is_valid_transcript(self, data: dict[str, Any]) -> bool:
        text = data.get("text")
        # Whitespace carries no content, and emitting it would commit a user turn
        # with no words.
        return isinstance(text, str) and bool(text.strip())

    def _handle_speech_end(self) -> None:
        self._utterance_speech_end_audio_pos = self._audio_position
        self._utterance_speech_end_wall = time.time()

        if self._active_endpointing != "vad":
            self._emit_end_of_speech()
        elif not self._eos_emitted_for_utterance:
            self._emit_end_of_speech()
            if self._final_received_for_utterance:
                self._try_commit_utterance()

        # The server's speech end is the utterance boundary, so the turn is over even
        # when the final is empty or never arrives. Completing unconditionally is what
        # lets a boundary-gated endpointing change promote; leaving the utterance open
        # would strand the stream in the old mode with the server in the new one.
        self._complete_utterance()

    def _try_commit_utterance(self) -> None:
        if self._pending_final_data is None or self._utterance_speech_end_audio_pos is None:
            return

        committed_data = self._pending_final_data
        if self._send_transcript_event(
            stt.SpeechEventType.FINAL_TRANSCRIPT,
            committed_data,
        ):
            self._logger.debug(
                "Sarvam realtime STT utterance committed",
                extra={
                    **self._build_log_context(),
                    "end_time": self._utterance_speech_end_audio_pos,
                    "speech_end_wall_time": self._utterance_speech_end_wall,
                },
            )
            if not self._eos_emitted_for_utterance:
                self._emit_end_of_speech()
            self._pending_final_data = None
            self._complete_utterance()

    def _flush_terminal_utterance(self) -> None:
        """Commit a buffered final transcript when the session ends mid-utterance.

        In VAD endpointing a ``transcript.final`` is held until ``vad.speech_end``
        supplies the speech-end position. When the input audio ends mid-utterance
        the server finalizes and closes without that event, so the speech-end
        position is anchored to the audio consumed so far instead of dropping the
        transcript. Safe to call more than once per session.
        """
        if self._pending_final_data is not None and self._utterance_speech_end_audio_pos is None:
            self._utterance_speech_end_audio_pos = self._audio_position
            if self._utterance_speech_end_wall is None:
                self._utterance_speech_end_wall = time.time()

        if not self._eos_emitted_for_utterance and self._pending_final_data is not None:
            self._emit_end_of_speech()

        self._try_commit_utterance()

    def _emit_end_of_speech(self) -> None:
        if self._eos_emitted_for_utterance:
            return

        # Emitted without alternatives so the agent pipeline treats it as a sentinel
        # it can hold and release with a concrete transcript. The speech-end timing
        # travels on the FINAL_TRANSCRIPT event instead.
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.END_OF_SPEECH,
                request_id=self._request_id,
            )
        )
        self._eos_emitted_for_utterance = True

    def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict[str, Any]) -> bool:
        text = data.get("text")
        if not isinstance(text, str) or not text.strip():
            return False

        language = data.get("language") or self._opts.language
        # Recognition confidence only: `language_confidence` is a language-identification
        # score and stays in metadata. The endpoint sends no per-segment confidence
        # today, and an absent value falls back to 1.0 (as `_extract_confidence` in
        # stt.py does) so it isn't averaged downstream as "no confidence".
        # bool is a subclass of int, so exclude it explicitly.
        confidence = data.get("confidence")
        if not isinstance(confidence, (int, float)) or isinstance(confidence, bool):
            confidence = 1.0

        metadata: dict[str, Any] = {
            key: data[key]
            for key in ("utterance_idx", "language_confidence")
            if key in data and data[key] is not None
        }
        if (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._utterance_speech_end_wall is not None
        ):
            metadata["speech_end_wall_time"] = self._utterance_speech_end_wall
        end_time = 0.0
        start_time = 0.0
        if event_type == stt.SpeechEventType.FINAL_TRANSCRIPT:
            start_s = data.get("start_s")
            end_s = data.get("end_s")
            if isinstance(start_s, (int, float)) and not isinstance(start_s, bool):
                start_time = max(float(start_s), 0.0)
            if isinstance(end_s, (int, float)) and not isinstance(end_s, bool):
                end_time = max(float(end_s), 0.0)
        if (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._utterance_speech_end_audio_pos is not None
            and end_time == 0.0
        ):
            end_time = self._utterance_speech_end_audio_pos
        elif (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._audio_position > 0
            and end_time == 0.0
        ):
            end_time = self._audio_position

        speech_data = stt.SpeechData(
            language=LanguageCode(language),
            text=text,
            start_time=start_time,
            end_time=end_time,
            confidence=float(confidence),
            metadata=metadata or None,
        )
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=event_type,
                request_id=self._request_id,
                alternatives=[speech_data],
            )
        )
        return True

    def _handle_session_end(self, data: dict[str, Any]) -> None:
        self._capture_server_ids(data)
        self._flush_terminal_utterance()
        audio_duration = data.get("audio_duration_s")
        if (
            isinstance(audio_duration, (int, float))
            and not isinstance(audio_duration, bool)
            and not self._server_audio_duration_reported
        ):
            # Report whatever audio is still buffered locally, then top up to Sarvam's
            # authoritative total so the session bills exactly once for it.
            self._audio_duration_collector.flush()
            server_audio_duration = max(float(audio_duration), 0.0)
            delta = max(server_audio_duration - self._total_reported_audio_duration, 0.0)
            if delta:
                self._emit_usage(delta)
            self._server_audio_duration_reported = True
        else:
            self._emit_local_usage_fallback()
        self._session_ended = True

    def _handle_error_event(self, data: dict[str, Any]) -> None:
        if not data.get("is_fatal", False):
            self._logger.warning(
                "Non-fatal Sarvam realtime STT error",
                extra={
                    **self._build_log_context(),
                    "error_code": data.get("code"),
                    "error_message": data.get("message"),
                    "status_code": data.get("status_code"),
                    "raw_message": data,
                },
            )
            return

        code = data.get("code", "unknown")
        status_code = data.get("status_code", -1)
        if not isinstance(status_code, int):
            status_code = -1
        self._logger.error(
            "Fatal Sarvam realtime STT error",
            extra={
                **self._build_log_context(),
                "error_code": code,
                "error_message": data.get("message", code),
                "status_code": status_code,
                "raw_message": data,
            },
        )
        raise APIStatusError(
            message=f"Sarvam realtime STT error: {data.get('message', code)}",
            status_code=status_code,
            request_id=self._request_id or None,
            body=data,
            retryable=code == "model_unavailable",
        )

    def _on_audio_duration_report(self, duration: float) -> None:
        self._local_audio_duration += duration
        self._emit_usage(duration)

    def _emit_local_usage_fallback(self) -> None:
        if self._server_audio_duration_reported:
            return
        self._audio_duration_collector.flush()

    def _emit_usage(self, duration: float) -> None:
        self._total_reported_audio_duration += duration
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.RECOGNITION_USAGE,
                request_id=self._request_id,
                recognition_usage=stt.RecognitionUsage(audio_duration=duration),
            )
        )

A single WebSocket session against Sarvam's realtime STT endpoint.

Audio pushed into the stream is forwarded in the configured wire encoding, and the events Sarvam returns are translated into LiveKit speech events. Audio duration is reported incrementally while the session runs and reconciled against the server's authoritative total when the session ends.

Create a realtime speech stream.

Args

stt

The parent instance that created this stream.

opts

Resolved options for this connection.

conn_options

Connection options for this stream.

http_session

aiohttp session used to open the WebSocket.

Ancestors

• livekit.agents.stt.stt.RecognizeStream
• abc.ABC

Instance variables

prop resolved_config : dict[str, Any] | None

Expand source code

@property
def resolved_config(self) -> dict[str, Any] | None:
    """Return the configuration resolved by Sarvam for this connection."""
    return dict(self._resolved_config) if self._resolved_config is not None else None

Return the configuration resolved by Sarvam for this connection.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close the connection, reporting any audio duration not yet billed.

    Agents normally end a stream here rather than waiting for ``session.end``, so
    the pending duration is flushed before the base class cancels the tasks that
    deliver usage metrics.
    """
    try:
        if not self._event_ch.closed:
            self._emit_local_usage_fallback()
            # Give the metrics monitor a chance to consume the usage event before
            # super().aclose() cancels it.
            await asyncio.sleep(0)
        if self._ws and not self._ws.closed:
            await self._ws.close()
    finally:
        self._ws = None
        await super().aclose()

Close the connection, reporting any audio duration not yet billed.

Agents normally end a stream here rather than waiting for session.end, so the pending duration is flushed before the base class cancels the tasks that deliver usage metrics.

def update_options(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str | None] = NOT_GIVEN,
return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
    mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
    endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str | None] = NOT_GIVEN,
    return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
    vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
) -> None:
    """Apply an option change to this live connection.

    Only the options explicitly passed here are changed, so per-stream overrides
    such as a ``language`` given to :meth:`STTRealtime.stream` survive an unrelated
    update. Connection-time options are retained at their current values and a
    warning is logged, since changing them would desynchronize the
    already-negotiated session. Every other change is queued as an in-band
    ``config.update`` sent before the next audio frame.

    Args:
        language: BCP-47 language code, or ``auto`` for adaptive identification.
        stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
        mode: Task applied to finals.
        endpointing: ``vad`` for server-side turn detection, or ``manual``.
        sample_rate: Audio sample rate in Hz; retained on a live stream.
        prompt: Context or terminology hint; ``None`` clears it.
        return_timestamps: Segment-level timestamps; retained on a live stream.
        vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
        vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
        vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
        vad_prefix_padding_ms: Speech-onset padding; retained on a live stream.

    Raises:
        ValueError: If an option falls outside the values the endpoint accepts.
    """
    previous_opts = self._opts
    requested: dict[str, Any] = {}
    if is_given(language):
        requested["language"] = language
    if is_given(stream_type):
        requested["stream_type"] = stream_type
    if is_given(mode):
        requested["mode"] = mode
    if is_given(endpointing):
        requested["endpointing"] = endpointing
    if is_given(sample_rate):
        requested["sample_rate"] = sample_rate
    if is_given(prompt):
        requested["prompt"] = prompt
    if is_given(return_timestamps):
        requested["return_timestamps"] = return_timestamps
    if is_given(vad_sot_threshold):
        requested["vad_sot_threshold"] = vad_sot_threshold
    if is_given(vad_min_speech_ms):
        requested["vad_min_speech_ms"] = vad_min_speech_ms
    if is_given(vad_min_silence_ms):
        requested["vad_min_silence_ms"] = vad_min_silence_ms
    if is_given(vad_prefix_padding_ms):
        requested["vad_prefix_padding_ms"] = vad_prefix_padding_ms

    if not requested:
        return

    opts = replace(previous_opts, **requested)
    connection_only_options: list[str] = []
    if opts.sample_rate != previous_opts.sample_rate:
        connection_only_options.append("sample_rate")
        opts = replace(opts, sample_rate=previous_opts.sample_rate)
    if opts.return_timestamps != previous_opts.return_timestamps:
        connection_only_options.append("return_timestamps")
        opts = replace(opts, return_timestamps=previous_opts.return_timestamps)
    if opts.vad_prefix_padding_ms != previous_opts.vad_prefix_padding_ms:
        connection_only_options.append("vad_prefix_padding_ms")
        opts = replace(opts, vad_prefix_padding_ms=previous_opts.vad_prefix_padding_ms)
    if connection_only_options:
        self._logger.warning(
            "Sarvam realtime STT connection-only option updates only apply to new streams",
            extra={
                **self._build_log_context(),
                "options": connection_only_options,
            },
        )

    self._opts = opts
    if opts.endpointing != previous_opts.endpointing:
        if opts.endpointing == "manual" and not self._flush_observed:
            self._logger.warning(
                "Sarvam realtime STT switched to manual endpointing without an external VAD; "
                "turns will not be delimited unless the agent framework flushes the stream. "
                "Configure a VAD on the AgentSession to receive end-of-turn boundaries.",
                extra=self._build_log_context(),
            )
        self._pending_endpointing = opts.endpointing
        self._endpointing_update_acknowledged = False
        self._endpointing_update_sent = False

    update = self._config_update_payload(previous_opts, opts)
    if update is not None:
        if self._pending_config_update is None:
            self._pending_config_update = update
        else:
            self._pending_config_update.update(update)

Apply an option change to this live connection.

Only the options explicitly passed here are changed, so per-stream overrides such as a language given to :meth:STTRealtime.stream() survive an unrelated update. Connection-time options are retained at their current values and a warning is logged, since changing them would desynchronize the already-negotiated session. Every other change is queued as an in-band config.update sent before the next audio frame.

Args

language

BCP-47 language code, or auto for adaptive identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals.

endpointing

vad for server-side turn detection, or manual.

sample_rate

Audio sample rate in Hz; retained on a live stream.

prompt

Context or terminology hint; None clears it.

return_timestamps

Segment-level timestamps; retained on a live stream.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Speech-onset padding; retained on a live stream.

Raises

ValueError

If an option falls outside the values the endpoint accepts.

class StreamingSpeechStream (*,
stt: STTRealtime,
opts: RealtimeSTTOptions,
conn_options: APIConnectOptions,
http_session: aiohttp.ClientSession)

Expand source code

class RealtimeSpeechStream(stt.SpeechStream):
    """A single WebSocket session against Sarvam's realtime STT endpoint.

    Audio pushed into the stream is forwarded in the configured wire encoding, and
    the events Sarvam returns are translated into LiveKit speech events. Audio
    duration is reported incrementally while the session runs and reconciled
    against the server's authoritative total when the session ends.
    """

    def __init__(
        self,
        *,
        stt: STTRealtime,
        opts: RealtimeSTTOptions,
        conn_options: APIConnectOptions,
        http_session: aiohttp.ClientSession,
    ) -> None:
        """Create a realtime speech stream.

        Args:
            stt: The parent instance that created this stream.
            opts: Resolved options for this connection.
            conn_options: Connection options for this stream.
            http_session: aiohttp session used to open the WebSocket.
        """
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
        self._opts = opts
        self._session = http_session
        self._ws: aiohttp.ClientWebSocketResponse | None = None
        self._request_id = ""
        self._session_id = ""
        self._resolved_config: dict[str, Any] | None = None
        self._session_ended = False
        self._utterance_idx: int | None = None
        self._utterance_in_progress = False
        self._active_endpointing = opts.endpointing
        self._pending_endpointing: RealtimeEndpointing | str | None = None
        self._endpointing_update_acknowledged = False
        self._endpointing_update_sent = False
        self._pending_config_update: dict[str, Any] | None = None
        self._manual_speech_started = False
        self._flush_observed = False
        self._pending_final_data: dict[str, Any] | None = None
        self._utterance_start_audio_pos = 0.0
        self._utterance_speech_end_audio_pos: float | None = None
        self._utterance_speech_end_wall: float | None = None
        self._final_received_for_utterance = False
        self._eos_emitted_for_utterance = False
        self._stream_started_at = time.time()
        self._audio_position = 0.0
        self._local_audio_duration = 0.0
        self._total_reported_audio_duration = 0.0
        self._server_audio_duration_reported = False
        self._audio_duration_collector = PeriodicCollector(
            callback=self._on_audio_duration_report,
            duration=5.0,
        )
        self._logger = logger

    @property
    def resolved_config(self) -> dict[str, Any] | None:
        """Return the configuration resolved by Sarvam for this connection."""
        return dict(self._resolved_config) if self._resolved_config is not None else None

    def update_options(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
        mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
        endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str | None] = NOT_GIVEN,
        return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
        vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
    ) -> None:
        """Apply an option change to this live connection.

        Only the options explicitly passed here are changed, so per-stream overrides
        such as a ``language`` given to :meth:`STTRealtime.stream` survive an unrelated
        update. Connection-time options are retained at their current values and a
        warning is logged, since changing them would desynchronize the
        already-negotiated session. Every other change is queued as an in-band
        ``config.update`` sent before the next audio frame.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals.
            endpointing: ``vad`` for server-side turn detection, or ``manual``.
            sample_rate: Audio sample rate in Hz; retained on a live stream.
            prompt: Context or terminology hint; ``None`` clears it.
            return_timestamps: Segment-level timestamps; retained on a live stream.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Speech-onset padding; retained on a live stream.

        Raises:
            ValueError: If an option falls outside the values the endpoint accepts.
        """
        previous_opts = self._opts
        requested: dict[str, Any] = {}
        if is_given(language):
            requested["language"] = language
        if is_given(stream_type):
            requested["stream_type"] = stream_type
        if is_given(mode):
            requested["mode"] = mode
        if is_given(endpointing):
            requested["endpointing"] = endpointing
        if is_given(sample_rate):
            requested["sample_rate"] = sample_rate
        if is_given(prompt):
            requested["prompt"] = prompt
        if is_given(return_timestamps):
            requested["return_timestamps"] = return_timestamps
        if is_given(vad_sot_threshold):
            requested["vad_sot_threshold"] = vad_sot_threshold
        if is_given(vad_min_speech_ms):
            requested["vad_min_speech_ms"] = vad_min_speech_ms
        if is_given(vad_min_silence_ms):
            requested["vad_min_silence_ms"] = vad_min_silence_ms
        if is_given(vad_prefix_padding_ms):
            requested["vad_prefix_padding_ms"] = vad_prefix_padding_ms

        if not requested:
            return

        opts = replace(previous_opts, **requested)
        connection_only_options: list[str] = []
        if opts.sample_rate != previous_opts.sample_rate:
            connection_only_options.append("sample_rate")
            opts = replace(opts, sample_rate=previous_opts.sample_rate)
        if opts.return_timestamps != previous_opts.return_timestamps:
            connection_only_options.append("return_timestamps")
            opts = replace(opts, return_timestamps=previous_opts.return_timestamps)
        if opts.vad_prefix_padding_ms != previous_opts.vad_prefix_padding_ms:
            connection_only_options.append("vad_prefix_padding_ms")
            opts = replace(opts, vad_prefix_padding_ms=previous_opts.vad_prefix_padding_ms)
        if connection_only_options:
            self._logger.warning(
                "Sarvam realtime STT connection-only option updates only apply to new streams",
                extra={
                    **self._build_log_context(),
                    "options": connection_only_options,
                },
            )

        self._opts = opts
        if opts.endpointing != previous_opts.endpointing:
            if opts.endpointing == "manual" and not self._flush_observed:
                self._logger.warning(
                    "Sarvam realtime STT switched to manual endpointing without an external VAD; "
                    "turns will not be delimited unless the agent framework flushes the stream. "
                    "Configure a VAD on the AgentSession to receive end-of-turn boundaries.",
                    extra=self._build_log_context(),
                )
            self._pending_endpointing = opts.endpointing
            self._endpointing_update_acknowledged = False
            self._endpointing_update_sent = False

        update = self._config_update_payload(previous_opts, opts)
        if update is not None:
            if self._pending_config_update is None:
                self._pending_config_update = update
            else:
                self._pending_config_update.update(update)

    @staticmethod
    def _config_update_payload(
        previous: RealtimeSTTOptions,
        current: RealtimeSTTOptions,
    ) -> dict[str, Any] | None:
        payload: dict[str, Any] = {"event": "config.update"}
        values = (
            ("language_code", previous.language, current.language),
            ("stream_type", previous.stream_type, current.stream_type),
            ("mode", previous.mode, current.mode),
            ("prompt", previous.prompt, current.prompt),
            ("endpointing", previous.endpointing, current.endpointing),
            ("threshold", previous.vad_sot_threshold, current.vad_sot_threshold),
            (
                "min_speech_duration_ms",
                previous.vad_min_speech_ms,
                current.vad_min_speech_ms,
            ),
            (
                "silence_duration_ms",
                previous.vad_min_silence_ms,
                current.vad_min_silence_ms,
            ),
        )
        for key, old_value, new_value in values:
            if old_value != new_value:
                if key == "prompt" and new_value is None:
                    new_value = ""
                payload[key] = new_value
        return payload if len(payload) > 1 else None

    @staticmethod
    def _ack_lists_endpointing(data: dict[str, Any]) -> bool | None:
        """Whether a ``config.updated`` acknowledges an endpointing change.

        The server echoes each applied key as ``"<key>=<value>"``, optionally
        suffixed when the change is deferred to the next utterance boundary.
        Returns ``None`` when ``applied`` is missing or not a list of strings, so
        the caller can fall back instead of stalling on an unexpected shape.
        """
        applied = data.get("applied")
        if not isinstance(applied, list) or not all(isinstance(e, str) for e in applied):
            return None
        return any(e.split("=", 1)[0].strip() == "endpointing" for e in applied)

    def _handle_config_updated(self, data: dict[str, Any]) -> None:
        if self._pending_endpointing is None:
            return
        if not self._endpointing_update_sent:
            # This acknowledges an earlier update; ours is still queued locally and
            # the server is still in the old mode.
            return
        if self._ack_lists_endpointing(data) is False:
            return

        self._endpointing_update_acknowledged = True
        self._apply_pending_endpointing()

    def _apply_pending_endpointing(self) -> None:
        if (
            self._pending_endpointing is not None
            and self._endpointing_update_acknowledged
            and not self._utterance_in_progress
        ):
            self._active_endpointing = self._pending_endpointing
            self._pending_endpointing = None
            self._endpointing_update_acknowledged = False
            self._endpointing_update_sent = False

    def _complete_utterance(self) -> None:
        self._utterance_in_progress = False
        self._apply_pending_endpointing()

    def _build_log_context(self) -> dict[str, Any]:
        return {
            "request_id": self._request_id,
            "session_id": self._session_id,
            "model": self._opts.model,
            "language": self._opts.language,
            "stream_type": self._opts.stream_type,
            "endpointing": self._opts.endpointing,
            "utterance_idx": self._utterance_idx,
        }

    @staticmethod
    def _extract_request_id(data: dict[str, Any]) -> str | None:
        request_id = data.get("request_id")
        if request_id is None:
            nested = data.get("data")
            if isinstance(nested, dict):
                request_id = nested.get("request_id")
            metadata = data.get("metadata")
            if request_id is None and isinstance(metadata, dict):
                request_id = metadata.get("request_id")
        if isinstance(request_id, str) and request_id:
            return request_id
        return None

    @staticmethod
    def _extract_session_id(data: dict[str, Any]) -> str | None:
        session_id = data.get("session_id")
        if isinstance(session_id, str) and session_id:
            return session_id
        return None

    def _capture_server_ids(self, data: dict[str, Any]) -> None:
        session_id = self._extract_session_id(data)
        if session_id is not None:
            self._session_id = session_id

        if not self._request_id:
            request_id = self._extract_request_id(data)
            if request_id is not None:
                self._request_id = request_id

    async def aclose(self) -> None:
        """Close the connection, reporting any audio duration not yet billed.

        Agents normally end a stream here rather than waiting for ``session.end``, so
        the pending duration is flushed before the base class cancels the tasks that
        deliver usage metrics.
        """
        try:
            if not self._event_ch.closed:
                self._emit_local_usage_fallback()
                # Give the metrics monitor a chance to consume the usage event before
                # super().aclose() cancels it.
                await asyncio.sleep(0)
            if self._ws and not self._ws.closed:
                await self._ws.close()
        finally:
            self._ws = None
            await super().aclose()

    async def _run(self) -> None:
        # A single connection attempt: this endpoint bills per connection, so the
        # stream never reconnects on its own (`stream()` also forces max_retry=0).
        ws: aiohttp.ClientWebSocketResponse | None = None
        try:
            ws = await self._connect_ws()
            self._ws = ws
            tasks = [
                asyncio.create_task(self._process_audio(ws)),
                asyncio.create_task(self._process_messages(ws)),
            ]
            try:
                await asyncio.gather(*tasks)
            finally:
                await utils.aio.gracefully_cancel(*tasks)
        except asyncio.TimeoutError as e:
            raise APITimeoutError("Timed out connecting to Sarvam realtime STT") from e
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(
                message=e.message,
                status_code=e.status,
                request_id=self._request_id or None,
                body=e.message,
            ) from e
        except aiohttp.ClientConnectorError as e:
            raise APIConnectionError("failed to connect to Sarvam realtime STT") from e
        finally:
            if ws is not None:
                await ws.close()
            self._ws = None

    def _reset_utterance_state(self) -> None:
        self._utterance_idx = None
        self._pending_final_data = None
        self._utterance_start_audio_pos = self._audio_position
        self._utterance_speech_end_audio_pos = None
        self._utterance_speech_end_wall = None
        self._final_received_for_utterance = False
        self._eos_emitted_for_utterance = False

    def _begin_manual_utterance(self) -> None:
        """Open a client-delimited turn.

        Sarvam emits no ``vad.speech_start`` under manual endpointing, so the client
        boundary is what starts an utterance. Resetting here keeps the per-utterance
        flags and timings from leaking across turns, including after an ``endpointing``
        switch from ``vad`` to ``manual``.
        """
        self._reset_utterance_state()
        self._utterance_in_progress = True
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.START_OF_SPEECH,
                request_id=self._request_id,
            )
        )

    def _end_manual_utterance(self) -> None:
        """Close a client-delimited turn and anchor its speech-end position."""
        self._utterance_speech_end_audio_pos = self._audio_position
        self._utterance_speech_end_wall = time.time()
        self._emit_end_of_speech()
        self._complete_utterance()

    async def _safe_send_str(
        self,
        ws: Any,
        payload: dict[str, Any],
    ) -> bool:
        """Send a JSON control message, tolerating a peer that already closed.

        Returns:
            Whether the payload reached the socket.
        """
        if ws.closed:
            return False

        try:
            await ws.send_str(json.dumps(payload))
        except (aiohttp.ClientConnectionResetError, ConnectionError):
            self._logger.debug(
                "Sarvam realtime STT WebSocket closed before send completed",
                extra={**self._build_log_context(), "payload": payload},
            )
            return False
        return True

    async def _safe_send_bytes(self, ws: Any, payload: bytes) -> None:
        if ws.closed:
            return

        try:
            await ws.send_bytes(payload)
        except (aiohttp.ClientConnectionResetError, ConnectionError):
            self._logger.debug(
                "Sarvam realtime STT WebSocket closed before audio send completed",
                extra={**self._build_log_context(), "payload_bytes": len(payload)},
            )

    async def _send_pending_config_update(self, ws: Any) -> None:
        payload = self._pending_config_update
        self._pending_config_update = None
        if payload is None:
            return
        if await self._safe_send_str(ws, payload) and "endpointing" in payload:
            # Only now can a config.updated acknowledgement refer to our change.
            self._endpointing_update_sent = True

    async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
        ws_url = _build_realtime_ws_url(self._opts.base_url, self._opts)
        headers = {
            "API-SUBSCRIPTION-KEY": self._opts.api_key,
            "User-Agent": USER_AGENT,
        }
        self._logger.debug(
            "Connecting to Sarvam realtime STT WebSocket", extra=self._build_log_context()
        )
        try:
            ws = await asyncio.wait_for(
                self._session.ws_connect(ws_url, headers=headers, heartbeat=30.0),
                self._conn_options.timeout,
            )
        except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e:
            self._logger.error(
                "Failed to connect to Sarvam realtime STT WebSocket",
                extra={**self._build_log_context(), "error": str(e), "url": ws_url},
                exc_info=True,
            )
            raise
        except aiohttp.ClientResponseError as e:
            self._logger.error(
                "Sarvam realtime STT WebSocket handshake failed",
                extra={
                    **self._build_log_context(),
                    "error": e.message,
                    "status_code": e.status,
                    "url": ws_url,
                },
                exc_info=True,
            )
            raise
        except Exception as e:
            self._logger.error(
                "Unexpected Sarvam realtime STT WebSocket connection error",
                extra={**self._build_log_context(), "error": str(e), "url": ws_url},
                exc_info=True,
            )
            raise APIConnectionError("failed to connect to Sarvam realtime STT") from e
        self._logger.debug(
            "Sarvam realtime STT WebSocket connected", extra=self._build_log_context()
        )
        return ws

    @utils.log_exceptions(logger=logger)
    async def _process_audio(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        samples_per_channel = max(int(self._opts.sample_rate * AUDIO_CHUNK_MS / 1000), 1)
        audio_bstream = utils.audio.AudioByteStream(
            sample_rate=self._opts.sample_rate,
            num_channels=1,
            samples_per_channel=samples_per_channel,
        )

        async for data in self._input_ch:
            # The server is done reading, so stop the pump instead of writing into a
            # socket whose reset would fail the whole stream (max_retry is forced to 0).
            if self._session_ended or ws.closed:
                break

            await self._send_pending_config_update(ws)
            frames: list[rtc.AudioFrame] = []
            if isinstance(data, rtc.AudioFrame):
                frames.extend(audio_bstream.write(data.data.tobytes()))
            if isinstance(data, self._FlushSentinel):
                self._flush_observed = True
                frames.extend(audio_bstream.flush())

            for frame in frames:
                if self._active_endpointing == "manual" and not self._manual_speech_started:
                    await self._safe_send_str(ws, {"event": "speech_start"})
                    self._manual_speech_started = True
                    self._begin_manual_utterance()

                self._audio_duration_collector.push(frame.duration)
                self._audio_position += frame.duration
                await self._safe_send_bytes(
                    ws, _encode_pcm_for_wire(self._opts.encoding, frame.data.tobytes())
                )

            if isinstance(data, self._FlushSentinel):
                self._audio_duration_collector.flush()
                if self._active_endpointing == "manual" and self._manual_speech_started:
                    await self._safe_send_str(ws, {"event": "speech_end"})
                    self._manual_speech_started = False
                    self._end_manual_utterance()

        self._emit_local_usage_fallback()
        if not self._session_ended:
            await self._safe_send_str(ws, {"event": "end"})

    @utils.log_exceptions(logger=logger)
    async def _process_messages(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        while True:
            msg = await ws.receive()
            if msg.type == aiohttp.WSMsgType.TEXT:
                try:
                    await self._handle_message(json.loads(msg.data))
                except json.JSONDecodeError as e:
                    if _looks_like_error_text(msg.data):
                        self._logger.error(
                            "Sarvam realtime STT non-JSON error message",
                            extra={**self._build_log_context(), "raw_message": msg.data},
                        )
                        raise APIStatusError(
                            message=f"Sarvam realtime STT non-JSON error message: {msg.data}",
                            request_id=self._request_id or None,
                            body={"raw_message": msg.data},
                        ) from e
                    self._logger.warning(
                        "Invalid JSON received from Sarvam realtime STT",
                        extra={**self._build_log_context(), "raw_data": msg.data},
                    )
                    continue
                if self._session_ended:
                    break
            elif msg.type == aiohttp.WSMsgType.ERROR:
                self._logger.error(
                    "Sarvam realtime STT WebSocket error",
                    extra={**self._build_log_context(), "raw_message": msg.data},
                )
                raise APIConnectionError(f"Sarvam realtime STT WebSocket error: {msg.data}")
            elif msg.type in (
                aiohttp.WSMsgType.CLOSED,
                aiohttp.WSMsgType.CLOSE,
                aiohttp.WSMsgType.CLOSING,
            ):
                close_code = ws.close_code if ws.close_code is not None else msg.data
                close_reason = msg.extra
                if self._session_ended and close_code in (1000, 1001, None):
                    self._flush_terminal_utterance()
                    self._emit_local_usage_fallback()
                    break
                if close_code in (1000, 1001, None) and not _looks_like_error_text(close_reason):
                    self._flush_terminal_utterance()
                    self._emit_local_usage_fallback()
                    break
                self._logger.error(
                    "Sarvam realtime STT WebSocket closed unexpectedly",
                    extra={
                        **self._build_log_context(),
                        "close_code": close_code,
                        "close_reason": close_reason,
                    },
                )
                raise self._status_error_from_close(close_code, close_reason)
            else:
                self._logger.debug(
                    "Unknown Sarvam realtime STT WebSocket message type",
                    extra={**self._build_log_context(), "message_type": str(msg.type)},
                )

    def _status_error_from_close(self, close_code: object, close_reason: object) -> APIStatusError:
        status_code = int(close_code) if isinstance(close_code, int) else -1
        retryable = close_code == 1013
        message = f"Sarvam realtime STT WebSocket closed unexpectedly: {close_reason}"
        if close_code == 1003:
            message = "Sarvam realtime STT authentication, quota, or rate limit error"
        elif close_code == 1008:
            message = "Sarvam realtime STT session timed out or exceeded the maximum duration"
        elif close_code == 1013:
            message = "Sarvam realtime STT backend temporarily unavailable"
        elif close_code == 4000:
            message = f"Sarvam realtime STT rejected the session: {close_reason}"

        return APIStatusError(
            message=message,
            status_code=status_code,
            request_id=self._request_id or None,
            body={
                "close_code": close_code,
                "close_reason": close_reason,
            },
            retryable=retryable,
        )

    async def _handle_message(self, data: dict[str, Any]) -> None:
        event = data.get("event")
        self._capture_server_ids(data)
        if event == "session.begin":
            config = data.get("config")
            self._resolved_config = dict(config) if isinstance(config, dict) else None
        self._log_stt_event(event, data)
        if event == "session.begin":
            return
        elif event == "vad.speech_start":
            self._reset_utterance_state()
            self._utterance_in_progress = True
            utterance_idx = data.get("utterance_idx")
            self._utterance_idx = utterance_idx if isinstance(utterance_idx, int) else None
            self._event_ch.send_nowait(
                stt.SpeechEvent(
                    type=stt.SpeechEventType.START_OF_SPEECH,
                    request_id=self._request_id,
                )
            )
        elif event == "vad.speech_end":
            self._handle_speech_end()
        elif event == "transcript.partial":
            self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data)
        elif event == "transcript.final":
            if self._active_endpointing == "vad":
                if self._is_valid_transcript(data):
                    self._pending_final_data = data
                    self._final_received_for_utterance = True
                    self._try_commit_utterance()
            elif self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data):
                self._final_received_for_utterance = True
                self._complete_utterance()
        elif event == "session.end":
            self._handle_session_end(data)
        elif event == "config.updated":
            self._handle_config_updated(data)
            return
        elif event == "error":
            self._handle_error_event(data)
        elif event == "pong":
            return
        else:
            self._logger.debug(
                "Unknown Sarvam realtime STT event",
                extra={**self._build_log_context(), "event": event, "data": data},
            )

    def _log_stt_event(self, event: object, data: dict[str, Any]) -> None:
        if event == "pong":
            return

        extra: dict[str, Any] = {
            **self._build_log_context(),
            "event": event,
            "utterance_idx": data.get("utterance_idx"),
        }
        if event in {"transcript.partial", "transcript.final"}:
            # Recognized speech is personal data, so only its length is safe for the
            # INFO record; the text itself stays in the opt-in DEBUG raw payload.
            text = data.get("text")
            if isinstance(text, str):
                extra["text_length"] = len(text)
            extra["language"] = data.get("language") or self._opts.language
            extra["confidence"] = data.get("language_confidence", data.get("confidence"))
        elif event == "vad.speech_start":
            extra["audio_position"] = self._audio_position
        elif event == "vad.speech_end":
            extra["audio_position"] = self._audio_position
        elif event == "session.begin":
            pass
        elif event == "session.end":
            extra["audio_duration_s"] = data.get("audio_duration_s")
        elif event == "config.updated":
            extra["applied"] = data.get("applied")
        elif event == "error":
            extra["error_code"] = data.get("code")
            extra["error_message"] = data.get("message")
            extra["status_code"] = data.get("status_code")
        else:
            return

        if event == "transcript.partial":
            self._logger.debug(
                "Sarvam realtime STT transcript.partial",
                extra={**extra, "raw_data": data},
            )
            return

        self._logger.info(f"Sarvam realtime STT {event}", extra=extra)
        self._logger.debug(
            "Sarvam realtime STT raw event",
            extra={**extra, "raw_data": data},
        )

    def _is_valid_transcript(self, data: dict[str, Any]) -> bool:
        text = data.get("text")
        # Whitespace carries no content, and emitting it would commit a user turn
        # with no words.
        return isinstance(text, str) and bool(text.strip())

    def _handle_speech_end(self) -> None:
        self._utterance_speech_end_audio_pos = self._audio_position
        self._utterance_speech_end_wall = time.time()

        if self._active_endpointing != "vad":
            self._emit_end_of_speech()
        elif not self._eos_emitted_for_utterance:
            self._emit_end_of_speech()
            if self._final_received_for_utterance:
                self._try_commit_utterance()

        # The server's speech end is the utterance boundary, so the turn is over even
        # when the final is empty or never arrives. Completing unconditionally is what
        # lets a boundary-gated endpointing change promote; leaving the utterance open
        # would strand the stream in the old mode with the server in the new one.
        self._complete_utterance()

    def _try_commit_utterance(self) -> None:
        if self._pending_final_data is None or self._utterance_speech_end_audio_pos is None:
            return

        committed_data = self._pending_final_data
        if self._send_transcript_event(
            stt.SpeechEventType.FINAL_TRANSCRIPT,
            committed_data,
        ):
            self._logger.debug(
                "Sarvam realtime STT utterance committed",
                extra={
                    **self._build_log_context(),
                    "end_time": self._utterance_speech_end_audio_pos,
                    "speech_end_wall_time": self._utterance_speech_end_wall,
                },
            )
            if not self._eos_emitted_for_utterance:
                self._emit_end_of_speech()
            self._pending_final_data = None
            self._complete_utterance()

    def _flush_terminal_utterance(self) -> None:
        """Commit a buffered final transcript when the session ends mid-utterance.

        In VAD endpointing a ``transcript.final`` is held until ``vad.speech_end``
        supplies the speech-end position. When the input audio ends mid-utterance
        the server finalizes and closes without that event, so the speech-end
        position is anchored to the audio consumed so far instead of dropping the
        transcript. Safe to call more than once per session.
        """
        if self._pending_final_data is not None and self._utterance_speech_end_audio_pos is None:
            self._utterance_speech_end_audio_pos = self._audio_position
            if self._utterance_speech_end_wall is None:
                self._utterance_speech_end_wall = time.time()

        if not self._eos_emitted_for_utterance and self._pending_final_data is not None:
            self._emit_end_of_speech()

        self._try_commit_utterance()

    def _emit_end_of_speech(self) -> None:
        if self._eos_emitted_for_utterance:
            return

        # Emitted without alternatives so the agent pipeline treats it as a sentinel
        # it can hold and release with a concrete transcript. The speech-end timing
        # travels on the FINAL_TRANSCRIPT event instead.
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.END_OF_SPEECH,
                request_id=self._request_id,
            )
        )
        self._eos_emitted_for_utterance = True

    def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict[str, Any]) -> bool:
        text = data.get("text")
        if not isinstance(text, str) or not text.strip():
            return False

        language = data.get("language") or self._opts.language
        # Recognition confidence only: `language_confidence` is a language-identification
        # score and stays in metadata. The endpoint sends no per-segment confidence
        # today, and an absent value falls back to 1.0 (as `_extract_confidence` in
        # stt.py does) so it isn't averaged downstream as "no confidence".
        # bool is a subclass of int, so exclude it explicitly.
        confidence = data.get("confidence")
        if not isinstance(confidence, (int, float)) or isinstance(confidence, bool):
            confidence = 1.0

        metadata: dict[str, Any] = {
            key: data[key]
            for key in ("utterance_idx", "language_confidence")
            if key in data and data[key] is not None
        }
        if (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._utterance_speech_end_wall is not None
        ):
            metadata["speech_end_wall_time"] = self._utterance_speech_end_wall
        end_time = 0.0
        start_time = 0.0
        if event_type == stt.SpeechEventType.FINAL_TRANSCRIPT:
            start_s = data.get("start_s")
            end_s = data.get("end_s")
            if isinstance(start_s, (int, float)) and not isinstance(start_s, bool):
                start_time = max(float(start_s), 0.0)
            if isinstance(end_s, (int, float)) and not isinstance(end_s, bool):
                end_time = max(float(end_s), 0.0)
        if (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._utterance_speech_end_audio_pos is not None
            and end_time == 0.0
        ):
            end_time = self._utterance_speech_end_audio_pos
        elif (
            event_type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and self._audio_position > 0
            and end_time == 0.0
        ):
            end_time = self._audio_position

        speech_data = stt.SpeechData(
            language=LanguageCode(language),
            text=text,
            start_time=start_time,
            end_time=end_time,
            confidence=float(confidence),
            metadata=metadata or None,
        )
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=event_type,
                request_id=self._request_id,
                alternatives=[speech_data],
            )
        )
        return True

    def _handle_session_end(self, data: dict[str, Any]) -> None:
        self._capture_server_ids(data)
        self._flush_terminal_utterance()
        audio_duration = data.get("audio_duration_s")
        if (
            isinstance(audio_duration, (int, float))
            and not isinstance(audio_duration, bool)
            and not self._server_audio_duration_reported
        ):
            # Report whatever audio is still buffered locally, then top up to Sarvam's
            # authoritative total so the session bills exactly once for it.
            self._audio_duration_collector.flush()
            server_audio_duration = max(float(audio_duration), 0.0)
            delta = max(server_audio_duration - self._total_reported_audio_duration, 0.0)
            if delta:
                self._emit_usage(delta)
            self._server_audio_duration_reported = True
        else:
            self._emit_local_usage_fallback()
        self._session_ended = True

    def _handle_error_event(self, data: dict[str, Any]) -> None:
        if not data.get("is_fatal", False):
            self._logger.warning(
                "Non-fatal Sarvam realtime STT error",
                extra={
                    **self._build_log_context(),
                    "error_code": data.get("code"),
                    "error_message": data.get("message"),
                    "status_code": data.get("status_code"),
                    "raw_message": data,
                },
            )
            return

        code = data.get("code", "unknown")
        status_code = data.get("status_code", -1)
        if not isinstance(status_code, int):
            status_code = -1
        self._logger.error(
            "Fatal Sarvam realtime STT error",
            extra={
                **self._build_log_context(),
                "error_code": code,
                "error_message": data.get("message", code),
                "status_code": status_code,
                "raw_message": data,
            },
        )
        raise APIStatusError(
            message=f"Sarvam realtime STT error: {data.get('message', code)}",
            status_code=status_code,
            request_id=self._request_id or None,
            body=data,
            retryable=code == "model_unavailable",
        )

    def _on_audio_duration_report(self, duration: float) -> None:
        self._local_audio_duration += duration
        self._emit_usage(duration)

    def _emit_local_usage_fallback(self) -> None:
        if self._server_audio_duration_reported:
            return
        self._audio_duration_collector.flush()

    def _emit_usage(self, duration: float) -> None:
        self._total_reported_audio_duration += duration
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.RECOGNITION_USAGE,
                request_id=self._request_id,
                recognition_usage=stt.RecognitionUsage(audio_duration=duration),
            )
        )

A single WebSocket session against Sarvam's realtime STT endpoint.

Audio pushed into the stream is forwarded in the configured wire encoding, and the events Sarvam returns are translated into LiveKit speech events. Audio duration is reported incrementally while the session runs and reconciled against the server's authoritative total when the session ends.

Create a realtime speech stream.

Args

stt

The parent instance that created this stream.

opts

Resolved options for this connection.

conn_options

Connection options for this stream.

http_session

aiohttp session used to open the WebSocket.

Ancestors

• livekit.agents.stt.stt.RecognizeStream
• abc.ABC

Instance variables

prop resolved_config : dict[str, Any] | None

Expand source code

@property
def resolved_config(self) -> dict[str, Any] | None:
    """Return the configuration resolved by Sarvam for this connection."""
    return dict(self._resolved_config) if self._resolved_config is not None else None

Return the configuration resolved by Sarvam for this connection.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close the connection, reporting any audio duration not yet billed.

    Agents normally end a stream here rather than waiting for ``session.end``, so
    the pending duration is flushed before the base class cancels the tasks that
    deliver usage metrics.
    """
    try:
        if not self._event_ch.closed:
            self._emit_local_usage_fallback()
            # Give the metrics monitor a chance to consume the usage event before
            # super().aclose() cancels it.
            await asyncio.sleep(0)
        if self._ws and not self._ws.closed:
            await self._ws.close()
    finally:
        self._ws = None
        await super().aclose()

Close the connection, reporting any audio duration not yet billed.

Agents normally end a stream here rather than waiting for session.end, so the pending duration is flushed before the base class cancels the tasks that deliver usage metrics.

def update_options(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str | None] = NOT_GIVEN,
return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
    mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
    endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str | None] = NOT_GIVEN,
    return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
    vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
) -> None:
    """Apply an option change to this live connection.

    Only the options explicitly passed here are changed, so per-stream overrides
    such as a ``language`` given to :meth:`STTRealtime.stream` survive an unrelated
    update. Connection-time options are retained at their current values and a
    warning is logged, since changing them would desynchronize the
    already-negotiated session. Every other change is queued as an in-band
    ``config.update`` sent before the next audio frame.

    Args:
        language: BCP-47 language code, or ``auto`` for adaptive identification.
        stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
        mode: Task applied to finals.
        endpointing: ``vad`` for server-side turn detection, or ``manual``.
        sample_rate: Audio sample rate in Hz; retained on a live stream.
        prompt: Context or terminology hint; ``None`` clears it.
        return_timestamps: Segment-level timestamps; retained on a live stream.
        vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
        vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
        vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
        vad_prefix_padding_ms: Speech-onset padding; retained on a live stream.

    Raises:
        ValueError: If an option falls outside the values the endpoint accepts.
    """
    previous_opts = self._opts
    requested: dict[str, Any] = {}
    if is_given(language):
        requested["language"] = language
    if is_given(stream_type):
        requested["stream_type"] = stream_type
    if is_given(mode):
        requested["mode"] = mode
    if is_given(endpointing):
        requested["endpointing"] = endpointing
    if is_given(sample_rate):
        requested["sample_rate"] = sample_rate
    if is_given(prompt):
        requested["prompt"] = prompt
    if is_given(return_timestamps):
        requested["return_timestamps"] = return_timestamps
    if is_given(vad_sot_threshold):
        requested["vad_sot_threshold"] = vad_sot_threshold
    if is_given(vad_min_speech_ms):
        requested["vad_min_speech_ms"] = vad_min_speech_ms
    if is_given(vad_min_silence_ms):
        requested["vad_min_silence_ms"] = vad_min_silence_ms
    if is_given(vad_prefix_padding_ms):
        requested["vad_prefix_padding_ms"] = vad_prefix_padding_ms

    if not requested:
        return

    opts = replace(previous_opts, **requested)
    connection_only_options: list[str] = []
    if opts.sample_rate != previous_opts.sample_rate:
        connection_only_options.append("sample_rate")
        opts = replace(opts, sample_rate=previous_opts.sample_rate)
    if opts.return_timestamps != previous_opts.return_timestamps:
        connection_only_options.append("return_timestamps")
        opts = replace(opts, return_timestamps=previous_opts.return_timestamps)
    if opts.vad_prefix_padding_ms != previous_opts.vad_prefix_padding_ms:
        connection_only_options.append("vad_prefix_padding_ms")
        opts = replace(opts, vad_prefix_padding_ms=previous_opts.vad_prefix_padding_ms)
    if connection_only_options:
        self._logger.warning(
            "Sarvam realtime STT connection-only option updates only apply to new streams",
            extra={
                **self._build_log_context(),
                "options": connection_only_options,
            },
        )

    self._opts = opts
    if opts.endpointing != previous_opts.endpointing:
        if opts.endpointing == "manual" and not self._flush_observed:
            self._logger.warning(
                "Sarvam realtime STT switched to manual endpointing without an external VAD; "
                "turns will not be delimited unless the agent framework flushes the stream. "
                "Configure a VAD on the AgentSession to receive end-of-turn boundaries.",
                extra=self._build_log_context(),
            )
        self._pending_endpointing = opts.endpointing
        self._endpointing_update_acknowledged = False
        self._endpointing_update_sent = False

    update = self._config_update_payload(previous_opts, opts)
    if update is not None:
        if self._pending_config_update is None:
            self._pending_config_update = update
        else:
            self._pending_config_update.update(update)

Apply an option change to this live connection.

Only the options explicitly passed here are changed, so per-stream overrides such as a language given to :meth:STTRealtime.stream() survive an unrelated update. Connection-time options are retained at their current values and a warning is logged, since changing them would desynchronize the already-negotiated session. Every other change is queued as an in-band config.update sent before the next audio frame.

Args

language

BCP-47 language code, or auto for adaptive identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals.

endpointing

vad for server-side turn detection, or manual.

sample_rate

Audio sample rate in Hz; retained on a live stream.

prompt

Context or terminology hint; None clears it.

return_timestamps

Segment-level timestamps; retained on a live stream.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Speech-onset padding; retained on a live stream.

Raises

ValueError

If an option falls outside the values the endpoint accepts.

class STT (*,
language: str = 'en-IN',
model: SarvamSTTModels | str = 'saaras:v4',
mode: SarvamSTTModes | str = 'transcribe',
api_key: str | None = None,
base_url: str | None = None,
http_session: aiohttp.ClientSession | None = None,
prompt: str | None = None,
high_vad_sensitivity: bool | None = None,
sample_rate: int = 16000,
flush_signal: bool | None = None,
input_audio_codec: str | None = None,
positive_speech_threshold: float | None = None,
negative_speech_threshold: float | None = None,
min_speech_frames: int | None = None,
first_turn_min_speech_frames: int | None = None,
negative_frames_count: int | None = None,
negative_frames_window: int | None = None,
start_speech_volume_threshold: float | None = None,
interrupt_min_speech_frames: int | None = None,
pre_speech_pad_frames: int | None = None,
num_initial_ignored_frames: int | None = None)

Expand source code

class STT(stt.STT):
    """Sarvam.ai Speech-to-Text implementation.

    This class provides speech-to-text functionality using the Sarvam.ai API.
    Sarvam.ai specializes in high-quality STT for Indian languages.

    Args:
        language: BCP-47 language code, e.g., "hi-IN", "en-IN"
        model: The Sarvam STT model to use
        mode: Mode for saaras:v3/v4 (transcribe/translate/verbatim/translit/codemix)
        api_key: Sarvam.ai API key (falls back to SARVAM_API_KEY env var)
        base_url: API endpoint URL
        http_session: Optional aiohttp session to use
        prompt: Optional prompt for STT translate (saaras models only)
    """

    def __init__(
        self,
        *,
        language: str = "en-IN",
        model: SarvamSTTModels | str = "saaras:v4",
        mode: SarvamSTTModes | str = "transcribe",
        api_key: str | None = None,
        base_url: str | None = None,
        http_session: aiohttp.ClientSession | None = None,
        prompt: str | None = None,
        high_vad_sensitivity: bool | None = None,
        sample_rate: int = 16000,
        flush_signal: bool | None = None,
        input_audio_codec: str | None = None,
        positive_speech_threshold: float | None = None,
        negative_speech_threshold: float | None = None,
        min_speech_frames: int | None = None,
        first_turn_min_speech_frames: int | None = None,
        negative_frames_count: int | None = None,
        negative_frames_window: int | None = None,
        start_speech_volume_threshold: float | None = None,
        interrupt_min_speech_frames: int | None = None,
        pre_speech_pad_frames: int | None = None,
        num_initial_ignored_frames: int | None = None,
    ) -> None:
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                # chunk timestamps don't seem to work despite the docs saying they do
                aligned_transcript=False,
            )
        )

        self._api_key = api_key or os.environ.get("SARVAM_API_KEY")
        if not self._api_key:
            raise ValueError(
                "Sarvam API key is required. "
                "Provide it directly or set SARVAM_API_KEY environment variable."
            )

        self._opts = SarvamSTTOptions(
            language=LanguageCode(language),
            api_key=self._api_key,
            model=model,
            mode=mode,
            base_url=base_url,
            prompt=prompt,
            high_vad_sensitivity=high_vad_sensitivity,
            sample_rate=sample_rate,
            flush_signal=flush_signal,
            input_audio_codec=input_audio_codec,
            positive_speech_threshold=positive_speech_threshold,
            negative_speech_threshold=negative_speech_threshold,
            min_speech_frames=min_speech_frames,
            first_turn_min_speech_frames=first_turn_min_speech_frames,
            negative_frames_count=negative_frames_count,
            negative_frames_window=negative_frames_window,
            start_speech_volume_threshold=start_speech_volume_threshold,
            interrupt_min_speech_frames=interrupt_min_speech_frames,
            pre_speech_pad_frames=pre_speech_pad_frames,
            num_initial_ignored_frames=num_initial_ignored_frames,
        )
        self._session = http_session
        self._logger = logger.getChild(self.__class__.__name__)
        self._streams = weakref.WeakSet[SpeechStream]()
        _warn_if_sunset_stt_model(model)

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

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

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()
        return self._session

    @staticmethod
    def _single_attempt_conn_options(conn_options: APIConnectOptions) -> APIConnectOptions:
        return APIConnectOptions(
            max_retry=0,
            retry_interval=conn_options.retry_interval,
            timeout=conn_options.timeout,
        )

    async def recognize(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.SpeechEvent:
        single_attempt_conn_options = self._single_attempt_conn_options(conn_options)
        return await super().recognize(
            buffer,
            language=language,
            conn_options=single_attempt_conn_options,
        )

    def _resolve_opts(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[SarvamSTTModels | str] = NOT_GIVEN,
        mode: NotGivenOr[SarvamSTTModes | str] = NOT_GIVEN,
    ) -> tuple[str, str, str]:
        """Resolve language, model and mode from overrides or defaults.

        Returns:
            Tuple of (language, model, mode).

        Raises:
            ValueError: If mode is explicitly given but not supported by the model.
        """
        resolved_language = LanguageCode(language) if is_given(language) else self._opts.language
        resolved_model = model if is_given(model) else self._opts.model
        if not isinstance(resolved_language, str):
            resolved_language = self._opts.language
        if not isinstance(resolved_model, str):
            resolved_model = self._opts.model
        if is_given(model):
            _warn_if_sunset_stt_model(resolved_model)

        if is_given(mode):
            resolved_mode = str(mode)
            # Validate: caller explicitly asked for a mode — error if unsupported
            _validate_mode_for_model(resolved_model, resolved_mode)
        else:
            resolved_mode = self._opts.mode

        _validate_language_for_model(resolved_model, resolved_language)

        return resolved_language, resolved_model, resolved_mode

    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[SarvamSTTModels | str] = NOT_GIVEN,
        mode: NotGivenOr[SarvamSTTModes | str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.SpeechEvent:
        """Recognize speech using Sarvam.ai API.

        Args:
            buffer: Audio buffer containing speech data
            language: BCP-47 language code (overrides the one set in constructor)
            model: Sarvam model to use (overrides the one set in constructor)
            conn_options: Connection options for API requests

        Returns:
            A SpeechEvent containing the transcription result

        Raises:
            APIConnectionError: On network connection errors
            APIStatusError: On API errors (non-200 status)
            APITimeoutError: On API timeout
        """
        opts_language, opts_model, opts_mode = self._resolve_opts(
            language=language,
            model=model,
            mode=mode,
        )

        wav_bytes = rtc.combine_audio_frames(buffer).to_wav_bytes()

        form_data = aiohttp.FormData()
        form_data.add_field("file", wav_bytes, filename="audio.wav", content_type="audio/wav")

        # Add model and language_code to the form data
        if opts_language:
            form_data.add_field("language_code", opts_language)
        if opts_model:
            form_data.add_field("model", str(opts_model))
        if _model_supports_mode(opts_model):
            form_data.add_field("mode", str(opts_mode))

        if not self._api_key:
            raise ValueError("API key cannot be None")
        headers = {
            "api-subscription-key": self._api_key,
            "User-Agent": USER_AGENT,
        }

        try:
            base_url, _ = _get_urls_for_model(opts_model)
            async with self._ensure_session().post(
                url=base_url,
                data=form_data,
                headers=headers,
                timeout=aiohttp.ClientTimeout(
                    total=conn_options.timeout,
                    sock_connect=conn_options.timeout,
                ),
            ) as res:
                if res.status != 200:
                    error_text = await res.text()
                    self._logger.error(f"Sarvam API error: {res.status} - {error_text}")
                    raise APIStatusError(
                        message=f"Sarvam API Error ({res.status}): {error_text}",
                        status_code=res.status,
                        body=error_text,
                    )

                response_json = await res.json()
                self._logger.debug(
                    "Sarvam API response received", extra={"lk.pii.response_json": response_json}
                )

                transcript_text = response_json.get("transcript", "")
                request_id = response_json.get("request_id", "")
                detected_language = response_json.get("language_code")
                if not isinstance(detected_language, str):
                    detected_language = LanguageCode(opts_language or "")
                else:
                    detected_language = LanguageCode(detected_language)

                start_time = 0.0
                end_time = 0.0

                # Try to get timestamps if available
                timestamps_data = response_json.get("timestamps")
                if timestamps_data and isinstance(timestamps_data, dict):
                    words_ts_start = timestamps_data.get("start_time_seconds")
                    words_ts_end = timestamps_data.get("end_time_seconds")
                    if isinstance(words_ts_start, list) and len(words_ts_start) > 0:
                        start_time = words_ts_start[0]
                    if isinstance(words_ts_end, list) and len(words_ts_end) > 0:
                        end_time = words_ts_end[-1]

                # If start/end times are still 0, use buffer duration as an estimate for end_time
                if start_time == 0.0 and end_time == 0.0:
                    end_time = _calculate_audio_duration(buffer)

                alternatives = [
                    stt.SpeechData(
                        language=detected_language,
                        text=transcript_text,
                        start_time=start_time,
                        end_time=end_time,
                        confidence=_extract_confidence(response_json, self._logger),
                    )
                ]

                return stt.SpeechEvent(
                    type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                    request_id=request_id,
                    alternatives=alternatives,
                )

        except asyncio.TimeoutError as e:
            self._logger.error(f"Sarvam API timeout: {e}")
            raise APITimeoutError("Sarvam API request timed out") from e
        except aiohttp.ClientError as e:
            self._logger.error(f"Sarvam API client error: {e}")
            raise APIConnectionError(f"Sarvam API connection error: {e}") from e
        except (APIStatusError, APIConnectionError, APITimeoutError):
            # Preserve provider-originated status/body/retry metadata.
            raise
        except Exception as e:
            self._logger.error(f"Error during Sarvam STT processing: {e}")
            raise APIConnectionError(f"Unexpected error in Sarvam STT: {e}") from e

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[SarvamSTTModels | str] = NOT_GIVEN,
        mode: NotGivenOr[SarvamSTTModes | str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        prompt: NotGivenOr[str] = NOT_GIVEN,
        high_vad_sensitivity: NotGivenOr[bool] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        flush_signal: NotGivenOr[bool] = NOT_GIVEN,
        input_audio_codec: NotGivenOr[str] = NOT_GIVEN,
        positive_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
        negative_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
        min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
        first_turn_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
        negative_frames_count: NotGivenOr[int] = NOT_GIVEN,
        negative_frames_window: NotGivenOr[int] = NOT_GIVEN,
        start_speech_volume_threshold: NotGivenOr[float] = NOT_GIVEN,
        interrupt_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
        pre_speech_pad_frames: NotGivenOr[int] = NOT_GIVEN,
        num_initial_ignored_frames: NotGivenOr[int] = NOT_GIVEN,
    ) -> SpeechStream:
        """Create a streaming transcription session."""
        opts_language, opts_model, opts_mode = self._resolve_opts(
            language=language,
            model=model,
            mode=mode,
        )

        # Handle prompt conversion from NotGiven to None
        final_prompt = prompt if isinstance(prompt, str) else self._opts.prompt

        opts_high_vad = (
            high_vad_sensitivity
            if is_given(high_vad_sensitivity)
            else self._opts.high_vad_sensitivity
        )
        opts_sample_rate = sample_rate if is_given(sample_rate) else self._opts.sample_rate
        opts_flush_signal = flush_signal if is_given(flush_signal) else self._opts.flush_signal
        opts_input_codec = (
            input_audio_codec if is_given(input_audio_codec) else self._opts.input_audio_codec
        )
        opts_positive_speech = (
            positive_speech_threshold
            if is_given(positive_speech_threshold)
            else self._opts.positive_speech_threshold
        )
        opts_negative_speech = (
            negative_speech_threshold
            if is_given(negative_speech_threshold)
            else self._opts.negative_speech_threshold
        )
        opts_min_speech = (
            min_speech_frames if is_given(min_speech_frames) else self._opts.min_speech_frames
        )
        opts_first_turn = (
            first_turn_min_speech_frames
            if is_given(first_turn_min_speech_frames)
            else self._opts.first_turn_min_speech_frames
        )
        opts_neg_count = (
            negative_frames_count
            if is_given(negative_frames_count)
            else self._opts.negative_frames_count
        )
        opts_neg_window = (
            negative_frames_window
            if is_given(negative_frames_window)
            else self._opts.negative_frames_window
        )
        opts_vol_threshold = (
            start_speech_volume_threshold
            if is_given(start_speech_volume_threshold)
            else self._opts.start_speech_volume_threshold
        )
        opts_interrupt = (
            interrupt_min_speech_frames
            if is_given(interrupt_min_speech_frames)
            else self._opts.interrupt_min_speech_frames
        )
        opts_pre_pad = (
            pre_speech_pad_frames
            if is_given(pre_speech_pad_frames)
            else self._opts.pre_speech_pad_frames
        )
        opts_initial_ignored = (
            num_initial_ignored_frames
            if is_given(num_initial_ignored_frames)
            else self._opts.num_initial_ignored_frames
        )
        single_attempt_conn_options = self._single_attempt_conn_options(conn_options)

        # Create options for the stream
        stream_opts = SarvamSTTOptions(
            language=opts_language,
            api_key=self._api_key if self._api_key else "",
            model=opts_model,
            mode=opts_mode,
            prompt=final_prompt,
            high_vad_sensitivity=opts_high_vad,
            sample_rate=opts_sample_rate,
            flush_signal=opts_flush_signal,
            input_audio_codec=opts_input_codec,
            positive_speech_threshold=opts_positive_speech,
            negative_speech_threshold=opts_negative_speech,
            min_speech_frames=opts_min_speech,
            first_turn_min_speech_frames=opts_first_turn,
            negative_frames_count=opts_neg_count,
            negative_frames_window=opts_neg_window,
            start_speech_volume_threshold=opts_vol_threshold,
            interrupt_min_speech_frames=opts_interrupt,
            pre_speech_pad_frames=opts_pre_pad,
            num_initial_ignored_frames=opts_initial_ignored,
        )

        # Create a fresh session for this stream to avoid conflicts
        stream_session = aiohttp.ClientSession()

        if not self._api_key:
            raise ValueError("API key cannot be None")
        stream = SpeechStream(
            stt=self,
            opts=stream_opts,
            conn_options=single_attempt_conn_options,
            api_key=self._api_key,
            http_session=stream_session,
        )
        self._streams.add(stream)
        return stream

Sarvam.ai Speech-to-Text implementation.

This class provides speech-to-text functionality using the Sarvam.ai API. Sarvam.ai specializes in high-quality STT for Indian languages.

Args

language

BCP-47 language code, e.g., "hi-IN", "en-IN"

model

The Sarvam STT model to use

mode

Mode for saaras:v3/v4 (transcribe/translate/verbatim/translit/codemix)

api_key

Sarvam.ai API key (falls back to SARVAM_API_KEY env var)

base_url

API endpoint URL

http_session

Optional aiohttp session to use

prompt

Optional prompt for STT translate (saaras models only)

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

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 recognize(self,
buffer: AudioBuffer,
*,
language: NotGivenOr[str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.agents.stt.stt.SpeechEvent

Expand source code

async def recognize(
    self,
    buffer: AudioBuffer,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> stt.SpeechEvent:
    single_attempt_conn_options = self._single_attempt_conn_options(conn_options)
    return await super().recognize(
        buffer,
        language=language,
        conn_options=single_attempt_conn_options,
    )

def stream(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
model: NotGivenOr[SarvamSTTModels | str] = NOT_GIVEN,
mode: NotGivenOr[SarvamSTTModes | str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0),
prompt: NotGivenOr[str] = NOT_GIVEN,
high_vad_sensitivity: NotGivenOr[bool] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
flush_signal: NotGivenOr[bool] = NOT_GIVEN,
input_audio_codec: NotGivenOr[str] = NOT_GIVEN,
positive_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
negative_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
first_turn_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
negative_frames_count: NotGivenOr[int] = NOT_GIVEN,
negative_frames_window: NotGivenOr[int] = NOT_GIVEN,
start_speech_volume_threshold: NotGivenOr[float] = NOT_GIVEN,
interrupt_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
pre_speech_pad_frames: NotGivenOr[int] = NOT_GIVEN,
num_initial_ignored_frames: NotGivenOr[int] = NOT_GIVEN) ‑> livekit.plugins.sarvam.stt.SpeechStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    model: NotGivenOr[SarvamSTTModels | str] = NOT_GIVEN,
    mode: NotGivenOr[SarvamSTTModes | str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    prompt: NotGivenOr[str] = NOT_GIVEN,
    high_vad_sensitivity: NotGivenOr[bool] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    flush_signal: NotGivenOr[bool] = NOT_GIVEN,
    input_audio_codec: NotGivenOr[str] = NOT_GIVEN,
    positive_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
    negative_speech_threshold: NotGivenOr[float] = NOT_GIVEN,
    min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
    first_turn_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
    negative_frames_count: NotGivenOr[int] = NOT_GIVEN,
    negative_frames_window: NotGivenOr[int] = NOT_GIVEN,
    start_speech_volume_threshold: NotGivenOr[float] = NOT_GIVEN,
    interrupt_min_speech_frames: NotGivenOr[int] = NOT_GIVEN,
    pre_speech_pad_frames: NotGivenOr[int] = NOT_GIVEN,
    num_initial_ignored_frames: NotGivenOr[int] = NOT_GIVEN,
) -> SpeechStream:
    """Create a streaming transcription session."""
    opts_language, opts_model, opts_mode = self._resolve_opts(
        language=language,
        model=model,
        mode=mode,
    )

    # Handle prompt conversion from NotGiven to None
    final_prompt = prompt if isinstance(prompt, str) else self._opts.prompt

    opts_high_vad = (
        high_vad_sensitivity
        if is_given(high_vad_sensitivity)
        else self._opts.high_vad_sensitivity
    )
    opts_sample_rate = sample_rate if is_given(sample_rate) else self._opts.sample_rate
    opts_flush_signal = flush_signal if is_given(flush_signal) else self._opts.flush_signal
    opts_input_codec = (
        input_audio_codec if is_given(input_audio_codec) else self._opts.input_audio_codec
    )
    opts_positive_speech = (
        positive_speech_threshold
        if is_given(positive_speech_threshold)
        else self._opts.positive_speech_threshold
    )
    opts_negative_speech = (
        negative_speech_threshold
        if is_given(negative_speech_threshold)
        else self._opts.negative_speech_threshold
    )
    opts_min_speech = (
        min_speech_frames if is_given(min_speech_frames) else self._opts.min_speech_frames
    )
    opts_first_turn = (
        first_turn_min_speech_frames
        if is_given(first_turn_min_speech_frames)
        else self._opts.first_turn_min_speech_frames
    )
    opts_neg_count = (
        negative_frames_count
        if is_given(negative_frames_count)
        else self._opts.negative_frames_count
    )
    opts_neg_window = (
        negative_frames_window
        if is_given(negative_frames_window)
        else self._opts.negative_frames_window
    )
    opts_vol_threshold = (
        start_speech_volume_threshold
        if is_given(start_speech_volume_threshold)
        else self._opts.start_speech_volume_threshold
    )
    opts_interrupt = (
        interrupt_min_speech_frames
        if is_given(interrupt_min_speech_frames)
        else self._opts.interrupt_min_speech_frames
    )
    opts_pre_pad = (
        pre_speech_pad_frames
        if is_given(pre_speech_pad_frames)
        else self._opts.pre_speech_pad_frames
    )
    opts_initial_ignored = (
        num_initial_ignored_frames
        if is_given(num_initial_ignored_frames)
        else self._opts.num_initial_ignored_frames
    )
    single_attempt_conn_options = self._single_attempt_conn_options(conn_options)

    # Create options for the stream
    stream_opts = SarvamSTTOptions(
        language=opts_language,
        api_key=self._api_key if self._api_key else "",
        model=opts_model,
        mode=opts_mode,
        prompt=final_prompt,
        high_vad_sensitivity=opts_high_vad,
        sample_rate=opts_sample_rate,
        flush_signal=opts_flush_signal,
        input_audio_codec=opts_input_codec,
        positive_speech_threshold=opts_positive_speech,
        negative_speech_threshold=opts_negative_speech,
        min_speech_frames=opts_min_speech,
        first_turn_min_speech_frames=opts_first_turn,
        negative_frames_count=opts_neg_count,
        negative_frames_window=opts_neg_window,
        start_speech_volume_threshold=opts_vol_threshold,
        interrupt_min_speech_frames=opts_interrupt,
        pre_speech_pad_frames=opts_pre_pad,
        num_initial_ignored_frames=opts_initial_ignored,
    )

    # Create a fresh session for this stream to avoid conflicts
    stream_session = aiohttp.ClientSession()

    if not self._api_key:
        raise ValueError("API key cannot be None")
    stream = SpeechStream(
        stt=self,
        opts=stream_opts,
        conn_options=single_attempt_conn_options,
        api_key=self._api_key,
        http_session=stream_session,
    )
    self._streams.add(stream)
    return stream

Create a streaming transcription session.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class STTRealtime (*,
language: str = 'en-IN',
stream_type: RealtimeStreamType | str = 'balanced',
mode: RealtimeMode | str = 'transcribe',
endpointing: RealtimeEndpointing | str = 'vad',
encoding: RealtimeEncoding | str = 'linear16',
sample_rate: int = 16000,
prompt: str | None = None,
return_timestamps: bool = False,
api_key: str | None = None,
base_url: str = 'wss://api.sarvam.ai/speech-to-text-realtime/ws',
http_session: aiohttp.ClientSession | None = None,
vad_sot_threshold: float | None = None,
vad_min_speech_ms: int | None = None,
vad_min_silence_ms: int | None = None,
vad_prefix_padding_ms: int | None = None)

Expand source code

class STTRealtime(stt.STT):
    """Speech-to-text using Sarvam's realtime WebSocket endpoint (``saaras:v3-realtime``).

    This endpoint streams interim and final transcripts over a single
    WebSocket connection and supports either server-side VAD or
    client-driven (manual) turn boundaries.
    """

    def __init__(
        self,
        *,
        language: str = "en-IN",
        stream_type: RealtimeStreamType | str = "balanced",
        mode: RealtimeMode | str = "transcribe",
        endpointing: RealtimeEndpointing | str = "vad",
        encoding: RealtimeEncoding | str = "linear16",
        sample_rate: int = 16000,
        prompt: str | None = None,
        return_timestamps: bool = False,
        api_key: str | None = None,
        base_url: str = SARVAM_STT_REALTIME_URL,
        http_session: aiohttp.ClientSession | None = None,
        vad_sot_threshold: float | None = None,
        vad_min_speech_ms: int | None = None,
        vad_min_silence_ms: int | None = None,
        vad_prefix_padding_ms: int | None = None,
    ) -> None:
        """Create a Sarvam realtime STT instance.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive language identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals: ``transcribe``, ``translate``, ``verbatim``,
                ``translit``, or ``codemix``.
            endpointing: ``vad`` for server-side turn detection, or ``manual`` when the
                caller delimits turns by flushing the stream.
            encoding: Wire encoding: ``linear16``, ``linear32``, ``mulaw``, or ``alaw``.
            sample_rate: Audio sample rate in Hz; ``8000`` or ``16000``.
            prompt: Optional context or terminology hint used to bias decoding.
            return_timestamps: Whether finals should carry segment-level start and end times.
            api_key: Sarvam API key. Falls back to the ``SARVAM_API_KEY`` environment variable.
            base_url: WebSocket URL of the realtime endpoint.
            http_session: Optional aiohttp session to reuse for the connection.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Audio retained before speech onset in ms
                (``vad`` endpointing only).

        Raises:
            ValueError: If no API key is provided or found in the environment, or if an
                option falls outside the values the endpoint accepts.
        """
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                aligned_transcript=False,
                offline_recognize=False,
            )
        )

        api_key = api_key or os.environ.get("SARVAM_API_KEY")
        if not api_key:
            raise ValueError(
                "Sarvam API key is required. "
                "Provide it directly or set SARVAM_API_KEY environment variable."
            )

        self._opts = RealtimeSTTOptions(
            language=language,
            api_key=api_key,
            stream_type=stream_type,
            mode=mode,
            endpointing=endpointing,
            encoding=encoding,
            sample_rate=sample_rate,
            base_url=base_url,
            prompt=prompt,
            return_timestamps=return_timestamps,
            vad_sot_threshold=vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms,
        )
        self._session = http_session
        self._owns_session = http_session is None
        self._streams = weakref.WeakSet[RealtimeSpeechStream]()

    @property
    def model(self) -> str:
        """Name of the Sarvam realtime model backing this instance."""
        return REALTIME_MODEL

    @property
    def provider(self) -> str:
        """Name of the speech-to-text provider."""
        return "Sarvam"

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            try:
                self._session = utils.http_context.http_session()
                self._owns_session = False
            except RuntimeError:
                self._session = aiohttp.ClientSession()
                self._owns_session = True
        return self._session

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

    def update_options(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
        mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
        endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str | None] = NOT_GIVEN,
        return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
        vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
    ) -> None:
        """Update options for this instance and every stream it created.

        Options that Sarvam only accepts at connection time (``sample_rate``,
        ``return_timestamps``, and ``vad_prefix_padding_ms``) take effect on
        newly created streams only.
        The remaining options are sent to active streams as an in-band
        ``config.update``, and the boundary-gated ones apply from the next
        utterance boundary.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive language identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals.
            endpointing: ``vad`` for server-side turn detection, or ``manual``.
            sample_rate: Audio sample rate in Hz; applies to new streams only.
            prompt: Context or terminology hint; ``None`` clears it.
            return_timestamps: Segment-level timestamps; applies to new streams only.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Audio retained before speech onset; new streams only.

        Raises:
            ValueError: If an option falls outside the values the endpoint accepts.
        """
        opts = RealtimeSTTOptions(
            language=language if is_given(language) else self._opts.language,
            api_key=self._opts.api_key,
            stream_type=stream_type if is_given(stream_type) else self._opts.stream_type,
            mode=mode if is_given(mode) else self._opts.mode,
            endpointing=endpointing if is_given(endpointing) else self._opts.endpointing,
            encoding=self._opts.encoding,
            sample_rate=sample_rate if is_given(sample_rate) else self._opts.sample_rate,
            base_url=self._opts.base_url,
            prompt=prompt if is_given(prompt) else self._opts.prompt,
            return_timestamps=return_timestamps
            if is_given(return_timestamps)
            else self._opts.return_timestamps,
            vad_sot_threshold=vad_sot_threshold
            if is_given(vad_sot_threshold)
            else self._opts.vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms
            if is_given(vad_min_speech_ms)
            else self._opts.vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms
            if is_given(vad_min_silence_ms)
            else self._opts.vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms
            if is_given(vad_prefix_padding_ms)
            else self._opts.vad_prefix_padding_ms,
        )
        self._opts = opts
        # Forward the given fields only, so a stream created with a per-stream
        # override (e.g. `stream(language=...)`) keeps it through unrelated updates.
        for stream in self._streams:
            stream.update_options(
                language=language,
                stream_type=stream_type,
                mode=mode,
                endpointing=endpointing,
                sample_rate=sample_rate,
                prompt=prompt,
                return_timestamps=return_timestamps,
                vad_sot_threshold=vad_sot_threshold,
                vad_min_speech_ms=vad_min_speech_ms,
                vad_min_silence_ms=vad_min_silence_ms,
                vad_prefix_padding_ms=vad_prefix_padding_ms,
            )

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> RealtimeSpeechStream:
        """Create a new realtime speech stream.

        Args:
            language: Overrides the instance language for this stream only.
            conn_options: Connection options. ``max_retry`` is forced to ``0`` because this
                endpoint bills per connection and must not silently reconnect.

        Returns:
            A stream that accepts audio frames and yields speech events.
        """
        conn_options = replace(conn_options, max_retry=0)
        opts = RealtimeSTTOptions(
            language=language if is_given(language) else self._opts.language,
            api_key=self._opts.api_key,
            stream_type=self._opts.stream_type,
            mode=self._opts.mode,
            endpointing=self._opts.endpointing,
            encoding=self._opts.encoding,
            sample_rate=self._opts.sample_rate,
            base_url=self._opts.base_url,
            prompt=self._opts.prompt,
            return_timestamps=self._opts.return_timestamps,
            vad_sot_threshold=self._opts.vad_sot_threshold,
            vad_min_speech_ms=self._opts.vad_min_speech_ms,
            vad_min_silence_ms=self._opts.vad_min_silence_ms,
            vad_prefix_padding_ms=self._opts.vad_prefix_padding_ms,
        )
        stream = RealtimeSpeechStream(
            stt=self,
            opts=opts,
            conn_options=conn_options,
            http_session=self._ensure_session(),
        )
        self._streams.add(stream)
        return stream

    async def aclose(self) -> None:
        """Close every stream created by this instance and any owned HTTP session."""
        for stream in list(self._streams):
            await stream.aclose()
        self._streams.clear()
        if self._owns_session and self._session and not self._session.closed:
            await self._session.close()

Speech-to-text using Sarvam's realtime WebSocket endpoint (saaras:v3-realtime).

This endpoint streams interim and final transcripts over a single WebSocket connection and supports either server-side VAD or client-driven (manual) turn boundaries.

Create a Sarvam realtime STT instance.

Args

language

BCP-47 language code, or auto for adaptive language identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals: transcribe, translate, verbatim, translit, or codemix.

endpointing

vad for server-side turn detection, or manual when the caller delimits turns by flushing the stream.

encoding

Wire encoding: linear16, linear32, mulaw, or alaw.

sample_rate

Audio sample rate in Hz; 8000 or 16000.

prompt

Optional context or terminology hint used to bias decoding.

return_timestamps

Whether finals should carry segment-level start and end times.

api_key

Sarvam API key. Falls back to the SARVAM_API_KEY environment variable.

base_url

WebSocket URL of the realtime endpoint.

http_session

Optional aiohttp session to reuse for the connection.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Audio retained before speech onset in ms (vad endpointing only).

Raises

ValueError

If no API key is provided or found in the environment, or if an option falls outside the values the endpoint accepts.

Ancestors

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

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    """Name of the Sarvam realtime model backing this instance."""
    return REALTIME_MODEL

Name of the Sarvam realtime model backing this instance.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    """Name of the speech-to-text provider."""
    return "Sarvam"

Name of the speech-to-text provider.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close every stream created by this instance and any owned HTTP session."""
    for stream in list(self._streams):
        await stream.aclose()
    self._streams.clear()
    if self._owns_session and self._session and not self._session.closed:
        await self._session.close()

Close every stream created by this instance and any owned HTTP session.

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

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> RealtimeSpeechStream:
    """Create a new realtime speech stream.

    Args:
        language: Overrides the instance language for this stream only.
        conn_options: Connection options. ``max_retry`` is forced to ``0`` because this
            endpoint bills per connection and must not silently reconnect.

    Returns:
        A stream that accepts audio frames and yields speech events.
    """
    conn_options = replace(conn_options, max_retry=0)
    opts = RealtimeSTTOptions(
        language=language if is_given(language) else self._opts.language,
        api_key=self._opts.api_key,
        stream_type=self._opts.stream_type,
        mode=self._opts.mode,
        endpointing=self._opts.endpointing,
        encoding=self._opts.encoding,
        sample_rate=self._opts.sample_rate,
        base_url=self._opts.base_url,
        prompt=self._opts.prompt,
        return_timestamps=self._opts.return_timestamps,
        vad_sot_threshold=self._opts.vad_sot_threshold,
        vad_min_speech_ms=self._opts.vad_min_speech_ms,
        vad_min_silence_ms=self._opts.vad_min_silence_ms,
        vad_prefix_padding_ms=self._opts.vad_prefix_padding_ms,
    )
    stream = RealtimeSpeechStream(
        stt=self,
        opts=opts,
        conn_options=conn_options,
        http_session=self._ensure_session(),
    )
    self._streams.add(stream)
    return stream

Create a new realtime speech stream.

Args

language

Overrides the instance language for this stream only.

conn_options

Connection options. max_retry is forced to 0 because this endpoint bills per connection and must not silently reconnect.

Returns

A stream that accepts audio frames and yields speech events.

def update_options(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str | None] = NOT_GIVEN,
return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
    mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
    endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str | None] = NOT_GIVEN,
    return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
    vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
) -> None:
    """Update options for this instance and every stream it created.

    Options that Sarvam only accepts at connection time (``sample_rate``,
    ``return_timestamps``, and ``vad_prefix_padding_ms``) take effect on
    newly created streams only.
    The remaining options are sent to active streams as an in-band
    ``config.update``, and the boundary-gated ones apply from the next
    utterance boundary.

    Args:
        language: BCP-47 language code, or ``auto`` for adaptive language identification.
        stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
        mode: Task applied to finals.
        endpointing: ``vad`` for server-side turn detection, or ``manual``.
        sample_rate: Audio sample rate in Hz; applies to new streams only.
        prompt: Context or terminology hint; ``None`` clears it.
        return_timestamps: Segment-level timestamps; applies to new streams only.
        vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
        vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
        vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
        vad_prefix_padding_ms: Audio retained before speech onset; new streams only.

    Raises:
        ValueError: If an option falls outside the values the endpoint accepts.
    """
    opts = RealtimeSTTOptions(
        language=language if is_given(language) else self._opts.language,
        api_key=self._opts.api_key,
        stream_type=stream_type if is_given(stream_type) else self._opts.stream_type,
        mode=mode if is_given(mode) else self._opts.mode,
        endpointing=endpointing if is_given(endpointing) else self._opts.endpointing,
        encoding=self._opts.encoding,
        sample_rate=sample_rate if is_given(sample_rate) else self._opts.sample_rate,
        base_url=self._opts.base_url,
        prompt=prompt if is_given(prompt) else self._opts.prompt,
        return_timestamps=return_timestamps
        if is_given(return_timestamps)
        else self._opts.return_timestamps,
        vad_sot_threshold=vad_sot_threshold
        if is_given(vad_sot_threshold)
        else self._opts.vad_sot_threshold,
        vad_min_speech_ms=vad_min_speech_ms
        if is_given(vad_min_speech_ms)
        else self._opts.vad_min_speech_ms,
        vad_min_silence_ms=vad_min_silence_ms
        if is_given(vad_min_silence_ms)
        else self._opts.vad_min_silence_ms,
        vad_prefix_padding_ms=vad_prefix_padding_ms
        if is_given(vad_prefix_padding_ms)
        else self._opts.vad_prefix_padding_ms,
    )
    self._opts = opts
    # Forward the given fields only, so a stream created with a per-stream
    # override (e.g. `stream(language=...)`) keeps it through unrelated updates.
    for stream in self._streams:
        stream.update_options(
            language=language,
            stream_type=stream_type,
            mode=mode,
            endpointing=endpointing,
            sample_rate=sample_rate,
            prompt=prompt,
            return_timestamps=return_timestamps,
            vad_sot_threshold=vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms,
        )

Update options for this instance and every stream it created.

Options that Sarvam only accepts at connection time (sample_rate, return_timestamps, and vad_prefix_padding_ms) take effect on newly created streams only. The remaining options are sent to active streams as an in-band config.update, and the boundary-gated ones apply from the next utterance boundary.

Args

language

BCP-47 language code, or auto for adaptive language identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals.

endpointing

vad for server-side turn detection, or manual.

sample_rate

Audio sample rate in Hz; applies to new streams only.

prompt

Context or terminology hint; None clears it.

return_timestamps

Segment-level timestamps; applies to new streams only.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Audio retained before speech onset; new streams only.

Raises

ValueError

If an option falls outside the values the endpoint accepts.

class STTStreaming (*,
language: str = 'en-IN',
stream_type: RealtimeStreamType | str = 'balanced',
mode: RealtimeMode | str = 'transcribe',
endpointing: RealtimeEndpointing | str = 'vad',
encoding: RealtimeEncoding | str = 'linear16',
sample_rate: int = 16000,
prompt: str | None = None,
return_timestamps: bool = False,
api_key: str | None = None,
base_url: str = 'wss://api.sarvam.ai/speech-to-text-realtime/ws',
http_session: aiohttp.ClientSession | None = None,
vad_sot_threshold: float | None = None,
vad_min_speech_ms: int | None = None,
vad_min_silence_ms: int | None = None,
vad_prefix_padding_ms: int | None = None)

Expand source code

class STTRealtime(stt.STT):
    """Speech-to-text using Sarvam's realtime WebSocket endpoint (``saaras:v3-realtime``).

    This endpoint streams interim and final transcripts over a single
    WebSocket connection and supports either server-side VAD or
    client-driven (manual) turn boundaries.
    """

    def __init__(
        self,
        *,
        language: str = "en-IN",
        stream_type: RealtimeStreamType | str = "balanced",
        mode: RealtimeMode | str = "transcribe",
        endpointing: RealtimeEndpointing | str = "vad",
        encoding: RealtimeEncoding | str = "linear16",
        sample_rate: int = 16000,
        prompt: str | None = None,
        return_timestamps: bool = False,
        api_key: str | None = None,
        base_url: str = SARVAM_STT_REALTIME_URL,
        http_session: aiohttp.ClientSession | None = None,
        vad_sot_threshold: float | None = None,
        vad_min_speech_ms: int | None = None,
        vad_min_silence_ms: int | None = None,
        vad_prefix_padding_ms: int | None = None,
    ) -> None:
        """Create a Sarvam realtime STT instance.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive language identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals: ``transcribe``, ``translate``, ``verbatim``,
                ``translit``, or ``codemix``.
            endpointing: ``vad`` for server-side turn detection, or ``manual`` when the
                caller delimits turns by flushing the stream.
            encoding: Wire encoding: ``linear16``, ``linear32``, ``mulaw``, or ``alaw``.
            sample_rate: Audio sample rate in Hz; ``8000`` or ``16000``.
            prompt: Optional context or terminology hint used to bias decoding.
            return_timestamps: Whether finals should carry segment-level start and end times.
            api_key: Sarvam API key. Falls back to the ``SARVAM_API_KEY`` environment variable.
            base_url: WebSocket URL of the realtime endpoint.
            http_session: Optional aiohttp session to reuse for the connection.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Audio retained before speech onset in ms
                (``vad`` endpointing only).

        Raises:
            ValueError: If no API key is provided or found in the environment, or if an
                option falls outside the values the endpoint accepts.
        """
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                aligned_transcript=False,
                offline_recognize=False,
            )
        )

        api_key = api_key or os.environ.get("SARVAM_API_KEY")
        if not api_key:
            raise ValueError(
                "Sarvam API key is required. "
                "Provide it directly or set SARVAM_API_KEY environment variable."
            )

        self._opts = RealtimeSTTOptions(
            language=language,
            api_key=api_key,
            stream_type=stream_type,
            mode=mode,
            endpointing=endpointing,
            encoding=encoding,
            sample_rate=sample_rate,
            base_url=base_url,
            prompt=prompt,
            return_timestamps=return_timestamps,
            vad_sot_threshold=vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms,
        )
        self._session = http_session
        self._owns_session = http_session is None
        self._streams = weakref.WeakSet[RealtimeSpeechStream]()

    @property
    def model(self) -> str:
        """Name of the Sarvam realtime model backing this instance."""
        return REALTIME_MODEL

    @property
    def provider(self) -> str:
        """Name of the speech-to-text provider."""
        return "Sarvam"

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            try:
                self._session = utils.http_context.http_session()
                self._owns_session = False
            except RuntimeError:
                self._session = aiohttp.ClientSession()
                self._owns_session = True
        return self._session

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

    def update_options(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
        mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
        endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str | None] = NOT_GIVEN,
        return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
        vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
        vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
    ) -> None:
        """Update options for this instance and every stream it created.

        Options that Sarvam only accepts at connection time (``sample_rate``,
        ``return_timestamps``, and ``vad_prefix_padding_ms``) take effect on
        newly created streams only.
        The remaining options are sent to active streams as an in-band
        ``config.update``, and the boundary-gated ones apply from the next
        utterance boundary.

        Args:
            language: BCP-47 language code, or ``auto`` for adaptive language identification.
            stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
            mode: Task applied to finals.
            endpointing: ``vad`` for server-side turn detection, or ``manual``.
            sample_rate: Audio sample rate in Hz; applies to new streams only.
            prompt: Context or terminology hint; ``None`` clears it.
            return_timestamps: Segment-level timestamps; applies to new streams only.
            vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
            vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
            vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
            vad_prefix_padding_ms: Audio retained before speech onset; new streams only.

        Raises:
            ValueError: If an option falls outside the values the endpoint accepts.
        """
        opts = RealtimeSTTOptions(
            language=language if is_given(language) else self._opts.language,
            api_key=self._opts.api_key,
            stream_type=stream_type if is_given(stream_type) else self._opts.stream_type,
            mode=mode if is_given(mode) else self._opts.mode,
            endpointing=endpointing if is_given(endpointing) else self._opts.endpointing,
            encoding=self._opts.encoding,
            sample_rate=sample_rate if is_given(sample_rate) else self._opts.sample_rate,
            base_url=self._opts.base_url,
            prompt=prompt if is_given(prompt) else self._opts.prompt,
            return_timestamps=return_timestamps
            if is_given(return_timestamps)
            else self._opts.return_timestamps,
            vad_sot_threshold=vad_sot_threshold
            if is_given(vad_sot_threshold)
            else self._opts.vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms
            if is_given(vad_min_speech_ms)
            else self._opts.vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms
            if is_given(vad_min_silence_ms)
            else self._opts.vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms
            if is_given(vad_prefix_padding_ms)
            else self._opts.vad_prefix_padding_ms,
        )
        self._opts = opts
        # Forward the given fields only, so a stream created with a per-stream
        # override (e.g. `stream(language=...)`) keeps it through unrelated updates.
        for stream in self._streams:
            stream.update_options(
                language=language,
                stream_type=stream_type,
                mode=mode,
                endpointing=endpointing,
                sample_rate=sample_rate,
                prompt=prompt,
                return_timestamps=return_timestamps,
                vad_sot_threshold=vad_sot_threshold,
                vad_min_speech_ms=vad_min_speech_ms,
                vad_min_silence_ms=vad_min_silence_ms,
                vad_prefix_padding_ms=vad_prefix_padding_ms,
            )

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> RealtimeSpeechStream:
        """Create a new realtime speech stream.

        Args:
            language: Overrides the instance language for this stream only.
            conn_options: Connection options. ``max_retry`` is forced to ``0`` because this
                endpoint bills per connection and must not silently reconnect.

        Returns:
            A stream that accepts audio frames and yields speech events.
        """
        conn_options = replace(conn_options, max_retry=0)
        opts = RealtimeSTTOptions(
            language=language if is_given(language) else self._opts.language,
            api_key=self._opts.api_key,
            stream_type=self._opts.stream_type,
            mode=self._opts.mode,
            endpointing=self._opts.endpointing,
            encoding=self._opts.encoding,
            sample_rate=self._opts.sample_rate,
            base_url=self._opts.base_url,
            prompt=self._opts.prompt,
            return_timestamps=self._opts.return_timestamps,
            vad_sot_threshold=self._opts.vad_sot_threshold,
            vad_min_speech_ms=self._opts.vad_min_speech_ms,
            vad_min_silence_ms=self._opts.vad_min_silence_ms,
            vad_prefix_padding_ms=self._opts.vad_prefix_padding_ms,
        )
        stream = RealtimeSpeechStream(
            stt=self,
            opts=opts,
            conn_options=conn_options,
            http_session=self._ensure_session(),
        )
        self._streams.add(stream)
        return stream

    async def aclose(self) -> None:
        """Close every stream created by this instance and any owned HTTP session."""
        for stream in list(self._streams):
            await stream.aclose()
        self._streams.clear()
        if self._owns_session and self._session and not self._session.closed:
            await self._session.close()

Speech-to-text using Sarvam's realtime WebSocket endpoint (saaras:v3-realtime).

This endpoint streams interim and final transcripts over a single WebSocket connection and supports either server-side VAD or client-driven (manual) turn boundaries.

Create a Sarvam realtime STT instance.

Args

language

BCP-47 language code, or auto for adaptive language identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals: transcribe, translate, verbatim, translit, or codemix.

endpointing

vad for server-side turn detection, or manual when the caller delimits turns by flushing the stream.

encoding

Wire encoding: linear16, linear32, mulaw, or alaw.

sample_rate

Audio sample rate in Hz; 8000 or 16000.

prompt

Optional context or terminology hint used to bias decoding.

return_timestamps

Whether finals should carry segment-level start and end times.

api_key

Sarvam API key. Falls back to the SARVAM_API_KEY environment variable.

base_url

WebSocket URL of the realtime endpoint.

http_session

Optional aiohttp session to reuse for the connection.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Audio retained before speech onset in ms (vad endpointing only).

Raises

ValueError

If no API key is provided or found in the environment, or if an option falls outside the values the endpoint accepts.

Ancestors

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

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    """Name of the Sarvam realtime model backing this instance."""
    return REALTIME_MODEL

Name of the Sarvam realtime model backing this instance.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    """Name of the speech-to-text provider."""
    return "Sarvam"

Name of the speech-to-text provider.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close every stream created by this instance and any owned HTTP session."""
    for stream in list(self._streams):
        await stream.aclose()
    self._streams.clear()
    if self._owns_session and self._session and not self._session.closed:
        await self._session.close()

Close every stream created by this instance and any owned HTTP session.

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

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> RealtimeSpeechStream:
    """Create a new realtime speech stream.

    Args:
        language: Overrides the instance language for this stream only.
        conn_options: Connection options. ``max_retry`` is forced to ``0`` because this
            endpoint bills per connection and must not silently reconnect.

    Returns:
        A stream that accepts audio frames and yields speech events.
    """
    conn_options = replace(conn_options, max_retry=0)
    opts = RealtimeSTTOptions(
        language=language if is_given(language) else self._opts.language,
        api_key=self._opts.api_key,
        stream_type=self._opts.stream_type,
        mode=self._opts.mode,
        endpointing=self._opts.endpointing,
        encoding=self._opts.encoding,
        sample_rate=self._opts.sample_rate,
        base_url=self._opts.base_url,
        prompt=self._opts.prompt,
        return_timestamps=self._opts.return_timestamps,
        vad_sot_threshold=self._opts.vad_sot_threshold,
        vad_min_speech_ms=self._opts.vad_min_speech_ms,
        vad_min_silence_ms=self._opts.vad_min_silence_ms,
        vad_prefix_padding_ms=self._opts.vad_prefix_padding_ms,
    )
    stream = RealtimeSpeechStream(
        stt=self,
        opts=opts,
        conn_options=conn_options,
        http_session=self._ensure_session(),
    )
    self._streams.add(stream)
    return stream

Create a new realtime speech stream.

Args

language

Overrides the instance language for this stream only.

conn_options

Connection options. max_retry is forced to 0 because this endpoint bills per connection and must not silently reconnect.

Returns

A stream that accepts audio frames and yields speech events.

def update_options(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str | None] = NOT_GIVEN,
return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    stream_type: NotGivenOr[RealtimeStreamType | str] = NOT_GIVEN,
    mode: NotGivenOr[RealtimeMode | str] = NOT_GIVEN,
    endpointing: NotGivenOr[RealtimeEndpointing | str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str | None] = NOT_GIVEN,
    return_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    vad_sot_threshold: NotGivenOr[float | None] = NOT_GIVEN,
    vad_min_speech_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_min_silence_ms: NotGivenOr[int | None] = NOT_GIVEN,
    vad_prefix_padding_ms: NotGivenOr[int | None] = NOT_GIVEN,
) -> None:
    """Update options for this instance and every stream it created.

    Options that Sarvam only accepts at connection time (``sample_rate``,
    ``return_timestamps``, and ``vad_prefix_padding_ms``) take effect on
    newly created streams only.
    The remaining options are sent to active streams as an in-band
    ``config.update``, and the boundary-gated ones apply from the next
    utterance boundary.

    Args:
        language: BCP-47 language code, or ``auto`` for adaptive language identification.
        stream_type: Latency profile: ``fast``, ``balanced``, or ``simulated``.
        mode: Task applied to finals.
        endpointing: ``vad`` for server-side turn detection, or ``manual``.
        sample_rate: Audio sample rate in Hz; applies to new streams only.
        prompt: Context or terminology hint; ``None`` clears it.
        return_timestamps: Segment-level timestamps; applies to new streams only.
        vad_sot_threshold: VAD activation threshold (``vad`` endpointing only).
        vad_min_speech_ms: Minimum speech duration in ms (``vad`` endpointing only).
        vad_min_silence_ms: End-of-turn silence in ms (``vad`` endpointing only).
        vad_prefix_padding_ms: Audio retained before speech onset; new streams only.

    Raises:
        ValueError: If an option falls outside the values the endpoint accepts.
    """
    opts = RealtimeSTTOptions(
        language=language if is_given(language) else self._opts.language,
        api_key=self._opts.api_key,
        stream_type=stream_type if is_given(stream_type) else self._opts.stream_type,
        mode=mode if is_given(mode) else self._opts.mode,
        endpointing=endpointing if is_given(endpointing) else self._opts.endpointing,
        encoding=self._opts.encoding,
        sample_rate=sample_rate if is_given(sample_rate) else self._opts.sample_rate,
        base_url=self._opts.base_url,
        prompt=prompt if is_given(prompt) else self._opts.prompt,
        return_timestamps=return_timestamps
        if is_given(return_timestamps)
        else self._opts.return_timestamps,
        vad_sot_threshold=vad_sot_threshold
        if is_given(vad_sot_threshold)
        else self._opts.vad_sot_threshold,
        vad_min_speech_ms=vad_min_speech_ms
        if is_given(vad_min_speech_ms)
        else self._opts.vad_min_speech_ms,
        vad_min_silence_ms=vad_min_silence_ms
        if is_given(vad_min_silence_ms)
        else self._opts.vad_min_silence_ms,
        vad_prefix_padding_ms=vad_prefix_padding_ms
        if is_given(vad_prefix_padding_ms)
        else self._opts.vad_prefix_padding_ms,
    )
    self._opts = opts
    # Forward the given fields only, so a stream created with a per-stream
    # override (e.g. `stream(language=...)`) keeps it through unrelated updates.
    for stream in self._streams:
        stream.update_options(
            language=language,
            stream_type=stream_type,
            mode=mode,
            endpointing=endpointing,
            sample_rate=sample_rate,
            prompt=prompt,
            return_timestamps=return_timestamps,
            vad_sot_threshold=vad_sot_threshold,
            vad_min_speech_ms=vad_min_speech_ms,
            vad_min_silence_ms=vad_min_silence_ms,
            vad_prefix_padding_ms=vad_prefix_padding_ms,
        )

Update options for this instance and every stream it created.

Options that Sarvam only accepts at connection time (sample_rate, return_timestamps, and vad_prefix_padding_ms) take effect on newly created streams only. The remaining options are sent to active streams as an in-band config.update, and the boundary-gated ones apply from the next utterance boundary.

Args

language

BCP-47 language code, or auto for adaptive language identification.

stream_type

Latency profile: fast, balanced, or simulated.

mode

Task applied to finals.

endpointing

vad for server-side turn detection, or manual.

sample_rate

Audio sample rate in Hz; applies to new streams only.

prompt

Context or terminology hint; None clears it.

return_timestamps

Segment-level timestamps; applies to new streams only.

vad_sot_threshold

VAD activation threshold (vad endpointing only).

vad_min_speech_ms

Minimum speech duration in ms (vad endpointing only).

vad_min_silence_ms

End-of-turn silence in ms (vad endpointing only).

vad_prefix_padding_ms

Audio retained before speech onset; new streams only.

Raises

ValueError

If an option falls outside the values the endpoint accepts.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class TTS (*,
target_language_code: SarvamTTSLanguages | str = 'en-IN',
model: SarvamTTSModels | str = 'bulbul:v3',
speaker: SarvamTTSSpeakers | str | None = None,
speech_sample_rate: int = 22050,
num_channels: int = 1,
pitch: float = 0.0,
pace: float = 1.0,
loudness: float = 1.0,
temperature: float = 0.6,
output_audio_bitrate: SarvamTTSOutputAudioBitrate | str = '128k',
min_buffer_size: int = 50,
max_chunk_length: int = 150,
enable_preprocessing: bool = False,
dict_id: str | None = None,
enable_cached_responses: bool | None = None,
api_key: str | None = None,
base_url: str = 'https://api.sarvam.ai/text-to-speech',
ws_url: str = 'wss://api.sarvam.ai/text-to-speech/ws',
http_session: aiohttp.ClientSession | None = None,
send_completion_event: bool = True,
output_audio_codec: str = 'mp3')

Expand source code

class TTS(tts.TTS):
    """Sarvam.ai Text-to-Speech implementation.

    This class provides text-to-speech functionality using the Sarvam.ai API.
    Sarvam.ai specializes in high-quality TTS for Indian languages.

    Args:
        target_language_code: BCP-47 language code for supported Indian languages
        model: Sarvam TTS model to use (bulbul:v2)
        speaker: Voice to use for synthesis
        speech_sample_rate: Audio sample rate in Hz
        num_channels: Number of audio channels (Sarvam outputs mono)
        pitch: Voice pitch adjustment (-0.75 to 0.75) - only supported in v2 for now
        pace: Speech rate multiplier (0.3 to 3.0)
        loudness: Volume multiplier (0.5 to 2.0) - only supported in v2 for now
        temperature: Sampling temperature (0.01 to 2.0), only used in v3 and v3-beta
        dict_id: Custom pronunciation dictionary ID (bulbul:v3 only)
        enable_cached_responses: Enable response caching beta feature (bulbul:v1/v2 only)
        output_audio_bitrate: Output audio bitrate (default 128k)
        min_buffer_size: Minimum character length for flushing (30 to 200)
        max_chunk_length: Maximum chunk length for sentence splitting (50 to 500)
        enable_preprocessing: Whether to use text preprocessing
        api_key: Sarvam.ai API key (required)
        base_url: API endpoint URL
        ws_url: WebSocket endpoint URL
        http_session: Optional aiohttp session to use
        output_audio_codec: Optionally choose the output codec format (mp3)
    """

    def __init__(
        self,
        *,
        target_language_code: SarvamTTSLanguages | str = "en-IN",
        model: SarvamTTSModels | str = "bulbul:v3",
        speaker: SarvamTTSSpeakers | str | None = None,
        speech_sample_rate: int = 22050,
        num_channels: int = 1,  # Sarvam output is mono WAV
        pitch: float = 0.0,
        pace: float = 1.0,
        loudness: float = 1.0,
        temperature: float = 0.6,
        output_audio_bitrate: SarvamTTSOutputAudioBitrate | str = "128k",
        min_buffer_size: int = 50,
        max_chunk_length: int = 150,
        enable_preprocessing: bool = False,
        dict_id: str | None = None,
        enable_cached_responses: bool | None = None,
        api_key: str | None = None,
        base_url: str = SARVAM_TTS_BASE_URL,
        ws_url: str = SARVAM_TTS_WS_URL,
        http_session: aiohttp.ClientSession | None = None,
        send_completion_event: bool = True,
        output_audio_codec: str = "mp3",
    ) -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=True),
            sample_rate=speech_sample_rate,
            num_channels=num_channels,
        )

        self._api_key = api_key or os.environ.get("SARVAM_API_KEY")
        if not self._api_key:
            raise ValueError(
                "Sarvam API key is required. Provide it directly or set SARVAM_API_KEY env var."
            )

        # Validate inputs early
        if not target_language_code or not target_language_code.strip():
            raise ValueError("Target language code is required and cannot be empty")
        if not model or not model.strip():
            raise ValueError("Model is required and cannot be empty")
        if speaker is None:
            # speaker = "shubh"
            if model == "bulbul:v3-beta" or model == "bulbul:v3":
                speaker = "shubh"
            else:
                speaker = "anushka"

        # Validate parameter ranges
        if not -0.75 <= pitch <= 0.75:
            logger.warning(
                "pitch value %.2f is outside the Sarvam API accepted range [-0.75, 0.75]; "
                "clamping to nearest bound. Please update your code.",
                pitch,
            )
            pitch = max(-0.75, min(0.75, pitch))
        if not 0.3 <= pace <= 3.0:
            raise ValueError("Pace must be between 0.3 and 3.0")
        if not 0.5 <= loudness <= 2.0:
            raise ValueError("Loudness must be between 0.5 and 2.0")
        if not 0.01 <= temperature <= 2.0:
            raise ValueError("Temperature must be between 0.01 and 2.0")
        if output_audio_bitrate not in ALLOWED_OUTPUT_AUDIO_BITRATES:
            raise ValueError(
                f"output_audio_bitrate must be one of {', '.join(sorted(ALLOWED_OUTPUT_AUDIO_BITRATES))}"
            )
        if not 30 <= min_buffer_size <= 200:
            raise ValueError("min_buffer_size must be between 30 and 200")
        if not 50 <= max_chunk_length <= 500:
            raise ValueError("max_chunk_length must be between 50 and 500")
        if speech_sample_rate not in [8000, 16000, 22050, 24000, 32000, 44100, 48000]:
            raise ValueError(
                "Sample rate must be one of 8000, 16000, 22050, 24000, 32000, 44100, or 48000 Hz"
            )
        if output_audio_codec not in ALLOWED_OUTPUT_AUDIO_CODECS:
            raise ValueError(
                f"output_audio_codec must be one of {','.join(sorted(ALLOWED_OUTPUT_AUDIO_CODECS))}"
            )

        # Validate model-speaker compatibility
        if not validate_model_speaker_compatibility(model, speaker):
            compatible_speakers = MODEL_SPEAKER_COMPATIBILITY.get(model, {}).get("all", [])
            raise ValueError(
                f"Speaker '{speaker}' is not compatible with model '{model}'. "
                f"Please choose a compatible speaker from: {', '.join(compatible_speakers)}"
            )

        # Initialize word tokenizer for streaming
        word_tokenizer = tokenize.basic.SentenceTokenizer()

        self._opts = SarvamTTSOptions(
            target_language_code=LanguageCode(target_language_code),
            model=model,
            speaker=speaker,
            speech_sample_rate=speech_sample_rate,
            pitch=pitch,
            pace=pace,
            loudness=loudness,
            temperature=temperature,
            output_audio_bitrate=output_audio_bitrate,
            min_buffer_size=min_buffer_size,
            max_chunk_length=max_chunk_length,
            enable_preprocessing=enable_preprocessing,
            dict_id=dict_id,
            enable_cached_responses=enable_cached_responses,
            api_key=self._api_key,
            base_url=base_url,
            ws_url=ws_url,
            word_tokenizer=word_tokenizer,
            send_completion_event=send_completion_event,
            output_audio_codec=output_audio_codec,
        )
        self._session = http_session
        self._streams = weakref.WeakSet[SynthesizeStream]()
        # Maps id(ws) -> background keepalive task that pings the server while
        # the connection sits idle in the pool. Sarvam closes idle connections
        # after 60 s; pinging every 30 s keeps them alive for reuse.
        self._ws_keepalive_tasks: dict[int, asyncio.Task[None]] = {}

        self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse](
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
            max_session_duration=3600,  # 1 hour
            mark_refreshed_on_get=False,
        )

    async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
        session = self._ensure_session()
        headers = {
            "api-subscription-key": self._opts.api_key,
            "User-Agent": USER_AGENT,
            "Accept": "*/*",
            "Accept-Encoding": "gzip, deflate, br",
        }
        # Add model parameter to URL like the client does
        ws_url = f"{self._opts.ws_url}?model={self._opts.model}&send_completion_event={self._opts.send_completion_event}"

        logger.info("Connecting to Sarvam TTS WebSocket")

        try:
            ws = await asyncio.wait_for(
                session.ws_connect(
                    ws_url,
                    headers=headers,
                    # Send protocol-level WebSocket PING frames every
                    # ``_WS_HEARTBEAT_INTERVAL`` seconds. aiohttp handles the
                    # PONG accounting on its own and will close the connection
                    # locally if the server stops responding -- this is what
                    # actually keeps the TCP connection alive while it sits
                    # idle in the pool (no ``receive()`` call to auto-pong).
                    heartbeat=_WS_HEARTBEAT_INTERVAL,
                ),
                timeout,
            )
        except Exception as e:
            logger.error(
                "Failed to connect to Sarvam TTS WebSocket",
                extra={"error": str(e), "url": ws_url},
                exc_info=True,
            )
            raise APIConnectionError(f"WebSocket connection failed: {e}") from e

        self._start_keepalive(ws)
        return ws

    async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        await self._stop_keepalive(ws)
        await ws.close()

    def _start_keepalive(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        """Spawn a background task that keeps ``ws`` alive with periodic pings.

        Idempotent: if a live keepalive task is already registered for ``ws``
        the call is a no-op. Callers that need a fresh task must invoke
        ``_stop_keepalive`` first.
        """
        if _KEEPALIVE_INTERVAL <= 0:
            return
        existing = self._ws_keepalive_tasks.get(id(ws))
        if existing is not None and not existing.done():
            return
        task = asyncio.create_task(self._keepalive_loop(ws), name="sarvam-tts-ws-keepalive")
        self._ws_keepalive_tasks[id(ws)] = task

    async def _stop_keepalive(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        """Cancel the keepalive task associated with ``ws`` (if any)."""
        task = self._ws_keepalive_tasks.pop(id(ws), None)
        if task is None or task.done():
            return
        task.cancel()
        with contextlib.suppress(BaseException):
            await task

    async def _keepalive_loop(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        """Keep a pooled WebSocket alive while it sits idle.

        This loop does two things in lockstep:

        1. Actively calls ``ws.receive()`` so that aiohttp can process
           server-initiated PONG frames (and any other messages). aiohttp
           only resets its internal "PONG not received" timer when a PONG
           is read via ``receive()`` -- without an active reader, the
           protocol-level heartbeat will tear the connection down even
           though the server is happily replying.

        2. Sends a Sarvam-defined ``{"type": "ping"}`` JSON message
           whenever ``receive()`` times out (i.e. nothing has come in for
           ``_KEEPALIVE_INTERVAL`` seconds). This resets Sarvam's
           server-side application idle timer (documented as 60 s).

        Any CLOSE/CLOSED/CLOSING/ERROR message from the server, or any
        write failure, evicts the connection from the pool so the next
        checkout creates a fresh one instead of handing the dead one out.
        """
        try:
            while not ws.closed:
                try:
                    msg = await asyncio.wait_for(ws.receive(), timeout=_KEEPALIVE_INTERVAL)
                except asyncio.TimeoutError:
                    # No incoming message within the interval -- send our
                    # app-level ping to reset Sarvam's idle timer.
                    if ws.closed:
                        return
                    try:
                        await ws.send_str(json.dumps({"type": "ping"}))
                    except asyncio.CancelledError:
                        raise
                    except Exception as e:
                        logger.debug(
                            "Sarvam TTS keepalive ping failed; evicting connection from pool",
                            extra={"error": str(e)},
                        )
                        with contextlib.suppress(Exception):
                            self._pool.remove(ws)
                        return
                    continue

                # We received something. CLOSE/CLOSED/CLOSING/ERROR means
                # the server tore the connection down -- evict and exit.
                if msg.type in (
                    aiohttp.WSMsgType.CLOSE,
                    aiohttp.WSMsgType.CLOSED,
                    aiohttp.WSMsgType.CLOSING,
                    aiohttp.WSMsgType.ERROR,
                ):
                    logger.debug(
                        "Sarvam TTS WebSocket closed while idle in pool; evicting connection",
                        extra={
                            "msg_type": str(msg.type),
                            "close_code": ws.close_code,
                        },
                    )
                    with contextlib.suppress(Exception):
                        self._pool.remove(ws)
                    return
                # Otherwise -- PONG, TEXT, BINARY, etc. -- discard. We are
                # idle in the pool so any unsolicited TTS traffic from a
                # previous request is no longer relevant. The act of
                # receiving has already reset aiohttp's heartbeat.
        except asyncio.CancelledError:
            pass

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

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

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()
        return self._session

    def update_options(
        self,
        *,
        model: str | None = None,
        target_language_code: SarvamTTSLanguages | str | None = None,
        speaker: str | None = None,
        pitch: float | None = None,
        pace: float | None = None,
        loudness: float | None = None,
        temperature: float | None = None,
        output_audio_bitrate: SarvamTTSOutputAudioBitrate | str | None = None,
        min_buffer_size: int | None = None,
        max_chunk_length: int | None = None,
        enable_preprocessing: bool | None = None,
        dict_id: str | None = None,
        enable_cached_responses: bool | None = None,
        send_completion_event: bool | None = None,
        output_audio_codec: str | None = None,
    ) -> None:
        """Update TTS options with validation."""
        if target_language_code is not None:
            if not target_language_code.strip():
                raise ValueError("Target language code cannot be empty")
            self._opts.target_language_code = LanguageCode(target_language_code)

        if model is not None:
            if not model.strip():
                raise ValueError("Model cannot be empty")
            self._opts.model = model
            if speaker is None and self._opts.speaker is not None:
                if not validate_model_speaker_compatibility(self._opts.model, self._opts.speaker):
                    compatible_speakers = MODEL_SPEAKER_COMPATIBILITY.get(self._opts.model, {}).get(
                        "all", []
                    )
                    raise ValueError(
                        f"Speaker '{self._opts.speaker}' incompatible with {self._opts.model}. "
                        f"Compatible speakers: {', '.join(compatible_speakers)}"
                    )
        if speaker is not None:
            if not speaker.strip():
                raise ValueError("Speaker cannot be empty")
            if not validate_model_speaker_compatibility(self._opts.model, speaker):
                compatible_speakers = MODEL_SPEAKER_COMPATIBILITY.get(self._opts.model, {}).get(
                    "all", []
                )
                raise ValueError(
                    f"Speaker '{speaker}' incompatible with {self._opts.model}. "
                    f"Compatible speakers: {', '.join(compatible_speakers)}"
                )
            self._opts.speaker = speaker

        if pitch is not None:
            if not -0.75 <= pitch <= 0.75:
                logger.warning(
                    "pitch value %.2f is outside the Sarvam API accepted range [-0.75, 0.75]; "
                    "clamping to nearest bound. Please update your code.",
                    pitch,
                )
                pitch = max(-0.75, min(0.75, pitch))
            self._opts.pitch = pitch

        if pace is not None:
            if not 0.3 <= pace <= 3.0:
                raise ValueError("Pace must be between 0.3 and 3.0")
            self._opts.pace = pace

        if loudness is not None:
            if not 0.5 <= loudness <= 2.0:
                raise ValueError("Loudness must be between 0.5 and 2.0")
            self._opts.loudness = loudness

        if temperature is not None:
            if not 0.01 <= temperature <= 2.0:
                raise ValueError("Temperature must be between 0.01 and 2.0")
            self._opts.temperature = temperature

        if output_audio_bitrate is not None:
            if output_audio_bitrate not in ALLOWED_OUTPUT_AUDIO_BITRATES:
                raise ValueError(
                    "output_audio_bitrate must be one of "
                    f"{', '.join(sorted(ALLOWED_OUTPUT_AUDIO_BITRATES))}"
                )
            self._opts.output_audio_bitrate = output_audio_bitrate

        if min_buffer_size is not None:
            if not 30 <= min_buffer_size <= 200:
                raise ValueError("min_buffer_size must be between 30 and 200")
            self._opts.min_buffer_size = min_buffer_size

        if max_chunk_length is not None:
            if not 50 <= max_chunk_length <= 500:
                raise ValueError("max_chunk_length must be between 50 and 500")
            self._opts.max_chunk_length = max_chunk_length

        if enable_preprocessing is not None:
            self._opts.enable_preprocessing = enable_preprocessing

        if dict_id is not None:
            self._opts.dict_id = dict_id

        if enable_cached_responses is not None:
            self._opts.enable_cached_responses = enable_cached_responses

        if send_completion_event is not None:
            self._opts.send_completion_event = send_completion_event

        if output_audio_codec is not None:
            if output_audio_codec not in ALLOWED_OUTPUT_AUDIO_CODECS:
                raise ValueError(
                    "output_audio_codec must be one of "
                    f"{','.join(sorted(ALLOWED_OUTPUT_AUDIO_CODECS))}"
                )
            self._opts.output_audio_codec = output_audio_codec

    # Implement the abstract synthesize method
    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions | None = None
    ) -> ChunkedStream:
        """Synthesize text to audio using Sarvam.ai TTS API."""
        if conn_options is None:
            conn_options = DEFAULT_API_CONNECT_OPTIONS
        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    def stream(
        self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> SynthesizeStream:
        """Create a streaming TTS session."""
        stream = SynthesizeStream(tts=self, conn_options=conn_options)
        self._streams.add(stream)
        return stream

    def prewarm(self) -> None:
        """Prewarm WebSocket connections."""
        self._pool.prewarm()

    async def aclose(self) -> None:
        """Close all active streams and connections."""
        for stream in list(self._streams):
            await stream.aclose()

        self._streams.clear()
        await self._pool.aclose()

Sarvam.ai Text-to-Speech implementation.

This class provides text-to-speech functionality using the Sarvam.ai API. Sarvam.ai specializes in high-quality TTS for Indian languages.

Args

target_language_code

BCP-47 language code for supported Indian languages

model

Sarvam TTS model to use (bulbul:v2)

speaker

Voice to use for synthesis

speech_sample_rate

Audio sample rate in Hz

num_channels

Number of audio channels (Sarvam outputs mono)

pitch

Voice pitch adjustment (-0.75 to 0.75) - only supported in v2 for now

pace

Speech rate multiplier (0.3 to 3.0)

loudness

Volume multiplier (0.5 to 2.0) - only supported in v2 for now

temperature

Sampling temperature (0.01 to 2.0), only used in v3 and v3-beta

dict_id

Custom pronunciation dictionary ID (bulbul:v3 only)

enable_cached_responses

Enable response caching beta feature (bulbul:v1/v2 only)

output_audio_bitrate

Output audio bitrate (default 128k)

min_buffer_size

Minimum character length for flushing (30 to 200)

max_chunk_length

Maximum chunk length for sentence splitting (50 to 500)

enable_preprocessing

Whether to use text preprocessing

api_key

Sarvam.ai API key (required)

base_url

API endpoint URL

ws_url

WebSocket endpoint URL

http_session

Optional aiohttp session to use

output_audio_codec

Optionally choose the output codec format (mp3)

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

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 all active streams and connections."""
    for stream in list(self._streams):
        await stream.aclose()

    self._streams.clear()
    await self._pool.aclose()

Close all active streams and connections.

def prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    """Prewarm WebSocket connections."""
    self._pool.prewarm()

Prewarm WebSocket connections.

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

Expand source code

def stream(
    self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
) -> SynthesizeStream:
    """Create a streaming TTS session."""
    stream = SynthesizeStream(tts=self, conn_options=conn_options)
    self._streams.add(stream)
    return stream

Create a streaming TTS session.

def synthesize(self, text: str, *, conn_options: APIConnectOptions | None = None) ‑> livekit.plugins.sarvam.tts.ChunkedStream

Expand source code

def synthesize(
    self, text: str, *, conn_options: APIConnectOptions | None = None
) -> ChunkedStream:
    """Synthesize text to audio using Sarvam.ai TTS API."""
    if conn_options is None:
        conn_options = DEFAULT_API_CONNECT_OPTIONS
    return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

Synthesize text to audio using Sarvam.ai TTS API.

def update_options(self,
*,
model: str | None = None,
target_language_code: SarvamTTSLanguages | str | None = None,
speaker: str | None = None,
pitch: float | None = None,
pace: float | None = None,
loudness: float | None = None,
temperature: float | None = None,
output_audio_bitrate: SarvamTTSOutputAudioBitrate | str | None = None,
min_buffer_size: int | None = None,
max_chunk_length: int | None = None,
enable_preprocessing: bool | None = None,
dict_id: str | None = None,
enable_cached_responses: bool | None = None,
send_completion_event: bool | None = None,
output_audio_codec: str | None = None) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: str | None = None,
    target_language_code: SarvamTTSLanguages | str | None = None,
    speaker: str | None = None,
    pitch: float | None = None,
    pace: float | None = None,
    loudness: float | None = None,
    temperature: float | None = None,
    output_audio_bitrate: SarvamTTSOutputAudioBitrate | str | None = None,
    min_buffer_size: int | None = None,
    max_chunk_length: int | None = None,
    enable_preprocessing: bool | None = None,
    dict_id: str | None = None,
    enable_cached_responses: bool | None = None,
    send_completion_event: bool | None = None,
    output_audio_codec: str | None = None,
) -> None:
    """Update TTS options with validation."""
    if target_language_code is not None:
        if not target_language_code.strip():
            raise ValueError("Target language code cannot be empty")
        self._opts.target_language_code = LanguageCode(target_language_code)

    if model is not None:
        if not model.strip():
            raise ValueError("Model cannot be empty")
        self._opts.model = model
        if speaker is None and self._opts.speaker is not None:
            if not validate_model_speaker_compatibility(self._opts.model, self._opts.speaker):
                compatible_speakers = MODEL_SPEAKER_COMPATIBILITY.get(self._opts.model, {}).get(
                    "all", []
                )
                raise ValueError(
                    f"Speaker '{self._opts.speaker}' incompatible with {self._opts.model}. "
                    f"Compatible speakers: {', '.join(compatible_speakers)}"
                )
    if speaker is not None:
        if not speaker.strip():
            raise ValueError("Speaker cannot be empty")
        if not validate_model_speaker_compatibility(self._opts.model, speaker):
            compatible_speakers = MODEL_SPEAKER_COMPATIBILITY.get(self._opts.model, {}).get(
                "all", []
            )
            raise ValueError(
                f"Speaker '{speaker}' incompatible with {self._opts.model}. "
                f"Compatible speakers: {', '.join(compatible_speakers)}"
            )
        self._opts.speaker = speaker

    if pitch is not None:
        if not -0.75 <= pitch <= 0.75:
            logger.warning(
                "pitch value %.2f is outside the Sarvam API accepted range [-0.75, 0.75]; "
                "clamping to nearest bound. Please update your code.",
                pitch,
            )
            pitch = max(-0.75, min(0.75, pitch))
        self._opts.pitch = pitch

    if pace is not None:
        if not 0.3 <= pace <= 3.0:
            raise ValueError("Pace must be between 0.3 and 3.0")
        self._opts.pace = pace

    if loudness is not None:
        if not 0.5 <= loudness <= 2.0:
            raise ValueError("Loudness must be between 0.5 and 2.0")
        self._opts.loudness = loudness

    if temperature is not None:
        if not 0.01 <= temperature <= 2.0:
            raise ValueError("Temperature must be between 0.01 and 2.0")
        self._opts.temperature = temperature

    if output_audio_bitrate is not None:
        if output_audio_bitrate not in ALLOWED_OUTPUT_AUDIO_BITRATES:
            raise ValueError(
                "output_audio_bitrate must be one of "
                f"{', '.join(sorted(ALLOWED_OUTPUT_AUDIO_BITRATES))}"
            )
        self._opts.output_audio_bitrate = output_audio_bitrate

    if min_buffer_size is not None:
        if not 30 <= min_buffer_size <= 200:
            raise ValueError("min_buffer_size must be between 30 and 200")
        self._opts.min_buffer_size = min_buffer_size

    if max_chunk_length is not None:
        if not 50 <= max_chunk_length <= 500:
            raise ValueError("max_chunk_length must be between 50 and 500")
        self._opts.max_chunk_length = max_chunk_length

    if enable_preprocessing is not None:
        self._opts.enable_preprocessing = enable_preprocessing

    if dict_id is not None:
        self._opts.dict_id = dict_id

    if enable_cached_responses is not None:
        self._opts.enable_cached_responses = enable_cached_responses

    if send_completion_event is not None:
        self._opts.send_completion_event = send_completion_event

    if output_audio_codec is not None:
        if output_audio_codec not in ALLOWED_OUTPUT_AUDIO_CODECS:
            raise ValueError(
                "output_audio_codec must be one of "
                f"{','.join(sorted(ALLOWED_OUTPUT_AUDIO_CODECS))}"
            )
        self._opts.output_audio_codec = output_audio_codec

Update TTS options with validation.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once