Is there any standard way to perform load tests to LiveKit Agents? I'm currently trying to use locust + LiveKit Python SDK but I'm having troubles to scale up these tests.

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

If I use

lk load-test

, will I be able to send custom prerecorded audios? Or just blank packets?

Thanks — I’m responding about whether

lk load-test

can send custom prerecorded audio. It cannot.

lk load-test

only sends blank (non‑audible) audio packets to simulate a target bitrate. If you need prerecorded audio in tests, I can help outline two options: • Build a custom load tester using a LiveKit SDK (Go/JS/Python) that publishes prerecorded audio files. • Modify or extend a client implementation to play files into a track and run that at scale (run across multiple machines or containers). Which option would you prefer, and what language/stack do you plan to use? Sources: Benchmarking | LiveKit docs

I'd prefer a load tester based on Locust and Python SDK

Thanks — I’m responding to your preference for a Locust + Python SDK load tester and the requested clarifications. Quick clarifying questions so I can give a focused, working example and scaling advice: 1. Do you need every Locust user to publish a prerecorded audio file, or only a subset (e.g., 10% publishers, rest subscribers)? 2. What target concurrency do you want to reach (approx. total clients)? 3. Will you run Locust workers across multiple machines/containers? 4. Which OS and Python version are you using, and do you control ulimits (e.g., can set

ulimit -n 65535

)? Answer these and I’ll provide a concise Locust test script that: • publishes prerecorded audio per user, • shows efficient client/session handling with the LiveKit Python SDK, • includes tuning and scaling tips for Locust and OS limits.

1. Yes, every Locust user needs to publish prerecorded audios. These audios simulate a conversation so, the Locust user must wait the first greeting of the agent and then, once the the agent is listening, publish the first prerecorded audio. Then, every time the agent is listening, the Locust user will publish the next audio until the list is exhausted 2. At least 100 users 3. I'd prefer to run it in just one machine (at least for 100 users) but I could scale it up 4. I'm running on mac OS with python 3.12.12 but If I had to run it across multiple machines/containers I'd use Linux. Yes, I do control ulimits.

Great — thanks for the details. I’ll give a focused, runnable plan and a compact Locust + LiveKit Python SDK example that matches your requirements: every Locust user publishes prerecorded audio segments in sequence, waits for the agent to greet, then publishes each next segment only when the agent is listening. I’ll also include practical tuning notes for running ~100 users on one macOS machine. Key assumptions I’ll follow • Agent emits an event or data message indicating it’s in the "listening" state (recommended). If not available, we can fallback to audio-based VAD heuristics. • Audio files are small WAV PCM files (2–5s), stored locally and identical for all users unless you tell me otherwise. • You have tokens that allow publishing audio to the room. Minimal Locust test approach (summary) • Each Locust User: a. Connects to LiveKit room with the Python SDK. b. Publishes a LocalAudioTrack backed by an

AudioSource

. c. Waits for agent greeting (agent -> "listening") before sending first segment. d. For each segment: wait for agent listening signal, stream WAV frames to

AudioSource

in 20ms chunks, then wait again. e. Clean disconnect when segments exhausted. Compact example • Save as

locustfile.py

. Set env vars:

LIVEKIT_URL

,

LIVEKIT_TOKEN

. Place WAVs under

./audio_segments/

. • I’ll keep this concise; let me know if you want the full file pasted into chat. Tuning & scaling tips for 100 users (one macOS machine) • ulimit: set

ulimit -n 65535

. • Event loop: install and enable

uvloop

for better asyncio performance. • CPU/RAM: 8+ cores and 16GB recommended; 100 users should be feasible on a modern MacBook Pro. • Network: ensure sufficient upload bandwidth (audio-only is light but concurrent signalling adds load). • Logging: minimize printing inside hot loops. • Start small (10 users), verify flow, then ramp to 100. • If hitting limits, run locust workers across multiple Linux containers/machines. Agent readiness detection (recommended) • Best: agent publishes a short data message like

"ready_for_input"

when in listening state — Locust listens for data messages and sets an asyncio.Event. • Fallback: consume agent audio track and set listening when audio stops (less precise). Next steps / choices for me to act on • I can paste a ready-to-run

locustfile.py

