vocal_helper.pipeline module

vocal_helper.pipeline

The top-level producer/consumer orchestrator.

Stages, in order :

[Source] → [VAD] → [Diar] → [ASR] → [LLM analyst]
q0 q1 q2 q3 q4

Every arrow is an asyncio.Queue. Each stage is a long-running coroutine that await``s on its inbox, processes the event, and pushes the result onto its outbox. Closing the upstream queue with a ``None sentinel cascades cleanly through the rest of the chain.

The pipeline is configured at construction time but only starts when Pipeline.start() is called. The caller can attach subscribers to any intermediate queue (q1 for VAD events, q3 for ASR output, etc.) for live UI / WebSocket fan-out.

Author

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

class vocal_helper.pipeline.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:

source (SourceFactory)
config (OfflinePipelineConfig | None)
strict (bool)

class vocal_helper.pipeline.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:

diar (dict)
asr (dict)
llm (dict | None)
qsize_pcm (int)
qsize_seg (int)

asr: dict

diar: dict

llm: dict | None = None

qsize_pcm: int = 200

qsize_seg: int = 32

class vocal_helper.pipeline.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:

source (SourceFactory)
config (PipelineConfig | None)
metrics (LiveMetrics | None)
strict (bool)

class vocal_helper.pipeline.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:

vad (dict)
eot (dict | None)
diar (dict)
diar_latency_s (float | None)
growing (dict | str | None)
rediar (dict | None)
asr (dict)
llm (dict | None)
qsize_pcm (int)
qsize_seg (int)
source_policy (str)

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

exception vocal_helper.pipeline.StageCrashed(failures)[source]

Bases: RuntimeError

A pipeline stage died, so the result is incomplete.

Raised at the end of Pipeline.run() / OfflinePipeline.run() when any stage crashed, because the alternative is worse than an exception: before this existed, a source that failed produced a normal, empty transcript. The crash was logged and the run returned successfully with nothing in it, so a caller saw silence and no reason for it — the same shape of failure the docstring of _drain_task() says once hid a broken pyannote API “undetected for hours”.

Pass strict=False to the pipeline to keep the lenient behaviour, which is a choice somebody then made on purpose rather than inherited by accident.

Parameters:: failures (list[tuple[str, BaseException]])
Return type:: None

vocal_helper.pipeline.choose_asr_engine(config, *, offline)[source]

Le moteur ASR à faire tourner : config["engine"], ou le meilleur mesuré si "auto".

"auto" (le défaut) suit les études de 2026-11, taux d’erreur sur les mots :

langue à découvrir → "nemotron", hors ligne comme en direct. Hors ligne (studies/asr_nemotron_vs_whisper_offline.py) : AMI 0,23 contre 0,38, voix mêlées 0,38 contre 0,54, Common Voice français 0,27 contre 0,69, trois fois plus vite. En direct, segment par segment après Silero (studies/asr_live_whisper_vs_nemotron.py) : AMI 0,24 contre 0,25, voix mêlées 0,38 contre 0,53, Common Voice français 0,27 contre 0,79, télévision 0,48 contre 0,83 — whisper devine mal la langue d’un segment court et le traduit en anglais ;
langue donnée, prompt ou modèle whisper nommé → "whisper" : à qui l’on donne la langue, whisper rattrape Nemotron sur Common Voice français (0,28) et le dépasse sur la télévision française, et il reste devant en anglais lu (0,012 contre 0,067).

offline est gardé pour les appelants ; la règle est la même des deux côtés.

Parameters:

config (dict)
offline (bool)

Return type:

str

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