Overview

ASR-based audio transcription implementations.

This package provides ASR (Automatic Speech Recognition) based transcription using cloud and self-hosted ASR engines.

Submodules

promptable_asr -- Promptable ASR implementations (OpenAI Whisper, Qwen).

Exported Classes

ASRBasedAudioToTranscript -- Abstract base for ASR-based converters.
ASRBasedGoogleCloudAudioToText -- Google Cloud Speech-to-Text.
ASRBasedOpenAIAudioToText -- OpenAI Whisper API.
ASRBasedProsaAudioToText -- Prosa ASR engine.
ASRBasedQwenAudioToText -- Qwen ASR engine.
ASRBasedSnowflakeAudioToText -- Snowflake Cortex AI.

`ASRBasedAudioToTranscript()`

Bases: BaseAudioToTranscript, ABC

Abstract base class for dedicated speech-to-text API integrations.

`ASRBasedGoogleCloudAudioToText(credentials_json, bucket_name, language_code='id-ID', alternative_language_codes=None, model='latest_long', timeout=5 * 60, proxy=None)`

Bases: ASRBasedAudioToTranscript

An audio to text converter using Google Cloud Speech-to-Text.

The ASRBasedGoogleCloudAudioToText class is responsible for converting audio to text using the Google Cloud Speech-to-Text. It supports various audio input formats and can handle audio from local files, base64 encoded strings, or URLs pointing to audio files.

Attributes:

Name	Type	Description
`speech_client`	`SpeechClient`	Google Cloud Speech-to-Text client.
`storage_client`	`Client`	Google Cloud Storage client.
`bucket_name`	`str`	Google Cloud Storage bucket name.
`language_code`	`str`	Language code for transcription.
`alternative_language_codes`	`list[str] \| None`	Alternative language codes for transcription. Up to 3 alternatives.
`model`	`str`	Transcription model name.
`timeout`	`int`	Timeout for the transcription request.
`proxy`	`str \| None`	The proxy URL to use for the YouTube request.

Initialize the ASRBasedGoogleCloudAudioToText instance.

Parameters:

Name	Type	Description	Default
`credentials_json`	`str \| dict`	Google Cloud API credentials can either be a file path or a dictionary.	required
`bucket_name`	`str`	Google Cloud Storage bucket name.	required
`language_code`	`str`	Language code for transcription. Defaults to "id-ID".	`'id-ID'`
`alternative_language_codes`	`list[str] \| None`	Alternative language codes for transcription. Up to 3 alternatives. If the list has more than 3 elements, the first 3 will be used. Defaults to None, in which ["id-ID", "en-US", "en-GB"] is used.	`None`
`model`	`str`	Transcription model name. Defaults to "latest_long".	`'latest_long'`
`timeout`	`int`	Timeout for the transcription request. Defaults to 5 minutes.	`5 * 60`
`proxy`	`str \| None`	The proxy URL to use for the YouTube request. Defaults to None.	`None`

`ASRBasedOpenAIAudioToText(api_key, model='whisper-1', language=None, prompt=None, temperature=0, timestamp_granularity='segment', proxy=None, skip_conversion_for_formats=None, base_url=None, merge_consecutive_duplicates=False)`

Bases: PromptableASRBasedAudioToTranscript

An audio to text converter using OpenAI Whisper.

The ASRBasedOpenAIAudioToText class is responsible for converting audio to text using OpenAI Whisper. It supports various audio sources such as file paths, base64 encoded strings, downloadable audio URLs, and YouTube URLs.

When base_url is omitted, the class operates using the official OpenAI API. When base_url points to a custom OpenAI-compatible endpoint, the API response is checked for an error field and a fallback AudioTranscript is produced when no segments or words are returned.

In both cases:

Audio is converted to mono FLAC before upload (unless the format appears in skip_conversion_for_formats).
Files exceeding 25 MB are progressively downsampled.

Both modes use the same synchronous openai.OpenAI client.

Attributes:

