Latency Optimization for Voice Agents

Why Latency Matters for Voice

Measuring the Latency Budget

import time

def voice_loop_timed():
    timing = {}

    # 1. Record
    t0 = time.perf_counter()
    audio, sr = record_until_silence()
    timing['record'] = time.perf_counter() - t0

    # 2. Transcribe
    t0 = time.perf_counter()
    audio_path = audio_to_file(audio, sr)
    user_text = transcribe_file(audio_path)
    timing['transcription'] = time.perf_counter() - t0

    # 3. LLM
    t0 = time.perf_counter()
    agent_text = agent.respond(user_text)
    timing['llm'] = time.perf_counter() - t0

    # 4. TTS
    t0 = time.perf_counter()
    say(agent_text)
    timing['tts'] = time.perf_counter() - t0

    print('Latency breakdown:')
    total = sum(timing.values())
    for step, duration in timing.items():
        print(f'  {step:15} {duration*1000:.0f}ms ({duration/total*100:.0f}%)')

Streaming TTS: Play While Generating

import threading
import queue
import sounddevice as sd
import numpy as np
import io

def stream_tts_and_play(text, voice='nova'):
    audio_queue = queue.Queue()

    def synthesize():
        from openai import OpenAI
        import os
        client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
        with client.audio.speech.with_streaming_response.create(
            model='tts-1', voice=voice, input=text
        ) as response:
            for chunk in response.iter_bytes(chunk_size=4096):
                audio_queue.put(chunk)
        audio_queue.put(None)  # sentinel

    synth_thread = threading.Thread(target=synthesize, daemon=True)
    synth_thread.start()

    # Collect and play (buffering first 2 chunks for smooth start)
    audio_buffer = b''
    min_buffer = 8192
    import pygame
    pygame.mixer.init()

    while True:
        chunk = audio_queue.get()
        if chunk is None:
            break
        audio_buffer += chunk
        if len(audio_buffer) >= min_buffer:
            # play buffer ...
            pass  # simplified — real impl streams to sounddevice

    synth_thread.join()

Sentence-by-Sentence Streaming TTS

import re
import concurrent.futures

def split_sentences(text):
    return [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]

def tts_and_play_streaming(text, voice='nova'):
    sentences = split_sentences(text)
    if not sentences:
        return

    # Prefetch next sentence while current is playing
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)

    # Kick off synthesis of first sentence
    futures = []
    for sentence in sentences:
        futures.append(executor.submit(cached_tts, sentence, voice))

    # Play each sentence as soon as it's ready
    for future in futures:
        audio_path = future.result(timeout=10)
        play_audio_file(audio_path)

    executor.shutdown(wait=False)

# Key insight: while sentence 1 plays (~2 seconds), sentence 2 is being synthesized in parallel

TTFT: Time to First Token Optimization

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def stream_llm_to_tts(user_text, conversation_history, voice='nova'):
    buffer = ''

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=conversation_history + [{'role': 'user', 'content': user_text}],
        stream=True  # streaming enabled
    )

    for chunk in stream:
        delta = chunk.choices[0].delta.content or ''
        buffer += delta

        # Check if a sentence is complete
        if buffer.endswith(('.', '!', '?')) and len(buffer) > 20:
            sentence = buffer.strip()
            print(f'Queuing TTS: {sentence[:50]}')
            # Synthesize and play immediately (non-blocking)
            audio_path = cached_tts(sentence, voice)
            play_audio_file(audio_path)
            buffer = ''

    # Flush remaining buffer
    if buffer.strip():
        audio_path = cached_tts(buffer.strip(), voice)
        play_audio_file(audio_path)

Pre-Generating Common Responses

import os

PRE_GENERATED = {
    'greeting':      'Hello! How can I help you today?',
    'thinking':      'Let me look that up for you.',
    'not_found':     'I could not find information on that. Could you rephrase?',
    'error':         'Sorry, something went wrong. Please try again.',
    'goodbye':       'Goodbye! Have a great day.',
    'clarify':       'Could you give me a bit more detail?',
    'working_on_it': 'Working on it, this may take a moment.'
}

pre_gen_cache = {}

def prewarm_responses(voice='nova'):
    for key, text in PRE_GENERATED.items():
        audio_path = cached_tts(text, voice=voice)
        pre_gen_cache[key] = audio_path
    print(f'Pre-generated {len(pre_gen_cache)} common responses')

def instant_response(key):
    path = pre_gen_cache.get(key)
    if path:
        play_audio_file(path)
    else:
        say(PRE_GENERATED.get(key, ''))

# Usage: while LLM is thinking, play a stall message instantly
instant_response('thinking')

Local TTS vs Cloud TTS Latency

import pyttsx3
import time

# Local TTS with pyttsx3
def local_tts(text, voice_index=0, rate=175):
    engine = pyttsx3.init()
    voices = engine.getProperty('voices')
    if voice_index < len(voices):
        engine.setProperty('voice', voices[voice_index].id)
    engine.setProperty('rate', rate)  # words per minute
    t0 = time.perf_counter()
    engine.say(text)
    engine.runAndWait()
    print(f'Local TTS latency: {(time.perf_counter()-t0)*1000:.0f}ms')

