Building a voice AI agent with GPT-4o realtime

Building a voice AI agent GPT-4o realtime API requires a persistent WebSocket connection, raw audio streaming, and careful session management. This tutorial walks through a minimal Python implementation that captures microphone input, sends PCM audio to OpenAI, and plays back the model’s responses with low latency. You will end up with a working full-duplex voice loop you can extend.

Prerequisites

• Python 3.10 or newer
• An OpenAI API key with access to the GPT-4o Realtime preview (gpt-4o-realtime-preview-2024-10-01)
• A microphone and speakers (or a headset)
• Install dependencies:

pip install websockets pyaudio python-dotenv

Create a .env file in your project root:

OPENAI_API_KEY=sk-...

If you are on Linux, you may need portaudio19-dev before pip install pyaudio.

Connect to the Realtime API

The endpoint expects the model as a query parameter and two headers. The OpenAI-Beta header is mandatory for the preview.

import os
import asyncio
import json
import base64
import websockets
from dotenv import load_dotenv

load_dotenv()

URL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"

async def connect():
    headers = {
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "OpenAI-Beta": "realtime=v1",
    }
    ws = await websockets.connect(URL, additional_headers=headers)
    return ws

Run a minimal coroutine that calls connect() and prints the first message. You should receive a session.created event. That is checkpoint one.

Expected output (truncated):

{
  "type": "session.created",
  "session": {
    "id": "sess_01ABC",
    "model": "gpt-4o-realtime-preview-2024-10-01",
    "audio_format": "pcm16",
    "sample_rate": 24000
  }
}

If you see 401, your key is missing or lacks preview access. If you see 404, the model name is wrong.

Configure the session

Before sending audio, lock in the audio format and set behavior. The realtime model defaults to 24kHz PCM16 mono. We enable server-side VAD so the model detects when you stop talking.

async def configure_session(ws):
    config = {
        "type": "session.update",
        "session": {
            "instructions": "You are a concise voice assistant. Answer in short sentences.",
            "audio_format": "pcm16",
            "sample_rate": 24000,
            "turn_detection": {
                "type": "server_vad",
                "silence_duration_ms": 600,
                "threshold": 0.5
            }
        }
    }
    await ws.send(json.dumps(config))

For a voice AI agent GPT-4o realtime API, server VAD removes manual push-to-talk logic. The server commits the audio buffer and triggers a response automatically.

Capture microphone audio

PyAudio reads frames from the input device. At 24kHz mono with 16-bit samples, 20ms equals 480 samples. We base64-encode and send as input_audio_buffer.append.

import pyaudio

CHUNK = 480
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 24000

def start_mic():
    pa = pyaudio.PyAudio()
    stream = pa.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                     input=True, frames_per_buffer=CHUNK)
    return pa, stream

async def send_mic(ws, stream):
    while True:
        data = stream.read(CHUNK, exception_on_overflow=False)
        b64 = base64.b64encode(data).decode()
        msg = {
            "type": "input_audio_buffer.append",
            "audio": b64
        }
        await ws.send(json.dumps(msg))
        await asyncio.sleep(0)  # yield to event loop

Do not block the loop. The sleep(0) keeps the coroutine cooperative with the receiver.

Play back model audio

The server streams response.audio.delta events. Each delta is base64 PCM16 at the session sample rate. Write it directly to the output stream.

def start_speaker():
    pa = pyaudio.PyAudio()
    stream = pa.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                     output=True, frames_per_buffer=CHUNK)
    return pa, stream

async def receive_audio(ws, out_stream):
    async for raw in ws:
        evt = json.loads(raw)
        if evt["type"] == "response.audio.delta":
            audio = base64.b64decode(evt["delta"])
            out_stream.write(audio)
        elif evt["type"] == "response.done":
            print("Agent finished speaking")
        elif evt["type"] == "error":
            print("Server error:", evt.get("error"))

Drive the conversation

Wire the pieces together. The server VAD will auto-commit and generate. We run the mic sender and receiver concurrently.

async def main():
    ws = await connect()
    await configure_session(ws)

    pa_in, mic = start_mic()
    pa_out, speaker = start_speaker()

    await asyncio.gather(
        send_mic(ws, mic),
        receive_audio(ws, speaker)
    )

if __name__ == "__main__":
    asyncio.run(main())

Run python voice_agent.py. Speak a clear sentence like “What’s the capital of France?” You should hear a spoken reply within a few hundred milliseconds of stopping. That is checkpoint two.

Inspect the event stream

During development, log every event type to understand the flow.

async def receive_audio(ws, out_stream):
    async for raw in ws:
        evt = json.loads(raw)
        print("EVENT:", evt["type"])
        if evt["type"] == "response.audio.delta":
            out_stream.write(base64.b64decode(evt["delta"]))

You will see input_audio_buffer.speech_started, input_audio_buffer.speech_stopped, response.created, and response.audio.delta sequences. This confirms the voice AI agent GPT-4o realtime API pipeline is live.

Manual commit flow

If you disable server VAD, you must commit and generate yourself:

async def manual_turn(ws):
    await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
    await ws.send(json.dumps({
        "type": "response.create",
        "response": {"instructions": "Reply briefly."}
    }))

Call manual_turn after the user stops speaking (e.g., via local VAD).

Handle errors and rate limits

The API sends error events with a code. Wrap sends and log disconnects.

try:
    await ws.send(json.dumps(msg))
except websockets.exceptions.ConnectionClosed as e:
    print(f"Socket dropped: {e}")

In production, a gateway such as n4n.ai provides automatic fallback when a provider is rate-limited or degraded, but the code above targets OpenAI directly for clarity.

Tune for latency

• Drop CHUNK to 240 samples (10ms) for lower glass-to-glass delay; monitor CPU.
• Set silence_duration_ms to 400–500 to make the agent cut in faster.
• Use response.create with per-turn instructions instead of rewriting the session.
• Keep the speaker buffer small; large frames_per_buffer adds playback lag.

Stop and cleanup

Catch KeyboardInterrupt to terminate PyAudio cleanly.

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pa_in.terminate()
    pa_out.terminate()

Leaving streams open will block your audio device.

Extend the agent

The session supports conversation.item.create with tools for function calling. You can stream tool results back as text or audio. For multi-model routing, the same OpenAI-compatible shape works behind a unified endpoint addressing 240+ models, but the realtime preview is currently OpenAI-only.

You now have a working voice AI agent GPT-4o realtime API client: open socket, configure session, stream mic, play deltas. From here, add barge-in handling by aborting the current response on speech_started, and add telemetry to measure turn-taking latency. Test with a headset first; echo cancellation is not built into the API.