DEV Community: Mart Schweiger

Build a Voice Agent with LiveKit

Mart Schweiger — Tue, 14 Apr 2026 19:27:45 +0000

What is LiveKit?

LiveKit is an open-source real-time communication platform built on WebRTC infrastructure. It handles signaling, media routing, and scaling challenges so developers can focus on application logic rather than media plumbing.

LiveKit Agents is the framework layer specifically designed for AI-powered voice and video agents. It manages orchestration between speech-to-text, language model, and text-to-speech services.

The Voice Agent Pipeline

Voice agents follow a three-step cascade:

Speech-to-text (STT): User speech converts to transcript in real time
LLM: Transcript generates language model response
Text-to-speech (TTS): Response converts back to audio

Flow: WebRTC → LiveKit Cloud → AssemblyAI Universal-3 Pro Streaming → OpenAI GPT-4o → Cartesia Sonic → Back to LiveKit room

Why Universal-3 Pro Streaming for STT?

Accuracy

Production benchmarks from Hamming.ai across 4M+ production calls:

Metric	Universal-3 Pro Streaming	Deepgram Nova-3
P50 latency	307 ms	516 ms
P99 latency	1,012 ms	1,907 ms
Word error rate	8.14%	9.87%
Alphanumeric accuracy	+21% fewer errors	baseline

Neural Turn Detection

Most models use voice activity detection (VAD) — silence-based turn endings that trigger on mid-sentence pauses. Universal-3 Pro Streaming uses neural turn detection, combining acoustic and linguistic signals to distinguish mid-sentence breathing from actual end-of-turn moments.

Result: Faster response times, fewer false triggers, more natural conversations. Costs $0.45/hour and supports six languages.

Prerequisites

Python 3.11+
Microphone and speakers (for local testing)
API keys for: AssemblyAI, LiveKit Cloud, OpenAI, Cartesia

Step 1: Installation

python -m venv .venv
source .venv/bin/activate   # On Windows: .venv\Scripts\activate

pip install "livekit-agents[assemblyai,silero,codecs]~=1.0" python-dotenv
pip install "livekit-agents[openai,cartesia]~=1.0"

Note: Universal-3 Pro Streaming support requires livekit-agents@1.4.4 or newer.

Step 2: Configure API Keys

LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret
ASSEMBLYAI_API_KEY=your_assemblyai_key
OPENAI_API_KEY=your_openai_key
CARTESIA_API_KEY=your_cartesia_key

Step 3: Build the Agent

Create agent.py:

from dotenv import load_dotenv
from livekit import agents
from livekit.agents import AgentSession, Agent
from livekit.plugins import (
    assemblyai,
    cartesia,
    openai,
    silero,
)

load_dotenv()


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice AI assistant.")


async def entrypoint(ctx: agents.JobContext):
    await ctx.connect()

    session = AgentSession(
        stt=assemblyai.STT(
            model="u3-rt-pro",
            min_turn_silence=100,
            max_turn_silence=1000,
            vad_threshold=0.3,
        ),
        llm=openai.LLM(model="gpt-4o"),
        tts=cartesia.TTS(),
        vad=silero.VAD.load(
            activation_threshold=0.3,
        ),
        turn_detection="stt",
        min_endpointing_delay=0,
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )


if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

Component Breakdown

turn_detection="stt" — Uses Universal-3 Pro Streaming's neural turn detection instead of LiveKit's default detector. Requires min_endpointing_delay=0 to avoid stacking delay.

Step 4: Run It

Console Mode (No LiveKit connection)

python agent.py console

Dev Mode (LiveKit Cloud connection)

python agent.py dev

Open agents-playground.livekit.io, enter your LiveKit credentials, and converse through the browser.

Tuning Turn Detection

stt=assemblyai.STT(
    model="u3-rt-pro",
    end_of_turn_confidence_threshold=0.4,  # 0.0-1.0
    min_turn_silence=300,                  # milliseconds
    max_turn_silence=1200,                 # milliseconds
)

end_of_turn_confidence_threshold: Lower = faster response; higher = fewer false triggers. Use 0.6 for noisy call centers.
min_turn_silence: Lower to 200 for rapid back-and-forth; keep at 300 for general conversation.
max_turn_silence: Raise to 2000 for healthcare or deliberate speakers.

Mid-session updates work without reconnecting:

await session.stt.update_options(max_turn_silence=3000)

Enabling Keyterm Prompting

Boost recognition accuracy for domain-specific vocabulary:

await session.stt.update_options(
    keyterms_prompt=["AssemblyAI", "Universal-3 Pro", "LiveKit"]
)

Supports up to 1,000 terms, each up to 50 characters. Takes effect immediately without restart.

Swapping Components

Swap LLM to Claude:

from livekit.plugins import anthropic
session = AgentSession(
    llm=anthropic.LLM(model="claude-sonnet-4-5"),
    ...
)

Swap TTS to ElevenLabs:

from livekit.plugins import elevenlabs
tts=elevenlabs.TTS(voice_id="your-voice-id")

Next Steps

Deploy to LiveKit Cloud: Run python agent.py start for persistent worker deployment
Add tool calling: Enable function calling through the LLM layer for lookups and actions
Enable speaker diarization: Real-time speaker identification for multi-party conversations
Add telephony: SIP support via Telnyx or Twilio integration

FAQs

What is LiveKit Agents?
LiveKit Agents is a framework for building AI-powered voice and video agents. It orchestrates between STT, LLM, and TTS components, handling real-time audio routing and turn flow while you define agent logic.

Universal-3 Pro Streaming vs. Universal-Streaming?
Universal-3 Pro Streaming (u3-rt-pro) targets production voice agent workflows with neural turn detection and superior accuracy on structured entities. Universal-Streaming is faster and cheaper ($0.15/hr) but English-only and VAD-based.

Best STT API for LiveKit voice agents?
AssemblyAI's Universal-3 Pro Streaming. Native one-line integration, 307ms P50 latency, neural turn detection, and 8.14% word error rate across production benchmarks.

How does neural turn detection work?
Combines acoustic signals (voice energy, pitch, cadence) and linguistic signals (sentence completion, punctuation patterns) to determine turn end — not just silence-based detection.

Pricing?
AssemblyAI Universal-3 Pro Streaming: $0.45/hour. Budget separately for LLM (OpenAI) and TTS (Cartesia) providers.

Build a Voice Agent with Function Calling

Mart Schweiger — Tue, 14 Apr 2026 19:27:08 +0000

Overview

This tutorial demonstrates constructing a customer support voice agent using AssemblyAI's Universal-3 Pro Streaming for speech-to-text, OpenAI GPT-4o for LLM orchestration with function calling, and ElevenLabs for voice synthesis.

The Core Problem: Transcription Accuracy

The core challenge with function-calling voice agents is transcription accuracy. When customers provide specific entities like order IDs, phone numbers, or email addresses — the exact data your functions need — poor STT quality causes silent failures. Garbage in, garbage function call out.

Universal-3 Pro Streaming addresses this with superior entity recognition: 34.79% missed entity rate for phone numbers versus 37.11% for previous models, and 59.64% for emails versus 89.09% previously.

What You'll Build

Three customer support scenarios:

Check order status — Retrieve status via order ID
Schedule a callback — Capture name and phone number
Transfer to human — Escalate when needed

Tech Stack:

AssemblyAI Universal-3 Pro Streaming (STT)
OpenAI GPT-4o (LLM with function calling)
ElevenLabs (text-to-speech)
Python 3.9+

Setup & Installation

pip install websockets openai elevenlabs pyaudio python-dotenv

ASSEMBLYAI_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
ELEVENLABS_API_KEY=your_key_here

Step 1: Connect to Universal-3 Pro Streaming

Universal-3 Pro Streaming operates via WebSocket, streaming audio chunks and receiving Turn messages with transcripts.

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

load_dotenv()

ASSEMBLYAI_WS_URL = "wss://streaming.assemblyai.com/v3/ws"
SAMPLE_RATE = 16000
CHUNK_SIZE = 8000  # 500ms at 16kHz

