feat: Add pipecat integration

py/noxfile.py

@@ -385,6 +385,29 @@ def test_livekit_agents(session, version):
     _run_tests(session, f"{INTEGRATION_DIR}/livekit_agents/test_livekit_agents.py", version=version, env=env)
 
 
+PIPECAT_VERSIONS = _get_matrix_versions("pipecat-ai")
+
+
+@nox.session()
+@nox.parametrize("version", PIPECAT_VERSIONS, ids=PIPECAT_VERSIONS)
+def test_pipecat(session, version):
+    if sys.version_info < (3, 11):
+        session.skip("Pipecat AI 1.x requires Python 3.11+")
+    if sys.version_info >= (3, 14):
+        session.skip("Pipecat AI's onnxruntime dependency does not ship Python 3.14 wheels")
+    _install_test_deps(session)
+    _install_group_locked(session, "test-pipecat")
+    _install_matrix_dep(session, "pipecat-ai", version)
+    # Pipecat imports NLTK, whose safe-import finder rejects dependencies
+    # loaded from Nox's virtualenv when it is beneath the current directory.
+    _run_tests(
+        session,
+        f"{INTEGRATION_DIR}/pipecat/test_pipecat.py",
+        version=version,
+        run_from_temp_dir=True,
+    )
+
+
 STRANDS_VERSIONS = _get_matrix_versions("strands-agents")
 
 
@@ -794,7 +817,15 @@ def _run_core_tests(session):
     )
 
 
-def _run_tests(session, test_path, ignore_path="", ignore_paths=None, env=None, version=None):
+def _run_tests(
+    session,
+    test_path,
+    ignore_path="",
+    ignore_paths=None,
+    env=None,
+    version=None,
+    run_from_temp_dir=False,
+):
     """Run tests against a wheel or the source code. Paths should be relative and start with braintrust."""
     env = env.copy() if env else {}
     if version:
@@ -811,19 +842,27 @@ def _run_tests(session, test_path, ignore_path="", ignore_paths=None, env=None,
         paths_to_ignore.extend(ignore_paths)
 
     if not wheel_flag:
-        # Run the tests in the src directory
+        # Run the tests in the src directory.
+        source_test_path = f"src/{test_path}"
+        source_ignore_paths = [f"src/{path}" for path in paths_to_ignore]
+        if run_from_temp_dir:
+            source_test_path = os.path.abspath(source_test_path)
+            source_ignore_paths = [os.path.abspath(path) for path in source_ignore_paths]
         test_args = [
             "pytest",
             # Disable the braintrust pytest plugin (registered via pytest11 entry
             # point) to avoid ImportPathMismatchError when the installed package
             # and the source tree both contain braintrust/conftest.py.
             "-p",
             "no:braintrust",
-            f"src/{test_path}",
+            source_test_path,
         ]
-        for path in paths_to_ignore:
-            test_args.append(f"--ignore=src/{path}")
-        session.run(*test_args, *common_args, *pytest_posargs, env=env)
+        test_args.extend(f"--ignore={path}" for path in source_ignore_paths)
+        if run_from_temp_dir:
+            with tempfile.TemporaryDirectory() as tmp, session.chdir(tmp):
+                session.run(*test_args, *common_args, *pytest_posargs, env=env)
+        else:
+            session.run(*test_args, *common_args, *pytest_posargs, env=env)
         return
 
     # Running the tests from the wheel involves a bit of gymnastics to ensure we don't import

py/pyproject.toml

@@ -184,6 +184,12 @@ test-livekit-agents = [
     "opentelemetry-sdk<1.39",
 ]
 
+test-pipecat = [
+    {include-group = "test"},
+    # pipecat-ai 1.3.0 imports websockets but does not install it transitively.
+    "websockets==15.0.1",
+]
+
 test-crewai = [
     {include-group = "test"},
     # CrewAI's no-network smoke test forces the LiteLLM fallback path via
@@ -363,6 +369,10 @@ latest = "litellm==1.95.0"
 latest = "livekit-agents==1.6.8"
 "1.3.1" = "livekit-agents==1.3.1"
 
+[tool.braintrust.matrix.pipecat-ai]
+latest = "pipecat-ai==1.4.0"
+"1.3.0" = "pipecat-ai==1.3.0"
+
 [tool.braintrust.matrix.claude-agent-sdk]
 latest = "claude-agent-sdk==0.2.129"
 "0.1.10" = "claude-agent-sdk==0.1.10"
@@ -514,6 +524,7 @@ mistral = ["mistralai"]
 openai = ["openai"]
 openai_agents = ["openai-agents"]
 openrouter = ["openrouter"]
+pipecat = ["pipecat-ai"]
 pydantic_ai = ["pydantic-ai-integration", "pydantic-ai-wrap-openai"]
 strands = ["strands-agents"]
 
@@ -539,5 +550,6 @@ huggingface-hub = "huggingface_hub"
 openai = "openai"
 openai-agents = "agents"
 openrouter = "openrouter"
+pipecat-ai = "pipecat"
 strands-agents = "strands"
 temporalio = "temporalio"

py/src/braintrust/auto.py

@@ -29,6 +29,7 @@
     OpenAIAgentsIntegration,
     OpenAIIntegration,
     OpenRouterIntegration,
+    PipecatIntegration,
     PydanticAIIntegration,
     StrandsIntegration,
     TemporalIntegration,
@@ -78,6 +79,7 @@ def auto_instrument(
     strands: bool = True,
     temporal: bool = True,
     livekit_agents: bool = True,
+    pipecat: bool = True,
 ) -> dict[str, bool]:
     """
     Auto-instrument supported AI/ML libraries for Braintrust tracing.
@@ -113,6 +115,7 @@ def auto_instrument(
         strands: Enable Strands Agents instrumentation (default: True)
         temporal: Enable Temporal instrumentation (default: True)
         livekit_agents: Enable LiveKit Agents instrumentation (default: True)
+        pipecat: Enable Pipecat AI instrumentation (default: True)
 
     Returns:
         Dict mapping integration name to whether it was successfully instrumented.
@@ -208,6 +211,8 @@ def auto_instrument(
         results["temporal"] = _instrument_integration(TemporalIntegration)
     if livekit_agents:
         results["livekit_agents"] = _instrument_integration(LiveKitAgentsIntegration)
+    if pipecat:
+        results["pipecat"] = _instrument_integration(PipecatIntegration)
 
     return results
 

py/src/braintrust/env.py

@@ -209,6 +209,8 @@ class BraintrustEnv:
     FAILED_PUBLISH_PAYLOADS_DIR = EnvVar("BRAINTRUST_FAILED_PUBLISH_PAYLOADS_DIR", EnvParser.STRING)
     ALL_PUBLISH_PAYLOADS_DIR = EnvVar("BRAINTRUST_ALL_PUBLISH_PAYLOADS_DIR", EnvParser.STRING)
     DISABLE_ATEXIT_FLUSH = EnvVar("BRAINTRUST_DISABLE_ATEXIT_FLUSH", EnvParser.BOOL)
+    CAPTURE_USER_AUDIO_ATTACHMENTS = EnvVar("BRAINTRUST_CAPTURE_USER_AUDIO_ATTACHMENTS", EnvParser.BOOL)
+    CAPTURE_AGENT_AUDIO_ATTACHMENTS = EnvVar("BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS", EnvParser.BOOL)
     OTEL_COMPAT = EnvVar("BRAINTRUST_OTEL_COMPAT", EnvParser.BOOL)
     # Opt out of the default OpenTelemetry-compatible hex span/trace IDs and use
     # legacy UUID-based IDs (and V3 span-component export) instead.

py/src/braintrust/integrations/__init__.py

@@ -19,6 +19,7 @@
 from .openai import OpenAIIntegration
 from .openai_agents import OpenAIAgentsIntegration
 from .openrouter import OpenRouterIntegration
+from .pipecat import PipecatIntegration
 from .pydantic_ai import PydanticAIIntegration
 from .strands import StrandsIntegration
 from .temporal import TemporalIntegration
@@ -46,6 +47,7 @@
     "OpenAIIntegration",
     "OpenAIAgentsIntegration",
     "OpenRouterIntegration",
+    "PipecatIntegration",
     "PydanticAIIntegration",
     "StrandsIntegration",
     "TemporalIntegration",

py/src/braintrust/integrations/auto_test_scripts/test_auto_pipecat.py

@@ -0,0 +1,102 @@
+import asyncio
+import importlib
+import inspect
+import os
+import tempfile
+from pathlib import Path
+
+from braintrust.auto import auto_instrument
+from braintrust.integrations.test_utils import autoinstrument_test_context
+
+
+def _ensure_nltk_punkt_tab():
+    data_dir = Path(tempfile.gettempdir()) / "braintrust-pipecat-nltk-data"
+    punkt_tab = data_dir / "tokenizers" / "punkt_tab"
+    punkt_tab.mkdir(parents=True, exist_ok=True)
+    os.environ.setdefault("NLTK_DATA", str(data_dir))
+
+
+def _import(path):
+    _ensure_nltk_punkt_tab()
+    module_name, attr = path.rsplit(".", 1)
+    return getattr(importlib.import_module(module_name), attr)
+
+
+def _worker_kwargs(**overrides):
+    PipelineWorker = _import("pipecat.pipeline.worker.PipelineWorker")
+    signature = inspect.signature(PipelineWorker)
+    kwargs = {"idle_timeout_secs": None}
+    for name, value in {
+        "enable_turn_tracking": False,
+        "enable_rtvi": False,
+        "check_dangling_tasks": False,
+    }.items():
+        if name in signature.parameters:
+            kwargs[name] = value
+    kwargs.update(overrides)
+    return kwargs
+
+
+def _runner_kwargs(**overrides):
+    WorkerRunner = _import("pipecat.workers.runner.WorkerRunner")
+    signature = inspect.signature(WorkerRunner)
+    kwargs = {"handle_sigint": False}
+    if "check_dangling_tasks" in signature.parameters:
+        kwargs["check_dangling_tasks"] = False
+    kwargs.update(overrides)
+    return kwargs
+
+
+async def main():
+    with autoinstrument_test_context("test_auto_pipecat", integration="pipecat") as memory_logger:
+        _ensure_nltk_punkt_tab()
+        results = auto_instrument()
+        assert results.get("pipecat") is True
+
+        EndFrame = _import("pipecat.frames.frames.EndFrame")
+        LLMContextFrame = _import("pipecat.frames.frames.LLMContextFrame")
+        Pipeline = _import("pipecat.pipeline.pipeline.Pipeline")
+        PipelineParams = _import("pipecat.pipeline.worker.PipelineParams")
+        PipelineWorker = _import("pipecat.pipeline.worker.PipelineWorker")
+        LLMContext = _import("pipecat.processors.aggregators.llm_context.LLMContext")
+        OpenAILLMService = _import("pipecat.services.openai.llm.OpenAILLMService")
+        WorkerRunner = _import("pipecat.workers.runner.WorkerRunner")
+
+        llm = OpenAILLMService(
+            api_key=os.environ["OPENAI_API_KEY"],
+            settings=OpenAILLMService.Settings(
+                model="gpt-4o-mini",
+                temperature=0.0,
+                max_completion_tokens=20,
+            ),
+        )
+        worker = PipelineWorker(
+            Pipeline([llm]),
+            **_worker_kwargs(
+                name="bt-auto-pipecat-worker",
+                params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
+            ),
+        )
+        context = LLMContext(
+            messages=[
+                {"role": "developer", "content": "Answer with exactly the requested text and no punctuation."},
+                {"role": "user", "content": "Say: braintrust auto pipecat"},
+            ]
+        )
+
+        @worker.event_handler("on_pipeline_started")
+        async def on_pipeline_started(_worker, _frame):
+            await worker.queue_frames([LLMContextFrame(context), EndFrame()])
+
+        runner = WorkerRunner(**_runner_kwargs())
+        await runner.add_workers(worker)
+        await asyncio.wait_for(runner.run(), timeout=20)
+
+        logs = memory_logger.pop()
+        names = {log.get("span_attributes", {}).get("name") for log in logs}
+        assert "pipecat_pipeline" in names
+        assert "pipecat_llm_response" in names
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

py/src/braintrust/integrations/livekit_agents/__init__.py

@@ -3,6 +3,12 @@
 
 
 def setup_livekit_agents() -> bool:
+    """Set up LiveKit Agents tracing.
+
+    Set ``BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS=false`` to omit agent
+    playback audio attachments while preserving the ``agent_speaking`` spans
+    and transcripts.
+    """
     return LiveKitAgentsIntegration.setup()
 
 

py/src/braintrust/integrations/livekit_agents/test_livekit_agents.py

@@ -205,6 +205,37 @@ def test_wrap_livekit_agents_wraps_real_agent_session():
     assert hasattr(inspect.getattr_static(AgentSession, "say"), "__wrapped__")
 
 
+@pytest.mark.asyncio
+async def test_livekit_agents_agent_audio_capture_respects_generic_env(monkeypatch):
+    class AudioOutput:
+        pass
+
+    class AudioFrame:
+        data = b"\x00\x00\x01\x00"
+        sample_rate = 16000
+        num_channels = 1
+        samples_per_channel = 2
+
+    async def capture_frame(frame):
+        return frame
+
+    output = AudioOutput()
+    setattr(output, tracing._PLAYBACK_HANDLER_ATTACHED_ATTR, True)
+    frame = AudioFrame()
+
+    monkeypatch.delenv("BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS", raising=False)
+    assert await tracing.traced_audio_output_capture_frame(capture_frame, output, (frame,), {}) is frame
+    assert bytes(getattr(output, tracing._PLAYBACK_AUDIO_ATTR)) == frame.data
+
+    monkeypatch.setenv("BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS", "false")
+    assert await tracing.traced_audio_output_capture_frame(capture_frame, output, (frame,), {}) is frame
+    assert getattr(output, tracing._PLAYBACK_AUDIO_ATTR, None) is None
+
+    monkeypatch.setenv("BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS", "true")
+    assert await tracing.traced_audio_output_capture_frame(capture_frame, output, (frame,), {}) is frame
+    assert bytes(getattr(output, tracing._PLAYBACK_AUDIO_ATTR)) == frame.data
+
+
 @pytest.mark.asyncio
 @pytest.mark.vcr
 async def test_livekit_agents_agent_speaking_e2e(memory_logger, livekit_server):

py/src/braintrust/integrations/livekit_agents/tracing.py

@@ -1,12 +1,12 @@
 import asyncio
 import contextlib
-import io
 import json
 import time
-import wave
 from contextvars import ContextVar
 from typing import Any
 
+from braintrust.env import BraintrustEnv
+from braintrust.integrations.utils import _pcm_to_wav
 from braintrust.logger import (
     NOOP_SPAN,
     Attachment,
@@ -440,7 +440,10 @@ async def traced_audio_output_capture_frame(
     if getattr(instance, _PLAYBACK_HANDLER_ATTACHED_ATTR, False):
         if getattr(instance, _PLAYBACK_START_ATTR, None) is None:
             setattr(instance, _PLAYBACK_START_ATTR, time.time())
-        _capture_playback_audio(instance, args[0] if args else kwargs.get("frame"))
+        if BraintrustEnv.CAPTURE_AGENT_AUDIO_ATTACHMENTS.get(True):
+            _capture_playback_audio(instance, args[0] if args else kwargs.get("frame"))
+        else:
+            _clear_playback_audio(instance)
     return await wrapped(*args, **kwargs)
 
 
@@ -677,7 +680,7 @@ def _pop_playback_audio(obj: Any) -> Attachment | None:
     audio = getattr(obj, _PLAYBACK_AUDIO_ATTR, None)
     metadata = getattr(obj, _PLAYBACK_AUDIO_METADATA_ATTR, None) or {}
     _clear_playback_audio(obj)
-    if not audio:
+    if not BraintrustEnv.CAPTURE_AGENT_AUDIO_ATTACHMENTS.get(True) or not audio:
         return None
     sample_rate = metadata.get("sample_rate")
     num_channels = metadata.get("num_channels")
@@ -695,17 +698,6 @@ def _pop_playback_audio(obj: Any) -> Attachment | None:
     )
 
 
-def _pcm_to_wav(audio: bytes, *, sample_rate: int, num_channels: int) -> bytes:
-    buffer = io.BytesIO()
-    with wave.open(buffer, "wb") as wav_file:
-        writer: Any = wav_file
-        writer.setnchannels(num_channels)  # pylint: disable=no-member
-        writer.setsampwidth(2)  # pylint: disable=no-member
-        writer.setframerate(sample_rate)  # pylint: disable=no-member
-        writer.writeframes(audio)  # pylint: disable=no-member
-    return buffer.getvalue()
-
-
 _AGENT_TURN_METADATA_FIELDS = (
     "generation_id",
     "speech_id",