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. Diarization: NemotronOnnxDiar (Nemotron 3, ONNX, no torch) when its graph is in the diarization-engines bundle, else GrowingBufferDiarStage over OnlineDiarStage (per-segment embedding, best installed of NeMo TitaNet / pyannote / sherpa, plus cosine-distance running-mean clustering at diar.JOIN_THRESHOLD).

  4. STT: asr={"engine": "auto"} picks Nemotron 3.5 in ONNX (vocal_helper.nemotron_asr) when the language is to be discovered, WhisperStage (pywhispercpp large-v3-turbo-q5_0) as soon as a language, a prompt or a whisper model is given; see vocal_helper.pipeline.choose_asr_engine(). Runs in asyncio.to_thread() so the loop is never blocked.

  5. LLM analyst (optional): GemmaAnalystStage. The model is resolved by best-engine-ai-helper from the committed vocal_helper/llm.brief.yaml (via resolve_engine()) and served over Ollama or vLLM ; 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={"engine": voh.resolve_engine()},
...         ),
...     )
...     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.AccelPlan(accelerator, threads, model_kwargs=<factory>, env=<factory>, why='')[source]

Bases: object

A resolved decision: which accelerator, how many threads, and how to get it.

Parameters:
accelerator: str
apply_env()[source]

Set this plan’s environment variables, without clobbering a caller’s.

A variable already set by whoever started the process wins: an operator who exported something did it on purpose, and silently overriding that is how a benchmark measures the wrong thing.

Return type:

None

describe()[source]

One line, for a log that has to explain itself later.

Examples

>>> line = resolve("cpu").describe()
>>> line.startswith("cpu, ") and line.endswith("asked for explicitly")
True
Return type:

str

env: dict[str, str]

Environment variables that must be set for this path to engage.

model_kwargs: dict

Extra keyword arguments for the whisper.cpp model constructor.

threads: int
why: str = ''

Why this one, in one clause — so a surprising real-time factor is explainable.

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

Bases: object

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

Parameters:

  • mode (str)

  • backend (str)

  • expected_der (float)

  • expected_rtf (float)

  • reason (str)

  • sherpa_num_clusters (int | None)

mode

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

Type:

str

backend

Diarization backend — "nemotron", "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

sherpa_num_clusters

When the backend is sherpa and the exact speaker count is known, the count to pin its clustering to (FastClusteringConfig(num_clusters=...)). None leaves sherpa on auto-clustering, so meeting-audio behaviour is unchanged. This is the lever that collapses sherpa’s 2-party telephony over-segmentation (pdbms diar-study §12).

Type:

int or None

backend: str
expected_der: float
expected_rtf: float
mode: str
property rationale: str

Alias for reason (some callers log plan.rationale).

reason: str
sherpa_num_clusters: int | None = None
class vocal_helper.CloudLivePipeline(asr_ws, diar_ws, *, pre_send=None)[source]

Bases: object

Paid live transcription and diarization: one PCM stream, two providers, fused.

The topology is a fan-out, not a chain: the same audio goes to Gladia (for words) and to pyannoteAI Live-1 (for speakers), concurrently with receiving from both, and the two answers are joined at the end. It has to be this way, because Gladia does not diarize live.

Both connections are injected rather than opened here, which is what makes this class testable against a local mock server and keeps credentials out of it. A caller owns the async with websockets.connect(...) scope for each, using the URLs from vocal_helper.asr_cloud.open_gladia_live_session() and vocal_helper.diar_cloud.open_pyannoteai_live_stream().

Parameters:

  • asr_ws (Any) – Open connection to the transcription socket.

  • diar_ws (Any) – Open connection to the diarization socket.

  • pre_send (callable, optional) – async (ndarray) -> ndarray, applied to every frame before it leaves the machine. Identity by default. This is the anonymisation injection point, and the reason it lives here is a separation of duties: this library provides transport and a hook, and never personal-data policy, which belongs to the caller who knows what must be removed.

Examples

>>> import asyncio, json, numpy as np
>>> class FakeWs:
...     def __init__(self, replies): self.replies, self.sent = replies, []
...     async def send(self, payload): self.sent.append(payload)
...     def __aiter__(self):
...         async def gen():
...             for r in self.replies: yield json.dumps(r)
...         return gen()
>>> async def demo():
...     asr = FakeWs([{"type": "final", "words": [
...         {"word": "oui", "time_begin": 0.1, "time_end": 0.4}]}])
...     diar = FakeWs([{"type": "diarization_speaker_start",
...                     "data": {"timestamp": 0.0, "speaker": "S0"}},
...                    {"type": "diarization_speaker_end",
...                     "data": {"timestamp": 1.0, "speaker": "S0"}}])
...     async def source():
...         yield np.zeros(1600, dtype=np.float32)
...     out = await CloudLivePipeline(asr, diar).run(source())
...     return [(u["speaker"], u["text"]) for u in out], len(asr.sent), len(diar.sent)
>>> asyncio.run(demo())
([('S0', 'oui')], 1, 1)
async run(pcm_source, *, stop=None)[source]

Fan pcm_source out to both providers and return the fused utterances.

Parameters:

  • pcm_source (async iterator of ndarray) – Mono float32 chunks, from any of vocal_helper.sources.

  • stop (asyncio.Event, keyword-only, optional) – Set externally — a “stop” button, a maximum-duration timer — to end the session without waiting for the source to exhaust itself. Neither a microphone nor an open-ended live URL ends on its own, so a live session without this has no way to finish.

Returns:

The fused record. Partial transcripts are excluded: they get revised, and a record must not contain retracted words.

Return type:

list of CloudUtterance

Notes

Three tasks run concurrently: sending audio, draining transcripts, draining speaker events. The two drain loops end when their socket closes, and this method does not close either one — the caller owns that scope. End of stream is signalled to the diarizer (which documents a message for it); Gladia documents no equivalent, so closing the socket is what ends it.

class vocal_helper.CloudOfflineConfig(transcriber='gladia', diarizer='pyannoteai', diarizer_model='precision-3', languages=None, code_switching=True, glossary=None, num_speakers=None, min_speakers=None, max_speakers=None, overlap_s=120.0)[source]

Bases: object

What a paid offline run should do.

Parameters:

  • transcriber (str, optional) – "gladia" (default, solaria-3) or "elevenlabs" (Scribe).

  • diarizer (str, optional) – "pyannoteai", or "" to keep the transcriber’s own speakers. Gladia diarizes its pre-recorded output, and Scribe bundles diarization too, but on a crowded panel the hosted diarizer is what the money is for — so it is on by default.

  • diarizer_model (str, optional) – Defaults to the provider’s most accurate batch model. Named rather than left to the client library’s default, which is the previous generation: on a paid call, accepting a default is not a neutral act.

  • languages (list of str, optional) – Expected languages. None lets the provider detect.

  • code_switching (bool, optional) – Whether one recording may switch between the declared languages.

  • glossary (list of str, optional) – Domain terms and proper nouns to bias toward. Never part of the billing ledger key, so editing it and re-running the same audio costs nothing.

  • num_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

  • min_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

  • max_speakers (int, optional) – Diarization hints. Prefer ``min_speakers``: a roster is a floor, not a cap, and an exact count will merge or split to hit the number even when the audio disagrees.

  • overlap_s (float, optional) – Shared audio between chunks when the media exceeds the provider’s cap.

Raises:

ValueError – On an unknown transcriber or diarizer, at construction time rather than halfway through a billed run.

Examples

>>> CloudOfflineConfig().transcriber
'gladia'
>>> CloudOfflineConfig(diarizer="").diarizer
''
>>> CloudOfflineConfig(diarizer="whisper")
Traceback (most recent call last):
    ...
ValueError: unknown diarizer 'whisper'; choose from '' or pyannoteai
code_switching: bool = True
diarizer: str = 'pyannoteai'
diarizer_model: str = 'precision-3'
glossary: list[str] | None = None
languages: list[str] | None = None
max_speakers: int | None = None
min_speakers: int | None = None
num_speakers: int | None = None
overlap_s: float = 120.0
transcriber: str = 'gladia'
class vocal_helper.CloudOfflinePipeline(config=None)[source]

Bases: object

Paid offline transcription and diarization: a file in, utterances out.

Unlike the local pipelines, there is no PCM stream to drive: the providers take a whole file, so this class is a plain call rather than an async generator. The output type is the same either way, so a consumer does not branch on the mode.

Parameters:

config (CloudOfflineConfig, optional) – Defaults to Gladia solaria-3 plus pyannoteAI, which is the combination this suite measured and bought.

Examples

>>> pipeline = CloudOfflinePipeline()
>>> pipeline.config.transcriber, pipeline.config.diarizer
('gladia', 'pyannoteai')
>>> pipeline.engine_name()
'gladia:solaria-3+pyannoteai:precision-3'
>>> CloudOfflinePipeline(CloudOfflineConfig(diarizer="")).engine_name()
'gladia:solaria-3'
engine_name()[source]

Name the engines that produced a result, for the record it ends up in.

Returns:

e.g. "gladia:solaria-3+pyannoteai:precision-3". The model, not just the provider: two runs through the same provider on different generations are not the same measurement, and a record that hid which one produced it could not be compared against itself a year later.

Return type:

str

run(audio)[source]

Transcribe and diarize audio, returning what the engines said.

Raw, deliberately. Rows come back as the providers produced them: a speaker the diarizer discovered for four words is still here, a block that merges thirty seconds of speech is still one row, and a row whose times collapsed to an instant still has none. Those are facts about the engines, and a library that quietly edited them would be deciding, on a caller’s behalf, which voices count.

The next step is explicit, and almost always wanted:

rows = CloudOfflinePipeline(config).run(audio)
rows, speakers = voh.consolidate(rows, min_speaker_s=120.0)

See vocal_helper.turns.consolidate() for the five rules it applies and the complaint behind each.

Parameters:

audio (str) – Media path. Any container; the waveform is what gets uploaded.

Returns:

In time order, each naming the engine that produced it.

Return type:

list of CloudUtterance

Raises:

RuntimeError – When a required credential is missing, or a provider refuses the call. Never a silent fallback to a local engine.

class vocal_helper.CloudUtterance[source]

Bases: Utterance

An Utterance from a hosted engine, carrying what it knows extra.

Subclassing with total=False makes every key below optional, so a CloudUtterance is accepted anywhere an Utterance is expected and no existing consumer has to change. That matters: Pipeline and OfflinePipeline have live callers, and widening their shared output type would have been a breaking change dressed up as an improvement.

Why the extra keys exist

Utterance.words is a list of (t0, t1, text) triplets, which has no room for a per-word confidence. Hosted transcribers return one, and it is the signal a downstream repair pass needs in order to know which spans to re-read rather than re-reading everything. So it travels alongside, index-aligned with words, instead of changing the triplet that callers already destructure.

engine: str

Which engine produced this utterance, e.g. "gladia:solaria-3".

A record names the engine that wrote it. Without this, a result from a degraded path is indistinguishable from a good one, and the reader blames the tool.

language: str | None
speaker: str
t0: float
t1: float
text: str
word_confidence: list[float]

Per-word confidence in [0, 1], index-aligned with words.

words: list[tuple[float, float, str]]
class vocal_helper.ConsistencyReport(drift=<factory>, collision=<factory>, assigned_label=<factory>, appearances=<factory>, label_time=<factory>)[source]

Bases: object

Identity quality, independent of boundary quality.

Parameters:
appearances: dict[str, list[tuple[float, float, str]]]

Per true speaker, their reference intervals with the label each one received.

assigned_label: dict[str, str]

Per true speaker, the label the one-to-one assignment gave them.

A speaker absent from this mapping received no label of their own, which happens when the hypothesis has fewer labels than there are speakers — a collapse.

collision: dict[str, float]

one minus the share of its time on its ASSIGNED speaker.

Type:

Per hypothesis label

describe()[source]

One line, for a log or a study table.

Return type:

str

drift: dict[str, float]

one minus the share of their speech on their ASSIGNED label.

Type:

Per true speaker

label_time: dict[str, float]

Seconds of reference speech each hypothesis label covers.

long_range_agreement(*, gap_s)[source]

Share of same-speaker appearance pairs, at least gap_s apart, that kept a label.

Parameters:

gap_s (float, keyword-only) – Minimum silence between the end of one appearance and the start of the next, for the pair to count. A large value isolates exactly the failure this module exists for: identity lost across a long absence.

