Live diarized transcription: per-speaker transcription streams (Approach 2)

Why use one transcription stream per speaker?

Live captions, meeting assistants, and call analytics need to answer one question: who said what? In this tutorial, we will combine pyannoteAI live diarization with one transcription stream per speaker. Each stream is assigned a pyannoteAI speaker label, so all text returned by that stream can be attributed directly to that speaker. The client combines timestamped results instead of matching transcript events to speaker turns. Sending each speaker’s speech to a separate transcription session may improve accuracy compared with sending all speakers to one session. This approach requires a known upper bound on the number of speakers. The client opens one transcription stream for each possible speaker, so transcription cost scales with that upper bound. It also buffers audio while waiting for speaker events, which adds transcription latency. Approach 1: single STT stream uses one transcription session for all speakers, but requires reconciliation logic. Evaluate both approaches with your audio, cost, and latency requirements.

Read Combining real-time diarization and transcription for the concepts behind both designs. This tutorial implements Approach 2.

Build a live speaker-attributed transcript

We will build a command-line program that listens to your microphone and prints a live transcript:

[SPEAKER_00] Hey, are you seeing the numbers from last week?
[SPEAKER_01] Yeah, I just pulled them up. Revenue is up about twelve percent.
[SPEAKER_00] Nice, that is ahead of the forecast.

This example sets max_speakers to 2. It supports up to two speakers and opens two AssemblyAI sessions before capture starts. Both sessions incur transcription costs for the full call duration, even if the conversation has only one speaker.

Prerequisites

A pyannoteAI API key from the dashboard
An AssemblyAI API key from the AssemblyAI dashboard
Python 3.10+
uv
Microphone access

The full script declares its dependencies inline. uv run installs them in an isolated environment automatically. Create a .env file:

PYANNOTEAI_API_KEY=sk_xxx
ASSEMBLYAI_API_KEY=xxxxxxxx

Architecture overview

The script captures each microphone frame once. It sends the original float32 frame to pyannoteAI and stores the frame in a short buffer. The router uses speaker events to send PCM16 speech to each active speaker’s AssemblyAI stream and silence to the other streams. Speaker events arrive after the corresponding audio. The delayed audio buffer holds audio for one second to give those events time to arrive before routing. If the router processes speech before its speaker-start event arrives, it can send silence instead of speech. This fixed delay does not guarantee that all events arrive in time; adjust it for your network and processing latency. The diagram shows selected streams for a general N; this example opens two. All N AssemblyAI streams receive the same amount of audio. Silence keeps each stream’s clock moving when its speaker is inactive. Word timestamps from all streams therefore use one common timeline.

Routing is not source separation. If multiple speakers talk into one microphone at the same time, each active stream receives the same mixed frame. Use separate input channels or a source-separation model when you must isolate overlapping voices.

1. Choose the speaker upper bound and open a pyannoteAI stream

Set max_speakers to the maximum number of speakers that can join the conversation. This value is N, the speaker-count upper bound. The script must know it before opening the transcription sessions. The router assigns each new pyannoteAI speaker label to one of the N slots. pyannoteAI live diarization tracks up to eight speakers, so use a value from 1 to 8:

max_speakers = 2  # N, the known speaker-count upper bound


def create_pyannote_stream() -> tuple[str, str]:
    response = requests.post(
        "https://api.pyannote.ai/v1/live",
        headers={"Authorization": f"Bearer {PYANNOTEAI_API_KEY}"},
    )
    response.raise_for_status()
    data = response.json()
    return data["id"], data["url"]

The response provides a stream ID and a single-use WebSocket URL. pyannoteAI accepts 16 kHz mono float32 PCM in 100 ms frames and returns stable speaker labels with start and end timestamps. The script holds audio for one second before routing it:

ROUTING_DELAY = 1.0

Speaker events describe audio that the microphone already captured. This delay gives those events time to arrive before the matching buffered frame is sent to AssemblyAI. Increase it if network or processing latency causes clipped turn starts.

2. Open N AssemblyAI streams

Create one AssemblyAI WebSocket for each of the N speaker slots before audio capture starts:

params = urlencode({
    "sample_rate": SAMPLE_RATE,
    "encoding": "pcm_s16le",
    "speech_model": "universal-3-5-pro",
    "speaker_labels": "false",
    "min_turn_silence": 100,
    "max_turn_silence": 1000,
    "continuous_partials": "true",
})
assembly_url = f"wss://streaming.assemblyai.com/v3/ws?{params}"

aai_streams = [
    await stack.enter_async_context(
        websockets.connect(
            assembly_url,
            additional_headers={"Authorization": ASSEMBLYAI_API_KEY},
        )
    )
    for _ in range(max_speakers)
]

universal-3-5-pro uses turn detection. min_turn_silence starts an end-of-turn check after a short pause, and max_turn_silence forces the turn to end after one second of silence. AssemblyAI speaker labels stay disabled because pyannoteAI supplies the speaker identity.

AssemblyAI bills each open streaming session by its duration, including periods that contain silence. Each session is billed independently, so N speaker slots produce N times the billed session duration of one stream. Always terminate every session when capture stops.

3. Route audio to each speaker

Store the pyannoteAI events with their timestamps:

def add_event(self, timestamp: float, speaker: str, started: bool) -> None:
    if started and speaker not in self.speaker_indexes:
        if len(self.speaker_indexes) < max_speakers:
            self.speaker_indexes[speaker] = len(self.speaker_indexes)
    self.events.append((timestamp, speaker, started))
    self.events.sort(key=lambda event: event[0])

For each buffered frame, allocate one row per speaker in a two-dimensional array. Copy samples to every row whose speaker is active. Leave the other rows at zero:

outputs = np.zeros(
    (max_speakers, len(frame.samples)),
    dtype=frame.samples.dtype,
)

for speaker in self.active:
    index = self.speaker_indexes.get(speaker)
    if index is not None:
        outputs[index, start_sample:end_sample] = frame.samples[start_sample:end_sample]

Convert each routed float32 array to little-endian PCM16 before sending it to AssemblyAI:

for stream, output in zip(aai_streams, outputs):
    pcm16 = (np.clip(output, -1, 1) * 32767).astype("<i2").tobytes()
    await stream.send(pcm16)

Each stream receives every frame. For example, while only SPEAKER_00 talks, stream 0 receives the microphone samples and the other N - 1 streams receive the same number of zero samples.

4. Handle partials, finals, and the shared timeline

AssemblyAI emits a Turn message more than once as a turn develops. Its transcript field contains the current text for the full turn, so each partial replaces the prior partial from that stream:

if not message["end_of_turn"]:
    timeline.update_partial(speaker, start, message["transcript"])

When end_of_turn is true, store the completed turn and remove that speaker’s partial:

timeline.finalize(
    speaker,
    message["turn_order"],
    start,
    message["transcript"],
)

The first word’s start value places the turn on the shared timeline. The terminal renderer sorts completed and partial turns by this timestamp. Results can arrive from the N AssemblyAI sessions in a different order, but their display order remains chronological.

Full code

Save this complete script as live_diarized_transcription_per_speaker.py:

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "sounddevice",
#     "numpy",
#     "requests",
#     "websockets>=13",
#     "python-dotenv",
# ]
# ///

import asyncio
import json
import os
import signal
from collections import deque
from contextlib import AsyncExitStack
from dataclasses import dataclass
from urllib.parse import urlencode

import numpy as np
import requests
import sounddevice as sd
import websockets
from dotenv import load_dotenv

load_dotenv()
PYANNOTEAI_API_KEY = os.environ["PYANNOTEAI_API_KEY"]
ASSEMBLYAI_API_KEY = os.environ["ASSEMBLYAI_API_KEY"]

SAMPLE_RATE = 16_000
CHUNK_SAMPLES = 1600
CHUNK_SECONDS = CHUNK_SAMPLES / SAMPLE_RATE
max_speakers = 2  # N, the known speaker-count upper bound
ROUTING_DELAY = 1.0


