vocal_helper.asr module

vocal_helper.asr

pywhispercpp-turbo ASR stage. Consumes DiarizedSegment and emits Utterance once whisper.cpp returns.

Concurrency model

whisper.cpp is CPU-bound and blocking — running it inline in the event loop would stall every other stage. The stage delegates each transcription to asyncio.to_thread() so the loop stays free for VAD / diarization / LLM analyst work.

For very short utterances (< min_segment_ms, default 250 ms) we skip whisper entirely and emit an empty-text Utterance. Whisper’s hallucination rate at sub-300 ms inputs is unacceptable for live captioning.

Author

Warith HARCHAOUI — https://linkedin.com/in/warith-harchaoui

class vocal_helper.asr.WhisperStage(*, model='large-v3-turbo-q5_0', language='auto', threads=6, word_timestamps=True, initial_prompt='', min_segment_ms=250, batch=False, max_chunk_s=24.0, warmup=False)[source]

Bases: object

Producer/consumer pywhispercpp transcription stage.

Parameters:

  • model (str) – whisper.cpp model name. Default "large-v3-turbo-q5_0" — the cheapest word-timestamp-capable variant.

  • language (str) – ISO-639-1 code ("en", "fr", …) or "auto" for language identification.

  • threads (int) – CPU threads handed to whisper.cpp.

  • word_timestamps (bool) – Emit per-word timestamps. Default True — matches the Utterance.words contract.

  • initial_prompt (str) – Bias / vocabulary prompt passed to whisper before every transcription. Empty by default. Strongly recommended : the 2026-06-30 sweep on AMI dev-slice (studies/whisper_prompt_lang_lock.py) showed a domain- aligned prompt cuts WER by 15-25 percentage points and saves up to 39 % RTF. Good prompts name the conversational domain and a handful of expected proper nouns / technical terms, e.g. "AMI meeting transcript: remote control design, " "marketing plan, user interface, requirements."

  • min_segment_ms (int) – Drop segments shorter than this — whisper hallucinates on very short inputs.

  • batch (bool) – Offline full-throttle mode. When True the stage stops transcribing one diarized segment at a time and instead packs consecutive segments into max_chunk_s windows, running a single whisper call per window (whisper.cpp pads every call to a fixed 30 s mel, so fewer/fuller calls amortise that cost). Each returned phrase is re-mapped back to the diarized segment whose local time window contains it, so the output contract is unchanged — still one Utterance per input segment, with the diarizer’s speaker id preserved. Off by default ; the streaming Pipeline never sets it. The 2026-07-09 sweep (studies/asr_offline_batching.py) measured 6.5× lower RTF and better WER vs the per-segment path.

  • max_chunk_s (float) – Max concatenated-audio length per batched whisper call. Only used when batch=True. Default 24 s (fills the 30 s window well — the sweep’s Pareto winner). Lowering toward 12 s trades a little speed for safety if inputs switch language every few seconds (a chunk straddling an en→fr switch forces one language).

  • warmup (bool) – When True, run() runs one throwaway inference on silence before consuming the queue, moving whisper’s ~1 s first-inference stall off the streaming hot path. Off by default ; the streaming Pipeline enables it.

Notes

The pywhispercpp model is loaded lazily on the first frame so cold pipelines don’t pay the ~ 1.5 GB download until they actually need it.

async run(inbox, outbox)[source]

Consume DiarizedSegment from inbox, push Utterance.

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

vocal_helper.asr.transcribe_pcm(pcm, sr, *, model='large-v3-turbo-q5_0', language='auto', threads=6, initial_prompt='')[source]

Synchronous one-shot transcription. Loads whisper.cpp on call.

initial_prompt is the same domain-bias lever as WhisperStage — empty by default, but strongly recommended (cuts WER 15-25 pp on AMI, saves up to 39 % RTF). See transcribe_pcm_with_language() to also get the discovered language.

Parameters:

  • pcm (ndarray[tuple[Any, ...], dtype[float32]])

  • sr (int)

  • model (str)

  • language (str)

  • threads (int)

  • initial_prompt (str)

Return type:

str

vocal_helper.asr.transcribe_pcm_with_language(pcm, sr, *, model='large-v3-turbo-q5_0', language='auto', threads=6, initial_prompt='')[source]

Synchronous one-shot transcription returning text and the language.

Same as transcribe_pcm(), but also surfaces the language whisper actually used. With language="auto" (the default) that is the language discovered from the audio — so callers can report the real language of the input instead of echoing back "auto".

Parameters:

  • pcm (NDArray[np.float32]) – Mono waveform.

  • sr (int) – Sample rate of pcm in Hz.

  • model (str, optional) – whisper.cpp model name (default DEFAULT_MODEL).

  • language (str, optional) – ISO-639-1 code, or "auto" to discover it (default DEFAULT_LANGUAGE).

  • threads (int, optional) – Inference threads (default DEFAULT_THREADS).

  • initial_prompt (str, optional) – Domain-bias prompt (default DEFAULT_INITIAL_PROMPT).

Returns:

(text, detected_language). The language is None only when whisper reported none (e.g. an empty / sub-threshold segment).

Return type:

(str, str or None)

Examples

>>> text, lang = transcribe_pcm_with_language(pcm, 16_000)
>>> lang
'fr'