Returns:

In [0, 1], or 1.0 when no pair is that far apart — nothing was asked of the diarizer at that range, so nothing is held against it.

Return type:

float

Examples

>>> reference = [(0.0, 5.0, "a"), (100.0, 105.0, "a")]
>>> kept = speaker_consistency(reference, [(0.0, 5.0, "S0"), (100.0, 105.0, "S0")])
>>> kept.long_range_agreement(gap_s=60.0)
1.0
>>> kept.long_range_agreement(gap_s=600.0)   # nothing that far apart
1.0
property misattributed_share: float

Share of labelled speech time that sits on the wrong speaker.

The time-weighted companion to worst_collision, and the number to quote when comparing two diarizers. A handful of tiny orphan clusters barely moves it; half the audio under a wrong name sends it to 0.5.

Returns:

In [0, 1]. Zero means every labelled second belongs to the speaker its label was assigned to.

Return type:

float

Examples

A tiny spurious label next to an otherwise perfect answer: the maximum saturates, the share stays honest.

>>> reference = [(0.0, 100.0, "alice")]
>>> almost = [(0.0, 99.9, "S0"), (99.9, 100.0, "S1")]
>>> report = speaker_consistency(reference, almost)
>>> report.worst_collision
1.0
>>> round(report.misattributed_share, 4)
0.001
property worst_collision: float

The most ambiguous label.

Saturates, and that is a known limit of the maximum. One spurious label holding a tenth of a second is assigned to nobody, scores 1.0 on its own, and pins this number at 1.0 whatever the rest of the recording looks like. It answers “is any label meaningless?” — useful as a tripwire, useless as a measure of how bad things are. For that, read misattributed_share.

property worst_drift: float

The least well-held speaker. One bad voice ruins a transcript, so the maximum is the honest summary and the mean would hide it.

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[int, ...], 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.Exemplar(vector, confidence)[source]

Bases: object

Une observation gardée, et ce qu’elle vaut.

Parameters:

  • vector (ndarray[tuple[int, ...], dtype[float32]])

  • confidence (float)

vector

L’empreinte, normalisée L2.

Type:

NDArray[np.float32]

confidence

Les secondes de parole propre dont elle a été tirée. C’est la seule mesure de qualité disponible sans vérité terrain, et elle est juste : une empreinte calculée sur huit secondes où le locuteur est seul vaut mieux qu’une calculée sur quatre dixièmes de seconde.

Type:

float

confidence: float
vector: ndarray[tuple[int, ...], dtype[float32]]
class vocal_helper.ExemplarClusterer(policy=<factory>)[source]

Bases: object

Rattache les locuteurs locaux d’une fenêtre, sans jamais moyenner.

L’interface est celle de IncrementalClusterer — assign prend des empreintes et des durées propres, rend des étiquettes — donc les deux sont interchangeables dans IncrementalDiarStage, et comparables à extraction identique.

Examples

Deux voix orthogonales ouvrent le flux, chacune reçoit son étiquette, et la correspondance tient quand elles se présentent dans l’autre sens.

>>> import numpy as np
>>> e0 = np.array([1.0, 0.0, 0.0], dtype=np.float32)
>>> e1 = np.array([0.0, 1.0, 0.0], dtype=np.float32)
>>> c = ExemplarClusterer()
>>> c.assign(np.stack([e0, e1]), [5.0, 5.0])
['S0', 'S1']
>>> c.assign(np.stack([e1, e0]), [5.0, 5.0])
['S1', 'S0']

Voici ce qu’un centroïde ne peut pas faire. La même voix revient dans un autre régime, assez loin pour être gardée comme second exemplaire mais assez près pour être reconnue :

>>> other = np.array([0.8, 0.6, 0.0], dtype=np.float32)
>>> c.assign(other[None, :], [5.0])
['S0']
>>> c.exemplar_counts
[2, 1]

Ce second régime est désormais reconnu exactement, parce que la mesure est la distance au meilleur point et non à une moyenne qui serait tombée entre les deux :

>>> float(round(c.distances(other[None, :])[0, 0], 6))
0.0

Un quasi-doublon ne prend pas de place, il renforce la confiance :

>>> c.assign((other + 0.01 * e0)[None, :], [5.0])
['S0']
>>> c.exemplar_counts
[2, 1]

Et la fusion globale, que l’absence de moyenne rend définissable. Deux voix quasi identiques vues en même temps sont forcément séparées par la contrainte d’injectivité — c’est le mécanisme même de la sur-segmentation :

>>> d = ExemplarClusterer()
>>> near = np.array([1.0, 0.1, 0.0], dtype=np.float32)
>>> d.assign(np.stack([e0, near]), [5.0, 5.0])
['S0', 'S1']
>>> d.merge()
{'S1': 'S0'}
>>> d.speaker_count
1
Parameters:

policy (ExemplarPolicy)

assign(embeddings, clean_durations)[source]

Étiqueter les locuteurs locaux d’une fenêtre.

Parameters:

  • embeddings (NDArray[np.float32]) – (k, d) — une empreinte par locuteur local actif.

  • clean_durations (list of float) – k durées de parole propre, qui servent de confiance.

Returns:

k étiquettes, dans l’ordre des lignes reçues.

Return type:

list of str

Raises:

ValueError – Si les deux entrées n’ont pas la même longueur, ou si les empreintes ne sont pas une matrice.

distances(embeddings)[source]

Distance de chaque empreinte au meilleur point de chaque locuteur.

Avec neighbours > 1, c’est le k-ième plus proche et non le plus proche : un seul exemplaire aberrant ne suffit alors plus à rapprocher deux locuteurs.

Parameters:

embeddings (NDArray[np.float32]) – (k, d), pas nécessairement normalisées.

Returns:

(k, m) de distances cosinus, m étant le nombre de locuteurs.

Return type:

NDArray[np.float32]

property exemplar_counts: list[int]

Combien d’exemplaires chaque locuteur garde.

merge()[source]

Réunir les locuteurs que la mémoire révèle identiques.

C’est ce que l’absence de moyenne rend possible. Deux ensembles d’exemplaires se comparent — on regarde si leurs observations se recouvrent — là où deux moyennes mobiles avaient déjà perdu de quoi le savoir. La passe est exacte, ne recalcule aucune empreinte, et répare après coup la sur-segmentation que la décision locale ne pouvait pas éviter : la contrainte d’injectivité doit séparer deux locuteurs locaux simultanés, même quand la segmentation s’est trompée et qu’il n’y en avait qu’un.

Returns:

{ancienne étiquette: étiquette survivante} pour les locuteurs absorbés. Vide si rien ne fusionne. À l’appelant de réécrire ses propres sorties avec cette table.

Return type:

dict

policy: ExemplarPolicy
property speaker_count: int

Nombre de locuteurs découverts.

property speaker_labels: list[str]

Les étiquettes, dans l’ordre de découverte.

class vocal_helper.ExemplarPolicy(rho_update_s=0.4, delta_new=0.6, reference='meilleur', neighbours=1, max_exemplars=12, diversity_margin=0.1, max_speakers=None, merge=True, merge_threshold=None)[source]

Bases: object

Les réglages du regroupement par exemplaires.

Parameters:

  • rho_update_s (float)

  • delta_new (float)

  • reference (str)

  • neighbours (int)

  • max_exemplars (int)

  • diversity_margin (float)

  • max_speakers (int | None)

  • merge (bool)

  • merge_threshold (float | None)

rho_update_s

Confiance minimale, en secondes de parole propre, pour qu’une observation soit gardée. En dessous, elle est quand même étiquetée — on ne perd personne — mais elle n’entre pas dans la mémoire du locuteur.

Type:

float

delta_new

Distance au meilleur point au-delà de laquelle on crée un locuteur plutôt que de forcer une jointure. Reste à 0,6 : trois réunions d’AMI donnent 0,750 / 0,600 / 0,600 comme optimum, et aucune normalisation n’a réduit cette dispersion.

Type:

float

reference

Comment mesurer la ressemblance à un locuteur, sans jamais moyenner.

"meilleur"

Distance au neighbours-ième exemplaire le plus proche. Garde la multimodalité, mais c’est un minimum sur un ensemble, donc un estimateur optimiste : il diminue mécaniquement quand l’ensemble grandit, si bien qu’un seul exemplaire bruité suffit à capturer un locuteur. Mesuré sur AMI : taux d’erreur 0,529 contre 0,299 pour le centroïde, et dérive 0,859. C’est ce biais que neighbours corrige partiellement, 0,529 puis 0,378 puis 0,363 pour k=1,2,3.

"confiance"

Distance à l’unique exemplaire de meilleure confiance. Un point observé, donc aucune moyenne, et pas de biais optimiste non plus puisque le nombre d’exemplaires n’entre pas dans la mesure.

"médoïde"

Distance au médoïde — l’exemplaire le plus central de l’ensemble, celui de plus forte similarité totale aux autres. Également un point observé, et robuste à un intrus isolé, qui ne peut pas devenir le plus central.

Type:

str

neighbours

Combien de voisins regarder, pour reference="meilleur" seulement. 1 est le meilleur point pur, donc un lien simple, sensible à un exemplaire aberrant faisant pont entre deux locuteurs. 2 ou 3 coupent le pont tout en gardant la multimodalité.

Type:

int

max_exemplars

Taille de la mémoire par locuteur. C’est ce qui garde le coût borné en direct : la distance à un locuteur coûte au plus max_exemplars produits scalaires.

Type:

int

diversity_margin

En dessous de cette distance à un exemplaire déjà gardé, une observation est un quasi-doublon : elle renforce la confiance de l’existant au lieu de prendre une place. C’est ce qui empêche la mémoire de se remplir de dix copies du même instant.

Type:

float

max_speakers

Plafond dur du nombre de locuteurs. None laisse le compte libre, c’est-à-dire deviné.

Type:

int, optional

merge

Faire ou non la reprise globale. True par défaut, parce que c’est l’apport principal de la représentation par exemplaires. Le mettre à False sert à isoler les deux effets dans une mesure : l’absence de moyenne d’une part, la reprise globale de l’autre.

Type:

bool

merge_threshold

Distance entre deux ensembles d’exemplaires en dessous de laquelle ExemplarClusterer.merge() les réunit. None reprend delta_new, ce qui est cohérent : deux locuteurs plus proches que le seuil de création n’auraient jamais dû être deux.

Type:

float, optional

