# How to Implement Custom Transcription Notifiers for Downstream Processing in Speech-to-Speech

> Learn to implement custom transcription notifiers for downstream processing in Hugging Face speech-to-speech. Subclass BaseHandler and override process for seamless integration.

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

---

**To implement custom transcription notifiers for downstream processing in the Hugging Face speech-to-speech pipeline, subclass `BaseHandler[STTOut, LLMIn]`, implement the `process` method to handle `PartialTranscription` and `Transcription` objects, and replace the default `TranscriptionNotifier` class before the pipeline is built.**

The Hugging Face `speech-to-speech` repository provides a modular, real-time voice conversation pipeline that routes Speech-to-Text (STT) output through a `TranscriptionNotifier` before reaching the Language Model (LLM). By creating **custom transcription notifiers**, you can intercept transcripts to perform auditing, filtering, or message broker integration without modifying the core STT or LLM handlers.

## Understanding the TranscriptionNotifier Architecture

The pipeline processes audio through a linear sequence of handlers:

```text
VAD → STT → TranscriptionNotifier → LLM → TTS

```

The `TranscriptionNotifier` is a thin `BaseHandler` implementation located in [`src/speech_to_speech/STT/transcription_notifier.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/STT/transcription_notifier.py) (lines 33-73). It converts each partial or final transcription into protocol-neutral events and pushes them onto a `text_output_queue` for downstream consumers.

In [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) (lines 404-414), the pipeline builder instantiates the notifier explicitly:

```python
transcription_notifier = TranscriptionNotifier(
    stop_event,
    queue_in=stt_output_queue,
    queue_out=text_prompt_queue,
    setup_kwargs={"text_output_queue": text_output_queue,
                  "should_listen": should_listen},
)

```

Because the builder references the class name directly, you can inject custom behavior by **monkey-patching** `TranscriptionNotifier` with your own subclass before the pipeline initializes.

## Creating a Custom TranscriptionNotifier

To build a custom notifier that archives every final transcript to disk, follow these implementation steps:

1. **Inherit from `BaseHandler[STTOut, LLMIn]`** – This establishes the correct type signatures for input and output queues.
2. **Implement `setup`** – Receive configuration including `text_output_queue`, `should_listen`, and any custom parameters.
3. **Implement `process`** – Inspect incoming messages for `PartialTranscription` and `Transcription` types, emit appropriate events, and optionally forward data downstream.
4. **Handle event emission** – Use `PartialTranscriptionEvent` and `TranscriptionCompletedEvent` from the events module to maintain compatibility with existing consumers.

Here is a complete implementation that archives final transcripts while preserving the default event flow:

```python

# my_pkg/custom_notifier.py

from __future__ import annotations
import logging
from queue import Queue
from threading import Event
from typing import Iterator

from speech_to_speech.baseHandler import BaseHandler
from speech_to_speech.pipeline.events import (
    PartialTranscriptionEvent,
    TranscriptionCompletedEvent,
)
from speech_to_speech.pipeline.handler_types import LLMIn, STTOut
from speech_to_speech.pipeline.messages import PartialTranscription, Transcription

logger = logging.getLogger(__name__)

class ArchiveNotifier(BaseHandler[STTOut, LLMIn]):
    """Custom notifier that writes final transcripts to a file for audit purposes."""

    def setup(
        self,
        text_output_queue: Queue | None = None,
        should_listen: Event | None = None,
        archive_path: str = "transcripts.log",
    ) -> None:
        self.text_output_queue = text_output_queue
        self.should_listen = should_listen
        self.archive_path = archive_path

    def _archive(self, transcript: str) -> None:
        with open(self.archive_path, "a", encoding="utf-8") as f:
            f.write(transcript + "\n")
        logger.info("Archived transcript (%d chars)", len(transcript))

    def process(self, transcription: STTOut) -> Iterator[LLMIn]:
        # Forward partial updates unchanged (optional)

        if isinstance(transcription, PartialTranscription):
            if self.text_output_queue and transcription.text:
                self.text_output_queue.put(
                    PartialTranscriptionEvent(
                        delta=str(transcription.text),
                        turn_id=transcription.turn_id,
                        turn_revision=transcription.turn_revision,
                    )
                )
            return

        # Final transcription – emit event and archive

        if isinstance(transcription, Transcription):
            text = transcription.text or ""
            self._archive(text)

            if self.text_output_queue is not None:
                self.text_output_queue.put(
                    TranscriptionCompletedEvent(
                        transcript=text,
                        language_code=transcription.language_code,
                        turn_id=transcription.turn_id,
                        turn_revision=transcription.turn_revision,
                        speech_stopped_at_s=transcription.speech_stopped_at_s,
                    )
                )
        else:
            # Back-compatibility: raw string output

            text = str(transcription)
            self._archive(text)

        # No downstream LLM payload – the LLM will be fed from the

        # upstream `text_prompt_queue`, exactly like the built-in notifier.

        return

```

## Wiring the Custom Notifier into the Pipeline

Because [`s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/s2s_pipeline.py) imports `TranscriptionNotifier` by name, you can replace the default implementation using monkey-patching before the pipeline builder executes:

```python

# my_pkg/pipeline_builder.py

from speech_to_speech.STT.transcription_notifier import TranscriptionNotifier
from my_pkg.custom_notifier import ArchiveNotifier

# Replace the default class before the pipeline is built

TranscriptionNotifier = ArchiveNotifier

# Now import and build the pipeline normally

from speech_to_speech.s2s_pipeline import build_pipeline
pipeline = build_pipeline(...)

```

Alternatively, you can modify [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) directly to instantiate your custom class instead of the default `TranscriptionNotifier`. Either approach ensures that every final transcription is archived to `transcripts.log` while the same events continue to flow downstream to the LLM and other consumers.

## Key Source Files

- **[`src/speech_to_speech/STT/transcription_notifier.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/STT/transcription_notifier.py)** – Reference implementation of the default notifier (lines 33-73).
- **[`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py)** – Pipeline builder where the notifier is instantiated (lines 404-414).
- **[`src/speech_to_speech/baseHandler.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/baseHandler.py)** – Abstract base class that all pipeline components extend.
- **[`src/speech_to_speech/pipeline/events.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/pipeline/events.py)** – Event definitions including `PartialTranscriptionEvent` and `TranscriptionCompletedEvent`.
- **[`src/speech_to_speech/pipeline/messages.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/pipeline/messages.py)** – Message structures for `PartialTranscription` and `Transcription`.

## Summary

- **Subclass `BaseHandler[STTOut, LLMIn]`** to create a custom transcription notifier that intercepts STT output before it reaches the LLM.
- **Implement `setup`** to receive the `text_output_queue` and `should_listen` event, then **implement `process`** to handle `PartialTranscription` and `Transcription` message types.
- **Emit events** using `PartialTranscriptionEvent` and `TranscriptionCompletedEvent` to maintain compatibility with existing downstream consumers.
- **Monkey-patch `TranscriptionNotifier`** before pipeline construction or modify the builder in [`s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/s2s_pipeline.py) to inject your custom logic.
- **Return `Iterator[LLMIn]`** from `process` to maintain type consistency with the handler interface, even when only emitting side effects.

## Frequently Asked Questions

### What is the difference between PartialTranscription and Transcription?

`PartialTranscription` represents interim speech-to-text results while the user is still speaking, containing incremental text updates. `Transcription` represents the final, committed transcript after speech has stopped. The built-in notifier emits `PartialTranscriptionEvent` for the former and `TranscriptionCompletedEvent` for the latter, allowing downstream consumers to distinguish between live previews and final results.

### Can I use multiple custom notifiers in the same pipeline?

The standard pipeline builder only instantiates a single notifier instance. To use multiple notifiers, create a composite handler that internally chains several processing steps, or modify the pipeline builder in [`s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/s2s_pipeline.py) to insert additional handlers between the STT and LLM stages. Each handler can write to the same `text_output_queue` or maintain separate queues for different consumers.

### How do I access the text_output_queue from my custom notifier?

The `text_output_queue` is passed through the `setup_kwargs` dictionary during pipeline construction. In your `setup` method, accept it as a parameter (type `Queue | None`) and store it as an instance attribute. This queue is the same object used by the RealtimeService and other downstream consumers, ensuring your events reach all attached listeners.

### Will custom notifiers affect the latency of the speech-to-speech pipeline?

The `process` method runs in the same thread as the STT handler, so blocking operations like disk I/O or network calls will introduce latency. For performance-critical applications, offload heavy processing to a separate thread or asynchronous task within your `process` method, or use the `text_output_queue` to pass events to a dedicated consumer thread without blocking the main pipeline flow.