Build a complete voice agent
On this page
Updated November 2, 2026 — Audio models. Several models in this course are deprecated:
tts-1,tts-1-hd, andgpt-4o-mini-ttswill stop on January 6, 2027 (successor:gpt-realtime-2.1-mini, which the OpenAI TTS guide states is not supported by/v1/audio/speech: speech synthesis migration moves to the Realtime API);gpt-realtime,gpt-4o-realtime, and legacy audio models on January 20, 2027 (gpt-realtime-2.1,gpt-realtime-2.1-mini,gpt-audio-1.5);whisper-1,gpt-4o-transcribe, and/v1/audio/translationson February 26, 2027 (gpt-transcribe,gpt-live-transcribe,v1/realtime/translations). For full-duplex voice, GPT-Live 1 has been available in the API since September 10, 2026.
Corrected November 3, 2026. Input transcription used
gpt-live-transcribe, which only works in a transcription session and supports neitherserver_vadnorsemantic_vad. In a conversation session like this one, where VAD validates each turn, the documentation specifiesgpt-transcribe, which transcribes validated turns and uses previous turns as context. Sources: Realtime transcription · GPT-Transcribe · GPT-Live-Transcribe.
Build a complete voice agent
This lesson brings together all the concepts covered so far to build a fully functional end-to-end voice agent. You will create an assistant that listens to the user, understands their request, executes actions, and responds by voice—all in real time via the Realtime API.
Agent architecture
Four modules, and above all one constraint that drives the entire architecture: everything must happen at once. The microphone captures while the model responds, events arrive while audio plays, and the user can interrupt at any moment. A voice agent is not a question-and-answer loop; it is a system where multiple streams advance in parallel—which is why we rely heavily on asyncio in what follows.
The VoiceAgent class
The class gathers the state that these parallel streams must share: the connection, the current session, and the response currently playing. Precisely because multiple tasks access this state simultaneously, an object is preferable to a set of global variables—when you need to cancel an ongoing response, you must know one exists.
import asyncio
import json
import base64
import websockets
import sounddevice as sd
import numpy as np
import pyaudio
class VoiceAgent:
"""Complete voice agent based on the Realtime API."""
def __init__(self, api_key: str, instructions: str, tools: list = None):
self.api_key = api_key
self.instructions = instructions
self.tools = tools or []
self.ws = None
self.is_playing = False
self.sample_rate = 24000
self.chunk_size = 2400 # 100ms
async def connect(self):
"""Establishes the connection and configures the session."""
url = "wss://api.openai.com/v1/realtime?model=gpt-realtime-2.1"
headers = {
"Authorization": f"Bearer {self.api_key}"
}
self.ws = await websockets.connect(url, additional_headers=headers)
# Wait for session.created
event = json.loads(await self.ws.recv())
print(f"Session: {event['session']['id']}")
# Configure
await self.ws.send(json.dumps({
"type": "session.update",
"session": {
"type": "realtime",
"output_modalities": ["audio"],
"instructions": self.instructions,
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"transcription": {"model": "gpt-transcribe", "languages": ["fr"]},
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 600
}
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "marin"
}
},
"tools": self.tools,
"tool_choice": "auto"
}
}))
await self.ws.recv() # session.updated
print("Voice agent ready.")
async def start(self):
"""Starts the voice agent."""
await self.connect()
await asyncio.gather(
self._capture_audio(),
self._process_events()
)Audio capture module
async def _capture_audio(self):
"""Captures microphone input and sends it to the Realtime API."""
loop = asyncio.get_event_loop()
queue = asyncio.Queue()
def callback(indata, frames, time_info, status):
pcm = (indata[:, 0] * 32767).astype(np.int16)
loop.call_soon_threadsafe(queue.put_nowait, pcm)
stream = sd.InputStream(
samplerate=self.sample_rate,
channels=1,
dtype="float32",
blocksize=self.chunk_size,
callback=callback
)
with stream:
while True:
pcm_data = await queue.get()
audio_b64 = base64.b64encode(pcm_data.tobytes()).decode()
await self.ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": audio_b64
}))Event processing module
async def _process_events(self):
"""Processes all events from the server."""
pa = pyaudio.PyAudio()
player = pa.open(
format=pyaudio.paInt16, channels=1,
rate=self.sample_rate, output=True
)
try:
async for raw in self.ws:
event = json.loads(raw)
t = event["type"]
if t == "response.output_audio.delta":
audio = base64.b64decode(event["delta"])
player.write(audio)
self.is_playing = True
elif t == "response.output_audio.done":
self.is_playing = False
elif t == "response.output_audio_transcript.delta":
print(event["delta"], end="", flush=True)
elif t == "input_audio_buffer.speech_started":
if self.is_playing:
await self.ws.send(json.dumps({
"type": "response.cancel"
}))
self.is_playing = False
elif t == "response.done":
await self._handle_response_done(event)
elif t == "error":
print(f"\nError: {event['error']['message']}")
finally:
player.close()
pa.terminate()Tool management module
This module transforms the agent from a simple voice chatbot into an assistant capable of taking action. The difference from classic function calling lies in the timing constraint: the user is on the line with no visual progress indicator. If execution takes more than a second or two, have the agent speak while it works—“Let me check that for you right now”—because silence in a voice conversation sounds like a breakdown.
async def _handle_response_done(self, event):
"""Handles function calls in the response."""
response = event["response"]
function_calls = [
o for o in response.get("output", [])
if o["type"] == "function_call"
]
if not function_calls:
print() # New line after transcription
return
for call in function_calls:
name = call["name"]
args = json.loads(call["arguments"])
call_id = call["call_id"]
print(f"\n→ {name}({args})")
# Execute the function (adapt to your business logic)
result = await self._execute_function(name, args)
await self.ws.send(json.dumps({
"type": "conversation.item.create",
"item": {
"type": "function_call_output",
"call_id": call_id,
"output": json.dumps(result)
}
}))
# Ask the model to continue
await self.ws.send(json.dumps({"type": "response.create"}))
async def _execute_function(self, name: str, args: dict) -> dict:
"""Dispatches the function call."""
# Implement according to your business logic
return {"error": f"Function {name} not implemented"}Launch the agent
async def main():
agent = VoiceAgent(
api_key="sk-...",
instructions=(
"You are a voice assistant for Corsen AI. "
"You help users manage their appointments "
"and get information about our services. "
"Respond in French, concisely and professionally."
),
tools=[
{
"type": "function",
"name": "check_availability",
"description": "Checks availability for a time slot",
"parameters": {
"type": "object",
"properties": {
"date": {"type": "string"},
"service": {"type": "string"}
},
"required": ["date"]
}
}
]
)
await agent.start()
asyncio.run(main())Key takeaways
- A complete voice agent combines four modules: audio, session, tools, and state management
- The
VoiceAgentclass encapsulates all logic behind a simple interface asyncio.gatherenables parallel microphone capture and event processing- Handling interruptions (canceling a response when the user speaks) is essential
- Tools transform the agent from a simple voice chatbot into an assistant capable of action