How to Implement a Custom TTS Backend Handler in the Hugging Face Speech-to-Speech Pipeline

To implement a custom TTS backend handler in the speech-to-speech repository, create a class inheriting from BaseHandler[TTSIn, TTSOut], implement the setup method for model initialization and the process generator for streaming audio output, then register it in s2s_pipeline.py with a matching arguments class.

The speech-to-speech pipeline from Hugging Face provides a modular architecture where each processing stage—speech-to-text, language model, and text-to-speech—runs as an independent handler. This design lets you swap TTS backends without modifying core pipeline logic. Whether you want to integrate a proprietary model, a lightweight on-device synthesizer, or an experimental architecture, the handler abstraction gives you full control over model loading, inference, and audio streaming.

Understanding the Handler Architecture

Every pipeline stage inherits from BaseHandler in src/speech_to_speech/baseHandler.py. This base class manages:

• Inter-handler communication via thread-safe queues
• Cancellation and interruption through CancelScope
• Performance timing and debug logging
• Cleanup and graceful shutdown

TTS handlers specifically work with two typed containers defined in src/speech_to_speech/pipeline/handler_types.py:

• TTSIn: Carries text to synthesize, language code, turn ID, and revision metadata for speculative turn handling
• TTSOut: Typically bytes or np.ndarray audio chunks, plus control messages from src/speech_to_speech/pipeline/messages.py

The two key methods you must implement are:

Method	Purpose	When Called
setup	Load model weights, configure device, initialize resources	Once at handler start
process	Generate audio chunks from input text	Called repeatedly in a loop

Reference implementations in src/speech_to_speech/TTS/qwen3_tts_handler.py and src/speech_to_speech/TTS/pocket_tts_handler.py demonstrate streaming inference, resampling, and proper cancellation handling.

Step 1: Create Your Handler Class

Create a new file in src/speech_to_speech/TTS/my_tts_handler.py. The following skeleton implements the complete contract while marking integration points for your specific model:

# src/speech_to_speech/TTS/my_tts_handler.py

from __future__ import annotations

import logging
from threading import Event
from time import perf_counter
from typing import Any, Iterator

import numpy as np
from rich.console import Console

from speech_to_speech.baseHandler import BaseHandler
from speech_to_speech.pipeline.cancel_scope import CancelScope
from speech_to_speech.pipeline.handler_types import TTSIn, TTSOut
from speech_to_speech.pipeline.messages import AUDIO_RESPONSE_DONE, EndOfResponse
from speech_to_speech.pipeline.speculative_turns import SpeculativeTurnTracker

logger = logging.getLogger(__name__)
console = Console()


class MyTTSHandler(BaseHandler[TTSIn, TTSOut]):
    """
    Custom TTS backend handler implementing the speech-to-speech pipeline contract.
    """

    def setup(
        self,
        should_listen: Event,
        *,
        device: str = "cpu",
        model_path: str | None = None,
        blocksize: int = 512,
        max_new_tokens: int = 100,
        gen_kwargs: dict[str, Any] | None = None,
        cancel_scope: CancelScope | None = None,
        speculative_turns: SpeculativeTurnTracker | None = None,
    ) -> None:
        """Initialize model and store configuration."""
        self.should_listen = should_listen
        self.cancel_scope = cancel_scope
        self.speculative_turns = speculative_turns
        self.device = device
        self.blocksize = blocksize
        self.max_new_tokens = max_new_tokens
        self.gen_kwargs = gen_kwargs or {}

        # TODO: Replace with actual model loading

        # self.model = MyTTSModel.from_pretrained(model_path, device=self.device)

        logger.info(f"Loading MyTTS model on {device} (stub)")

    @property
    def min_time_to_debug(self) -> float:
        """Suppress debug logs for chunks faster than this threshold."""
        return 0.1  # 100ms

    def process(self, tts_input: TTSIn) -> Iterator[TTSOut]:
        """Generate audio from text, yielding int16 chunks."""
        # Handle end-of-response signaling

        speculative = getattr(self, "speculative_turns", None)
        if isinstance(tts_input, EndOfResponse):
            if speculative and not speculative.is_latest_after_reopen_grace(
                tts_input.turn_id, tts_input.turn_revision
            ):
                return
            yield AUDIO_RESPONSE_DONE
            return

        # Filter stale speculative turns

        if speculative and not speculative.is_latest_after_reopen_grace(
            tts_input.turn_id, tts_input.turn_revision
        ):
            logger.debug("Dropping stale TTS input turn=%s", tts_input.turn_id)
            return
        
        if speculative:
            speculative.commit(tts_input.turn_id, tts_input.turn_revision)

        # TODO: Implement text coalescing for batched generation

        text = tts_input.text or ""
        console.print(f"[green]ASSISTANT: {text}")

        # Streaming generation loop with cancellation support

        start = perf_counter()
        first_chunk = True

        # TODO: Replace with actual model streaming generator

        # for audio_float in self.model.generate_stream(text, **self.gen_kwargs):

        for _ in range(1):  # Stub: single silence block

            # Check for cancellation

            if self.cancel_scope:
                gen = self.cancel_scope.generation
                if gen is not None and self.cancel_scope.is_stale(gen):
                    logger.info("TTS generation cancelled (interruption)")
                    return

            if first_chunk:
                logger.debug(f"Time to first audio: {perf_counter() - start:.3f}s")
                first_chunk = False

            # Convert float32 [-1, 1] to int16 for pipeline compatibility

            dummy_float = np.zeros(self.blocksize, dtype=np.float32)
            audio_int16 = (dummy_float * 32768).astype(np.int16)
            yield audio_int16

