> This page is for version Nightly (default). > For other versions, use one of these documentation indexes: > - Nightly (default): https://docs.nvidia.com/nemo/labs-voice-agent/nightly/llms.txt > For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.nvidia.com/nemo/labs-voice-agent/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/labs-voice-agent/_mcp/server. # nemo_voice_agent.pipecat.services.nemo.audio_logger ## Module Contents ### Classes | Name | Description | | --------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | | [`AudioLogger`](#nemo_voice_agent-pipecat-services-nemo-audio_logger-AudioLogger) | Utility class for logging audio data and transcriptions during voice agent interactions. | | [`RTVIAudioLoggerObserver`](#nemo_voice_agent-pipecat-services-nemo-audio_logger-RTVIAudioLoggerObserver) | Observer that triggers audio logging when TranscriptionFrame is pushed. | ### API ```python 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 **File structure** log_dir/ ├── session_YYYYMMDD_HHMMSS/ │ ├── user/ │ │ ├── 00001_HHMMSS.wav │ │ ├── 00001_HHMMSS.json │ │ ├── 00002_HHMMSS.wav │ │ └── 00002_HHMMSS.json │ ├── agent/ │ │ ├── 00001_HHMMSS.wav │ │ ├── 00001_HHMMSS.json │ └── session_metadata.json # 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`** Base directory for storing logs (default: "./audio_logs") --- **`session_id`** Optional custom session ID. If None, auto-generated from timestamp --- **`enabled`** Whether logging is enabled (default: True) --- **`_agent_counter`** --- **`_lock`** --- **`_stereo_audio_buffer_left`** --- **`_stereo_audio_buffer_right`** --- **`_stereo_conversation_file`** --- **`_stereo_conversation_filename`** --- **`_turn_index`** --- **`_user_counter`** --- **`agent_dir`** --- **`continuous_user_audio_buffer`** --- **`log_dir`** --- **`session_dir`** --- **`session_metadata`** --- **`session_start_time`** --- **`turn_audio_buffer`** --- **`turn_transcription_buffer`** --- **`user_dir`** --- ```python 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`** Audio data as bytes or numpy array --- **`channel`** "left" for agent, "right" for user --- **`start_time`** Start time in seconds from session start --- **`sample_rate`** Sample rate of the input audio --- ```python 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. ```python 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`** Audio data as bytes (int16) or numpy array --- **`orig_sr`** Original sample rate --- **`target_sr`** Target sample rate --- **Returns:** `np.ndarray` Resampled audio as numpy array (float32) ```python 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`** Audio data as bytes or numpy array --- **`file_path`** Path to save the WAV file --- **`sample_rate`** Audio sample rate in Hz --- **`num_channels`** Number of audio channels (default: 1) --- ```python nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._save_metadata_json( metadata: dict, file_path: pathlib.Path ) ``` Save metadata to a JSON file. ```python nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger._save_session_metadata() ``` Save the session metadata to disk. ```python 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`** Raw audio data as bytes --- ```python 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. ```python nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger.finalize_session() ``` Finalize the session and save final metadata. ```python 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. ```python 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`** "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. ```python 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`** Generated audio data as bytes or numpy array --- **`text`** Input text that was synthesized --- **`sample_rate`** Audio sample rate in Hz (default: 22050) --- **`num_channels`** Number of audio channels (default: 1) --- **`additional_metadata`** Additional metadata to include --- **`tts_generation_time`** 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 ```python 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). ```python 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`** Whether this audio is a backchannel utterance (default: False) --- ```python 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`** The cutoff time in seconds from session start. If None, uses current time from session start. --- ```python 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. ```python 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`** Timestamp when the audio was received --- **`is_first_frame`** Whether this is the first frame of a turn (default: False) --- **`additional_metadata`** Additional metadata to include (e.g., model, backend info) --- ```python 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`** Timestamp when the audio was received --- **`transcription`** Transcribed text --- **`sample_rate`** Audio sample rate in Hz (default: 16000) --- **`num_channels`** Number of audio channels (default: 1) --- **`is_first_frame`** Whether this is the first frame of a turn (default: False) --- **`is_backchannel`** Whether this is a backchannel utterance (default: False) --- **`additional_metadata`** Additional metadata to include --- **Returns:** `Optional[dict]` Dictionary with logged file paths, or None if logging is disabled ```python 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. ```python 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.