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

> Learn to implement a custom STT handler for new speech-to-speech models in Hugging Face. Subclass BaseHandler, load your model in setup(), and process audio for transcriptions in s2s_pipeline.py.

- Repository: [Hugging Face/speech-to-speech](https://github.com/huggingface/speech-to-speech)
- Tags: how-to-guide
- Published: 2026-08-06

---

**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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/my_model_handler.py). The library organizes legacy and reference handlers in this directory.

### Step 2: ImportBase Classes and Dependencies

```python
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

```python
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

```python
    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

```python
    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`](https://github.com/huggingface/speech-to-speech/blob/main/archive/STT/__init__.py):

```python
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:

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/archive/STT/moonshine_handler.py) file demonstrates this pattern with the Moonshine model family. Key excerpts showing the same structural elements:

```python
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:

```python
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:

```bash
pytest -q

```

## Summary

- **Inherit from `BaseHandler[VADAudio]`** — located in [`src/speech_to_speech/baseHandler.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/archive/STT/moonshine_handler.py)
- **Wire into `S2SPipeline`** — pass handler class and kwargs to `stt_handler` in [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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.