vocal_helper package

Submodules

Module contents

Vocal Helper

Producer/consumer pipeline turning a live PCM stream into diarized, transcribed utterances and (optionally) a rolling LLM summary.

Stages, all stitched by Pipeline :

  1. Source — any async iterator of PcmFrame. Three shipped : sources.from_microphone() (live mic via capture-helper), sources.from_wav_file() (replay a mono 16 kHz WAV at real-time or burst speed), sources.from_numpy_array() (in-memory PCM, for tests).

  2. VAD — Silero v5 ONNX on CPU (SileroVADStage). 32 ms windows, activity_threshold=0.5, default min_silence_ms=300.

  3. Online diarization — OnlineDiarStage. Per-segment embedding (pyannote/embedding by default, TitaNet via NeMo with backend='nemo') + cosine-distance running-mean clustering with join_threshold=0.30 (calibrated on AMI dev-slice N=8, 2026-06-30 sweep).

  4. STT — WhisperStage. pywhispercpp turbo (large-v3-turbo-q5_0), threads default 6, word timestamps on. Runs in asyncio.to_thread() so the loop is never blocked.

  5. LLM analyst (optional) — GemmaAnalystStage. Ollama serves gemma4:e4b (auto-selects the -mlx variant on Apple-Silicon) ; the stage keeps a rolling summary of everything older than recent_window_s = 60 seconds and emits a fresh SummarySnapshot after every accepted utterance.

Quickstart

>>> import asyncio, vocal_helper as voh
>>>
>>> async def main():
...     pipeline = voh.Pipeline(
...         source=lambda: voh.sources.from_microphone(),
...         config=voh.PipelineConfig(
...             diar={"backend": "pyannote"},
...             llm={"model": "gemma4:e4b"},
...         ),
...     )
...     async for ev in pipeline.run():
...         if isinstance(ev, dict) and "text" in ev:
...             print(f"[{ev['t0']:.1f}s {ev['speaker']}] {ev['text']}")
...         elif isinstance(ev, dict) and "summary" in ev:
...             print(f"--- summary ---\n{ev['summary']}")
>>>
>>> asyncio.run(main())

Usage Example

>>> import asyncio, vocal_helper as voh
>>>
>>> async def main():
...     pipeline = voh.Pipeline(
...         source=lambda: voh.sources.from_wav_file("clip.wav"),
...         config=voh.PipelineConfig(diar={"backend": "pyannote"}),
...     )
...     async for ev in pipeline.run():
...         if "text" in ev:
...             print(ev["text"])
>>>
>>> asyncio.run(main())

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

class vocal_helper.BackendPlan(mode, backend, expected_der, expected_rtf, reason)[source]

Bases: object

One routing decision: which diarizer, and its quality + speed.

Parameters:
mode

"online" (streaming OnlineDiarStage) or "offline" (whole-buffer OfflineDiarStage).

Type:

str

backend

Diarization backend — "pyannote", "nemo" or "sherpa" — to pass straight to the stage’s diar={"backend": ...} config.

Type:

str

expected_der

Representative diarization error rate (quality — lower is better) for this scenario, from the pdbms study / ADR 0002.

Type:

float

expected_rtf

Representative real-time factor (speed — < 1 is faster than real time) for this scenario.

Type:

float

reason

Human-readable justification citing the deciding measurement, surfaced to the operator so the choice is never a black box.

Type:

str

backend: str
expected_der: float
expected_rtf: float
mode: str
reason: str
class vocal_helper.DiarizedSegment[source]

Bases: TypedDict

Voiced segment with a global speaker id attached.

Emitted by vocal_helper.diar.OnlineDiarStage after each voiced segment has been embedded and matched against the running speaker centroids. speaker is a stable string id of the form "S0", "S1" — same speaker across the whole session.

pcm: ndarray[tuple[Any, ...], dtype[float32]]
sample_rate: int
speaker: str
t0: float
t1: float
class vocal_helper.EOTPair(true_turn_end_s, detector_commit_s)[source]

Bases: object

One aligned ground-truth / detector datum.

