Live diarized transcription: single transcription stream (Approach 1)

Why speaker attribution is hard in real time

Live captions, meeting assistants, and call analytics need both the words being spoken and a speaker label for each turn before the conversation ends. pyannoteAI streaming diarization provides speaker labels and turn boundaries as audio arrives. A streaming speech-to-text service provides the words. Because both APIs emit events independently, the client must combine their outputs into one live, speaker-attributed transcript. This approach opens one diarization stream and one transcription stream. The transcription stream receives audio from all speakers. One transcription session keeps session costs independent of the speaker count, but requires reconciliation logic to associate transcript events with speaker turns. The per-speaker approach uses one transcription session per speaker slot and buffers audio before routing it. Evaluate both approaches with your audio, cost, and latency requirements.

Build a live speaker-attributed transcript

In this tutorial, we will build a Python script that listens to your microphone and prints a live, speaker-attributed transcript:

[SPEAKER_00] Hello
[SPEAKER_00] This is transcribing while diarization is running
[SPEAKER_01] Now another person is speaking, and the transcript switches speakers.
[SPEAKER_00] And now the first speaker is back.

Read Combining real-time diarization and transcription for the concepts behind both designs. This tutorial implements Approach 1. To give each known speaker a separate transcription stream, see Approach 2: per-speaker transcription streams.

Prerequisites

A pyannoteAI API key from the dashboard
An OpenAI API key
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
OPENAI_API_KEY=sk-xxx

Architecture overview

The script captures each microphone frame once and sends it to both streaming APIs: pyannoteAI receives 16 kHz mono float32 PCM and emits speaker start and end events. OpenAI receives the same audio converted to 24 kHz PCM16 and emits transcript deltas and completed segments. Shared state joins these two event streams.

This approach sends mixed microphone audio to one transcription stream. It does not separate simultaneous speakers, so overlapping speech can share a transcript segment.

1. Open two streaming sessions

First, create a pyannoteAI stream. The response contains an ID and a single-use WebSocket URL:

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


stream_id, url = create_stream()
client = AsyncOpenAI(api_key=OPENAI_API_KEY)

async with websockets.connect(url) as ws:
    async with client.realtime.connect(extra_query={"intent": "transcription"}) as conn:
        await conn.session.update(session={
            "type": "transcription",
            "audio": {"input": {
                "format": {"type": "audio/pcm", "rate": OPENAI_RATE},
                "transcription": {"model": "gpt-realtime-whisper", "language": "en"},
                "turn_detection": None,
            }},
        })
        ...

Both connections now exist before microphone capture starts. The pyannoteAI WebSocket accepts 16 kHz mono float32 PCM chunks every 100 ms. See Streaming Diarization for the full stream format and event reference. pyannoteAI controls the speaker boundaries and triggers each transcription commit. OpenAI uses gpt-realtime-whisper with turn detection disabled to transcribe those turns.

2. Capture the microphone once and fan out

The microphone callback puts each 100 ms frame into one asynchronous queue. Capturing once keeps both APIs on the same audio timeline:

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,
        callback=callback,
    )
    stream.start()
    return stream

Send the original float32 frame to pyannoteAI. Convert that same frame to the format OpenAI expects, then append it to the OpenAI input buffer:

async def pump(queue: "asyncio.Queue[np.ndarray]", ws, convo: Conversation, conn) -> None:
    while True:
        frame = await queue.get()
        await ws.send(frame.astype("<f4").tobytes())
        pcm16 = (np.clip(upsample(frame), -1, 1) * 32767).astype("<i2").tobytes()
        await conn.input_audio_buffer.append(audio=base64.b64encode(pcm16).decode())
        convo.uncommitted = True

3. Build the reconciler

The reconciler tracks four pieces of shared state:

active contains the speakers pyannoteAI currently hears. The most recently active speaker labels live transcript deltas.
pending is a queue of pyannoteAI speaker labels waiting for OpenAI to acknowledge a committed audio buffer.
pyannote_speaker_by_item associates each OpenAI transcript item with its pyannoteAI speaker label.
uncommitted records whether the OpenAI buffer contains audio that can be committed.

class Conversation:
    def __init__(self) -> None:
        self.active: list[str] = []
        self.pending: deque[str] = deque()
        self.pyannote_speaker_by_item: dict[str, str] = {}
        self.uncommitted = False

    def started(self, speaker: str) -> None:
        if speaker not in self.active:
            self.active.append(speaker)

    def ended(self, speaker: str) -> None:
        if speaker in self.active:
            self.active.remove(speaker)

    @property
    def speaker(self) -> str | None:
        return self.active[-1] if self.active else None

    def committed(self, item_id: str) -> None:
        if self.pending:
            self.pyannote_speaker_by_item[item_id] = self.pending.popleft()

    def speaker_for(self, item_id: str) -> str | None:
        return self.pyannote_speaker_by_item.get(item_id, self.speaker)

    def completed(self, item_id: str) -> str | None:
        return self.pyannote_speaker_by_item.pop(item_id, self.speaker)

