How the OpenAI Realtime API Integrates with the Speech‑to‑Speech Pipeline: Architecture and Code Guide

The OpenAI Realtime API integrates with the Hugging Face speech‑to‑speech pipeline through a WebSocket‑based bridge that converts OpenAI events into internal pipeline handlers, using a shared RuntimeConfig state object to synchronize VAD, STT, LLM, and TTS components across the audio‑in → transcribe → generate → synthesize flow.

The huggingface/speech-to-speech repository provides a production‑ready implementation that can replace or augment OpenAI's native Realtime API service. This article breaks down how the integration works at the source code level, from WebSocket handling to pipeline orchestration.

Core Integration Components

The OpenAI Realtime API integration comprises four tightly‑coupled modules that bridge external WebSocket connections with internal speech processing:

Component	Role	Source File
RuntimeConfig	Mutable session state mirroring OpenAI's RealtimeSessionCreateRequest	src/speech_to_speech/api/openai_realtime/runtime_config.py
RealtimeService	Event dispatcher parsing WebSocket events and emitting pipeline events	src/speech_to_speech/api/openai_realtime/service.py
WebSocket Router	FastAPI endpoint connecting external clients to the internal service	src/speech_to_speech/api/openai_realtime/websocket_router.py
Pipeline Builder	Constructs handler chains (VAD → STT → LLM → TTS) wired to the service	src/speech_to_speech/s2s_pipeline.py

RuntimeConfig: The Shared Session State

In runtime_config.py, the RuntimeConfig class serves as the canonical source of truth for all pipeline parameters. It mirrors OpenAI's RealtimeSessionCreateRequest schema and provides thread‑safe access to:

• Turn detection settings — turn_detection_enabled, interrupt_response_enabled
• Audio format negotiation — input/output sample rates, encoding
• Chat history — conversation items for context preservation

The GIL guarantees atomic attribute writes, eliminating the need for explicit locks when handlers read or update configuration.

RealtimeService: Event Translation Layer

The service.py module implements RealtimeService, which translates between OpenAI's WebSocket protocol and internal pipeline events. Key methods include:

• parse_client_event() — deserializes JSON events (InputAudioBufferAppend, ConversationItemCreate, SessionUpdate)
• dispatch_pipeline_event() — routes parsed data to appropriate handlers
• build_session_created() / build_session_updated() — constructs server‑side response events

This translation enables features like barge‑in interruption: when RuntimeConfig.interrupt_response_enabled is true, the VAD handler can signal the TTS handler to abort ongoing synthesis.

WebSocket Router and Pipeline Unit

The websocket_router.py exposes the /v1/realtime endpoint. Per connection, it:

1. Creates a RealtimeServer instance from a connection pool
2. Registers a unique connection ID with RealtimeService.register()
3. Instantiates a PipelineUnit via _build_realtime_pipeline_unit() in s2s_pipeline.py

Each PipelineUnit (defined in pipeline_unit.py) wraps the handler chain and maintains isolated state for concurrent connections.

Data Flow: From WebSocket to Synthesized Audio

Understanding the OpenAI Realtime API integration requires tracing how audio bytes become spoken responses:

1. Client connects to ws://host:8000/v1/realtime via the FastAPI endpoint
2. Audio stream begins — PCM chunks arrive as InputAudioBufferAppend events
3. Service parses and dispatches — RealtimeService.append_pcm() enqueues audio for VAD
4. VAD triggers STT — voice activity detection emits PartialTranscriptionEvent
5. LLM generates text — transcription prompts the chat model, yielding AssistantTextEvent
6. TTS synthesizes audio — text streams to the TTS handler, producing AudioOutput chunks
7. Server events returned — RealtimeService packages audio as AudioBufferAppend events for the client

The entire pipeline operates on streaming data structures, minimizing latency between speech input and synthesized output.

Practical Implementation Examples

Running a Local Realtime Server

Start the complete pipeline with default handlers:

# Terminal 1: Start the server

python -m speech_to_speech.server \
    --stt-handler whisper \
    --llm-handler chat \
    --tts-handler qwen3 \
    --enable-realtime-api

# Terminal 2: Run the demo client

python scripts/listen_and_play_realtime.py \
    --api-key $OPENAI_API_KEY \
    --model gpt-4o-mini \
    --voice "alloy"