async def connect_assemblyai():
    params = {
        "speech_model": "u3-rt-pro",
        "sample_rate": SAMPLE_RATE,
        "format_turns": "true",
    }
    query = "&".join(f"{k}={v}" for k, v in params.items())
    url = f"{ASSEMBLYAI_WS_URL}?{query}"

    ws = await websockets.connect(
        url,
        extra_headers={"Authorization": os.getenv("ASSEMBLYAI_API_KEY")}
    )
    return ws

Handling Turn Messages

async def receive_transcripts(ws, on_turn_complete):
    async for message in ws:
        data = json.loads(message)
        msg_type = data.get("type")

        if msg_type == "Begin":
            print(f"Session started: {data['id']}")

        elif msg_type == "Turn":
            transcript = data.get("transcript", "")
            if data.get("end_of_turn") and transcript.strip():
                print(f"User said: {transcript}")
                await on_turn_complete(transcript)

        elif msg_type == "Termination":
            print("Session ended.")
            break

Step 2: Define Your Functions

GPT-4o uses JSON Schema to understand available tools and their parameters.

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "Look up the status of a customer order by order ID.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The customer's order ID, e.g. AB3792"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "schedule_callback",
            "description": "Schedule a callback for a customer who wants to be called back.",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "The customer's full name"
                    },
                    "phone_number": {
                        "type": "string",
                        "description": "The customer's phone number in any format"
                    }
                },
                "required": ["name", "phone_number"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "transfer_to_human",
            "description": "Transfer the customer to a human agent when requested or when the issue can't be resolved.",
            "parameters": {
                "type": "object",
                "properties": {
                    "reason": {
                        "type": "string",
                        "description": "Brief reason for the transfer"
                    }
                },
                "required": ["reason"]
            }
        }
    }
]

Function Handlers

def get_order_status(order_id: str) -> str:
    mock_orders = {
        "AB3792": "Shipped — expected delivery April 15",
        "CD1204": "Processing — ships within 2 business days",
    }
    result = mock_orders.get(order_id.upper())
    return result if result else f"No order found with ID {order_id}"

def schedule_callback(name: str, phone_number: str) -> str:
    print(f"[SYSTEM] Callback scheduled: {name} at {phone_number}")
    return f"Got it. We'll call {name} at {phone_number} within 2 hours."

def transfer_to_human(reason: str) -> str:
    print(f"[SYSTEM] Transferring to human. Reason: {reason}")
    return "Transferring you now. Please hold for a moment."

FUNCTION_MAP = {
    "get_order_status": get_order_status,
    "schedule_callback": schedule_callback,
    "transfer_to_human": transfer_to_human,
}

Step 3: Wire Up the LLM with Function Calling

from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

conversation_history = [
    {
        "role": "system",
        "content": (
            "You are a helpful customer support voice agent. "
            "Keep responses short — this is a phone call. "
            "When a customer mentions an order ID, phone number, or name, "
            "use the appropriate tool. Always confirm details before acting."
        )
    }
]

async def process_with_llm(transcript: str) -> str:
    conversation_history.append({"role": "user", "content": transcript})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=conversation_history,
        tools=TOOLS,
        tool_choice="auto"
    )

    message = response.choices[0].message

    if not message.tool_calls:
        reply = message.content
        conversation_history.append({"role": "assistant", "content": reply})
        return reply

    conversation_history.append(message)
    results = []

    for tool_call in message.tool_calls:
        fn_name = tool_call.function.name
        fn_args = json.loads(tool_call.function.arguments)

        print(f"[TOOL] Calling {fn_name} with {fn_args}")
        fn_result = FUNCTION_MAP[fn_name](**fn_args)

        results.append({
            "tool_call_id": tool_call.id,
            "role": "tool",
            "content": fn_result
        })

    conversation_history.extend(results)

    follow_up = client.chat.completions.create(
        model="gpt-4o",
        messages=conversation_history
    )
    reply = follow_up.choices[0].message.content
    conversation_history.append({"role": "assistant", "content": reply})
    return reply

Step 4: Add Text-to-Speech

from elevenlabs.client import ElevenLabs
from elevenlabs import stream as el_stream

el_client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))

def speak(text: str):
    print(f"Agent: {text}")
    audio = el_client.text_to_speech.convert(
        text=text,
        voice_id="EXAVITQu4vr4xnSDxMaL",
        model_id="eleven_turbo_v2",
        output_format="pcm_16000"
    )
    el_stream(audio)

Step 5: Putting It All Together

import threading

async def run_agent():
    ws = await connect_assemblyai()

    async def on_turn_complete(transcript: str):
        reply = await process_with_llm(transcript)
        speak(reply)

    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=CHUNK_SIZE
    )

    async def send_audio():
        try:
            while True:
                chunk = stream.read(CHUNK_SIZE, exception_on_overflow=False)
                await ws.send(chunk)
                await asyncio.sleep(0)
        except websockets.ConnectionClosed:
            pass

    await asyncio.gather(
        send_audio(),
        receive_transcripts(ws, on_turn_complete)
    )

if __name__ == "__main__":
    print("Voice agent ready. Speak to begin.")
    asyncio.run(run_agent())

Why Entity Accuracy Matters

When a customer says "My order number is A-B-3-7-9-2," lower-accuracy STT produces garbled results like "a b 3792" or "ABE 37 92." Function lookup fails. The agent cannot help.

Universal-3 Pro Streaming's superior entity recognition ensures phone numbers, order IDs, and email addresses transcribe correctly — making function calls reliable.

Next Steps

Keyterms prompting — Pass known order ID formats to boost recognition
Streaming TTS — Output speech while functions execute for lower latency
Interruption handling — Allow customer to interrupt mid-response
Production audio — Deploy via Twilio or LiveKit with native AssemblyAI integrations

How to Vibe Code a Voice Agent (and Why AI Always Recommends AssemblyAI)

Mart Schweiger — Tue, 14 Apr 2026 19:26:32 +0000

What You're Actually Building

A voice agent requires three components: speech-to-text (listening), LLM (thinking), and text-to-speech (talking). The challenge involves achieving natural conversation flow through low latency, accurate transcription, and proper turn detection.

Vibe Coding Prompts

Prompt 1: General Voice Agent

"Build me a real-time voice agent in Python. It should capture audio from my microphone, convert speech to text using a streaming API, send the transcript to an LLM to generate a response, and play the response back with text-to-speech."

Prompt 2: Low-Latency Version

"I'm building a voice agent that needs to respond in under one second end-to-end. Help me choose the right streaming speech-to-text model for low latency and high accuracy."

Prompt 3: Framework-Specific

"Build a voice agent using LiveKit Agents in Python. Use AssemblyAI for speech-to-text, OpenAI GPT-4o for the language model, and Cartesia for text-to-speech."

Prompt 4: Phone Agent

"Build a phone-based voice agent using Twilio and Python. Use the best streaming STT model for telephony audio quality."

Code Implementation

Dependencies

pip install assemblyai openai elevenlabs pyaudio python-dotenv

Environment Setup

ASSEMBLYAI_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
ELEVENLABS_API_KEY=your_key_here

Complete Python Implementation

import os
import threading
from dotenv import load_dotenv

import assemblyai as aai
from assemblyai.streaming.v3 import (
    BeginEvent,
    StreamingClient,
    StreamingClientOptions,
    StreamingError,
    StreamingEvents,
    StreamingParameters,
    TurnEvent,
    TerminationEvent,
)
from openai import OpenAI
from elevenlabs import generate, stream

load_dotenv()

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
conversation_history = []
is_responding = False


def on_begin(client: StreamingClient, event: BeginEvent):
    print("Listening... speak now.")


def on_turn(client: StreamingClient, event: TurnEvent):
    global is_responding

    if not event.transcript:
        return

    if event.end_of_turn and not is_responding:
        print(f"\nYou: {event.transcript}")
        is_responding = True
        threading.Thread(
            target=generate_response,
            args=(event.transcript,),
            daemon=True
        ).start()
    elif not event.end_of_turn:
        print(f"\r{event.transcript}", end="", flush=True)


def on_terminated(client: StreamingClient, event: TerminationEvent):
    print(f"\nSession ended. Audio processed: {event.audio_duration_seconds}s")


