Speech-to-Speech Threading Model: How VAD, STT, LLM, and TTS Communicate via Queues

The huggingface/speech-to-speech pipeline employs a producer-consumer architecture where Voice Activity Detection (VAD), Speech-to-Text (STT), Large Language Model (LLM), and Text-to-Speech (TTS) handlers run in isolated threads and exchange data exclusively through thread-safe queue.Queue objects.

The open-source huggingface/speech-to-speech repository implements a real-time, full-duplex voice assistant capable of processing live audio streams. At its core, the system relies on a multi-threaded pipeline architecture where each processing stage operates independently, communicating through synchronized queues to prevent blocking and enable speculative turn handling.

Queue and Event Infrastructure

The pipeline initialization begins in src/speech_to_speech/s2s_pipeline.py with initialize_queues_and_events() (lines 364-378), which instantiates the shared communication layer:

• stop_event: Global threading.Event for coordinated shutdown
• should_listen: Controls audio capture activation state
• recv_audio_chunks_queue: Raw audio input (AudioInItem)
• spoken_prompt_queue: VAD-filtered voice segments (VADOutItem)
• stt_output_queue: Transcription results (STTOutItem)
• text_prompt_queue: Normalized prompts for the LLM (TextPromptItem)
• lm_response_queue: Raw LLM output chunks (LMOutItem)
• lm_processed_queue: Processed text ready for synthesis (TTSInItem)
• send_audio_chunks_queue: Final audio output (AudioOutItem)

These queue.Queue instances provide thread-safe FIFO buffers between producers and consumers, eliminating race conditions without requiring explicit locks in handler code.

The Handler Chain: Queue Wiring and Data Flow

The _build_pipeline_handlers() function (lines 381-495) constructs the pipeline by injecting queue references into each handler. The architecture follows a strict linear flow where each stage consumes from one queue and produces to the next.

VAD Handler (Voice Activity Detection)

Located in src/speech_to_speech/VAD/vad_handler.py, the VADHandler monitors recv_audio_chunks_queue for incoming audio frames. When speech is detected, it emits VADAudio objects to spoken_prompt_queue. The handler also manages speculative turn tracking through SpeculativeTurnTracker, allowing the system to reopen interrupted conversations by signaling via text_output_queue when real-time transcription events occur.

STT Handler (Speech-to-Text)

All speech recognition implementations inherit from BaseSTTHandler in src/speech_to_speech/STT/base_stt_handler.py. This base class provides queue-level gating logic via should_process_input() and should_emit_output(), which automatically drops stale audio chunks when the user interrupts a turn. The handler consumes VADAudio from spoken_prompt_queue and publishes PartialTranscription or Transcription objects to stt_output_queue.

LLM Handler (Language Model)

The BaseLanguageModelHandler in src/speech_to_speech/LLM/language_model.py operates as a streaming text generator. It pulls GenerateResponseRequest objects from text_prompt_queue (populated by the TranscriptionNotifier bridge), generates tokens using either transformers or MLX backends, and yields LLMResponseChunk objects into lm_response_queue. This design allows the LLM to stream partial responses to the TTS stage before generation completes.

TTS Handler (Text-to-Speech)

Concrete TTS implementations like FacebookMMSTTSHandler in src/speech_to_speech/TTS/facebookmms_handler.py read from lm_processed_queue (populated by LMOutputProcessor). The handler synthesizes np.int16 audio chunks and pushes them to send_audio_chunks_queue. It respects cancellation scopes and speculative-turn checks, ensuring audio generation halts immediately when a new user utterance begins.

Thread Management and Execution Model

Each handler inherits from BaseHandler, which implements a persistent run() loop executed by ThreadManager from utils/thread_manager.py. When pipeline_manager.start() is invoked, the manager spawns one thread per handler. Each thread executes a blocking loop:

1. Poll queue_in.get() for incoming data
2. Execute process(item) for business logic
3. Put results into queue_out.put()

This architecture decouples I/O-bound operations (audio streaming) from CPU-bound operations (LLM inference), maximizing throughput across heterogeneous hardware.

Real-Time Concurrency and Speculative Turns

For production deployments handling multiple concurrent sessions, _build_realtime_pipeline_unit() creates isolated pipeline instances—each with dedicated queues and events—managed by thread pools. The speculative turn mechanism, implemented in src/speech_to_speech/pipeline/speculative_turns.py, coordinates across threads to detect barge-in events (user interruptions), allowing VAD to signal STT and LLM to cancel in-flight operations via queue poisoning and event flags.

Minimal Pipeline Implementation

The following example demonstrates programmatic construction of the threaded pipeline:

from speech_to_speech.s2s_pipeline import (
    parse_arguments,
    prepare_all_args,
    initialize_queues_and_events,
    build_pipeline,
)

# Initialize configuration

args = parse_arguments()
prepare_all_args(
    args.module_kwargs,
    args.whisper_stt_handler_kwargs,
    args.language_model_handler_kwargs,
    args.facebook_mms_tts_handler_kwargs,
    # ... other handler kwargs

)

# Create thread-safe queues and events

queues = initialize_queues_and_events()

# Build the pipeline with queue injections

pipeline_manager = build_pipeline(
    args.module_kwargs,
    args.socket_receiver_kwargs,
    args.socket_sender_kwargs,
    # ... handler configurations

    queues,
)

# Start all handler threads

pipeline_manager.start()
pipeline_manager.wait()  # Blocks until stop_event is set

This pattern mirrors the main() entry point in s2s_pipeline.py while exposing the queue initialization for custom integrations.

Summary

• Queue-Driven Architecture: All inter-handler communication uses queue.Queue objects to ensure thread safety without explicit locking.
• Isolated Threads: Each pipeline stage (VAD, STT, LLM, TTS) runs in a dedicated thread managed by ThreadManager, preventing I/O blocking.
• Speculative Turn Support: The queue-based design enables rapid cancellation and turn reopening when users interrupt the assistant.
• Base Handler Abstraction: BaseSTTHandler and BaseLanguageModelHandler provide standardized queue gating, stale-turn filtering, and cancellation logic across all implementations.
• Scalable Real-Time Mode: The architecture supports multiple isolated pipeline units for concurrent WebSocket sessions, each with independent queue sets.

Frequently Asked Questions

How does the speech-to-speech pipeline handle thread safety between components?

The pipeline relies exclusively on Python's queue.Queue class, which implements thread-safe FIFO operations using internal mutexes and condition variables. Handlers never share state directly; they communicate by putting typed message objects (like VADAudio or LLMResponseChunk) into queues and getting them from the next stage. This eliminates race conditions and simplifies concurrency management.

What happens when a user interrupts the assistant mid-sentence?

The speculative turn tracking system detects new speech via VAD and propagates cancellation signals through the queue chain. The BaseSTTHandler drops stale audio chunks using should_process_input(), while TTS handlers check cancel_scope before generating audio. This ensures the pipeline immediately pivots to processing the new utterance rather than completing the outdated response.

Can the queue sizes be configured to prevent memory issues?

While the provided analysis focuses on the default implementation, queue.Queue accepts a maxsize parameter during instantiation in initialize_queues_and_events(). When queues reach capacity, put() operations block until space is available, providing natural backpressure to prevent unbounded memory growth during high-latency LLM inference or slow TTS synthesis.

How are multiple concurrent sessions handled in production deployments?

For real-time server deployments, the repository uses _build_realtime_pipeline_unit() to create isolated pipeline instances—each containing its own set of queues, events, and handlers. ThreadManager maintains separate thread pools for each session, ensuring that blocking operations in one conversation do not impact others. This architecture supports horizontal scaling across WebSocket connections.

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