Module agents.inference.stt

Classes

class BaseInferenceConfig (**data: Any)
Expand source code 
class BaseInferenceConfig(BaseModel):
    model_config = ConfigDict(extra="allow")

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__
The names of the class variables defined on the model.
__private_attributes__
Metadata about the private attributes of the model.
__signature__
The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__
Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__
The core schema of the model.
__pydantic_custom_init__
Whether the model has a custom __init__ function.
__pydantic_decorators__
Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__
A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.origin] and [__args__][genericalias.args] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.
__pydantic_parent_namespace__
Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__
The name of the post-init method for the model, if defined.
__pydantic_root_model__
Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__
The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__
The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__
A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__
A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__
A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__
The names of fields explicitly set during instantiation.
__pydantic_private__
Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

  • pydantic.main.BaseModel

Subclasses

Class variables

var model_config
class DeepgramSTTInferenceConfig (**data: Any)
Expand source code 
class DeepgramSTTInferenceConfig(BaseInferenceConfig):
    input_sample_rate: int = 48000
    interim_results: bool = True
    punctuate: bool = True
    smart_format: bool = True
    endpointing: int = 50
    # v2 flux
    eager_eot_threshold: float = 0.6
    eot_threshold: float = 0.8
    eot_timeout_ms: int = 7000
    keyterm: list[str] | None = None

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__
The names of the class variables defined on the model.
__private_attributes__
Metadata about the private attributes of the model.
__signature__
The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__
Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__
The core schema of the model.
__pydantic_custom_init__
Whether the model has a custom __init__ function.
__pydantic_decorators__
Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__
A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.origin] and [__args__][genericalias.args] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.
__pydantic_parent_namespace__
Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__
The name of the post-init method for the model, if defined.
__pydantic_root_model__
Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__
The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__
The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__
A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__
A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__
A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__
The names of fields explicitly set during instantiation.
__pydantic_private__
Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var eager_eot_threshold : float
var endpointing : int
var eot_threshold : float
var eot_timeout_ms : int
var input_sample_rate : int
var interim_results : bool
var keyterm : list[str] | None
var model_config
var punctuate : bool
var smart_format : bool
class InferenceProvider (*args, **kwds)
Expand source code 
class InferenceProvider(str, Enum):
    DEEPGRAM = "deepgram"
    SARVAM = "sarvam"
    GOOGLE = "google"
    CARTESIA = "cartesia"

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var CARTESIA
var DEEPGRAM
var GOOGLE
var SARVAM
class STT (*,
provider: str,
model_id: str,
language: str = 'en-US',
config: Dict[str, Any] | None = None,
enable_streaming: bool = True,
base_url: str | None = None)
Expand source code 
class STT(BaseSTT):
    """
    VideoSDK Inference Gateway STT Plugin.

    A lightweight Speech-to-Text client that connects to VideoSDK's Inference Gateway.
    Supports multiple providers (Google, Sarvam, Deepgram) through a unified interface.

    Example:
        # Using factory methods (recommended)
        stt = STT.google(language="en-US")
        stt = STT.sarvam(language="en-IN")

        # Using generic constructor
        stt = STT(provider="google", model_id="chirp_3", config={"language": "en-US"})
    """

    def __init__(
        self,
        *,
        # provider: InferenceProvider,
        provider: str,
        model_id: str,
        language: str = "en-US",
        config: Dict[str, Any] | None = None,
        enable_streaming: bool = True,
        base_url: str | None = None,
    ) -> None:
        """
        Initialize the VideoSDK Inference STT plugin.

        Args:
            provider: STT provider name (e.g., "google", "sarvamai", "deepgram")
            model_id: Model identifier for the provider
            language: Language code (default: "en-US")
            config: Provider-specific configuration dictionary
            enable_streaming: Enable streaming transcription (default: True)
            base_url: Custom inference gateway URL (default: production gateway)
        """
        super().__init__()

        self._videosdk_token = os.getenv("VIDEOSDK_AUTH_TOKEN")
        if not self._videosdk_token:
            raise ValueError(
                "VIDEOSDK_AUTH_TOKEN environment variable must be set for authentication"
            )

        self.provider = provider
        self.model_id = model_id
        self.language = language
        self.config = config or {}
        self.enable_streaming = enable_streaming
        self.base_url = base_url or VIDEOSDK_INFERENCE_URL

        # WebSocket state
        self._session: Optional[aiohttp.ClientSession] = None
        self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
        self._ws_task: Optional[asyncio.Task] = None
        self._config_sent: bool = False

        # Speech state tracking
        self._is_speaking: bool = False
        self._last_transcript: str = ""

        # Metrics tracking
        self._stt_start_time: Optional[float] = None
        self._connecting: bool = False

        # Audio duration tracking (seconds) accumulated since the last FINAL transcript.
        # Used to populate SpeechData.duration so accounting can bill per-utterance audio.
        self._pending_audio_bytes: int = 0
        self._input_sample_rate: int = int(
            (config or {}).get("input_sample_rate", 48000)
        )
        self._bytes_per_sample: int = 2  # 16-bit PCM mono

    # ==================== Factory Methods ====================

    @staticmethod
    def google(
        *,
        model_id: str = "chirp_3",
        language: str = "en-US",
        languages: list[str] | None = None,
        interim_results: bool = True,
        punctuate: bool = True,
        location: str = "asia-south1",
        input_sample_rate: int = 48000,
        output_sample_rate: int = 16000,
        enable_streaming: bool = True,
        base_url: str | None = None,
        config: Optional[Dict] = None,
    ) -> "STT":
        """
        Create an STT instance configured for Google Cloud Speech-to-Text.

        Args:
            model_id: Google STT model (default: "chirp_3"). Options: "chirp_3", "chirp_2", "latest_long", "latest_short"
            language: Primary language code (default: "en-US")
            languages: List of languages for auto-detection (default: [language])
            interim_results: Return interim transcription results (default: True)
            punctuate: Add punctuation to transcripts (default: True)
            location: Google Cloud region (default: "asia-south1")
            input_sample_rate: Input audio sample rate (default: 48000)
            output_sample_rate: Output sample rate for processing (default: 16000)
            enable_streaming: Enable streaming mode (default: True)
            base_url: Custom inference gateway URL

        Returns:
            Configured STT instance for Google
        """
        config = {
            "model": model_id,
            "language": language,
            "languages": languages or [language],
            "input_sample_rate": input_sample_rate,
            "output_sample_rate": output_sample_rate,
            "interim_results": interim_results,
            "punctuate": punctuate,
            "location": location,
            **(config or {}),
        }
        return STT(
            provider="google",
            model_id=model_id,
            language=language,
            config=config,
            enable_streaming=enable_streaming,
            base_url=base_url,
        )

    @staticmethod
    def sarvam(
        *,
        model_id: str = "saaras:v3",
        language: str = "en-IN",
        input_sample_rate: int = 48000,
        output_sample_rate: int = 16000,
        enable_streaming: bool = True,
        base_url: str | None = None,
        config: Optional[Dict] = None,
    ) -> "STT":
        """
        Create an STT instance configured for Sarvam AI.

        Args:
            model_id: Sarvam model (default: "saaras:v3")
            language: Language code (default: "en-IN"). Supports Indian languages.
            input_sample_rate: Input audio sample rate (default: 48000)
            output_sample_rate: Output sample rate for processing (default: 16000)
            enable_streaming: Enable streaming mode (default: True)
            base_url: Custom inference gateway URL

        Returns:
            Configured STT instance for Sarvam AI
        """
        config = {
            "model": model_id,
            "language": language,
            "input_sample_rate": input_sample_rate,
            "output_sample_rate": output_sample_rate,
            **(config or {}),
        }
        return STT(
            provider="sarvamai",
            model_id=model_id,
            language=language,
            config=config,
            enable_streaming=enable_streaming,
            base_url=base_url,
        )

    @staticmethod
    def deepgram(
        *,
        model_id: str = "nova-2",
        language: str = "en-US",
        input_sample_rate: int = 48000,
        interim_results: bool = True,
        punctuate: bool = True,
        smart_format: bool = True,
        endpointing: int = 50,
        enable_streaming: bool = True,
        base_url: str | None = None,
        config: Optional[Dict] = None,
        eager_eot_threshold: float = 0.6,
        eot_threshold: float = 0.8,
        eot_timeout_ms: int = 7000,
        keyterm: list[str] | None = None,
    ) -> "STT":
        """
        Create an STT instance configured for Deepgram.

        Args:
            model_id: Deepgram model (default: "nova-2")
            language: Language code (default: "en-US")
            input_sample_rate: Input audio sample rate (default: 48000)
            interim_results: Return interim transcription results (default: True)
            punctuate: Add punctuation to transcripts (default: True)
            smart_format: Enable smart formatting (default: True)
            endpointing: Endpointing threshold in ms (default: 50)
            enable_streaming: Enable streaming mode (default: True)
            base_url: Custom inference gateway URL

        Returns:
            Configured STT instance for Deepgram
        """
        # config = config or DeepgramSTTInferenceConfig()

        # default_config = config.model_dump(exclude_none=True)
        # _config = {
        #     "model": model_id,
        #     "language": language,
        #     **default_config,
        # }
        config = {
            "model": model_id,
            "language": language,
            "input_sample_rate": input_sample_rate,
            "interim_results": interim_results,
            "punctuate": punctuate,
            "smart_format": smart_format,
            "endpointing": endpointing,
            "eager_eot_threshold": eager_eot_threshold,
            "eot_threshold": eot_threshold,
            "eot_timeout_ms": eot_timeout_ms,
            "keyterm": keyterm,
            **(config or {}),
        }
        return STT(
            provider="deepgram",
            model_id=model_id,
            language=language,
            enable_streaming=enable_streaming,
            config=config,
            base_url=base_url,
        )

    @staticmethod
    def assemblyai(
        *,
        model_id: str = "universal-streaming-english",
        input_sample_rate: int = 48000,
        region: str = "US",
        format_turns: bool = True,
        keyterms_prompt: list[str] | None = None,
        end_of_turn_confidence_threshold: float = 0.5,
        min_end_of_turn_silence_when_confident: int = 800,
        max_turn_silence: int = 2000,
        language_detection: bool = True,
        enable_streaming: bool = True,
        base_url: str | None = None,
        config: Optional[Dict] = None,
    ) -> "STT":
        """
        Create an STT instance configured for AssemblyAI Universal Streaming.

        Args:
            model_id: AssemblyAI speech model (default: "universal-streaming-english")
                Options: "universal-streaming-english", "universal-streaming-multilingual"
            input_sample_rate: Input audio sample rate (default: 48000)
            region: Service region (default: "US"). Options: "US", "EU"
            format_turns: Whether to format turns (default: True)
            keyterms_prompt: List of words/phrases to boost recognition (optional)
            end_of_turn_confidence_threshold: Confidence threshold for end-of-turn (default: 0.5)
            min_end_of_turn_silence_when_confident: Min silence in ms when confident (default: 800)
            max_turn_silence: Max silence in ms before end-of-turn (default: 2000)
            language_detection: Enable automatic language detection (default: True)
            enable_streaming: Enable streaming mode (default: True)
            base_url: Custom inference gateway URL

        Returns:
            Configured STT instance for AssemblyAI
        """
        resolved_config = {
            "speech_model": model_id,
            "input_sample_rate": input_sample_rate,
            "region": region,
            "format_turns": format_turns,
            "keyterms_prompt": keyterms_prompt,
            "end_of_turn_confidence_threshold": end_of_turn_confidence_threshold,
            "min_end_of_turn_silence_when_confident": min_end_of_turn_silence_when_confident,
            "max_turn_silence": max_turn_silence,
            "language_detection": language_detection,
            **(config or {}),
        }
        return STT(
            provider="assemblyai",
            model_id=model_id,
            language="en",
            enable_streaming=enable_streaming,
            config=resolved_config,
            base_url=base_url,
        )

    # ==================== Core Methods ====================

    async def flush(self) -> None:
        """Signal end-of-speech to the inference server."""
        if not self._ws or self._ws.closed:
            return
        try:
            await self._ws.send_str(json.dumps({"type": "flush"}))
        except Exception as e:
            logger.debug(f"[InferenceSTT] Flush error: {e}")

    async def process_audio(
        self,
        audio_frames: bytes,
        language: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        """
        Process audio frames and send to the inference server.

        Args:
            audio_frames: Raw PCM audio bytes (16-bit, typically 48kHz stereo)
            language: Optional language override
            **kwargs: Additional arguments (unused)
        """
        # logger.info(f"[STT DEBUG] process_audio called | bytes={len(audio_frames)}")
        if not self.enable_streaming:
            logger.warning("Non-streaming mode not yet supported for inference STT")
            return

        try:
            if self._connecting:
                return

            if not self._ws or self._ws.closed:
                self._connecting = True
                try:
                    await self._connect_ws()
                    if not self._ws_task or self._ws_task.done():
                        self._ws_task = asyncio.create_task(
                            self._listen_for_responses()
                        )
                finally:
                    self._connecting = False

            if not self._config_sent:
                await self._send_config()

            await self._send_audio(audio_frames)
            await asyncio.sleep(0)

        except Exception as e:
            logger.error(f"[InferenceSTT] Error in process_audio: {e}")
            self.emit("error", str(e))
            await self._cleanup_connection()

    async def _connect_ws(self) -> None:
        """Establish WebSocket connection to the inference gateway."""
        # logger.info(f"[STT DEBUG] WS URL: {ws_url}")

        if not self._session:
            self._session = aiohttp.ClientSession()

        # Build WebSocket URL with query parameters
        ws_url = (
            f"{self.base_url}/v1/stt"
            f"?provider={self.provider}"
            f"&secret={self._videosdk_token}"
            f"&modelId={self.model_id}"
        )
        # logger.info(f"[STT DEBUG] Connecting to: {ws_url}")
        try:
            logger.info(
                f"[InferenceSTT] Connecting to {self.base_url} (provider={self.provider})"
            )
            self._ws = await self._session.ws_connect(ws_url, heartbeat=20)
            self._config_sent = False
            logger.info(f"[InferenceSTT] Connected successfully")
        except Exception as e:
            logger.error(f"[InferenceSTT] Connection failed: {e}")
            raise

    async def _send_config(self) -> None:
        """Send configuration message to the inference server."""
        if not self._ws:
            return

        config_message = {"type": "config", "data": self.config}
        try:
            await self._ws.send_str(json.dumps(config_message))
            self._config_sent = True
            logger.info(f"[InferenceSTT] Config sent: {self.config}")
        except Exception as e:
            logger.error(f"[InferenceSTT] Failed to send config: {e}")
            raise

    async def _send_audio(self, audio_bytes: bytes) -> None:
        """Send audio data to the inference server."""
        # logger.info(f"[STT DEBUG] sending audio chunk | bytes={len(audio_bytes)}")
        if not self._ws:
            return

        # Track STT start time for metrics
        if self._stt_start_time is None:
            self._stt_start_time = time.perf_counter()

        # Track audio bytes for per-utterance duration accounting.
        self._pending_audio_bytes += len(audio_bytes)

        # Encode audio as base64 for JSON transmission
        audio_message = {
            "type": "audio",
            "data": base64.b64encode(audio_bytes).decode("utf-8"),
        }
        try:
            await self._ws.send_str(json.dumps(audio_message))
        except Exception as e:
            logger.error(f"[InferenceSTT] Failed to send audio: {e}")
            raise

    async def _listen_for_responses(self) -> None:
        """Background task to listen for WebSocket responses from the server."""
        if not self._ws:
            return

        try:
            async for msg in self._ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    await self._handle_message(msg.data)
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    logger.error(
                        f"[InferenceSTT] WebSocket error: {self._ws.exception()}"
                    )
                    self.emit("error", f"WebSocket error: {self._ws.exception()}")
                    break
                elif msg.type == aiohttp.WSMsgType.CLOSED:
                    logger.info("[InferenceSTT] WebSocket closed by server")
                    break

        except asyncio.CancelledError:
            logger.debug("[InferenceSTT] WebSocket listener cancelled")
        except Exception as e:
            logger.error(f"[InferenceSTT] Error in WebSocket listener: {e}")
            self.emit("error", str(e))
        finally:
            await self._cleanup_connection()

    async def _handle_message(self, raw_message: str) -> None:
        """
        Handle incoming messages from the inference server.

        Args:
            raw_message: Raw JSON message string from server
        """
        # logger.info(f"[STT DEBUG] raw message from server: {raw_message}")
        try:
            data = json.loads(raw_message)
            msg_type = data.get("type")

            if msg_type == "event":
                await self._handle_event(data.get("data", {}))
            elif msg_type == "error":
                error_msg = data.get("data", {}).get("error") or data.get(
                    "message", "Unknown error"
                )
                logger.error(f"[InferenceSTT] Server error: {error_msg}")
                self.emit("error", error_msg)

        except json.JSONDecodeError as e:
            logger.error(f"[InferenceSTT] Failed to parse message: {e}")
        except Exception as e:
            logger.error(f"[InferenceSTT] Error handling message: {e}")

    async def _handle_event(self, event_data: Dict[str, Any]) -> None:
        """
        Handle event messages from the inference server.

        Args:
            event_data: Event data dictionary
        """
        event_type = event_data.get("eventType")

        if event_type == "TRANSCRIPT":
            text = event_data.get("text", "")
            language = event_data.get("language", self.language)
            is_final = event_data.get("is_final", True)
            confidence = event_data.get("confidence", 1.0)

            # before we send the text if there is an small intruption like khasi and it trigger the endspeech and send the blank text to llm so sarvam tts cant processed that blank space
            # if not self._has_enough_content(text):
            #     logger.debug(f"[InferenceSTT] Invalid transcript skipped: '{text}'")
            #     return
            if text.strip():
                logger.info(f"[STT] {text} | Final: {is_final}")
            self._last_transcript = text.strip()

            duration = 0.0
            if is_final and self._pending_audio_bytes > 0:
                sample_count = self._pending_audio_bytes / self._bytes_per_sample
                duration = sample_count / self._input_sample_rate
                self._pending_audio_bytes = 0

            response = STTResponse(
                event_type=(
                    SpeechEventType.FINAL if is_final else SpeechEventType.INTERIM
                ),
                data=SpeechData(
                    text=text.strip(),
                    language=language,
                    confidence=confidence,
                    duration=duration,
                ),
                metadata={
                    "provider": self.provider,
                    "model": self.model_id,
                },
            )

            if self._transcript_callback:
                await self._transcript_callback(response)

            if is_final:
                self._stt_start_time = None

        elif event_type == "START_SPEECH":
            if not self._is_speaking:
                self._is_speaking = True
                global_event_emitter.emit("speech_started")
                logger.debug("[InferenceSTT] Speech started")

        elif event_type == "END_SPEECH":
            if self._is_speaking:
                self._is_speaking = False
                if self._last_transcript:
                    global_event_emitter.emit("speech_stopped")
                    logger.debug("[InferenceSTT] Speech ended (transcript present)")
                else:
                    logger.debug(
                        "[InferenceSTT] Speech ended but no transcript — suppressing speech_stopped"
                    )
        elif event_type == "DENOISE_AUDIO":
            logger.warning("[STT] Received DENOISE_AUDIO event — ignoring")
            return

    async def _cleanup_connection(self) -> None:
        """Clean up WebSocket connection resources."""
        if self._ws and not self._ws.closed:
            try:
                await self._ws.close()
            except Exception:
                pass
        self._ws = None
        self._config_sent = False

    async def aclose(self) -> None:
        """Clean up all resources."""
        logger.info(f"[InferenceSTT] Closing STT (provider={self.provider})")

        # Cancel listener task
        if self._ws_task:
            self._ws_task.cancel()
            try:
                await self._ws_task
            except asyncio.CancelledError:
                pass
            self._ws_task = None

        # Close WebSocket
        await self._cleanup_connection()

        # Close HTTP session
        if self._session and not self._session.closed:
            await self._session.close()
            self._session = None

        # Call parent cleanup
        await super().aclose()
        logger.info(f"[InferenceSTT] Closed successfully")

    # ==================== Properties ====================

    @property
    def label(self) -> str:
        """Get a descriptive label for this STT instance."""
        return f"videosdk.inference.STT.{self.provider}.{self.model_id}"

VideoSDK Inference Gateway STT Plugin.

A lightweight Speech-to-Text client that connects to VideoSDK's Inference Gateway. Supports multiple providers (Google, Sarvam, Deepgram) through a unified interface.

Example

Using factory methods (recommended)

stt = STT.google(language="en-US") stt = STT.sarvam(language="en-IN")

Using generic constructor

stt = STT(provider="google", model_id="chirp_3", config={"language": "en-US"})

Initialize the VideoSDK Inference STT plugin.

Args

provider
STT provider name (e.g., "google", "sarvamai", "deepgram")
model_id
Model identifier for the provider
language
Language code (default: "en-US")
config
Provider-specific configuration dictionary
enable_streaming
Enable streaming transcription (default: True)
base_url
Custom inference gateway URL (default: production gateway)

Ancestors

  • videosdk.agents.stt.stt.STT
  • videosdk.agents.event_emitter.EventEmitter
  • typing.Generic

Static methods

def assemblyai(*,
model_id: str = 'universal-streaming-english',
input_sample_rate: int = 48000,
region: str = 'US',
format_turns: bool = True,
keyterms_prompt: list[str] | None = None,
end_of_turn_confidence_threshold: float = 0.5,
min_end_of_turn_silence_when_confident: int = 800,
max_turn_silence: int = 2000,
language_detection: bool = True,
enable_streaming: bool = True,
base_url: str | None = None,
config: Optional[Dict] = None) ‑> STT
Expand source code 
@staticmethod
def assemblyai(
    *,
    model_id: str = "universal-streaming-english",
    input_sample_rate: int = 48000,
    region: str = "US",
    format_turns: bool = True,
    keyterms_prompt: list[str] | None = None,
    end_of_turn_confidence_threshold: float = 0.5,
    min_end_of_turn_silence_when_confident: int = 800,
    max_turn_silence: int = 2000,
    language_detection: bool = True,
    enable_streaming: bool = True,
    base_url: str | None = None,
    config: Optional[Dict] = None,
) -> "STT":
    """
    Create an STT instance configured for AssemblyAI Universal Streaming.

    Args:
        model_id: AssemblyAI speech model (default: "universal-streaming-english")
            Options: "universal-streaming-english", "universal-streaming-multilingual"
        input_sample_rate: Input audio sample rate (default: 48000)
        region: Service region (default: "US"). Options: "US", "EU"
        format_turns: Whether to format turns (default: True)
        keyterms_prompt: List of words/phrases to boost recognition (optional)
        end_of_turn_confidence_threshold: Confidence threshold for end-of-turn (default: 0.5)
        min_end_of_turn_silence_when_confident: Min silence in ms when confident (default: 800)
        max_turn_silence: Max silence in ms before end-of-turn (default: 2000)
        language_detection: Enable automatic language detection (default: True)
        enable_streaming: Enable streaming mode (default: True)
        base_url: Custom inference gateway URL

    Returns:
        Configured STT instance for AssemblyAI
    """
    resolved_config = {
        "speech_model": model_id,
        "input_sample_rate": input_sample_rate,
        "region": region,
        "format_turns": format_turns,
        "keyterms_prompt": keyterms_prompt,
        "end_of_turn_confidence_threshold": end_of_turn_confidence_threshold,
        "min_end_of_turn_silence_when_confident": min_end_of_turn_silence_when_confident,
        "max_turn_silence": max_turn_silence,
        "language_detection": language_detection,
        **(config or {}),
    }
    return STT(
        provider="assemblyai",
        model_id=model_id,
        language="en",
        enable_streaming=enable_streaming,
        config=resolved_config,
        base_url=base_url,
    )

Create an STT instance configured for AssemblyAI Universal Streaming.

Args

model_id
AssemblyAI speech model (default: "universal-streaming-english") Options: "universal-streaming-english", "universal-streaming-multilingual"
input_sample_rate
Input audio sample rate (default: 48000)
region
Service region (default: "US"). Options: "US", "EU"
format_turns
Whether to format turns (default: True)
keyterms_prompt
List of words/phrases to boost recognition (optional)
end_of_turn_confidence_threshold
Confidence threshold for end-of-turn (default: 0.5)
min_end_of_turn_silence_when_confident
Min silence in ms when confident (default: 800)
max_turn_silence
Max silence in ms before end-of-turn (default: 2000)
language_detection
Enable automatic language detection (default: True)
enable_streaming
Enable streaming mode (default: True)
base_url
Custom inference gateway URL

Returns

Configured STT instance for AssemblyAI

def deepgram(*,
model_id: str = 'nova-2',
language: str = 'en-US',
input_sample_rate: int = 48000,
interim_results: bool = True,
punctuate: bool = True,
smart_format: bool = True,
endpointing: int = 50,
enable_streaming: bool = True,
base_url: str | None = None,
config: Optional[Dict] = None,
eager_eot_threshold: float = 0.6,
eot_threshold: float = 0.8,
eot_timeout_ms: int = 7000,
keyterm: list[str] | None = None) ‑> STT
Expand source code 
@staticmethod
def deepgram(
    *,
    model_id: str = "nova-2",
    language: str = "en-US",
    input_sample_rate: int = 48000,
    interim_results: bool = True,
    punctuate: bool = True,
    smart_format: bool = True,
    endpointing: int = 50,
    enable_streaming: bool = True,
    base_url: str | None = None,
    config: Optional[Dict] = None,
    eager_eot_threshold: float = 0.6,
    eot_threshold: float = 0.8,
    eot_timeout_ms: int = 7000,
    keyterm: list[str] | None = None,
) -> "STT":
    """
    Create an STT instance configured for Deepgram.

    Args:
        model_id: Deepgram model (default: "nova-2")
        language: Language code (default: "en-US")
        input_sample_rate: Input audio sample rate (default: 48000)
        interim_results: Return interim transcription results (default: True)
        punctuate: Add punctuation to transcripts (default: True)
        smart_format: Enable smart formatting (default: True)
        endpointing: Endpointing threshold in ms (default: 50)
        enable_streaming: Enable streaming mode (default: True)
        base_url: Custom inference gateway URL

    Returns:
        Configured STT instance for Deepgram
    """
    # config = config or DeepgramSTTInferenceConfig()

    # default_config = config.model_dump(exclude_none=True)
    # _config = {
    #     "model": model_id,
    #     "language": language,
    #     **default_config,
    # }
    config = {
        "model": model_id,
        "language": language,
        "input_sample_rate": input_sample_rate,
        "interim_results": interim_results,
        "punctuate": punctuate,
        "smart_format": smart_format,
        "endpointing": endpointing,
        "eager_eot_threshold": eager_eot_threshold,
        "eot_threshold": eot_threshold,
        "eot_timeout_ms": eot_timeout_ms,
        "keyterm": keyterm,
        **(config or {}),
    }
    return STT(
        provider="deepgram",
        model_id=model_id,
        language=language,
        enable_streaming=enable_streaming,
        config=config,
        base_url=base_url,
    )

Create an STT instance configured for Deepgram.

Args

model_id
Deepgram model (default: "nova-2")
language
Language code (default: "en-US")
input_sample_rate
Input audio sample rate (default: 48000)
interim_results
Return interim transcription results (default: True)
punctuate
Add punctuation to transcripts (default: True)
smart_format
Enable smart formatting (default: True)
endpointing
Endpointing threshold in ms (default: 50)
enable_streaming
Enable streaming mode (default: True)
base_url
Custom inference gateway URL

Returns

Configured STT instance for Deepgram

def google(*,
model_id: str = 'chirp_3',
language: str = 'en-US',
languages: list[str] | None = None,
interim_results: bool = True,
punctuate: bool = True,
location: str = 'asia-south1',
input_sample_rate: int = 48000,
output_sample_rate: int = 16000,
enable_streaming: bool = True,
base_url: str | None = None,
config: Optional[Dict] = None) ‑> STT
Expand source code 
@staticmethod
def google(
    *,
    model_id: str = "chirp_3",
    language: str = "en-US",
    languages: list[str] | None = None,
    interim_results: bool = True,
    punctuate: bool = True,
    location: str = "asia-south1",
    input_sample_rate: int = 48000,
    output_sample_rate: int = 16000,
    enable_streaming: bool = True,
    base_url: str | None = None,
    config: Optional[Dict] = None,
) -> "STT":
    """
    Create an STT instance configured for Google Cloud Speech-to-Text.

    Args:
        model_id: Google STT model (default: "chirp_3"). Options: "chirp_3", "chirp_2", "latest_long", "latest_short"
        language: Primary language code (default: "en-US")
        languages: List of languages for auto-detection (default: [language])
        interim_results: Return interim transcription results (default: True)
        punctuate: Add punctuation to transcripts (default: True)
        location: Google Cloud region (default: "asia-south1")
        input_sample_rate: Input audio sample rate (default: 48000)
        output_sample_rate: Output sample rate for processing (default: 16000)
        enable_streaming: Enable streaming mode (default: True)
        base_url: Custom inference gateway URL

    Returns:
        Configured STT instance for Google
    """
    config = {
        "model": model_id,
        "language": language,
        "languages": languages or [language],
        "input_sample_rate": input_sample_rate,
        "output_sample_rate": output_sample_rate,
        "interim_results": interim_results,
        "punctuate": punctuate,
        "location": location,
        **(config or {}),
    }
    return STT(
        provider="google",
        model_id=model_id,
        language=language,
        config=config,
        enable_streaming=enable_streaming,
        base_url=base_url,
    )

Create an STT instance configured for Google Cloud Speech-to-Text.

Args

model_id
Google STT model (default: "chirp_3"). Options: "chirp_3", "chirp_2", "latest_long", "latest_short"
language
Primary language code (default: "en-US")
languages
List of languages for auto-detection (default: [language])
interim_results
Return interim transcription results (default: True)
punctuate
Add punctuation to transcripts (default: True)
location
Google Cloud region (default: "asia-south1")
input_sample_rate
Input audio sample rate (default: 48000)
output_sample_rate
Output sample rate for processing (default: 16000)
enable_streaming
Enable streaming mode (default: True)
base_url
Custom inference gateway URL

Returns

Configured STT instance for Google

def sarvam(*,
model_id: str = 'saaras:v3',
language: str = 'en-IN',
input_sample_rate: int = 48000,
output_sample_rate: int = 16000,
enable_streaming: bool = True,
base_url: str | None = None,
config: Optional[Dict] = None) ‑> STT
Expand source code 
@staticmethod
def sarvam(
    *,
    model_id: str = "saaras:v3",
    language: str = "en-IN",
    input_sample_rate: int = 48000,
    output_sample_rate: int = 16000,
    enable_streaming: bool = True,
    base_url: str | None = None,
    config: Optional[Dict] = None,
) -> "STT":
    """
    Create an STT instance configured for Sarvam AI.

    Args:
        model_id: Sarvam model (default: "saaras:v3")
        language: Language code (default: "en-IN"). Supports Indian languages.
        input_sample_rate: Input audio sample rate (default: 48000)
        output_sample_rate: Output sample rate for processing (default: 16000)
        enable_streaming: Enable streaming mode (default: True)
        base_url: Custom inference gateway URL

    Returns:
        Configured STT instance for Sarvam AI
    """
    config = {
        "model": model_id,
        "language": language,
        "input_sample_rate": input_sample_rate,
        "output_sample_rate": output_sample_rate,
        **(config or {}),
    }
    return STT(
        provider="sarvamai",
        model_id=model_id,
        language=language,
        config=config,
        enable_streaming=enable_streaming,
        base_url=base_url,
    )

Create an STT instance configured for Sarvam AI.

Args

model_id
Sarvam model (default: "saaras:v3")
language
Language code (default: "en-IN"). Supports Indian languages.
input_sample_rate
Input audio sample rate (default: 48000)
output_sample_rate
Output sample rate for processing (default: 16000)
enable_streaming
Enable streaming mode (default: True)
base_url
Custom inference gateway URL

Returns

Configured STT instance for Sarvam AI

Instance variables

prop label : str
Expand source code 
@property
def label(self) -> str:
    """Get a descriptive label for this STT instance."""
    return f"videosdk.inference.STT.{self.provider}.{self.model_id}"

Get a descriptive label for this STT instance.

Methods

async def aclose(self) ‑> None
Expand source code 
async def aclose(self) -> None:
    """Clean up all resources."""
    logger.info(f"[InferenceSTT] Closing STT (provider={self.provider})")

    # Cancel listener task
    if self._ws_task:
        self._ws_task.cancel()
        try:
            await self._ws_task
        except asyncio.CancelledError:
            pass
        self._ws_task = None

    # Close WebSocket
    await self._cleanup_connection()

    # Close HTTP session
    if self._session and not self._session.closed:
        await self._session.close()
        self._session = None

    # Call parent cleanup
    await super().aclose()
    logger.info(f"[InferenceSTT] Closed successfully")

Clean up all resources.

async def flush(self) ‑> None
Expand source code 
async def flush(self) -> None:
    """Signal end-of-speech to the inference server."""
    if not self._ws or self._ws.closed:
        return
    try:
        await self._ws.send_str(json.dumps({"type": "flush"}))
    except Exception as e:
        logger.debug(f"[InferenceSTT] Flush error: {e}")

Signal end-of-speech to the inference server.

async def process_audio(self, audio_frames: bytes, language: Optional[str] = None, **kwargs: Any) ‑> None
Expand source code 
async def process_audio(
    self,
    audio_frames: bytes,
    language: Optional[str] = None,
    **kwargs: Any,
) -> None:
    """
    Process audio frames and send to the inference server.

    Args:
        audio_frames: Raw PCM audio bytes (16-bit, typically 48kHz stereo)
        language: Optional language override
        **kwargs: Additional arguments (unused)
    """
    # logger.info(f"[STT DEBUG] process_audio called | bytes={len(audio_frames)}")
    if not self.enable_streaming:
        logger.warning("Non-streaming mode not yet supported for inference STT")
        return

    try:
        if self._connecting:
            return

        if not self._ws or self._ws.closed:
            self._connecting = True
            try:
                await self._connect_ws()
                if not self._ws_task or self._ws_task.done():
                    self._ws_task = asyncio.create_task(
                        self._listen_for_responses()
                    )
            finally:
                self._connecting = False

        if not self._config_sent:
            await self._send_config()

        await self._send_audio(audio_frames)
        await asyncio.sleep(0)

    except Exception as e:
        logger.error(f"[InferenceSTT] Error in process_audio: {e}")
        self.emit("error", str(e))
        await self._cleanup_connection()

Process audio frames and send to the inference server.

Args

audio_frames
Raw PCM audio bytes (16-bit, typically 48kHz stereo)
language
Optional language override
**kwargs
Additional arguments (unused)