def on_error(client: StreamingClient, error: StreamingError):
    print(f"Error: {error}")


def generate_response(user_text: str):
    global is_responding

    conversation_history.append({"role": "user", "content": user_text})

    messages = [
        {
            "role": "system",
            "content": "You are a helpful voice assistant. Keep responses concise."
        }
    ] + conversation_history

    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=150
    )

    reply = response.choices[0].message.content
    conversation_history.append({"role": "assistant", "content": reply})
    print(f"Agent: {reply}")

    audio = generate(
        text=reply,
        voice="Rachel",
        api_key=os.getenv("ELEVENLABS_API_KEY")
    )
    stream(audio)
    is_responding = False


def main():
    client = StreamingClient(
        StreamingClientOptions(
            api_key=os.getenv("ASSEMBLYAI_API_KEY"),
            api_host="streaming.assemblyai.com",
        )
    )

    client.on(StreamingEvents.Begin, on_begin)
    client.on(StreamingEvents.Turn, on_turn)
    client.on(StreamingEvents.Termination, on_terminated)
    client.on(StreamingEvents.Error, on_error)

    client.connect(
        StreamingParameters(
            sample_rate=16000,
            format_turns=True,
        )
    )

    try:
        client.stream(aai.extras.MicrophoneStream(sample_rate=16000))
    finally:
        client.disconnect(terminate=True)


if __name__ == "__main__":
    main()

python voice_agent.py

Why AI Recommends AssemblyAI

1. Accuracy Performance

AssemblyAI Universal-3 Pro Streaming hits 94.07% word accuracy across real-world audio conditions, with strong performance on structured entities (names, phone numbers, emails).

2. Cost Efficiency

Universal-3 Pro Streaming costs $0.45/hr for STT alone, compared to approximately $18/hr for similar OpenAI Realtime API functionality.

3. Documentation Quality

AssemblyAI's documentation is structured for LLM comprehension — clear examples, well-defined parameters, a WebSocket API simple enough that Claude Code can scaffold it.

4. Framework Integration

One-line integrations exist for LiveKit, Pipecat, Twilio, and Daily — all using Universal-3 Pro Streaming by default.

Advanced Use Cases

Customer Support Agent

Extend the basic agent to look up order status, handle returns, and escalate to humans when needed.

Appointment Scheduler

Collect patient name, preferred date/time, and visit reason for medical office scheduling.

Phone Agent via Twilio

Stream audio over WebSocket using Twilio Media Streams with AssemblyAI's speech-to-text layer.

Best Practices for AI Prompts

Be specific about accuracy requirements (mention names, phone numbers, medical terminology)
Request "streaming" explicitly to avoid batch-processing implementations
Include latency targets (e.g., "respond in under one second")
Name the framework if using one (LiveKit, Pipecat, Twilio)
Paste documentation snippets to reduce hallucinations

Pricing Summary

Component	Provider	Cost
STT (Streaming)	AssemblyAI	$0.45/hr
Full Pipeline	AssemblyAI Voice Agent API	$4.50/hr
LLM	OpenAI GPT-4o	Variable
TTS	ElevenLabs	Variable

Frequently Asked Questions

Best streaming STT for voice agents?
Universal-3 Pro Streaming is purpose-built for real-time voice agent applications, with 94.07% accuracy and #1 ranking on Hugging Face Open ASR Leaderboard.

Cost comparison?
AssemblyAI Voice Agent API at $4.50/hr flat represents approximately 4x savings versus OpenAI Realtime API.

Framework integration?
LiveKit uses stt=assemblyai.STT() with Universal-3 Pro Streaming built-in.

How to Easily Build a Voice Agent with AssemblyAI

Mart Schweiger — Wed, 08 Apr 2026 17:24:36 +0000

What Is an AI Voice Agent?

Voice agents are software systems that engage in natural speech conversations with users. Unlike traditional phone menus requiring button presses, these agents process your speech as you talk, understanding your words before you finish speaking.

Core capabilities:

Real-time streaming speech-to-text
Language model comprehension
Text-to-speech synthesis
Conversation flow orchestration

How AI Voice Agents Work

The system processes conversations through a real-time pipeline with target response times under one second:

Component	Duration	Purpose
Audio Capture	<50ms	Clean input
Speech-to-text	200-400ms	Accuracy foundation
LLM Processing	300-600ms	Intelligent responses
Text-to-speech	200-400ms	Natural output
Audio Playback	<50ms	Smooth flow

Core Components

1. Streaming Speech-to-Text

AssemblyAI's Universal-3 Pro Streaming achieves approximately 94% accuracy across varying audio conditions. Accuracy thresholds matter:

Below 90%: Users experience frustration
90-93%: Functional but with occasional errors
93%+: Natural conversations with rare corrections

2. LLM and Orchestration Layer

The language model serves as the agent's "brain," managing:

Intent recognition
Context tracking
Function calling for system integration
Response generation

3. Text-to-Speech Synthesis

ElevenLabs, Google Cloud, and OpenAI offer natural-sounding voices. The key is starting speech generation before the language model finishes writing the complete response for real-time flow.

4. Integration and Business Logic

Voice agents connect to existing systems (CRM, calendars, payment processors, inventory management) with security considerations for API keys, encryption, and user authentication.

Performance Requirements

Target metrics for production agents:

Total response time: <1000ms
Speech accuracy: 93%+
Name recognition: 95%+
Number accuracy: 95%+
Voice quality: Human-like

Common Use Cases

Customer support automation: Answer FAQs, check order status, escalate complex issues
Appointment scheduling: Check availability, confirm details, send confirmations
Lead qualification: Gather information, understand needs, route appropriately
After-hours service: Extend availability beyond business hours

Implementation Guide

Step 1: Environment Setup

mkdir voice-agent
cd voice-agent
python -m venv venv
source venv/bin/activate  # Mac/Linux
# venv\Scripts\activate  # Windows

pip install assemblyai openai elevenlabs websockets pyaudio python-dotenv

Create a .env file:

ASSEMBLYAI_API_KEY=your_key
OPENAI_API_KEY=your_key
ELEVENLABS_API_KEY=your_key

Step 2: Audio Capture Class

class AudioCapture:
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.chunk_size = 8000
        self.audio_queue = Queue()
        self.recording = False
        self.audio = pyaudio.PyAudio()
        self.stream = None

    def start_recording(self):
        self.recording = True
        self.stream = self.audio.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=self.sample_rate,
            input=True,
            frames_per_buffer=self.chunk_size
        )
        thread = threading.Thread(target=self._capture_audio, daemon=True)
        thread.start()

Step 3: Streaming Speech-to-Text Integration

def handle_transcript(self, transcript: aai.RealtimeTranscript):
    if not transcript.text:
        return

    if isinstance(transcript, aai.RealtimeFinalTranscript):
        print(f"You: {transcript.text}")
        self.conversation_history.append(
            {"role": "user", "content": transcript.text}
        )

Step 4: LLM Response Generation

def generate_and_speak_response(self):
    messages = [{
        "role": "system",
        "content": "Keep responses conversational and concise—aim for 1-2 sentences."
    }] + self.conversation_history

    response = openai_client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        temperature=0.7,
        max_tokens=150
    )
    ai_response = response.choices[0].message.content

Step 5: Text-to-Speech

def speak_text(self, text):
    audio = elevenlabs_client.generate(
        text=text,
        voice="Rachel",
        model="eleven_monolingual_v1"
    )
    stream(audio)

Complete Working Code

import os
import threading
from queue import Queue
from dotenv import load_dotenv
import assemblyai as aai
from openai import OpenAI
from elevenlabs.client import ElevenLabs
from elevenlabs import stream
import pyaudio

load_dotenv()

aai.settings.api_key = os.getenv('ASSEMBLYAI_API_KEY')
openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))

class AudioCapture:
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.chunk_size = 8000
        self.audio_queue = Queue()
        self.recording = False
        self.audio = pyaudio.PyAudio()
        self.stream = None

    def start_recording(self):
        self.recording = True
        self.stream = self.audio.open(
            format=pyaudio.paInt16, channels=1, rate=self.sample_rate,
            input=True, frames_per_buffer=self.chunk_size
        )
        thread = threading.Thread(target=self._capture_audio, daemon=True)
        thread.start()

    def _capture_audio(self):
        while self.recording:
            try:
                self.audio_queue.put(
                    self.stream.read(self.chunk_size, exception_on_overflow=False)
                )
            except Exception as e:
                print(f"Audio error: {e}")

