is there a way to track user volume?

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

please provide an example of how we could write a txt file with the colume of the user and the AI agent

Do you mean audio volume levels (loudness) or the number of participants? Also which environment/language do you want the example in — Python Agent, Node/JS (frontend), or another? And what output format do you prefer (CSV with timestamp, participant type, identity, volume)? Sources: Subscribing to tracks | LiveKit docs | Job lifecycle | LiveKit docs | Agent speech and audio | LiveKit docs

I mean volume levels, like loudness. When looking at our tracks it seems like users speaks quietly, although when calling them directly their audio is fine

Responding to your request: below is a concise Python Agent example that logs loudness (audio level) for both a remote user and the AI agent into a CSV (text) file with timestamp, identity, role, and level. Important notes: • This uses the

audio_level

event available on tracks/participants (range 0.0–1.0). • Identify "agent" vs "user" by participant.identity (adjust logic if you use metadata). • Sample rate here is 1s; change as needed.

import asyncio
from datetime import datetime
from livekit import agents, rtc

LOG_PATH = "volume_log.csv"

# Helper to append CSV header
with open(LOG_PATH, "w", buffering=1) as f:
    f.write("timestamp,identity,role,level\n")

def log_level(identity: str, role: str, level: float):
    ts = datetime.utcnow().isoformat()
    with open(LOG_PATH, "a", buffering=1) as f:
        f.write(f"{ts},{identity},{role},{level:.4f}\n")

async def monitor_remote_participant(participant: rtc.RemoteParticipant):
    # subscribe to existing audio tracks
    for pub in participant.audio_tracks:
        if pub.track:
            pub.track.on("audioLevel", lambda lvl, id=participant.identity: log_level(id, "user", lvl))

    # handle future published tracks
    @participant.on("trackPublished")
    def _on_published(pub: rtc.RemoteTrackPublication):
        @pub.on("subscribed")
        def _on_subscribed(track: rtc.RemoteAudioTrack):
            track.on("audioLevel", lambda lvl, id=participant.identity: log_level(id, "user", lvl))

async def monitor_local_audio(local_participant: rtc.LocalParticipant):
    for pub in local_participant.audio_tracks:
        if pub.track:
            pub.track.on("audioLevel", lambda lvl, id=local_participant.identity: log_level(id, "agent", lvl))

    @local_participant.on("trackPublished")
    def _on_local_published(pub: rtc.LocalTrackPublication):
        if pub.track and pub.track.kind == rtc.TrackKind.KIND_AUDIO:
            pub.track.on("audioLevel", lambda lvl, id=local_participant.identity: log_level(id, "agent", lvl))

@agents.room_job()
def entrypoint(job: agents.JobContext):
    return VolumeAgent().run(job)

class VolumeAgent(agents.JobProcess):
    async def run(self, job: agents.JobContext):
        # connect to the room (auto_subscribe as appropriate)
        room = await job.connect(auto_subscribe=agents.AutoSubscribe.AUDIO_ONLY)

        # Start monitoring existing remote participants
        for p in room.remote_participants.values():
            asyncio.create_task(monitor_remote_participant(p))

        # Monitor future participant connections
        @room.on("participantConnected")
        def _on_participant_connected(participant: rtc.RemoteParticipant):
            asyncio.create_task(monitor_remote_participant(participant))

        # Monitor agent's local audio
        asyncio.create_task(monitor_local_audio(room.local_participant))

        # Keep agent running; adjust exit conditions as needed
        try:
            while True:
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            return

If you want: • a Node/browser example instead, • RMS-based server-side volume from raw audio frames, • or help identifying why users sound quiet (noise suppression, gain, client-side scaling), tell me which you prefer and I’ll provide it. Sources: Subscribing to tracks | LiveKit docs | Understanding "silence detected on local audio track" warning | Agent speech and audio | LiveKit docs | Agent insights in LiveKit Cloud | LiveKit docs

could you give me a class that does this, wich I can just call on in my agent. And thell me where I should add the class

Below is a self-contained Python class you can drop into your agent code. It encapsulates monitoring and CSV logging of audio levels for remote participants (users) and the local agent. After the class I show where to instantiate and start it in your agent's

run

method. Save this class in a module (e.g.,

volume_monitor.py

) and import/use it from your agent file.

# volume_monitor.py
import asyncio
from datetime import datetime
from typing import Optional
from livekit import rtc