delta_new: float = 0.6
diversity_margin: float = 0.1
max_exemplars: int = 12
max_speakers: int | None = None
merge: bool = True
merge_threshold: float | None = None
neighbours: int = 1
reference: str = 'meilleur'
rho_update_s: float = 0.4
class vocal_helper.GemmaAnalystStage(*, engine, recent_window_s=60.0, flush_every_n=5, flush_every_s=60.0, 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:

  • engine (dict) – Resolved engine descriptor from best_engine_ai_helper.ensure(<vocal_helper package dir>) — it names the backend (Ollama / vLLM), the base URL, and the text model to serve. No default model is baked in here.

  • 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.

  • 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.GrowingBufferDiarStage(backend='auto', policy=<factory>, offline=None, online=None, on_revision=None)[source]

Bases: object

Diarisation en direct par re-diarisation périodique du tampon entier.

L’étage accumule l’audio et, aux instants que la politique fixe, le fait diariser en entier par un moteur différé. Les étiquettes de chaque passe sont rattachées à celles de la précédente par recouvrement temporel, de sorte que le lecteur voie des identités stables ; les changements sont rendus comme des révisions explicites.

Parameters:

  • backend (str) – Moteur différé, passé à OfflineDiarStage. "auto" prend le meilleur installé.

  • policy (GrowingBufferPolicy, optional) – Le calendrier des passes.

  • offline (object, optional) – Un diariseur déjà construit, qui doit exposer diarize(pcm, sr) -> [(t0, t1, étiquette), …]. Sert aux tests et permet de brancher un autre moteur sans toucher à cette classe.

  • online (object, optional) – L’étage en ligne qui donne l’étiquette provisoire, en une demi- seconde, le temps que la première passe arrive. None ne publie rien avant la première passe.

  • on_revision (callable, optional) – Appelée pour chaque étiquette démentie par une passe. Sans elle, les corrections sont calculées et perdues.

Examples

Un diariseur de pacotille suffit à montrer le mécanisme : les étiquettes qu’il rend changent de nom à chaque passe, et l’étage les stabilise.

>>> import numpy as np
>>> class Flighty:
...     def __init__(self): self.calls = 0
...     def diarize(self, pcm, sr, **kw):
...         self.calls += 1
...         tag = f"local{self.calls}"
...         return [(0.0, len(pcm) / sr, tag)]
>>> stage = GrowingBufferDiarStage(
...     offline=Flighty(), policy=GrowingBufferPolicy(first_s=1.0, ratio=2.0)
... )
>>> segments = stage.diarize(np.zeros(16000 * 4, dtype=np.float32), 16000)
>>> [label for _, _, label in segments]
['P0']
>>> stage.speaker_labels
['P0']

Trois passes ont eu lieu, et le locuteur n’a été découvert qu’une fois.

>>> stage.passes
3
backend: str = 'auto'
diarize(pcm, sr, *, num_speakers=None, min_speakers=None, max_speakers=None)[source]

Dérouler un tampon comme un direct, et rendre l’état final.

Utile pour mesurer : la dernière passe voit tout, donc le résultat est celui du différé pleine fenêtre, obtenu en ayant payé le prix du direct. Pour un vrai flux, appeler refresh() aux instants que GrowingBufferPolicy.times() indique et lire les révisions.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence.

  • num_speakers (int, optional) – Transmis au moteur différé à chaque passe.

  • min_speakers (int, optional) – Transmis au moteur différé à chaque passe.

  • max_speakers (int, optional) – Transmis au moteur différé à chaque passe.

Returns:

[(t0, t1, locuteur), …], la même forme que les autres étages.

Return type:

list of tuple

property discovered_speakers: int

Combien de locuteurs ont été découverts.

offline: Any = None
on_revision: Any = None
online: Any = None
property passes: int

Combien de passes de re-diarisation ont eu lieu.

policy: GrowingBufferPolicy
refresh(pcm, sr, **hints)[source]

Rediariser tout le tampon, et rendre ce que ça change.

Parameters:

  • pcm (NDArray[np.float32]) – Tout l’audio depuis le début, pas seulement le nouveau.

  • sr (int) – Sa fréquence.

  • **hints – Transmis au moteur différé, par exemple max_speakers.

Returns:

Les passages dont l’étiquette change par rapport à ce qui avait été publié. Vide à la première passe, puisque rien n’avait été dit.

Return type:

list of LabelRevision

reset()[source]

Oublier le flux, en gardant la configuration.

Return type:

None

async run(inbox, outbox)[source]

Consommer des segments voisés, publier aussitôt, corriger ensuite.

Même contrat que l’étage en ligne, donc c’est un remplacement direct : un consommateur qui lit outbox n’a rien à changer. Ce qui est nouveau, c’est qu’une étiquette déjà publiée peut être démentie, et on_revision le dit — un affichage qui ignore les révisions montre des étiquettes dont on lui a signalé qu’elles sont fausses.

L’étage garde tout l’audio voisé, puisque chaque passe rediarise depuis le début : à 16 kHz en flottant simple cela fait 64 ko par seconde de parole, soit environ 230 Mo pour une heure de réunion. C’est le prix de l’exactitude, et il est borné par la durée de la séance.

Parameters:

  • inbox – Files d’attente de segments voisés puis diarisés, comme pour les autres étages.

  • outbox – Files d’attente de segments voisés puis diarisés, comme pour les autres étages.

Return type:

None

segments()[source]

La diarisation publiée, telle que la dernière passe l’a laissée.

Return type:

list[tuple[float, float, str]]

property speaker_labels: list[str]

Les locuteurs publiés, dans l’ordre de découverte.

class vocal_helper.GrowingBufferPolicy(first_s=60.0, ratio=1.6, max_period_s=None, min_speaker_s=0.0, gap_s=0.8)[source]

Bases: object

Quand rafraîchir, et jusqu’où laisser le retard grandir.

Parameters:
first_s

Instant de la première passe. En dessous, le diariseur n’a pas de quoi travailler et rendrait un seul locuteur.

Type:

float

ratio

Facteur d’espacement. 1,6 mesure un coût de 2,6 à 3,3 fois l’audio pour un retard valant 0,375 fois la séance ; descendre vers 1,2 raccourcit le retard et monte le coût.

Type:

float

max_period_s

Attente maximale entre deux passes. None laisse le retard croître avec la séance, ce qui va pour une réunion et non pour une journée.

Type:

float, optional

min_speaker_s

Temps de parole minimal pour qu’une étiquette soit publiée. Écarte les locuteurs d’une seconde que le diariseur invente sur un bruit.

Type:

float

gap_s

Silence considéré comme une séparation de tours, pour récolter le bruit de fond qui reconstitue les creux entre segments voisés.

Type:

float

first_s: float = 60.0
gap_s: float = 0.8
max_period_s: float | None = None
min_speaker_s: float = 0.0
ratio: float = 1.6
times(duration_s)[source]

Les instants de rafraîchissement pour cette durée.

Parameters:

duration_s (float)

Return type:

list[float]

worst_staleness_s(duration_s)[source]

Le plus long délai avant qu’un passage reçoive son étiquette définitive.

Examples

>>> GrowingBufferPolicy(first_s=60.0, ratio=1.6).worst_staleness_s(1200.0)
377.48736
Parameters:

duration_s (float)

Return type:

float

class vocal_helper.Hardware(system, machine, apple_silicon, performance_threads, whisper_backends=<factory>)[source]

Bases: object

What this machine offers, as far as transcription is concerned.

Parameters:
apple_silicon: bool
describe()[source]

One line naming the machine and what it can run.

Returns:

e.g. "Darwin/arm64, 8 fast threads, backends: metal".

Return type:

str

Examples

>>> "threads" in probe().describe()
True
machine: str

Processor architecture, e.g. "arm64".

performance_threads: int

Cores worth giving to whisper.cpp — see the module docstring on why not all.

system: str

"Darwin", "Linux", "Windows".

whisper_backends: frozenset[str]

Which of {"metal", "coreml", "ane", "cuda"} the installed build carries.

class vocal_helper.IncrementalClusterer(policy=<factory>)[source]

Bases: object

Rattache les locuteurs locaux d’une fenêtre aux locuteurs globaux.

L’objet est volontairement sans audio et sans modèle : il reçoit des empreintes et des durées, il rend des étiquettes. Tout le comportement décrit dans l’en-tête du module se teste donc directement.

Examples

Deux voix orthogonales ouvrent le flux ; chacune reçoit son étiquette.

>>> import numpy as np
>>> e0 = np.array([1.0, 0.0], dtype=np.float32)
>>> e1 = np.array([0.0, 1.0], dtype=np.float32)
>>> c = IncrementalClusterer()
>>> c.assign(np.stack([e0, e1]), [1.0, 1.0])
['S0', 'S1']

La fenêtre suivante les présente dans l’autre sens. La contrainte d’injectivité et l’algorithme hongrois les suivent sans les confondre — c’est ce qu’un plus-proche-centroïde glouton ne garantit pas.

>>> c.assign(np.stack([e1, e0]), [1.0, 1.0])
['S1', 'S0']

Une empreinte mélangée, à mi-chemin des deux voix, reste sous delta_new : elle est rattachée au plus proche. Mais sa part propre est trop courte, donc elle n’écrit dans aucun centroïde.

>>> blend = np.array([0.7071, 0.7071], dtype=np.float32)
>>> c.assign(blend[None, :], [0.1])
['S0']
>>> c.updates
4

Une troisième voix, elle, est trop loin des deux premières : le flux découvre un locuteur de plus plutôt que de forcer la jointure.

>>> e2 = np.array([-1.0, 0.0], dtype=np.float32)
>>> c.assign(e2[None, :], [1.0])
['S2']
>>> c.speaker_count
3

Sous plafond, le même flux ne crée rien et se rabat sur le plus proche.

>>> capped = IncrementalClusterer(policy=IncrementalPolicy(max_speakers=2))
>>> capped.assign(np.stack([e0, e1]), [1.0, 1.0])
['S0', 'S1']
>>> capped.assign(e2[None, :], [1.0])
['S1']
>>> capped.speaker_count
2
Parameters:

policy (IncrementalPolicy)

assign(embeddings, clean_durations)[source]

Étiqueter les locuteurs locaux d’une fenêtre.

Parameters:

  • embeddings (NDArray[np.float32]) – (k, d) — une empreinte par locuteur local actif. Normalisée ici, l’appelant n’a pas à s’en soucier.

  • clean_durations (list of float) – k durées, en secondes, de la part de parole propre de chaque locuteur local — les trames où il est actif et seul. C’est ce que rho_update filtre.

Returns:

k étiquettes globales, dans l’ordre des lignes reçues.

Return type:

list of str

Raises:

ValueError – Si les deux entrées n’ont pas la même longueur.

centroids()[source]

Les centroïdes courants, (m, d) normalisés L2.

Return type:

ndarray[tuple[int, …], dtype[float32]]

policy: IncrementalPolicy
property speaker_count: int

Nombre de locuteurs globaux découverts jusqu’ici.

property speaker_labels: list[str]

Les étiquettes globales, dans l’ordre de leur découverte.

property updates: int

Nombre de mises à jour de centroïde acceptées par rho_update.

class vocal_helper.IncrementalDiarStage(*, segmenter='auto', embedder='auto', policy=None, window_s=5.0, step_s=2.5, local_max_speakers=3, min_embed_s=0.5)[source]

Bases: object

Diarisation en ligne par fenêtre glissante, segmenteur au choix.

L’étage fait glisser une fenêtre sur le flux, fait segmenter localement chaque fenêtre par le backend qu’on lui donne, embarque la part propre de chaque locuteur local, puis confie le rattachement à IncrementalClusterer. Ni le segmenteur ni l’encodeur ne sont imposés : c’est tout l’intérêt d’avoir porté l’algorithme plutôt que de dépendre de diart, dont le segmenteur local est câblé sur pyannote.

Parameters:

  • segmenter (str) – Backend de segmentation locale, passé à OfflineDiarStage. "auto" (défaut) prend le meilleur backend installé ; "sherpa" donne un chemin sans torch.

  • embedder (str) – Backend d’empreinte vocale, passé à OnlineDiarStage. "auto" par défaut.

  • policy (IncrementalPolicy, optional) – Les réglages du regroupement. Défauts de diart.

  • window_s (float) – Longueur de la fenêtre glissante. 5 s, la valeur de référence.

  • step_s (float) – Pas de la fenêtre. La méthode d’origine avance de 0,5 s parce que son segmenteur local est un petit modèle dédié ; les nôtres sont des diariseurs complets, donc le défaut est ici 2,5 s — un recouvrement de moitié, qui garde le vote de recouvrement sans rendre le RTF absurde.

  • local_max_speakers (int) – Plafond passé au segmenteur local. 3, la limite usuelle d’un modèle de segmentation locale, et un plafond est ce qui empêche la prolifération d’étiquettes.

  • min_embed_s (float) – Part propre minimale pour tenter une empreinte. Sous ce seuil, le locuteur local est ignoré : mieux vaut ne rien dire que de verser une empreinte de bruit dans le regroupement.

diarize(pcm, sr, *, num_speakers=None, min_speakers=None, max_speakers=None)[source]

Diariser un tampon entier par fenêtres glissantes.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence d’échantillonnage.

  • num_speakers (int, optional) – Acceptés pour la parité d’interface avec les autres étages. Le regroupement incrémental devine le compte par delta_new ; un compte exact n’a pas de sens au fil de l’eau, puisque les locuteurs arrivent les uns après les autres.

  • min_speakers (int, optional) – Acceptés pour la parité d’interface avec les autres étages. Le regroupement incrémental devine le compte par delta_new ; un compte exact n’a pas de sens au fil de l’eau, puisque les locuteurs arrivent les uns après les autres.

  • max_speakers (int, optional) – Plafond dur, qui l’emporte sur celui de la politique. C’est le réglage utile : la mesure du 2026-11-01 montre qu’un plafond vaut le compte exact sur le moteur différé.

Returns:

[(t0, t1, locuteur), …] trié par début, la même forme que tous les autres étages de diarisation du paquet.

Return type:

list of tuple

property discovered_speakers: int

Nombre de locuteurs que le flux a découverts.

extract(pcm, sr)[source]

Segmenter et embarquer chaque fenêtre, sans encore regrouper.

C’est la moitié coûteuse du travail, et la seule qui touche aux modèles. Son résultat ne dépend d’aucun réglage du regroupement, donc il se garde et se rejoue — c’est ce qui rend un balayage abordable.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence d’échantillonnage.

Returns:

Une entrée par fenêtre ayant livré au moins une empreinte.

Return type:

list of WindowFeatures

label(windows, duration_s)[source]

Regrouper des fenêtres extraites en une diarisation.

La moitié bon marché : du numpy et un algorithme hongrois. Repart du regroupeur courant, donc appeler reset() avant de rejouer les mêmes fenêtres sous un autre réglage.

Parameters:
Returns:

[(t0, t1, locuteur), …] trié par début.

Return type:

list of tuple

reset()[source]

Oublier les locuteurs découverts, en gardant la politique.

Examples

>>> import numpy as np
>>> stage = IncrementalDiarStage()
>>> stage.clusterer.assign(np.eye(2, dtype=np.float32), [1.0, 1.0])
['S0', 'S1']
>>> stage.reset()
>>> stage.discovered_speakers
0

La politique décide du regroupement : une politique par exemplaires donne un regroupement sans centroïde, à extraction identique, donc les deux sont directement comparables.

>>> from vocal_helper.exemplar import ExemplarPolicy
>>> other = IncrementalDiarStage(policy=ExemplarPolicy())
>>> type(other.clusterer).__name__
'ExemplarClusterer'
Return type:

None

property speaker_labels: list[str]

Les étiquettes globales, dans l’ordre de découverte.

windows(duration_s)[source]

Les bornes des fenêtres couvrant duration_s.

La dernière fenêtre est ramenée à la fin du flux plutôt que tronquée, pour que la fin de l’audio soit vue autant que le reste.

Examples

>>> stage = IncrementalDiarStage(window_s=5.0, step_s=2.5)
>>> stage.windows(10.0)
[(0.0, 5.0), (2.5, 7.5), (5.0, 10.0)]
>>> stage.windows(6.0)
[(0.0, 5.0), (1.0, 6.0)]
>>> stage.windows(3.0)
[(0.0, 3.0)]
Parameters:

duration_s (float)

Return type:

list[tuple[float, float]]

class vocal_helper.IncrementalPolicy(tau_active=0.5, rho_update_s=0.4, delta_new=0.6, delta_scale=None, delta_quantile=0.5, delta_position=None, max_speakers=None, ema_alpha=None)[source]

Bases: object

Les réglages du regroupement incrémental.

Les valeurs par défaut reprennent celles de la méthode dont ce module s’inspire, hors max_speakers, qui est un ajout.

Parameters:

  • tau_active (float)

  • rho_update_s (float)

  • delta_new (float)

  • delta_scale (float | None)

  • delta_quantile (float)

  • delta_position (float | None)

  • max_speakers (int | None)

  • ema_alpha (float | None)

tau_active

Seuil d’activité de la segmentation locale. Sous ce seuil, un locuteur local n’est pas considéré comme parlant dans la fenêtre.

Type:

float

rho_update_s

Part de parole propre minimale, en secondes, pour qu’un locuteur local soit autorisé à mettre son centroïde à jour.

Type:

float

delta_new

Distance cosinus au-delà de laquelle on crée un locuteur global plutôt que de forcer une jointure. Ignoré si delta_scale est posé.

Type:

float

delta_quantile

Quantile des distances inter-locuteurs servant d’échelle, passé à inter_speaker_scale(). Sans effet si delta_scale est absent.

Type:

float

delta_position

Si posé, delta_new est pris entre l’échelle intra-locuteur et l’échelle inter-locuteurs de cet enregistrement, par separating_threshold() — la seule calibration mesurée qui n’ait besoin d’aucune constante choisie à la main. L’emporte sur delta_scale.

Type:

float, optional

delta_scale

Si posé, delta_new est recalculé pour chaque enregistrement comme delta_scale fois l’échelle inter-locuteurs mesurée par inter_speaker_scale(). C’est la seule façon qu’on ait trouvée de tenir deux enregistrements différents avec un seul réglage : un seuil absolu ne transfère pas, une fraction de l’échelle locale peut.

Type:

float, optional

max_speakers

Plafond dur du nombre de locuteurs globaux. None (défaut) laisse le compte libre — c’est le mode « devine ». Un plafond est la garde la plus efficace contre la prolifération d’étiquettes : la mesure du 2026-11-01 montre qu’imposer un plafond au moteur différé vaut exactement le compte exact (DER 0,242 contre 0,299 en découverte libre).

Type:

int, optional

ema_alpha

Coefficient de la moyenne glissante des centroïdes. None prend la moyenne courante exacte, ce que fait diart.

Type:

float

delta_new: float = 0.6
delta_position: float | None = None
delta_quantile: float = 0.5
delta_scale: float | None = None
ema_alpha: float | None = None
max_speakers: int | None = None
rho_update_s: float = 0.4
tau_active: float = 0.5
class vocal_helper.LabelRevision(t0, t1, old_speaker, new_speaker)[source]

Bases: object

One already-published span whose speaker turned out to be somebody else.

Parameters:
new_speaker: str
old_speaker: str
t0: float
t1: float
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.LiveBudget(per_unit, effective_rtf, headroom, bottleneck)[source]

Bases: object

Whether a set of concurrently-running stages can hold a live stream.

Parameters:
bottleneck: str
describe()[source]

One line: the busiest unit, the headroom, and where the load sits.

Return type:

str

effective_rtf: float

The busiest unit. This, not the total, is what a live pipeline must beat.

headroom: float

how many times faster than real time the chain runs.

Type:

1 / effective_rtf

holds(*, margin=2.0)[source]

Whether the chain holds a live stream with room to spare.

Parameters:

margin (float, keyword-only, optional) – Required headroom. The default of 2 is not timidity: a machine under other load, a longer utterance, a model a size up, or a thermal throttle each eat a factor well under two, and crossing 1 is not a graceful degradation — it is a delay that grows for the rest of the session.

Return type:

bool

per_unit: dict[str, float]

stages sharing one unit serialise.

Type:

Summed real-time factor on each unit

class vocal_helper.LiveMetrics(started_at=<factory>, _emissions=<factory>, _latest=<factory>, _revisions=0, _queue_waits=<factory>, _violations=<factory>)[source]

Bases: object

Collects live-session timings. One instance per session.

Parameters:

  • started_at (float, optional) – Monotonic clock reading at the moment audio capture began. Defaults to now, which is right when the collector is created alongside the source and wrong if it is created later — pass it explicitly when replaying a file, where “now” and “the start of the audio” are different instants.

  • _emissions (list[_Emission])

  • _latest (dict[tuple[str, float], _Emission])

  • _revisions (int)

  • _queue_waits (dict[str, list[float]])

  • _violations (list[dict])

Examples

>>> m = LiveMetrics(started_at=0.0)
>>> m.record(speaker="S0", t0=0.0, t1=3.0, text="bonjour", now_s=3.85)
>>> round(m.report()["latency_p50_s"], 2)
0.85
latencies()[source]

Per-item seconds between the end of the speech and its delivery.

From the END of the utterance: that is when a listener starts waiting. Measuring from the start would charge a long sentence for being long, and would reward a pipeline that chops speech into fragments.

Return type:

list[float]

record(*, speaker, t0, t1, text, now_s=None)[source]

Record one item reaching the consumer.

t0/t1 are positions in the AUDIO timeline; now_s is the monotonic clock when the item was handed over (defaults to now). Re-recording the same (speaker, t0) with different text counts as a revision — which is what flicker is — while identical text is a duplicate delivery and is not.

Parameters:
Return type:

None

record_queue_wait(stage, seconds)[source]

Record how long one item waited in stage’s inbox before being picked up.

This is the half of latency a total cannot show. CADENCE.md: user-visible latency is the sum of queue waits plus per-stage compute, so a slow session with no queue wait is a slow model, and the same number with a large queue wait is a starved consumer. Opposite fixes.

Parameters:
Return type:

None

report()[source]

One flat dict: latency quantiles, revisions, per-queue waits, violations.

Return type:

dict

revision_rate()[source]

Share of deliveries that rewrote something already shown, in [0, 1].

Zero is not automatically good: a pipeline that only emits finalised text cannot flicker and pays for it in latency instead. The pair is the measurement, never either alone.

Return type:

float

started_at: float
summary()[source]

A one-line human reading of report(), for a log or a CLI tail.

Return type:

str

property violations: list[dict]

Items emitted before their audio existed. Non-empty means do not trust the rest.

class vocal_helper.NemotronOnnxDiar(graph=None, *, latency_s=1.04, mode=None, providers=None, threads=None, threshold=0.5)[source]

Bases: object

Diarisation en flux par le graphe Nemotron, sans torch.

Parameters:

  • graph (str or Path, optional) – Le fichier .onnx. None le cherche dans le bundle de poids, puis le télécharge une fois seul (resolve_graph()), sans jeton.

  • latency_s (float, optional) – Budget de latence, en secondes. C’est le seul réglage à connaître : le calendrier s’en déduit, en prenant le morceau le plus long que ce budget et le graphe permettent. Le défaut est DEFAULT_LATENCY_S, c’est-à-dire le calendrier publié à basse latence, parce qu’il est nettement meilleur dès qu’il y a de la réverbération — et une salle de réunion en a. None prend le plus long contexte viable, 6,4 s, qui coûte un tiers de calcul en moins et n’est à égalité qu’en salle anéchoïque.

  • mode (str, optional) – Un calendrier nommé de STREAMING_MODES, qui l’emporte sur latency_s quand il est donné. Utile pour reproduire exactement une configuration publiée. Ce n’est jamais celui que le graphe porte dans ses métadonnées : ce dernier est le calendrier hors ligne et il dépasse l’encodage positionnel tracé, voir MAX_SEQUENCE_FRAMES.

  • providers (list of str, optional) – Les moteurs d’exécution, par ordre de préférence. None prend CoreML s’il est disponible — c’est la voie du GPU et du moteur neuronal sur Apple Silicon, onnxruntime n’ayant pas de moteur Metal — puis le CPU, qui suffit de toute façon.

  • threads (int, optional) – Fils pour le moteur CPU. None — le défaut — laisse onnxruntime choisir, et c’est le meilleur choix : forcer huit fils sur une machine à huit cœurs performants mesure 0,953 de facteur temps réel contre 0,528 en laissant faire, soit presque deux fois pire. L’argument existe pour brider la consommation, pas pour accélérer.

  • threshold (float) – Seuil de parole appliqué aux probabilités à 10 ms.

diarize(pcm, sr, *, num_speakers=None, min_speakers=None, max_speakers=None)[source]

Diariser un tampon, en le faisant défiler comme un flux.

Les trois arguments de compte sont acceptés pour la parité d’interface avec les autres étages du paquet, et ignorés : ce modèle ordonne ses huit voies par ordre d’arrivée et n’a pas de compte à recevoir.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence, qui doit être 16 kHz.

  • num_speakers (int | None)

  • min_speakers (int | None)

  • max_speakers (int | None)

Returns:

[(t0, t1, locuteur), …] trié par début, la même forme que les autres étages de diarisation.

Return type:

list of tuple

load()[source]

Ouvrir la session tout de suite, comme les autres moteurs de diarisation.

vocal_helper.diar.OfflineDiarStage appelle load() sur chaque moteur avant le premier tampon. Sans cette méthode, backend="nemotron" (donc "auto" dès que le graphe est dans le bundle) levait AttributeError à la première diarisation hors ligne.

Return type:

None

locate()[source]

Le chemin du graphe : celui donné, sinon resolve_graph().

Raises:

FileNotFoundError – Si ni le bundle ni le fichier unique ne fournissent le moteur.

Return type:

Path

probabilities(pcm, sr)[source]

Les probabilités d’activité par locuteur, (T, 8) à 10 ms.

Ce que diarize() seuille. Leur maximum par trame est une détection de parole, mesurée contre Silero dans studies/vad_nemotron_vs_silero.py.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence, qui doit être 16 kHz.

Returns:

Une ligne par trame de 10 ms, une colonne par voie de locuteur.

Return type:

NDArray[np.float32]

property providers: list[str]

Les moteurs d’exécution que la session a réellement retenus.

async run(inbox, outbox)[source]

Consommer des segments voisés, les étiqueter, les republier.

Même contrat que les autres étages : un segment voisé entre, le même segment ressort avec son locuteur et son PCM, parce que la transcription en aval a besoin de l’audio. Le modèle n’est donc pas utilisé pour découper la parole — le détecteur d’activité l’a déjà fait — mais pour dire qui parle.

Le modèle est un modèle de flux : il veut de l’audio continu avec une horloge juste. Or la file ne porte que les segments voisés, donc les creux sont absents. L’étage reconstitue donc la ligne de temps en comblant les silences, sans quoi le cache de locuteurs verrait un montage et non une séance.

Un segment n’est publié que lorsque le modèle a dépassé sa fin : c’est la latence du calendrier choisi — 1,04 s par défaut — et elle est le prix de l’identité, pas un défaut d’implémentation.

Parameters:

  • inbox – Files de segments voisés puis diarisés.

  • outbox – Files de segments voisés puis diarisés.

Return type:

None

property schedule: NemotronSchedule

Le calendrier effectif, celui du graphe ou celui demandé.

class vocal_helper.NemotronSchedule(num_speakers=8, emb_dim=512, n_mels=128, subsampling=8, upsample_factor=8, spkcache_len=264, fifo_len=40, chunk_len=340, chunk_right_context=40, spkcache_update_period=300, sil_threshold=0.2, pred_score_threshold=0.25, strong_boost_rate=0.75, weak_boost_rate=1.5, min_pos_scores_rate=0.5, scores_boost_latest=0.05, spkcache_sil_frames_per_spk=1)[source]

Bases: object

Le calendrier de flux, en trames de 80 ms.

Les valeurs vivent dans les métadonnées du graphe, donc elles ne sont pas recopiées ici : un moteur mis à jour apporte les siennes, et un écart entre le code et le fichier est impossible par construction.

Parameters:

  • num_speakers (int)

  • emb_dim (int)

  • n_mels (int)

  • subsampling (int)

  • upsample_factor (int)

  • spkcache_len (int)

  • fifo_len (int)

  • chunk_len (int)

  • chunk_right_context (int)

  • spkcache_update_period (int)

  • sil_threshold (float)

  • pred_score_threshold (float)

  • strong_boost_rate (float)

  • weak_boost_rate (float)

  • min_pos_scores_rate (float)

  • scores_boost_latest (float)

  • spkcache_sil_frames_per_spk (int)

num_speakers

Nombre maximal de locuteurs que le modèle peut distinguer.

Type:

int

emb_dim

Dimension des empreintes du cache et de la file.

Type:

int

n_mels

Bandes attendues en entrée.

Type:

int

subsampling

Rapport entre les trames de 10 ms et celles de 80 ms.

Type:

int

upsample_factor

Rapport inverse, pour les probabilités à 10 ms.

Type:

int

spkcache_len, fifo_len, chunk_len, chunk_right_context

Longueurs du cache, de la file, du morceau et de son contexte droit.

Type:

int

spkcache_update_period

Fréquence de rafraîchissement du cache.

Type:

int

sil_threshold, pred_score_threshold

Seuils de silence et de score.

Type:

float

strong_boost_rate, weak_boost_rate, min_pos_scores_rate

Les taux qui équilibrent le cache entre locuteurs.

Type:

float

scores_boost_latest

Prime donnée aux trames les plus récentes.

Type:

float

spkcache_sil_frames_per_spk

Créneaux de silence réservés par locuteur dans le cache.

Type:

int

chunk_len: int = 340
chunk_right_context: int = 40
emb_dim: int = 512
fifo_len: int = 40
classmethod from_metadata(metadata)[source]

Lire le calendrier des métadonnées du graphe.

Les clés inconnues sont ignorées et les absentes gardent leur défaut, pour qu’un export enrichi n’empêche pas le chargement.

Examples

>>> NemotronSchedule.from_metadata(
...     {"vernacula.diar.fifo_len": "64", "vernacula.diar.inconnu": "x"}
... ).fifo_len
64
Parameters:

metadata (dict[str, str])

Return type:

NemotronSchedule

property latency_s: float

Audio à attendre avant de traiter un morceau, en secondes.

Examples

>>> round(NemotronSchedule(chunk_len=9, chunk_right_context=4).latency_s, 3)
1.04
min_pos_scores_rate: float = 0.5
n_mels: int = 128
num_speakers: int = 8
pred_score_threshold: float = 0.25
scores_boost_latest: float = 0.05
sil_threshold: float = 0.2
spkcache_len: int = 264
spkcache_sil_frames_per_spk: int = 1
spkcache_update_period: int = 300
strong_boost_rate: float = 0.75
subsampling: int = 8
upsample_factor: int = 8
weak_boost_rate: float = 1.5
class vocal_helper.OfflineDiarStage(*, backend='auto', ideal_duration_s=None, overlap_s=10.0, stitch_threshold=0.35, sherpa_cluster_threshold=0.5, sherpa_num_clusters=-1, pyannoteai_model=None, 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, *, num_speakers=None, min_speakers=None, max_speakers=None)[source]

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

Parameters:

  • num_speakers (int, optional) – Exact speaker count, when the set really is closed.

  • min_speakers (int, optional) – A FLOOR — the hint to reach for whenever a roster is known. Panel shows announce their panel, so the count is available before a sample is processed; but the roster is a floor and not a total, because the episode also carries reporters, archive clips and narrators nobody announced. Forcing an exact count has to put those extra voices somewhere, and where it puts them is inside a real person.

  • max_speakers (int, optional) – A ceiling. Leave unset unless a hard upper bound is genuinely known.

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

  • sr (int)

Return type:

list[tuple[float, float, str]]

Notes

These reach the backend’s OWN clustering rather than being applied afterwards, and that is the point. Repairing an over-estimated count after the fact is measurably not possible with the same embedding that produced it — the clusterer has already spent that information (see merge_speakers()). A count hint is only useful where it can still change the decision.

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 (str)

  • ideal_duration_s (float | None)

  • overlap_s (float)

  • stitch_threshold (float)

  • sherpa_cluster_threshold (float)

  • sherpa_num_clusters (int)

  • pyannoteai_model (str | None)

  • device (str | None)

class vocal_helper.OfflinePipeline(*, source, config=None, strict=True)[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={"engine": voh.resolve_engine()},
...         ),
...     )
...     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]