@dataclass
class AudioFrame:
    start: float
    samples: np.ndarray

    @property
    def end(self) -> float:
        return self.start + len(self.samples) / SAMPLE_RATE


class SpeakerRouter:
    def __init__(self) -> None:
        self.frames: deque[AudioFrame] = deque()
        self.events: list[tuple[float, str, bool]] = []
        self.active: set[str] = set()
        self.speaker_indexes: dict[str, int] = {}
        self.latest_time = 0.0
        self.warned_about_limit = False

    def add_frame(self, frame: AudioFrame) -> None:
        self.frames.append(frame)
        self.latest_time = frame.end

    def add_event(self, timestamp: float, speaker: str, started: bool) -> None:
        if started and speaker not in self.speaker_indexes:
            if len(self.speaker_indexes) < max_speakers:
                self.speaker_indexes[speaker] = len(self.speaker_indexes)
            elif not self.warned_about_limit:
                print(f"More than {max_speakers} speakers detected; extra speakers are muted.")
                self.warned_about_limit = True
        self.events.append((timestamp, speaker, started))
        self.events.sort(key=lambda event: event[0])

    def speaker_for_index(self, index: int) -> str:
        for speaker, speaker_index in self.speaker_indexes.items():
            if speaker_index == index:
                return speaker
        return f"SPEAKER_{index:02d}"

    def apply_event(self, speaker: str, started: bool) -> None:
        if started:
            self.active.add(speaker)
            return
        self.active.discard(speaker)

    def copy_active(self, outputs: np.ndarray, samples: np.ndarray, start: int, end: int) -> None:
        for speaker in self.active:
            index = self.speaker_indexes.get(speaker)
            if index is not None:
                outputs[index, start:end] = samples[start:end]

    def route(self, frame: AudioFrame) -> list[bytes]:
        outputs = np.zeros(
            (max_speakers, len(frame.samples)),
            dtype=frame.samples.dtype,
        )

        while self.events and self.events[0][0] <= frame.start:
            _, speaker, started = self.events.pop(0)
            self.apply_event(speaker, started)

        sample = 0
        while self.events and self.events[0][0] < frame.end:
            timestamp, speaker, started = self.events.pop(0)
            event_sample = round((timestamp - frame.start) * SAMPLE_RATE)
            event_sample = max(sample, min(len(frame.samples), event_sample))
            self.copy_active(outputs, frame.samples, sample, event_sample)
            self.apply_event(speaker, started)
            sample = event_sample

        self.copy_active(outputs, frame.samples, sample, len(frame.samples))
        return [
            (np.clip(output, -1, 1) * 32767).astype("<i2").tobytes()
            for output in outputs
        ]

    async def flush(self, streams, force: bool = False) -> None:
        cutoff = self.latest_time - ROUTING_DELAY
        while self.frames and (force or self.frames[0].end <= cutoff):
            frame = self.frames.popleft()
            for stream, pcm16 in zip(streams, self.route(frame)):
                await stream.send(pcm16)
            if force:
                await asyncio.sleep(CHUNK_SECONDS)


class TranscriptTimeline:
    def __init__(self) -> None:
        self.finals: dict[tuple[str, int], tuple[float, str, str]] = {}
        self.partials: dict[str, tuple[float, str, str]] = {}

    def update_partial(self, speaker: str, start: float, text: str) -> None:
        self.partials[speaker] = (start, speaker, text)
        self.render()

    def finalize(self, speaker: str, order: int, start: float, text: str) -> None:
        self.partials.pop(speaker, None)
        self.finals[(speaker, order)] = (start, speaker, text)
        self.render()

    def render(self) -> None:
        rows = [*self.finals.values(), *self.partials.values()]
        rows.sort(key=lambda row: row[0])
        print("\033[H\033[J", end="")
        print("Listening - speak now (Ctrl+C to stop)\n")
        for row in rows:
            suffix = " ..." if row in self.partials.values() else ""
            print(f"[{row[1]}] {row[2]}{suffix}")