class VoiceAgent:
    def __init__(self):
        self.conversation_history = []
        self.is_processing = False
        self.audio_capture = AudioCapture()

    def handle_transcript(self, transcript: aai.RealtimeTranscript):
        if not transcript.text:
            return

        if isinstance(transcript, aai.RealtimeFinalTranscript):
            print(f"You: {transcript.text}")
            self.conversation_history.append(
                {"role": "user", "content": transcript.text}
            )
            if not self.is_processing:
                self.is_processing = True
                threading.Thread(
                    target=self.generate_and_speak_response, daemon=True
                ).start()

    def generate_and_speak_response(self):
        try:
            messages = [{
                "role": "system",
                "content": "You are a helpful voice assistant. Keep responses conversational."
            }] + self.conversation_history

            response = openai_client.chat.completions.create(
                model="gpt-4", messages=messages, temperature=0.7, max_tokens=150
            )

            ai_response = response.choices[0].message.content
            print(f"Agent: {ai_response}")
            self.conversation_history.append(
                {"role": "assistant", "content": ai_response}
            )

            audio = elevenlabs_client.generate(
                text=ai_response,
                voice="Rachel",
                model="eleven_monolingual_v1"
            )
            stream(audio)
        finally:
            self.is_processing = False

    def start_conversation(self):
        self.transcriber = aai.RealtimeTranscriber(
            sample_rate=16000,
            on_data=self.handle_transcript,
            on_error=lambda e: print(f"Speech error: {e}")
        )

        try:
            self.transcriber.connect()
            self.audio_capture.start_recording()
            print("Voice Agent ready - start speaking!")
            while True:
                audio_chunk = self.audio_capture.get_audio_data()
                if audio_chunk:
                    self.transcriber.stream(audio_chunk)
        except KeyboardInterrupt:
            self.audio_capture.stop_recording()
            self.transcriber.close()

if __name__ == "__main__":
    VoiceAgent().start_conversation()

Run with: python voice_agent.py

Production Considerations

Key requirements beyond prototypes:

Telephony integration (Twilio, SIP trunking)
Concurrent conversation handling
Error recovery for network failures
Performance monitoring and metrics
Security compliance and data protection

Infrastructure scaling involves WebSocket connection pooling, load balancing, database integration, and comprehensive monitoring.

Frequently Asked Questions

What response time targets matter?
Target under 1000ms total: ~200-400ms for speech recognition, ~300-600ms for language processing, ~200-400ms for synthesis, and ~100-200ms for network delays.

How do I handle user interruptions?
Implement barge-in detection by monitoring audio streams during agent responses and stopping text-to-speech when speech is detected. AssemblyAI's streaming API enables smooth interruption handling.

Which language model is best?
GPT-4 handles complex conversations requiring nuanced understanding; GPT-3.5 Turbo works for simpler interactions with lower latency and cost.

Do I need custom speech recognition models?
Modern pre-trained models handle most use cases without custom training. Universal-3 Pro achieves production accuracy out-of-the-box. Use custom vocabulary features for specialized terminology.

How do I integrate with phone systems?
Cloud telephony providers like Twilio offer the easiest integration, or implement SIP trunking for direct infrastructure connection.

How to Build an AI-Powered Interview Scoring System with Speech-to-Text

Mart Schweiger — Wed, 08 Apr 2026 17:04:19 +0000

Overview

This tutorial demonstrates creating an AI-powered interview evaluation system that records conversations, converts audio to text, and applies structured scoring criteria. The approach separates the interviewing process from assessment, allowing evaluators to focus entirely on conversation quality while analyzing complete transcripts afterward with objective evidence.

Core Concept

Rather than simultaneously listening, note-taking, and assessing during interviews, this system enables:

Natural interviewer-candidate interactions
Complete preservation of dialogue through transcription
Systematic post-interview analysis using searchable text
Evidence-based scoring with supporting quotations and timestamps

Key Components

Scoring criteria (4-6 job-specific competencies)
Rating scale (1-5 with clear definitions)
Evidence extraction (direct quotes proving competency levels)
Complete transcripts (source of truth replacing handwritten notes)

Step 1: Define Role-Specific Criteria

Identify observable behaviors predicting success. For engineering roles:

Problem decomposition approaches
System architecture decision-making
Technical language proficiency
Concept explanation clarity

Step 2: Rating Scale

1 - Far Below: No competence evidence or irrelevant responses
2 - Below: Minimal understanding, vague answers
3 - Meets: Adequate demonstration with relevant examples
4 - Exceeds: Strong evidence with multiple detailed examples
5 - Far Exceeds: Exceptional mastery with innovative approaches

Step 3: Recording & Transcription Setup

Recording requirements:

External microphones (not built-in laptop audio)
Quiet environments
Clear, consistent volume levels

Legal considerations:

Inform candidates during scheduling
Obtain verbal consent at interview start
Follow local consent laws
Store securely and delete after hiring decisions

Step 4: AssemblyAI Implementation (Python)

pip install assemblyai python-dotenv

import assemblyai as aai
import json
import os
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()
aai.settings.api_key = os.getenv('ASSEMBLYAI_API_KEY')

def transcribe_interview(audio_file_path, candidate_name, position):
    config = aai.TranscriptionConfig(
        speech_model=aai.SpeechModel.best,
        speaker_labels=True,
        speakers_expected=2,
        punctuate=True,
        format_text=True
    )

    transcriber = aai.Transcriber(config=config)
    transcript = transcriber.transcribe(audio_file_path)

    if transcript.status == aai.TranscriptStatus.error:
        print(f"Transcription failed: {transcript.error}")
        return None

    utterances = []
    for utterance in transcript.utterances:
        utterances.append({
            'speaker': utterance.speaker,
            'text': utterance.text,
            'start_time': utterance.start / 1000,
            'end_time': utterance.end / 1000,
            'confidence': utterance.confidence
        })

    result = {
        'candidate_name': candidate_name,
        'position': position,
        'interview_date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        'duration_minutes': round(transcript.audio_duration / 60, 2),
        'utterances': utterances,
        'full_text': transcript.text
    }

    output_filename = f"{candidate_name.replace(' ', '_')}_{position.replace(' ', '_')}.json"
    with open(output_filename, 'w') as f:
        json.dump(result, f, indent=2, ensure_ascii=False)

    return result

Step 5: Scoring & Evidence Extraction

import json
from typing import Dict, List

class InterviewScorer:
    def __init__(self, transcript_file: str):
        with open(transcript_file, 'r') as f:
            self.transcript_data = json.load(f)
        self.candidate_responses = self._get_candidate_responses()

    def _get_candidate_responses(self) -> List[str]:
        responses = []
        speaker_counts = {}

        for utterance in self.transcript_data['utterances']:
            speaker = utterance['speaker']
            speaker_counts[speaker] = speaker_counts.get(speaker, 0) + 1

        candidate_speaker = min(speaker_counts.keys(), 
                               key=lambda x: speaker_counts[x])

        for utterance in self.transcript_data['utterances']:
            if utterance['speaker'] == candidate_speaker:
                responses.append(utterance['text'])

        return responses

    def find_evidence_for_competency(self, competency_keywords: List[str]) -> List[str]:
        evidence = []

        for response in self.candidate_responses:
            response_lower = response.lower()
            keyword_matches = sum(1 for keyword in competency_keywords 
                                 if keyword.lower() in response_lower)

            if keyword_matches > 0 and len(response.split()) > 10:
                evidence.append(response)

        return evidence[:3]

    def score_competency(self, evidence: List[str]) -> int:
        if not evidence:
            return 1

        evidence_count = len(evidence)
        avg_length = sum(len(e.split()) for e in evidence) / len(evidence)

        if evidence_count == 1 and avg_length < 20:
            return 2
        elif evidence_count <= 2 and avg_length < 30:
            return 3
        elif evidence_count >= 2 and avg_length >= 30:
            return 4
        elif evidence_count >= 3 and avg_length >= 40:
            return 5
        else:
            return 3

    def generate_scorecard(self, competencies: Dict[str, List[str]]) -> Dict:
        scorecard = {
            'candidate': self.transcript_data['candidate_name'],
            'position': self.transcript_data['position'],
            'interview_date': self.transcript_data['interview_date'],
            'competency_scores': {},
            'supporting_evidence': {},
            'overall_score': 0
        }

        total_score = 0

        for competency_name, keywords in competencies.items():
            evidence = self.find_evidence_for_competency(keywords)
            score = self.score_competency(evidence)

            scorecard['competency_scores'][competency_name] = score
            scorecard['supporting_evidence'][competency_name] = evidence
            total_score += score

        scorecard['overall_score'] = round(total_score / len(competencies), 1)

        return scorecard