stage_failures: list[tuple[str, BaseException]]

(stage name, exception) for every stage that died this run — see StageCrashed. Filled even when strict=False.

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='auto', join_threshold=0.5, ema_alpha=0.1, min_segment_ms=500, device=None, max_speakers=None, refine_on_close=False, min_cluster_size=2, merge_threshold=None, merge_linkage='complete')[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. See JOIN_THRESHOLD.

  • 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)

  • merge_linkage (str)

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[int, ...], dtype[float32]]
sample_rate: int
t0: float
class vocal_helper.PeriodicOfflineDiarStage(*, online, offline, policy=None, on_revision=None)[source]

Bases: object

Live diarization at offline quality: a fast provisional label, then a correction.

Composes rather than replaces. The existing online stage gives a label within about half a second, which is what a live display needs; this stage then runs the offline diarizer periodically over a bounded window and reports the labels that were wrong. Neither half is sufficient alone: the online stage is three to four times worse, and the offline one cannot answer in half a second.

The audio path stays once-only: a revision carries labels, never audio, so nothing downstream re-transcribes anything. Transcription hears each segment exactly as many times as the online stage would have let it.

Parameters:

  • online (Any) – An OnlineDiarStage-shaped object, used for the provisional label. Injected so this class is testable without an embedder.

  • offline (Any) – An OfflineDiarStage-shaped object exposing diarize(pcm, sr, *, num_speakers=...). Injected for the same reason.

  • policy (RediarPolicy, optional)

  • on_revision (callable, optional) – (list[LabelRevision]) -> None, called after each refresh. None means revisions are only recorded on revisions, which is the right default for a batch caller and the wrong one for a live display — a display that ignores revisions shows labels it has been told are wrong.

