How to Add a Custom STT Backend to the Speech-to-Speech Pipeline

To add a custom STT backend, inherit from BaseSTTHandler, create a configuration dataclass extending ArgumentsBase, register your identifier in the STTBackend enum, and wire the handler into the get_stt_handler() dispatch function in s2s_pipeline.py.

The Hugging Face speech-to-speech repository dynamically constructs its processing pipeline, allowing seamless swapping of Speech-to-Text (STT) backends through a centralized dispatch mechanism. Whether integrating a proprietary model or an experimental architecture, you can add a custom STT backend by implementing six concrete integration points that hook into the existing speculative-turn and stale-input filtering infrastructure.

Understanding the STT Handler Architecture

Before writing code, understand the two core components that manage backend instantiation.

The Base Handler Contract

All STT handlers inherit from BaseSTTHandler defined in src/speech_to_speech/STT/base_stt_handler.py (lines 16-22). This base class enforces a strict constructor signature: stop_event, queue_in, queue_out, and setup_kwargs. By inheriting from this class, your implementation automatically receives speculative-turn filtering and stale-input dropping, ensuring consistent behavior with built-in backends like Whisper and Paraformer.

The Dispatch Mechanism

The pipeline instantiates STT backends through get_stt_handler() in src/speech_to_speech/s2s_pipeline.py (lines 749-789). This function inspects module_kwargs.stt—populated from CLI flags or JSON configuration—to determine which concrete handler class to return. The value must match an entry in the STTBackend enum defined in src/speech_to_speech/arguments_classes/module_arguments.py.

Step-by-Step Implementation Guide

Follow these six steps to fully integrate a custom STT backend.

Step 1: Create the Handler Class

Implement a new file at src/speech_to_speech/STT/<your_name>_handler.py. Inherit from BaseSTTHandler and implement the run() method to process STTIn messages (typically VADAudio objects) and emit STTOut messages (PartialTranscription or Transcription).

from speech_to_speech.STT.base_stt_handler import BaseSTTHandler
from speech_to_speech.pipeline.handler_types import STTIn, STTOut

class MySTTHandler(BaseSTTHandler):
    """Example custom STT backend receiving VADAudio and emitting Transcription objects."""

    def __init__(self, stop_event, queue_in, queue_out, setup_kwargs):
        super().__init__(stop_event, queue_in, queue_out, setup_kwargs)
        # Load model using configuration passed via setup_kwargs

        self.model_path = setup_kwargs.get("model_path", "default.pt")
        self.device = setup_kwargs.get("device", "cpu")

    def run(self) -> None:
        """Main processing loop called by BaseHandler.run()."""
        while not self.stop_event.is_set():
            audio: STTIn = self.queue_in.get()
            if not self.should_process_input(audio):
                continue
            
            # Replace with actual inference logic

            text = self.transcribe(audio)
            
            from speech_to_speech.pipeline.messages import Transcription
            output = Transcription(
                turn_id=audio.turn_id,
                turn_revision=audio.turn_revision,
                created_at_s=audio.created_at_s,
                text=text,
                mode="final",
            )
            
            if self.should_emit_output(output):
                self.queue_out.put(output)

    def transcribe(self, audio: STTIn) -> str:
        """Implement your model inference here."""
        return "transcribed text"

Key requirements: Respect should_process_input() to avoid processing stale audio, and use should_emit_output() before queueing results to leverage the built-in speculative-turn logic.

Step 2: Define Configuration Arguments

Create src/speech_to_speech/arguments_classes/<your_name>_stt_arguments.py to expose CLI flags for your backend. Extend ArgumentsBase so HfArgumentParser automatically generates prefixed flags like --my_stt_model_path.

from dataclasses import dataclass
from speech_to_speech.arguments_classes.module_arguments import ArgumentsBase

@dataclass
class MySTTHandlerArguments(ArgumentsBase):
    """Configuration for the custom STT backend."""
    model_path: str = "models/my_stt.pt"
    device: str = "cpu"
    language: str = "en"

Step 3: Expose the Backend in ModuleArguments

Add your identifier to the STTBackend enum in src/speech_to_speech/arguments_classes/module_arguments.py. This creates the valid CLI value --stt my_stt.

from enum import Enum

class STTBackend(str, Enum):
    WHISPER = "whisper"
    WHISPER_MLX = "whisper-mlx"
    MLX_AUDIO_WHISPER = "mlx-audio-whisper"
    FASTER_WHISPER = "faster-whisper"
    PARAKEET_TDT = "parakeet-tdt"
    PARAFORMER = "paraformer"
    MY_STT = "my_stt"  # ← Your custom backend

