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

> Learn to implement a custom TTS backend handler in Hugging Face's speech-to-speech pipeline. Extend BaseHandler, define setup and process, and register your handler for custom TTS solutions.

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

---

**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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/TTS/qwen3_tts_handler.py) and [`src/speech_to_speech/TTS/pocket_tts_handler.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/TTS/my_tts_handler.py). The following skeleton implements the complete contract while marking integration points for your specific model:

```python

# 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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/arguments_classes/my_tts_arguments.py):

```python

# 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`](https://github.com/huggingface/speech-to-speech/blob/main/qwen3_tts_arguments.py) and [`pocket_tts_arguments.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) at three locations:

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

```python
from speech_to_speech.arguments_classes.my_tts_arguments import MyTTSHandlerArguments

```

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

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

    my_tts_handler_kwargs: MyTTSHandlerArguments = MyTTSHandlerArguments()

```

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

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/tests/test_my_tts_handler_backend.py) following the patterns in [`tests/test_qwen3_tts_handler_backend.py`](https://github.com/huggingface/speech-to-speech/blob/main/tests/test_qwen3_tts_handler_backend.py):

```python

# 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:

```python

# 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`](https://github.com/huggingface/speech-to-speech/blob/main/qwen3_tts_handler.py) | Accumulate rapid-fire text chunks to reduce synthesis calls |
| **Resampling** | [`pocket_tts_handler.py`](https://github.com/huggingface/speech-to-speech/blob/main/pocket_tts_handler.py) | Match pipeline's expected sample rate (typically 16kHz or 24kHz) |
| **Language tags** | [`qwen3_tts_handler.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/handler_types.py) and respect control messages from [`messages.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`.