Notes

Labels change retroactively, and that has to be visible. A line shown as S1 may become S2. Keeping a label known to be wrong would be worse, and silently rewriting history would be worse still: the display must re-colour, which is a deliberate behaviour, not an accident of ordering.

property discovered_speakers: int

How many speakers have been discovered, never assumed.

Determining the number of people talking is part of what diarization does, not an input to it. Nothing in the free path forces a count: the first refresh runs the engine in auto-detect, each later one runs with a floor while the set is still growing, and only once it has stopped growing is the count pinned — to the number discovered, not to a number supplied.

A caller that genuinely knows the count can still pass hints, and a caller that does not gets an answer rather than a guess.

Returns:

Zero before the first refresh.

Return type:

int

refresh(*, now_s, sample_rate, force=False)[source]

Re-diarize the trailing window and return the labels that were wrong.

Returns an empty list when no refresh is due, when there is nothing in the window, or when the offline diarizer declined — a refresh that fails must cost the provisional labels nothing.

Parameters:

  • now_s (float, keyword-only) – Audio time reached.

  • sample_rate (int, keyword-only)

  • force (bool, keyword-only, optional) – Refresh even if the period has not elapsed. Used when the stream closes: the period is a rate limit for a session that continues, and this one does not. Resetting the clock instead does not work — the period is measured forward from the last refresh, not backward from now.

Return type:

list of LabelRevision

refreshes

Refresh count, for a caller that wants to report the cadence.

revisions: list[LabelRevision]

Every revision this stage has produced, newest last.

async run(inbox, outbox)[source]

Consume voiced segments, publish provisionally, and revise periodically.

Same contract as the online stage, so this is a drop-in: a consumer that reads outbox needs no change. What is new is that labels already in outbox may be superseded, which is what on_revision reports.

Return type:

None

property speaker_labels: list[str]

The discovered speakers, in discovery order.

class vocal_helper.Pipeline(*, source, config=None, metrics=None, strict=True)[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(
...             asr={"model": "large-v3-turbo-q5_0"},
...             llm={"engine": voh.resolve_engine()},
...         ),
...     )
...     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.

dropped_frames: int

Frames discarded under source_policy="drop_oldest". Zero under back-pressure, by construction: that policy cannot lose a sample.

async run()[source]

Run the pipeline ; yield every Utterance and SummarySnapshot.

Return type:

AsyncIterator[Utterance | SummarySnapshot]

stage_failures: list[tuple[str, BaseException]]

