Is there any way we can record audio before and after background noise cancelleation with livekit cloud Observability ?

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

I am using following setup 1. Livekit owned phone numbers 2. Python Agent is running on livekit cloud 3. using ai_coustics.audio_enhancement for background noise cancellation

Thanks — I’m replying about capturing pre- and post-noise-cancellation audio given your setup (LiveKit-owned phone numbers, Python agent running on LiveKit Cloud, and

ai_coustics.audio_enhancement

). Short answer • Observability records only the post-noise-cancellation audio. It does not provide a built-in way to record or export the raw audio before

ai_coustics.audio_enhancement

is applied. • To get both versions you’ll need to capture the raw audio yourself (outside Observability) and keep the Observability recording for the processed audio. How you can implement a workaround with your setup 1. Capture raw audio on the agent before enhancement ◦ Since your Python agent runs on LiveKit Cloud, add code in the agent process to tap the incoming audio frames prior to calling

ai_coustics.audio_enhancement

. ◦ Buffer those raw frames (for example, as PCM chunks or WAV) and either store them temporarily or stream them to your storage endpoint. 1. Keep Observability for post-processed audio ◦ Observability will continue to record the audio that has had

ai_coustics.audio_enhancement

applied. Use those recordings as your “after” audio. 1. Correlate recordings ◦ Ensure you timestamp or otherwise tag both recordings (session ID, call ID, and precise timestamps) so you can align raw and processed audio for comparison or analysis. Implementation notes and tips • Format: Store raw frames in a common format (16-bit PCM, sample rate matching the session) to simplify comparison against Observability output. • Performance: Buffering and exporting raw audio in real time may increase CPU/network usage on the agent — consider batching uploads or streaming to a dedicated ingest endpoint. • Privacy/compliance: Make sure storing raw (pre-processed) audio meets your privacy and regulatory requirements. • Where processing happens:

ai_coustics.audio_enhancement

is applied server-side in your agent flow; capturing the raw frames must occur before you call that enhancement function. If you’d like, I can: • Provide a small Python sketch showing where to intercept and save raw frames in an agent flow. • Suggest an efficient upload/streaming pattern for storing raw audio in near real-time. • Help with timestamping and sync strategies so you can align raw and processed audio during analysis. Which would you prefer? Also, confirm whether your agent is using the LiveKit Python SDK audio stream APIs directly (I can tailor the sample accordingly).

Can i use this to record call before background voice cancellation: from livekit import api async def entrypoint(ctx: JobContext): req = api.RoomCompositeEgressRequest( room_name=ctx.room.name, audio_only=True, file_outputs=[ api.EncodedFileOutput( file_type=api.EncodedFileType.OGG, filepath="livekit/my-room-test.ogg", s3=api.S3Upload( bucket=os.getenv("AWS_BUCKET_NAME"), region=os.getenv("AWS_REGION"), access_key=os.getenv("AWS_ACCESS_KEY_ID"), secret=os.getenv("AWS_SECRET_ACCESS_KEY"), ), ) ], ) lkapi = api.LiveKitAPI() await lkapi.egress.start_room_composite_egress(req) await lkapi.aclose() # ... continue with your agent logic

Thanks — I’m replying about whether that RoomCompositeEgress request will capture audio before

ai_coustics.audio_enhancement

(background noise cancellation). Short answer • No. Egress (RoomCompositeEgress) records the audio as it exists in the room after server-side processing. Observability and egress capture the post-noise-cancellation stream, so that code will not give you the raw pre-enhancement audio. Why • Noise cancellation from

ai_coustics.audio_enhancement

is applied server-side in the agent flow. The audio surfaced to the room and to egress/observability reflects that processed stream, so egress files will contain the post-processed audio. What you can do instead (recommended) 1. Intercept raw frames in your agent before calling

ai_coustics.audio_enhancement

. ◦ In your Python agent code, capture and buffer the incoming audio frames (PCM/WAV) before you pass them to

ai_coustics.audio_enhancement

