How to Implement a Custom STT Handler for New Models in HuggingFace Speech‑to‑Speech

Subclass BaseHandler[VADAudio], implement setup() to load your model, and implement process() to yield (transcription, language_code) tuples—then wire it into the S2SPipeline in s2s_pipeline.py.

The HuggingFace speech‑to‑speech repository provides a modular, queue‑based pipeline where each processing stage runs as an independent thread. To add support for a new speech‑to‑text (STT) model, you create a custom STT handler that inherits from BaseHandler and conforms to the library's messaging protocol. This guide walks through the implementation using the actual source code structure, with MoonshineSTTHandler in archive/STT/moonshine_handler.py serving as the reference pattern.

Core Architecture: The BaseHandler Abstraction

Every handler in the speech‑to‑speech pipeline extends BaseHandler from src/speech_to_speech/baseHandler.py. This base class manages:

• Thread lifecycle — automatic start/stop with stop_event signaling
• Queue management — queue_in receives VADAudio objects; queue_out delivers results downstream
• Control messages — PipelineControlMessage and SESSION_END from src/speech_to_speech/pipeline/control.py allow per‑session state resets
• Timing and logging — built‑in performance tracking with threshold warnings

The generic signature BaseHandler[T] lets you specify the input type. For STT handlers, this is always VADAudio, defined in src/speech_to_speech/pipeline/messages.py.

Required Methods for a Custom STT Handler

A minimal custom STT handler must implement three hook methods:

Method	Purpose	Called When
setup(self, **kwargs)	Load model, tokenizer, device; configure generation parameters	Once, before the thread starts processing
warmup(self)	Run dummy inference to prime caches (optional but recommended)	From setup() or explicitly after initialization
process(self, vad_audio: VADAudio)	Execute inference on audio chunk and yield results	For every item in queue_in

The process() method must yield a tuple. The convention is (transcription: str, language_code: str), though you can adapt this to your pipeline's needs.

Complete Implementation Walkthrough

Step 1: Create the Handler File

Create a new file in archive/STT/—for example, my_model_handler.py. The library organizes legacy and reference handlers in this directory.

Step 2: ImportBase Classes and Dependencies

import logging
import torch
from speech_to_speech.baseHandler import BaseHandler
from speech_to_speech.pipeline.messages import VADAudio

logger = logging.getLogger(__name__)

Step 3: Define the Handler Class with Type Annotation

class MyModelSTTHandler(BaseHandler[VADAudio]):
    """Custom STT handler wrapping a HuggingFace Transformers speech-to-text model."""
    
    def setup(self, model_path: str, device: str = "cpu", **gen_kwargs):
        from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
        
        self.device = device
        self.gen_kwargs = gen_kwargs
        
        # Load processor and model

        self.processor = AutoProcessor.from_pretrained(model_path)
        self.model = AutoModelForSpeechSeq2Seq.from_pretrained(model_path).to(device)
        
        # Prime the model to avoid first-call latency

        self.warmup()

Step 4: Implement Warm‑Up to Eliminate Cold Start

    def warmup(self):
        """Execute dummy inference to initialize CUDA kernels and allocate buffers."""
        logger.info(f"Warming up {self.__class__.__name__} …")
        dummy_audio = torch.randn(1, 16000).to(self.device)  # 1 second of fake audio

        with torch.no_grad():
            _ = self.model.generate(dummy_audio, **self.gen_kwargs)

Step 5: Implement the Core process() Method

    def process(self, vad_audio: VADAudio):
        """
        Receive VADAudio, run STT inference, yield (transcription, language).
        
        VADAudio attributes:
            - audio: numpy array of float32 waveform samples
            - sample_rate: integer sampling frequency (typically 16000)
        """
        # Prepare inputs for the model

        inputs = self.processor(
            vad_audio.audio,
            sampling_rate=vad_audio.sample_rate,
            return_tensors="pt"
        )
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        # Generate transcription

        with torch.no_grad():
            generated_ids = self.model.generate(**inputs, **self.gen_kwargs)
        
        # Decode to text

        transcription = self.processor.batch_decode(
            generated_ids,
            skip_special_tokens=True
        )[0]
        
        # Yield result tuple; language detection can be added here

        yield (transcription, "en")

Step 6: Expose the Handler for Import

Add to archive/STT/__init__.py:

from .my_model_handler import MyModelSTTHandler

__all__ = ["MyModelSTTHandler"]

Step 7: Integrate into the Pipeline

In your application code or a custom pipeline script, instantiate and inject your handler:

from threading import Event
from speech_to_speech.s2s_pipeline import S2SPipeline
from speech_to_speech.STT.my_model_handler import MyModelSTTHandler

stop_event = Event()

# Instantiate with configuration

stt_handler = MyModelSTTHandler(
    stop_event=stop_event,
    queue_in=vad_queue,      # queue receiving VADAudio from voice activity detection

    queue_out=text_queue,    # queue delivering results to LLM handler

    model_path="openai/whisper-small",
    device="cuda",
    max_new_tokens=256
)