Parameters:

  • true_turn_end_s (float)

  • detector_commit_s (float)

true_turn_end_s

Absolute (mix-time) second at which the speaker actually released the floor. In practice pulled from a hand-labelled turn boundary or a bridged word-level RTTM.

Type:

float

detector_commit_s

Absolute second at which the detector emitted its end-of-turn decision. For a semantic-EOT model this is typically true_turn_end + inference_latency on hits.

Type:

float

detector_commit_s: float
true_turn_end_s: float
class vocal_helper.GemmaAnalystStage(*, model='gemma3:4b', recent_window_s=60.0, flush_every_n=5, flush_every_s=60.0, host=None, prompt_template='You are a meeting note-taker. Update the running summary below by integrating the new utterances. Keep it concise (≤ 6 bullet points), preserve speaker attributions, and drop low-signal small talk. Output only the updated summary, nothing else.\n\nCurrent summary:\n{summary}\n\nNew utterances (older → newer):\n{new_block}\n')[source]

Bases: object

Producer/consumer LLM analyst with a rolling summary.

Parameters:

  • model (str) – Ollama model tag. Default "gemma4:e4b" — Gemma 4 4B effective, the canonical light analyst across the AI Helpers suite. On Apple-Silicon, ollama resolves the -mlx variant automatically when present.

  • recent_window_s (float) – How many seconds of verbatim transcript to keep before folding into the summary. Default 60 s.

  • flush_every_n (int) – Update the summary every flush_every_n new utterances that crossed the recent window. Default 5.

  • host (str, optional) – Ollama host URL. Defaults to the OLLAMA_HOST env var or http://localhost:11434.

  • prompt_template (str) – Override the canonical summarisation prompt. Two placeholders : {summary} (current digest) and {new_block} (newly evicted utterances). Default keeps a ≤ 6-bullet meeting digest.

  • flush_every_s (float | None)

async run(inbox, outbox)[source]

Consume Utterance from inbox, push SummarySnapshot.

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

class vocal_helper.LangRegion(lang, t0, t1)[source]

Bases: object

A contiguous mono-language span of audio (seconds), ISO-639-1 lang.

Parameters:
lang: str
t0: float
t1: float
class vocal_helper.OfflineDiarStage(*, backend='pyannote', ideal_duration_s=None, overlap_s=10.0, stitch_threshold=0.35, device=None)[source]

Bases: object

Offline diarization on the full PCM buffer.

Designed for batch / file-based use : the upstream source is expected to drain end-to-end, the stage collects the full PCM, then hands it to the canonical offline backend.

Backends

  • "pyannote" — pyannote/speaker-diarization-3.1. The production default for any meeting / podcast / lecture (pdbms §10.5 : AMI dev-slice median DER 0.116, inside Bredin 2023’s 0.188 band).

  • "nemo" — NVIDIA Sortformer (the nvidia/diar_sortformer_v1 checkpoint). Better for short clips ≤ 60 s, struggles past its 90 s training cap. Auto-chunked when len > IDEAL_DURATION_S.

Long-form chunking

For inputs longer than ideal_duration_s the stage replicates the pdbms ChunkedOfflineDiarizer strategy at minimal cost :

  1. Split the audio into chunks of ideal_duration_s with overlap_s (default 10 s) of shared content at boundaries.

  2. Run the backend on each chunk.

  3. Embed each chunk-local speaker on its concatenated audio.

  4. Cluster all chunk-local embeddings via cosine AHC (stitch_threshold=0.35, the value selected in the 2026-06-30 stitch_threshold sweep on AMI dev-slice N=8 where t∈{0.30..0.40} forms the operating plateau).

The full pdbms variant adds VAD-aware cut-point selection and pink-noise pad ; vocal-helper trades these for a simpler hard-cut + zero-pad pair to keep the dependency surface small. For mission- critical AMI-style work, use pdbms.diar.offline_chunked.ChunkedOfflineDiarizer directly.