. 1. Save or stream those raw frames separately. ◦ Example options: write to disk and upload to S3, stream to an ingest endpoint, or push to another service for storage/analysis. 1. Use RoomCompositeEgress (or Observability) to get the post-processed audio. ◦ Keep your egress job like the snippet you showed to produce the “after” recording. 1. Correlate files by session/room ID and timestamps so you can align raw vs processed audio. Offer to help • If you want, I can provide a small Python sketch showing where to intercept and save raw frames in the agent flow, or suggest an efficient streaming/upload pattern for near-real-time capture. Also let me know whether your agent receives audio via the LiveKit Python SDK audio callbacks or another input method so I can tailor the sample.

Please provide me a Python sketch

Thanks — I’m responding to your request for a Python sketch that captures raw audio frames before calling

ai_coustics.audio_enhancement

, while keeping LiveKit egress/observability for the post-processed audio. Below is a concise sketch showing the key ideas. It assumes: • Your Python agent uses an async loop that receives audio frames (PCM) from LiveKit or a telephony interface. • You will call

ai_coustics.audio_enhancement

on those frames for processing. • You want to write raw frames to disk (or stream/upload them) for “before” recordings, and keep your RoomCompositeEgress (or Observability) as the “after” recording. Adjust the I/O specifics to match how your agent receives frames (callback, queue, or SDK stream).

import os
import wave
import asyncio
from typing import AsyncIterator

# Placeholder: replace with actual import/usage of your ai_coustics API
# from ai_coustics import audio_enhancement

# Config
RAW_OUTPUT_DIR = "/tmp/raw_audio"
os.makedirs(RAW_OUTPUT_DIR, exist_ok=True)
SAMPLE_RATE = 16000          # match your session's sample rate
SAMPLE_WIDTH = 2             # bytes per sample (16-bit PCM)
CHANNELS = 1

# Utility: write PCM bytes to a WAV file
def write_pcm_to_wav(filepath: str, pcm_bytes: bytes, sample_rate: int = SAMPLE_RATE):
    with wave.open(filepath, "wb") as wf:
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(SAMPLE_WIDTH)
        wf.setframerate(sample_rate)
        wf.writeframes(pcm_bytes)

