Chapter 13: Audio Customization

Your agent talks, listens, and takes action. But it sounds like every other voice agent out there: the same latency on the first greeting, the same flat delivery of abbreviations, the same inability to whisper or pause for effect. The difference between a demo and a production-quality agent comes down to audio polish. This chapter walks through pre-synthesized audio for instant playback, TTS caching to avoid redundant synthesis, custom text replacement before synthesis, SSML markup for fine-grained speech control, and volume scaling for consistent output levels.

Pre-Synthesized Audio for Instant Playback

The most noticeable latency in a voice agent is the greeting. The caller connects, there is a brief silence while the LLM generates a response and the TTS synthesizes it, and then the agent finally speaks. That delay (typically 500ms to 1.5 seconds) makes the agent feel sluggish before the conversation even starts.

The fix is to synthesize your greeting before the call arrives. At agent startup, you send the greeting text through your TTS provider, collect the resulting audio frames, and store them. When a caller connects, you play the cached audio immediately with no round-trip to the LLM or TTS service.

from livekit import rtc
from livekit.agents import Agent, AgentSession, ChatContext
from livekit.plugins import cartesia

GREETING = "Hi there, thanks for calling! How can I help you today?"

class ReceptionistAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a friendly receptionist.",
        )
        self._greeting_audio: list[rtc.AudioFrame] = []

    async def on_enter(self):
        # Synthesize the greeting and cache the frames
        tts = cartesia.TTS(model="sonic-2")
        frames = []
        async for event in tts.synthesize(GREETING):
            frames.append(event.frame)
        self._greeting_audio = frames

        # Play instantly when the session starts
        await self.session.say(
            GREETING,
            audio=self._generate_cached_audio(),
        )

    async def _generate_cached_audio(self):
        for frame in self._greeting_audio:
            yield frame

The session.say() method accepts an audio parameter: an async generator of audio frames. When you provide pre-synthesized frames, the framework skips TTS entirely and streams those frames directly to the caller. The text parameter is still required so the LLM knows what was said and can maintain conversation context.

This pattern works for any predictable utterance: greetings, hold messages, transfer announcements, or closing statements. If you know what the agent will say before the caller speaks, pre-synthesize it.

TTS Caching Strategies

Pre-synthesizing a single greeting is useful. But many agents repeat the same phrases across hundreds of calls: “Let me look that up for you,” “Can you spell that for me?”, “I’ve got you booked, you’ll receive a confirmation shortly.” Synthesizing these every time wastes money and adds latency.

A simple in-memory cache keyed by the text string solves this:

from collections import OrderedDict
from livekit import rtc
from livekit.agents import Agent

class CachedTTSMixin:
    """Mixin that caches TTS output frames keyed by text."""

    def __init__(self, *args, max_cache_size: int = 100, **kwargs):
        super().__init__(*args, **kwargs)
        self._tts_cache: OrderedDict[str, list[rtc.AudioFrame]] = OrderedDict()
        self._max_cache_size = max_cache_size

    async def say_cached(self, text: str):
        if text in self._tts_cache:
            self._tts_cache.move_to_end(text)
            frames = self._tts_cache[text]
            await self.session.say(text, audio=self._iter_frames(frames))
            return

        # Synthesize and cache
        frames = []
        async for event in self.session.tts.synthesize(text):
            frames.append(event.frame)

        if len(self._tts_cache) >= self._max_cache_size:
            self._tts_cache.popitem(last=False)  # evict oldest
        self._tts_cache[text] = frames

        await self.session.say(text, audio=self._iter_frames(frames))

    async def _iter_frames(self, frames: list[rtc.AudioFrame]):
        for frame in frames:
            yield frame


class ReceptionistAgent(CachedTTSMixin, Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a friendly receptionist.",
            max_cache_size=50,
        )

The OrderedDict acts as an LRU cache. When a phrase is reused, it moves to the end. When the cache is full, the least recently used phrase is evicted. For agents handling high call volumes, this can reduce TTS costs significantly. Common phrases are synthesized once and reused for the lifetime of the process.

For persistence across restarts, serialize the frames to disk or Redis. Audio frames are just byte buffers with metadata (sample rate, channels, samples per channel), so they serialize cleanly.