Chunking is a memory ceiling, not a quality lever. The 2026-07-14 offline map-reduce study (full stack VAD + ASR + diar on AMI, doc/studies/offline-mapreduce-study.md) found DER strictly monotone in chunk size — whole-buffer is best (median DER 0.143 vs 0.170 at 300 s, and cliffs to 0.31 / 0.50 at 120 s / 60 s as speaker fragmentation outruns the stitch) — and ASR destabilises when chunked (a long-window whisper loop drove one meeting to WER 1.17). So the pyannote default now runs whole-buffer for any realistic input (ideal_duration_s = 3600 s) and only chunks past ~1 h as a memory backstop. NeMo is the exception: its Sortformer 90 s training cap forces chunking at ideal_duration_s = 60 s.

param backend:

Backend to use. Default "pyannote". "sherpa" is the portable ONNX pipeline (community-1 seg + TitaNet-large emb, no torch) — see ADR 0002.

type backend:

“pyannote” | “nemo” | “sherpa”

param ideal_duration_s:

Whole-buffer ceiling : inputs longer than this are chunked + stitched, shorter ones run as a single call. Default depends on the backend — 3600 s for pyannote (effectively whole-buffer for any realistic meeting; chunking is a memory backstop only), 60 s for NeMo (forced by its Sortformer 90 s cap).

type ideal_duration_s:

float, optional

param overlap_s:

Overlap between adjacent chunks. Default 10 s.

type overlap_s:

float

param stitch_threshold:

Cosine-distance threshold for cross-chunk AHC stitching. Default 0.35.

type stitch_threshold:

float

param device:

Torch device for the pyannote pipeline + embedder. None (default) auto-picks CUDA > MPS > CPU. On Apple Silicon CPU is ~ 10× slower than MPS, so the auto-pick matters in practice. Has no effect on the NeMo backend.

type device:

“cpu” | “cuda” | “mps”, optional

Notes

Model weights load from the self-hosted diarization-engines bundle (see resolve_diarization_engines()) — no HuggingFace token is used or accepted.

diarize(pcm, sr)[source]

Return [(t0, t1, speaker), …] sorted by start time.

Parameters:

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

  • sr (int)

Return type:

list[tuple[float, float, str]]

async run(inbox, outbox)[source]

Drain inbox of PcmFrame ; emit DiarizedSegment.

Collects every frame until the upstream sends None, then runs diarize on the full buffer in a worker thread and emits one DiarizedSegment per identified speaker span.

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

Parameters:

  • backend (BackendName)

  • ideal_duration_s (float | None)

  • overlap_s (float)

  • stitch_threshold (float)

  • device (str | None)

class vocal_helper.OfflinePipeline(*, source, config=None)[source]

Bases: object

End-to-end batch chain : source → offline diar → ASR → LLM.

Trades the live VAD + per-segment online clustering for a single call to OfflineDiarStage on the full PCM buffer (with long-form chunking baked in). This is the best-quality path when the input is fully available — typical use cases : meeting recordings, podcasts, voicemail batches, lecture archives.

Usage

>>> import asyncio, vocal_helper as voh
>>>
>>> async def main():
...     pipeline = voh.OfflinePipeline(
...         source=lambda: voh.sources.from_wav_file("meeting.wav",
...                                                  real_time=False),
...         config=voh.OfflinePipelineConfig(
...             diar={"backend": "pyannote"},
...             asr={"language": "en"},
...             llm={"model": "gemma4:e4b"},
...         ),
...     )
...     async for ev in pipeline.run():
...         print(ev)
...
>>> asyncio.run(main())

The cadence trade

Because the four stages are decoupled by queues, each can run at its own pace : the diarizer waits for the entire PCM, the ASR streams through diarized segments as the diarizer finishes emitting them, the LLM analyst aggregates utterances as they land. The only end-to-end blocker is the offline diarizer itself.

async run()[source]

Run the batch chain ; yield every Utterance and SummarySnapshot.

Return type:

AsyncIterator[Utterance | SummarySnapshot]

subscribe_diarized(cb)[source]

Register an async callback fired for every DiarizedSegment.

Parameters:

cb (Callable[[DiarizedSegment], Awaitable[None]])