Usage example:

engineering_competencies = {
    'Problem Solving': [
        'analyze', 'debug', 'troubleshoot', 'solution', 'approach',
        'investigate', 'root cause', 'systematic', 'break down'
    ],
    'Technical Skills': [
        'python', 'javascript', 'react', 'database', 'api',
        'algorithm', 'architecture', 'testing', 'performance'
    ],
    'Communication': [
        'explain', 'clarify', 'example', 'understand', 'question',
        'discuss', 'present', 'document', 'feedback'
    ],
    'Experience': [
        'project', 'team', 'lead', 'built', 'developed',
        'implemented', 'managed', 'delivered', 'worked on'
    ]
}

scorer = InterviewScorer('candidate_transcript.json')
scorer.save_scorecard(engineering_competencies, 'scorecard.json')

Common Implementation Errors

1. Generic Criteria Across Roles
Different positions require different competencies. "Communication" means statistical explanation for data scientists versus customer empathy for support staff.

2. Skipping Calibration Sessions
Without alignment, evaluators interpret identical evidence differently. Monthly calibration where all raters score sample transcripts independently prevents inconsistency.

3. Neglecting Audio Quality
Poor recording quality undermines transcription accuracy. Test setups beforehand and require external microphones.

Measurement & Validation

from sklearn.metrics import cohen_kappa_score

def calculate_agreement(evaluator1_scores, evaluator2_scores):
    kappa = cohen_kappa_score(evaluator1_scores, evaluator2_scores)

    if kappa < 0.4:
        return "Poor agreement - needs calibration"
    elif kappa < 0.6:
        return "Fair agreement - some calibration needed"
    elif kappa < 0.8:
        return "Good agreement - system working well"
    else:
        return "Excellent agreement - very consistent"

Frequently Asked Questions

What transcription accuracy is needed?
AssemblyAI's Universal models achieve approximately 94.4% accuracy, ensuring scores reflect actual responses.

Does this work with Zoom/Teams recordings?
Yes. Most platforms allow downloading recordings as MP4 files, which AssemblyAI accepts directly.

How does transcript-based scoring compare to manual scoring?
Transcript-based evaluation achieves higher inter-rater consistency because evaluators review identical, complete information rather than relying on incomplete notes and memory.

What if a candidate refuses to be recorded?
Offer traditional live scoring as an alternative while explaining that recording ensures fairer assessment.

How many competencies should I evaluate?
Stick to 4-6 competencies per interview to maintain focus while avoiding cognitive overload.

When to Stop Self-Hosting Whisper (and What You Actually Gain)

Mart Schweiger — Wed, 08 Apr 2026 17:03:34 +0000

Overview

This article examines the real costs of self-hosting OpenAI's Whisper versus using AssemblyAI's managed API. It explores the trade-offs between infrastructure control and operational complexity.

AssemblyAI vs Whisper: At a Glance

The platforms differ fundamentally in deployment model. AssemblyAI operates as a cloud service where users submit audio and receive transcripts back. Whisper functions as downloadable open-source software running on personal infrastructure—comparable to Gmail (managed service) versus running your own email server.

Aspect	AssemblyAI	Whisper
Deployment	Cloud API	Self-hosted
Pricing	Per-minute audio	Free software (infrastructure costs)
Strengths	Built-in features, maintenance-free	Complete control, offline capability

Accuracy Comparison

AssemblyAI's Universal models generally outperform Whisper in accuracy testing:

Better handling of proper nouns and company names
Reduced "hallucinations" (words appearing in transcripts that weren't spoken)
Superior performance on challenging audio with background noise
Stronger support across diverse accents

Both platforms support multilingual transcription, with AssemblyAI offering 99-language support through Universal-2.

Feature Gap Analysis

AssemblyAI includes built-in capabilities requiring separate integration work with Whisper:

Speaker diarization (automatic speaker identification)
Real-time streaming via WebSocket API
Sentiment analysis and content detection
Auto chapters (segmenting long audio)
PII redaction (removing sensitive information)
Custom vocabulary support

Cost Breakdown

Monthly Volume	AssemblyAI Cost	Whisper Infrastructure Cost
1,000 minutes	$2.50	~$50
10,000 minutes	$25	~$200
100,000 minutes	$250	~$800 + engineering

Hidden self-hosting expenses:

Initial setup: 40+ hours
Ongoing maintenance and security patches
Downtime risks when servers fail
Capacity planning for traffic spikes
DevOps expertise requirements

Implementation Complexity

AssemblyAI requires minimal code:

import assemblyai as aai
aai.settings.api_key = "your-api-key"
transcriber = aai.Transcriber()
config = aai.TranscriptionConfig(
    speech_models=["universal-3-pro", "universal-2"]
)
transcript = transcriber.transcribe("audio.mp3", config=config)
print(transcript.text)

Whisper setup involves:

Installing CUDA drivers for GPU acceleration
Downloading large model files (several gigabytes)
Python environment configuration
Managing VRAM requirements (10GB+ for large models)
Audio preprocessing implementation

When to Choose Each Platform

Choose AssemblyAI for:

Fast feature shipping
Real-time transcription needs
Advanced features (diarization, sentiment analysis)
Predictable costs
Compliance-heavy applications

Choose Whisper when:

Complete data control is required
Offline processing is necessary
Custom model modifications are needed
ML engineering resources are available

Frequently Asked Questions

Can both platforms be used together?
Yes, many developers use hybrid approaches where AssemblyAI handles real-time features while Whisper processes batch jobs.

How long does switching take?
Transitioning from Whisper to AssemblyAI typically requires days; switching away requires weeks of infrastructure work.

Which handles specialized terminology better?
AssemblyAI's custom vocabulary feature supports industry-specific terms more effectively, particularly in healthcare and legal domains.

Does AssemblyAI work offline?
No—it requires internet connectivity. Only Whisper offers completely offline operation.

How do model improvements work?
AssemblyAI automatically deploys improvements without breaking changes. Whisper requires manual testing and migration.

How to Build a Voice Agent with Python in 5 Minutes

Mart Schweiger — Wed, 08 Apr 2026 17:03:33 +0000

Overview

This tutorial demonstrates creating a complete voice agent that listens, thinks, and responds naturally using Python. The application processes speech in real-time, generates intelligent responses, and speaks back to users in under 100 lines of code.

The solution combines three APIs:

AssemblyAI's Universal-3 Pro Streaming for speech-to-text
OpenAI's GPT-4 for conversational AI
ElevenLabs for voice synthesis

Prerequisites

Requirements:

Python 3.9 or higher
Three API keys (AssemblyAI, OpenAI, ElevenLabs)
Computer with microphone and speakers

Installation

pip install "assemblyai>=1.0.0" openai "elevenlabs>=1.0.0" pyaudio python-dotenv

Package purposes:

assemblyai: Real-time speech recognition with Universal-3 Pro Streaming
openai: GPT model integration
elevenlabs: Natural voice synthesis
pyaudio: Microphone access
python-dotenv: API key management

Configuration

Create a .env file:

ASSEMBLYAI_API_KEY=your_assemblyai_key_here
OPENAI_API_KEY=your_openai_key_here
ELEVENLABS_API_KEY=your_elevenlabs_key_here

Voice Agent Components

A voice agent operates through three interconnected stages:

Component	Role	Why Streaming Matters
AssemblyAI	Speech-to-text	Transcribes audio as it arrives for faster LLM response
OpenAI GPT-4	Language model	Generates responses token-by-token for immediate TTS
ElevenLabs	Text-to-speech	Plays audio while generating remaining content

Streaming architecture eliminates delays that make conversations feel unnatural. Batch processing creates awkward pauses between user input and agent response.

Speech-to-Text Setup

Create voice_agent.py:

import assemblyai as aai
from assemblyai.streaming.v3 import (
    BeginEvent,
    StreamingClient,
    StreamingClientOptions,
    StreamingError,
    StreamingEvents,
    StreamingParameters,
    TurnEvent,
    TerminationEvent,
)
from dotenv import load_dotenv
import os

load_dotenv()

class VoiceAgent:
    def __init__(self):
        self.client = StreamingClient(
            StreamingClientOptions(
                api_key=os.getenv('ASSEMBLYAI_API_KEY'),
                api_host="streaming.assemblyai.com",
            )
        )

        self.client.on(StreamingEvents.Begin, self.on_begin)
        self.client.on(StreamingEvents.Turn, self.on_turn)
        self.client.on(StreamingEvents.Termination, self.on_terminated)
        self.client.on(StreamingEvents.Error, self.on_error)

        self.is_processing = False

    def on_begin(self, event: BeginEvent):
        print("Listening... Start speaking!")

    def on_turn(self, turn: TurnEvent):
        if not turn.transcript:
            return

        if turn.end_of_turn:
            print(f"You said: {turn.transcript}")
        else:
            print(f"Hearing: {turn.transcript}", end="\r")

    def on_error(self, error: StreamingError):
        print(f"Error: {error}")

    def on_terminated(self, event: TerminationEvent):
        print("Connection closed")

The system provides two transcript types: partial (incomplete sentences shown in real-time) and final (complete when punctuation-triggered pauses occur).

Language Model Integration

Add OpenAI processing to the VoiceAgent class:

from openai import OpenAI

class VoiceAgent:
    def __init__(self):
        # Previous code...

        self.openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

        self.conversation = [
            {"role": "system", "content": """You are a helpful voice assistant.

Keep responses short and conversational.

Talk like you're having a normal conversation with someone."""}
        ]

    def process_with_llm(self, user_text):
        self.conversation.append({"role": "user", "content": user_text})

        response_text = ""

        stream = self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=self.conversation,
            stream=True,
            temperature=0.7,
            max_tokens=150
        )

        print("Assistant: ", end="")

        for chunk in stream:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                response_text += content
                print(content, end="", flush=True)

        print()

        self.conversation.append({"role": "assistant", "content": response_text})

        self.speak(response_text)