4. Attribute completed text

When pyannoteAI reports the end of a speaker turn, add that pyannoteAI speaker label to pending and commit the OpenAI audio buffer. Committing the buffer instructs OpenAI to complete the transcript segment for that turn:

elif msg.get("type") == "diarization_speaker_end":
    convo.ended(data["speaker"])
    if convo.uncommitted:
        convo.pending.append(data["speaker"])
        convo.uncommitted = False
        await conn.input_audio_buffer.commit()

OpenAI acknowledges the commit with an item_id. Associate that ID with the pending pyannoteAI label. Completion events can arrive out of order, so use item_id to recover the correct pyannoteAI speaker:

elif et.endswith("input_audio_buffer.committed"):
    convo.committed(event.item_id)
elif et.endswith("transcription.completed"):
    sp = convo.completed(event.item_id)
    text = event.transcript.strip()
    if text:
        tag = f"{color(sp)}[{sp}]{_RESET} " if sp else ""
        print(f"\r{tag}{text}\033[K")

The script uses OpenAI’s item ID only to match transcription events. Every speaker label comes from pyannoteAI.

5. Update partial attribution as text arrives

OpenAI sends partial transcript deltas before a segment is complete. Store partial text by item_id, then display it with the associated pyannoteAI label:

elif et.endswith("transcription.delta"):
    partial = partials.get(event.item_id, "") + event.delta
    partials[event.item_id] = partial
    sp = convo.speaker_for(event.item_id)
    tag = f"{color(sp)}[{sp}]{_RESET} " if sp else ""
    print(f"\r{tag}{partial}\033[K", end="", flush=True)

Each delta uses the latest pyannoteAI attribution available for that transcript item. Once OpenAI acknowledges the commit, item_id fixes the completed text to the queued pyannoteAI label.

Full code

Here is the complete script:

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

import asyncio
import base64
import json
import os
import signal
from collections import deque

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

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

SAMPLE_RATE = 16_000
OPENAI_RATE = 24_000
CHUNK = 1600

_COLORS = ["\033[32m", "\033[33m", "\033[34m", "\033[35m", "\033[36m", "\033[31m", "\033[37m", "\033[93m"]
_RESET = "\033[0m"
_colors: dict[str, str] = {}


def color(speaker: str) -> str:
    _colors.setdefault(speaker, _COLORS[len(_colors) % len(_COLORS)])
    return _colors[speaker]


class Conversation:
    def __init__(self) -> None:
        self.active: list[str] = []
        self.pending: deque[str] = deque()
        self.pyannote_speaker_by_item: dict[str, str] = {}
        self.uncommitted = False

    def started(self, speaker: str) -> None:
        if speaker not in self.active:
            self.active.append(speaker)

    def ended(self, speaker: str) -> None:
        if speaker in self.active:
            self.active.remove(speaker)

    @property
    def speaker(self) -> str | None:
        return self.active[-1] if self.active else None

    def committed(self, item_id: str) -> None:
        if self.pending:
            self.pyannote_speaker_by_item[item_id] = self.pending.popleft()

    def speaker_for(self, item_id: str) -> str | None:
        return self.pyannote_speaker_by_item.get(item_id, self.speaker)

    def completed(self, item_id: str) -> str | None:
        return self.pyannote_speaker_by_item.pop(item_id, self.speaker)


def upsample(x: np.ndarray) -> np.ndarray:
    n = len(x) * OPENAI_RATE // SAMPLE_RATE
    return np.interp(np.linspace(0, len(x) - 1, n), np.arange(len(x)), x).astype(np.float32)


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


async def pyannote_task(ws, convo: Conversation, conn) -> None:
    async for raw in ws:
        msg = json.loads(raw)
        data = msg.get("data", {})
        if msg.get("type") == "diarization_speaker_start":
            convo.started(data["speaker"])
        elif msg.get("type") == "diarization_speaker_end":
            convo.ended(data["speaker"])
            if convo.uncommitted:
                convo.pending.append(data["speaker"])
                convo.uncommitted = False
                await conn.input_audio_buffer.commit()
        elif msg.get("type") == "error":
            print(f"\npyannote error: {msg.get('message')}")