(stage name, exception) for every stage that died this run. Populated even when strict=False, so a lenient caller can still ask what went wrong.

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>, diar_latency_s=1.04, growing='auto', rediar=None, asr=<factory>, llm=None, qsize_pcm=200, qsize_seg=32, source_policy='backpressure')[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
diar_latency_s: float | None = 1.04
eot: dict | None = None
growing: dict | str | None = 'auto'
llm: dict | None = None
qsize_pcm: int = 200
qsize_seg: int = 32
rediar: dict | None = None
source_policy: str = 'backpressure'
vad: dict
class vocal_helper.RediarPolicy(window_s=180.0, ref_budget_s=12.0, gap_s=0.8, period_s=60.0, min_speaker_s=5.0, backend='pyannote', max_speakers=12, stable_rounds_before_pinning=1)[source]

Bases: object

How often to re-diarize, over how much audio, and what that costs.

Parameters:

  • window_s (float) – Trailing audio re-diarized at each refresh.

  • ref_budget_s (float) – Audio harvested per already-known speaker for the primer.

  • gap_s (float) – Room tone inserted between primer pieces and before the window. Below the segmentation model’s min_duration_off the model does not cut there, the references bleed into each other, and the read-back becomes ambiguous.

  • period_s (float) – New audio accumulated between refreshes.

  • min_speaker_s (float) – In-window speech a new cluster must clear to be promoted to a speaker.

  • backend (str) – The offline diarizer. "pyannote" is the measured default at an error rate of 0.122; "sherpa" is the torch-free option at 0.174.

  • max_speakers (int)

  • stable_rounds_before_pinning (int)

backend: str = 'pyannote'
gap_s: float = 0.8
holds(*, asr_rtf, backend_rtf, speakers=6, vad_rtf=0.01, margin=2.0)[source]

Whether the whole system stays under a real-time factor of 1.

Assumes the worst case for placement — transcription and diarization on the same compute unit, so their costs add. That is the truth on a single-GPU machine, and assuming otherwise would be optimism dressed as arithmetic.

Parameters:

  • speakers (int, keyword-only, optional) – Expected speaker count. Six by default, which is a panel: budget for the room you might get, not the room you started with.

  • margin (float, keyword-only, optional) – Required headroom. Two by default: crossing 1 is not a graceful degradation but a delay that grows for the rest of the session.

  • asr_rtf (float)

  • backend_rtf (float)

  • vad_rtf (float)

Return type:

bool

Examples

>>> RediarPolicy().holds(asr_rtf=0.033, backend_rtf=0.067)
True
>>> RediarPolicy(period_s=10.0).holds(asr_rtf=0.033, backend_rtf=0.067)
False
max_speakers: int = 12

Hard ceiling on discovered speakers.

The lesson of DIARIZATION-TROUBLES.md: an unbounded greedy clusterer minted 203 labels for a four-speaker podcast. A ceiling is not a quality knob, it is the difference between a transcript and a list of 203 strangers. Twelve is past any panel and short of any runaway.

max_speakers_within(*, asr_rtf, backend_rtf, margin=2.0, ceiling=24)[source]

How many speakers this policy can carry before the budget breaks.

The number to know before a session, not during: when the answer is four and a fifth person joins, the pipeline starts falling behind, and the honest response is to widen the period rather than to discover it live.

Returns:

The largest speaker count that still holds. 0 when even one speaker does not fit, which means the policy is wrong for this machine.

Return type:

int

Parameters:

Examples

>>> RediarPolicy().max_speakers_within(asr_rtf=0.033, backend_rtf=0.067)
18

Without a GPU, transcription costs five times more and the room shrinks — still a real panel, but no longer unlimited:

>>> RediarPolicy().max_speakers_within(asr_rtf=0.164, backend_rtf=0.067)
9

Refresh too often and nothing fits, because the primer is re-diarized every time: a twenty-second period spends more on repeating itself than on new audio.

>>> RediarPolicy(period_s=20.0).max_speakers_within(asr_rtf=0.033, backend_rtf=0.067)
0
min_speaker_s: float = 5.0
period_s: float = 60.0
primer_s(*, speakers)[source]

Primer length for a given number of known speakers.

One reference plus one gap each, and one more gap before the window so the model also cuts between the primer and the audio being diarized.

Examples

>>> RediarPolicy().primer_s(speakers=0)
0.0
>>> round(RediarPolicy().primer_s(speakers=1), 1)
13.6
>>> round(RediarPolicy().primer_s(speakers=6), 1)
77.6
Parameters:

speakers (int)

Return type:

float

ref_budget_s: float = 12.0
stable_rounds_before_pinning: int = 1

Consecutive refreshes with no new speaker before the count gets pinned.

Discovery and stability pull opposite ways, and both are needed. While the count is still growing, the diarizer must be free to find a voice nobody has heard yet, so it runs with a floor. Once the set has stopped growing, the count is pinned, and that is what makes cross-window identity clean: the engine returns exactly as many clusters as there are references, every one of them maps back to a reference, and there is nothing left over to mint a new name from.

Measured: without this, the identity drifted worse than the online stage it was meant to fix — seven labels for four speakers, long-range agreement 0.297 against the online stage’s 0.969.

stage_rtf(backend_rtf, *, speakers)[source]

What this stage costs as a real-time factor, at a given speaker count.

Constant in the session length. Not constant in the speaker count, and that dependency is the one a live budget has to carry: speakers arrive during a session.

Examples

>>> round(RediarPolicy().stage_rtf(0.067, speakers=2), 3)
0.23
>>> round(RediarPolicy().stage_rtf(0.067, speakers=8), 3)
0.316
Parameters:
Return type:

float

static unbounded_rtf(*, buffer_s, period_s, backend_rtf)[source]

What re-diarizing the whole buffer would cost, for comparison.

A function rather than a comment, because the number is the argument: small enough to look fine in a demo, impossible in use. That is exactly how this design error survives a review.

Examples

>>> RediarPolicy.unbounded_rtf(buffer_s=180.0, period_s=60.0, backend_rtf=0.067)
0.201
>>> round(RediarPolicy.unbounded_rtf(buffer_s=3600.0, period_s=60.0, backend_rtf=0.067), 1)
4.0
Parameters:
Return type:

float

window_s: float = 180.0
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(*, engine, stt_model='large-v3-turbo-q5_0', max_merge_s=4.0, min_incomplete_ms=800)[source]

Bases: object

Producer/consumer EOT gating stage.

Parameters:

  • engine (dict) – Resolved engine descriptor from best_engine_ai_helper.ensure(<vocal_helper package dir>) — names the backend (Ollama / vLLM), the base URL, and the text model used as the EOT classifier. No default model is baked in.

  • 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).

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.Sink(queue, *, policy='lossless', name='')[source]

Bases: object

One destination of a fan-out: a queue plus what to do when it is full.

Parameters:

  • queue (asyncio.Queue) – Where items go. Its maxsize is what makes the policy matter at all; an unbounded queue never blocks and never drops, it just grows until the process dies.

  • policy (str, optional) – LOSSLESS (default) or LOSSY.

  • name (str, optional) – Used in on_wait reports, so a queue-wait figure can be attributed to a consumer rather than to “somewhere in the pipeline”.

dropped

Items dropped because this sink was full. Non-zero is information, not an error — it says a lossy consumer could not keep up, which is exactly what it opted into.

class vocal_helper.SpeakerReference(speaker, pcm, primer_t0=0.0, primer_t1=0.0)[source]

Bases: object

Harvested audio standing for one speaker, and where it sits in the primer.

primer_t0/primer_t1 are filled once the primer is assembled: they are what the read-back looks at, so a reference without them cannot anchor anything.

Parameters:
property duration_s: float

Length of the harvested audio, in the primer’s time base.

pcm: ndarray[tuple[int, ...], dtype[float32]]
primer_t0: float = 0.0
primer_t1: float = 0.0
speaker: str
class vocal_helper.StageLoad(name, unit, rtf)[source]

Bases: object

One pipeline stage, the compute unit it occupies, and its real-time factor.

Parameters:
name: str
rtf: float
unit: str

"cpu", "gpu" (Metal or MPS — the same silicon), or "ane".

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[int, ...], dtype[float32]]
sample_rate: int
t0: float
t1: float
class vocal_helper.WhisperStage(*, model='large-v3-turbo-q5_0', language='auto', threads=None, accelerator='auto', 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.

  • accelerator (str)

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

class vocal_helper.WindowFeatures(t0, local, embeddings, clean_durations, speakers)[source]

Bases: object

Une fenêtre glissante, sa segmentation locale et ses empreintes.

L’extraction (segmenter, découper la part propre, embarquer) coûte des secondes par minute d’audio ; le regroupement coûte des microsecondes. Les séparer rend un balayage de delta_new ou de rho_update_s cent fois moins cher, puisqu’il rejoue ces fenêtres en cache au lieu de refaire tourner les modèles.

Parameters:
t0

Début de la fenêtre dans le flux, en secondes.

Type:

float

local

La segmentation locale, en temps relatifs à la fenêtre.

Type:

list of tuple

embeddings

(k, d) — une empreinte par locuteur local retenu.

Type:

NDArray[np.float32]

clean_durations

k durées de part propre, dans l’ordre des empreintes.

Type:

list of float

speakers

k identifiants locaux, dans l’ordre des empreintes.

Type:

list of str

clean_durations: list[float]
embeddings: ndarray[tuple[int, ...], dtype[float32]]
local: list[tuple[float, float, str]]
speakers: list[str]
t0: float
vocal_helper.consolidate(utterances, *, min_speaker_s=120.0, merge_gap_s=3.0, max_turn_s=30.0, max_s=9.0, target_s=6.0, gap_s=0.3, max_words=45, pad_rate_s=0.34)[source]

Turn raw engine output into readable, clickable rows with stable speaker ids.

Applies the five rules of the module docstring, in the order that docstring explains, and relabels the surviving speakers S0, S1, … by descending talk time.

Parameters:

  • utterances (list of CloudUtterance) – Raw rows from any engine, with whatever speaker labels it used.

  • min_speaker_s (float, keyword-only, optional) – A discovered speaker must exceed this total duration to survive. Two minutes by default. Lower it when a real but quiet participant matters; the default is tuned for a panel where a four-word voice is a passer-by.

  • merge_gap_s (float, keyword-only, optional) – Same-speaker rows closer than the gap merge, up to the cap.

  • max_turn_s (float, keyword-only, optional) – Same-speaker rows closer than the gap merge, up to the cap.

  • max_s (float and int, keyword-only, optional) – When a row is cut, and where: past max_s it is split, preferring a silence of gap_s once the running row passes target_s, and always before max_words tokens.

  • target_s (float and int, keyword-only, optional) – When a row is cut, and where: past max_s it is split, preferring a silence of gap_s once the running row passes target_s, and always before max_words tokens.

  • gap_s (float and int, keyword-only, optional) – When a row is cut, and where: past max_s it is split, preferring a silence of gap_s once the running row passes target_s, and always before max_words tokens.

  • max_words (float and int, keyword-only, optional) – When a row is cut, and where: past max_s it is split, preferring a silence of gap_s once the running row passes target_s, and always before max_words tokens.

  • pad_rate_s (float, keyword-only, optional) – Seconds per word given to a zero-width row. The default is a calm French rate, roughly three words a second.

Returns:

(rows, speakers) where speakers maps each new id to the original engine label it came from — kept so a mis-assignment can be traced back to what the engine actually said, rather than disappearing into a renaming.

Return type:

tuple

Examples

>>> raw = [
...     {"t0": 0.0, "t1": 5.0, "speaker": "spk_1", "text": "Bonjour à tous",
...      "words": [(0.0, 5.0, "Bonjour à tous")], "language": "fr"},
...     {"t0": 5.0, "t1": 6.0, "speaker": "waiter", "text": "Un café",
...      "words": [(5.0, 6.0, "Un café")], "language": "fr"},
... ]
>>> rows, speakers = consolidate(raw, min_speaker_s=2.0)
>>> [(r["speaker"], r["text"]) for r in rows]
[('S0', 'Bonjour à tous')]
>>> speakers
{'S0': 'spk_1'}

With no floor, the brief voice survives and is simply ranked second:

>>> rows, speakers = consolidate(raw, min_speaker_s=0.0)
>>> [r["speaker"] for r in rows], speakers
(['S0', 'S1'], {'S0': 'spk_1', 'S1': 'waiter'})
vocal_helper.credential(name, default='')[source]

Resolve one credential, environment first.

Parameters:

  • name (str) – The credential name, conventionally upper-case ("GLADIA_API_KEY", "PYANNOTEAI_API_KEY", "ELEVENLABS_API_KEY").

  • default (str, optional) – Returned when nothing is configured anywhere. Defaults to "", which every cloud stage treats as “this mode is unavailable”.

Returns:

The value, or default. Never None, so a caller can test it with a plain truth check.

Return type:

str

Examples

>>> import os
>>> os.environ["PYANNOTEAI_API_KEY"] = "p-456"
>>> credential("PYANNOTEAI_API_KEY")
'p-456'
>>> del os.environ["PYANNOTEAI_API_KEY"]
>>> credential("PYANNOTEAI_API_KEY", default="absent")
'absent'
vocal_helper.credentials_status()[source]

Report which paid services are configured, without revealing any value.

Returns:

One entry per name in PAID_SERVICES, plus one per service that is True only when every key that service needs is present. A service with a single missing key reads as unavailable, which is the honest answer: the paid offline mode cannot run on half its providers.

Return type:

dict of bool

Examples

>>> status = credentials_status()
>>> set(status) >= {"cloud_offline", "GLADIA_API_KEY"}
True
>>> all(isinstance(v, bool) for v in status.values())
True
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

async vocal_helper.fanout(source, sinks, *, on_wait=None)[source]

Pull from source and forward every item to every sink until the stream ends.

The terminating None sentinel always reaches every sink, whatever its policy: losing an end-of-stream marker is not a late frame, it is a consumer that waits forever. On a lossy sink it makes room for itself the same way data does, rather than blocking on a queue nobody is draining — the first version awaited it unconditionally and deadlocked. On cancellation the sentinel is pushed too, so no downstream stage is left hanging on a session that was torn down.

Parameters:

  • source (asyncio.Queue) – The producer’s queue. A bare queue in sinks is wrapped as a lossless Sink, so the simple call stays simple.

  • on_wait (callable, optional) – (sink_name, seconds) for every put that had to wait. Feed it to LiveMetrics.record_queue_wait() to find out which consumer is pacing the rest.

  • sinks (Sequence[Sink | Queue[T | None]])

Return type:

None

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.inter_speaker_scale(windows, quantile=0.5)[source]

L’échelle des distances entre locuteurs différents, pour cet audio.

Un seuil cosinus absolu ne transfère pas d’un enregistrement à l’autre : la salle, le micro et la distance changent l’échelle même des distances. Deux réunions AMI de quatre personnes l’ont montré sans ambiguïté — le seuil qui donne le meilleur taux d’erreur sur l’une (0,8) fusionne deux vrais locuteurs sur l’autre, dérive 1,000.

La calibration est pourtant gratuite, et elle vient de la structure même du problème : deux locuteurs locaux présents dans la même fenêtre sont des personnes différentes par construction, puisque la segmentation locale les a distingués. La médiane de leurs distances donne donc l’échelle inter-locuteurs propre à cet enregistrement, sans vérité terrain, sans réglage et sans second passage.

Parameters:

  • windows (list of WindowFeatures) – Ce que IncrementalDiarStage.extract() a rendu.

  • quantile (float) – Quel quantile de ces distances retenir. La médiane (0,5) décrit l’enregistrement en moyenne, et c’est insuffisant : deux réunions AMI ont rendu 0,834 et 0,847 — indiscernables — alors que leur seuil optimal diffère d’un tiers. Ce qui les sépare n’est pas l’échelle moyenne mais la paire la plus confusable, que la médiane noie. Un quantile bas la fait remonter.

Returns:

La distance au quantile demandé entre locuteurs connus différents. nan si aucune fenêtre n’a vu deux locuteurs à la fois — un monologue, où la question ne se pose pas.

Return type:

float

Examples

Deux voix orthogonales, vues ensemble : l’échelle est leur distance, 1,0.

>>> import numpy as np
>>> pair = np.eye(2, dtype=np.float32)
>>> w = WindowFeatures(t0=0.0, local=[], embeddings=pair,
...                    clean_durations=[1.0, 1.0], speakers=["A", "B"])
>>> float(inter_speaker_scale([w]))
1.0

Un monologue ne renseigne sur rien.

>>> lone = WindowFeatures(t0=0.0, local=[], embeddings=pair[:1],
...                       clean_durations=[1.0], speakers=["A"])
>>> bool(np.isnan(inter_speaker_scale([lone])))
True
vocal_helper.intra_speaker_scale(windows, *, min_overlap_s=1.0, quantile=0.5)[source]

L’échelle des distances pour un même locuteur, pour cet audio.

Le pendant de inter_speaker_scale(), et il s’obtient tout aussi gratuitement. Les fenêtres voisines se recouvrent — la moitié d’une fenêtre par défaut — donc un locuteur qui parle dans le recouvrement est embarqué deux fois, une fois par chaque fenêtre. Les étiquettes locales ne sont pas partagées d’une fenêtre à l’autre, mais le temps l’est : deux locuteurs locaux dont les intervalles absolus se chevauchent franchement sont presque sûrement la même personne.

La distance entre ces deux empreintes mesure donc ce que l’enregistrement fait subir à une voix qui ne change pas : réverbération, mouvement, bruit. C’est l’information qui manquait pour poser un seuil, parce que ce qui décide d’une fusion n’est ni la distance inter-locuteurs ni la distance intra-locuteur seule, mais l’écart entre les deux.

Parameters:

  • windows (list of WindowFeatures) – Ce que IncrementalDiarStage.extract() a rendu, dans l’ordre.

  • min_overlap_s (float) – Recouvrement temporel minimal pour admettre que c’est la même voix. Au-dessous, l’appariement devient douteux et polluerait la mesure.

  • quantile (float) – Quel quantile de ces distances retenir. La médiane par défaut : ici on veut le comportement courant d’une voix, pas son pire moment.

Returns:

La distance au quantile demandé entre deux vues d’un même locuteur. nan si aucune paire n’a pu être appariée.

Return type:

float

Examples

Le même locuteur vu par deux fenêtres voisines, avec une empreinte qui a légèrement bougé : l’échelle est ce déplacement.

>>> import numpy as np
>>> a = np.array([[1.0, 0.0]], dtype=np.float32)
>>> moved = np.array([[0.8, 0.6]], dtype=np.float32)
>>> w0 = WindowFeatures(t0=0.0, local=[(1.0, 4.0, "A")], embeddings=a,
...                     clean_durations=[3.0], speakers=["A"])
>>> w1 = WindowFeatures(t0=2.5, local=[(0.0, 1.5, "B")], embeddings=moved,
...                     clean_durations=[1.5], speakers=["B"])
>>> round(intra_speaker_scale([w0, w1]), 3)
0.2

Sans recouvrement temporel, rien n’est apparié.

>>> far = WindowFeatures(t0=100.0, local=[(0.0, 1.5, "B")], embeddings=moved,
...                      clean_durations=[1.5], speakers=["B"])
>>> bool(np.isnan(intra_speaker_scale([w0, far])))
True
vocal_helper.join_words(words)[source]

Rebuild a row’s text from its word tokens, respecting each engine’s spacing.

Two conventions exist and they must not be mixed. Gladia and whisper encode a word boundary as a leading space on each token (" Après", " C", then "'est" with none), so those are concatenated verbatim: inserting our own spaces would shatter elisions into "C 'est -à -dire". ElevenLabs tokens carry no leading space, so those are space-joined, with apostrophes, hyphens and trailing punctuation glued to the previous token.

Which convention is in play is detected, not configured: a caller should not have to know, and a wrong setting is silently unreadable French.

Parameters:

words (list of tuple) – (t0, t1, text) triplets.

Returns:

The row’s text, stripped.

Return type:

str

Examples

>>> join_words([])
''
>>> join_words([(0.0, 1.0, " c"), (1.0, 2.0, "'est-à-dire")])
"c'est-à-dire"
>>> join_words([(0.0, 1.0, "oui"), (1.0, 2.0, "!")])
'oui!'
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[int, …], dtype[float64]], list[str], ndarray[tuple[int, …], dtype[float64]]]