Text-to-Speech Output

from elevenlabs.client import ElevenLabs
from elevenlabs import stream as play_stream
import threading

class VoiceAgent:
    def __init__(self):
        # Previous code...

        self.elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))
        self.voice_id = "EXAVITQu4vr4xnSDxMaL"  # Sarah voice

    def speak(self, text):
        def generate_and_play():
            try:
                audio_stream = self.elevenlabs_client.text_to_speech.stream(
                    voice_id=self.voice_id,
                    text=text,
                    model_id="eleven_turbo_v2_5",
                )

                play_stream(audio_stream)

            except Exception as e:
                print(f"Voice error: {e}")

        thread = threading.Thread(target=generate_and_play, daemon=True)
        thread.start()

Available voices:

Sarah (EXAVITQu4vr4xnSDxMaL): Clear, professional female
Josh (TxGEqnHWrfWFTfGW9XjX): Warm, friendly male
Elli (MF3mGyEYCl7XYWbV9V6O): Young, energetic female

Complete Implementation

import assemblyai as aai
import os
import sys
import threading

from assemblyai.streaming.v3 import (
    BeginEvent,
    StreamingClient,
    StreamingClientOptions,
    StreamingError,
    StreamingEvents,
    StreamingParameters,
    TurnEvent,
    TerminationEvent,
)

from elevenlabs.client import ElevenLabs
from elevenlabs import stream as play_stream
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

class VoiceAgent:

    def __init__(self):
        self.client = StreamingClient(
            StreamingClientOptions(
                api_key=os.getenv('ASSEMBLYAI_API_KEY'),
                api_host="streaming.assemblyai.com",
            )
        )

        self.client.on(StreamingEvents.Begin, self.on_begin)
        self.client.on(StreamingEvents.Turn, self.on_turn)
        self.client.on(StreamingEvents.Termination, self.on_terminated)
        self.client.on(StreamingEvents.Error, self.on_error)

        self.elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))
        self.openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

        self.is_processing = False
        self.voice_id = "EXAVITQu4vr4xnSDxMaL"

        self.conversation = [
            {"role": "system", "content": "You are a helpful voice assistant. Keep responses short and conversational."}
        ]

    def on_begin(self, event: BeginEvent):
        print("\n Voice Agent Ready! Start speaking...\n")

    def on_turn(self, turn: TurnEvent):
        if not turn.transcript:
            return

        if turn.end_of_turn:
            print(f"You: {turn.transcript}")

            if not self.is_processing:
                self.is_processing = True
                self.process_with_llm(turn.transcript)
                self.is_processing = False
        else:
            print(f"Listening: {turn.transcript}...", end="\r")

    def on_error(self, error: StreamingError):
        print(f"\n Error: {error}\n")

    def on_terminated(self, event: TerminationEvent):
        print("\n Voice Agent stopped\n")

    def process_with_llm(self, user_text):
        self.conversation.append({"role": "user", "content": user_text})

        response_text = ""

        stream = self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=self.conversation,
            stream=True,
            temperature=0.7,
            max_tokens=150
        )

        print("Agent: ", end="")

        for chunk in stream:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                response_text += content
                print(content, end="", flush=True)

        print()

        self.conversation.append({"role": "assistant", "content": response_text})

        self.speak(response_text)

    def speak(self, text):
        def generate_and_play():
            try:
                audio_stream = self.elevenlabs_client.text_to_speech.stream(
                    voice_id=self.voice_id,
                    text=text,
                    model_id="eleven_turbo_v2_5",
                )
                play_stream(audio_stream)
            except Exception as e:
                print(f"Voice error: {e}")

        voice_thread = threading.Thread(target=generate_and_play)
        voice_thread.daemon = True
        voice_thread.start()

    def start(self):
        self.client.connect(
            StreamingParameters(
                sample_rate=16000,
                speech_model="u3-rt-pro",
            )
        )

        try:
            self.client.stream(aai.extras.MicrophoneStream(sample_rate=16000))
        except KeyboardInterrupt:
            self.stop()

    def stop(self):
        print("\nStopping voice agent...")
        self.client.disconnect(terminate=True)
        sys.exit(0)

if __name__ == "__main__":
    agent = VoiceAgent()
    agent.start()

Running the Agent

python voice_agent.py

When "Voice Agent Ready! Start speaking..." appears, speak naturally into your microphone.

Test prompts:

"What's the weather like today?"
"Tell me a quick joke"
"Help me plan dinner"
"Explain how WiFi works simply"

Cost Estimate

Approximate hourly costs: $0.50–$1.00

AssemblyAI Universal-3 Pro Streaming: ~$0.45/hour
OpenAI GPT-4: ~$0.30/hour
ElevenLabs voice synthesis: ~$0.20/hour

Frequently Asked Questions

Do I need WebSocket knowledge?
No. The AssemblyAI Python SDK handles WebSocket connections, reconnection logic, and audio streaming protocols automatically.

Can I replace AssemblyAI with another STT service?
Technically possible, but you'd lose built-in punctuation-based turn detection and require manual WebSocket and audio streaming implementation.

Does this work with languages other than English?
Yes. Configure: StreamingParameters(speech_model="u3-rt-pro", sample_rate=16_000, prompt="Transcribe Spanish.").

Can I use this with frameworks like Pipecat or LiveKit?
Yes. AssemblyAI offers first-party integrations for Pipecat, LiveKit, Vapi, and Twilio.

Build an Agora Transcription Bot with AssemblyAI Universal-3 Pro