The demo script in scripts/listen_and_play_realtime.py creates a WebSocketTransport that speaks the OpenAI Realtime protocol to your local server rather than OpenAI's hosted endpoint.

Programmatic Service Integration

For custom applications, instantiate components directly:

from speech_to_speech.api.openai_realtime.service import RealtimeService
from speech_to_speech.api.openai_realtime.runtime_config import RuntimeConfig
from speech_to_speech.s2s_pipeline import build_pipeline

# Initialize shared configuration

config = RuntimeConfig()

# Build the handler pipeline

pipeline = build_pipeline(
    stt_handler="whisper",
    llm_handler="chat",
    tts_handler="qwen3",
    runtime_config=config,
)

# Create service and register connection

service = RealtimeService(pipeline=pipeline, runtime_config=config)
connection_id = service.register()

# Stream audio data (normally from WebSocket)

service.append_pcm(
    connection_id,
    pcm_bytes=audio_chunk,
    src_rate=16000
)

# Retrieve generated response events

response_events = service.begin_audio_response(connection_id)

This pattern mirrors websocket_router.py internals while enabling custom preprocessing or postprocessing.

Dynamic Session Updates

Modify behavior mid‑conversation using OpenAI‑compatible session updates:

from speech_to_speech.api.openai_realtime.runtime_config import RuntimeConfig
from openai.types.realtime.realtime_session_create_request import RealtimeSessionCreateRequest

# Construct partial update (unset fields preserved)

update = RealtimeSessionCreateRequest(
    turn_detection=TurnDetection(
        type="server_vad",
        interrupt_response=False  # Disable barge‑in

    )
)

# Apply atomically to shared config

service.runtime_config.apply_session_update(update)

The apply_session_update() method in runtime_config.py performs a shallow merge, preserving existing values for unspecified fields.

Handler Configuration for Realtime Pipelines

The build_pipeline() function in s2s_pipeline.py accepts handler specifications that determine latency and quality tradeoffs:

Handler	Role	Typical Latency
silero_vad / pyannote_vad	Voice endpoint detection	100-300ms
whisper / faster_whisper	Speech-to-text transcription	200-800ms
chat / mlc_chat	LLM text generation	50-500ms/token
qwen3 / mimic3	Text-to-speech synthesis	50-200ms/chunk

These handlers communicate through PipelineUnit queues, with backpressure management to prevent memory growth during long conversations.

Summary

• The OpenAI Realtime API integration centers on RealtimeService in service.py, which translates WebSocket events to internal pipeline events
• RuntimeConfig provides thread‑safe, shared state across all pipeline handlers, enabling dynamic features like interruptible responses
• The WebSocket router (websocket_router.py) manages connection lifecycle, instantiating isolated PipelineUnit instances per client
• Handler chains are constructed via build_pipeline() in s2s_pipeline.py, with VAD, STT, LLM, and TTS stages interconnected through queue-based streaming
• Full OpenAI protocol compatibility allows drop‑in replacement of hosted Realtime API with local or self‑hosted infrastructure

Frequently Asked Questions

What OpenAI Realtime events does the pipeline support?

The implementation handles SessionCreate, SessionUpdate, ConversationItemCreate, InputAudioBufferAppend, and ResponseCreate events from clients, and emits SessionCreated, SessionUpdated, ConversationItemCreated, and AudioBufferAppend events in response. Unsupported events are logged and discarded rather than raising errors.

Can I use custom STT or TTS models with the Realtime API integration?

Yes. The build_pipeline() function accepts any handler registered in the HANDLER_REGISTRY. Pass stt_handler="your_custom_stt" or implement a handler matching the BaseSTTHandler interface in src/speech_to_speech/handlers/base.py. The pipeline will wire your custom handler into the Realtime event flow automatically.

How does barge‑in interruption work technically?

When RuntimeConfig.interrupt_response_enabled is true, the VAD handler monitors for new speech during TTS playback. Upon detection, it emits an InterruptEvent that the PipelineUnit routes to the TTS handler's abort() method. The TTS handler clears its synthesis queue, and RealtimeService sends a ResponseCancelled event to the client, allowing immediate LLM processing of the new utterance.

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