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

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:

VAD → STT → TranscriptionNotifier → LLM → TTS

The TranscriptionNotifier is a thin BaseHandler implementation located in 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 (lines 404-414), the pipeline builder instantiates the notifier explicitly:

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:

# 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 imports TranscriptionNotifier by name, you can replace the default implementation using monkey-patching before the pipeline builder executes:

# 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 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 – Reference implementation of the default notifier (lines 33-73).
• src/speech_to_speech/s2s_pipeline.py – Pipeline builder where the notifier is instantiated (lines 404-414).
• src/speech_to_speech/baseHandler.py – Abstract base class that all pipeline components extend.
• src/speech_to_speech/pipeline/events.py – Event definitions including PartialTranscriptionEvent and TranscriptionCompletedEvent.
• 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 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 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.

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