AI-agenten · Les

Latency optimaliseren voor voice-agents

Streaming-TTS, responses opdelen in chunks en de latency tot het eerste woord minimaliseren.

Les 4 van 413 stappen

Latency optimaliseren voor voice-agents is een gratis AI-agenten-les op CoddyKit. Dit is les 4 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject AI-agenten. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus AI-agenten bevat in totaal 4 lessen.

Waarom vertraging belangrijk is voor spraak

In tekstchat is een vertraging van 3 seconden acceptabel. In een stemgesprek voelt alles boven 1,5 seconden onnatuurlijk aan en wordt de gespreksstroom verstoord.

De vertraging van spraak bestaat uit drie hoofdonderdelen: transcriptietijd (STT), verwerkingstijd van het LLM (TTFT + generatie) en TTS-synthesetijd. Door elk onderdeel te optimaliseren, wordt de gebruikerservaring samen aanzienlijk beter.

Het vertragingsbudget meten

Meet elk onderdeel voordat je gaat optimaliseren. Voeg tijdmetingen aan de lus toe, zodat je weet waar de tijd daadwerkelijk aan wordt besteed.

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: afspelen tijdens het genereren

De grootste afzonderlijke verbetering voor de vertraging is streaming-TTS. Wacht niet tot de volledige audio is gesynthetiseerd, maar begin de eerste audiobrok binnen ongeveer 200 ms af te spelen terwijl de rest tegelijkertijd wordt gegenereerd.

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()

Streaming-TTS per zin

De meest praktische streamingaanpak is om het antwoord van het LLM op te splitsen in zinnen en deze één voor één te synthetiseren en af te spelen. De eerste zin begint binnen ongeveer 300 ms met afspelen.

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: tijd tot het eerste token optimaliseren

TTFT (Time to First Token) is de vertraging tussen het verzenden van het LLM-verzoek en het ontvangen van het eerste token van het antwoord. Door TTFT te optimaliseren horen gebruikers het begin van het antwoord sneller.

Gebruik streamingantwoorden van het LLM en stuur tokens naar TTS zodra een zinsgrens wordt gedetecteerd.

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)

Veelgebruikte antwoorden vooraf genereren

Sommige antwoorden van een agent zijn voorspelbaar: begroetingen, foutmeldingen en aanduidingen dat de agent nadenkt, zoals "Laat me dat voor je controleren." Genereer de audio hiervan vooraf bij het opstarten, zodat deze direct en zonder vertraging kan worden afgespeeld.

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')

Vertraging van lokale TTS versus cloud-TTS

Cloud-TTS voegt tijd toe voor de heen-en-terugreis over het netwerk, ongeveer 100 tot 300 ms. Voor korte antwoorden of veelgebruikte zinnen kan een lokale TTS-engine sneller zijn, ten koste van de stemkwaliteit.

pyttsx3 is de eenvoudigste offline-TTS-oplossing voor Python.

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

Een sneller LLM-model kiezen

De modelkeuze heeft grote invloed op TTFT. GPT-4o-mini is voor de meeste taken van stemagenten 5 tot 10 keer sneller dan GPT-4o. Gebruik het kleinste model dat aan de kwaliteitseisen voldoet.

# 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}"')

Antwoorden cachen voor herhaalde vragen

Gebruikers stellen vaak herhaaldelijk dezelfde of vergelijkbare vragen. Cache reacties van de LLM en TTS voor identieke vragen, zodat je ze bij herhaling direct kunt leveren.

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

Doel voor latentie van begin tot eind

Met alle optimalisaties op hun plaats hoort een goed gebouwde spraakagent minder dan 1 seconde nodig te hebben vanaf het einde van de spraak van de gebruiker tot het begin van het antwoord van de agent.

# 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')

Asynchrone transcriptie overlappen

Begin onmiddellijk met het opnemen van de volgende invoer van de gebruiker terwijl de agent het antwoord genereert en uitspreekt. Door de fasen van de verwerkingsketen te overlappen, verklein je de tijd zonder activiteit tussen beurten.

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()

Kennischeck

Welke enkele optimalisatie zorgt doorgaans voor de grootste vermindering van latentie bij spraakagents?

Beantwoord dit in de interactieve les

Samenvatting: latentieoptimalisatie voor spraakagents

Stack voor latentieoptimalisatie van spraak: LLM-antwoorden streamen (start TTS bij zinsgrenzen zodra tokens binnenkomen), TTS-weergave per zin (speel af tijdens het synthetiseren), TTS-caching (direct opnieuw afspelen van herhaalde zinnen), veelgebruikte antwoorden vooraf genereren (begroetingen, opvulzinnen) en een snel LLM-model kiezen (gpt-4o-mini voor eenvoudige vragen).

Doel: minder dan 1 seconde vanaf het einde van de spraak van de gebruiker tot de eerste audiobyte. Meet elk onderdeel — transcriptie is vaak 40% van de totale latentie. Lokale TTS ruilt kwaliteit in voor snelheid; cloud-TTS met streaming per zin biedt voor de meeste agents de beste balans.

Gratis beginnen

Leer AI-agenten met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen: 60
Lessen: 239

Gratis beginnen met leren Bekijk de volledige cursus

Veelgestelde vragen

Is de les “Latency optimaliseren voor voice-agents” gratis?

Ja — de volledige tekst van “Latency optimaliseren voor voice-agents” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus AI-agenten wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus AI-agenten bevat in totaal 4 lessen.

Wat leer ik in “Latency optimaliseren voor voice-agents”?

Streaming-TTS, responses opdelen in chunks en de latency tot het eerste woord minimaliseren. Je oefent met AI-agenten door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met AI-agenten te beginnen?

Ervaring vooraf is niet nodig. AI-agenten op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 4 van 4.

Hoe lang duurt de les “Latency optimaliseren voor voice-agents”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over AI-agenten?

Ja. Elke les over AI-agenten bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

← Terug naar AI-agenten