OpenAI’s Realtime API enables low-latency, streaming voice conversations with GPT-4o. Unlike traditional speech-to-text → LLM → text-to-speech pipelines, the Realtime API handles all three in a single WebSocket connection with sub-500ms response times. This guide walks through building voice AI applications with the Realtime API.

What Makes the Realtime API Different

Traditional voice AI pipeline:

Audio → Whisper (STT) → GPT-4 → TTS → Audio
         ~1s              ~1s    ~1s
         = 3+ seconds total latency

Realtime API:

Audio → GPT-4o Realtime → Audio
         ~300-600ms total

Additional capabilities:

Interruption handling: the model stops mid-sentence if the user starts speaking
Function calling: trigger actions while speaking (check weather, control devices)
Custom voice: choose from multiple voices (alloy, echo, fable, onyx, nova, shimmer)
Audio formats: PCM16, G.711 µ-law and A-law, Opus
Native VAD: voice activity detection built-in

Getting Access

The Realtime API uses the standard OpenAI API key:

export OPENAI_API_KEY="sk-..."

Current model: gpt-4o-realtime-preview-2024-12-17 Pricing: ~$0.06/min for audio input, ~$0.24/min for audio output (check openai.com for current rates).

Basic WebSocket Connection

The Realtime API uses WebSocket:

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

OPENAI_API_KEY = "your-key"
SAMPLE_RATE = 24000
CHANNELS = 1

async def realtime_session():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }
    
    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure the session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["audio", "text"],
                "instructions": "You are a helpful assistant. Be concise.",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",   # auto-detect when user stops speaking
                    "threshold": 0.5,
                    "silence_duration_ms": 500
                }
            }
        }))
        
        # Send audio and receive responses
        async for message in ws:
            event = json.loads(message)
            await handle_event(event, ws)

Sending Audio

Capture microphone input and send as base64-encoded PCM16:

import sounddevice as sd
import numpy as np
import base64

def capture_audio_chunk(duration_ms=100, sample_rate=24000):
    samples = int(sample_rate * duration_ms / 1000)
    audio = sd.rec(samples, samplerate=sample_rate, channels=1, dtype='int16')
    sd.wait()
    return base64.b64encode(audio.tobytes()).decode('utf-8')

async def stream_microphone(ws):
    """Continuously stream microphone to the API"""
    with sd.InputStream(samplerate=24000, channels=1, dtype='int16') as stream:
        while True:
            audio_chunk, _ = stream.read(2400)  # 100ms chunks
            audio_b64 = base64.b64encode(audio_chunk.tobytes()).decode()
            
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": audio_b64
            }))
            await asyncio.sleep(0.1)

Handling Events

The API sends events as JSON messages:

async def handle_event(event, ws):
    event_type = event.get("type")
    
    if event_type == "response.audio.delta":
        # Received audio chunk — play it
        audio_data = base64.b64decode(event["delta"])
        audio_array = np.frombuffer(audio_data, dtype=np.int16)
        sd.play(audio_array, samplerate=24000)
    
    elif event_type == "response.audio_transcript.delta":
        # Streaming transcript of model's speech
        print(event.get("delta", ""), end="", flush=True)
    
    elif event_type == "conversation.item.input_audio_transcription.completed":
        # User's speech transcribed
        print(f"\nUser: {event['transcript']}")
    
    elif event_type == "response.done":
        print("\n[Response complete]")
    
    elif event_type == "input_audio_buffer.speech_started":
        # User started speaking — optionally cancel current response
        await ws.send(json.dumps({"type": "response.cancel"}))
    
    elif event_type == "error":
        print(f"Error: {event['error']['message']}")

Function Calling (Tools)

Define functions the model can call during conversation:

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "tools": [
            {
                "type": "function",
                "name": "get_weather",
                "description": "Get current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string", "description": "City name"}
                    },
                    "required": ["city"]
                }
            }
        ],
        "tool_choice": "auto"
    }
}))

When the model calls the function:

elif event_type == "response.function_call_arguments.done":
    function_name = event["name"]
    args = json.loads(event["arguments"])
    
    # Execute the function
    if function_name == "get_weather":
        result = get_weather(args["city"])  # your implementation
    
    # Send result back
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result)
        }
    }))
    
    # Trigger a response
    await ws.send(json.dumps({"type": "response.create"}))

Browser Implementation

For web apps, use the official OpenAI Realtime API client or raw WebRTC:

const pc = new RTCPeerConnection();

// Add microphone track
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
stream.getTracks().forEach(track => pc.addTrack(track, stream));

// Get ephemeral token from your backend (don't expose API key in browser)
const tokenResponse = await fetch('/api/realtime-token');
const { client_secret } = await tokenResponse.json();

// Connect
const offer = await pc.createOffer();
await pc.setLocalDescription(offer);

const response = await fetch(
  'https://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17',
  {
    method: 'POST',
    headers: { 'Authorization': `Bearer ${client_secret.value}` },
    body: offer.sdp
  }
);

Use Cases

Customer service agents: voice-driven support with tool access to your database
Language learning: conversation practice with real-time feedback
Accessibility tools: voice interface for applications
Smart home control: voice commands that trigger API calls
Interview practice: AI interviewer with consistent questioning

The Realtime API’s low latency and native interruption handling make it the first practical foundation for truly conversational AI applications.