Attach call recordings to Langfuse traces

open-telemetry/langfuse/README.md

@@ -55,6 +55,50 @@ uv run bot.py
 
 Open your browser to [https://cloud.langfuse.com](https://cloud.langfuse.com) to view traces.
 
+## Call Recordings
+
+Each call's audio is attached to its trace, giving two levels of playback in the Langfuse UI:
+
+- **The whole call**, on the trace root — stereo, with the user on the left channel and the bot on the right, so barge-in stays audible.
+- **Each turn**, on that turn's `turn` span — the user's speech as the span's `input`, the bot's reply as its `output`.
+
+Langfuse media travels over the REST media API rather than OTLP, and is linked to a trace by id. Everything uploads after the pipeline shuts down, so recording adds nothing to the latency of the call itself.
+
+`langfuse_recording.py` holds the whole integration. Pipecat reports each turn's audio with the turn number attached, so what's left here is the Langfuse upload and you can lift the file into your own bot:
+
+```python
+audiobuffer = AudioBufferProcessor(
+    num_channels=2, enable_turn_audio=True, auto_start_recording=True
+)
+# ...place audiobuffer in the pipeline, after transport.output()...
+
+recorder = LangfuseRecorder.from_env()
+if recorder:
+    recorder.attach(audiobuffer, worker)
+```
+
+Then stop the recording while the pipeline is still alive, and upload once it isn't:
+
+```python
+@transport.event_handler("on_client_disconnected")
+async def on_client_disconnected(transport, client):
+    if recorder:
+        await recorder.stop_and_collect()
+    await worker.cancel()
+
+await runner.run()
+
+if recorder:
+    await recorder.upload()
+```
+
+Things worth knowing before running this against a real project:
+
+- **Credentials come from the OTLP config by default.** Langfuse authenticates its OTLP endpoint with HTTP Basic over the same key pair the REST API wants, so the recorder reads the keys out of `OTEL_EXPORTER_OTLP_HEADERS` and the host out of `OTEL_EXPORTER_OTLP_ENDPOINT`. Media and spans therefore reach the same project by construction. Setting `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_HOST` overrides that, and then it's on you to keep them pointed at the same project — mismatched keys give a 401 on upload while traces keep working, since OTLP authenticates separately.
+- **Per-turn clips are capped** at `MAX_TURN_CLIPS` (40) to stay under Langfuse's API rate limit — 30/min on Hobby, 100/min on Core. A longer call keeps its whole-call recording and the first 40 turns of clips. Raise the cap when self-hosting.
+- **The whole-call recording is capped** at `MAX_RECORDING_BYTES` (100MB, roughly 20 minutes of stereo 24kHz). Past that the upload is truncated and the recorder says so.
+- **Turn 1 begins when the pipeline starts**, not when the user first speaks, so a bot greeting is filed under turn 1 alongside whatever the user says next.
+
 ## Langfuse-Specific Configuration
 
 In the `bot.py` file, note the HTTP exporter configuration:

open-telemetry/langfuse/bot.py

@@ -19,6 +19,7 @@
     LLMContextAggregatorPair,
     LLMUserAggregatorParams,
 )
+from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -34,6 +35,8 @@
 from pipecat.utils.tracing.setup import setup_tracing
 from pipecat.workers.runner import WorkerRunner
 
+from langfuse_recording import LangfuseRecorder
+
 load_dotenv(override=True)
 
 IS_TRACING_ENABLED = bool(os.getenv("ENABLE_TRACING"))
@@ -117,6 +120,15 @@ async def on_function_calls_started(service, function_calls):
 
     conversation_id = str(uuid.uuid4())
 
+    # Stereo keeps the user on the left channel and the bot on the right, so
+    # barge-in stays audible in the recording. Turn audio adds one clip per
+    # speaker per turn, which is what the per-turn players in Langfuse play.
+    audiobuffer = AudioBufferProcessor(
+        num_channels=2,
+        enable_turn_audio=True,
+        auto_start_recording=True,
+    )
+
     pipeline = Pipeline(
         [
             transport.input(),
@@ -125,6 +137,9 @@ async def on_function_calls_started(service, function_calls):
             llm,
             tts,
             transport.output(),
+            # After the output transport, so both the user's audio and the bot's
+            # pass through it.
+            audiobuffer,
             assistant_aggregator,
         ]
     )
@@ -142,6 +157,10 @@ async def on_function_calls_started(service, function_calls):
         additional_span_attributes={"langfuse.session.id": conversation_id},
     )
 
+    recorder = LangfuseRecorder.from_env()
+    if recorder:
+        recorder.attach(audiobuffer, worker)
+
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
@@ -151,13 +170,21 @@ async def on_client_connected(transport, client):
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):
         logger.info(f"Client disconnected")
+        # Collect the audio before cancelling: the recording lives in the
+        # pipeline, which stops existing right after.
+        if recorder:
+            await recorder.stop_and_collect()
         await worker.cancel()
 
     runner = WorkerRunner(handle_sigint=False)
 
     await runner.add_workers(worker)
     await runner.run()
 
+    # Media is linked to the trace by id, so the spans no longer have to be open.
+    if recorder:
+        await recorder.upload()
+
 
 async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""

open-telemetry/langfuse/env.example

@@ -15,4 +15,15 @@ OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel"
 OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic <base64_encoded_api_keys>"
 
 # Set to any value to enable console output for debugging
-# OTEL_CONSOLE_EXPORT=true
+# OTEL_CONSOLE_EXPORT=true
+
+# Call recordings. Nothing to set: the recorder reads the same key pair and host
+# out of OTEL_EXPORTER_OTLP_HEADERS and OTEL_EXPORTER_OTLP_ENDPOINT above, so the
+# audio always lands in the project the traces went to.
+#
+# Set these to upload media somewhere else, e.g. a self-hosted deployment. Both
+# keys are needed, and the host has to be the same project as the OTLP endpoint
+# or the audio attaches to a trace that isn't there.
+# LANGFUSE_PUBLIC_KEY=pk-lf-...
+# LANGFUSE_SECRET_KEY=sk-lf-...
+# LANGFUSE_HOST="https://cloud.langfuse.com"