Attach per-turn call audio to the Langfuse trace

open-telemetry/README.md

@@ -34,7 +34,7 @@ This organization helps you track conversation-to-conversation and turn-to-turn
 | Demo                            | Description                                                               |
 | ------------------------------- | ------------------------------------------------------------------------- |
 | [Jaeger Tracing](./jaeger/)     | Tracing with Jaeger, an open-source end-to-end distributed tracing system |
-| [Langfuse Tracing](./langfuse/) | Tracing with Langfuse, a specialized platform for LLM observability       |
+| [Langfuse Tracing](./langfuse/) | Tracing with Langfuse, a specialized platform for LLM observability. Also attaches the call recording so you can play it back from the trace |
 | [LangSmith Tracing](./langsmith/) | Tracing with LangSmith, LangChain's platform for LLM observability       |
 | [Opik Tracing](./opik/)         | Tracing with Opik, an open-source tracing and evaluation platform         |
 

open-telemetry/langfuse/README.md

@@ -35,6 +35,11 @@ OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20<base64_encoded_api_key>
 DEEPGRAM_API_KEY=your_key_here
 CARTESIA_API_KEY=your_key_here
 OPENAI_API_KEY=your_key_here
+
+# Optional: needed only to attach the call recording (see below)
+LANGFUSE_PUBLIC_KEY=pk-lf-...
+LANGFUSE_SECRET_KEY=sk-lf-...
+LANGFUSE_HOST=https://cloud.langfuse.com
 ```
 
 ### 3. Set up a venv and install Dependencies
@@ -73,12 +78,127 @@ setup_tracing(
 )
 ```
 