Return type:

None

subscribe_utterances(cb)[source]

Register an async callback fired for every Utterance.

Parameters:

cb (Callable[[Utterance], Awaitable[None]])

Return type:

None

Parameters:
class vocal_helper.OfflinePipelineConfig(diar=<factory>, asr=<factory>, llm=None, qsize_pcm=200, qsize_seg=32)[source]

Bases: object

Configuration object for OfflinePipeline.

Same shape as PipelineConfig minus the streaming-specific vad block — the offline diarizer ingests the full audio and relies on the backend’s own VAD / segmentation.

Parameters:
asr: dict
diar: dict
llm: dict | None = None
qsize_pcm: int = 200
qsize_seg: int = 32
class vocal_helper.OnlineDiarStage(*, backend='nemo', join_threshold=0.3, ema_alpha=0.1, min_segment_ms=500, device=None, max_speakers=None, refine_on_close=False, min_cluster_size=2, merge_threshold=None)[source]

Bases: object

Producer/consumer online speaker diarizer.

Parameters:

  • backend ("pyannote" | "nemo" | "sherpa") – Which embedding model to use. Default "pyannote". "sherpa" runs TitaNet-large through onnxruntime (no torch) — see _SherpaEmbedder.

  • join_threshold (float) – Cosine-distance threshold below which a new segment joins an existing centroid. Default 0.30 — calibrated on AMI dev-slice N=8 in 2026-06-30 stitch_threshold sweep, where the pyannote/embedding distribution exhibits a clear DER minimum at the 0.30-0.45 plateau.

  • ema_alpha (float) – Exponential-moving-average coefficient for centroid updates. Default 0.1.

  • min_segment_ms (int) – Minimum voiced-segment duration to attempt embedding. Default 500 ms (pyannote/embedding’s convolutional kernels choke on shorter inputs).

  • device ("cpu" | "cuda" | "mps", optional) – Torch device for the pyannote embedder. None (default) auto-picks CUDA > MPS > CPU. Has no effect on the NeMo backend.

  • max_speakers (int | None)

  • refine_on_close (bool)

  • min_cluster_size (int)

  • merge_threshold (float | None)

Notes

Model weights load from the self-hosted diarization-engines bundle (see resolve_diarization_engines()) — no HuggingFace token is used or accepted.

async run(inbox, outbox)[source]

Consume VoicedSegment from inbox, push DiarizedSegment.

In streaming mode (refine_on_close=False) each segment is labelled and emitted as it arrives. In batch mode (refine_on_close=True) segments are buffered, globally re-clustered when the stream closes, then emitted with corrected labels — see _run_refine().

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

class vocal_helper.PcmFrame[source]

Bases: TypedDict

One PCM frame at the configured sample rate.

Mirrors capture_helper.MicFrame and podcast_helper.PcmFrame so a producer from either library can feed this pipeline directly.

pcm: ndarray[tuple[Any, ...], dtype[float32]]
sample_rate: int
t0: float
class vocal_helper.Pipeline(*, source, config=None)[source]

Bases: object

End-to-end audio→text(→summary) producer/consumer chain.

Usage

>>> import asyncio, vocal_helper as voh
>>>
>>> async def main():
...     pipeline = voh.Pipeline(
...         source=lambda: voh.sources.from_microphone(),
...         config=voh.PipelineConfig(
...             diar={"backend": "pyannote"},
...             asr={"model": "large-v3-turbo-q5_0"},
...             llm={"model": "gemma4:e4b"},
...         ),
...     )
...     async for event in pipeline.run():
...         print(event)
...
>>> asyncio.run(main())

Yielded events are a mix of Utterance and SummarySnapshot (the latter only when llm is configured). For per-stage observation see Pipeline.subscribe_voiced() / subscribe_diarized.

async run()[source]

Run the pipeline ; yield every Utterance and SummarySnapshot.

Return type:

AsyncIterator[Utterance | SummarySnapshot]

subscribe_diarized(cb)[source]

Async callback for every DiarizedSegment after diarization.