Name	Type	Description
`client`	`OpenAI`	The OpenAI client instance used for API requests.
`model`	`str`	The identifier of the OpenAI Whisper model to use for transcription.
`language`	`str \| None`	The language of the input audio content.
`prompt`	`str \| None`	The text prompt to guide the model's style or continue a previous audio segment.
`temperature`	`float`	The sampling temperature to control output randomness.
`timestamp_granularity`	`str`	The timestamp detail levels to include in the output.
`proxy`	`str \| None`	The proxy URL to use for the YouTube request.
`skip_conversion_for_formats`	`list[str]`	List of audio formats that should skip mono FLAC conversion.
`merge_consecutive_duplicates`	`bool`	Whether to merge consecutive duplicated transcripts.

Initialize the ASRBasedOpenAIAudioToText instance.

Parameters:

Name	Type	Description	Default
`api_key`	`str`	The API key for authentication with OpenAI (or the custom endpoint).	required
`model`	`str`	The model identifier to use for transcription. Defaults to "whisper-1".	`'whisper-1'`
`language`	`str \| None`	The language of the input audio content. Defaults to None.	`None`
`prompt`	`str \| PromptBuilder \| None`	The text prompt or prompt builder to guide transcription behavior. Defaults to None.	`None`
`temperature`	`float`	The sampling temperature to control output randomness. Defaults to 0.	`0`
`timestamp_granularity`	`str`	The timestamp detail levels to include in the output. The granularity can be "segment" or "word". When set to "segment", OpenAI will return transcripts divided into segments of speech. When set to "word", it will include word-level timestamps. Defaults to "segment".	`'segment'`
`proxy`	`str \| None`	The proxy URL to use for the YouTube request. Defaults to None.	`None`
`skip_conversion_for_formats`	`list[str] \| None`	List of audio formats (e.g., ['mp3', 'wav']) that should skip the mono FLAC conversion process. Formats are case-insensitive and can be specified with or without a leading dot. If None or empty, all audio will be converted to mono FLAC. Defaults to None.	`None`
`base_url`	`str \| None`	Base URL of a custom OpenAI-compatible Whisper endpoint (e.g. `"https://my-whisper-server/v1"`). When provided, the class checks for API-level errors, and produces a fallback transcript when the response contains no segments or words. Defaults to `None` (standard OpenAI API).	`None`
`merge_consecutive_duplicates`	`bool`	Whether to merge consecutive transcripts that have the exact same text, combining their time ranges. Defaults to False.	`False`

`ASRBasedProsaAudioToText(api_key, base_url=None, url=None, model='stt-general', polling_interval=MIN_POLLING_INTERVAL)`

Bases: ASRBasedAudioToTranscript

An audio to text converter using Prosa STT.

The ASRBasedProsaAudioToText class is responsible for converting audio to text using the Prosa STT API. It supports various audio input formats and can handle audio from local files, base64 encoded strings, or URLs pointing to audio files.

Attributes:

Name	Type	Description
`url`	`str`	The URL of the Prosa STT API.
`api_key`	`str`	The API key for authenticating with the Prosa STT API.
`model`	`str`	The model to use for the transcription.
`polling_interval`	`int`	The interval between polling requests to the Prosa STT API.

Initializes a new instance of the ASRBasedProsaAudioToText class.

Parameters:

Name	Type	Description	Default
`api_key`	`str`	The API key for the Prosa STT API.	required
`base_url`	`str \| None`	The base URL of the Prosa STT API. Defaults to None and falls back to DEFAULT_PROSA_STT_BASE_URL.	`None`
`url`	`str \| None`	Deprecated alias for `base_url`. Defaults to None.	`None`
`model`	`str`	The model to use for the transcription. Defaults to "stt-general".	`'stt-general'`
`polling_interval`	`int`	The interval between polling requests to the Prosa STT API. Defaults to MIN_POLLING_INTERVAL, which is set to 5 seconds.	`MIN_POLLING_INTERVAL`

