How to Handle Real-Time Speech-to-Speech Processing with the Hugging Face speech-to-speech Library

Real-time speech-to-speech processing in the huggingface/speech-to-speech repository relies on a modular, streaming pipeline that continuously captures microphone audio, detects speech activity, transcribes it, generates responses with a language model, and synthesizes replies with minimal latency.

The speech-to-speech library implements this as an async event-driven system where audio flows through dedicated handlers for voice activity detection (VAD), speech-to-text (STT), large language model (LLM) inference, and text-to-speech (TTS). Each component processes data in small chunks—typically ~20 ms—enabling sub-second round-trip conversation flow.

Core Pipeline Architecture

The repository organizes real-time S2S into six interconnected stages. Understanding this data flow is essential for customization and debugging.

Audio Capture and Streaming

The entry point is scripts/listen_and_play_realtime.py, which runs the audio streamer in a dedicated background thread.

# From scripts/listen_and_play_realtime.py – WebSocket audio producer

async def audio_producer(ws):
    def callback(indata, frames, time, status):
        wav = (indata[:, 0] * 32767).astype(np.int16).tobytes()
        asyncio.run_coroutine_threadsafe(ws.send(wav), loop)

    with sd.InputStream(
        samplerate=16000,
        channels=1,
        dtype="float32",
        blocksize=320,  # 20 ms at 16 kHz

        callback=callback
    ):
        await asyncio.Future()

The streamer yields raw PCM chunks without waiting for complete utterances, eliminating the primary source of latency in traditional batch systems.

Voice Activity Detection (VAD)

VAD segmentation lives in src/speech_to_speech/VAD/vad_handler.py and vad_iterator.py. The implementation uses frame-wise classification to mark speech boundaries:

• Speech start: Confidence exceeds voice_activity_threshold (default 0.5)
• Turn closure: Silence persists beyond max_silence_ms (configurable, typically 600–800 ms)

The VadIterator class manages turn state and buffers audio segments for downstream processing.

from speech_to_speech.arguments_classes.vad_arguments import VadArguments

vad_args = VadArguments(
    model_name="silero-vad",
    voice_activity_threshold=0.6,
    min_speech_ms=150,    # Filter out coughs and noise bursts

    max_silence_ms=600,   # End turn after 600 ms silence

    vad_padding_ms=200,   # Keep 200 ms buffer around speech

)

Speech-to-Text Transcription

The STT handler receives complete VAD segments and streams transcripts back to the pipeline. Configuration happens through argument classes like WhisperSttArguments in src/speech_to_speech/arguments_classes/whisper_stt_arguments.py:

from speech_to_speech.arguments_classes.whisper_stt_arguments import WhisperSttArguments

stt_args = WhisperSttArguments(
    model_name="openai/whisper-medium",
    language="en",
    task="transcribe",
    use_fast=True,
)

The handler processes audio as soon as VAD signals turn completion—no waiting for additional silence or manual endpoint detection.

Language Model Inference

The LLM stage generates textual responses using classes defined in src/speech_to_speech/arguments_classes/language_model_arguments.py. For voice-specific prompt formatting, the library provides src/speech_to_speech/LLM/voice_prompt.py.

Text-to-Speech Synthesis

Multiple TTS backends are available in src/speech_to_speech/TTS/:

Handler	Model	Streaming Support
qwen3_tts_handler.py	Qwen/Qwen3-Chat-tts	Yes
kokoro_handler.py	Kokoro	Yes
chatTTS_handler.py	ChatTTS	Yes
pocket_tts_handler.py	Piper/Local	Yes

All handlers implement the same interface: they receive text chunks and yield audio chunks for immediate playback.

Orchestration with SpeechToSpeechPipeline

The SpeechToSpeechPipeline class in src/speech_to_speech/s2s_pipeline.py wires everything together:

from speech_to_speech.s2s_pipeline import SpeechToSpeechPipeline
from speech_to_speech.arguments_classes import (
    WebsocketStreamerArguments,
    VadArguments,
    WhisperSttArguments,
    LanguageModelArguments,
    Qwen3TtsArguments,
)

pipeline = SpeechToSpeechPipeline(
    streamer_args=WebsocketStreamerArguments(
        host="0.0.0.0", port=8765, sample_rate=16000
    ),
    vad_args=VadArguments(
        model_name="silero-vad",
        vad_padding_ms=200,
        silence_threshold_ms=800,
    ),
    stt_args=WhisperSttArguments(
        model_name="openai/whisper-medium",
        language="en",
    ),
    llm_args=LanguageModelArguments(
        model_name="meta-llama/Meta-Llama-3.1-8B-Instruct",
        max_new_tokens=256,
        temperature=0.7,
    ),
    tts_args=Qwen3TtsArguments(
        model_name="Qwen/Qwen3-Chat-tts",
        voice="female",
    ),
)

# Blocks, running the real-time event loop

pipeline.run()

The run() method creates an asyncio event loop that:

1. Launches the audio streamer via src/speech_to_speech/utils/thread_manager.py
2. Feeds frames to the VAD iterator
3. Queues completed turns for STT processing
4. Streams transcripts to the LLM
5. Pipes LLM output to TTS
6. Returns synthesized audio to the client

All inter-component communication uses typed messages from src/speech_to_speech/pipeline/messages.py, ensuring type safety and clean extension points.

Latency Optimization Techniques

Speculative Turn Generation

The src/speech_to_speech/pipeline/speculative_turns.py module reduces perceived latency by running a fast "draft" model in parallel with the main LLM:

from speech_to_speech.pipeline.speculative_turns import SpeculativeTurnManager

spec_manager = SpeculativeTurnManager(
    fast_llm_name="meta-llama/Meta-Llama-3.1-8B-Instruct",
    slow_llm_name="meta-llama/Meta-Llama-3.1-70B-Instruct",
    max_speculative_tokens=32,
)

# Returns provisional response immediately, swaps when slow model finishes

response = await spec_manager.generate(conversation_history)

Cancel Scopes for Interruption Handling

src/speech_to_speech/pipeline/cancel_scope.py implements cooperative cancellation. When new speech is detected mid-generation, in-flight STT/LLM/TTS operations abort promptly, keeping the conversation responsive.

Back-Pressured Queues

src/speech_to_speech/pipeline/queue_types.py uses asyncio.Queue with size limits to prevent memory accumulation when inference lags behind real-time audio.

Complete Client-Server Example

Server (runs the full pipeline)

# server.py – deploys the SpeechToSpeechPipeline as WebSocket endpoint

from speech_to_speech.s2s_pipeline import SpeechToSpeechPipeline
from speech_to_speech.arguments_classes import (
    WebsocketStreamerArguments,
    VadArguments,
    WhisperSttArguments,
    LanguageModelArguments,
    Qwen3TtsArguments,
)

pipeline = SpeechToSpeechPipeline(
    streamer_args=WebsocketStreamerArguments(host="0.0.0.0", port=8765),
    vad_args=VadArguments(model_name="silero-vad"),
    stt_args=WhisperSttArguments(model_name="openai/whisper-base"),
    llm_args=LanguageModelArguments(model_name="HuggingFaceTB/SmolLM-1.7B-Instruct"),
    tts_args=Qwen3TtsArguments(model_name="Qwen/Qwen3-Chat-tts"),
)

if __name__ == "__main__":
    pipeline.run()

Client (streams microphone, plays responses)

# client.py – pairs with scripts/listen_and_play_realtime.py

import asyncio
import websockets
import sounddevice as sd
import numpy as np

SERVER = "ws://localhost:8765"
SAMPLE_RATE = 16000
CHUNK_MS = 20
CHUNK_SIZE = int(SAMPLE_RATE * CHUNK_MS / 1000)


async def stream_microphone(ws):
    """Send 20ms PCM chunks to server."""
    def callback(indata, frames, time, status):
        pcm = (indata[:, 0] * 32767).astype(np.int16).tobytes()
        asyncio.run_coroutine_threadsafe(ws.send(pcm), loop)

    with sd.InputStream(
        samplerate=SAMPLE_RATE,
        channels=1,
        dtype="float32",
        blocksize=CHUNK_SIZE,
        callback=callback
    ):
        await asyncio.Future()


async def play_responses(ws):
    """Receive and play TTS audio from server."""
    async for message in ws:
        audio = np.frombuffer(message, dtype=np.int16)
        audio = audio.astype(np.float32) / 32767
        sd.play(audio, SAMPLE_RATE, blocking=True)


async def main():
    async with websockets.connect(SERVER) as ws:
        await asyncio.gather(
            stream_microphone(ws),
            play_responses(ws),
        )


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Run server and client in separate terminals for a complete real-time speech-to-speech conversation.

Key Implementation Files

File	Purpose
src/speech_to_speech/s2s_pipeline.py	Central orchestrator
src/speech_to_speech/pipeline/messages.py	Typed inter-component messages
src/speech_to_speech/pipeline/speculative_turns.py	Low-latency speculative generation
src/speech_to_speech/pipeline/cancel_scope.py	Cooperative cancellation for interruptions
src/speech_to_speech/VAD/vad_handler.py	VAD processing logic
src/speech_to_speech/VAD/vad_iterator.py	Turn segmentation state machine
scripts/listen_and_play_realtime.py	Reference WebSocket client implementation
demo/server.py	Full demonstration server

Summary

• Real-time speech-to-speech processing in huggingface/speech-to-speech uses chunked streaming (≈20 ms) across VAD, STT, LLM, and TTS stages
• The SpeechToSpeechPipeline in s2s_pipeline.py orchestrates components via async message passing
• Speculative turns and cancel scopes cut perceived latency and handle user interruptions gracefully
• Multiple TTS backends (qwen3_tts_handler.py, kokoro_handler.py, etc.) support model flexibility without pipeline changes
• Client-server deployment uses WebSocket transport with raw PCM for minimal overhead

Frequently Asked Questions

What hardware is required for real-time speech-to-speech processing?

A modern GPU with 16GB+ VRAM handles the full pipeline smoothly. CPU-only operation is possible with smaller models (SmolLM-1.7B, Whisper Base, Piper TTS) but increases latency. The client requires only a microphone, speakers, and network connectivity.

How does the pipeline handle overlapping speech or interruptions?

The cancel scope mechanism in pipeline/cancel_scope.py monitors for new VAD activity. When detected, it aborts pending STT, LLM, and TTS operations, clears the queue, and begins processing the new turn immediately.

Can I use custom STT or TTS models not included in the repository?

Yes. Implement the handler interface defined in pipeline/handler_types.py and pass your custom arguments class to SpeechToSpeechPipeline. The message-based architecture in pipeline/messages.py ensures compatibility without modifying core pipeline code.

What latency should I expect end-to-end?

With appropriate hardware and speculative turns enabled, typical round-trip latency is 300–800 ms from speech end to audio playback start. Actual performance depends on model sizes, network conditions, and VAD configuration.

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