class VolumeMonitor:
    """
    Monitor audio levels for remote participants (users) and the local participant (agent).
    Writes CSV rows: timestamp,identity,role,level
    """

    def __init__(self, room: rtc.Room, log_path: str = "volume_log.csv", sample_interval: float = 1.0):
        self.room = room
        self.log_path = log_path
        self.sample_interval = sample_interval
        self._tasks = []
        # write header
        with open(self.log_path, "w", buffering=1) as f:
            f.write("timestamp,identity,role,level\n")

    def _write_row(self, identity: str, role: str, level: float):
        ts = datetime.utcnow().isoformat()
        with open(self.log_path, "a", buffering=1) as f:
            f.write(f"{ts},{identity},{role},{level:.4f}\n")

    def _attach_track_handlers(self, identity: str, role: str, track: rtc.AudioTrack):
        # handler invoked frequently by SDK when audio level updates
        try:
            track.on("audioLevel", lambda lvl: self._write_row(identity, role, lvl))
        except Exception:
            # if event wiring fails, fallback to periodic sampling if track exposes .audio_level
            async def sampler():
                while True:
                    lvl = getattr(track, "audio_level", None)
                    if lvl is not None:
                        self._write_row(identity, role, float(lvl))
                    await asyncio.sleep(self.sample_interval)
            t = asyncio.create_task(sampler())
            self._tasks.append(t)

    async def _monitor_existing_remote(self):
        # Attach to already-connected remote participants
        for participant in list(self.room.remote_participants.values()):
            await self._setup_remote_participant(participant)

    async def _setup_remote_participant(self, participant: rtc.RemoteParticipant):
        identity = participant.identity
        # attach handlers for current audio publications
        for pub in list(participant.audio_tracks):
            if pub.track:
                self._attach_track_handlers(identity, "user", pub.track)

        # when new track published later
        @participant.on("trackPublished")
        def _on_published(pub: rtc.RemoteTrackPublication):
            @pub.on("subscribed")
            def _on_subscribed(track: rtc.RemoteAudioTrack):
                self._attach_track_handlers(identity, "user", track)

    def _setup_remote_events(self):
        @self.room.on("participantConnected")
        def _on_participant_connected(participant: rtc.RemoteParticipant):
            # start async task to attach handlers (keeps sync boundaries safe)
            t = asyncio.create_task(self._setup_remote_participant(participant))
            self._tasks.append(t)

    def _setup_local_audio(self):
        lp = self.room.local_participant
        identity = lp.identity
        # attach handlers for existing local audio tracks
        for pub in list(lp.audio_tracks):
            if pub.track:
                self._attach_track_handlers(identity, "agent", pub.track)

        # when agent publishes audio later
        @lp.on("trackPublished")
        def _on_local_published(pub: rtc.LocalTrackPublication):
            if pub.track and pub.track.kind == rtc.TrackKind.KIND_AUDIO:
                self._attach_track_handlers(identity, "agent", pub.track)

    async def start(self):
        """
        Start monitoring: attach handlers and begin any necessary tasks.
        Call from your agent after connecting to the room.
        """
        # attach to current room state
        await self._monitor_existing_remote()
        self._setup_remote_events()
        self._setup_local_audio()

    async def stop(self, cancel_tasks: bool = True):
        """
        Stop monitoring. Cancels internal tasks if cancel_tasks True.
        """
        if cancel_tasks:
            for t in list(self._tasks):
                t.cancel()
        # no persistent resources to close beyond tasks

Where to add and how to use it in your agent: 1. Save the class above as

volume_monitor.py

(or paste into your agent module). 2. In your agent file, after connecting to the room in your

run

method, instantiate and start it:

# agent.py (snippet)
from livekit import agents
from volume_monitor import VolumeMonitor

class MyAgent(agents.JobProcess):
    async def run(self, job: agents.JobContext):
        room = await job.connect(auto_subscribe=agents.AutoSubscribe.AUDIO_ONLY)

        # Create and start the volume monitor
        monitor = VolumeMonitor(room, log_path="volume_log.csv", sample_interval=1.0)
        await monitor.start()

        try:
            # your existing loop / logic
            while True:
                await asyncio.sleep(1)
        finally:
            await monitor.stop()

Notes and tips: • Identification: the class treats any remote participant as "user" and the local participant as "agent". If you need more strict role detection, change the identity-based logic before writing rows. • Sampling vs events: the class prefers SDK

audioLevel

events; if not available it falls back to periodic sampling of

track.audio_level

. • Log file location: use an absolute path if your agent runs in a container or different working directory. • If you want per-second averaging instead of raw events, I can modify the class to accumulate samples and write averaged values. Would you like me to adapt this to a different file layout (single-file agent) or add RMS-from-raw-audio framing for a server-side loudness metric?