Mart Schweiger — Fri, 03 Apr 2026 14:09:47 +0000

Build an Agora Transcription Bot with AssemblyAI Universal-3 Pro

This tutorial walks through building a real-time transcription bot in Python that joins an Agora channel as a silent observer, captures each participant's audio as raw PCM frames, and streams it to AssemblyAI Universal-3 Pro Streaming for speaker-aware transcripts.

The full source is available at github.com/kelseyefoster/voice-agent-agora-universal-3-pro.

Why This Stack?

Agora's Python Server SDK lets a server-side bot join channels, subscribe to participant audio as raw PCM frames, and optionally publish audio back — without any browser or mobile client. This PCM stream format aligns directly with what AssemblyAI Universal-3 Pro Streaming expects, making the integration unusually clean.

Metric	AssemblyAI Universal-3 Pro	Agora Built-in STT
P50 latency	307ms	~600–900ms
Word Error Rate	8.9%	~14–18%
Speaker diarization	✅ Real-time	❌
Languages	99+	Limited

Prerequisites

Python 3.9+
Agora Console account — App ID and App Certificate
AssemblyAI API key

Quick Start

git clone https://github.com/kelseyefoster/voice-agent-agora-universal-3-pro
cd voice-agent-agora-universal-3-pro

pip install -r requirements.txt
cp .env.example .env
# Fill in AGORA_APP_ID, AGORA_APP_CERT, ASSEMBLYAI_API_KEY

python bot.py --channel my-channel

The bot joins the channel, opens one AssemblyAI WebSocket per participant, and prints completed turn transcripts to stdout. Press Ctrl+C to stop cleanly.

Environment Variables

AGORA_APP_ID=your_agora_app_id
AGORA_APP_CERT=your_agora_certificate
AGORA_CHANNEL=my-channel
AGORA_BOT_UID=9999
ASSEMBLYAI_API_KEY=your_assemblyai_api_key

How It Works

1. Join the channel as an audience bot

from agora.rtc.agora_service import AgoraService, AgoraServiceConfig
from agora.rtc.rtc_connection import RTCConnConfig
from agora.rtc.agora_base import ClientRoleType, ChannelProfileType, AudioScenarioType

cfg = AgoraServiceConfig()
cfg.appid = AGORA_APP_ID
cfg.enable_audio_processor = True
cfg.audio_scenario = AudioScenarioType.AUDIO_SCENARIO_CHORUS

service = AgoraService()
service.initialize(cfg)

conn_cfg = RTCConnConfig(
    client_role_type=ClientRoleType.CLIENT_ROLE_AUDIENCE,
    channel_profile=ChannelProfileType.CHANNEL_PROFILE_LIVE_BROADCASTING,
)
connection = service.create_rtc_connection(conn_cfg)
connection.connect(token, channel, str(bot_uid))

2. Configure 16 kHz audio output before subscribing

agora_channel = connection.get_local_user()

# Set BEFORE subscribe_all_audio — eliminates resampling
agora_channel.set_playback_audio_frame_before_mixing_parameters(
    num_of_channels=1,
    sample_rate=16000,
)
agora_channel.subscribe_all_audio()

Each PcmAudioFrame will contain 160 samples of 16-bit little-endian PCM at 16 kHz mono — exactly what AssemblyAI expects.

3. Open one AssemblyAI WebSocket per participant

AAI_WS_URL = (
    "wss://streaming.assemblyai.com/v3/ws"
    f"?sample_rate=16000"
    "&speech_model=u3-rt-pro"
    "&format_turns=true"
)

async def stream_participant(agora_channel, uid: int, api_key: str):
    headers = {"Authorization": api_key}
    async with websockets.connect(AAI_WS_URL, additional_headers=headers) as ws:
        begin = json.loads(await ws.recv())
        print(f"[uid={uid}] Session: {begin['id']}")

        async def send_audio():
            async for frame in agora_channel.get_audio_frames(uid):
                await ws.send(frame.data)

        async def recv_transcripts():
            async for message in ws:
                event = json.loads(message)
                if event["type"] == "Turn" and event.get("end_of_turn"):
                    print(f"[uid={uid}] {event['transcript']}")

        await asyncio.gather(send_audio(), recv_transcripts())

4. Track participants dynamically

active_streams: dict[int, asyncio.Task] = {}

def on_user_joined(uid: int):
    task = asyncio.create_task(stream_participant(agora_channel, uid, api_key))
    active_streams[uid] = task

def on_user_left(uid: int, reason: int):
    if uid in active_streams:
        active_streams[uid].cancel()
        del active_streams[uid]

connection.register_observer_callback("on_user_joined", on_user_joined)
connection.register_observer_callback("on_user_offline", on_user_left)

5. Terminate cleanly

async def close_stream(ws):
    await ws.send(json.dumps({"type": "Terminate"}))
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "Termination":
            print(f"Audio processed: {event['audio_duration_seconds']}s")
            break

Production Token Generation

pip install agora-token-builder

from agora_token_builder import RtcTokenBuilder, Role_Subscriber
import time

def generate_bot_token(app_id, app_cert, channel, uid):
    expire = int(time.time()) + 3600
    return RtcTokenBuilder.buildTokenWithUid(
        app_id, app_cert, channel, uid, Role_Subscriber, expire
    )

Extending the Bot

The end_of_turn transcript is a clean signal to drive downstream logic:

if event["type"] == "Turn" and event.get("end_of_turn"):
    transcript = event["transcript"]

    # Option A: send to an LLM
    await send_to_llm(uid, transcript)

    # Option B: store in a database
    await db.insert(uid=uid, text=transcript)

    # Option C: trigger a webhook
    await post_webhook({"uid": uid, "text": transcript})

Resources

Twilio Phone Agent with AssemblyAI Universal-3 Pro Streaming

Mart Schweiger — Fri, 03 Apr 2026 14:09:11 +0000

Twilio Phone Agent with AssemblyAI Universal-3 Pro Streaming

Build an AI phone agent that handles real calls using Twilio Voice + Media Streams and the AssemblyAI Universal-3 Pro Streaming model for real-time speech-to-text.

The key detail: Twilio streams 8kHz μ-law audio. AssemblyAI Universal-3 Pro accepts pcm_mulaw at sample_rate=8000 natively — no resampling, no format conversion.

Architecture

Incoming call
     │
  Twilio Voice
     │ TwiML → open WebSocket
     ▼
Your server (/media-stream WebSocket)
     │                        │
     │ mulaw 8kHz audio       │ synthesized mulaw audio
     ▼                        ▲
AssemblyAI Universal-3 Pro    ElevenLabs TTS
     │ transcript + turn signal
     ▼
  OpenAI GPT-4o

Prerequisites

Python 3.11+
AssemblyAI API key
Twilio account with a phone number
OpenAI API key
ElevenLabs API key
ngrok for local development

Quick Start

git clone https://github.com/kelseyefoster/voice-agent-twilio-universal-3-pro
cd voice-agent-twilio-universal-3-pro

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

uvicorn server:app --host 0.0.0.0 --port 8000
ngrok http 8000

Configure Twilio

Twilio Console > Phone Numbers > your number > Voice & Fax
Set A Call Comes In to Webhook: https://your-ngrok-url.ngrok.io/incoming-call
Call your Twilio number

AssemblyAI WebSocket Parameters for Twilio

ASSEMBLYAI_WS_URL = (
    "wss://streaming.assemblyai.com/v3/ws"
    "?speech_model=u3-rt-pro"
    "&encoding=pcm_mulaw"      # must match Twilio's audio format
    "&sample_rate=8000"        # must match Twilio's 8kHz stream
    "&end_of_turn_confidence_threshold=0.5"
    "&min_turn_silence=400"
)

Phone calls have more background noise than browser audio. The higher confidence threshold and longer min_turn_silence reduce false triggers.

Post-Call Transcription

import assemblyai as aai
transcriber = aai.Transcriber()
transcript = transcriber.transcribe(recording_url)
print(transcript.text)

Keyterm Prompting

ASSEMBLYAI_WS_URL += "&keyterms_prompt=YourBrand&keyterms_prompt=SpecialTerm"