Key integration points marked with TODO:

• Model loading: Replace the stub with your model's initialization logic
• Text coalescing: For low-latency streaming, accumulate short text segments before generation (see qwen3_tts_handler.py for the coalesce_inputs pattern)
• Streaming generator: Your model should yield audio chunks incrementally rather than buffering complete utterances

Step 2: Create an Arguments Class

The pipeline uses dataclasses for clean CLI integration. Create src/speech_to_speech/arguments_classes/my_tts_arguments.py:

# src/speech_to_speech/arguments_classes/my_tts_arguments.py

from dataclasses import dataclass, field


@dataclass
class MyTTSHandlerArguments:
    device: str = "cpu"
    model_path: str | None = None
    blocksize: int = 512
    max_new_tokens: int = 100
    gen_kwargs: dict = field(default_factory=dict)

This mirrors the pattern in qwen3_tts_arguments.py and pocket_tts_arguments.py, enabling automatic CLI argument generation and type validation.

Step 3: Register in the Pipeline

Modify src/speech_to_speech/s2s_pipeline.py at three locations:

Import your arguments class (around line 42, with other handler arguments):

from speech_to_speech.arguments_classes.my_tts_arguments import MyTTSHandlerArguments

Add to the pipeline arguments dataclass (around line 105):

@dataclass
class SpeechToSpeechArgs:
    # ... existing handler arguments ...

    my_tts_handler_kwargs: MyTTSHandlerArguments = MyTTSHandlerArguments()

Instantiate and wire your handler in create_pipeline (around line 201):

from speech_to_speech.TTS.my_tts_handler import MyTTSHandler

# Inside create_pipeline method:

my_tts = MyTTSHandler(
    stop_event,
    queue_in=tts_queue_in,
    queue_out=tts_queue_out,
    setup_args=(should_listen,),
    setup_kwargs=self.args.my_tts_handler_kwargs.__dict__,
)
pipeline.append(my_tts)

The queue_in and queue_out connections should match the existing TTS handler placement—typically receiving from the LLM handler and sending to the audio playback handler.

Step 4: Write Tests

Create tests/test_my_tts_handler_backend.py following the patterns in tests/test_qwen3_tts_handler_backend.py:

# tests/test_my_tts_handler_backend.py

import pytest
import numpy as np
from threading import Event

from speech_to_speech.TTS.my_tts_handler import MyTTSHandler
from speech_to_speech.pipeline.handler_types import TTSIn
from speech_to_speech.pipeline.messages import AUDIO_RESPONSE_DONE, EndOfResponse
from speech_to_speech.pipeline.cancel_scope import CancelScope


@pytest.fixture
def handler():
    stop_event = Event()
    h = MyTTSHandler(stop_event, queue_in=None, queue_out=None)
    # Mock setup to avoid heavy model loading

    h.should_listen = Event()
    h.cancel_scope = None
    h.speculative_turns = None
    h.blocksize = 512
    return h


def test_process_yields_int16_chunks(handler):
    """Verify output format matches pipeline expectations."""
    tts_input = TTSIn(text="Hello", turn_id=1, turn_revision=0)
    
    chunks = list(handler.process(tts_input))
    
    assert len(chunks) > 0
    assert all(isinstance(c, np.ndarray) for c in chunks if c is not AUDIO_RESPONSE_DONE)
    assert all(c.dtype == np.int16 for c in chunks if isinstance(c, np.ndarray))


