Phase 7: Voice Agent Frameworks · 55 min · Python · OpenAI SDK · websockets

OpenAI Realtime API: Native Speech-to-Speech

Native speech-to-speech is the future. gpt-realtime-2 eliminates the text intermediary entirely.

Hiring signal: OpenAI Realtime API proficiency is increasingly required as native audio models become standard.

What you will learn

Use OpenAI Realtime API (gpt-realtime-2) for native speech-to-speech conversation
Configure server VAD with eagerness parameter (low/medium/high)
Implement tool calling during Realtime sessions
Manage session lifecycle: create, update, close

The Problem

Traditional voice agents require three separate services: ASR (Deepgram), LLM (OpenAI), TTS (Cartesia). You wire them together, handle streaming between them, manage latency. OpenAI's Realtime API collapses all three into one WebSocket connection — speech goes in, speech comes out.

The Concept

Realtime API Architecture

Feature	Realtime API	Traditional Pipeline
Services	1 (OpenAI)	3 (ASR + LLM + TTS)
Latency	~300-500ms	~400-600ms
Audio format	PCM16 24kHz	Varies
Voice options	OpenAI voices	Any TTS provider
Customization	Limited	Full control
Cost	$0.06/min (text+audio)	Varies by provider
Barge-in	Built-in	Manual

What's the main trade-off of using OpenAI's Realtime API vs a custom ASR+LLM+TTS pipeline?

Realtime API is simpler but less customizable — you're locked into OpenAI's voices, models, and pricing

WebSocket Events

Direction	Event	Purpose
Client → API	`session.update`	Configure voice, instructions, tools
Client → API	`input_audio_buffer.append`	Send audio chunk
Client → API	`input_audio_buffer.commit`	End of user speech
Client → API	`response.create`	Trigger response
API → Client	`response.audio.delta`	Audio chunk from agent
API → Client	`response.audio.done`	Agent finished speaking
API → Client	`conversation.item.input_audio_transcription`	ASR transcript
API → Client	`response.function_call_arguments.done`	Tool call result

Session Configuration

import websockets
import json

async def create_realtime_session():
    """Create an OpenAI Realtime API session."""
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, additional_headers=headers) as ws:
        # Configure session
        session_config = {
            "type": "session.update",
            "session": {
                "voice": "alloy",
                "instructions": "You are a helpful voice agent. Keep responses to 1-3 sentences.",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 500
                },
                "tools": [
                    {
                        "type": "function",
                        "name": "check_flight_status",
                        "description": "Check the status of a flight",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "flight_number": {"type": "string"}
                            },
                            "required": ["flight_number"]
                        }
                    }
                ]
            }
        }
        await ws.send(json.dumps(session_config))

Server-Side VAD

The Realtime API has built-in server-side VAD with configurable threshold:

"turn_detection": {
    "type": "server_vad",      # OpenAI handles VAD
    "threshold": 0.5,           # Sensitivity (0-1)
    "silence_duration_ms": 500  # Endpointing silence threshold
}

Available Voices

Voice	Character	Best For
alloy	Neutral, balanced	General purpose
echo	Warm, friendly	Customer service
fable	Expressive, dramatic	Storytelling
onyx	Deep, authoritative	Professional
nova	Energetic, young	Casual, upbeat
shimmer	Calm, soothing	Healthcare, support

Handling Audio

async def handle_realtime_audio(ws):
    """Handle bidirectional audio with Realtime API."""
    # Send audio to API
    async def send_audio():
        while True:
            audio_chunk = await mic_queue.get()
            event = {
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_chunk).decode()
            }
            await ws.send(json.dumps(event))

    # Receive audio from API
    async def receive_audio():
        async for message in ws:
            data = json.loads(message)
            if data["type"] == "response.audio.delta":
                audio = base64.b64decode(data["delta"])
                await speaker_queue.put(audio)
            elif data["type"] == "response.audio.done":
                print("Agent finished speaking")

    await asyncio.gather(send_audio(), receive_audio())

Function Calling

# The API calls your function and waits for the result
async def handle_function_call(ws, data):
    """Handle function call from Realtime API."""
    if data["type"] == "response.function_call_arguments.done":
        tool_name = data["name"]
        args = json.loads(data["arguments"])

        # Execute the function
        if tool_name == "check_flight_status":
            result = await check_flight_status(**args)

        # Send result back to API
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": data["call_id"],
                "output": json.dumps(result)
            }
        }))

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers Build It, Use It, Ship It, Evaluation, Key Terms, Common Pitfalls, Interview Framing — plus a hands-on lab, quiz, and project artifact.

Create a free account to unlock Phase 0 and Phase 1 of every course — no credit card.

Key terms in this lesson

Agent