(complete code) that implements the flow above (WAV loader, AudioSource streaming in 20ms chunks, agent-ready via data message + fallback to silence). • Or I can provide a Dockerized version and a small script to generate publish-only JWTs. Which do you want now: the full locustfile code, or the Dockerized + token helper? Sources: Camera & microphone | LiveKit docs | Agent session | LiveKit docs

What we're doing now is create a room connection, subscribe to the participant_attributes_changed event to get the agent status, publish an audio track, and use the capture_frame of source to publish the audio. Should we use another approach?

You’re on the right track — your approach will work, but there are a few refinements to make it more reliable and scalable. Quick validation • Connecting to the room, publishing a LocalAudioTrack backed by an

AudioSource

, and calling

capture_frame

to push prerecorded audio is the correct pattern for programmatic playback. • Listening to participant attribute changes can indicate agent state, but it’s not the most robust/specific signal for turn-taking. Recommended improvements 1. Prefer explicit agent signals ◦ Use the agent’s

agent_state_changed

event or have the agent publish a small data message (e.g.,

"ready_for_input"

on a data channel) when it’s listening. This is precise and cheap to detect. 1. Fallback to audio-based detection only if needed ◦ If explicit signals aren’t available, consume the agent’s audio track and detect silence/vad to infer listening. This is less precise and more error-prone under noise/jitter. 1. Optimize audio publishing ◦ Stream frames in ~20ms chunks to

AudioSource.capture_frame

and match sample rate/channels (prefer 48kHz mono). Avoid huge frames or tight loops that block the event loop. 1. Concurrency & performance tips ◦ Minimize logging inside hot loops, use

uvloop

for better asyncio performance, and set

ulimit -n

high (e.g., 65535). ◦ Reuse connections/sources where possible; avoid recreating tracks per segment. 1. Error handling & timeouts ◦ Add timeouts when waiting for agent signals, and backoff/retry on capture failures to avoid hanging users. 1. Scaling note ◦ The Python SDK + Locust will work for ~100 users on a well-tuned machine. If you need higher scale, consider the Go-based

lk load-test

tool or distribute Locust workers across machines. If you want, I can: • Provide a small code snippet showing authoritative detection via a data message handler and the ideal

capture_frame

loop, or • Show a minimal VAD-based fallback snippet for silence detection. Which snippet would help you most?

Okay, provide the small code snippet

Below is a concise, ready-to-run snippet showing the recommended pattern: • Listen for the agent readiness signal (

agent_state_changed

or

participant

attribute

lk.agent.state == "listening"

). • Publish a

LocalAudioTrack

backed by an

AudioSource

. • Stream a WAV file in 20ms chunks via

AudioSource.capture_frame

with an asyncio sleep to simulate real time. Save as part of your Locust task or run inside an async function after connecting the Room. Adjust env vars, file path, and agent identity.

import asyncio
import wave
from pathlib import Path
from livekit import rtc

# assumptions: room is an already-connected rtc.Room, and local_participant available
# agent_identity is the agent's participant.identity string
# wav_path is Path to a 16-bit PCM WAV (prefer 48kHz mono or resample to 48k)