Parameters:

cb (Callable[[DiarizedSegment], Awaitable[None]])

Return type:

None

subscribe_utterances(cb)[source]

Async callback for every Utterance after ASR.

Parameters:

cb (Callable[[Utterance], Awaitable[None]])

Return type:

None

subscribe_voiced(cb)[source]

Async callback for every VoicedSegment after VAD.

Parameters:

cb (Callable[[VoicedSegment], Awaitable[None]])

Return type:

None

Parameters:
class vocal_helper.PipelineConfig(vad=<factory>, eot=None, diar=<factory>, asr=<factory>, llm=None, qsize_pcm=200, qsize_seg=32)[source]

Bases: object

Configuration object for Pipeline.

Pass per-stage settings as dicts. The pipeline forwards them verbatim to each stage’s constructor.

Parameters:
asr: dict
diar: dict
eot: dict | None = None
llm: dict | None = None
qsize_pcm: int = 200
qsize_seg: int = 32
vad: dict
class vocal_helper.RegionVerdict(t0, t1, primary, speechbrain, sb_prob, agree)[source]

Bases: object

One region cross-checked against the independent SpeechBrain LID.

Parameters:
agree: bool
primary: str
sb_prob: float
speechbrain: str
t0: float
t1: float
class vocal_helper.SemanticEOTStage(*, eot_model='qwen2.5:3b', stt_model='large-v3-turbo-q5_0', max_merge_s=4.0, min_incomplete_ms=800, host=None)[source]

Bases: object

Producer/consumer EOT gating stage.

Parameters:

  • eot_model (str) – Ollama model used as the EOT classifier. Default qwen2.5:3b — small enough to run at ~ 50 ms / classification on Apple Silicon while broadly equivalent in capability to the LiveKit turn-detector’s 0.5B target.

  • stt_model (str) – pywhispercpp model used for the partial transcript pass. Default large-v3-turbo-q5_0 — same as the downstream WhisperStage. We could cache one instance shared by both stages in a future revision.

  • max_merge_s (float) – Maximum total duration of a merged-on-incomplete chain. After this we force-emit regardless of the classifier’s verdict.

  • min_incomplete_ms (int) – Segments shorter than this are presumed back-channels (acks / breaths) and gated by the classifier ; longer segments are emitted directly without an LLM call (cheap heuristic).

  • host (str, optional) – Ollama host URL. Defaults to the OLLAMA_HOST env var or http://localhost:11434.

async run(inbox, outbox)[source]

Consume :class:`VoicedSegment`s, gate them by semantic EOT.

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

class vocal_helper.SileroVADStage(*, activity_threshold=0.5, min_silence_ms=300, min_speech_ms=300, edge_pad_ms=200, sample_rate=16000)[source]

Bases: object

Producer/consumer Silero-VAD stage.

Parameters:

  • activity_threshold (float) – Silero score above which a window is considered voiced. Default 0.5 (canonical).

  • min_silence_ms (int) – Trailing silence required to close a voiced run. Default 300 ms.

  • min_speech_ms (int) – Reject runs shorter than this. Default 300 ms.

  • edge_pad_ms (int) – Lead / trail padding kept around the voiced span so the ASR sees a natural envelope. Default 200 ms.

  • sample_rate (int) – Required to be 16 000 — Silero v5 is trained at 16 kHz.

Notes

The stage owns a single Silero model instance (loaded lazily on first frame). The model is CPU-only ONNX so this is cheap.

async run(inbox, outbox)[source]

Consume PcmFrames from inbox, push VoicedSegments to outbox.

Exits cleanly when a None sentinel is received. Forwards None to outbox so downstream stages can stop too.

Parameters:

  • inbox (Queue)

  • outbox (Queue)

Return type:

None

class vocal_helper.SummarySnapshot[source]

Bases: TypedDict

One running summary as produced by the optional LLM analyst.

recent is the verbatim transcript of the last recent_window_s seconds ; summary is the LLM’s running digest of everything older than that.

model: str
recent: str
summary: str
t0: float
class vocal_helper.Utterance[source]

