How to record a session using recorder_io

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

Can you share full code

Here is a complete example of how to use

RecorderIO

to record a session in LiveKit:

import asyncio
from livekit import RecorderIO

async def record_session(agent_session, input_audio, output_audio):
    # Initialize RecorderIO
    recorder_io = RecorderIO(
        agent_session=agent_session,
        sample_rate=48000,  # Optional, default is 48000 Hz
        loop=asyncio.get_event_loop()
    )

    # Wrap input and output audio streams
    input_audio = recorder_io.record_input(input_audio)
    output_audio = recorder_io.record_output(output_audio)

    # Start recording to a file
    await recorder_io.start(output_path="recording.ogg")

    print("Recording started. Press Ctrl+C to stop.")

    try:
        # Keep the recording running
        await asyncio.sleep(3600)  # Adjust duration as needed
    except KeyboardInterrupt:
        print("Stopping recording...")
    finally:
        # Stop and finalize the recording
        await recorder_io.aclose()
        print("Recording saved to recording.ogg")

# Example usage
# Replace `agent_session`, `input_audio`, and `output_audio` with actual objects
# asyncio.run(record_session(agent_session, input_audio, output_audio))

This script gently walks you through initializing

RecorderIO

, wrapping your input and output audio streams, and starting the recording to an OGG file. When you’re ready to stop, simply interrupt the process, and it will save your recording for you. Just be sure to replace

agent_session

,

input_audio

, and

output_audio

with the actual objects from your setup. If you have any questions or need a hand with anything else, please don’t hesitate to reach out—I’m here to help!

how to add this in my agent code ?

To help you smoothly integrate

RecorderIO

into your agent code, here’s a simple step-by-step guide: 1. Import the Required Modules: Make sure to import

RecorderIO

along with any other modules you’ll need in your agent code. 2. Initialize RecorderIO: Create an instance of

RecorderIO

during your agent session setup, passing in the session and any optional parameters like the sample rate. 3. Wrap Audio Streams: Use

record_input()

and

record_output()

to wrap the input and output audio streams your agent uses. 4. Start Recording: When the session begins, call

start()

with the file path where you want to save the recording (for example,

recording.ogg

). 5. Stop Recording: Don’t forget to call

aclose()

to properly finalize the recording when the session ends or the agent disconnects. Here’s a handy example to illustrate how you can put this all together:

import asyncio
from livekit import RecorderIO

class Agent:
    def __init__(self, session, input_audio, output_audio):
        self.session = session
        self.input_audio = input_audio
        self.output_audio = output_audio
        self.recorder = RecorderIO(agent_session=session, sample_rate=48000)

    async def start_session(self):
        # Wrap audio streams
        self.input_audio = self.recorder.record_input(self.input_audio)
        self.output_audio = self.recorder.record_output(self.output_audio)

        # Start recording
        await self.recorder.start(output_path="recording.ogg")
        print("Recording started.")

    async def end_session(self):
        # Stop recording
        await self.recorder.aclose()
        print("Recording finalized.")

# Example usage
# Replace `session`, `input_audio`, and `output_audio` with actual objects
# agent = Agent(session, input_audio, output_audio)
# asyncio.run(agent.start_session())

Feel free to integrate this into your agent’s lifecycle so the recording starts and stops exactly when it should. If you run into any questions or need a hand with anything else, I’m here to help!