# Or use with S2SPipeline factory

pipeline = S2SPipeline(
    stt_handler=MyModelSTTHandler,
    stt_kwargs=dict(
        model_path="openai/whisper-small",
        device="cuda",
        max_new_tokens=256,
        do_sample=False
    ),
    # ... other handler configurations

)

pipeline.start()

# Audio flows: microphone → VAD → MyModelSTTHandler → LLM → TTS → speaker

Reference Implementation: MoonshineSTTHandler

The archive/STT/moonshine_handler.py file demonstrates this pattern with the Moonshine model family. Key excerpts showing the same structural elements:

class MoonshineSTTHandler(BaseHandler[VADAudio]):
    def setup(self, model_name="moonshine/base", torch_dtype="float16", **gen_kwargs):
        self.torch_dtype = getattr(torch, torch_dtype)
        self.gen_kwargs = gen_kwargs
        
        # Moonshine-specific loading

        self.tokenizer = moonshine.load_tokenizer()
        self.model = moonshine.load_model(model_name)
        
        self.warmup()

    def warmup(self):
        # Model-specific dummy generation

        dummy = torch.zeros(1, 16000, dtype=self.torch_dtype)
        _ = self.model.generate(dummy, **self.gen_kwargs)

    def process(self, vad_audio: VADAudio):
        # Moonshine expects batch dimension

        pred_ids = self.model.generate(vad_audio.audio[None, :], **self.gen_kwargs)
        pred_text = self.tokenizer.decode_batch(pred_ids)[0]
        yield (pred_text, "en")

Advanced: Hooks for Input/Output Filtering

BaseHandler provides optional hooks for fine‑grained control:

• should_process_input(item) — return False to skip stale or cancelled audio chunks
• should_emit_output(result) — return False to suppress unwanted outputs

Override these in your handler if your pipeline requires filtering based on timestamps, confidence scores, or session state.

Testing Your Custom STT Handler

Verify implementation with a minimal test script:

import wave
import numpy as np
import queue
from threading import Event

from speech_to_speech.pipeline.messages import VADAudio
from speech_to_speech.STT.my_model_handler import MyModelSTTHandler

# Load test audio

with wave.open("tests/audio/sample.wav") as wf:
    frames = wf.readframes(wf.getnframes())
    audio = np.frombuffer(frames, dtype=np.int16).astype(np.float32) / 32768.0
    vad_audio = VADAudio(
        audio=audio,
        sample_rate=wf.getframerate()
    )

# Set up queues and run handler directly

stop_event = Event()
q_in = queue.Queue()
q_out = queue.Queue()

handler = MyModelSTTHandler(
    stop_event=stop_event,
    queue_in=q_in,
    queue_out=q_out,
    model_path="openai/whisper-tiny",
    device="cpu"
)

# Start handler thread

handler.start()
q_in.put(vad_audio)

# Retrieve result

result = q_out.get(timeout=30)
print(f"Transcription: {result[0]}")
print(f"Language: {result[1]}")

# Signal shutdown

stop_event.set()
handler.join()

Run the full test suite to ensure integration compatibility:

pytest -q

Summary

• Inherit from BaseHandler[VADAudio] — located in src/speech_to_speech/baseHandler.py
• Implement setup() — load model, tokenizer, configure device and generation parameters
• Call warmup() — eliminate first‑inference latency with dummy forward passes
• Yield from process() — return (transcription, language_code) tuples for each VADAudio input
• Register in archive/STT/ — follow the MoonshineSTTHandler pattern from archive/STT/moonshine_handler.py
• Wire into S2SPipeline — pass handler class and kwargs to stt_handler in src/speech_to_speech/s2s_pipeline.py

Frequently Asked Questions

What input format does process() receive?

The process() method receives a VADAudio object from src/speech_to_speech/pipeline/messages.py. This dataclass contains audio (numpy array of float32 samples) and sample_rate (integer, typically 16000 Hz). The voice activity detection (VAD) stage upstream segments continuous audio into these chunks.

Can I change the output format from (transcription, language)?

Yes. While the convention is a two‑element tuple, downstream handlers in your pipeline must expect the structure you yield. If you need additional metadata—confidence scores, word‑level timestamps, or alternative hypotheses—yield a dictionary or custom dataclass instead, and update consuming handlers accordingly.

How do I handle different sampling rates?

Resampling is typically handled upstream in the VAD or audio capture stage. If your model requires a specific rate, check vad_audio.sample_rate in process() and resample using librosa or torchaudio before feeding to the model. The Whisper processor in the example above accepts variable rates and handles resampling internally.

Why is warmup() important for real‑time pipelines?

The first CUDA kernel launch, memory allocation, and model graph compilation introduce significant latency—often hundreds of milliseconds. warmup() executes dummy inference during initialization so that real user audio experiences consistent, low‑latency processing. Without it, the first utterance in a session stalls noticeably.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}