Bases: TypedDict

Transcribed diarized segment.

Emitted by vocal_helper.asr.WhisperStage. words is a list of (t0, t1, text) triplets when the underlying backend supports word-level timestamps, else a single triplet spanning the whole utterance.

language: str | None
speaker: str
t0: float
t1: float
text: str
words: list[tuple[float, float, str]]
class vocal_helper.VoicedSegment[source]

Bases: TypedDict

A contiguous run of voiced speech as detected by VAD.

Emitted by vocal_helper.vad.SileroVADStage when speech ends (i.e. after min_silence_ms of trailing silence) — the PCM buffer holds the full voiced span minus the trailing silence.

pcm: ndarray[tuple[Any, ...], dtype[float32]]
sample_rate: int
t0: float
t1: float
class vocal_helper.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.cross_check_regions(pcm, regions, sample_rate=16000, *, supported=None)[source]

Corroborate each region’s language with the independent SpeechBrain LID.

An agree=False verdict is a genuine disagreement between two models that share only the audio — a signal to inspect, not a code bug. Regions shorter than 1 s are too short to judge and pass through as agreeing.

Parameters:
Return type:

list[RegionVerdict]

vocal_helper.detect_language(pcm, *, model='large-v3-turbo-q5_0', threads=6, offset_ms=0, supported=None)[source]

Discover the actual language of pcm (single window).

Runs whisper.cpp’s auto_detect_language and, by default, returns whisper’s true argmax over its full language head — the language the input actually is, with no default and no restriction. supported is an optional, opt-in routing guard: pass the ISO-639-1 codes you can actually route and detection is re-ranked strictly within that set (the “mind the codes” guard — on a short window whisper may rank a close relative such as Galician gl above the routable Spanish es). Leave it None to let the input speak for itself.

Parameters:

  • pcm (NDArray[np.float32]) – Mono waveform window at the model’s sample rate.

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

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

  • offset_ms (int, optional) – Offset into pcm handed to whisper’s detector (default 0).

  • supported (tuple[str, ...] or None, optional) – Opt-in routable-language whitelist. None (default) means discover freely — return whisper’s own top language. A tuple restricts the answer to those codes (e.g. DEFAULT_SUPPORTED_LANGS).

Returns:

(iso_639_1_code, probability) — the discovered language and its posterior. When supported is given, the top-ranked code within it.

Return type:

(str, float)

Examples

>>> code, prob = detect_language(pcm)
>>> code
'fr'
vocal_helper.detect_language_regions(pcm, sample_rate=16000, *, model='large-v3-turbo-q5_0', threads=6, window_s=10.0, hop_s=3.0, smooth_s=6.0, min_region_s=8.0, refine_s=4.0, snap_s=1.0, supported=None)[source]

Partition pcm into mono-language regions (posterior-curve method).

A language switch has no sharp acoustic cue, so instead of hunting for a boundary we (1) sample a posterior curve over overlapping windows, (2) Gaussian-smooth it, (3) take the per-frame argmax, (4) place change points where it flips, (5) absorb sub-min_region_s regions, (6) locally refine each change point, and (7) snap it to the nearest silence. Each region’s language is discovered from the audio, never defaulted. The regions must be transcribed before committing to a language, so each is transcribed in its own. supported is an opt-in routing whitelist (None = discover freely, see detect_language()).

Empty or too-short-to-identify audio has no language to discover, so this returns an empty list (no region invented) rather than guessing one.

Parameters:
Return type:

list[LangRegion]

vocal_helper.detect_language_regions_fast(pcm, sample_rate=16000, *, conf_gate=0.5, model='large-v3-turbo-q5_0', threads=6, supported=None)[source]

Partition pcm into language regions, fast path for monolingual audio.

Most real recordings are one language end to end, yet detect_language_regions() pays for a full overlapping-window posterior scan on every file. This wrapper first runs a single cheap global detect_language() over the whole signal: when that detection clears conf_gate on a routable language the file is taken to be monolingual and returned as one region (a single whisper call instead of dozens). Only when the global detection is uncertain — a genuinely code-switched or noisy file — does it fall back to the accurate posterior-curve segmentation of detect_language_regions().

