vocal_helper.parallel_pipelines module

vocal_helper.parallel_pipelines

Fan-out primitive for research A/B comparisons — imports the Pipecat ParallelPipelines pattern (https://docs.pipecat.ai/guides/learn/pipeline) into pdbms.

Motivation

Every diar / STT / EOT study we run has the same shape :

load audio → run VAD → for each backend branch → collect results

The naive script re-loads the audio and re-runs the VAD once per branch. That’s wasteful (VAD is fast but not free ; a 30-min meeting × 5 backends × 3 re-runs adds up on the multi-hour cascades). Pipecat’s ParallelPipelines solves this by carrying one upstream produce through multiple downstream consumers, with an ordering contract on the outputs.

This module gives the same primitive to pdbms study scripts. It’s intentionally small — the goal is a clean pattern, not a full framework.

Two shapes are supported :

1. run_parallel_sync() — synchronous. The caller supplies a list of (name, callable) branches. Each callable takes the shared upstream input and returns a per-branch result. They run in parallel via concurrent.futures. Best for CPU-bound backends that don’t share GPU state.

2. run_parallel_async() — asyncio-based. The caller supplies coroutines. Each coroutine takes the shared input and returns a result. They run concurrently via asyncio.gather(). Best for I/O-bound backends (Ollama HTTP, network models).

Output contract : both variants return a dict {branch_name: (result, wall_time_s)} preserving the order of the input branch list, so downstream summary tables compare like for like.

Example

from pdbms.utils.parallel_pipelines import run_parallel_sync

def run_pyannote(audio): ...
def run_titanet(audio): ...

results = run_parallel_sync(
    upstream_input=audio_pcm,
    branches=[
        ("pyannote", run_pyannote),
        ("titanet",  run_titanet),
    ],
)

for name, (segs, wall) in results.items():
    print(name, len(segs), wall)

Author

Warith Harchaoui — https://www.linkedin.com/in/warith-harchaoui/

async vocal_helper.parallel_pipelines.run_parallel_async(upstream_input, branches)

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:

• upstream_input (T)

• branches (list[tuple[str, Callable[[T], Awaitable[R]]]])

Return type:

dict[str, tuple[R, float]]

vocal_helper.parallel_pipelines.run_parallel_sync(upstream_input, branches, *, max_workers=None)

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