Step 4: Register in get_stt_handler()

Wire your handler into the dispatch logic in src/speech_to_speech/s2s_pipeline.py. Add an elif branch inside get_stt_handler() (around line 750) that imports your class and returns it wrapped with with_speculative_turns()—the same pattern used for built-in handlers.

elif module_kwargs.stt == "my_stt":
    from speech_to_speech.STT.my_stt_handler import MySTTHandler
    return with_speculative_turns(
        MySTTHandler(
            stop_event,
            queue_in=spoken_prompt_queue,
            queue_out=text_prompt_queue,
            setup_kwargs=vars(my_stt_handler_kwargs),
        )
    )

Critical: Use with_speculative_turns() to wrap your handler instance. This wrapper enables the pipeline's speculative-turn detection, preventing redundant processing when the user is still speaking.

Step 5: Update Argument Parser Ordering (Optional)

If your arguments dataclass introduces field names that clash with other backends (e.g., a generic model_path that conflicts with TTS settings), adjust the pre-parse logic in parse_arguments() within s2s_pipeline.py. The parser resolves collisions based on registration order; ensure your STT argument class is instantiated before conflicting classes.

Step 6: Write Integration Tests

Validate your integration in tests/test_my_stt_handler.py. Verify that the pipeline builds correctly with --stt my_stt and that audio flows through your handler to downstream components.

import json
import pathlib
from speech_to_speech.s2s_pipeline import parse_arguments, build_pipeline

def test_my_stt_integration(tmp_path):
    cfg = {
        "module_kwargs": {"stt": "my_stt", "tts": "melo_tts", "mode": "local"},
        "my_stt_handler_kwargs": {"model_path": str(tmp_path / "model.pt"), "device": "cpu"},
        # Include other required kwargs for TTS, VAD, etc.

    }
    
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps(cfg))
    
    # Parse and build pipeline

    args = parse_arguments()
    # Initialize required queues and events

    # queues = initialize_queues_and_events()

    # manager = build_pipeline(args.module_kwargs, ..., queues)

    
    # Assert pipeline constructed successfully and test message flow

Summary

• Inherit from BaseSTTHandler in src/speech_to_speech/STT/base_stt_handler.py to receive speculative-turn and stale-input filtering automatically.
• Create an arguments dataclass extending ArgumentsBase to expose configuration via HfArgumentParser.
• Add your identifier to the STTBackend enum in module_arguments.py to validate CLI inputs.
• Register the handler in get_stt_handler() in s2s_pipeline.py (lines 749-789) using the with_speculative_turns() wrapper.
• Adjust parser ordering only if field name collisions occur with other pipeline components.
• Test thoroughly by instantiating the pipeline with your STT identifier and verifying STTOut message flow.

Frequently Asked Questions

What interface must my custom STT handler implement?

Your handler must inherit from BaseSTTHandler and implement a run() method that reads STTIn objects from queue_in and writes STTOut objects to queue_out. The constructor must accept four parameters: stop_event, queue_in, queue_out, and setup_kwargs. According to the source code in src/speech_to_speech/STT/base_stt_handler.py, the base class provides should_process_input() and should_emit_output() methods that you should call to maintain consistency with the pipeline's filtering logic.

How does the pipeline handle configuration arguments for custom backends?

The pipeline uses HfArgumentParser to convert CLI flags or JSON configuration files into strongly-typed dataclasses. When you create a dataclass extending ArgumentsBase in src/speech_to_speech/arguments_classes/, the parser automatically generates prefixed command-line flags (e.g., --my_stt_device). These values are passed to your handler's setup_kwargs parameter during instantiation in get_stt_handler().

Can I use a completely custom base class instead of BaseSTTHandler?

Yes, you can inherit directly from BaseHandler[STTIn, STTOut] if you need to bypass the speculative-turn logic entirely. However, you will lose the automatic stale-input dropping and turn-tracking features implemented in BaseSTTHandler. To maintain pipeline stability, you must manually implement equivalent filtering logic to prevent processing audio from aborted turns.

How do I test my custom STT backend without running the full pipeline?

Unit tests should verify that your handler correctly processes mock VADAudio objects and emits Transcription messages with proper turn_id and turn_revision fields. For integration testing, use the parse_arguments() function to load a JSON configuration specifying your STT backend, then call build_pipeline() to ensure the dispatch logic in s2s_pipeline.py correctly instantiates your class without import errors. Test with a subprocess or thread to verify queue-based message passing works correctly under the pipeline's threading model.

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