async def publish_prerecorded_when_agent_listening(room: rtc.Room, agent_identity: str, wav_path: Path):
    agent_ready = asyncio.Event()

    # Prefer authoritative event
    @room.on("agent_state_changed")
    def _on_agent_state(ev):
        # ev.state expected e.g. "listening", "speaking"
        if getattr(ev, "state", None) == "listening" and ev.participant.identity == agent_identity:
            agent_ready.set()

    # Fallback: watch participant attributes (if agent sets lk.agent.state attribute)
    @room.on("participant_updated")
    def _on_participant_updated(participant):
        if participant.identity == agent_identity:
            val = participant.get_attribute("lk.agent.state")  # SDK-specific API for attributes
            if val == "listening":
                agent_ready.set()

    # wait for agent greeting / listening
    try:
        await asyncio.wait_for(agent_ready.wait(), timeout=30.0)
    except asyncio.TimeoutError:
        raise RuntimeError("Agent did not become listening within 30s")

    # create audio source and publish track (48kHz mono)
    source = rtc.AudioSource(48000, 1)
    track = rtc.LocalAudioTrack.create_audio_track("prerecorded_audio", source)
    await room.local_participant.publish_track(track)

    # stream WAV in 20ms chunks (960 samples @ 48kHz)
    with wave.open(str(wav_path), "rb") as wf:
        # If WAV not 48kHz/mono, convert or resample beforehand.
        sample_rate = wf.getframerate()
        num_channels = wf.getnchannels()
        bytes_per_sample = wf.getsampwidth()
        chunk_frames = 960  # 20ms @ 48kHz

        while True:
            frames = wf.readframes(chunk_frames)
            if not frames:
                break

            # If WAV sample_rate != 48000 or num_channels !=1, pre-convert/resample before running test.
            audio_frame = rtc.AudioFrame(
                data=frames,
                sample_rate=48000,
                num_channels=1,
                samples_per_channel=len(frames) // (bytes_per_sample * 1),
            )

            await source.capture_frame(audio_frame)
            await asyncio.sleep(0.02)  # simulate real-time 20ms pacing

    # optional: signal end-of-speech (mute/unpublish)
    await room.local_participant.set_microphone_muted(True)
    await room.local_participant.unpublish_track(track)

Best practices reminders: • Prefer the

agent_state_changed

event or an explicit data message from the agent (e.g.,

"ready_for_input"

) for reliable turn signaling. • Use 48kHz mono PCM WAVs or resample before the test to avoid runtime resampling overhead. • Send frames in ~20ms chunks and use

await asyncio.sleep(0.02)

to simulate real-time and avoid overwhelming the event loop. • Keep logging out of the inner loop and set high

ulimit -n

for many concurrent sessions. If you want, I can provide: • a small data-message handler example (agent sends

"ready_for_input"

), or • a full minimal Locust task wrapping this snippet. Which would you prefer? Sources: Agent session | LiveKit docs | Camera & microphone | LiveKit docs | Turns overview | LiveKit docs | Web and mobile frontends | LiveKit docs | Capturing the First Few Seconds of a Call for Processing

Okay, but in you're example the frame size is unknown. I mean you're not limiting to 20ms chunks, right? is that correct?

Good question — yes, the example intends 20ms chunks but only enforces that when the WAV is already 48 kHz mono. Concretely: • 20 ms @ 48 kHz = 0.02 * 48000 = 960 samples. • If samples are 16-bit (2 bytes) and mono, each 20ms chunk = 960 * 2 = 1920 bytes. • If your WAV is not 48 kHz mono, you must resample/convert beforehand or compute chunk size from the file's sample rate and channel count. Helper: compute chunk frames for any sample rate / desired ms:

def frames_for_ms(sample_rate: int, ms: int) -> int:
    return int(sample_rate * (ms / 1000.0))

Revised loop that enforces 20ms chunks based on file metadata:

with wave.open(str(wav_path), "rb") as wf:
    sr = wf.getframerate()
    chans = wf.getnchannels()
    sampwidth = wf.getsampwidth()  # bytes per sample per channel
    frames_per_chunk = frames_for_ms(sr, 20)     # frames == samples per channel
    bytes_per_chunk = frames_per_chunk * chans * sampwidth

    while True:
        data = wf.readframes(frames_per_chunk)
        if not data:
            break

        # If sr != 48000 or chans != 1, resample/convert offline or do it here (costly)
        audio_frame = rtc.AudioFrame(
            data=data,
            sample_rate=sr,           # prefer to match AudioSource
            num_channels=chans,
            samples_per_channel=frames_per_chunk,
        )
        await source.capture_frame(audio_frame)
        await asyncio.sleep(0.02)  # 20ms pacing

Recommendation: pre-convert all WAVs to the AudioSource format you choose (48 kHz, mono, 16-bit) to avoid runtime resampling and ensure each captured frame equals ~20ms of audio.

Currently I'm sending 16kHz mono 16-bit. Here you have a snippet:

@dataclass
class AudioClip:
    """Audio data from a WAV file, ready to be sent to LiveKit."""
    audio_data: bytes  # Raw PCM samples (16-bit signed, little-endian)
    sample_rate: int
    channels: int
    
    # LiveKit frame size: 20ms at 16kHz = 320 samples = 640 bytes
    FRAME_SAMPLES: int = 320
    FRAME_DURATION_SEC: float = 0.02  # 20ms

    @classmethod
    def from_wav_file(cls, file_path: str) -> "AudioClip":
        with wave.open(file_path, "rb") as wav_file:
            sample_rate = wav_file.getframerate()
            channels = wav_file.getnchannels()
            audio_data = wav_file.readframes(wav_file.getnframes())
            return cls(audio_data=audio_data, sample_rate=sample_rate, channels=channels)
    
    @property
    def frame_bytes(self) -> int:
        """Bytes per frame (320 samples * 2 bytes per sample)."""
        return self.FRAME_SAMPLES * 2
    
    @property
    def total_frames(self) -> int:
        """Total number of 20ms frames in this clip."""
        return len(self.audio_data) // self.frame_bytes
...
for i in range(0, len(audio_clip.audio_data), frame_bytes):
    chunk = audio_clip.audio_data[i:i + frame_bytes]
    if len(chunk) < frame_bytes:
        break  # Skip incomplete final frame
    
    # Unpack 320 samples (16-bit signed, little-endian)
    samples = struct.unpack(f"<{AudioClip.FRAME_SAMPLES}h", chunk)
    samples_array = array.array("h", samples)
    
    # Create and send frame
    frame = rtc.AudioFrame.create(SAMPLE_RATE, NUM_CHANNELS, AudioClip.FRAME_SAMPLES)
    frame.data[:] = samples_array
    await source.capture_frame(frame)
    
    frames_sent += 1
    if frames_sent % 50 == 0:
        logger.debug("Audio progress", frames_sent=frames_sent, total=total_frames)
    
    # Real-time pacing: 20ms per frame
    await asyncio.sleep(AudioClip.FRAME_DURATION_SEC)

Do you see anything wrong?

Your snippet is mostly correct—you’re chunking 20ms frames and using `capture_frame`—but two practical fixes will make it robust at scale: 1) Use a monotonic clock for precise pacing (avoid naive repeated

asyncio.sleep(0.02)

which drifts):

import time
start = time.monotonic()
frames_sent = 0

for i in range(0, len(audio_clip.audio_data), frame_bytes):
    chunk = audio_clip.audio_data[i:i+frame_bytes]
    if len(chunk) != frame_bytes:
        break

    samples = struct.unpack(f"<{AudioClip.FRAME_SAMPLES}h", chunk)
    frame = rtc.AudioFrame.create(16000, 1, AudioClip.FRAME_SAMPLES)
    frame.data[:] = array.array("h", samples)
    await source.capture_frame(frame)

    frames_sent += 1
    target = frames_sent * AudioClip.FRAME_DURATION_SEC
    sleep = target - (time.monotonic() - start)
    if sleep > 0:
        await asyncio.sleep(sleep)

2) Increase

AudioSource

buffer to avoid “failed to capture frame” under load:

source = rtc.AudioSource(sample_rate=16000, num_channels=1, queue_size_ms=2000)

Additional checks/best practices • Validate WAV metadata: ensure sample_rate == 16000 and channels == 1 (or pre-convert). • Pre-load/decode clips into memory to avoid per-frame I/O. • Skip or zero-pad the final incomplete frame (you already skip it). • Minimize logging inside the sending loop. • On macOS/Linux, raise

ulimit -n

and use

uvloop

for better asyncio performance when scaling to ~100 users. If you want, I can: • Show a tiny helper that validates/resamples WAVs before test run, or • Paste a complete minimal Locust task that integrates this loop. Which do you prefer? Sources: Resolving "InvalidState - failed to capture frame" errors in AudioSource | Processing raw media tracks | LiveKit docs

I can't send you the whole locustfile as it's too long to be sent by slack but it gets stuck sending the first clip. Any idea why it's happening?

Sorry that’s blocking you — let’s troubleshoot quickly. A few targeted checks and questions will help find why it hangs on the first clip. Please check/send (or answer) the following: 1. Do you see any error logs or exceptions (SDK logs, Locust output, or stderr)? If yes, paste the last ~50 lines. 2. Is the agent actually reaching the "listening" state? Add a log right before you start streaming to confirm you passed the readiness wait. 3. Is the call to

await source.capture_frame(...)

