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)[source]

Bases: object

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

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

Usage

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

The cadence trade

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

async run()[source]

Run the batch chain ; yield every Utterance and SummarySnapshot.

Return type:

AsyncIterator[Utterance | SummarySnapshot]

subscribe_diarized(cb)[source]

Register an async callback fired for every DiarizedSegment.

Parameters:

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

Return type:

None

subscribe_utterances(cb)[source]

Register an async callback fired for every Utterance.

Parameters:

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

Return type:

None

Parameters:
class vocal_helper.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:
asr: dict
diar: dict
llm: dict | None = None
qsize_pcm: int = 200
qsize_seg: int = 32
class vocal_helper.pipeline.Pipeline(*, source, config=None)[source]

Bases: object

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

Usage

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

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

async run()[source]

Run the pipeline ; yield every Utterance and SummarySnapshot.

Return type:

AsyncIterator[Utterance | SummarySnapshot]

subscribe_diarized(cb)[source]

Async callback for every DiarizedSegment after diarization.

Parameters:

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

Return type:

None

subscribe_utterances(cb)[source]

Async callback for every Utterance after ASR.

Parameters:

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

Return type:

None

subscribe_voiced(cb)[source]

Async callback for every VoicedSegment after VAD.

Parameters:

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

Return type:

None

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

Bases: object

Configuration object for Pipeline.

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

Parameters:
asr: dict
diar: dict
eot: dict | None = None
llm: dict | None = None
qsize_pcm: int = 200
qsize_seg: int = 32
vad: dict