vocal_helper.live_budget(stages)[source]

Combine concurrent stages into the one number a live pipeline must beat.

Not a sum. In a producer/consumer pipeline the stages run concurrently, so two stages on different compute units overlap and the chain costs the slower of the two, not both. Two stages on the same unit do not overlap: they queue for it, and there the costs do add.

So the rule is: add within a unit, take the maximum across units. Treating stage real-time factors as simply additive, as the obvious reading does, overstates the cost of a well-placed chain and — worse — hides the real problem, which is never the total but always one saturated unit.

This is the lever for a live session: the question is not “is each stage fast enough” but “are the expensive stages on different silicon”.

Parameters:

stages (list of StageLoad) – The stages that run concurrently on the same audio.

Return type:

LiveBudget

Examples

Both heavy stages on the GPU: they serialise, and the CPU sits idle.

>>> budget = live_budget([
...     StageLoad("vad", "cpu", 0.010),
...     StageLoad("asr", "gpu", 0.033),
...     StageLoad("diar", "gpu", 0.030),
... ])
>>> round(budget.effective_rtf, 3), budget.bottleneck
(0.063, 'gpu')
>>> budget.holds()
True

Moving diarization off the GPU does not help when the alternative is an expensive CPU backend: the bottleneck simply changes hands, for the worse.

>>> worse = live_budget([
...     StageLoad("vad", "cpu", 0.010),
...     StageLoad("asr", "gpu", 0.033),
...     StageLoad("diar", "cpu", 0.580),
... ])
>>> round(worse.effective_rtf, 3), worse.bottleneck
(0.59, 'cpu')

With no accelerator at all, everything piles onto one unit and the margin collapses to the point where a slower machine decouples:

>>> cramped = live_budget([
...     StageLoad("vad", "cpu", 0.010),
...     StageLoad("asr", "cpu", 0.164),
...     StageLoad("diar", "cpu", 0.580),
... ])
>>> round(cramped.effective_rtf, 3), cramped.holds()
(0.754, False)
vocal_helper.make_embedder(backend='nemo', *, model_path=None)[source]

Build a speaker embedder: the public way to get one.

Exists because every consumer in the suite was reaching into this module’s private names for it — notes-helper imported _TitaNetEmbedder, _SherpaEmbedder and _resolve_sherpa_models; sidekick imported _SherpaEmbedder. Three repositories depending on a leading underscore is not a dependency, it is a promise nobody made, and it is exactly what has to go before vocal-helper can be “the tool you use for any free local mode”.

Parameters:

  • backend (str, optional) – "nemo" (default) or "sherpa" — see EMBEDDER_BACKENDS. The default is TitaNet via NeMo, picked by studies/diar_embedding_backend.py: a 0.354 separability margin against 0.201 for pyannote/embedding, a 76 % uplift.

  • model_path (str, optional) – sherpa only: an explicit ONNX path. None resolves it the usual way (resolve_sherpa_models()), which never reaches the hub.

Returns:

An object exposing load() and embed(pcm, sr) -> np.ndarray.

Return type:

Any

Raises:

ValueError – On an unknown backend — named here rather than failing later inside a model load.

vocal_helper.map_by_overlap(left, right)[source]

La correspondance un-pour-un entre deux diarisations, par temps partagé.

Les noms de grappes sont arbitraires et interchangeables ; ce qui doit être conservé d’une passe à l’autre, c’est qui est qui. C’est un appariement bijectif, donc résolu par algorithme hongrois et non par un argmax par étiquette : un argmax peut envoyer deux étiquettes de gauche sur la même à droite, ce qui fusionnerait deux locuteurs sans le dire.

Parameters:

  • left (list of tuple) – Deux diarisations (t0, t1, étiquette) sur la même horloge.

  • right (list of tuple) – Deux diarisations (t0, t1, étiquette) sur la même horloge.

Returns:

{étiquette de gauche: étiquette de droite}, seulement pour les paires qui partagent effectivement du temps.

Return type:

dict of str

Examples

>>> map_by_overlap([(0.0, 2.0, "a"), (2.0, 4.0, "b")],
...                [(0.0, 2.0, "y"), (2.0, 4.0, "x")])
{'a': 'y', 'b': 'x'}

Rien de partagé, rien d’apparié — plutôt qu’un rapprochement inventé.

>>> map_by_overlap([(0.0, 1.0, "a")], [(9.0, 10.0, "z")])
{}
vocal_helper.merge_speakers(embeddings, labels, *, durations=None, threshold=0.6, linkage='complete', min_cluster_size=1, min_speakers=None, max_speakers=None, centre=True)[source]

Fold an over-estimated set of speaker labels back down using voiceprints.

Every diarizer we measure on real panel audio finds too many speakers rather than too few: on C dans l’air episodes announcing a six-person panel, pyannote community-1 returns 13–18 labels, NeMo 20–24, and the online path over 100. DIARIZATION-TROUBLES.md records the same signature — 203 labels for four people.

When this works, and when it cannot

It depends entirely on WHERE the extra labels came from, and the difference is not a detail — it is the difference between a near-perfect repair and no repair at all.

Random fragmentation is repairable. A streaming clusterer that mints a duplicate because its centroid drifted, or spawns a one-off label for a laugh, splits a speaker in a way unrelated to how that speaker actually sounds. Each fragment is then a random sample of the same voice, and the fragments look alike. Measured on AMI with such splits, the voiceprints separate same-speaker from different-speaker fragments at AUC 0.99 and the merge recovers the true count in 94 % of cases.

A clusterer’s considered opinion is not. When the extra labels come from asking a clusterer for too many speakers, the groups it returns are precisely the ones it judged FAR APART — using the same cosine similarity this function would use to merge them. The information needed for the repair has already been consumed. Measured on the same meetings, that AUC collapses from 0.99 to 0.60 median, 0.27 worst — at or below chance. No threshold and no linkage recovers it, because the signal is not there.

The practical rule: use this to repair fragmentation, never to undo a deliberate over-clustering. For the latter the only real fix is a different, stronger embedding, or a diarizer asked for the right count in the first place. The shipped default is deliberately conservative for exactly this reason — see MERGE_THRESHOLD.

That asymmetry is the reason this function exists. Over-clustering is repairable and under-clustering is not. When one person has been split across three labels, each piece is still pure, and a merge can reassemble them. When two people have been collapsed into one label, their speech is mixed inside a single cluster and nothing downstream can separate them again without redoing the segmentation. So the safe operating point for any diarizer in this suite is to over-split deliberately and repair here, rather than to gamble on hitting the count exactly.