Custom tts_node for Text Replacement

Sometimes the LLM generates text that sounds wrong when spoken aloud. “LOL” should not be pronounced as a word. “Dr.” should be “Doctor.” “3pm” should be “3 P.M.” The LLM does not know how the TTS will handle these, and TTS providers are inconsistent about abbreviation expansion.

The solution is to override the tts_node in your agent pipeline. This node sits between the LLM output and TTS input, giving you a chance to transform the text before synthesis:

import re
from livekit.agents import Agent, tts as tts_module

# Replacement rules: pattern -> spoken form
TEXT_REPLACEMENTS = {
    r"\bLOL\b": "ha ha ha",
    r"\bOMG\b": "oh my god",
    r"\bASAP\b": "as soon as possible",
    r"\bDr\.": "Doctor",
    r"\bMr\.": "Mister",
    r"\bMrs\.": "Missus",
    r"\bSt\.": "Street",
    r"\b(\d{1,2})(am|pm)\b": r"\1 \2",
    r"\b(\d{1,2}):(\d{2})(am|pm)\b": r"\1 \2 \3",
}


class ReceptionistAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a friendly receptionist.",
        )

    async def tts_node(
        self,
        text: AsyncIterable[str],
        model_settings: ModelSettings,
    ) -> AsyncIterable[tts_module.SynthesizedAudio]:
        async def transform_text():
            async for chunk in text:
                transformed = chunk
                for pattern, replacement in TEXT_REPLACEMENTS.items():
                    transformed = re.sub(
                        pattern, replacement, transformed, flags=re.IGNORECASE
                    )
                yield transformed

        # Pass transformed text to the default TTS pipeline
        async for audio in Agent.tts_node(self, transform_text(), model_settings):
            yield audio

The tts_node receives an async iterable of text chunks from the LLM and must yield SynthesizedAudio objects. By wrapping the input text stream with a transformation function and passing it to the parent implementation, you get text replacement without reimplementing the TTS pipeline.

This pattern is also useful for profanity filtering, brand name pronunciation fixes, or injecting SSML tags around specific words.

SSML Support

SSML (Speech Synthesis Markup Language) gives you fine-grained control over how text is spoken: pronunciation, pacing, emphasis, and pauses. Support varies across TTS providers, but the following tags are widely supported, though with provider-specific limitations.

Core SSML Tags

<break>: Insert a pause. Use this between sentences for natural pacing or before important information to create emphasis through silence.

Your appointment is confirmed. <break time="500ms"/> Your reference number is
<say-as interpret-as="characters">A B C 1 2 3</say-as>.

<say-as>: Control interpretation of text. Critical for phone numbers, dates, and codes that should be spelled out rather than read as words.

Your confirmation code is <say-as interpret-as="characters">XJ47</say-as>.
Call us at <say-as interpret-as="telephone">+14155551234</say-as>.

<phoneme>: Specify exact pronunciation using IPA (International Phonetic Alphabet). Use this for brand names, technical terms, or words the TTS consistently mispronounces.

Welcome to <phoneme alphabet="ipa" ph="vɔɪs.klɔː">VoiceClaw</phoneme>.

<prosody>: Adjust rate, pitch, and volume. Useful for reading important information slowly or adding emphasis.

<prosody rate="slow">Your total is forty-seven dollars and fifty cents.</prosody>
<prosody pitch="+10%">Great news!</prosody>

Provider-Specific Notes

Cartesia supports <break> and <phoneme> natively. For <say-as>, you may need to spell out characters manually in the text rather than relying on the tag. Cartesia also supports emotion and speed control through voice settings rather than SSML.

ElevenLabs supports <break>, <phoneme>, and <prosody>. It does not support <say-as> in all modes, so test with your specific model. ElevenLabs also offers pronunciation dictionaries as an alternative to inline <phoneme> tags.

Rime has strong SSML support across all four tags. It is often the most predictable provider for SSML-heavy use cases.

Injecting SSML in the TTS Node

You can combine the text replacement pattern from the previous section with SSML injection:

import re
from livekit.agents import Agent, tts as tts_module