def create_pyannote_stream() -> tuple[str, str]:
    response = requests.post(
        "https://api.pyannote.ai/v1/live",
        headers={"Authorization": f"Bearer {PYANNOTEAI_API_KEY}"},
    )
    response.raise_for_status()
    data = response.json()
    return data["id"], data["url"]


def microphone(loop, queue: "asyncio.Queue[np.ndarray]") -> sd.InputStream:
    def enqueue(frame: np.ndarray) -> None:
        try:
            queue.put_nowait(frame)
        except asyncio.QueueFull:
            pass

    def callback(indata, frames, time, status):
        loop.call_soon_threadsafe(enqueue, indata[:, 0].copy())

    stream = sd.InputStream(
        samplerate=SAMPLE_RATE,
        channels=1,
        dtype="float32",
        blocksize=CHUNK_SAMPLES,
        callback=callback,
    )
    stream.start()
    return stream


async def receive_pyannote(
    stream,
    router: SpeakerRouter,
    stop: asyncio.Event,
    done: asyncio.Event,
) -> None:
    try:
        async for raw in stream:
            message = json.loads(raw)
            message_type = message.get("type")
            data = message.get("data", {})
            if message_type == "diarization_speaker_start":
                router.add_event(data["timestamp"], data["speaker"], True)
            if message_type == "diarization_speaker_end":
                router.add_event(data["timestamp"], data["speaker"], False)
            if message_type == "error":
                print(f"pyannoteAI error: {message.get('message')}")
                stop.set()
    finally:
        done.set()
        if not stop.is_set():
            stop.set()


def turn_start(message: dict, fallback: float) -> float:
    words = message.get("words", [])
    if not words:
        return fallback
    return words[0]["start"] / 1000


async def receive_assemblyai(
    stream,
    index: int,
    router: SpeakerRouter,
    timeline: TranscriptTimeline,
    stop: asyncio.Event,
    done: asyncio.Event,
) -> None:
    try:
        async for raw in stream:
            message = json.loads(raw)
            if message.get("type") == "Turn" and message.get("transcript", "").strip():
                speaker = router.speaker_for_index(index)
                text = message["transcript"].strip()
                fallback = max(0.0, router.latest_time - ROUTING_DELAY)
                start = turn_start(message, fallback)
                if message["end_of_turn"]:
                    timeline.finalize(speaker, message["turn_order"], start, text)
                else:
                    timeline.update_partial(speaker, start, text)
            if message.get("type") == "Termination":
                done.set()
            if message.get("type") == "Error":
                print(f"AssemblyAI error for stream {index}: {message}")
                stop.set()
    finally:
        done.set()
        if not stop.is_set():
            stop.set()


async def pump_audio(
    queue: "asyncio.Queue[np.ndarray]",
    pyannote_stream,
    assembly_streams,
    router: SpeakerRouter,
    stop: asyncio.Event,
) -> None:
    samples_sent = 0
    while not stop.is_set():
        try:
            samples = await asyncio.wait_for(queue.get(), timeout=0.1)
        except asyncio.TimeoutError:
            continue
        await pyannote_stream.send(samples.astype("<f4").tobytes())
        router.add_frame(AudioFrame(samples_sent / SAMPLE_RATE, samples))
        samples_sent += len(samples)
        await router.flush(assembly_streams)