param embeddings:

One unit-norm voiceprint per segment. None marks a segment that was never embedded; those keep their incoming label untouched rather than being guessed at.

type embeddings:

list of ndarray or None

param labels:

The provisional speaker id per segment, index-aligned with embeddings.

type labels:

list of str

param durations:

Seconds per segment. Used to weight the per-speaker voiceprint, because a 20 s turn is far stronger evidence of who is talking than a 1 s one, and an unweighted mean lets a crowd of short backchannels outvote the speech that actually identifies somebody. Defaults to equal weight.

type durations:

list of float, optional

param threshold:

Cosine distance at which merging stops. See MERGE_THRESHOLD.

type threshold:

float, optional

param linkage:

"complete" (default), "average" or "single" — see MERGE_LINKAGE.

type linkage:

str, optional

param min_cluster_size:

Merged clusters holding fewer segments than this are dissolved and their segments reassigned to the nearest surviving cluster. This is what absorbs the one-off micro-speakers that overlap, laughter and jingles produce.

type min_cluster_size:

int, optional

param min_speakers:

A FLOOR the merge may never go below. Panel shows announce their panel, so the roster is known before a single sample is processed — and it is a floor rather than a total, since an episode also carries reporters, archive clips and off-screen narrators nobody announces. Merging past the floor destroys the very separation the roster proves exists, which is why a floor is the right hint to pass a diarizer and an exact count is not.

type min_speakers:

int, optional

param max_speakers:

A ceiling, applied after the floor. Useful when a hard upper bound is genuinely known; leave unset otherwise, since guessing it wrong forces real speakers together.

type max_speakers:

int, optional

param centre:

Subtract the mean of the per-speaker voiceprints before comparing them, then re-normalise. Every segment of one recording shares a component that is not the speaker — the microphone, the room, the codec — and on single-device audio (a meeting through one array, a panel through one broadcast chain) that shared component can dominate the cosine geometry and swamp the smaller differences that actually distinguish people. Removing it widened the safe band from 94 % of AMI cases at a single threshold (0.30, uncentred) to 94 % across a three-notch span.

The mean is taken over SPEAKERS, not over segments, and that detail matters: a host holding 40 % of a panel’s airtime drags a segment-mean towards himself, so centering would subtract mostly him rather than the shared channel. One vote per provisional speaker removes that failure before it can happen.

Borrowed from notes_helper.diarize.cluster, which has done this for its local path all along; measuring it is what moved it from one consumer’s private trick to everyone’s default. Note that it changes what threshold means — see MERGE_THRESHOLD.

type centre:

bool, optional

returns:

One label per input segment, renamed compactly to "S0", "S1", … in order of first appearance. Segments whose embedding was None keep their original label.

rtype:

list of str

Notes

Pure and model-free: it takes vectors and returns labels, touching no audio and loading nothing. That is deliberate — it lets any consumer repair the output of any diarizer, including offline and paid cloud ones whose clustering we do not control.

Parameters:
Return type:

list[str]

vocal_helper.read_back_identities(segments, references, *, min_coverage=0.0)[source]

Recover global speaker identity by reading which local label covered each reference.

The engine diarized [primer | window] in one call, so its own clustering already placed each window turn with the matching reference. Identity therefore travels by time, not by embedding geometry: for each reference span, read which local label covered it.

This is the part ported from the native block-wise diarizer, and it is why that design needs no embeddings of its own.

The read-back is a bijection, so it is resolved by optimal assignment and not by a per-reference argmax. The distinction is not cosmetic. A greedy argmax lets one local label win two references — which happens whenever the engine merges two primer speakers into a single cluster — and the second write then overwrites the first, so one known speaker silently loses its label and is reported as absent. Optimal assignment cannot do that: the merged label takes one reference, the other reference stays unmatched, and the caller sees a speaker it could not place instead of a wrong answer.

Parameters:

  • segments (list of tuple) – (t0, t1, local_label) over the primer-plus-window time base.

  • references (list of SpeakerReference) – With their primer spans filled in.

  • min_coverage (float) – Fraction of a reference’s own duration that the matched label must cover for the pairing to count. 0.0 (default) accepts any shared time at all; raising it refuses a label that merely brushed the reference.

Returns:

{local_label: global_speaker}. A local label absent from the result covered no reference: it is a candidate new speaker, and the caller promotes it only if it clears min_speaker_s — guessing instead would merge a passer-by into a participant.

Return type:

dict of str

Examples

>>> import numpy as np
>>> silence = np.zeros(1, dtype=np.float32)
>>> refs = [SpeakerReference("S0", silence, 0.0, 12.0),
...         SpeakerReference("S1", silence, 12.8, 24.8)]
>>> segs = [(0.0, 12.0, "c1"), (12.8, 24.8, "c0"), (30.0, 40.0, "c2")]
>>> read_back_identities(segs, refs)
{'c0': 'S1', 'c1': 'S0'}

c2 is deliberately absent: it covered no reference, so it is a newcomer and not something to be guessed at.

A reference nobody covered yields nothing for that speaker, rather than a wrong answer:

>>> read_back_identities([(50.0, 60.0, "c9")], refs)
{}

One label straddling both references — the engine merged two primer speakers — is assigned to one of them, and the other is reported as unplaced. A per-reference argmax would instead have overwritten the first mapping and lost S0 without a word:

>>> read_back_identities([(0.0, 24.8, "c0")], refs)
{'c0': 'S0'}
vocal_helper.refresh_times(duration_s, *, first_s, ratio, max_period_s=None)[source]

Les instants de rafraîchissement, espacés géométriquement.

Un pas constant ferait croître le coût comme le carré de la durée, puisque chaque passe rediarise tout depuis le début. Un pas géométrique le garde proportionnel : la somme des préfixes reste un petit multiple de la durée totale. max_period_s borne l’attente au prix d’un coût plus élevé, pour une séance longue où un retard proportionnel serait inacceptable.

Examples

>>> refresh_times(100.0, first_s=10.0, ratio=2.0)
[10.0, 20.0, 40.0, 80.0, 100.0]

Plus court que la première passe : une seule, à la fin.

>>> refresh_times(8.0, first_s=10.0, ratio=2.0)
[8.0]

Avec un plafond, l’attente ne grandit plus indéfiniment.

>>> refresh_times(100.0, first_s=10.0, ratio=2.0, max_period_s=25.0)
[10.0, 20.0, 40.0, 65.0, 90.0, 100.0]
Parameters:
Return type:

list[float]

vocal_helper.resolve_engine(*, endpoint=None)[source]

Resolve vocal_helper’s LLM engine descriptor via best-engine-ai-helper.

Loads vocal_helper/llm.engine.yaml if present, else resolves it from the committed vocal_helper/llm.brief.yaml on first use (and writes it). The returned descriptor names the backend (Ollama / vLLM), base URL, and text model; it is what the LLM analyst and the semantic-EOT stage consume via best_engine_ai_helper.llm.chat. No default model constant is involved — the model is read only from this resolved engine. Entry points call this once and thread the result into both stages.

Parameters:

endpoint (str or None) – Override the serving base URL passed to best_engine_ai_helper.ensure (e.g. a remote Ollama / vLLM host). None uses the backend default.

Returns:

The resolved engine descriptor.

Return type:

dict

vocal_helper.resolve_sherpa_models()[source]

Public name for the sherpa ONNX resolution — (segmentation, embedding) paths.

Same function as the private one it wraps, published because consumers already depend on it. Never reaches HuggingFace: environment override, then the self-hosted bundle, then the shared ai-helpers mirror.

Return type:

tuple[str, str]

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, num_speakers=None, torch_free=False, pyannote_available=True, nemo_available=True, nemotron_available=False)[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.

  • num_speakers (int or None, optional) – Exact known speaker count (distinct from the max_speakers ceiling). When the router picks sherpa and this is set, the plan carries it as sherpa_num_clusters to pin sherpa’s clustering — the fix for its 2-party telephony over-segmentation (pdbms diar-study §12.1). Leaving it None keeps sherpa on auto-clustering, so meeting behaviour is unchanged. 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.

  • nemotron_available (bool, optional) – Whether the Nemotron 3 ONNX graph is in the diarization-engines bundle. When it is, every offline run routes to nemotron: it leads vocal_helper.diar.OFFLINE_PREFERENCE on error rate alone, is torch-free, and has neither Sortformer’s length hang nor its 4-speaker cap. Default False, so a caller that does not probe keeps the duration crossover; the CLI and the HTTP surface probe.

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'
>>> # torch-free with a known 2-party count pins sherpa's clustering
>>> select_diarization(live=False, torch_free=True, num_speakers=2).sherpa_num_clusters
2
>>> # without a known count, sherpa stays on auto-clustering (meeting default)
>>> select_diarization(live=False, torch_free=True).sherpa_num_clusters is None
True

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.separating_threshold(windows, *, position=0.5)[source]

Le seuil qui sépare « même voix » de « voix différentes », pour cet audio.

Les deux échelles se mesurent sans vérité terrain : l’une sur des locuteurs simultanés, donc différents par construction, l’autre sur un locuteur vu par deux fenêtres voisines, donc identique par construction. Le seuil se pose entre les deux, et il suit automatiquement l’enregistrement — c’est ce qu’aucune constante ne savait faire, puisque le seuil optimal variait d’un tiers entre deux réunions du même corpus.

Parameters:
Returns:

Le seuil. nan si l’une des deux échelles n’a pas pu être mesurée — à l’appelant de retomber sur sa valeur par défaut.

Return type:

float

Examples

>>> import numpy as np
>>> same = np.array([[1.0, 0.0]], dtype=np.float32)
>>> moved = np.array([[0.8, 0.6]], dtype=np.float32)
>>> pair = np.eye(2, dtype=np.float32)
>>> w0 = WindowFeatures(t0=0.0, local=[(1.0, 4.0, "A")], embeddings=same,
...                     clean_durations=[3.0], speakers=["A"])
>>> w1 = WindowFeatures(t0=2.5, local=[(0.0, 1.5, "B")], embeddings=moved,
...                     clean_durations=[1.5], speakers=["B"])
>>> w2 = WindowFeatures(t0=5.0, local=[(0.0, 1.0, "C"), (1.0, 2.0, "D")],
...                     embeddings=pair, clean_durations=[1.0, 1.0],
...                     speakers=["C", "D"])
>>> round(separating_threshold([w0, w1, w2]), 3)
0.6
vocal_helper.speaker_consistency(reference, hypothesis, *, min_share=0.0)[source]

Measure whether identities were held, ignoring boundary accuracy.

Parameters:

  • reference (list of tuple) – (t0, t1, true_speaker). From a synthetic recipe, where identity is exact by construction, or from a hand annotation.

  • hypothesis (list of tuple) – (t0, t1, label) as produced by a diarizer.

  • min_share (float, keyword-only, optional) – Ignore a label that holds less than this share of a speaker’s time, as boundary spill rather than a genuine identity split. 0.0 counts everything, which is the strict reading and the default: a frame of spill is still an error, it is simply a small one, and the share-weighted drift already makes it small.

Return type:

ConsistencyReport

Examples

Two voices merged into one label. Both numbers now report it, because the assignment can only give S0 to one of them: the other is unassigned and therefore wrong everywhere.

>>> reference = [(0.0, 10.0, "alice"), (10.0, 20.0, "bob")]
>>> merged = [(0.0, 20.0, "S0")]
>>> report = speaker_consistency(reference, merged)
>>> report.worst_drift, report.worst_collision
(1.0, 0.5)
>>> sorted(report.assigned_label)        # only one of the two got a label
['alice']

One voice split in half: the mirror image. Alice keeps one of the two labels, so half her speech is lost to the other — and the orphan label belongs to nobody, which scores a collision of 1.

>>> split = [(0.0, 10.0, "S0"), (10.0, 20.0, "S1")]
>>> report = speaker_consistency([(0.0, 20.0, "alice")], split)
>>> report.worst_drift, report.worst_collision
(0.5, 1.0)

That an unowned label costs a full collision is deliberate, and it is what makes this a direct detector of over-segmentation: the repository’s open high-severity bug is a clusterer that minted 203 labels for four speakers, and 199 of them would be unowned here.

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[int, ...], 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'