Building Real-Time Voice Agents Using ElevenLabs

# Install the SDK
pip install elevenlabs

# Python: Conversational AI Agent (Fully Managed)
import asyncio
from elevenlabs.client import ElevenLabs
from elevenlabs.conversational_ai.conversation import Conversation
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface

client = ElevenLabs(api_key="your-api-key")

# Agent ID created in ElevenLabs Console
AGENT_ID = "your-agent-id"

async def main():
    conversation = Conversation(
        client=client,
        agent_id=AGENT_ID,
        
        # Use the default mic/speaker for local testing
        audio_interface=DefaultAudioInterface(),
        
        # Callback when agent responds
        callback_agent_response=lambda response: print(f"Agent: {response}"),
        
        # Callback when user speech is recognized
        callback_user_transcript=lambda transcript: print(f"User: {transcript}"),
        
        # Optional: handle conversation end
        callback_agent_response_correction=lambda original, corrected: print(
            f"Corrected: {corrected}"
        ),
    )
    
    conversation.start_session()
    print("Voice agent ready — speak now.")
    
    # Wait for conversation to end (user says goodbye or session timeout)
    await asyncio.sleep(300)
    conversation.end_session()

asyncio.run(main())

# FastAPI webhook server that ElevenLabs calls with conversation context
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json

app = FastAPI()
openai_client = OpenAI()

@app.post("/llm-webhook")
async def custom_llm(request: Request):
    body = await request.json()
    
    # ElevenLabs sends conversation history in OpenAI-compatible format
    messages = body.get("messages", [])
    
    # Add your system prompt and any RAG context here
    system_message = {
        "role": "system",
        "content": """You are a helpful customer support agent for Acme Corp.
        Be concise — voice responses should be under 3 sentences.
        You have access to order history and can process refunds."""
    }
    
    full_messages = [system_message] + messages
    
    # Stream response back to ElevenLabs (streaming is critical for latency)
    async def generate():
        stream = openai_client.chat.completions.create(
            model="gpt-4o",
            messages=full_messages,
            stream=True,
            max_tokens=150,  # Keep responses short for voice
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                # Return chunks in ElevenLabs-expected format
                yield chunk.choices[0].delta.content
    
    return StreamingResponse(generate(), media_type="text/plain")

import asyncio
from elevenlabs import ElevenLabs, VoiceSettings
from openai import OpenAI

eleven = ElevenLabs(api_key="your-elevenlabs-key")
openai_client = OpenAI()

async def speak_streaming_response(user_input: str):
    """
    Pipeline: LLM streams text → TTS starts playing first sentence
    while LLM is still generating the rest.
    """
    
    # 1. Get LLM response as a stream
    llm_stream = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": user_input}],
        stream=True,
        max_tokens=200,
    )
    
    def text_generator():
        """Extract text from LLM stream chunks"""
        for chunk in llm_stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta
    
    # 2. Pipe LLM stream directly to ElevenLabs streaming TTS
    # ElevenLabs buffers to sentence boundaries internally
    audio_stream = eleven.generate(
        text=text_generator(),  # Accepts a generator!
        voice="Rachel",
        model="eleven_turbo_v2_5",  # Fastest model, lowest latency
        voice_settings=VoiceSettings(
            stability=0.5,
            similarity_boost=0.75,
            style=0.0,
            use_speaker_boost=True
        ),
        stream=True,
    )
    
    # 3. Stream audio chunks to output (e.g., WebSocket, pyaudio)
    for chunk in audio_stream:
        if chunk:
            yield chunk  # Each chunk is raw PCM audio bytes

# Play locally
import pyaudio
async def play_response(user_input: str):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)
    
    async for audio_chunk in speak_streaming_response(user_input):
        stream.write(audio_chunk)
    
    stream.close()