AI Voice Agent Guide 2026: Build Speech-to-Speech AI with Real-Time APIs

Microphone → STT → LLM → TTS → Speaker
              ↓      ↓      ↓
           Deepgram  GPT  ElevenLabs

Microphone → OpenAI Realtime API → Speaker
              (STT + LLM + TTS combined)

Microphone → Deepgram STT → Claude → ElevenLabs TTS → Speaker
              (Best STT)    (Best reasoning)  (Best voices)

import asyncio
import websockets
import json
import base64
import pyaudio

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

async def voice_agent():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    # Use Crazyrouter for cost savings
    # url = "wss://crazyrouter.com/v1/realtime?model=gpt-4o-realtime-preview"
    
    headers = {
        "Authorization": "Bearer YOUR_API_KEY",
        "OpenAI-Beta": "realtime=v1"
    }
    
    audio = pyaudio.PyAudio()
    
    # Input stream (microphone)
    input_stream = audio.open(
        format=FORMAT, channels=CHANNELS,
        rate=RATE, input=True, frames_per_buffer=CHUNK
    )
    
    # Output stream (speaker)
    output_stream = audio.open(
        format=FORMAT, channels=CHANNELS,
        rate=RATE, output=True, frames_per_buffer=CHUNK
    )
    
    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure the agent
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": """You are a helpful customer service agent for a tech company. 
                Be concise, friendly, and professional. 
                If you don't know something, say so honestly.""",
                "voice": "nova",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 800
                }
            }
        }))
        
        # Send audio from microphone
        async def send_audio():
            while True:
                data = input_stream.read(CHUNK, exception_on_overflow=False)
                encoded = base64.b64encode(data).decode()
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": encoded
                }))
                await asyncio.sleep(0.01)
        
        # Receive and play audio
        async def receive_audio():
            async for message in ws:
                event = json.loads(message)
                if event["type"] == "response.audio.delta":
                    audio_bytes = base64.b64decode(event["delta"])
                    output_stream.write(audio_bytes)
                elif event["type"] == "response.audio_transcript.delta":
                    print(f"Agent: {event['delta']}", end="", flush=True)
                elif event["type"] == "input_audio_buffer.speech_started":
                    print("\n[User speaking...]")
        
        await asyncio.gather(send_audio(), receive_audio())

asyncio.run(voice_agent())

from openai import OpenAI
import deepgram
import elevenlabs
import asyncio

# Use Crazyrouter for the LLM component
llm_client = OpenAI(
    api_key="YOUR_CRAZYROUTER_KEY",
    base_url="https://crazyrouter.com/v1"
)

# Deepgram for STT (fastest, most accurate)
dg_client = deepgram.DeepgramClient("YOUR_DEEPGRAM_KEY")

# ElevenLabs for TTS (most natural voices)
el_client = elevenlabs.ElevenLabs(api_key="YOUR_ELEVENLABS_KEY")

class VoiceAgent:
    def __init__(self):
        self.conversation_history = []
        self.system_prompt = """You are a friendly AI assistant. 
        Keep responses under 2 sentences for natural conversation flow.
        Be warm, helpful, and concise."""
    
    async def listen(self, audio_stream) -> str:
        """Convert speech to text using Deepgram."""
        response = await dg_client.listen.live.v("1").transcribe(
            audio_stream,
            model="nova-2",
            language="en",
            smart_format=True,
            interim_results=True
        )
        return response.results.channels[0].alternatives[0].transcript
    
    def think(self, user_text: str) -> str:
        """Generate response using LLM via Crazyrouter."""
        self.conversation_history.append({"role": "user", "content": user_text})
        
        response = llm_client.chat.completions.create(
            model="claude-sonnet-4-20250514",  # Best for conversational AI
            messages=[
                {"role": "system", "content": self.system_prompt},
                *self.conversation_history[-10:]  # Last 10 turns for context
            ],
            max_tokens=150  # Keep responses short for voice
        )
        
        assistant_text = response.choices[0].message.content
        self.conversation_history.append({"role": "assistant", "content": assistant_text})
        return assistant_text
    
    def speak(self, text: str) -> bytes:
        """Convert text to speech using ElevenLabs."""
        audio = el_client.generate(
            text=text,
            voice="Rachel",
            model="eleven_turbo_v2_5",
            stream=True
        )
        return b"".join(audio)

agent = VoiceAgent()

# Using Twilio + Voice Agent for phone calls
from flask import Flask, Response
from twilio.twiml.voice_response import VoiceResponse, Gather

app = Flask(__name__)

@app.route("/incoming-call", methods=["POST"])
def incoming_call():
    response = VoiceResponse()
    gather = Gather(
        input="speech",
        action="/process-speech",
        language="en-US",
        speech_timeout="auto"
    )
    gather.say("Hello! I'm your AI assistant. How can I help you today?")
    response.append(gather)
    return Response(str(response), mimetype="text/xml")

@app.route("/process-speech", methods=["POST"])
def process_speech():
    from flask import request
    user_speech = request.form.get("SpeechResult", "")
    
    # Use Crazyrouter LLM to generate response
    llm_response = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a phone customer service agent. Be brief and helpful."},
            {"role": "user", "content": user_speech}
        ],
        max_tokens=100
    )
    
    agent_text = llm_response.choices[0].message.content
    
    response = VoiceResponse()
    gather = Gather(
        input="speech",
        action="/process-speech",
        speech_timeout="auto"
    )
    gather.say(agent_text)
    response.append(gather)
    return Response(str(response), mimetype="text/xml")

system_prompt = """Respond in 1-2 sentences maximum. 
Be conversational and natural. Avoid lists or technical jargon unless asked."""

if not transcription or len(transcription.strip()) < 2:
    return "I didn't quite catch that. Could you repeat?"