whisper-speech-transcription

name	whisper-speech-transcription
metadata	{"category":"NLP Audio and Speech AI"}
description	Deploy OpenAI Whisper and Faster-Whisper for high-throughput batch and streaming speech-to-text (STT) transcription. Triggers when implementing CTranslate2 GPU acceleration, Voice Activity Detection (Silero VAD), word-level timestamps, speaker diarization (PyAnnote), FFmpeg audio pre-processing, dynamic language detection, or WebSocket streaming transcription APIs.
compatibility	Python (>= 3.9), faster-whisper (>= 1.0.0), PyTorch, Silero VAD, FFmpeg, CUDA (>= 11.8)

Whisper Speech Transcription & Diarization

Production patterns for deploying high-performance Speech-to-Text (STT) pipelines using Faster-Whisper, Silero VAD, PyAnnote Speaker Diarization, and WebSockets.

---

1. System Architecture

+----------------------+      +---------------------------------+      +---------------------------------+
| Input Audio Stream   | ---> | FFmpeg Audio Pre-processing     | ---> | Silero VAD                      |
| (Microphone/WAV/MP3) |      | (Resample to 16kHz Mono PCM)    |      | (Voice Activity Chunk Filter)   |
+----------------------+      +---------------------------------+      +---------------------------------+
                                                                                       |
                                                                                       v
+----------------------+      +---------------------------------+      +---------------------------------+
| Output JSON Transcript| <--- | PyAnnote Speaker Diarization    | <--- | Faster-Whisper Engine           |
| (Words, Timestamps)  |      | (Assign Speaker IDs)            |      | (CTranslate2 FP16/INT8 GPU)     |
+----------------------+      +---------------------------------+      +---------------------------------+

---

2. High-Performance Batch Transcription Engine (transcribe_engine.py)

Using faster-whisper (CTranslate2) for 4x faster execution and 50% lower VRAM usage compared to standard PyTorch implementations.

import os
import logging
from faster_whisper import WhisperModel
import torch

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class HighPerformanceWhisperEngine:
    def __init__(self, model_size: str = "large-v3", device: str = "cuda", compute_type: str = "float16"):
        logger.info(f"Loading Faster-Whisper model '{model_size}' on {device} ({compute_type})...")
        
        # Fallback to float32 on CPU
        if not torch.cuda.is_available():
            device = "cpu"
            compute_type = "int8"

        self.model = WhisperModel(
            model_size_or_path=model_size,
            device=device,
            compute_type=compute_type,
            cpu_threads=4,
            num_workers=2
        )

    def transcribe_file(self, audio_filepath: str, language: str = None) -> dict:
        """
        Transcribes an audio file with Voice Activity Detection (VAD) and word timestamps.
        """
        segments, info = self.model.transcribe(
            audio_filepath,
            language=language,
            beam_size=5,
            best_of=5,
            vad_filter=True,  # Built-in Silero VAD
            vad_parameters=(
                min_speech_duration_ms=,
                max_speech_duration_s=,
                min_silence_duration_ms=,
                speech_pad_ms=
            ),
            word_timestamps=
        )

        logger.info()

        formatted_segments = []
        full_text = []

         segment  segments:
            full_text.append(segment.text)
            
            words_list = []
             segment.words:
                 w  segment.words:
                    words_list.append({
                        : w.word,
                        : (w.start, ),
                        : (w.end, ),
                        : (w.probability, )
                    })

            formatted_segments.append({
                : segment.,
                : (segment.start, ),
                : (segment.end, ),
                : segment.text.strip(),
                : words_list
            })

         {
            : info.language,
            : (info.duration, ),
            : .join(full_text).strip(),
            : formatted_segments
        }

 __name__ == :
    engine = HighPerformanceWhisperEngine(model_size=, compute_type=)
    result = engine.transcribe_file()
    (result[])

---

3. Combining Whisper with PyAnnote Speaker Diarization (diarized_transcribe.py)

Assign precise speaker labels ("Speaker A", "Speaker B") to Whisper segment timestamps.

import torch
from pyannote.audio import Pipeline
from transcribe_engine import HighPerformanceWhisperEngine
import numpy as np

def transcribe_with_diarization(audio_filepath: str, hf_token: str):
    # 1. Initialize Whisper STT Engine
    whisper_engine = HighPerformanceWhisperEngine(model_size="large-v3")
    stt_result = whisper_engine.transcribe_file(audio_filepath)

    # 2. Initialize PyAnnote Speaker Diarization Pipeline
    diarization_pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )
    if torch.cuda.is_available():
        diarization_pipeline.to(torch.device("cuda"))

    # 3. Execute Diarization
    diarization = diarization_pipeline(audio_filepath)

    # 4. Merge Diarization Speakers with Whisper Segments
    final_transcript = []
    
    for segment in stt_result["segments"]:
        seg_start = segment["start"]
        seg_end = segment["end"]

        # Find speaker with maximum overlap during segment timeframe
        speaker_overlaps = {}
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            overlap_start = max(seg_start, turn.start)
            overlap_end = min(seg_end, turn.end)
            
            if overlap_end > overlap_start:
                duration = overlap_end - overlap_start
                speaker_overlaps[speaker] = speaker_overlaps.get(speaker, 0.0) + duration

        assigned_speaker = max(speaker_overlaps, key=speaker_overlaps.get) if speaker_overlaps  

        final_transcript.append({
            : assigned_speaker,
            : seg_start,
            : seg_end,
            : segment[]
        })

     final_transcript

---

4. Real-Time Streaming Audio Server via WebSockets (websocket_server.py)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from faster_whisper import WhisperModel
import numpy as np
import asyncio

app = FastAPI(title="Whisper Live Streaming Audio API")

# Load small/fast model for low-latency streaming
MODEL = WhisperModel("tiny.en", device="cuda", compute_type="float16")

@app.websocket("/ws/transcribe")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    audio_buffer = bytearray()
    SAMPLE_RATE = 16000
    BYTES_PER_SAMPLE = 2  # 16-bit PCM
    CHUNK_DURATION_SEC = 2
    REQUIRED_BYTES = SAMPLE_RATE * BYTES_PER_SAMPLE * CHUNK_DURATION_SEC

    try:
        while True:
            data = await websocket.receive_bytes()
            audio_buffer.extend(data)

            if len(audio_buffer) >= REQUIRED_BYTES:
                # Convert raw PCM int16 bytes to float32 NumPy array
                pcm_data = np.frombuffer(audio_buffer[:REQUIRED_BYTES], dtype=np.int16).astype(np.float32) / 32768.0
                audio_buffer = audio_buffer[REQUIRED_BYTES:]

                # Run Whisper inference on chunk
                segments, _ = MODEL.transcribe(pcm_data, language="en", beam_size=1)
                
                text_chunk = " ".join([s.text for s in segments]).strip()
                if text_chunk:
                     websocket.send_json({: text_chunk})

     WebSocketDisconnect:
        ()

---

5. Production Optimization Guidelines

1. Audio Pre-processing: Convert input streams using FFmpeg to 16kHz single-channel mono PCM (ffmpeg -i input.mp3 -ar 16000 -ac 1 -f s16le output.raw) prior to passing to model buffers.
2. Batching: Use vad_filter=True to strip silence chunks before invoking CTranslate2 matrix computations; this increases throughput by up to 300% on dialogue files.
3. Quantization Selection: Use compute_type="float16" on NVIDIA GPUs (Ampere/Ada architectures) and compute_type="int8" for CPU deployments.