how can i have stream generator for wav files to play in the background audio player?

Thanks for reaching out, Sebastián. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

in Python Agents, the WAV files are:

Audio: pcm_s16le ([1][0][0][0] / 0x0001), 48000 Hz, mono, s16, 768 kb/s

i want to do this because im ocassionally seeing stream timeout on the audio mixer when using the path to the files

Thanks — that's helpful. Below is a concise, ready-to-use Python example that: • Reads a 48 kHz, mono, 16-bit PCM WAV file (

pcm_s16le

) in small chunks, • Yields those chunks as

rtc.AudioFrame

objects via an async generator, • Works with

BackgroundAudioPlayer.play()

to avoid relying on passing the file path (reduces mixer timeout risk). Adjust

FRAME_SAMPLES

if you need different frame timing.

import wave
import asyncio
from typing import AsyncIterator
import rtc
from livekit.agents import BackgroundAudioPlayer, AudioConfig

SAMPLE_RATE = 48000
NUM_CHANNELS = 1
SAMPLE_WIDTH_BYTES = 2  # 16-bit = 2 bytes
FRAME_MS = 20
FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_MS / 1000)  # 960 samples for 20ms

async def wav_file_stream(path: str) -> AsyncIterator[rtc.AudioFrame]:
    # open WAV with wave (expects PCM s16le)
    with wave.open(path, "rb") as wf:
        assert wf.getframerate() == SAMPLE_RATE
        assert wf.getnchannels() == NUM_CHANNELS
        assert wf.getsampwidth() == SAMPLE_WIDTH_BYTES

        while True:
            raw = wf.readframes(FRAME_SAMPLES)
            if not raw:
                break

            frame = rtc.AudioFrame(
                data=raw,
                sample_rate=SAMPLE_RATE,
                num_channels=NUM_CHANNELS,
                samples_per_channel=min(FRAME_SAMPLES, len(raw) // (SAMPLE_WIDTH_BYTES * NUM_CHANNELS)),
            )
            yield frame

            # keep timing consistent with real-time playback
            await asyncio.sleep(FRAME_MS / 1000.0)

async def play_wav_via_stream(path: str, volume: float = 1.0):
    player = BackgroundAudioPlayer()
    await player.play(AudioConfig(source=wav_file_stream(path), volume=volume))

# Example usage:
# await play_wav_via_stream("/path/to/file.wav", volume=0.8)

Notes: • The generator yields 20 ms frames (960 samples) which matches common WebRTC frame sizing; change

FRAME_MS

if required. •

await asyncio.sleep(...)

maintains real-time pacing so the mixer doesn't detect a stalled stream. • If your files vary, add defensive checks and handle resampling or channel conversion before yielding frames. Would you like a version that: • streams variable-length files continuously (looping), • removes the sleep and uses timestamps instead, or • performs on-the-fly resampling if a file isn’t exactly 48 kHz/mono?

i wanted on prewarm, to create a global iterator that returns frames to be consumed by the background audio player

Thanks — I’m responding to your request for a global prewarmed async iterator that yields 48 kHz mono 16-bit PCM frames for BackgroundAudioPlayer. Below is a corrected, safe pattern to prewarm WAV files into memory and expose a per-consumer async iterator. It: • Preloads files on prewarm (so playback avoids file-path timeouts), • Yields 20 ms

rtc.AudioFrame

chunks with real-time pacing, • Gives each consumer its own iterator state (call

.frames()

per player), • Handles format mismatches by raising a clear error you can catch (optionally change to skip/log).

import asyncio
import wave
from typing import AsyncIterator, List
import rtc
from livekit.agents import BackgroundAudioPlayer, AudioConfig

SAMPLE_RATE = 48000
NUM_CHANNELS = 1
SAMPLE_WIDTH_BYTES = 2
FRAME_MS = 20
FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_MS / 1000)  # 960
BYTES_PER_SAMPLE = SAMPLE_WIDTH_BYTES * NUM_CHANNELS
BYTES_PER_FRAME = FRAME_SAMPLES * BYTES_PER_SAMPLE

