Add Pipecat voice traces and opt-in audio

examples/README.md

@@ -79,6 +79,7 @@ Unless noted otherwise, every example below uses `braintrust.auto_instrument()`.
 | `openai_agents/` | OpenAI Agents SDK `Runner` running an agent |
 | `openrouter/` | OpenRouter chat completion routed to OpenAI |
 | `otel/` | OpenTelemetry interop — `BraintrustSpanProcessor`, filtering, distributed tracing |
+| [pipecat/](pipecat/) | Cascade and realtime voice agents with tool calls and recordings — uses `setup_pipecat()` |
 | `pydantic_ai/` | Pydantic AI agent run inside a `start_span` for a permalink |
 | `strands/` | Strands `Agent` against `gpt-4o-mini` |
 | `temporal/` | Distributed Temporal workflow tracing via `BraintrustPlugin` |

examples/pipecat/README.md

@@ -0,0 +1,28 @@
+# Pipecat voice tracing
+
+Two small voice agents using real OpenAI services and the local Braintrust SDK:
+
+| Example | Pipeline |
+| --- | --- |
+| `cascade.py` | Speech → OpenAI transcription → LLM + order lookup → synthesized speech |
+| `realtime.py` | Speech → OpenAI Realtime + order lookup → speech |
+
+Requires Python 3.11–3.13, `uv`, `OPENAI_API_KEY`, and `BRAINTRUST_API_KEY` in your environment. From the repository root, run either command:
+
+```sh
+uv run --project examples/pipecat python examples/pipecat/cascade.py
+uv run --project examples/pipecat python examples/pipecat/realtime.py
+```
+
+Each command installs its dependencies, runs one conversation, and prints a trace link in the `example-pipecat` project. OpenAI usage is billable.
+
+The bundled `order.wav` (about 82 KiB) asks “Where is my order number one zero four two?” `common.py` supplies file input and silent output so no microphone, speaker, browser, or phone setup is needed. The local `lookup_order` tool returns a fictional delivery date. Listen to the captured audio in the trace.
+
+Instrumentation is enabled with:
+
+```python
+logger = braintrust.init_logger(project="example-pipecat")
+setup_pipecat(capture_audio_attachments=True)
+```
+
+Pipecat metrics are enabled in `PipelineParams`. The trace includes turns, model calls, tool execution, metrics, and Ogg audio attachments. Set `capture_audio_attachments=False` to trace without recording audio.

examples/pipecat/cascade.py