def test_end_of_response_emits_done(handler):
    """Verify proper end-of-stream signaling."""
    end_msg = EndOfResponse(turn_id=1, turn_revision=0)
    
    results = list(handler.process(end_msg))
    
    assert AUDIO_RESPONSE_DONE in results


def test_respects_cancellation(handler):
    """Verify generator stops when cancel_scope marks generation stale."""
    handler.cancel_scope = CancelScope()
    handler.cancel_scope.start_generation()
    # Simulate stale generation

    handler.cancel_scope.generation = 0
    handler.cancel_scope._latest_generation = 1  # Higher = newer

    tts_input = TTSIn(text="Long text", turn_id=1, turn_revision=0)
    chunks = list(handler.process(tts_input))
    
    # Should yield nothing or stop early when cancellation detected

    # Exact behavior depends on your implementation's cancellation checks

Mock heavy dependencies with monkeypatch to keep tests fast and deterministic.

Running Your Custom Handler

Execute the pipeline with your new backend:

# scripts/run_with_my_tts.py

from speech_to_speech.s2s_pipeline import SpeechToSpeechArgs, SpeechToSpeechPipeline
from speech_to_speech.arguments_classes.my_tts_arguments import MyTTSHandlerArguments

if __name__ == "__main__":
    args = SpeechToSpeechArgs(
        my_tts_handler_kwargs=MyTTSHandlerArguments(
            device="cuda",
            model_path="checkpoints/my-tts-v1",
            blocksize=1024,
            gen_kwargs={"temperature": 0.8, "top_p": 0.95},
        )
    )
    pipeline = SpeechToSpeechPipeline(args)
    pipeline.run()

Common Implementation Patterns

When building production-ready TTS handlers, consider these patterns from the reference implementations:

Pattern	Source	Purpose
Input coalescing	qwen3_tts_handler.py	Accumulate rapid-fire text chunks to reduce synthesis calls
Resampling	pocket_tts_handler.py	Match pipeline's expected sample rate (typically 16kHz or 24kHz)
Language tags	qwen3_tts_handler.py	Pass tts_input.language to multilingual models
Turn revision tracking	Both handlers	Handle mid-utterance interruptions and corrections

The speculative turn tracker enables the pipeline to discard TTS output when the LLM revises its response. Always check is_latest_after_reopen_grace() before yielding audio.

Summary

• Inherit from BaseHandler[TTSIn, TTSOut] and implement setup and process to create a TTS backend handler
• Use typed containers from handler_types.py and respect control messages from messages.py
• Implement cancellation checks via CancelScope for responsive interruption handling
• Create matching arguments classes for clean CLI and programmatic configuration
• Register in s2s_pipeline.py by adding imports, dataclass fields, and handler instantiation
• Write unit tests mocking model loading and verifying cancellation, format compliance, and edge cases

Frequently Asked Questions

What audio format should my TTS handler output?

The speech-to-speech pipeline expects 16-bit integer PCM audio (np.int16) as the standard format. Your process generator should yield np.ndarray objects with dtype=np.int16 and samples in the range -32768 to 32767. If your model produces float32 output in the [-1, 1] range, convert using (audio_float * 32768).astype(np.int16) as shown in the implementation example.

How do I handle interruptions and cancellation?

Check self.cancel_scope.is_stale() within your streaming loop. The CancelScope object tracks generation epochs; when the user interrupts or the LLM produces new output, the scope marks older generations stale. Exit your generator immediately when this occurs to free compute resources and prevent audio artifacts. See src/speech_to_speech/pipeline/cancel_scope.py for the full API.

Can I batch multiple text inputs for efficiency?

Yes, implement input coalescing by accumulating TTSIn objects briefly before synthesis. The qwen3_tts_handler.py implementation shows this pattern: buffer inputs with short timeouts, concatenate text segments, and generate once. Balance latency against throughput based on your use case—real-time conversational applications typically use 50-150ms coalescing windows.

Where should I configure model-specific hyperparameters?

Add fields to your arguments dataclass in src/speech_to_speech/arguments_classes/, then access them via setup_kwargs in the handler's setup method. This keeps configuration declarative and enables CLI overrides. For runtime-adjustable parameters (like voice ID or speaking rate), consider adding them to TTSIn and reading from tts_input in process.

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