class PrewarmedWavStream:
    def __init__(self, paths: List[str], loop: bool = True):
        self.loop = loop
        self._buffers: List[bytes] = []
        self._frames_per_file: List[int] = []
        self._prewarmed = asyncio.Event()
        self._stop = False
        self._lock = asyncio.Lock()
        self._paths = paths

    async def prewarm(self):
        """Load files into memory. Call this during prewarm/startup."""
        buffers: List[bytes] = []
        frames_counts: List[int] = []
        for p in self._paths:
            with wave.open(p, "rb") as wf:
                if wf.getframerate() != SAMPLE_RATE or wf.getnchannels() != NUM_CHANNELS or wf.getsampwidth() != SAMPLE_WIDTH_BYTES:
                    raise RuntimeError(f"Unsupported wav format: {p} (expected 48kHz mono s16)")
                raw = wf.readframes(wf.getnframes())
                buffers.append(raw)
                frames_counts.append(wf.getnframes())
        async with self._lock:
            self._buffers = buffers
            self._frames_per_file = frames_counts
            self._prewarmed.set()

    async def stop(self):
        self._stop = True

    async def frames(self) -> AsyncIterator[rtc.AudioFrame]:
        """Per-consumer async iterator that yields continuous frames across files."""
        await self._prewarmed.wait()
        idx = 0
        while not self._stop:
            async with self._lock:
                if not self._buffers:
                    return
                buf = self._buffers[idx]
            offset = 0
            buf_len = len(buf)
            while offset < buf_len and not self._stop:
                chunk = buf[offset: offset + BYTES_PER_FRAME]
                if not chunk:
                    break
                samples_in_chunk = len(chunk) // BYTES_PER_SAMPLE
                frame = rtc.AudioFrame(
                    data=chunk,
                    sample_rate=SAMPLE_RATE,
                    num_channels=NUM_CHANNELS,
                    samples_per_channel=samples_in_chunk,
                )
                yield frame
                offset += BYTES_PER_FRAME
                await asyncio.sleep(FRAME_MS / 1000.0)
            idx += 1
            if idx >= len(self._buffers):
                if self.loop:
                    idx = 0
                else:
                    break

Usage:

_global_stream = PrewarmedWavStream(["/path/a.wav", "/path/b.wav"], loop=True)

# during agent startup / prewarm
await _global_stream.prewarm()

# when playing (each consumer gets its own iterator)
player = BackgroundAudioPlayer()
await player.play(AudioConfig(source=_global_stream.frames(), volume=0.9))

Options I can provide next (pick any): • Make prewarm resilient (skip/log bad files instead of raising). • Add final-frame padding so every yielded frame is exactly 20 ms. • Stream large files from disk without full preload to save memory. • Add on-the-fly resampling/conversion for non-48kHz/mono files. Which option would you like me to implement? Also tell me if you want prewarm to fail fast (current behavior) or quietly skip bad files.

better, how can i catch timeout errors so code doesnt break?

You're asking how to catch stream timeout errors so your prewarmed iterator doesn't crash the agent. Below is a concise pattern showing how to catch the common exceptions around playback, handle them gracefully, and optionally retry or fall back to silence. Key ideas: • Wrap

player.play()

and the iterator consumption in try/except. • Catch network/RTC/mixer errors and asyncio timeouts. • Provide a fallback iterator (silence) or retry with backoff. • Ensure the iterator itself yields predictable frames and stops cleanly. Code (integrate into your existing PrewarmedWavStream and play logic):

import asyncio
import logging
from typing import AsyncIterator
import rtc
from livekit.agents import BackgroundAudioPlayer, AudioConfig

logger = logging.getLogger(__name__)

# simple silence frame of FRAME_SAMPLES samples
SILENCE_BYTES = b"\x00" * BYTES_PER_FRAME

async def silent_frames() -> AsyncIterator[rtc.AudioFrame]:
    while True:
        yield rtc.AudioFrame(
            data=SILENCE_BYTES,
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            samples_per_channel=FRAME_SAMPLES,
        )
        await asyncio.sleep(FRAME_MS / 1000.0)

async def safe_play(player: BackgroundAudioPlayer, source_iter: AsyncIterator[rtc.AudioFrame], volume: float = 1.0, retries: int = 2):
    attempt = 0
    while attempt <= retries:
        try:
            # Wrap the play call which consumes the async iterator
            await player.play(AudioConfig(source=source_iter, volume=volume))
            return  # success
        except asyncio.CancelledError:
            # let cancellation propagate
            raise
        except asyncio.TimeoutError as e:
            logger.warning("BackgroundAudioPlayer timeout on attempt %d: %s", attempt, e)
        except Exception as e:
            # catch RTC/mixer/network-related errors from the player
            logger.exception("BackgroundAudioPlayer error on attempt %d: %s", attempt, e)

        # retry/backoff logic
        attempt += 1
        backoff = min(2 ** attempt, 8)
        await asyncio.sleep(backoff)

    # if all retries failed, fallback to silence so mixer stays healthy
    logger.error("All play attempts failed; falling back to silence")
    try:
        await player.play(AudioConfig(source=silent_frames(), volume=0.0))
    except Exception:
        # last-resort: log and return to avoid crashing
        logger.exception("Fallback to silence also failed")

# Example usage:
# await safe_play(player, _global_stream.frames(), volume=0.8)