On a corpus of support-call recordings this cut per-file language identification from ~73 s to ~1 s with identical region output on the monolingual majority, while the low-confidence fallback still recovers the switches. It is a drop-in replacement for detect_language_regions() wherever code-switching is the exception rather than the rule.

Parameters:

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

  • sample_rate (int, optional) – Sample rate of pcm in Hz (default DEFAULT_SR).

  • conf_gate (float, optional) – Minimum whole-file detection probability required to accept the single-region fast path (default DEFAULT_FAST_CONF_GATE). Below it, the robust detect_language_regions() scan runs.

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

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

  • supported (tuple[str, ...] or None, optional) – Opt-in routable-language whitelist. None (default) discovers the language freely (see detect_language()).

Returns:

A single region spanning the file on the fast path, or the full multi-region partition on the fallback path. Always ≥ 1 region.

Return type:

list[LangRegion]

Examples

>>> regions = detect_language_regions_fast(pcm, 16_000)
>>> regions[0].lang
'en'
vocal_helper.detect_language_speechbrain(pcm, *, supported=None)[source]

Identify the language of pcm with SpeechBrain VoxLingua107.

An ECAPA-TDNN spoken-language classifier that shares nothing with whisper but the audio — the independent second opinion. VoxLingua107 labels are "<iso>: <English name>" (e.g. "fr: French"); the ISO-639-1 prefix is returned. By default the classifier’s true label is returned so the second opinion stays genuinely independent — it is never silently remapped to a preferred language. Pass supported only if you want an out-of-whitelist label dropped to the closest routable one.

Parameters:

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

  • supported (tuple[str, ...] or None, optional) – Opt-in routable-language whitelist. None (default) returns the true VoxLingua107 label, even if it is outside any routing set. A tuple re-ranks the classifier’s posterior within that set instead.

Returns:

(iso_639_1_code, probability).

Return type:

(str, float)

vocal_helper.eot_score(pairs, *, latency_bands_ms=(300, 600, 1200), tolerance_ms=50)

Full LiveKit-style report on a set of paired detections.

Parameters:

  • pairs (Iterable[EOTPair]) – Iterable of aligned (true_end, detector_commit) pairs.

  • latency_bands_ms (Sequence[int]) – Latency budgets to score the hang rate against. Default is LiveKit’s canonical (300, 600, 1200) ms bands.

  • tolerance_ms (int) – Acceptable early-commit slack before we count a false-cutoff. Default 50 ms sits below human turn-taking noise floor (Heldner & Edlund 2010 median gap 200 ms).

Returns:

{"n": int, "median_latency_ms": float, "false_cutoff_rate": float, "hang_rate_at_ms": {band: rate, ...}}.

Return type:

dict

vocal_helper.false_cutoff_rate(pairs, *, tolerance_s)[source]

Share of pairs where the detector fired tolerance_s early.

A “false cutoff” is a decision to end the turn while the speaker is still speaking — the perceptual failure LiveKit’s metric targets. Setting tolerance_s = 0 treats any early commit as a false cutoff ; realistic tolerances are 50-150 ms (below human turn-taking noise floor).

Parameters:
Return type:

float

vocal_helper.hang_rate(pairs, *, latency_budget_s)[source]

Share of pairs where the detector waited > latency_budget_s.

The other-side failure : the speaker finished but the agent kept waiting, producing perceived agent hesitation. latency_budget_s is the acceptable upper bound (LiveKit uses 300 / 600 / 1200 ms bands).

Parameters:
Return type:

float

vocal_helper.language_posterior_curve(pcm, sample_rate=16000, *, model='large-v3-turbo-q5_0', threads=6, window_s=10.0, hop_s=3.0, supported=None)[source]

Sample a per-window language posterior over time.

Returns (centers[T], langs[L], P[T, L]) where centers are window centre times (s), langs the candidate codes, and each row of P a posterior over langs from whisper’s auto_detect_language on a window_s window centred there. Overlapping windows (hop_s < window_s) give a finely sampled curve.

