Module Contents

Classes

Name	Description
`AudioLogger`	Utility class for logging audio data and transcriptions during voice agent interactions.
`RTVIAudioLoggerObserver`	Observer that triggers audio logging when TranscriptionFrame is pushed.

API

class nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger(
    log_dir: typing.Union[str, pathlib.Path] = './audio_logs',
    session_id: typing.Optional[str] = None,
    enabled: bool = True,
    user_audio_sample_rate: int = 16000,
    pre_roll_time_sec: float = 0.8,
    round_precision: int = 2
)

Utility class for logging audio data and transcriptions during voice agent interactions.

This logger saves:

Audio files in WAV format
Transcriptions with metadata in JSON format
Session information and metadata

12/19/2025 Note: Stereo conversation recording is implemented,

but -0.8 seconds offset needs to be applied to make the session sound synced.

Parameters:

log_dir

Union[str, Path]" default="'./audio_logs'

Base directory for storing logs (default: ”./audio_logs”)

session_id

Optional[str]" default="None

Optional custom session ID. If None, auto-generated from timestamp

enabled

bool" default="True

Whether logging is enabled (default: True)

_agent_counter

= 0

_lock

= threading.Lock()

_stereo_audio_buffer_left

list = []

_stereo_audio_buffer_right

list = []

_stereo_conversation_file

_stereo_conversation_filename

= 'conversation_stereo.wav'

_turn_index

= 0

_user_counter

= 0

agent_dir

= self.session_dir / 'agent'

continuous_user_audio_buffer

= []

log_dir

= Path(log_dir)

session_dir

= self.log_dir / session_id

session_metadata

session_start_time

= datetime.now()

turn_audio_buffer

= []

turn_transcription_buffer

= []

user_dir

= self.session_dir / 'user'

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._append_to_stereo_conversation(
    audio_data: typing.Union[bytes, numpy.ndarray],
    channel: str,
    start_time: float,
    sample_rate: int
)

Append audio to the stereo conversation buffer at the correct time position.

Parameters:

audio_data

Union[bytes, np.ndarray]

Audio data as bytes or numpy array

channel

str

“left” for agent, “right” for user

start_time

float

Start time in seconds from session start

sample_rate

int

Sample rate of the input audio

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._get_next_counter(
    speaker: str
) -> int

Get the next counter value for a speaker in a thread-safe manner.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._resample_audio(
    audio_data: typing.Union[bytes, numpy.ndarray],
    orig_sr: int,
    target_sr: int
) -> numpy.ndarray

Resample audio data to a target sample rate using librosa.

Parameters:

audio_data

Union[bytes, np.ndarray]

Audio data as bytes (int16) or numpy array

orig_sr

int

Original sample rate

target_sr

int

Target sample rate

Returns: np.ndarray

Resampled audio as numpy array (float32)

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._save_audio_wav(
    audio_data: typing.Union[bytes, numpy.ndarray],
    file_path: pathlib.Path,
    sample_rate: int,
    num_channels: int = 1
)

Save audio data to a WAV file.

Parameters:

audio_data

Union[bytes, np.ndarray]

Audio data as bytes or numpy array

file_path

Path

Path to save the WAV file

sample_rate

int

Audio sample rate in Hz

num_channels

int" default="1

Number of audio channels (default: 1)

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._save_metadata_json(
    metadata: dict,
    file_path: pathlib.Path
)

Save metadata to a JSON file.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._save_session_metadata()

Save the session metadata to disk.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.append_continuous_user_audio(
    audio_data: bytes
)

Append audio data to the continuous user audio buffer for stereo conversation.

This method should be called for EVERY audio frame received from the user, regardless of VAD state, to record the complete conversation audio.

Parameters:

audio_data

bytes

Raw audio data as bytes

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.clear_user_audio_buffer()

Clear the user audio buffer if the user stopped speaking detected by VAD.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.finalize_session()

Finalize the session and save final metadata.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.get_time_from_start_of_session(
    timestamp: datetime.datetime = None
) -> float

Get the time from the start of the session to the given datetime string.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.increment_turn_index(
    speaker: str = None
) -> int

Increment the turn index if the speaker has changed.

Parameters:

speaker

str" default="None