# Comparison (rough benchmarks):
# pyttsx3 (local):     ~50ms start, robotic quality
# OpenAI TTS-1:        ~200-400ms, good quality
# ElevenLabs turbo:    ~150-250ms, excellent quality
# OpenAI TTS-1-hd:     ~400-800ms, best quality

# Recommendation:
# Real-time voice agent -> OpenAI TTS-1 or ElevenLabs turbo
# Quality recording     -> OpenAI TTS-1-hd or ElevenLabs standard

Choosing a Faster LLM Model

# Model latency comparison (rough benchmarks for voice agent use case):
MODEL_BENCHMARKS = {
    'gpt-4o-mini':   {'ttft_ms': 300,  'quality': 'good',      'cost': 'very low'},
    'gpt-4o':        {'ttft_ms': 800,  'quality': 'excellent',  'cost': 'medium'},
    'claude-haiku':  {'ttft_ms': 250,  'quality': 'good',       'cost': 'very low'},
    'claude-sonnet': {'ttft_ms': 600,  'quality': 'excellent',  'cost': 'medium'},
    'llama3-8b':     {'ttft_ms': 100,  'quality': 'decent',     'cost': 'free (local)'},
}

# Strategy: use fast model for simple factual queries,
# fall back to powerful model for complex reasoning
def select_model(question):
    # Short, simple questions -> fast model
    if len(question.split()) < 15:
        return 'gpt-4o-mini'
    # Complex, multi-step -> better model
    if any(kw in question.lower() for kw in ['analyze', 'compare', 'explain why', 'write a']):
        return 'gpt-4o'
    return 'gpt-4o-mini'

if __name__ == '__main__':
    for q in ['What time is it in Tokyo?', 'Analyze why sales dropped last quarter and compare to competitors']:
        print(f'{select_model(q)!r} chosen for: "{q}"')

Response Caching for Repeat Questions

import hashlib
import json

RESPONSE_CACHE = {}  # in production: use Redis with TTL

def get_cache_key(user_text):
    # Normalize: lowercase, strip punctuation
    import re
    normalized = re.sub(r'[^a-z0-9 ]', '', user_text.lower()).strip()
    return hashlib.md5(normalized.encode()).hexdigest()

def cached_agent_respond(user_text, voice='nova'):
    key = get_cache_key(user_text)

    if key in RESPONSE_CACHE:
        print('Response cache hit!')
        entry = RESPONSE_CACHE[key]
        play_audio_file(entry['audio_path'])
        return entry['text']

    # Cache miss
    text = agent.respond(user_text)
    audio_path = cached_tts(text, voice=voice)

    RESPONSE_CACHE[key] = {'text': text, 'audio_path': audio_path}
    play_audio_file(audio_path)
    return text

End-to-End Latency Target

# Target latency budget breakdown (1000ms total):

LATENCY_BUDGET = {
    'silence_detection_end':   0,    # user stops speaking
    'audio_processing':        50,   # RMS check, noise gate
    'transcription_whisper':   400,  # STT API call
    'llm_ttft':                300,  # time to first token (gpt-4o-mini)
    'tts_first_sentence':      200,  # first sentence synthesized
    'playback_start':          1000  # TOTAL: user hears response within 1 second
}

# Optimizations applied:
# - Streaming LLM responses (saves 200-500ms vs waiting for full response)
# - Sentence-by-sentence TTS (play while rest generates)
# - gpt-4o-mini instead of gpt-4o (saves 500ms TTFT)
# - TTS cache for common phrases (saves 200ms on cached phrases)
# - Pre-generated stall messages ('Let me check...' plays instantly)

print('Target: < 1000ms from speech end to first audio byte played')
print('Typical with optimizations: 600-900ms')

Async Transcription Overlap

import threading
import time

class PipelineOverlapAgent:
    def __init__(self):
        self.next_audio = None
        self.recording_thread = None

    def start_background_recording(self):
        def record():
            self.next_audio = record_until_silence()
        self.recording_thread = threading.Thread(target=record, daemon=True)
        self.recording_thread.start()

    def get_recorded_audio(self, timeout=30):
        if self.recording_thread:
            self.recording_thread.join(timeout=timeout)
        audio = self.next_audio
        self.next_audio = None
        return audio

    def conversation_turn(self, audio, sr):
        # Transcribe and run agent
        text = transcribe_file(audio_to_file(audio, sr))
        if not text.strip():
            return

        # Start recording NEXT turn BEFORE speaking
        # (user can start speaking as soon as agent starts)
        response = agent.respond(text)

        # Start next recording while speaking
        self.start_background_recording()

        # Speak current response
        speak_with_interrupt(response)

        # Next audio is already being captured in background
        return self.get_recorded_audio()

Knowledge Check

Recap: Latency Optimization for Voice Agents