async def openai_task(conn, convo: Conversation) -> None:
    partials: dict[str, str] = {}
    async for event in conn:
        et = event.type
        if et.endswith("transcription_session.created"):
            print(f"OpenAI session:  {event.session.id}")
        elif et.endswith("input_audio_buffer.committed"):
            convo.committed(event.item_id)
        elif et.endswith("transcription.delta"):
            partial = partials.get(event.item_id, "") + event.delta
            partials[event.item_id] = partial
            sp = convo.speaker_for(event.item_id)
            tag = f"{color(sp)}[{sp}]{_RESET} " if sp else ""
            print(f"\r{tag}{partial}\033[K", end="", flush=True)
        elif et.endswith("transcription.completed"):
            partials.pop(event.item_id, None)
            sp = convo.completed(event.item_id)
            text = event.transcript.strip()
            if text:
                tag = f"{color(sp)}[{sp}]{_RESET} " if sp else ""
                print(f"\r{tag}{text}\033[K")


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, callback=callback)
    stream.start()
    return stream


async def pump(queue: "asyncio.Queue[np.ndarray]", ws, convo: Conversation, conn) -> None:
    while True:
        frame = await queue.get()
        await ws.send(frame.astype("<f4").tobytes())
        pcm16 = (np.clip(upsample(frame), -1, 1) * 32767).astype("<i2").tobytes()
        await conn.input_audio_buffer.append(audio=base64.b64encode(pcm16).decode())
        convo.uncommitted = True


async def main() -> None:
    loop = asyncio.get_running_loop()
    convo = Conversation()
    queue: "asyncio.Queue[np.ndarray]" = asyncio.Queue(maxsize=50)

    stream_id, url = create_stream()
    print(f"pyannoteAI stream: {stream_id}")

    client = AsyncOpenAI(api_key=OPENAI_API_KEY)
    async with websockets.connect(url) as ws:
        async with client.realtime.connect(extra_query={"intent": "transcription"}) as conn:
            await conn.session.update(session={
                "type": "transcription",
                "audio": {"input": {
                    "format": {"type": "audio/pcm", "rate": OPENAI_RATE},
                    "transcription": {"model": "gpt-realtime-whisper", "language": "en"},
                    "turn_detection": None,
                }},
            })

            mic = microphone(loop, queue)
            print("\nListening — speak now (Ctrl+C to stop)\n")

            stop = asyncio.Event()
            loop.add_signal_handler(signal.SIGINT, stop.set)
            tasks = [
                asyncio.create_task(pyannote_task(ws, convo, conn)),
                asyncio.create_task(openai_task(conn, convo)),
                asyncio.create_task(pump(queue, ws, convo, conn)),
            ]
            await asyncio.wait({*tasks, asyncio.create_task(stop.wait())}, return_when=asyncio.FIRST_COMPLETED)

            print("\nStopping...")
            mic.stop()
            mic.close()
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            await ws.send(json.dumps({"type": "end_of_stream"}))


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

Run the script

Save the full code as live_diarized_transcription.py, then run it:

uv run live_diarized_transcription.py

Speak into your microphone. Partial text updates in place while OpenAI streams deltas. Completed turns print as separate lines with speaker labels like [SPEAKER_00]. Example output:

uv run live_diarized_transcription.py
pyannoteAI stream: 7d3f4a21-9b6c-4f8a-8f12-3c9d8b6e2a44

Listening — speak now (Ctrl+C to stop)

[SPEAKER_00] Hello
[SPEAKER_00] This is transcribing while diarization is running
[SPEAKER_00] The speaker label stays attached to this turn
[SPEAKER_01] Now another person is speaking, and the live transcript switches speakers.
[SPEAKER_01] It keeps printing completed turns as they arrive.
[SPEAKER_00] And now the first speaker is back.
^C
Stopping...

Next steps

Compare this reconciliation logic with per-speaker transcription streams (Approach 2).
Read Combining real-time diarization and transcription for the tradeoffs between both approaches.
Replace terminal output with WebSocket broadcasts to your frontend.
Store completed {speaker, text} turns in your database.
Add timestamps from pyannote events if your UI needs time-aligned captions.

Getting Started

Pre-recorded audio

Live audio

More tutorials

Webhooks

Support

Administration

Live diarized transcription: single transcription stream (Approach 1)

Why speaker attribution is hard in real time

Build a live speaker-attributed transcript

Prerequisites

Architecture overview

1. Open two streaming sessions

2. Capture the microphone once and fan out

3. Build the reconciler

4. Attribute completed text

5. Update partial attribution as text arrives

Full code

Run the script

Next steps

​Why speaker attribution is hard in real time

​Build a live speaker-attributed transcript

​Prerequisites

​Architecture overview

​1. Open two streaming sessions

​2. Capture the microphone once and fan out

​3. Build the reconciler

​4. Attribute completed text

​5. Update partial attribution as text arrives

​Full code

​Run the script

​Next steps

Why speaker attribution is hard in real time

Build a live speaker-attributed transcript

Prerequisites

Architecture overview

1. Open two streaming sessions

2. Capture the microphone once and fan out

3. Build the reconciler

4. Attribute completed text

5. Update partial attribution as text arrives

Full code

Run the script

Next steps