# Example async generator that yields incoming raw PCM frames from LiveKit / telephony
# Replace this with your actual frame source (callbacks / SDK)
async def incoming_pcm_frames() -> AsyncIterator[bytes]:
    """
    Yield raw PCM bytes for each frame/chunk you receive.
    This is a placeholder to illustrate how you would capture frames.
    """
    # Example: generate silence for illustration
    for i in range(10):
        await asyncio.sleep(0.02)  # simulate arrival timing
        yield b"\x00" * (SAMPLE_RATE // 50 * SAMPLE_WIDTH)  # ~20ms of silence
    return

# Main processing loop
async def process_call(room_id: str, call_id: str):
    raw_chunks = []
    chunk_index = 0

    async for pcm_chunk in incoming_pcm_frames():
        # 1) Capture raw chunk (before enhancement)
        raw_chunks.append(pcm_chunk)

        # Optional: stream/upload incrementally here instead of buffering everything
        # await upload_chunk_to_storage(pcm_chunk, metadata={...})

        # 2) Call the enhancement function (server-side in your flow)
        # enhanced_pcm = await audio_enhancement(pcm_chunk)
        # For sketch, we just pass the same chunk through:
        enhanced_pcm = pcm_chunk

        # 3) Send enhanced_pcm onward to the agent logic (STT/LLM/room)
        # await forward_enhanced_to_agent(enhanced_pcm)

        chunk_index += 1

    # After call ends, write buffered raw audio to WAV for "pre-enhancement" file
    raw_filepath = os.path.join(RAW_OUTPUT_DIR, f"{room_id}-{call_id}-raw.wav")
    write_pcm_to_wav(raw_filepath, b"".join(raw_chunks), SAMPLE_RATE)
    print("Wrote raw pre-enhancement audio to:", raw_filepath)

    # Your RoomCompositeEgress / Observability will produce the post-enhancement audio file.
    # Correlate by using room_id / call_id and timestamps.

# Example usage
async def entrypoint(ctx):
    # spawn or handle multiple calls as needed
    await process_call(room_id=ctx.room.name, call_id="call-123")

Notes and suggestions • Alignment: ◦ Include timestamps or sequence numbers with each chunk (e.g., attach UTC micros) when you upload or stream raw chunks so you can align them precisely with the egress/observability audio later. • Performance: ◦ For production, avoid buffering everything in memory. Stream chunks to S3 or to an ingest service as they arrive (multipart uploads, or append to an object store). ◦ Consider batching (e.g., 1–5 second buffers) before writing/uploading. • Formats: ◦ Use 16-bit PCM WAV at the same sample rate as the session to simplify comparison. • Privacy/compliance: ◦ Ensure storing raw audio complies with your policies and local regulations. • Integration points: ◦ If your LiveKit Python agent receives audio through callbacks, hook into that callback to capture frames. ◦ If using a telephony bridge that provides frames via gRPC or other stream API, capture frames right before calling

ai_coustics.audio_enhancement

. If you want, tell me: • How your agent currently receives audio frames (LiveKit SDK callbacks, gRPC, etc.), and • Whether you prefer streaming raw chunks to S3 or writing local files first. I can then provide a tailored code example showing: • How to incrementally upload to S3 (multipart), • Or how to hook into specific LiveKit Python SDK callbacks to grab frames.

LiveKit SDK callbacks and i want to write writing local files

Thanks — I’m replying with a concise, targeted Python sketch that uses LiveKit SDK callbacks to capture raw audio frames to local WAV files before noise cancellation, while you keep LiveKit egress/observability for the post-processed audio. Key points • Use the

track_subscribed

callback to access incoming audio via

rtc.AudioStream

. Frames you read there are the raw frames (pre-enhancement). • Write those frames to a local WAV file (16-bit PCM, matching the session sample rate). • Continue to use your existing

ai_coustics.audio_enhancement

call and LiveKit egress/observability to get the post-processed recording. • Add timestamps/IDs to filenames so you can correlate raw and processed files. Sketch (adapt to your imports and project structure):

import os
import wave
import asyncio
from livekit import rtc
from livekit.agents import JobContext, AgentServer

# Config - match your LiveKit session settings
SAMPLE_RATE = 16000
SAMPLE_WIDTH_BYTES = 2
CHANNELS = 1
RAW_DIR = "/tmp/raw_audio"
os.makedirs(RAW_DIR, exist_ok=True)

async def capture_raw_audio(track: rtc.RemoteAudioTrack, out_path: str):
    """
    Consume raw frames from rtc.AudioStream and write them to a WAV file.
    These frames are before any ai_coustics enhancement.
    """
    stream = rtc.AudioStream(track)
    # Open WAV for streaming write
    with wave.open(out_path, "wb") as wf:
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(SAMPLE_WIDTH_BYTES)
        wf.setframerate(SAMPLE_RATE)

        async for frame in stream:
            # frame.data is raw PCM bytes (match your SDK)
            wf.writeframes(frame.data)
    # When track ends / stream exits, this function returns and file is finalized.

# Example agent entrypoint using the LiveKit agent server decorator / pattern
server = AgentServer()  # Replace with actual server setup you use

@server.rtc_session(agent_name="audio-capture-agent")
async def entrypoint(ctx: JobContext):
    # Called when a track is subscribed in the room
    @ctx.room.on("track_subscribed")
    async def on_track_subscribed(
        track: rtc.RemoteAudioTrack,
        publication: rtc.RemoteTrackPublication,
        participant: rtc.RemoteParticipant
    ):
        # Only handle audio tracks
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return

        # Build a filename that helps you correlate later (room, participant, timestamp)
        filename = f"{ctx.room.name}_{participant.identity}_{publication.sid}.wav"
        out_path = os.path.join(RAW_DIR, filename)

        # Start capturing raw audio in the background (non-blocking)
        asyncio.create_task(capture_raw_audio(track, out_path))

    # Start your agent session as usual, with noise cancellation enabled
    # The noise cancellation will be applied downstream; egress/observability will record the post-processed audio.
    await server.start_session(ctx)  # Replace with your actual session start call

Practical notes and tips • Match sample rate/sampwidth/channels to your LiveKit audio config (common values: 16000 Hz, 16-bit, mono). • The sketch writes directly to local disk. For production, consider rotating files, disk space, or streaming/uploading chunks to S3 if storage is a concern. • Correlation: use room name, participant identity, publication SID, and timestamps in filenames to align raw files with egress/observability outputs. • Performance: writing per-frame is fine for short calls; for high throughput, buffer frames (e.g., 1–5s) and write periodically to reduce IO overhead. • Privacy/compliance: ensure storing raw pre-processed audio meets your legal and privacy requirements. If you want, I can: • Provide a version that buffers then flushes every N seconds. • Show an example that uploads chunks to S3 as they’re captured. • Tailor the sample to the exact LiveKit agent server/session API you use (share a short snippet of your current entrypoint). Sources: Agent session | LiveKit Documentation | Speech-to-text (STT) models overview | LiveKit Documentation | Capturing the First Few Seconds of a Call for Processing

Will it store both the channels ? and can you please provide a example to steam it GCP

Thanks — I’m replying to your two questions: (1) will the approach store multiple channels, and (2) an example to stream captured raw audio to GCP. I’ll assume you want to capture raw frames from the LiveKit SDK callback and write them locally and/or stream them to Google Cloud Storage (GCS). If your setup differs, tell me which part to adjust. 1) Channels (mono vs. multi-channel) • The frames you receive from