Deploy

# Railway
railway login && railway init && railway up

# Render — Web Service with start command:
# uvicorn server:app --host 0.0.0.0 --port $PORT

Resources

Vapi Voice Agent with AssemblyAI Universal-3 Pro Streaming

Mart Schweiger — Fri, 03 Apr 2026 14:08:35 +0000

Vapi Voice Agent with AssemblyAI Universal-3 Pro Streaming

Use AssemblyAI Universal-3 Pro Streaming as the speech-to-text engine inside your Vapi voice agent — and get neural turn detection, keyterm prompting, and 307ms P50 latency inside Vapi's managed voice platform.

What is Vapi?

Vapi handles telephony, turn-taking, and orchestration so you don't have to. It supports 14+ speech-to-text providers. You bring your AssemblyAI key, Vapi handles the rest.

Setup: Add AssemblyAI to Vapi

Step 1 — Add Your API Key

Go to dashboard.vapi.ai
Navigate to Settings > Transcriber Providers
Add your AssemblyAI API key

Step 2 — Create an Assistant (Dashboard)

Click Create Assistant
Under Transcriber, select Assembly AI
Choose u3-rt-pro under Model
Save and test via the web call button

Step 3 — Create an Assistant (API)

git clone https://github.com/kelseyefoster/voice-agent-vapi-assemblyai
cd voice-agent-vapi-assemblyai

pip install -r requirements.txt
cp .env.example .env

python create_assistant.py create

This creates a fully configured assistant:

{
  "transcriber": {
    "provider": "assembly-ai",
    "model": "u3-rt-pro",
    "language": "en",
    "keytermsPrompt": ["YourBrand", "SpecialTerm"],
    "confidenceThreshold": 0.4
  }
}

Quick Start (Full)

# Create an assistant
python create_assistant.py create

# Make an outbound test call
python create_assistant.py call --assistant-id <id> --phone +1XXXXXXXXXX

# Start the webhook server
uvicorn webhook_server:app --port 8000

Keyterm Prompting

"keytermsPrompt": [
    "hemoglobin A1c",
    "HIPAA",
    "Jardiance",
    "deductible"
]

Up to 100 keyterms, 50 characters each. Changes take effect on the next call without restarting the assistant.

When to Choose AssemblyAI in Vapi

Use Case	Recommendation
Fastest streaming latency	AssemblyAI (307ms P50)
Account/serial codes	AssemblyAI (+21% fewer alphanumeric errors)
Medical terminology	AssemblyAI (keyterm prompting)
Interruption handling	AssemblyAI (punctuation-based turn detection)
Multilingual callers	AssemblyAI (native code switching)

Resources

Pipecat Voice Agent with AssemblyAI Universal-3 Pro Streaming

Mart Schweiger — Fri, 03 Apr 2026 14:08:00 +0000

Pipecat Voice Agent with AssemblyAI Universal-3 Pro Streaming

Build a real-time voice agent using Pipecat — Daily.co's open-source Voice AI framework — with the AssemblyAI Universal-3 Pro Streaming model as the speech-to-text engine.

Pipecat's modular pipeline design means you can swap any component without touching the rest. AssemblyAI has a first-party Pipecat plugin with full Universal-3 Pro Streaming support — no manual WebSocket wiring required.

Why AssemblyAI in Pipecat?

Metric	AssemblyAI Universal-3 Pro	Deepgram Nova-3
P50 latency	307 ms	516 ms
P99 latency	1,012 ms	1,907 ms
Word Error Rate	8.14%	9.87%
Neural turn detection	✅	❌ (VAD only)
Mid-session prompting	✅	❌
Anti-hallucination	✅	❌
Real-time diarization	✅	❌

The 41% latency advantage is noticeable in live conversation — and neural turn detection means fewer awkward double-responses when users pause mid-thought.

Prerequisites

Python 3.11+
AssemblyAI API key
Daily.co API key
OpenAI API key
Cartesia API key

Quick Start

git clone https://github.com/kelseyefoster/voice-agent-pipecat-universal-3-pro
cd voice-agent-pipecat-universal-3-pro

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

python create_room.py
python bot.py --url https://your-name.daily.co/your-room

Open the room URL in your browser and begin conversing.

Key Configuration Examples

Keyterm Prompting

stt = AssemblyAISTTService(
    connection_params=AssemblyAIConnectionParams(
        api_key=os.environ["ASSEMBLYAI_API_KEY"],
        speech_model="u3-rt-pro",
        keyterms_prompt=["AssemblyAI", "Universal-3", "Pipecat", "YourBrandName"],
    )
)

Supports up to 1,000 terms per session — particularly valuable for medical, legal, and financial domains.

Real-Time Speaker Diarization

connection_params=AssemblyAIConnectionParams(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    speech_model="u3-rt-pro",
    speaker_labels=True,
    max_speakers=2,
)

Multilingual Support

connection_params=AssemblyAIConnectionParams(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    speech_model="u3-rt-pro",
    language_detection=True,
)

Supported languages: English, Spanish, French, German, Italian, Portuguese.

Turn Detection Tuning

connection_params=AssemblyAIConnectionParams(
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
    speech_model="u3-rt-pro",
    end_of_turn_confidence_threshold=0.7,
    min_end_of_turn_silence_when_confident=300,
    max_turn_silence=1000,
)

Deploy to PipecatCloud

pip install pipecatcloud
pcc auth login
pcc init
pcc secrets set my-agent-secrets --file .env
pcc deploy

Resources

Daily.co Voice Agent with AssemblyAI Universal-3 Pro Streaming

Mart Schweiger — Fri, 03 Apr 2026 14:07:24 +0000

Daily.co Voice Agent with AssemblyAI Universal-3 Pro Streaming

Build a WebRTC voice agent using Daily.co for real-time audio transport and the AssemblyAI Universal-3 Pro Streaming model for speech-to-text — without Pipecat.

This is the bare-metal Daily.co integration. It shows exactly how Daily's audio tracks connect to the AssemblyAI WebSocket — useful when you want to embed a voice agent into a custom Daily.co application without pulling in a full pipeline framework.

Architecture

Browser / Phone (Daily.co room participant)
        │ WebRTC audio
        ▼
  Daily.co room
        │ PCM audio via daily-python SDK
        ▼
  This bot (daily-python)
        │ raw PCM bytes
        ▼
  AssemblyAI Universal-3 Pro WebSocket
        │ transcript + neural turn signal
        ▼
  OpenAI GPT-4o → Cartesia TTS → PCM audio
        ▼
  Bot sends audio back into Daily room

Prerequisites

Quick Start

git clone https://github.com/kelseyefoster/voice-agent-dailyco-universal-3-pro
cd voice-agent-dailyco-universal-3-pro

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env
# Edit .env with your API keys

# Create a room and bot token
python create_room.py

# Start the bot
python bot.py --room-url https://yourname.daily.co/room --token <bot-token>

Open the room URL in your browser and start speaking.

How Audio Flows

Daily.co calls on_audio_data whenever a remote participant speaks. The bot forwards raw PCM bytes directly to the AssemblyAI WebSocket — no conversion needed at 16 kHz:

def on_audio_data(self, participant_id, audio_data, sample_rate, num_channels):
    if self.aai_ws and not self.aai_ws.closed:
        asyncio.create_task(self.aai_ws.send(audio_data))

When Universal-3 Pro detects an end-of-turn, the bot generates a response with GPT-4o, synthesizes audio with Cartesia, and injects it back into the room.

AssemblyAI Connection Parameters

AAI_WS_URL = (
    "wss://streaming.assemblyai.com/v3/ws"
    "?speech_model=u3-rt-pro"
    "&encoding=pcm_s16le"
    f"&sample_rate={SAMPLE_RATE}"
    "&end_of_turn_confidence_threshold=0.4"
    "&min_turn_silence=300"
    f"&token={ASSEMBLYAI_API_KEY}"
)

Why Direct Daily.co Instead of Pipecat?

Pipecat excels for production voice agents with complex pipelines (VAD, context management, interruption handling). This tutorial targets those who want bare-metal access to primitives — useful when embedding a voice agent into a custom Daily.co application without full framework overhead.