Is there a way we could detect that a person has too much background noise and trigger an action from the agent side ?

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

We are using Python, and the idea would be to do it during the call, exactly for what you mentioned • Play a verbal prompt (ask the user to move to a quieter place or mute). We are using BVC room_options = RoomOptions( audio_input=AudioInputOptions( noise_cancellation=noise_cancellation.BVC(), ), ) And use a TTS - LLM - SST pipeline, but where could i attach so i can analyze that background noise level ?

Thanks — this is clear. You should analyze noise level right after you receive/process each inbound audio frame in your agent’s audio input pipeline (after BVC runs), then trigger the TTS prompt when your noise threshold is exceeded. Where to attach the analysis • In the agent-side audio input handler that receives audio frames/chunks from the room. Attach noise-metric calculation immediately after noise_cancellation processing and before sending audio into your STT/LLM pipeline. That lets you measure residual/background energy post-filtering (what the agent actually hears). What to measure (simple, reliable signals) • Residual noise energy: compute RMS or short-time energy of the filtered audio chunk. • SNR proxy: compare speech-activity energy vs background energy using VAD or voice activity frames. • Repeated STT failures or low confidence: count consecutive low-confidence transcriptions. Example flow (high-level) 1. Receive audio chunk from room. 2. Run BVC (already configured). 3. Compute metric(s) on the filtered chunk (RMS, SNR proxy, VAD ratio). 4. Apply debounce/threshold rules (e.g., noise metric > threshold for N consecutive chunks). 5. If threshold crossed, call your TTS to play the verbal prompt into the room and/or mark the caller UI. Minimal Python pseudocode

from collections import deque
import numpy as np

THRESH = 0.02           # tune per your audio level units
CONSECUTIVE = 3

recent = deque(maxlen=CONSECUTIVE)

def rms(samples):
    return float(np.sqrt(np.mean(np.square(samples))))

async def on_audio_chunk(filtered_pcm):
    # filtered_pcm: numpy array of float32/int16 samples after BVC
    level = rms(filtered_pcm)
    recent.append(level)
    if len(recent) == CONSECUTIVE and all(l > THRESH for l in recent):
        await play_tts_prompt("Please move to a quieter place or mute when not speaking")
        recent.clear()
    else:
        # continue STT -> LLM pipeline
        await send_to_stt(filtered_pcm)

Tuning & reliability tips • Calibrate THRESH using sample calls in your environment. • Use a short debounce window (2–5 chunks) to avoid false positives. • Combine RMS with VAD or STT confidence for fewer spurious prompts. • Rate-limit prompts (e.g., once per minute per participant) to avoid annoyance. If you want, I can: • Provide a concrete code snippet that hooks into LiveKit’s Python agent SDK callbacks you’re using. • Suggest precise thresholds and VAD parameters if you share sample audio levels or your buffer/frame size. Which would you prefer?

Please provide me the code snippet, because im not sure which event i should attach this or which pipeline node

Thanks — below is a concise, ready-to-use Python snippet that shows exactly where to attach real-time noise analysis in your agent pipeline (using BVC) and how to trigger a TTS prompt during the call. What this does • Subscribes to inbound audio in

on_track_subscribed

. • Reads raw frames from an

AudioStream

after BVC runs. • Computes RMS per frame, debounces sustained high-noise, and calls

session.say()

to play a verbal prompt. • Includes simple rate-limiting to avoid repeated prompts. Code snippet

import asyncio
import time
import numpy as np
from collections import deque
from livekit import rtc
from livekit.agents import AgentSession, stt

# Tuning params
NOISE_THRESHOLD = 0.02     # tune this with your audio (RMS normalized to [-1,1])
SUSTAIN_SECONDS = 1.0      # noise must persist this long to trigger
FRAME_DURATION = 0.02      # assumed frame length in seconds (20ms typical)
RATE_LIMIT_SECONDS = 60    # minimum seconds between prompts per participant

async def analyze_audio_for_noise(session: AgentSession, audio_stream: rtc.AudioStream, participant_sid: str):
    frames_needed = int(SUSTAIN_SECONDS / FRAME_DURATION)
    noisy_count = 0
    last_prompt_time = 0

    async for event in audio_stream:
        frame = event.frame
        # Convert int16 PCM -> float32 normalized to [-1,1]
        samples = np.frombuffer(frame.data, dtype=np.int16).astype(np.float32) / 32768.0
        rms = float(np.sqrt(np.mean(np.square(samples))))

        if rms > NOISE_THRESHOLD:
            noisy_count += 1
        else:
            noisy_count = max(0, noisy_count - 1)

        if noisy_count >= frames_needed:
            now = time.time()
            if now - last_prompt_time >= RATE_LIMIT_SECONDS:
                last_prompt_time = now
                noisy_count = 0
                # Play TTS prompt into the room (agent voice)
                await session.say("I'm having trouble hearing you due to background noise. Could you please move to a quieter place or mute when not speaking?")
            else:
                noisy_count = 0  # reset to avoid immediate retrigger

@server.rtc_session()
async def entrypoint(ctx: rtc.JobContext):
    # Create your AgentSession with BVC configured in room_options as you already have
    session = AgentSession(
        ctx=ctx,
        stt=stt.DeepgramSTT(),  # or your STT
        # tts=..., other options
    )
    await session.start()

    @ctx.room.on("track_subscribed")
    def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            # Wrap the inbound track so we can read raw frames after processing (BVC is applied inbound)
            audio_stream = rtc.AudioStream(track)
            # Start background task to monitor noise for this participant
            asyncio.create_task(analyze_audio_for_noise(session, audio_stream, participant.sid))

Notes and tuning • Place this alongside your existing BVC

room_options

configuration — BVC will have been applied to the inbound audio the agent receives. • Tune

NOISE_THRESHOLD

,

SUSTAIN_SECONDS

, and

RATE_LIMIT_SECONDS

with sample calls to avoid false positives. • For fewer spurious prompts, combine RMS with STT confidence or VAD: require both high RMS and low STT confidence before calling

session.say()

. • If you want a snippet that uses your exact Agent class or STT/TTS setup, paste those bits and I’ll adapt this to match. Would you like the same snippet adapted to use STT confidence in the trigger condition? Sources: Pipeline nodes and hooks | LiveKit docs | Enhanced noise cancellation | LiveKit docs | Processing raw media tracks | LiveKit docs | Playing Audio | LiveKit docs