`polling_interval` `property` `writable`

Get the polling interval in seconds.

Returns:

Name	Type	Description
`int`	`int`	The current polling interval.

`ASRBasedQwenAudioToText(api_key='EMPTY', model='Qwen/Qwen3-ASR-1.7B', base_url='http://localhost:8000/v1', prompt=None, align_text=None, temperature=0.01, max_tokens=1024, timestamp_granularity='segment', segment_gap_threshold=DEFAULT_SEGMENT_GAP_THRESHOLD, proxy=None, timeout=DEFAULT_REQUEST_TIMEOUT)`

Bases: PromptableASRBasedAudioToTranscript

An audio to text converter using Qwen3-ASR served via a vLLM-compatible endpoint.

This class sends audio to a vLLM OpenAI-compatible chat completions endpoint as a base64-encoded audio_url data URI. The model returns a structured response containing a <|language|> tag, a <|text|> tag with the full transcription, and a <|timestamps|> tag with word-level timing information.

The timestamp_granularity parameter controls the output format:

"word" — each word is returned as a separate AudioTranscript with its own start and end times.
"segment" — consecutive words are merged into segments based on the time gap between them. A new segment is started whenever the gap between one word's end time and the next word's start time exceeds segment_gap_threshold seconds.

Attributes:

Name	Type	Description
`client`	`AsyncOpenAI`	The OpenAI-compatible async client used for API requests.
`model`	`str`	The model identifier (e.g. `"Qwen/Qwen3-ASR-1.7B"`).
`prompt`	`str \| None`	Optional text prompt to inject domain knowledge or context.
`align_text`	`str \| None`	The ground truth transcription to use for forced alignment.
`temperature`	`float`	The sampling temperature for generation.
`max_tokens`	`int`	The maximum number of tokens to generate.
`timestamp_granularity`	`str`	The timestamp detail level — `"word"` or `"segment"`.
`segment_gap_threshold`	`float`	The maximum gap in seconds between consecutive words before starting a new segment (only used when `timestamp_granularity` is `"segment"`).
`proxy`	`str \| None`	The proxy URL to use for YouTube requests.

Initialize the ASRBasedQwenAudioToText instance.

Parameters:

Name	Type	Description	Default
`api_key`	`str`	The API key for authentication with the endpoint.	`'EMPTY'`
`model`	`str`	The model identifier. Defaults to "Qwen/Qwen3-ASR-1.7B".	`'Qwen/Qwen3-ASR-1.7B'`
`base_url`	`str`	The base URL of the vLLM-compatible endpoint. Defaults to "http://localhost:8000/v1".	`'http://localhost:8000/v1'`
`prompt`	`str \| PromptBuilder \| None`	Optional text prompt or prompt builder used to inject domain knowledge or context. Defaults to None.	`None`
`align_text`	`str \| None`	The ground truth transcription to use for forced alignment. If provided, the server will bypass generation and instead return accurate timestamps for the text. Defaults to None.	`None`
`temperature`	`float`	The sampling temperature. Defaults to 0.01.	`0.01`
`max_tokens`	`int`	The maximum number of tokens to generate. Defaults to 1024.	`1024`
`timestamp_granularity`	`str`	The timestamp detail level. Use `"word"` for word-level transcripts or `"segment"` to merge words into segments split by time gaps. Must be one of `"word"` or `"segment"`. Defaults to "segment".	`'segment'`
`segment_gap_threshold`	`float`	The maximum gap in seconds between consecutive words before a new segment is started. Only applies when `timestamp_granularity` is `"segment"`. Defaults to 2.0.	`DEFAULT_SEGMENT_GAP_THRESHOLD`
`proxy`	`str \| None`	The proxy URL for YouTube requests. Defaults to None.	`None`
`timeout`	`float`	The timeout in seconds for API requests. Defaults to 120.0.	`DEFAULT_REQUEST_TIMEOUT`

Raises:

Type	Description
`ValueError`	If `timestamp_granularity` is not `"word"` or `"segment"`.