Notes and recommendations: • If timeouts originate from the iterator (e.g., too-slow yields), ensure you keep

await asyncio.sleep(FRAME_MS/1000)

in your iterator to maintain real-time pacing. • If you want tighter detection of slow iterators, wrap iteration with

asyncio.wait_for(iterator.__anext__(), timeout=some_ms)

inside a custom adapter to raise clear timeouts per-frame. • Use logging and metrics to surface frequent timeouts so you can decide whether to increase retries, prewarm more, or change memory/disk strategy. Would you like the adapter that applies per-frame

asyncio.wait_for

with a configurable timeout so you can detect stalled iterators earlier?

nono, in the regular play method from background player

You're asking how to catch timeout/errors from the regular

BackgroundAudioPlayer.play()

call so playback doesn't crash. The

play()

call returns/awaits a PlayHandle and may raise exceptions if playback fails. Use a try/except around the await, optionally inspect/handle specific exceptions, and provide fallbacks (retry, silence, or skip). Concise pattern to use:

import asyncio
import logging
from livekit.agents import BackgroundAudioPlayer, AudioConfig

logger = logging.getLogger(__name__)

async def play_with_handling(player: BackgroundAudioPlayer, source, volume: float = 1.0, retries: int = 2):
    attempt = 0
    while attempt <= retries:
        try:
            # play() accepts a path or an AsyncIterator[rtc.AudioFrame]
            await player.play(AudioConfig(source=source, volume=volume))
            return  # success
        except asyncio.CancelledError:
            raise
        except asyncio.TimeoutError as e:
            logger.warning("Playback timeout (attempt %d): %s", attempt, e)
        except Exception as e:
            # Catch RTC/mixer/FFmpeg/file errors from play()
            logger.exception("Playback error (attempt %d): %s", attempt, e)

        attempt += 1
        backoff = min(2 ** attempt, 8)
        await asyncio.sleep(backoff)

    # All retries failed: fallback to silence or safe no-op
    logger.error("All play attempts failed; falling back to silence/no-op")
    try:
        # If you have a silent iterator, use it; otherwise skip playback
        await player.play(AudioConfig(source=silent_frames(), volume=0.0))
    except Exception:
        logger.exception("Fallback to silence also failed")

# Example silent_frames generator (20ms s16le 48k mono)
FRAME_MS = 20
SAMPLE_RATE = 48000
FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_MS / 1000)
BYTES_PER_FRAME = FRAME_SAMPLES * 2  # 16-bit mono

async def silent_frames():
    silence = b"\x00" * BYTES_PER_FRAME
    while True:
        yield rtc.AudioFrame(
            data=silence,
            sample_rate=SAMPLE_RATE,
            num_channels=1,
            samples_per_channel=FRAME_SAMPLES,
        )
        await asyncio.sleep(FRAME_MS / 1000.0)

Notes and best practices: • Keep real-time pacing in iterators (await sleep per frame). If the iterator stalls, the mixer may time out. • If you want earlier detection of a stalled source, supply an AsyncIterator and wrap each

__anext__()

call with

asyncio.wait_for(..., timeout=per_frame_timeout)

inside an adapter to raise timeouts per frame. • Log metrics/counts for timeouts so you can tune retries/backoff or prewarm behavior. • Prefer prewarmed in-memory iterators (your previous PrewarmedWavStream) when file-path playback is flaky. If you want, I can: • Provide the adapter that wraps an AsyncIterator with per-frame

asyncio.wait_for

, • Add a non-blocking play pattern that returns the PlayHandle so you can cancel/monitor it, or • Show an example that distinguishes common exception types you see in your logs. Which would you like? Sources: Agent speech and audio | LiveKit docs

but i have it this way:

async def play_notification_sound() -> bool:
            """Play notification sound twice before agent speaks.

            Helper function to avoid code duplication across RPC methods.

            Returns:
                True if completed without interruption, False if interrupted
            """
            if background_audio:
                <http://logger.info|logger.info>("Playing notification sound twice")
                for i in range(2):
                    # Check for interruption before starting each sound
                    if assistant.is_interruption_requested():
                        <http://logger.info|logger.info>(
                            f"Notification sound interrupted before sound {i + 1}/2"
                        )
                        return False

                    handle = background_audio.play(NOTIFICATION_AUDIO_FILE)
                    # Wait for the notification to finish playing
                    while handle and not handle.done():
                        # Check for user interruption during playback
                        if assistant.is_interruption_requested():
                            <http://logger.info|logger.info>(
                                f"Notification sound interrupted during sound {i + 1}/2"
                            )
                            handle.stop()
                            return False
                        await asyncio.sleep(0.1)
                    logger.debug(f"Notification sound {i + 1}/2 completed")
            return True