“user” or “agent”. If provided, only increments if this is different from the current speaker. If None, always increments.

Returns: int

The current turn index after any increment.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.log_agent_audio(
    audio_data: typing.Union[bytes, numpy.ndarray],
    text: str,
    sample_rate: int = 22050,
    num_channels: int = 1,
    additional_metadata: typing.Optional[dict] = None,
    tts_generation_time: typing.Optional[float] = None
) -> typing.Optional[dict]

Log agent audio and text (from TTS).

Parameters:

audio_data

Union[bytes, np.ndarray]

Generated audio data as bytes or numpy array

text

str

Input text that was synthesized

sample_rate

int" default="22050

Audio sample rate in Hz (default: 22050)

num_channels

int" default="1

Number of audio channels (default: 1)

additional_metadata

Optional[dict]" default="None

Additional metadata to include

tts_generation_time

Optional[float]" default="None

Time when TTS generation started (seconds from session start). Used to calculate actual start_time for first segment of a turn.

Returns: Optional[dict]

Dictionary with logged file paths, or None if logging is disabled

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.save_stereo_conversation()

Save the stereo conversation buffer to a WAV file. Left channel = Agent, Right channel = User.

User audio comes from continuous_user_audio_buffer (not affected by VAD).

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.save_user_audio(
    is_backchannel: bool = False,
    float_divisor: float = 32768.0
)

Save the user audio to the disk.

Parameters:

is_backchannel

bool" default="False

Whether this audio is a backchannel utterance (default: False)

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.set_agent_cutoff_time(
    cutoff_time: typing.Optional[float] = None
)

Set the cutoff time for the most recent agent audio entry.

This method should be called when TTS is interrupted by user speech. The cutoff_time represents when the agent audio was actually cut off, which may be earlier than the natural end_time.

Parameters:

cutoff_time

Optional[float]" default="None

The cutoff time in seconds from session start. If None, uses current time from session start.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.set_agent_turn_start_time()

Set the start time for the current agent turn.

This should be called when BotStartedSpeakingFrame is received, which indicates the audio is actually starting to play (not just generated). This provides more accurate timing than capturing time during TTS generation.

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.stage_turn_audio_and_transcription(
    timestamp_now: datetime.datetime,
    is_first_frame: bool = False,
    additional_metadata: typing.Optional[dict] = None
)

Stage the complete turn audio and accumulated transcriptions.

This method is called when a final transcription is received. It joins all accumulated audio and transcription chunks and stages them together.

Parameters:

timestamp_now

datetime

Timestamp when the audio was received

is_first_frame

bool" default="False

Whether this is the first frame of a turn (default: False)

additional_metadata

Optional[dict]" default="None

Additional metadata to include (e.g., model, backend info)

nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.stage_user_audio(
    timestamp_now: datetime.datetime,
    transcription: str,
    sample_rate: int = 16000,
    num_channels: int = 1,
    is_first_frame: bool = False,
    is_backchannel: bool = False,
    additional_metadata: typing.Optional[dict] = None
) -> typing.Optional[dict]

Stage user audio metadata and transcription (from STT). This data will be saved when the turn is complete by save_user_audio method. Audio data is retrieved from continuous_user_audio_buffer based on timestamps.

Parameters:

timestamp_now

datetime

Timestamp when the audio was received

transcription

str

Transcribed text

sample_rate

int" default="16000

Audio sample rate in Hz (default: 16000)

num_channels

int" default="1

Number of audio channels (default: 1)

is_first_frame

bool" default="False

Whether this is the first frame of a turn (default: False)

is_backchannel

bool" default="False

Whether this is a backchannel utterance (default: False)

additional_metadata

Optional[dict]" default="None

Additional metadata to include

Returns: Optional[dict]

Dictionary with logged file paths, or None if logging is disabled

class nemo_voice_agent.pipecat.services.nemo.audio_logger.RTVIAudioLoggerObserver(
    audio_logger: nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger
)

Bases: BaseObserver

Observer that triggers audio logging when TranscriptionFrame is pushed.

nemo_voice_agent.pipecat.services.nemo.audio_logger.RTVIAudioLoggerObserver.on_push_frame(
    data: pipecat.observers.base_observer.FramePushed
)

async

Handle frame push events and save user audio on TranscriptionFrame.