async def main() -> None:
    loop = asyncio.get_running_loop()
    stop = asyncio.Event()
    loop.add_signal_handler(signal.SIGINT, stop.set)

    stream_id, pyannote_url = create_pyannote_stream()
    print(f"pyannoteAI stream: {stream_id}")

    params = urlencode({
        "sample_rate": SAMPLE_RATE,
        "encoding": "pcm_s16le",
        "speech_model": "universal-3-5-pro",
        "speaker_labels": "false",
        "min_turn_silence": 100,
        "max_turn_silence": 1000,
        "continuous_partials": "true",
    })
    assembly_url = f"wss://streaming.assemblyai.com/v3/ws?{params}"

    async with AsyncExitStack() as stack:
        pyannote_stream = await stack.enter_async_context(websockets.connect(pyannote_url))
        assembly_streams = [
            await stack.enter_async_context(
                websockets.connect(
                    assembly_url,
                    additional_headers={"Authorization": ASSEMBLYAI_API_KEY},
                )
            )
            for _ in range(max_speakers)
        ]

        queue: "asyncio.Queue[np.ndarray]" = asyncio.Queue(maxsize=50)
        router = SpeakerRouter()
        timeline = TranscriptTimeline()
        mic = microphone(loop, queue)
        timeline.render()

        pyannote_done = asyncio.Event()
        assembly_done = [asyncio.Event() for _ in assembly_streams]
        pyannote_task = asyncio.create_task(
            receive_pyannote(pyannote_stream, router, stop, pyannote_done)
        )
        assembly_tasks = [
            asyncio.create_task(
                receive_assemblyai(
                    stream,
                    index,
                    router,
                    timeline,
                    stop,
                    assembly_done[index],
                )
            )
            for index, stream in enumerate(assembly_streams)
        ]
        pump_task = asyncio.create_task(
            pump_audio(queue, pyannote_stream, assembly_streams, router, stop)
        )
        tasks = [pyannote_task, *assembly_tasks, pump_task]

        await stop.wait()
        mic.stop()
        mic.close()
        await pump_task
        await pyannote_stream.send(json.dumps({"type": "end_of_stream"}))
        await pyannote_done.wait()
        await router.flush(assembly_streams, force=True)
        for stream in assembly_streams:
            await stream.send(json.dumps({"type": "Terminate"}))
        await asyncio.gather(*(done.wait() for done in assembly_done))
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)


if __name__ == "__main__":
    asyncio.run(main())

Run the script

Run the file from the directory that contains .env:

uv run live_diarized_transcription_per_speaker.py

Speak into your microphone. Partial text updates in place. Completed AssemblyAI turns remain on screen and sort by their first word timestamp. Press Ctrl+C to stop. The script stops microphone capture, waits for pyannoteAI to finalize its speaker events, flushes the delayed audio, and waits for all N AssemblyAI sessions to terminate with their final turns.

Next steps

To compare both designs, read Combining real-time diarization and transcription.
To use one transcription stream with reconciliation logic and avoid paying for one session per speaker slot, see Approach 1: single STT stream.
Set max_speakers to the smallest reliable upper bound for your conversation. The script opens and pays for that many AssemblyAI sessions for the full call, including silent sessions. If more speakers join, the router mutes the extra speakers.
pyannoteAI live diarization supports up to eight speakers. If more than eight speakers join, it may merge speakers under the same labels.
For production use, measure pyannoteAI event latency on your network and set ROUTING_DELAY above a high percentile of that measurement.

You can use another streaming transcription provider if it accepts continuous audio or explicit silence for each speaker stream. Change the audio conversion, connection setup, and event handler to match that provider. Partials may append text or replace the complete partial, turn-final signals use provider-specific fields, and timestamps may use a session clock or start at the first speech frame. Keep every stream on one common audio clock before merging its results.

Getting Started

Pre-recorded audio

Live audio

More tutorials

Webhooks

Support

Administration

Live diarized transcription: per-speaker transcription streams (Approach 2)

Why use one transcription stream per speaker?

Build a live speaker-attributed transcript

Prerequisites

Architecture overview

1. Choose the speaker upper bound and open a pyannoteAI stream

2. Open N AssemblyAI streams

3. Route audio to each speaker

4. Handle partials, finals, and the shared timeline

Full code

Run the script

Next steps

​Why use one transcription stream per speaker?

​Build a live speaker-attributed transcript

​Prerequisites

​Architecture overview

​1. Choose the speaker upper bound and open a pyannoteAI stream

​2. Open N AssemblyAI streams

​3. Route audio to each speaker

​4. Handle partials, finals, and the shared timeline

​Full code

​Run the script

​Next steps

Why use one transcription stream per speaker?

Build a live speaker-attributed transcript

Prerequisites

Architecture overview

1. Choose the speaker upper bound and open a pyannoteAI stream

2. Open N AssemblyAI streams

3. Route audio to each speaker

4. Handle partials, finals, and the shared timeline

Full code

Run the script

Next steps