@@ -0,0 +1,62 @@
+"""Audio → transcription → model/tool → synthesized speech, traced by Braintrust."""
+
+import asyncio
+import os
+
+import braintrust
+from braintrust.integrations.pipecat import setup_pipecat
+from common import INSTRUCTIONS, RecordedInput, SilentOutput, order_context, run_conversation
+
+# Pipecat requires Python 3.11+ and is installed by this example, not the shared lint environment.
+# pylint: disable=import-error
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair, LLMUserAggregatorParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.openai.stt import OpenAISTTService
+from pipecat.services.openai.tts import OpenAITTSService
+from pipecat.transports.base_transport import TransportParams
+
+
+async def main():
+    logger = braintrust.init_logger(project="example-pipecat")
+    setup_pipecat(capture_audio_attachments=True)
+
+    api_key = os.environ["OPENAI_API_KEY"]
+    transport = TransportParams(audio_in_enabled=True, audio_out_enabled=True)
+    source, output = RecordedInput(transport), SilentOutput(transport)
+    aggregators = LLMContextAggregatorPair(
+        order_context(),
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+    pipeline = Pipeline(
+        [
+            source,
+            OpenAISTTService(api_key=api_key),
+            aggregators.user(),
+            OpenAILLMService(
+                api_key=api_key,
+                settings=OpenAILLMService.Settings(
+                    model="gpt-4.1-mini",
+                    system_instruction=INSTRUCTIONS,
+                ),
+            ),
+            OpenAITTSService(
+                api_key=api_key,
+                settings=OpenAITTSService.Settings(
+                    model="gpt-4o-mini-tts",
+                    voice="alloy",
+                ),
+            ),
+            output,
+            aggregators.assistant(),
+        ]
+    )
+    with logger.start_span(name="cascade") as trace:
+        await run_conversation(pipeline, aggregators, source)
+    await asyncio.to_thread(logger.flush)
+    print(f"Trace: {trace.link()}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/pipecat/common.py

@@ -0,0 +1,115 @@
+"""Example application plumbing: play a WAV instead of opening a microphone."""
+
+import asyncio
+import wave
+from pathlib import Path
+
+# Pipecat requires Python 3.11+ and is installed by this example, not the shared lint environment.
+# pylint: disable=import-error
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.utils import create_stream_resampler
+from pipecat.frames.frames import InputAudioRawFrame, LLMRunFrame
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.transports.base_input import BaseInputTransport
+from pipecat.transports.base_output import BaseOutputTransport
+from pipecat.workers.runner import WorkerRunner
+
+
+async def lookup_order(params):
+    """A fictional local order database; replace with your application's tool."""
+    await params.result_callback({"order_id": params.arguments["order_id"], "delivery": "Friday"})
+
+
+def order_context():
+    return LLMContext(
+        tools=ToolsSchema(
+            standard_tools=[
+                FunctionSchema(
+                    name="lookup_order",
+                    description="Look up an order's delivery date",
+                    properties={"order_id": {"type": "string"}},
+                    required=["order_id"],
+                    handler=lookup_order,
+                )
+            ]
+        ),
+    )
+
+
+INSTRUCTIONS = (
+    "Speak English. Greet the caller briefly and ask for their order number. "
+    "When they supply it, always call lookup_order. "
+    "After its result, say only: Your order arrives Friday."
+)
+
+
+class RecordedInput(BaseInputTransport):
+    async def start(self, frame):
+        await super().start(frame)
+        await self.set_transport_ready(frame)
+
+    async def play(self, sample_rate):
+        with wave.open(str(Path(__file__).with_name("order.wav"))) as audio:
+            pcm = audio.readframes(audio.getnframes())
+        pcm = await create_stream_resampler().resample(pcm, 16000, sample_rate)
+        # Include silence around the request so native VAD determines its boundaries.
+        pcm = b"\0" * sample_rate + pcm + b"\0" * (sample_rate * 2)
+        frame_bytes = sample_rate // 50 * 2
+        for offset in range(0, len(pcm), frame_bytes):
+            await self.push_audio_frame(InputAudioRawFrame(pcm[offset : offset + frame_bytes], sample_rate, 1))
+            await asyncio.sleep(0.02)
+
+
+class SilentOutput(BaseOutputTransport):
+    """Accept agent audio without requiring an audio device; listen in the trace."""
+
+    async def start(self, frame):
+        await super().start(frame)
+        await self.set_transport_ready(frame)
+
+    async def write_audio_frame(self, frame):
+        return True
+
+
+async def run_conversation(pipeline, aggregators, source, *, realtime=False):
+    sample_rate = 24000 if realtime else 16000
+    worker = PipelineWorker(
+        pipeline,
+        params=PipelineParams(
+            audio_in_sample_rate=sample_rate,
+            audio_out_sample_rate=24000,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+    greeted, finished = asyncio.Event(), asyncio.Event()
+
+    @aggregators.assistant().event_handler("on_assistant_turn_stopped")
+    async def assistant_turn(aggregator, message):
+        print(f"Agent: {message.content}", flush=True)
+        if "friday" in message.content.lower():
+            finished.set()
+        else:
+            greeted.set()
+
+    @aggregators.user().event_handler("on_user_turn_message_added")
+    async def user_turn(aggregator, message):
+        print(f"Caller: {message.content}", flush=True)
+
+    @worker.event_handler("on_pipeline_started")
+    async def started(worker, frame):
+        await worker.queue_frame(LLMRunFrame())
+        await asyncio.wait_for(greeted.wait(), 30)
+        await source.play(sample_rate)
+
+    task = asyncio.create_task(WorkerRunner(handle_sigint=False).run(worker))
+    try:
+        await asyncio.wait_for(finished.wait(), 60)
+        await worker.stop_when_done()
+        await task
+    finally:
+        if not task.done():
+            await worker.cancel()
+            await task

examples/pipecat/pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "braintrust-pipecat-example"
+version = "0.1.0"
+description = "Trace cascade and realtime Pipecat voice agents"
+requires-python = ">=3.11,<3.14"
+dependencies = [
+    "braintrust[audio]",
+    "pipecat-ai[openai,silero]==1.12.0",
+]
+
+[tool.uv.sources]
+braintrust = { path = "../../py", editable = true }

examples/pipecat/realtime.py

@@ -0,0 +1,50 @@
+"""OpenAI Realtime speech-to-speech with a local tool, traced by Braintrust."""
+
+import asyncio
+import os
+
+import braintrust
+from braintrust.integrations.pipecat import setup_pipecat
+from common import INSTRUCTIONS, RecordedInput, SilentOutput, order_context, run_conversation
+
+# Pipecat requires Python 3.11+ and is installed by this example, not the shared lint environment.
+# pylint: disable=import-error
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.services.openai.realtime import events
+from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService
+from pipecat.transports.base_transport import TransportParams
+
+
+async def main():
+    logger = braintrust.init_logger(project="example-pipecat")
+    setup_pipecat(capture_audio_attachments=True)
+
+    transport = TransportParams(audio_in_enabled=True, audio_out_enabled=True)
+    source, output = RecordedInput(transport), SilentOutput(transport)
+    aggregators = LLMContextAggregatorPair(order_context())
+    model = OpenAIRealtimeLLMService(
+        api_key=os.environ["OPENAI_API_KEY"],
+        settings=OpenAIRealtimeLLMService.Settings(
+            model="gpt-realtime",
+            system_instruction=INSTRUCTIONS,
+            session_properties=events.SessionProperties(
+                audio=events.AudioConfiguration(
+                    input=events.AudioInput(
+                        transcription=events.InputAudioTranscription(model="gpt-4o-mini-transcribe", language="en"),
+                        turn_detection=events.TurnDetection(),
+                    ),
+                    output=events.AudioOutput(voice="alloy"),
+                )
+            ),
+        ),
+    )
+    pipeline = Pipeline([source, aggregators.user(), model, output, aggregators.assistant()])
+    with logger.start_span(name="realtime") as trace:
+        await run_conversation(pipeline, aggregators, source, realtime=True)
+    await asyncio.to_thread(logger.flush)
+    print(f"Trace: {trace.link()}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

py/noxfile.py

@@ -467,7 +467,7 @@ def test_pipecat(session, version):
     # loaded from Nox's virtualenv when it is beneath the current directory.
     _run_tests(
         session,
-        f"{INTEGRATION_DIR}/pipecat/test_pipecat.py",
+        f"{INTEGRATION_DIR}/pipecat" if version == LATEST else f"{INTEGRATION_DIR}/pipecat/test_pipecat.py",
         version=version,
         run_from_temp_dir=True,
     )
@@ -786,6 +786,13 @@ def test_pytest_plugin(session, version):
     _run_tests(session, f"{WRAPPER_DIR}/pytest_plugin/test_plugin.py")
 
 
+@nox.session()
+def test_audio(session):
+    _install_test_deps(session)
+    _install_group_locked(session, "test-audio")
+    _run_tests(session, "braintrust/_audio")
+
+
 @nox.session()
 def test_core(session):
     _install_test_deps(session)
@@ -956,6 +963,8 @@ def _run_core_tests(session):
         SRC_DIR,
         ignore_paths=[
             WRAPPER_DIR,
+            "braintrust/_audio/test_recording.py",
+            "braintrust/_audio/test_segments.py",
             *_integration_subdirs_to_ignore(),
             CONTRIB_DIR,
             DEVSERVER_DIR,

py/pyproject.toml

@@ -45,6 +45,7 @@ braintrust = "braintrust.wrappers.pytest_plugin.plugin"
 braintrust = "braintrust.integrations.harbor:HarborPlugin"
 
 [project.optional-dependencies]
+audio = ["numpy>=1.26", "soundfile>=0.13.1"]
 cli = ["boto3", "python-dotenv", "uv", "starlette", "uvicorn"]
 # TODO: remove the doc extra in the next major version.
 doc = []
@@ -195,8 +196,15 @@ test-livekit-agents = [
     "opentelemetry-sdk<1.39",
 ]
 
+test-audio = [
+    {include-group = "test"},
+    "numpy>=1.26",
+    "soundfile>=0.13.1",
+]
+
 test-pipecat = [
     {include-group = "test"},
+    "soundfile>=0.13.1",
     # pipecat-ai 1.3.0 imports websockets but does not install it transitively.
     "websockets==15.0.1",
 ]

py/src/braintrust/_audio/__init__.py

@@ -0,0 +1,33 @@
+"""Internal audio API shared by integrations; codecs remain optional and lazy."""
+
+from .alignment import InputRanges
+from .attachments import UploadFailed, prepare_recording, upload_recording
+from .budget import source_budget
+from .export import AlignmentPublisher, segment_descriptor
+from .jobs import RecordingJobs
+from .options import RecordingOptions
+from .recording import encode_audio
+from .segments import SegmentedRecording
+from .timeline import ClipTimeline, ms_to_samples, pcm_bytes_to_ms, samples_to_ms
+from .worker import RecordingBusy, encode_in_worker
+
+
+__all__ = [
+    "AlignmentPublisher",
+    "ClipTimeline",
+    "InputRanges",
+    "RecordingBusy",
+    "RecordingJobs",
+    "RecordingOptions",
+    "SegmentedRecording",
+    "UploadFailed",
+    "encode_audio",
+    "encode_in_worker",
+    "ms_to_samples",
+    "pcm_bytes_to_ms",
+    "prepare_recording",
+    "samples_to_ms",
+    "segment_descriptor",
+    "source_budget",
+    "upload_recording",
+]