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?
The Realtime API collapses ASR+LLM+TTS into one service, which is simpler (one WebSocket, one provider) but less customizable. You can't use Deepgram for STT (better accuracy), Cartesia for TTS (lower latency), or Claude for LLM (different personality). You're locked into OpenAI's voices (alloy, echo, fable, etc.), GPT-4o models, and their pricing ($0.06/min for audio). The latency is comparable (~300-500ms), and barge-in is built-in. The trade-off: simplicity and speed-to-market vs control and flexibility. Use Realtime API for prototypes and simple agents; use custom pipeline for production agents needing specific providers.
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)
}
}))