returning or hanging? Add a timestamped log immediately before and after the capture_frame call. 4. Are you using

AudioSource(..., queue_size_ms=...)

? If not, try increasing it to 2000 ms. 5. Are your WAV files confirmed 16kHz mono 16-bit and frame-aligned (len % frame_bytes == 0)? Log sample_rate/channels/total_frames. 6. Check CPU, memory, and FD usage while a worker is stuck (top, vm_stat, lsof -p <pid>) — do you see resource exhaustion? 7. Are you running under the Locust async runner correctly (async tasks inside Locust User)? Any blocking synchronous code elsewhere? 8. Do you see repeated log messages like “failed to capture frame”, queue full, or asyncio timeouts? Quick fixes to try now: • Use monotonic pacing (time-based) instead of fixed asyncio.sleep loops (reduce drift). • Increase

queue_size_ms

on AudioSource. • Ensure you call

await room.connect(...)

and await publication success before streaming. • Temporarily reduce concurrency to 1 user and reproduce; attach a verbose log to observe sequence. • Add a 1–2s small sleep after publish before sending the first frame to let the pipeline warm up. If you share the small logs from checks 1–3 and the audio file metadata (answer 5), I’ll pinpoint the exact likely cause and a precise fix.

1. There's no error logs or exceptions 2. Yes, we wait for the agent to be in listening state before sending any chunk 3. capture_frame is hanging but not always in the same chunk. 4. I'm using _queue_size_ms_=1000 5. Yes, they are. sample_rate = 16000, channles = 1, total_frames = 137 6. There's no resource exhaustion. htop shows 2.5% of CPU 7. Yes, I'm using async tasks inside a Locust User and I'm doing polling to read a shared variable to get data from the Room connection 8. No, there are no error messages of any kind

Thanks — helpful details. Since

capture_frame

intermittently hangs (no errors, low CPU, queue_size_ms=1000), most likely causes are back-pressure from the AudioSource queue or an event-loop/block causing capture to wait. Try these targeted changes and checks to unblock it. Immediate fixes to try 1. Increase buffer: ◦ Set

queue_size_ms=5000

on

AudioSource

to give more headroom:

source = rtc.AudioSource(sample_rate=16000, num_channels=1, queue_size_ms=5000)

1. Await publish fully and warm up: ◦ After

publish_track

, wait a short grace period before sending first frame:

pub = await room.local_participant.publish_track(track)
     await asyncio.sleep(0.5)

1. Handle back-pressure with a retry and timeout: ◦ Wrap

capture_frame

in a retry with a timeout so it doesn’t hang forever:

async def safe_capture(source, frame, timeout=2.0):
         try:
             await asyncio.wait_for(source.capture_frame(frame), timeout=timeout)
         except asyncio.TimeoutError:
             # backoff and retry once
             await asyncio.sleep(0.05)
             await asyncio.wait_for(source.capture_frame(frame), timeout=timeout)

1. Use monotonic pacing (you already do) and ensure your pacing logic yields control to the loop (avoid CPU spin). 2. Ensure single AudioSource per Locust user: ◦ Don’t share a source/track between users or reuse one source concurrently from multiple tasks. Checks to run if it still hangs • Add timestamped logs immediately before and after

capture_frame

to confirm it’s that call blocking and to measure duration. • Temporarily increase log verbosity in the SDK to detect internal warnings. • Run a single user (no Locust concurrency) and loop the same clip repeatedly to see if it ever hangs—if not, it’s interaction with Locust scheduling. • Verify you’re not polling a shared variable in a tight loop without

await

(can starve event loop). Replace polling with an

asyncio.Event

set by the Room callbacks. Recommended robust pattern • Publish track, await publish success, sleep ~0.5s. • For each 20ms frame: a. build

AudioFrame

b. call

await asyncio.wait_for(source.capture_frame(frame), timeout=2.0)

c. if timeout, back off (sleep 50–200ms) and retry once d. use monotonic clock pacing to schedule next send Would you like me to: • paste a minimal instrumented snippet that includes logging, safe_capture, and the warm-up delay so you can drop it into your locustfile; or • review a small paste of the part that publishes the track + the capture loop (the few hundred lines you can paste here)?