supported is opt-in: None (default) discovers freely — the posterior axis is whisper’s full language head (adopted from the first usable window) so any language can surface. A tuple restricts (and renormalises over) that routable set instead.

Parameters:
Return type:

tuple[ndarray[tuple[Any, …], dtype[float64]], list[str], ndarray[tuple[Any, …], dtype[float64]]]

async vocal_helper.run_parallel_async(upstream_input, branches)[source]

Run every async branch on the same upstream input concurrently.

Same shape as run_parallel_sync() but for coroutines. Uses asyncio.gather() so a failure in one branch cancels the others ; wrap in try/except at the branch level if per-branch fault isolation matters.

Parameters:
Return type:

dict[str, tuple[R, float]]

vocal_helper.run_parallel_sync(upstream_input, branches, *, max_workers=None)[source]

Run every branch on the same upstream input in parallel.

Parameters:

  • upstream_input (T) – The shared value every branch consumes — typically an audio buffer, a segment list, or a DiarSegment stream. Branches must not mutate it (defensive copy inside each branch is the caller’s responsibility).

  • branches (list[tuple[str, Callable[[T], R]]]) – Ordered list of (name, callable) pairs. The order in the returned dict matches the input order (Python 3.7+ dict insertion order is guaranteed).

  • max_workers (int | None) – Concurrency cap. Defaults to len(branches) which is usually what you want for a compare-N-backends sweep.

Returns:

{name: (result, wall_seconds)} in the input order.

Return type:

dict

vocal_helper.select_diarization(*, live, duration_s=None, max_speakers=None, torch_free=False, pyannote_available=True, nemo_available=True)[source]

Route to the diarization backend that the experiments justify.

Encodes the pdbms unified-study crossover (see the module docstring) as a single, testable decision over the conditions that actually move DER (quality) and RTF (speed): live-vs-batch, duration, speaker count, and torch-availability. The returned plan carries both axes explicitly.

Parameters:

  • live (bool) – True for a live stream (streaming diarizer), False for a batch file (whole-buffer offline diarizer — the reliable default for files).

  • duration_s (float or None, optional) – Audio duration in seconds when known (a file’s length is cheap to read). None means “unknown” and is treated as long-form — the safe, robust branch. Default None.

  • max_speakers (int or None, optional) – Known upper bound on the number of speakers. Used only to keep audio with more than SORTFORMER_MAX_SPEAKERS speakers off NeMo Sortformer (its hard 4-speaker cap). None = unknown. Default None.

  • torch_free (bool, optional) – True when the deployment cannot install PyTorch — routes to the sherpa onnxruntime backend regardless of length. Default False.

  • pyannote_available (bool, optional) – Whether the pyannote backend can actually run (extra installed + bundle present). When a rule would pick pyannote but it is unavailable, the router falls back rather than choosing an unrunnable backend. Default True.

  • nemo_available (bool, optional) – Whether the NeMo Sortformer backend can actually run (the nemo extra is importable). When the short/dense rule would pick nemo but it is not installed, the router falls through to the robust pyannote branch rather than emitting an unrunnable backend. Default True.

Returns:

The chosen mode + backend, its expected_der / expected_rtf, and the reason.

Return type:

BackendPlan

Examples

>>> select_diarization(live=False, duration_s=45.0, max_speakers=3).backend
'nemo'
>>> select_diarization(live=False, duration_s=1800.0).backend
'pyannote'
>>> select_diarization(live=True, torch_free=True).backend
'sherpa'
>>> # a short clip still routes to pyannote when the nemo extra is absent
>>> select_diarization(live=False, duration_s=45.0, nemo_available=False).backend
'pyannote'

Notes

The router decides which diarizer and reports its scenario quality/speed; the online/offline stages keep their own tuned knobs (join threshold, refine pass, chunk ceiling). Numbers were re-validated on this machine (studies/router_profile_validation.py, 2026-07-19) against ground truth; sherpa is from ADR 0002.

vocal_helper.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.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'