SSML_RULES = {
    # Spell out anything that looks like a reference code
    r"\b([A-Z]{2,}\d{2,})\b": r'<say-as interpret-as="characters">\1</say-as>',
    # Add a pause before prices
    r"(\$[\d,.]+)": r'<break time="200ms"/>\1',
    # Slow down phone numbers
    r"(\+?\d[\d\s\-]{8,})": r'<prosody rate="slow">\1</prosody>',
}


class ReceptionistAgent(Agent):
    async def tts_node(self, text, model_settings):
        async def inject_ssml():
            async for chunk in text:
                result = chunk
                for pattern, replacement in SSML_RULES.items():
                    result = re.sub(pattern, replacement, result)
                yield result

        async for audio in Agent.tts_node(self, inject_ssml(), model_settings):
            yield audio

Always test SSML modifications with your specific TTS provider and model. Tags that work in one provider’s documentation may be silently ignored or produce unexpected results in practice.

Volume Control

Volume inconsistency is a subtle but real problem. Different TTS providers output at different levels. Callers on speakerphone in a noisy truck need louder audio than someone on a headset. And some synthesized phrases are naturally quieter than others.

Audio frames in LiveKit are PCM data: arrays of 16-bit signed integers. Adjusting volume is multiplying each sample by a scaling factor:

import struct
from livekit import rtc
from livekit.agents import Agent, tts as tts_module

def scale_audio_frame(frame: rtc.AudioFrame, volume: float) -> rtc.AudioFrame:
    """Scale audio frame volume. 1.0 = unchanged, 0.5 = half, 2.0 = double."""
    data = frame.data
    samples = struct.unpack(f"<{len(data) // 2}h", data)
    scaled = []
    for sample in samples:
        new_val = int(sample * volume)
        new_val = max(-32768, min(32767, new_val))  # clamp to 16-bit range
        scaled.append(new_val)
    new_data = struct.pack(f"<{len(scaled)}h", *scaled)
    return rtc.AudioFrame(
        data=new_data,
        sample_rate=frame.sample_rate,
        num_channels=frame.num_channels,
        samples_per_channel=frame.samples_per_channel,
    )

Applying Volume in tts_node

To adjust volume for all TTS output, override tts_node and scale every frame:

class ReceptionistAgent(Agent):
    def __init__(self, volume: float = 1.0):
        super().__init__(instructions="You are a friendly receptionist.")
        self._volume = volume

    async def tts_node(self, text, model_settings):
        async for event in Agent.tts_node(self, text, model_settings):
            if self._volume != 1.0:
                event.frame = scale_audio_frame(event.frame, self._volume)
            yield event

Applying Volume in realtime_audio_output_node

For agents using a realtime model (like GPT-4o Realtime), audio does not flow through tts_node. Instead, it comes through realtime_audio_output_node. The same scaling approach applies:

class RealtimeAgent(Agent):
    def __init__(self, volume: float = 1.2):
        super().__init__(instructions="You are a friendly receptionist.")
        self._volume = volume

    async def realtime_audio_output_node(self, audio, model_settings):
        async for frame in Agent.realtime_audio_output_node(
            self, audio, model_settings
        ):
            if self._volume != 1.0:
                frame = scale_audio_frame(frame, self._volume)
            yield frame

A common production pattern is to set a baseline volume boost (1.2x to 1.5x works well for phone calls) and expose a per-call adjustment through session metadata. When a caller says “I can’t hear you,” the agent can increase its volume mid-call by updating the scaling factor.

Be careful with values above 2.0. Aggressive scaling causes clipping. Samples that exceed the 16-bit range get clamped, producing distortion. If you need significantly louder audio, it is better to adjust the TTS provider’s output settings or use audio normalization rather than brute-force scaling.

Bringing It Together

These techniques compose naturally. A production agent might pre-synthesize its greeting, cache common phrases, run text through abbreviation expansion and SSML injection in tts_node, and apply a volume boost on the output. Each layer is independent, so you can adopt them incrementally as your agent’s requirements grow. Start with pre-synthesized greetings for immediate impact, add caching when your TTS costs become noticeable, and layer in SSML and volume control when you need that extra level of polish.