rtc.AudioStream

reflect the incoming track format. If the incoming track is mono, the frames are mono; if it’s multi-channel, frames will contain interleaved samples for each channel. • To preserve all channels, do NOT force mono when creating the WAV (i.e., set channels to the actual number of channels in the stream) and write the raw PCM bytes unmodified. If you instead set the WAV header to 1 channel while writing multi-channel bytes, the file will be interpreted incorrectly. • The code sketch below shows how to detect and use an explicit

channels

value. If your SDK doesn’t expose channel count on the frame/track object, you may need to confirm the session settings (typical telephony flows are mono; other sources can be stereo or multi-channel). 2) Sketch: capture raw frames via LiveKit SDK callback, write local WAV (multi-channel aware), and stream to Google Cloud Storage • This sketch: ◦ Hooks

track_subscribed

to start capturing. ◦ Writes a local WAV file preserving channel count and sample rate. ◦ Streams the same PCM chunks to GCS using resumable uploads (recommended for longer files). The example uses the Google Cloud Storage Python client with a simple resumable upload writer. You’ll need: • Google Cloud credentials set in the environment (e.g.,

GOOGLE_APPLICATION_CREDENTIALS

). • The

google-cloud-storage

Python package installed. Example sketch (adapt to your project structure and actual LiveKit imports):

import os
import wave
import asyncio
from datetime import datetime
from google.cloud import storage

# LiveKit imports - adapt to your installed SDK paths
from livekit import rtc
from livekit.agents import JobContext, AgentServer

# Config
RAW_DIR = "/tmp/raw_audio"
os.makedirs(RAW_DIR, exist_ok=True)
GCS_BUCKET = os.getenv("GCS_BUCKET")
SAMPLE_RATE = 16000         # set to your session sample rate
SAMPLE_WIDTH_BYTES = 2      # 16-bit PCM => 2 bytes
# If you can detect channels from track, override CHANNELS per-track
DEFAULT_CHANNELS = 1

# GCS helper: open a resumable upload writer for a target object
def gcs_resumable_writer(bucket_name: str, object_name: str):
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(object_name)
    # open a writable stream (resumable)
    return blob.open("wb")  # returns file-like object

