"""
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 :class:`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
:meth:`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
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
from collections.abc import AsyncIterator, Awaitable, Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import best_engine_ai_helper as beh
from vocal_helper.asr import DEFAULT_MODEL as DEFAULT_WHISPER_MODEL
from vocal_helper.asr import WhisperStage
from vocal_helper.diar import OfflineDiarStage, OnlineDiarStage
# Optional in-flight module — the semantic EOT stage lives in Warith's
# WIP branch and is not always on disk. Guarded so the base pipeline
# stays importable when the module is absent.
try:
from vocal_helper.eot import SemanticEOTStage # type: ignore[assignment]
except Exception: # pragma: no cover — optional
SemanticEOTStage = None # type: ignore[assignment, misc]
from vocal_helper.live_metrics import LiveMetrics
from vocal_helper.llm import GemmaAnalystStage
from vocal_helper.types import (
DiarizedSegment,
PcmFrame,
SummarySnapshot,
Utterance,
VoicedSegment,
)
from vocal_helper.vad import SileroVADStage
# Module-scoped logger named ``vocal_helper.pipeline`` — callers (and tests)
# filter on this name to observe swallowed stage / subscriber exceptions, and
# ``exc_info=True`` attaches the real traceback to the record. os-helper's
# ``osh.warning`` logs to the root logger without a name or ``exc_info``, so it
# cannot express this named-logger-with-traceback contract ; stdlib logging is
# the right tool here.
logger = logging.getLogger(__name__)
# The vocal_helper package directory — where the committed ``llm.brief.yaml``
# lives and where ``best-engine-ai-helper`` writes the gitignored
# ``llm.engine.yaml`` it resolves from it.
_PACKAGE_DIR = Path(__file__).resolve().parent
[docs]
def resolve_engine(*, endpoint: str | None = None) -> dict[str, Any]:
"""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
-------
dict
The resolved engine descriptor.
"""
return beh.ensure(_PACKAGE_DIR, endpoint=endpoint)
[docs]
class StageCrashed(RuntimeError):
"""A pipeline stage died, so the result is incomplete.
Raised at the end of :meth:`Pipeline.run` / :meth:`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 :func:`_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.
"""
def __init__(self, failures: list[tuple[str, BaseException]]) -> None:
names = ", ".join(name for name, _exc in failures)
super().__init__(
f"{len(failures)} pipeline stage(s) crashed: {names}. "
"The result is incomplete; see the chained exception."
)
self.failures = failures
async def _drain_task(
t: asyncio.Task, failures: list[tuple[str, BaseException]] | None = None
) -> None:
"""Await a cancelled / long-running task on pipeline shutdown.
``asyncio.CancelledError`` is expected — the task got the ``cancel()``
we sent in the ``finally`` block. Anything else is a stage
exception that has *not* been surfaced through the queue path (e.g.
the diar backend raised on the last forward and we never emitted
the ``None`` sentinel). Swallowing those silently is what let the
pyannote 3.x ``DiarizeOutput`` API break wait undetected for hours ;
always log them.
"""
try:
await t
except asyncio.CancelledError:
# Normal shutdown path.
return
except Exception as exc: # noqa: BLE001 — final barrier ; record, log, keep draining.
# Named logger + ``exc_info`` so the crash is observable with a full
# traceback ; this silent-swallow path once deadlocked the suite.
logger.warning(
"vocal_helper.pipeline: task %r crashed on shutdown",
t.get_name(),
exc_info=True,
)
# Logging alone is invisible to a PROGRAM. Recording it is what lets run() refuse to
# report success on a run that lost a stage. Draining continues either way, so one
# dead stage never leaves the others un-awaited.
if failures is not None:
failures.append((t.get_name(), exc))
async def _invoke_subscribers(
subscribers: list[Callable[..., Awaitable[None]]],
item: object,
stage: str,
) -> None:
"""Fan-out an item to user callbacks with per-callback isolation.
A crashing subscriber must not break the pipeline — but the caller
should still find out. We log a warning per failure with the stage
name, callback name and full traceback ; the pipeline keeps
forwarding to the remaining subscribers.
"""
for cb in subscribers:
try:
await cb(item)
except Exception: # noqa: BLE001 — user code ; log, then continue.
# Named logger + ``exc_info`` keeps the full traceback so a crashing
# subscriber is diagnosable without breaking the fan-out.
logger.warning(
"vocal_helper.pipeline: subscriber %r on stage %r raised",
getattr(cb, "__qualname__", repr(cb)),
stage,
exc_info=True,
)
# Bounded queue sizes — large enough to absorb a ~ 1 s burst, small
# enough that back-pressure pushes upstream before memory explodes.
_DEFAULT_QSIZE_PCM = 200 # 200 × 20 ms = 4 s of audio in flight
_DEFAULT_QSIZE_SEG = 32 # plenty for a slow ASR / LLM consumer
SourceFactory = Callable[[], AsyncIterator[PcmFrame]]
[docs]
@dataclass
class PipelineConfig:
"""Configuration object for :class:`Pipeline`.
Pass per-stage settings as dicts. The pipeline forwards them
verbatim to each stage's constructor.
"""
vad: dict = field(default_factory=dict)
# ``None`` disables the semantic end-of-turn gating stage —
# default behaviour matches the v0.1.0 release where Silero VAD's
# silence threshold is the only turn-end signal. Provide a dict to
# enable :class:`SemanticEOTStage` between VAD and diarization
# (cuts ~ 39 % of mid-sentence breaks per the LiveKit turn-detector
# white paper, at the cost of one extra LLM hop per voiced segment).
# Splatted into :class:`SemanticEOTStage` — so an enabled EOT block must
# carry a resolved ``engine`` (from :func:`resolve_engine`), e.g.
# ``eot={"engine": resolve_engine()}``.
eot: dict | None = None
diar: dict = field(default_factory=dict)
# OFFLINE-QUALITY DIARIZATION ON A LIVE STREAM, by re-diarizing the WHOLE
# buffer periodically. Enabled by default, because the alternative is the
# worst thing this package measures: the online stage alone gives a label in
# about half a second and an error rate of 0.369 with SIXTY-FIVE labels for
# four speakers on AMI dev-slice, a nearest-centroid clusterer having no way
# to hold identity over an hour.
#
# A dict turns on :class:`vocal_helper.growing.GrowingBufferDiarStage`, which
# keeps that fast label and then re-runs the offline diarizer over everything
# heard so far. Because the last pass sees the whole session, the final
# answer IS the full-file one: 0.176 median on the same eight meetings, the
# right speaker count, to the third decimal. The passes are spaced
# geometrically, so the cost stays proportional to the session rather than to
# its square — 2.6 to 3.3 times the audio, a real-time factor of 0.10 to 0.16.
#
# Splatted into :class:`vocal_helper.growing.GrowingBufferPolicy`, except for
# ``on_revision``, which is pulled out and passed to the stage. Supply it: a
# display that ignores revisions shows labels it has been told are wrong.
#
# growing={"ratio": 1.4, "on_revision": my_recolour_callback}
#
# ``None`` leaves the online stage alone, which is only reasonable when
# nothing may ever be revised. What this costs is not quality but FRESHNESS:
# speech after the last pass has no authoritative label yet, and at
# ``ratio=1.6`` that wait is 0.375 times the session — fine for twenty
# minutes, not for two hours, where ``max_period_s`` bounds it.
# ``"auto"`` (the default) means "use it, unless ``rediar`` was explicitly
# configured" — a default must not collide with an explicit choice. A dict is
# an explicit request, and then asking for ``rediar`` too is refused rather
# than silently resolved. ``None`` disables it.
# LE SEUL RÉGLAGE DE LATENCE À CONNAÎTRE pour la diarisation en direct, en
# secondes. Le défaut, 1,04 s, est le calendrier publié à basse latence et
# la valeur de :data:`vocal_helper.nemotron.DEFAULT_LATENCY_S` : mesuré en
# salle réverbérée à 0,6 s de RT60 (``studies/diar_far_field.py``), il fait
# 0,252 de taux d'erreur contre 0,350 au plus long contexte (6,4 s), et un
# accord sur les locuteurs au-delà de 300 s de 0,668 contre 0,372. Une salle
# de réunion réverbère ; c'est elle qui fixe le défaut.
#
# ``None`` prend le plus long contexte viable, 6,4 s. Il n'est à égalité
# qu'en salle anéchoïque (0,092 contre 0,090) et coûte alors un tiers de
# calcul en moins (facteur temps réel 0,166 contre 0,258) : le bon choix pour
# un enregistrement propre fait de près. ``0.32`` donne le plus court.
#
# Ce qui compte en direct est le RTF TOTAL du système, les étages
# travaillant à des cadences différentes ; le budget tient aux deux.
#
# Sans effet sur les étages qui ne sont pas Nemotron.
diar_latency_s: float | None = 1.04
growing: dict | str | None = "auto"
# The EARLIER approach to the same problem, kept because it bounds memory:
# re-diarize a trailing WINDOW with a synthetic primer carrying one reference
# per known speaker, so identity crosses windows. On five minutes of AMI
# IS1008a it halved the online error (0.666 → 0.348, offline 0.149,
# ``studies/asr_acceleration_rtf.md``). On the WHOLE meetings it did worse
# than the online stage alone: 0.679 against 0.597 on IS1008a, 0.817 against
# 0.616 on ES2011a (``studies/rediar_live_der.json``). Prefer ``growing``
# unless the session is long enough that holding its audio is the binding
# constraint, and measure on your own material before relying on it.
#
# Splatted into :class:`vocal_helper.rediar.RediarPolicy`, same ``on_revision``
# convention. Setting both is refused rather than silently resolved.
rediar: dict | None = None
asr: dict = field(default_factory=dict)
# ``None`` disables the LLM analyst — useful when you only need
# the transcript without summarisation. When set, the dict is splatted
# into :class:`GemmaAnalystStage` and must carry a resolved ``engine``
# (from :func:`resolve_engine`), e.g. ``llm={"engine": resolve_engine()}``.
llm: dict | None = None
qsize_pcm: int = _DEFAULT_QSIZE_PCM
qsize_seg: int = _DEFAULT_QSIZE_SEG
# WHAT HAPPENS WHEN THE PIPELINE FALLS BEHIND. The two answers are both right,
# for different sources, and picking the wrong one loses audio silently.
#
# ``"backpressure"`` (default) — the source waits for room. Correct for a FILE:
# reading more slowly costs nothing and not a sample is lost. It is also the
# only policy under which the transcript is guaranteed complete.
#
# ``"drop_oldest"`` — the source never waits; when the queue is full its oldest
# frame is discarded to make room, and the drop is counted. This is the policy
# for a LIVE source, and the reason is that back-pressure does not reach a
# microphone: blocking the loop that drains it makes a buffer grow inside
# capture-helper or the operating system, where this library cannot see it, and
# the audio is then lost with nobody logging it.
#
# Why the OLDEST and not the newest frame: dropping the newest bounds nothing.
# The pipeline keeps chewing through a stale backlog, stays maximally late
# forever, and every subsequent frame is refused too. Dropping the oldest keeps
# the queue full of the most RECENT audio, so latency stays bounded at the queue
# depth, which is what a live caption needs. A gap is a gap either way; only one
# of the two recovers.
source_policy: str = "backpressure"
[docs]
class Pipeline:
"""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 :class:`Utterance` and
:class:`SummarySnapshot` (the latter only when ``llm`` is
configured). For per-stage observation see
:meth:`Pipeline.subscribe_voiced` / ``subscribe_diarized``.
"""
def __init__(
self,
*,
source: SourceFactory,
config: PipelineConfig | None = None,
metrics: LiveMetrics | None = None,
strict: bool = True,
) -> None:
"""Wire up queues, stages and subscriber lists from ``config``.
Nothing runs here — construction only allocates the inter-stage queues
and instantiates each stage. The chain starts on :meth:`run`.
Parameters
----------
strict : bool, optional
``True`` (default) raises :class:`StageCrashed` when a stage died, because a run
that lost a stage did not succeed — and the old behaviour returned a normal,
empty transcript with the reason only in a log. ``False`` restores leniency as a
deliberate choice; ``stage_failures`` is filled either way.
metrics : LiveMetrics, optional
Collector for what the session costs the person watching it: latency from
end-of-speech to delivery, revision rate, queue waits. ``None`` (default)
records nothing. Pass one and read :meth:`LiveMetrics.report` afterwards —
offline quality has WER and DER, and this is the live equivalent.
"""
self.source_factory = source
self.config = config or PipelineConfig()
self.metrics = metrics
self.strict = strict
#: ``(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.
self.stage_failures: list[tuple[str, BaseException]] = []
#: Frames discarded under ``source_policy="drop_oldest"``. Zero under
#: back-pressure, by construction: that policy cannot lose a sample.
self.dropped_frames: int = 0
# One bounded queue per hand-off. PCM gets the deep buffer (bursty,
# 20 ms frames) ; the downstream segment queues stay shallow so a slow
# ASR / LLM consumer back-pressures upstream instead of hoarding memory.
self._q_pcm: asyncio.Queue[PcmFrame | None] = asyncio.Queue(maxsize=self.config.qsize_pcm)
self._q_voiced: asyncio.Queue[VoicedSegment | None] = asyncio.Queue(
maxsize=self.config.qsize_seg
)
self._q_diar: asyncio.Queue[DiarizedSegment | None] = asyncio.Queue(
maxsize=self.config.qsize_seg
)
self._q_utt: asyncio.Queue[Utterance | None] = asyncio.Queue(maxsize=self.config.qsize_seg)
self._q_summary: asyncio.Queue[SummarySnapshot | None] = asyncio.Queue(
maxsize=self.config.qsize_seg
)
self._vad = SileroVADStage(**self.config.vad)
self._eot: SemanticEOTStage | None = (
SemanticEOTStage(**self.config.eot) if self.config.eot is not None else None
)
self._online_diar = OnlineDiarStage(**self.config.diar)
self._diar: Any = self._online_diar
growing = self.config.growing
# Un backend nommé dans ``diar`` est un choix, et un choix l'emporte sur
# le défaut, comme ``rediar`` : ni Nemotron ne le remplace, ni la passe
# différée de la fenêtre croissante ne le contourne par son propre
# ``"auto"``. ``"nemotron"`` nommé demande justement Nemotron.
diar_backend = str(self.config.diar.get("backend", "auto"))
explicit_backend = diar_backend not in ("auto", "nemotron")
if growing == "auto":
# Un défaut s'efface devant un choix explicite.
growing = None if self.config.rediar is not None else {}
# MIEUX QUE LA FENÊTRE CROISSANTE QUAND IL EST LÀ. Mesuré sur un
# mixage à vérité exacte, en salle anéchoïque puis réverbérée à 0,3
# et 0,6 s de temps de décroissance : Nemotron donne 0,090 / 0,105 /
# 0,252 de taux d'erreur quand pyannote — donc la fenêtre croissante,
# dont la dernière passe EST la réponse de pyannote — donne 0,299 /
# 0,296 / 0,553. Deux à trois fois mieux dans toutes les salles, avec
# le bon nombre de locuteurs partout et une latence d'une seconde au
# lieu de plusieurs minutes.
#
# Le moteur n'est pas toujours là : il vit dans le bundle de poids, et
# une installation qui ne l'a pas téléchargé doit dégrader vers la
# fenêtre croissante plutôt que de refuser de démarrer.
if growing is not None and not explicit_backend:
try:
from vocal_helper.nemotron import NemotronOnnxDiar
candidate = NemotronOnnxDiar(latency_s=self.config.diar_latency_s)
candidate.locate()
except Exception: # noqa: BLE001 - absence du moteur, pas une faute
candidate = None
if candidate is not None:
self._diar = candidate
growing = None
elif growing is not None and self.config.rediar is not None:
raise ValueError(
"growing et rediar répondent au même besoin par deux chemins ; "
"en poser les deux explicitement ne dit pas lequel doit l'emporter. "
"Choisir growing pour la qualité du différé, rediar pour une "
"mémoire bornée."
)
if growing is not None:
from vocal_helper.diar import OfflineDiarStage
from vocal_helper.growing import GrowingBufferDiarStage, GrowingBufferPolicy
options = dict(growing)
on_revision = options.pop("on_revision", None)
# Seuls les moteurs différés locaux et gratuits passent : un moteur
# payant ne se répète pas à chaque passe sans qu'on le demande ici.
local = diar_backend in ("pyannote", "nemo", "sherpa")
backend = options.pop("backend", diar_backend if local else "auto")
self._diar = GrowingBufferDiarStage(
backend=backend,
policy=GrowingBufferPolicy(**options),
offline=OfflineDiarStage(backend=backend),
online=self._online_diar,
on_revision=on_revision,
)
elif self.config.rediar is not None:
from vocal_helper.diar import OfflineDiarStage
from vocal_helper.rediar import PeriodicOfflineDiarStage, RediarPolicy
options = dict(self.config.rediar)
on_revision = options.pop("on_revision", None)
policy = RediarPolicy(**options)
# The offline backend comes from the policy, not from ``config.diar``:
# the online stage's backend is an EMBEDDER choice and the offline one is
# a whole pipeline, and conflating them would silently ask for a
# diarizer that does not exist.
self._diar = PeriodicOfflineDiarStage(
online=self._online_diar,
offline=OfflineDiarStage(backend=policy.backend),
policy=policy,
on_revision=on_revision,
)
# Streaming enables warm-up by default (whisper's cold first inference,
# the Nemotron graph's load) so the first caption doesn't stall. Any
# explicit ``warmup`` in ``config.asr`` wins ; the offline
# pipeline leaves it off (batch throughput doesn't care).
self._asr = _asr_stage({"warmup": True, **self.config.asr})
self._llm: GemmaAnalystStage | None = (
GemmaAnalystStage(**self.config.llm) if self.config.llm is not None else None
)
self._voiced_subs: list[Callable[[VoicedSegment], Awaitable[None]]] = []
self._diar_subs: list[Callable[[DiarizedSegment], Awaitable[None]]] = []
self._utt_subs: list[Callable[[Utterance], Awaitable[None]]] = []
# ----- public API ----------------------------------------------------
[docs]
def subscribe_voiced(self, cb: Callable[[VoicedSegment], Awaitable[None]]) -> None:
"""Async callback for every :class:`VoicedSegment` after VAD."""
self._voiced_subs.append(cb)
[docs]
def subscribe_diarized(self, cb: Callable[[DiarizedSegment], Awaitable[None]]) -> None:
"""Async callback for every :class:`DiarizedSegment` after diarization."""
self._diar_subs.append(cb)
[docs]
def subscribe_utterances(self, cb: Callable[[Utterance], Awaitable[None]]) -> None:
"""Async callback for every :class:`Utterance` after ASR."""
self._utt_subs.append(cb)
[docs]
async def run(self) -> AsyncIterator[Utterance | SummarySnapshot]:
"""Run the pipeline ; yield every Utterance and SummarySnapshot."""
tasks: list[asyncio.Task] = []
# Inbound — read the source and push frames into q_pcm.
tasks.append(asyncio.create_task(self._source_loop(), name="voh.source"))
# VAD : q_pcm → q_voiced (with tee to subscribers).
tasks.append(
asyncio.create_task(
self._vad.run(self._q_pcm, self._q_voiced),
name="voh.vad",
)
)
# Tee voiced subscribers + forward to diar (or to EOT then diar).
q_voiced_for_diar: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._tee(self._q_voiced, q_voiced_for_diar, self._voiced_subs),
name="voh.tee.voiced",
)
)
if self._eot is not None:
# Insert semantic EOT gating between VAD and diarization —
# holds back segments that look mid-thought, merges them
# with their successor, and forwards the merged
# super-segment downstream. The diar queue therefore
# sees fewer, larger segments at higher semantic completeness.
q_voiced_post_eot: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._eot.run(q_voiced_for_diar, q_voiced_post_eot),
name="voh.eot",
)
)
tasks.append(
asyncio.create_task(
self._diar.run(q_voiced_post_eot, self._q_diar),
name="voh.diar",
)
)
else:
tasks.append(
asyncio.create_task(
self._diar.run(q_voiced_for_diar, self._q_diar),
name="voh.diar",
)
)
# Tee diar subscribers + forward to ASR.
q_diar_for_asr: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._tee(self._q_diar, q_diar_for_asr, self._diar_subs),
name="voh.tee.diar",
)
)
tasks.append(
asyncio.create_task(
self._asr.run(q_diar_for_asr, self._q_utt),
name="voh.asr",
)
)
# Tee utterance subscribers + forward to LLM (if configured) and
# to the yielded stream.
q_utt_for_llm: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
q_utt_for_output: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._tee_two(self._q_utt, q_utt_for_output, q_utt_for_llm, self._utt_subs),
name="voh.tee.utt",
)
)
if self._llm is not None:
tasks.append(
asyncio.create_task(
self._llm.run(q_utt_for_llm, self._q_summary),
name="voh.llm",
)
)
else:
# Drain to /dev/null so back-pressure doesn't stall the tee.
tasks.append(
asyncio.create_task(
self._drain(q_utt_for_llm),
name="voh.llm.disabled",
)
)
# Push an immediate None into q_summary so the merger knows
# there is no summary stream.
await self._q_summary.put(None)
try:
async for ev in self._merge(
q_utt_for_output, self._q_summary, watch=tasks if self.strict else None
):
yield ev
finally:
for t in tasks:
if not t.done():
t.cancel()
for t in tasks:
await _drain_task(t, self.stage_failures)
if self.strict and self.stage_failures:
raise StageCrashed(self.stage_failures) from self.stage_failures[0][1]
# ----- internal coroutines ------------------------------------------
async def _source_loop(self) -> None:
"""Pump frames from the source into ``q_pcm``, under the configured policy.
See :attr:`PipelineConfig.source_policy` for why there are two policies and
which source each belongs to. A drop is **counted and reported**, never
silent: audio that goes missing without a number attached is the one failure
nobody can debug afterwards.
"""
drop_oldest = getattr(self.config, "source_policy", "backpressure") == "drop_oldest"
try:
async for frame in self.source_factory():
if not drop_oldest:
await self._q_pcm.put(frame)
continue
try:
self._q_pcm.put_nowait(frame)
except asyncio.QueueFull:
# Make room by discarding the oldest frame, then take the new one.
# If the queue drained between the two calls there is nothing to
# discard, and the fallback await below cannot block for long.
with contextlib.suppress(asyncio.QueueEmpty):
self._q_pcm.get_nowait()
self.dropped_frames += 1
if self.dropped_frames == 1:
import os_helper as osh
osh.warning(
" pipeline: falling behind the live source; dropping the oldest "
"frames to bound latency. The count is reported when the run ends."
)
try:
self._q_pcm.put_nowait(frame)
except asyncio.QueueFull:
await self._q_pcm.put(frame)
finally:
if self.dropped_frames:
import os_helper as osh
osh.warning(
f" pipeline: {self.dropped_frames} frame(s) dropped to keep up with the "
f"live source. The transcript has gaps, and they are these."
)
# Always cap the stream with the None sentinel — even if the source
# raised — so the sentinel cascades and every stage shuts down.
#
# Under "drop_oldest" the queue may be FULL at this moment, and a plain
# await here would wait for a consumer that is itself waiting for the
# sentinel: the whole pipeline hangs on shutdown. The sentinel is the one
# message that must never be dropped, so it is given room by discarding a
# frame — losing audio at the very end of a stream that is already over
# costs nothing, and losing the sentinel costs everything.
if drop_oldest:
while True:
try:
self._q_pcm.put_nowait(None)
break
except asyncio.QueueFull:
with contextlib.suppress(asyncio.QueueEmpty):
self._q_pcm.get_nowait()
else:
await self._q_pcm.put(None)
async def _tee(
self,
inbox: asyncio.Queue,
outbox: asyncio.Queue,
subscribers: list[Callable[..., Awaitable[None]]],
*,
stage: str = "tee",
) -> None:
"""Forward every item inbox→outbox, fanning each out to ``subscribers`` too."""
while True:
item = await inbox.get()
# Forward first — the sentinel must propagate before we consider
# returning, so the downstream stage always sees end-of-stream.
await outbox.put(item)
if item is None:
return
# Subscribers are observers only ; a crash in one is isolated + logged.
await _invoke_subscribers(subscribers, item, stage)
async def _tee_two(
self,
inbox: asyncio.Queue,
out_a: asyncio.Queue,
out_b: asyncio.Queue,
subscribers: list[Callable[..., Awaitable[None]]],
*,
stage: str = "tee_two",
) -> None:
"""Fan every item to two outboxes (output + LLM) plus the subscriber list."""
while True:
item = await inbox.get()
# Both branches must receive the sentinel, else the merger / LLM
# stage would block forever waiting on a stream that never closes.
await out_a.put(item)
await out_b.put(item)
if item is None:
return
await _invoke_subscribers(subscribers, item, stage)
async def _drain(self, inbox: asyncio.Queue) -> None:
"""Swallow a queue to /dev/null so an unused branch can't back-pressure the tee."""
while True:
item = await inbox.get()
if item is None:
return
async def _merge(
self,
utt_q: asyncio.Queue,
summary_q: asyncio.Queue,
watch: list[asyncio.Task] | None = None,
) -> AsyncIterator[Utterance | SummarySnapshot]:
"""Interleave Utterance + SummarySnapshot streams until both end."""
# One-shot ``get`` wrapped as a task so ``asyncio.wait`` can race the
# two queues and yield whichever event lands first.
async def reader(q: asyncio.Queue) -> Utterance | SummarySnapshot | None:
"""Await and return the next item from ``q`` (one queue read).
Parameters
----------
q : asyncio.Queue
The utterance or summary queue to read a single item from.
Returns
-------
Utterance or SummarySnapshot or None
The next queued event, or ``None`` when the stream's
end-of-stream sentinel is reached.
"""
return await q.get()
utt_task = asyncio.create_task(reader(utt_q))
sum_task = asyncio.create_task(reader(summary_q))
utt_done = False
sum_done = False
# Loop until BOTH streams have delivered their None sentinel — a closed
# stream drops out of ``pending`` so we stop re-reading it.
watched = set(watch or ())
while not (utt_done and sum_done):
pending = {t for t, done in [(utt_task, utt_done), (sum_task, sum_done)] if not done}
if not pending:
break
done, _ = await asyncio.wait(
pending | {t for t in watched if not t.done()},
return_when=asyncio.FIRST_COMPLETED,
)
if _stage_crashed(done, watched):
# A dead stage never sends its None downstream, so waiting on
# the queues would hang forever. run() records the crash and
# raises StageCrashed once the stages are drained.
utt_task.cancel()
sum_task.cancel()
return
for d in done:
if d is not utt_task and d is not sum_task:
continue
item = d.result()
# For each finished read : a None retires that stream, otherwise
# we yield the event and immediately re-arm a fresh read of it.
if d is utt_task:
if item is None:
utt_done = True
else:
# Record what the consumer is about to see, at the moment it sees
# it. Opt-in (``metrics=`` on the Pipeline) and a dict insert when
# on, so an unmeasured session pays nothing. See
# :mod:`vocal_helper.live_metrics` for why latency and revision
# rate have to be read as a pair.
if self.metrics is not None:
self.metrics.record(
speaker=str(item.get("speaker", "")),
t0=float(item.get("t0", 0.0)),
t1=float(item.get("t1", 0.0)),
text=str(item.get("text", "")),
)
yield item
utt_task = asyncio.create_task(reader(utt_q))
elif d is sum_task:
if item is None:
sum_done = True
else:
yield item
sum_task = asyncio.create_task(reader(summary_q))
def _stage_crashed(done: set[asyncio.Task], watched: set[asyncio.Task]) -> bool:
"""Whether a watched stage task in ``done`` ended on an exception.
Retires every finished watched task from ``watched``, so a stage that ended
cleanly is not waited on again.
"""
crashed = False
for task in done & watched:
watched.discard(task)
if not task.cancelled() and task.exception() is not None:
crashed = True
return crashed
# ===========================================================================
# OFFLINE PIPELINE
# ===========================================================================
[docs]
def choose_asr_engine(config: dict, *, offline: bool) -> str:
"""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.
"""
engine = str(config.get("engine", "auto")).lower()
if engine != "auto":
return engine
language = config.get("language")
model = config.get("model")
if (
language not in (None, "", "auto")
or config.get("initial_prompt")
or model not in (None, "", DEFAULT_WHISPER_MODEL)
):
return "whisper"
return "nemotron"
def _asr_stage(config: dict, *, offline: bool = False) -> Any:
"""L'étage ASR que :func:`choose_asr_engine` désigne : whisper.cpp ou Nemotron 3.5.
``"nemotron"`` est l'encodeur de Nemotron 3.5 Transcribe en ONNX
(:class:`vocal_helper.nemotron_asr.NemotronAsrStage`), sans torch ; le
graphe vient de ``config["graph"]``, de ``$VH_NEMOTRON_ASR_GRAPH``, ou est
téléchargé une fois depuis ``deraison.ai/nemotron-onnx``.
"""
options = dict(config)
options.pop("engine", None)
engine = choose_asr_engine(config, offline=offline)
if engine == "whisper":
return WhisperStage(**options)
if engine == "nemotron":
from vocal_helper.nemotron_asr import NemotronAsrStage
return NemotronAsrStage(**options)
raise ValueError(f"moteur ASR inconnu {engine!r} : 'auto', 'whisper' ou 'nemotron'")
[docs]
@dataclass
class OfflinePipelineConfig:
"""Configuration object for :class:`OfflinePipeline`.
Same shape as :class:`PipelineConfig` minus the streaming-specific
``vad`` block — the offline diarizer ingests the full audio and
relies on the backend's own VAD / segmentation.
"""
diar: dict = field(default_factory=dict)
asr: dict = field(default_factory=dict)
llm: dict | None = None
qsize_pcm: int = _DEFAULT_QSIZE_PCM
qsize_seg: int = _DEFAULT_QSIZE_SEG
[docs]
class OfflinePipeline:
"""End-to-end batch chain : source → offline diar → ASR → LLM.
Trades the live VAD + per-segment online clustering for a single
call to :class:`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.
"""
def __init__(
self,
*,
source: SourceFactory,
config: OfflinePipelineConfig | None = None,
strict: bool = True,
) -> None:
"""Wire up the batch chain (no VAD): queues, offline diar, ASR, LLM."""
self.source_factory = source
self.config = config or OfflinePipelineConfig()
self.strict = strict
#: ``(stage name, exception)`` for every stage that died this run — see
#: :class:`StageCrashed`. Filled even when ``strict=False``.
self.stage_failures: list[tuple[str, BaseException]] = []
# Same bounded-queue rationale as the streaming pipeline — deep PCM
# buffer, shallow segment queues that back-pressure the diarizer.
self._q_pcm: asyncio.Queue[PcmFrame | None] = asyncio.Queue(maxsize=self.config.qsize_pcm)
self._q_diar: asyncio.Queue[DiarizedSegment | None] = asyncio.Queue(
maxsize=self.config.qsize_seg
)
self._q_utt: asyncio.Queue[Utterance | None] = asyncio.Queue(maxsize=self.config.qsize_seg)
self._q_summary: asyncio.Queue[SummarySnapshot | None] = asyncio.Queue(
maxsize=self.config.qsize_seg
)
self._diar = OfflineDiarStage(**self.config.diar)
# Offline defaults to the full-throttle batched ASR path (one
# whisper call per concatenated ``max_chunk_s`` window instead of
# one per diarized segment ; ~6× lower RTF at on-par WER per the
# 2026-07-09 sweep). Callers opt back into the per-segment path
# with ``OfflinePipelineConfig(asr={"batch": False})`` ; any
# explicit ``batch`` in ``config.asr`` wins.
asr_cfg = {"batch": True, **self.config.asr}
# Hors ligne, le moteur par défaut est Nemotron 3.5 quand la langue est à
# découvrir, whisper dès qu'on lui donne langue, prompt ou modèle
# (:func:`choose_asr_engine`) ; ``asr={"engine": "whisper"}`` le force.
self._asr = _asr_stage(asr_cfg, offline=True)
self._llm: GemmaAnalystStage | None = (
GemmaAnalystStage(**self.config.llm) if self.config.llm is not None else None
)
self._diar_subs: list[Callable[[DiarizedSegment], Awaitable[None]]] = []
self._utt_subs: list[Callable[[Utterance], Awaitable[None]]] = []
# ----- public API ----------------------------------------------------
[docs]
def subscribe_diarized(self, cb: Callable[[DiarizedSegment], Awaitable[None]]) -> None:
"""Register an async callback fired for every :class:`DiarizedSegment`."""
self._diar_subs.append(cb)
[docs]
def subscribe_utterances(self, cb: Callable[[Utterance], Awaitable[None]]) -> None:
"""Register an async callback fired for every :class:`Utterance`."""
self._utt_subs.append(cb)
[docs]
async def run(self) -> AsyncIterator[Utterance | SummarySnapshot]:
"""Run the batch chain ; yield every Utterance and SummarySnapshot."""
tasks: list[asyncio.Task] = []
# Inbound — feed the whole PCM buffer to the offline diarizer.
tasks.append(asyncio.create_task(self._source_loop(), name="voh.offline.source"))
tasks.append(
asyncio.create_task(
self._diar.run(self._q_pcm, self._q_diar),
name="voh.offline.diar",
)
)
q_diar_for_asr: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._tee(self._q_diar, q_diar_for_asr, self._diar_subs),
name="voh.offline.tee.diar",
)
)
tasks.append(
asyncio.create_task(
self._asr.run(q_diar_for_asr, self._q_utt),
name="voh.offline.asr",
)
)
q_utt_for_llm: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
q_utt_for_output: asyncio.Queue = asyncio.Queue(maxsize=self.config.qsize_seg)
tasks.append(
asyncio.create_task(
self._tee_two(self._q_utt, q_utt_for_output, q_utt_for_llm, self._utt_subs),
name="voh.offline.tee.utt",
)
)
if self._llm is not None:
tasks.append(
asyncio.create_task(
self._llm.run(q_utt_for_llm, self._q_summary),
name="voh.offline.llm",
)
)
else:
# No analyst configured — drain the LLM branch and hand the merger
# an immediate None so it never waits on a summary stream that
# will never produce one.
tasks.append(
asyncio.create_task(
self._drain(q_utt_for_llm),
name="voh.offline.llm.disabled",
)
)
await self._q_summary.put(None)
try:
async for ev in self._merge(
q_utt_for_output, self._q_summary, watch=tasks if self.strict else None
):
yield ev
finally:
# Cancel any still-running stage, then await each so exceptions
# (other than the expected CancelledError) get surfaced + logged.
for t in tasks:
if not t.done():
t.cancel()
for t in tasks:
await _drain_task(t, self.stage_failures)
if self.strict and self.stage_failures:
raise StageCrashed(self.stage_failures) from self.stage_failures[0][1]
# ----- internal coroutines (mirrors the streaming pipeline) ----------
async def _source_loop(self) -> None:
"""Pump frames from the source factory into ``q_pcm`` until it's exhausted."""
try:
async for frame in self.source_factory():
await self._q_pcm.put(frame)
finally:
# Sentinel-cap the stream even on source error, so diar shuts down.
await self._q_pcm.put(None)
async def _tee(
self,
inbox: asyncio.Queue,
outbox: asyncio.Queue,
subscribers: list[Callable[..., Awaitable[None]]],
*,
stage: str = "tee",
) -> None:
"""Forward every item inbox→outbox, fanning each out to ``subscribers`` too."""
while True:
item = await inbox.get()
# Propagate first so the sentinel always reaches the next stage.
await outbox.put(item)
if item is None:
return
await _invoke_subscribers(subscribers, item, stage)
async def _tee_two(
self,
inbox: asyncio.Queue,
out_a: asyncio.Queue,
out_b: asyncio.Queue,
subscribers: list[Callable[..., Awaitable[None]]],
*,
stage: str = "tee_two",
) -> None:
"""Fan every item to two outboxes (output + LLM) plus the subscriber list."""
while True:
item = await inbox.get()
# Both branches must see the sentinel or the merger blocks forever.
await out_a.put(item)
await out_b.put(item)
if item is None:
return
await _invoke_subscribers(subscribers, item, stage)
async def _drain(self, inbox: asyncio.Queue) -> None:
"""Swallow a queue to /dev/null so an unused branch can't back-pressure the tee."""
while True:
item = await inbox.get()
if item is None:
return
async def _merge(
self,
utt_q: asyncio.Queue,
summary_q: asyncio.Queue,
watch: list[asyncio.Task] | None = None,
) -> AsyncIterator[Utterance | SummarySnapshot]:
"""Interleave Utterance + SummarySnapshot streams until both end."""
# One-shot queue read as a task so ``asyncio.wait`` can race both queues.
async def reader(q: asyncio.Queue) -> Utterance | SummarySnapshot | None:
"""Await and return the next item from ``q`` (one queue read).
Parameters
----------
q : asyncio.Queue
The utterance or summary queue to read a single item from.
Returns
-------
Utterance or SummarySnapshot or None
The next queued event, or ``None`` when the stream's
end-of-stream sentinel is reached.
"""
return await q.get()
utt_task = asyncio.create_task(reader(utt_q))
sum_task = asyncio.create_task(reader(summary_q))
utt_done = False
sum_done = False
# Run until both streams have signalled end via their None sentinel.
watched = set(watch or ())
while not (utt_done and sum_done):
pending = {t for t, done in [(utt_task, utt_done), (sum_task, sum_done)] if not done}
if not pending:
break
done, _ = await asyncio.wait(
pending | {t for t in watched if not t.done()},
return_when=asyncio.FIRST_COMPLETED,
)
if _stage_crashed(done, watched):
# A dead stage never sends its None downstream, so waiting on
# the queues would hang forever. run() records the crash and
# raises StageCrashed once the stages are drained.
utt_task.cancel()
sum_task.cancel()
return
for d in done:
if d is not utt_task and d is not sum_task:
continue
item = d.result()
# None retires the stream ; a real event is yielded then the
# read is re-armed to keep the interleave going.
if d is utt_task:
if item is None:
utt_done = True
else:
yield item
utt_task = asyncio.create_task(reader(utt_q))
elif d is sum_task:
if item is None:
sum_done = True
else:
yield item
sum_task = asyncio.create_task(reader(summary_q))