Good — you provided the exact code. Below is a minimal, safe update to your

play_notification_sound

that: • Awaits

background_audio.play(...)

and wraps it in try/except to catch playback errors (including timeouts), • Monitors the returned PlayHandle and handles interruption requests by stopping the handle, • Retries play a configurable number of times before giving up, • Returns False on interruption or on unrecoverable playback failure. Apply this patch-like snippet into your code:

async def play_notification_sound(retries: int = 1) -> bool:
    if not background_audio:
        return True

    <http://logger.info|logger.info>("Playing notification sound twice")
    for i in range(2):
        if assistant.is_interruption_requested():
            <http://logger.info|logger.info>(f"Notification sound interrupted before sound {i + 1}/2")
            return False

        attempt = 0
        while attempt <= retries:
            try:
                handle = background_audio.play(NOTIFICATION_AUDIO_FILE)
                # If play() returns a handle-like object, await completion by polling
                while handle and not handle.done():
                    if assistant.is_interruption_requested():
                        <http://logger.info|logger.info>(
                            f"Notification sound interrupted during sound {i + 1}/2"
                        )
                        try:
                            handle.stop()
                        except Exception:
                            logger.exception("Failed to stop play handle")
                        return False
                    await asyncio.sleep(0.1)
                # Completed successfully
                logger.debug(f"Notification sound {i + 1}/2 completed")
                break
            except asyncio.CancelledError:
                raise
            except asyncio.TimeoutError as e:
                logger.warning("Playback timeout on attempt %d: %s", attempt, e)
            except Exception as e:
                logger.exception("Playback error on attempt %d: %s", attempt, e)

            attempt += 1
            backoff = min(2 ** attempt, 4)
            await asyncio.sleep(backoff)

        else:
            # Exited retries loop without break -> failed all attempts
            logger.error("Failed to play notification sound after retries")
            # Optional: continue to next sound, or return False to abort
            return False

    return True

If you want, I can: • Add a fallback to play silence instead of returning False, • Make the retry/backoff configurable globally, • Or provide a small adapter that enforces a per-frame timeout for custom iterators. Which would you prefer?

what is the backoff?

You're asking what "backoff" means in the retry code. Backoff is the delay inserted between retry attempts to avoid immediate repeated failures and to give transient issues time to recover. In the example code the backoff is computed as: • backoff = min(2 ** attempt, 4) (or 8 in other snippets) This is exponential backoff: • attempt = 0 -> backoff = 1s • attempt = 1 -> backoff = 2s • attempt = 2 -> backoff = 4s • attempt >= 3 -> backoff capped at 4s (or 8s where capped) Why use it: • Prevents hammering the mixer/decoder on transient errors. • Gives short recovery time while keeping total retry time bounded. You can adjust: • Base and exponent (e.g., 1.5 ** attempt) • Add jitter: multiply by random.uniform(0.8, 1.2) to avoid synchronized retries • Increase cap or max attempts for more aggressive retrying Want me to provide a tiny snippet that adds jitter and makes base/cap configurable?

one retry is enough, to be honest i'd rather continue with code than failing on playing this sound

Got it — you want a single retry (or effectively one attempt) and to continue execution if playback fails. Below is a minimal patch to your

play_notification_sound

that: • Attempts playback once (no retries), • Catches/playback exceptions and logs them, • Stops playback on interruption, • Always returns so the agent continues (returns False only for interruption). Replace your function body with this:

async def play_notification_sound() -> bool:
    if not background_audio:
        return True

    <http://logger.info|logger.info>("Playing notification sound twice")
    for i in range(2):
        if assistant.is_interruption_requested():
            <http://logger.info|logger.info>(f"Notification sound interrupted before sound {i + 1}/2")
            return False

        try:
            handle = background_audio.play(NOTIFICATION_AUDIO_FILE)
        except Exception as e:
            logger.exception("Failed to start playback for notification sound: %s", e)
            # continue execution without blocking on playback
            continue

        # Monitor the play handle; stop if interrupted
        try:
            while handle and not handle.done():
                if assistant.is_interruption_requested():
                    <http://logger.info|logger.info>(f"Notification sound interrupted during sound {i + 1}/2")
                    try:
                        handle.stop()
                    except Exception:
                        logger.exception("Failed to stop play handle")
                    return False
                await asyncio.sleep(0.1)
            logger.debug(f"Notification sound {i + 1}/2 completed")
        except Exception as e:
            # Any runtime error while monitoring playback — log and continue
            logger.exception("Playback monitoring error for notification sound: %s", e)
            # do not abort the agent; proceed to next step
            continue

    return True

If you want, I can instead: • Start playback but not wait at all (fire-and-forget), or • Fall back to a silent play instead of skipping when start fails. Which behavior would you prefer?