async def capture_and_stream(track: rtc.RemoteAudioTrack, out_wav_path: str,
                             gcs_bucket: str, gcs_object: str,
                             sample_rate=SAMPLE_RATE,
                             sampwidth=SAMPLE_WIDTH_BYTES,
                             channels=DEFAULT_CHANNELS):
    """
    Read frames from track AudioStream, write to WAV preserving channels,
    and stream raw PCM to GCS in parallel.
    """
    stream = rtc.AudioStream(track)

    # Open local WAV for writing
    wf = wave.open(out_wav_path, "wb")
    wf.setnchannels(channels)
    wf.setsampwidth(sampwidth)
    wf.setframerate(sample_rate)

    # Open GCS writer (binary) for raw PCM (no WAV header) or for .wav if you prefer.
    # If you want to upload a proper WAV to GCS, write WAV locally and then upload the file at the end.
    gcs_obj_path = gcs_object
    gcs_writer = None
    if gcs_bucket:
        gcs_writer = gcs_resumable_writer(gcs_bucket, gcs_obj_path)

    try:
        async for frame in stream:
            # frame.data should be raw PCM bytes, interleaved if multi-channel.
            pcm = frame.data

            # Write to WAV (preserves channels per header)
            wf.writeframes(pcm)

            # Stream raw PCM to GCS as you go
            if gcs_writer:
                gcs_writer.write(pcm)
    finally:
        # finalize local WAV file
        try:
            wf.close()
        except Exception:
            pass
        # finalize GCS writer
        if gcs_writer:
            try:
                gcs_writer.close()
            except Exception:
                pass

# Example AgentServer entrypoint with track_subscribed handler
server = AgentServer()  # adapt to your server initialization

@server.rtc_session(agent_name="audio-capture-agent")
async def entrypoint(ctx: JobContext):
    @ctx.room.on("track_subscribed")
    async def on_track_subscribed(track: rtc.RemoteAudioTrack,
                                  publication: rtc.RemoteTrackPublication,
                                  participant: rtc.RemoteParticipant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return

        # Determine channels if available on track/publication; fallback to DEFAULT_CHANNELS
        channels = getattr(track, "channels", DEFAULT_CHANNELS)
        sample_rate = getattr(track, "sample_rate", SAMPLE_RATE)
        sampwidth = SAMPLE_WIDTH_BYTES

        ts = datetime.utcnow().strftime("%Y%m%dT%H%M%S")
        filename = f"{ctx.room.name}_{participant.identity}_{publication.sid}_{ts}.wav"
        local_path = os.path.join(RAW_DIR, filename)

        # GCS object name (we upload raw PCM stream as .pcm, or you can upload WAV after closed)
        gcs_object_name = f"raw_audio/{filename}"

        # Start background capture and streaming
        asyncio.create_task(capture_and_stream(
            track=track,
            out_wav_path=local_path,
            gcs_bucket=GCS_BUCKET,
            gcs_object=gcs_object_name,
            sample_rate=sample_rate,
            sampwidth=sampwidth,
            channels=channels
        ))

    # Start the agent session as you normally do (with noise cancellation enabled downstream)
    await server.start_session(ctx)  # adapt to your actual start call

Notes, best practices, and caveats • Channel detection: The sketch attempts to use

track.channels

or

track.sample_rate

if exposed. If the SDK doesn’t expose these, you must use your session configuration or the telephony bridge settings. Most PSTN inbound calls are mono. • WAV vs raw PCM: The example writes a WAV locally (header + PCM). For streaming to GCS you can either: ◦ Stream raw PCM bytes directly (no header) and later reconstruct into WAV on download/processing, or ◦ Buffer locally and upload the completed WAV file at the end (simpler to consume). • Resumable uploads: The

blob.open("wb")

call uses a resumable upload under the hood. For very long calls, consider chunked/multipart strategies or uploading completed WAVs after rotation. • File rotation and memory/IO: For long calls, avoid writing every small frame directly to disk. Buffer for 1–5 seconds, or let wave.writeframes be called but rely on OS buffering. If storage is limited, stream directly to GCS and delete local files after upload. • Post-processed audio: LiveKit egress/observability will record the processed audio (post-noise-cancellation). Use filenames with room/participant/publication/timestamps to correlate raw vs processed recordings. • Privacy/compliance: Ensure you have consent and meet regulatory requirements before storing raw pre-processed audio. If you’d like: • I can change the sketch to buffer and flush every N seconds (recommended) and show how to upload each chunk as a separate object (useful for near-real-time processing). • I can show a flow that finalizes the WAV locally and uploads the completed WAV to GCS after the track ends. • Share a short snippet of your actual LiveKit track object (attributes you see) and I’ll tailor the channel/sample-rate detection. Sources: Agent session | LiveKit Documentation | Capturing the First Few Seconds of a Call for Processing