Attach call recordings to Langfuse traces

changelog/5285.added.2.md

@@ -0,0 +1,5 @@
+- Added `AudioBufferProcessor.set_turn_tracker()`. Attach the pipeline's
+  `TurnTrackingObserver` and the `on_user_turn_audio_data` /
+  `on_bot_turn_audio_data` events report the tracker's turn number for each
+  clip, so turn audio can be matched to its turn (turn spans in tracing,
+  per-turn evals, storage keys) without hand-rolled counting.

changelog/5285.added.3.md

@@ -0,0 +1,2 @@
+- Added `pcm_to_wav()` to `pipecat.audio.utils`, wrapping raw s16le PCM (what
+  `AudioBufferProcessor` emits from its audio event handlers) in a WAV container.

changelog/5285.added.md

@@ -0,0 +1,6 @@
+- Added `LangfuseRecordingUploader` in `pipecat.utils.tracing.langfuse`. It captures
+  audio from an `AudioBufferProcessor` and, after the pipeline shuts down, uploads it
+  through the Langfuse media API and links it to the conversation's trace: the whole call
+  (stereo, user left / bot right) playable from the trace root, and each turn's user and
+  bot audio on that turn's `turn` span as its `input` and `output`. Uses only aiohttp; no
+  `langfuse` package dependency.

changelog/5285.changed.md

@@ -0,0 +1,4 @@
+- BREAKING: `AudioBufferProcessor`'s `on_user_turn_audio_data` and
+  `on_bot_turn_audio_data` events now pass a trailing `turn_number` argument
+  (0 when no turn tracker is attached). Existing handlers for these two events
+  need the parameter added to their signature.

src/pipecat/audio/utils.py

@@ -12,6 +12,8 @@
 """
 
 import audioop
+import io
+import wave
 
 import loudness
 import numpy as np
@@ -106,6 +108,30 @@ def interleave_stereo_audio(left_audio: bytes, right_audio: bytes) -> bytes:
     return stereo.astype(np.int16).tobytes()
 
 
+def pcm_to_wav(pcm: bytes, sample_rate: int, num_channels: int = 1) -> bytes:
+    """Wrap raw PCM audio in a WAV container.
+
+    The PCM data is expected to be signed 16-bit little-endian samples, which
+    is what Pipecat pipelines carry (e.g. what ``AudioBufferProcessor`` emits
+    from its audio event handlers).
+
+    Args:
+        pcm: Raw PCM audio data (16-bit signed integers).
+        sample_rate: Sample rate of the audio in Hz.
+        num_channels: Number of interleaved channels in the PCM data.
+
+    Returns:
+        A complete in-memory WAV file as bytes.
+    """
+    with io.BytesIO() as buffer:
+        with wave.open(buffer, "wb") as wav_file:
+            wav_file.setnchannels(num_channels)
+            wav_file.setsampwidth(2)
+            wav_file.setframerate(sample_rate)
+            wav_file.writeframes(pcm)
+        return buffer.getvalue()
+
+
 def normalize_value(value, min_value, max_value):
     """Normalize a value to the range [0, 1] and clamp it to bounds.
 

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -42,8 +42,10 @@ class AudioBufferProcessor(FrameProcessor):
 
     - on_audio_data: Triggered when buffer_size is reached, providing merged audio
     - on_track_audio_data: Triggered when buffer_size is reached, providing separate tracks
-    - on_user_turn_audio_data: Triggered when user turn has ended, providing that user turn's audio
-    - on_bot_turn_audio_data: Triggered when bot turn has ended, providing that bot turn's audio
+    - on_user_turn_audio_data: Triggered when user turn has ended, providing that user turn's
+      audio and its turn number (see :meth:`set_turn_tracker`)
+    - on_bot_turn_audio_data: Triggered when bot turn has ended, providing that bot turn's
+      audio and its turn number (see :meth:`set_turn_tracker`)
     - on_recording_started: Triggered when recording starts (state transitions to active)
     - on_recording_stopped: Triggered after recording stops and the final audio has been emitted
 
@@ -94,6 +96,7 @@ def __init__(
         self._bot_speaking = False
         self._user_turn_audio_buffer = bytearray()
         self._bot_turn_audio_buffer = bytearray()
+        self._turn_number = 0
 
         self._recording = False
 
@@ -155,6 +158,27 @@ def merge_audio_buffers(self) -> bytes:
         else:
             return b""
 
+    def set_turn_tracker(self, turn_tracker):
+        """Number turn audio events with the tracker's turn numbers.
+
+        The pipeline is usually built before the object that owns the turn
+        tracker (e.g. the pipeline worker), so the tracker is attached after
+        construction rather than passed to it.
+
+        Args:
+            turn_tracker: A ``TurnTrackingObserver``, e.g. the pipeline
+                worker's ``turn_tracking_observer``. When set, the
+                ``on_user_turn_audio_data`` and ``on_bot_turn_audio_data``
+                events report the turn number the audio belongs to, so a clip
+                can be matched to that turn elsewhere (turn spans in tracing,
+                per-turn evals, storage keys). Without a tracker the turn
+                number is always 0.
+        """
+
+        @turn_tracker.event_handler("on_turn_started")
+        async def on_turn_started(tracker, turn_number: int):
+            self._turn_number = turn_number
+
     async def start_recording(self):
         """Start recording audio from both user and bot.
 
@@ -359,12 +383,20 @@ async def _process_turn_recording(self, frame: Frame, resampled_audio: bytes | N
         # _process_recording so it is always up-to-date here.
         if isinstance(frame, UserStoppedSpeakingFrame):
             await self._call_event_handler(
-                "on_user_turn_audio_data", self._user_turn_audio_buffer, self.sample_rate, 1
+                "on_user_turn_audio_data",
+                self._user_turn_audio_buffer,
+                self.sample_rate,
+                1,
+                self._turn_number,
             )
             self._user_turn_audio_buffer = bytearray()
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self._call_event_handler(
-                "on_bot_turn_audio_data", self._bot_turn_audio_buffer, self.sample_rate, 1
+                "on_bot_turn_audio_data",
+                self._bot_turn_audio_buffer,
+                self.sample_rate,
+                1,
+                self._turn_number,
             )
             self._bot_turn_audio_buffer = bytearray()