Before adding any timeout to capture_frame, I'd like to clarify something. We're running the whole room logic (send audio and get agent state) in a different thread that runs the async tasks. Is it the right way to run this kind of tests using locust?

Short answer: running the LiveKit room logic in a separate thread is likely the cause. The LiveKit Python SDK is async and expects its operations (including

capture_frame

) to run in the same asyncio event loop; moving room/SDK calls across threads or different loops can cause deadlocks/hangs even with no errors. What to do next (practical, minimal changes) • Stop using a separate thread for the Room/AudioSource. Run all LiveKit async code in the same asyncio event loop that Locust uses. • Use Locust's async support (async User/task). Example pattern: ◦ async def on_start(): await room.connect(...) ◦ async def task(self): await publish/stream_frames(...) ◦ use asyncio.create_task(...) only for helper coroutines within that same loop. • Replace any cross-thread polling of shared vars with asyncio primitives: use asyncio.Event, Queue, or callbacks on the Room to set events so the sending loop waits with await event.wait() instead of busy polling. • Keep one AudioSource per simulated user and do not share it across threads/tasks. Why this fixes your hanging capture_frame • capture_frame relies on the SDK's internal async queue and the same event loop for progress; threading breaks that coordination and can block capture_frame unpredictably. Quick checklist to validate after making the change • room.connect, publish_track, and capture_frame all execute on the same loop/thread. • Replace polling with asyncio.Event for agent readiness. • Add a small warm-up delay after publish (0.25–0.5s). • Increase queue_size_ms (e.g., 2000) if you still see intermittent backpressure. • Instrument with timestamped logs before/after capture_frame to confirm it no longer blocks. If you want, I can: • Paste a minimal async Locust User example that runs the Room and capture loop in the same event loop and uses asyncio.Event for agent readiness. Which would you prefer? Sources: Resolving &amp;#34;InvalidState - failed to capture frame&amp;#34; errors in AudioSource | Going beyond a single-core

It's running on a different thread because Locust does not support async it uses gevent so, I'm creating a new thread and running all Livekit related tasks on this thread and everything in the same event loop.

Thanks — that context helps. Running the LiveKit async code on a separate thread while Locust uses gevent can still cause subtle event-loop conflicts even if you create a dedicated asyncio loop on that thread. Gevent monkey‑patching and the rest of Locust’s runtime can interact with networking, blocking syscalls, or scheduling in ways that make SDK async calls (like

capture_frame

) intermittently hang. Two safe options Option A — Use Locust’s AsyncUser (recommended) • Locust supports async users via

AsyncUser

/

FastHttpUser

. Run your LiveKit async logic directly inside an async Locust user so everything runs on a single asyncio event loop managed by Locust. • Advantages: no thread/loop boundaries, predictable scheduling, fewer subtle deadlocks. • I can provide a minimal AsyncUser example that connects, waits for agent readiness via

asyncio.Event

, and streams 20ms frames with monotonic pacing. Option B — If you must keep a separate thread, ensure strict isolation • Create a single dedicated thread with its own asyncio loop and ensure no gevent/Locust code interacts with sockets or blocking calls used by that thread. • Use only asyncio primitives inside that thread (no cross-thread polling). Communicate between Locust tasks and that thread via thread-safe queues (e.g., asyncio loop.call_soon_threadsafe to set events) and avoid busy polling. • Increase

AudioSource(queue_size_ms=5000)

, add capture timeouts and retry, and instrument logs — but this is fragile and still more error-prone than Option A. Why your current approach likely causes intermittent hangs • Gevent monkey patching can alter socket/IO behavior used by the SDK or underlying libraries. • Cross-thread boundaries with an asyncio loop mean any small blocking syscall or scheduling difference can stall the SDK’s internal pump, causing

capture_frame

to await indefinitely. • The intermittent nature fits an event-loop scheduling deadlock rather than pure resource exhaustion. Recommendation • Move to Locust’s AsyncUser and run the LiveKit flow in that async task. This will almost certainly eliminate the intermittent

capture_frame

hang. • If you want, I’ll paste a concise AsyncUser locustfile now (connect → wait for agent via

asyncio.Event

→ publish track → stream 20ms frames with monotonic pacing). Which would you prefer: the full AsyncUser snippet, or a thread‑safe adapter pattern for your current setup?