+### Recording the conversation audio
+
+Set `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`, and `LANGFUSE_HOST` and the demo also
+attaches the call audio to the trace, so you can listen while reading the spans. Leave them
+unset and you get traces without audio.
+
+You get audio at two levels:
+
+- **The whole call**, on the trace root. Stereo, user on the left channel and bot on the
+  right, so an interruption reads as overlap.
+- **Each turn**, on that turn's `turn` span. The user's speech is filed as the span's
+  **input** and the bot's reply as its **output**, so clicking a turn plays just that turn
+  and you can hear each side separately.
+
+Two credentials for one service looks odd, so here is why. Spans travel over OTLP, which
+only needs the pre-encoded `OTEL_EXPORTER_OTLP_HEADERS`. Langfuse media does not travel
+over OTLP: audio is uploaded through the media REST API, and that call needs the keys
+unencoded.
+
+The heavy lifting is Pipecat's `LangfuseRecordingUploader`
+(`pipecat.utils.tracing.langfuse`), fed by an
+[`AudioBufferProcessor`](https://docs.pipecat.ai/pipecat/fundamentals/recording-audio).
+This is the whole integration:
+
+```python
+from pipecat.utils.tracing.langfuse import LangfuseRecordingUploader
+
+# Stereo whole-call recording, plus a clip per speaker per turn.
+audiobuffer = AudioBufferProcessor(
+    num_channels=2,
+    buffer_size=0,
+    enable_turn_audio=True,
+)
+
+uploader = LangfuseRecordingUploader.from_env() if IS_TRACING_ENABLED else None
+if uploader:
+    # The turn tracker numbers the turns, which is how each clip finds its span.
+    uploader.attach(audiobuffer, turn_tracker=worker.turn_tracking_observer)
+
+# ...the processor goes after transport.output() in the pipeline, so it records what
+# was actually played, including bot speech cut short by an interruption...
+
+@transport.event_handler("on_client_connected")
+async def on_client_connected(transport, client):
+    await audiobuffer.start_recording()
+
+@transport.event_handler("on_client_disconnected")
+async def on_client_disconnected(transport, client):
+    # Collect the audio while the pipeline is still up; upload after it shuts down.
+    await uploader.stop_and_collect(audiobuffer, worker)
+    await worker.cancel()
+
+# After the runner returns:
+await uploader.upload(worker)
+```
+
+> `LangfuseRecordingUploader` is not in a pipecat release yet
+> ([pipecat-ai/pipecat#5285](https://github.com/pipecat-ai/pipecat/pull/5285)), so this
+> example's `pyproject.toml` installs pipecat from that PR's branch. It will move back to
+> a version pin once the PR ships.
+
+Three behaviors of the uploader worth knowing:
+
+- **Nothing is written into the span payload.** Langfuse renders a player from the media
+  link alone, so a clip only needs the trace id, and for per-turn audio the turn's span id.
+  Those ids outlive the spans, which is why all uploading happens after the pipeline has
+  shut down and a slow upload never delays teardown.
+- **Per-turn uploads are capped** at `max_turn_clips` turns (40 by default). Each clip costs
+  two calls against Langfuse's general API rate limit, which is 30/min on Hobby and 100/min
+  on Core, so a very long call degrades to "the first N turns have audio" instead of a wall
+  of 429s. A 429 is retried once using `Retry-After`.
+- **The recording is capped** at 100MB of audio and truncated with a warning past that.
+  Stereo 24kHz 16-bit is roughly 350MB per hour, so a long-running agent would otherwise
+  grow without bound.
+
+### Running the audio eval
+
+The bot exposes an `eval` transport, so Pipecat's eval harness can drive it with
+synthesized speech and judge the result. This is the quickest way to check that the
+recording still works after a change.
+
+`--trigger-disconnect` is required: the upload happens in `on_client_disconnected`, so
+without it the recording is never finalized.
+
+```bash
+# Terminal 1: the bot, with tracing enabled so a real upload happens
+uv run bot.py -t eval --port 7860
+
+# Terminal 2: from a pipecat checkout
+uv run pipecat eval run scripts/release-evals/scenarios/interruption_audio.yaml \
+  --bot-url ws://localhost:7860 \
+  -a -v --trigger-disconnect \
+  --record-dir /tmp/langfuse-eval-recordings -t 90
+```
+
+The eval writes its own copy of the recording to `--record-dir`, which is handy as a
+reference to compare against the audio that landed on the trace. `interruption_audio`
+is the interesting scenario because it barges in on the bot. Requires
+`pipecat-ai[evals]` and a judge (Ollama serving `gemma4:12b` by default).
+
 ## Troubleshooting
 
 - **No Traces in Langfuse**: Ensure that your credentials are correct and follow this [troubleshooting guide](https://langfuse.com/faq/all/missing-traces)
 - **Connection Errors**: Verify network connectivity to Langfuse
 - **Authorization Issues**: Check that your base64 encoding is correct and the API keys are valid
+- **`Failed to export span batch code: 401`**: check that the space in
+  `OTEL_EXPORTER_OTLP_HEADERS` is written as `%20`, not as a literal space. Header values in
+  that variable are URL encoded per the OTLP spec, and a literal space makes the SDK discard
+  the header without sending it, which Langfuse answers with a 401.
+- **Traces but no audio player**: `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` are probably unset. The bot logs `Langfuse recording disabled` at startup when that is the case, and `Langfuse recording attached: mediaId=...` on success.
+- **Traces and an upload log, but still no player**: make sure `LANGFUSE_HOST` and
+  `OTEL_EXPORTER_OTLP_ENDPOINT` name the same region. Mixing EU and US puts the audio in one
+  project and the trace in another. The bot warns about this at startup.
+- **Whole-call audio but no per-turn audio**: the processor needs `enable_turn_audio=True`
+  and the uploader needs the turn tracker, otherwise there is no way to match a clip to its
+  span. The bot logs `no turn tracker, per-turn audio disabled` in that case.
+- **A just-finished trace 404s in the UI**: give it a minute. The API returns the trace
+  immediately, but the trace view can lag behind ingestion.
 
 ## References
 
 - [Langfuse OpenTelemetry Documentation](https://langfuse.com/docs/opentelemetry/get-started)
+- [Langfuse Multi-Modality (media attachments)](https://langfuse.com/docs/observability/features/multi-modality)
+- [Pipecat: Recording Conversation Audio](https://docs.pipecat.ai/pipecat/fundamentals/recording-audio)

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
@@ -31,6 +32,7 @@
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.utils.tracing.langfuse import LangfuseRecordingUploader
 from pipecat.utils.tracing.setup import setup_tracing
 from pipecat.workers.runner import WorkerRunner
 
@@ -62,6 +64,17 @@ async def get_current_weather(params: FunctionCallParams, location: str, format:
     await params.result_callback({"conditions": "nice", "temperature": "75"})
 
 
+def _eval_transport_params():
+    """Params for the eval transport, imported lazily.
+
+    The eval harness ships as an optional extra (``pipecat-ai[evals]``), so importing it
+    at module scope would make the demo unrunnable without it.
+    """
+    from pipecat.evals.transport import EvalTransportParams
+
+    return EvalTransportParams(audio_in_enabled=True, audio_out_enabled=True)
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -78,6 +91,8 @@ async def get_current_weather(params: FunctionCallParams, location: str, format:
         audio_in_enabled=True,
         audio_out_enabled=True,
     ),
+    # Lets `pipecat eval run` drive this bot. See the README for the command.
+    "eval": lambda: _eval_transport_params(),
 }
 
 
@@ -117,6 +132,19 @@ async def on_function_calls_started(service, function_calls):
 
     conversation_id = str(uuid.uuid4())
 
+    # Records the call so it can be played back from the Langfuse trace. Stereo keeps the
+    # user on the left and the bot on the right, so an interruption reads as overlap.
+    # buffer_size=0 means the whole recording arrives in a single on_audio_data event when
+    # recording stops. enable_turn_audio also emits a clip per speaker per turn, which is
+    # what gets attached to the individual turn spans.
+    audiobuffer = AudioBufferProcessor(
+        num_channels=2,
+        buffer_size=0,
+        enable_turn_audio=True,
+    )
+
+    uploader = LangfuseRecordingUploader.from_env() if IS_TRACING_ENABLED else None
+
     pipeline = Pipeline(
         [
             transport.input(),
@@ -125,6 +153,9 @@ async def on_function_calls_started(service, function_calls):
             llm,
             tts,
             transport.output(),
+            # After transport.output() so we capture what was actually played, including
+            # bot speech cut short by an interruption.
+            audiobuffer,
             assistant_aggregator,
         ]
     )
@@ -142,22 +173,40 @@ async def on_function_calls_started(service, function_calls):
         additional_span_attributes={"langfuse.session.id": conversation_id},
     )
 
+    if uploader:
+        # The turn tracker numbers the turns, which is how each clip finds its turn span.
+        uploader.attach(audiobuffer, turn_tracker=worker.turn_tracking_observer)
+
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
+        await audiobuffer.start_recording()
         # Kick off the conversation.
         await worker.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):
         logger.info(f"Client disconnected")
+
+        # Collect the audio while the pipeline is still up. Uploading it can wait: Langfuse
+        # links media by trace and observation id, so no span needs to be open for it.
+        if uploader:
+            await uploader.stop_and_collect(audiobuffer, worker)
+        else:
+            await audiobuffer.stop_recording()
+
         await worker.cancel()
 
     runner = WorkerRunner(handle_sigint=False)
 
     await runner.add_workers(worker)
     await runner.run()
 
+    # Upload after the pipeline has shut down, so a slow upload never delays teardown or
+    # holds the call open. The trace and turn span ids outlive the spans themselves.
+    if uploader:
+        await uploader.upload(worker)
+
 
 async def bot(runner_args: RunnerArguments):
     """Main bot entry point compatible with Pipecat Cloud."""

open-telemetry/langfuse/env.example

@@ -12,7 +12,20 @@ OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel"
 # 🏠 Local deployment (>= v3.22.0)
 # OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:3000/api/public/otel"
 
-OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic <base64_encoded_api_keys>"
+# The space after "Basic" must be written as %20. Header values in this variable are
+# URL encoded per the OTLP spec, and a literal space makes the SDK drop the header
+# silently, so every span export comes back 401.
+OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic%20<base64_encoded_api_keys>"
 
 # Set to any value to enable console output for debugging
-# OTEL_CONSOLE_EXPORT=true
+# OTEL_CONSOLE_EXPORT=true
+
+# Optional: attach the call recording to the trace so you can play it back.
+# Traces go out over OTLP above, but Langfuse media is uploaded through the REST API,
+# which needs the keys unencoded. Leave these unset to get traces without audio.
+LANGFUSE_PUBLIC_KEY=pk-lf-...
+LANGFUSE_SECRET_KEY=sk-lf-...
+# Match the data region you chose above.
+LANGFUSE_HOST="https://cloud.langfuse.com"
+# LANGFUSE_HOST="https://us.cloud.langfuse.com"
+# LANGFUSE_HOST="http://localhost:3000"

open-telemetry/langfuse/pyproject.toml

@@ -4,7 +4,10 @@ version = "0.1.0"
 description = "A Pipecat example using Langfuse tracing"
 requires-python = ">=3.11"
 dependencies = [
-    "pipecat-ai[daily,webrtc,websocket,silero,cartesia,deepgram,openai,tracing,runner]>=1.4.0",
+    # LangfuseRecordingUploader (pipecat.utils.tracing.langfuse) is not released yet, so
+    # install pipecat from the PR branch (pipecat-ai/pipecat#5285). Switch back to a
+    # version pin once it ships.
+    "pipecat-ai[daily,webrtc,websocket,silero,cartesia,deepgram,openai,tracing,runner] @ git+https://github.com/pipecat-ai/pipecat.git@jh/turn-audio-turn-number",
     "pipecatcloud>=0.7.1",
     "opentelemetry-exporter-otlp-proto-http",
 ]