`ASRBasedSnowflakeAudioToText(session_config=None, timestamp_granularity=None, proxy=None, auto_cleanup_stage_file=True)`

Bases: ASRBasedAudioToTranscript

An audio to text converter using Snowflake AI_TRANSCRIBE.

The ASRBasedSnowflakeAudioToText class transcribes audio via Snowflake Cortex using the ai_transcribe Snowpark function. Audio must reside on a Snowflake stage with server-side encryption. When the input is not already a stage path, the converter uploads the resolved audio bytes to a temporary stage before transcription.

When session_config is provided, a Snowpark session is established at init time and reused across invocations. Call close when finished, or use the instance as a context manager (with ASRBasedSnowflakeAudioToText(...) as transcriber:), to release that session deterministically. When omitted, get_active_session() is used at invoke time, suitable for code running inside Snowflake (notebooks, Streamlit); do not call close in that mode.

This class is not thread-safe. Do not invoke convert concurrently on the same instance.

Supported features

Local file, bytes, URL, and YouTube audio sources
Direct transcription from an existing Snowflake stage path
Optional word-level or speaker-level timestamps (see module References [1])
Automatic cleanup of uploaded stage files

Attributes:

Name	Type	Description
`session_config`	`SnowflakeSessionConfig \| None`	Validated Snowflake connection parameters.
`timestamp_granularity`	`str \| None`	Timestamp detail level forwarded to `ai_transcribe`. Use `"word"` for per-word segments or `"speaker"` for per-speaker turns with `speaker_label`. When omitted, Snowflake returns one segment without timestamps. See module References [1].
`proxy`	`str \| None`	Proxy URL for resolving YouTube audio sources.
`auto_cleanup_stage_file`	`bool`	Whether to remove uploaded stage files after transcription.
`stage_name`	`str`	Name of the temporary Snowflake stage used for uploads.

Examples:

transcriber = ASRBasedSnowflakeAudioToText(
    session_config=SnowflakeSessionConfig(
        account="my_account",
        user="my_user",
        password="secret",
        role="ANALYST",
        warehouse="COMPUTE_WH",
    ),
)
transcripts = await transcriber.convert("path/to/audio.ogg")
print(transcripts[0].text)

transcriber = ASRBasedSnowflakeAudioToText(
    session_config={"connection_name": "my_conn"},
)
transcripts = await transcriber.convert("path/to/audio.ogg")
print(transcripts[0].text)

# Inside Snowflake (notebook / Streamlit) with word-level timestamps
transcriber = ASRBasedSnowflakeAudioToText(timestamp_granularity="word")
transcripts = await transcriber.convert("@mystage/audio.ogg")
for segment in transcripts:
    print(segment.start_time, segment.text)

# Synchronous usage via syncify
from gllm_core.concurrency import syncify

transcriber = ASRBasedSnowflakeAudioToText(session_config={"connection_name": "my_conn"})
transcripts = syncify(transcriber.convert)("path/to/audio.wav")

with ASRBasedSnowflakeAudioToText(session_config={"connection_name": "my_conn"}) as transcriber:
    transcripts = await transcriber.convert("path/to/audio.wav")

# Via modality converter builder
from gllm_multimodal.builder.modality_converter_builder import build_modality_converter
from gllm_multimodal.constants import Modality, ModalityConverterApproach, ModalityConverterTask

converter = build_modality_converter(
    Modality.AUDIO,
    Modality.TEXT,
    task_type=ModalityConverterTask.TRANSCRIPT,
    approach_type=ModalityConverterApproach.SNOWFLAKE,
    session_config={"connection_name": "my_conn"},
)
transcripts = await converter.convert("path/to/audio.wav")

Initialize the ASRBasedSnowflakeAudioToText instance.

Parameters:

Name	Type	Description	Default
`session_config`	`SnowflakeSessionConfig \| dict[str, Any] \| None`	Connection parameters for `Session.builder.configs()`. Accepts a `SnowflakeSessionConfig` instance or a dictionary of connector parameters, e.g. `{"connection_name": "my_conn"}` (References [3]) or `{"account": "...", "user": "...", "password": "..."}` (References [2]). Defaults to None, in which case `get_active_session()` is called at invoke time.	`None`
`timestamp_granularity`	`str \| None`	`ai_transcribe` timestamp granularity. Supported values are `"word"` (one segment per word with timestamps) and `"speaker"` (one segment per speaker turn with timestamps and `speaker_label`). When omitted, the entire file is transcribed as a single segment without timestamps. See module References [1]. Defaults to None.	`None`
`proxy`	`str \| None`	Proxy URL forwarded when resolving YouTube audio sources. Defaults to None.	`None`
`auto_cleanup_stage_file`	`bool`	Whether to remove uploaded files from the stage after transcription completes. Defaults to True.	`True`

Raises:

Type	Description
`ValueError`	If `timestamp_granularity` is not a supported value.

`enter()`

Enter a context manager that closes the session on exit.

Returns:

Name	Type	Description
`Self`	`Self`	This converter instance.

`exit(*_args)`

Close the Snowpark session when leaving the context manager.

This method is a no-op if the session is already closed.

Parameters:

Name	Type	Description	Default
`*_args`	`object`	The arguments passed to the context manager.	`()`

`close()`

Close the Snowpark session created by this instance.

Only closes sessions opened via session_config. When using get_active_session(), this method is a no-op because Snowflake owns the session lifecycle.

`PromptableASRBasedAudioToTranscript()`

Bases: ASRBasedAudioToTranscript, ABC

Abstract base for ASR integrations that support a prompt parameter.

Flattens str / PromptBuilder into ASR-ready prompt strings for backends that need a single text prompt (e.g. Whisper, Qwen). Subclasses must assign self.prompt in their own __init__ and call super().__init__().

Use _warn_deprecated_string_prompt (imported from this module) when accepting a plain-string prompt to emit the standardized deprecation notice.

Example

Implementing a custom promptable ASR converter

class MyAudioToText(PromptableASRBasedAudioToTranscript):
    def __init__(self, prompt: str | PromptBuilder | None = None, ...) -> None:
        super().__init__()
        self.prompt = prompt
        if isinstance(prompt, str):
            _warn_deprecated_string_prompt()

Initialize prompt cache state.

`clear_prompt_context()`

Remove cached rendered prompt components.

`set_prompt_context(prompt_context=None)`

Render and cache prompt components from the provided context.

Parameters:

Name	Type	Description	Default
`prompt_context`	`dict[str, Any] \| None`	Context variables for prompt rendering. Defaults to None.	`None`

`SnowflakeSessionConfig`

Bases: BaseModel

Validated configuration for creating a Snowflake Snowpark session.

All fields are optional individually, but the configuration must provide a connection bootstrap: connection_name, both account and user, or a pre-existing connection object for an open Python connector connection. Additional connector parameters may be supplied via extra fields and are passed to Session.builder.configs().

See module References [2] for snowflake.connector.connect() parameter names (account, user, password, role, warehouse, etc.) and References [3] for connection_name via connections.toml.

Sensitive values such as passwords, tokens, and private keys are automatically redacted from repr(), str(), and to_loggable_dict via redact_sensitive_data. Use to_session_config only when passing credentials to Snowflake, never for logging.

Attributes:

Name	Type	Description
`connection_name`	`str \| None`	Named connection from a Snowflake `connections.toml` file. See module References [3].
`account`	`str \| None`	Snowflake account identifier.
`user`	`str \| None`	Snowflake username.
`password`	`str \| None`	Snowflake password.
`role`	`str \| None`	Default role for the session.
`warehouse`	`str \| None`	Default warehouse for the session.
`database`	`str \| None`	Default database for the session.
`schema`	`str \| None`	Default schema for the session.
`authenticator`	`str \| None`	Authentication method (e.g. `"externalbrowser"`).
`private_key`	`str \| None`	PEM-encoded private key for key-pair authentication.
`private_key_file`	`str \| None`	Path to a private key file.
`private_key_file_pwd`	`str \| None`	Passphrase for the private key file.
`token`	`str \| None`	OAuth or programmatic access token.
`host`	`str \| None`	Custom host for the Snowflake account.
`port`	`int \| None`	Custom port for the Snowflake account.
`protocol`	`str \| None`	Connection protocol (e.g. `"https"`).
`connection`	`Any \| None`	An existing `snowflake.connector.Connection` to reuse when creating the Snowpark session. See module References [2].

`repr()`

Return a redacted string representation safe for logging.

Returns:

Name	Type	Description
`str`	`str`	A redacted string representation of the configuration.

`str()`

Return a redacted string representation safe for logging.

Returns:

Name	Type	Description
`str`	`str`	A redacted string representation of the configuration.

`to_loggable_dict()`

Return a copy of the configuration with sensitive values redacted for logging.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Connection parameters safe to include in logs or error messages.

`to_session_config()`

Convert the configuration to a Snowpark Session.builder.configs() dictionary.

The returned mapping uses the same keys accepted by snowflake.connector.connect() and Snowpark session creation. See module References [2].

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Connection parameters with `None` values excluded.

Note

Credential security

This method includes credential values and must not be logged directly. Use to_loggable_dict for logging or debugging output.

`validate_connection_params()`

Ensure the configuration contains enough information to connect.

Returns:

Name	Type	Description
`Self`	`Self`	The validated configuration instance.

Raises:

Type	Description
`ValueError`	If no connection bootstrap parameters are provided.

Overview

Submodules

Exported Classes

ASRBasedAudioToTranscript()

ASRBasedGoogleCloudAudioToText(credentials_json, bucket_name, language_code='id-ID', alternative_language_codes=None, model='latest_long', timeout=5 * 60, proxy=None)

ASRBasedOpenAIAudioToText(api_key, model='whisper-1', language=None, prompt=None, temperature=0, timestamp_granularity='segment', proxy=None, skip_conversion_for_formats=None, base_url=None, merge_consecutive_duplicates=False)

ASRBasedProsaAudioToText(api_key, base_url=None, url=None, model='stt-general', polling_interval=MIN_POLLING_INTERVAL)

polling_interval property writable

ASRBasedSnowflakeAudioToText(session_config=None, timestamp_granularity=None, proxy=None, auto_cleanup_stage_file=True)

__enter__()

__exit__(*_args)

close()

PromptableASRBasedAudioToTranscript()

clear_prompt_context()

set_prompt_context(prompt_context=None)

SnowflakeSessionConfig

__repr__()

__str__()

to_loggable_dict()

to_session_config()

validate_connection_params()

`ASRBasedAudioToTranscript()`

`ASRBasedGoogleCloudAudioToText(credentials_json, bucket_name, language_code='id-ID', alternative_language_codes=None, model='latest_long', timeout=5 * 60, proxy=None)`

`ASRBasedOpenAIAudioToText(api_key, model='whisper-1', language=None, prompt=None, temperature=0, timestamp_granularity='segment', proxy=None, skip_conversion_for_formats=None, base_url=None, merge_consecutive_duplicates=False)`

`ASRBasedProsaAudioToText(api_key, base_url=None, url=None, model='stt-general', polling_interval=MIN_POLLING_INTERVAL)`

`polling_interval` `property` `writable`

`ASRBasedSnowflakeAudioToText(session_config=None, timestamp_granularity=None, proxy=None, auto_cleanup_stage_file=True)`

`enter()`

`exit(*_args)`

`close()`

`PromptableASRBasedAudioToTranscript()`

`clear_prompt_context()`

`set_prompt_context(prompt_context=None)`

`SnowflakeSessionConfig`

`repr()`

`str()`

`to_loggable_dict()`

`to_session_config()`

`validate_connection_params()`