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

> Learn how the OpenAI Realtime API integrates with the speech-to-speech pipeline. Discover the architecture and code for this WebSocket bridge, synchronizing VAD, STT, LLM, and TTS components for seamless audio processing.

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

---

**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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) |

### RuntimeConfig: The Shared Session State

In [`runtime_config.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/s2s_pipeline.py)

Each `PipelineUnit` (defined in [`pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/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:

```bash

# 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`](https://github.com/huggingface/speech-to-speech/blob/main/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:

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py) internals while enabling custom preprocessing or postprocessing.

### Dynamic Session Updates

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

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)) manages connection lifecycle, instantiating isolated `PipelineUnit` instances per client
- **Handler chains** are constructed via `build_pipeline()` in [`s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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.