---
title: "Pipecat Tracing"
description: "Automatically trace Pipecat voice and multimodal agent pipelines."
sidebar:
  label: "Pipecat"
seo:
  title: "Pipecat Tracing with Judgment | Agent Framework Integration Docs"
  description: "Trace Pipecat agent executions with Judgment, including tool calls, handoffs, spans, and behavior for production agent monitoring."
---

**Pipecat integration** captures traces from your Pipecat voice and multimodal agents, including conversation turns and LLM, STT, and TTS service calls.

Pipecat has built-in OpenTelemetry tracing. Judgment installs itself as the global OpenTelemetry tracer provider, so enabling tracing on your pipeline routes all Pipecat spans to Judgment — no OTLP exporter setup required.

{/* @test-flow
  id: integration-pipecat
  lang: python
  env: JUDGMENT_API_KEY, JUDGMENT_ORG_ID
  skip: Requires a live voice transport and STT/LLM/TTS service credentials that CI cannot provide
*/}

## Quickstart

1. ### Install Dependencies

    **uv**

    ```bash
    uv add judgeval pipecat-ai
    ```

    **pip**

    ```bash
    pip install judgeval pipecat-ai
    ```

2. ### Install the Judgment Tracer Provider

    Initialize the tracer and install it as the global OpenTelemetry tracer provider **before** building your pipeline.

    ```python
    from judgeval import JudgmentTracerProvider, Tracer

    Tracer.init(project_name="pipecat_project")
    JudgmentTracerProvider.install_as_global_tracer_provider()
    ```

    > **Warning**
    >
    > Do not call Pipecat's `setup_tracing()`. It installs its own global tracer
    > provider, and OpenTelemetry enforces first-writer-wins, so the Judgment
    > provider would either be ignored or overwritten.

3. ### Enable Tracing on Your Pipeline

    Set `enable_tracing=True` on your `PipelineTask` and `enable_metrics=True` in `PipelineParams` so service-level attributes (model, token usage, TTFB) are captured.

    ```python
    from pipecat.pipeline.task import PipelineParams, PipelineTask

    task = PipelineTask(
        pipeline,
        params=PipelineParams(enable_metrics=True), # [!code highlight]
        enable_tracing=True, # [!code highlight]
        conversation_id="customer-123",
    )
    ```

    Each conversation appears as a single trace in Judgment, with turn spans and service spans nested underneath.

## Complete Example

A typical Pipecat voice bot wires a transport together with STT, LLM, and TTS services. Install Judgment as the global tracer provider once at startup, then enable tracing on the task — every span the pipeline emits is routed to Judgment automatically.

```python title="bot.py"

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport

from judgeval import JudgmentTracerProvider, Tracer  # [!code ++]

Tracer.init(project_name="pipecat_project")  # [!code highlight]
JudgmentTracerProvider.install_as_global_tracer_provider()  # [!code highlight]


async def main():
    transport = DailyTransport(
        os.getenv("DAILY_ROOM_URL"),
        None,
        "Voice Bot",
        DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
    tts = CartesiaTTSService(
        api_key=os.getenv("CARTESIA_API_KEY"),
        voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",
    )

    messages = [{"role": "system", "content": "You are a helpful voice assistant."}]
    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    pipeline = Pipeline(
        [
            transport.input(),
            stt,
            context_aggregator.user(),
            llm,
            tts,
            transport.output(),
            context_aggregator.assistant(),
        ]
    )

    task = PipelineTask(
        pipeline,
        params=PipelineParams(enable_metrics=True),  # [!code highlight]
        enable_tracing=True,  # [!code highlight]
        conversation_id="customer-123",
    )

    runner = PipelineRunner()
    await runner.run(task)


if __name__ == "__main__":
    asyncio.run(main())
```

> **Info**
>
> Service spans (model name, token usage, character count, TTFB) only appear when
> `enable_metrics=True` is set in `PipelineParams`. Turn spans are emitted as long
> as `enable_tracing=True` — turn tracking is on by default.

## What's Captured

Pipecat structures its spans hierarchically, so a single trace represents one complete conversation:

```
conversation
└── turn
    ├── stt   (speech-to-text)
    ├── llm   (model call)
    └── tts   (text-to-speech)
```

| Span | Attributes captured |
| --- | --- |
| **Conversation** | `conversation.id`, `conversation.type` |
| **Turn** | `turn.number`, `turn.duration_seconds`, `turn.was_interrupted` |
| **LLM** | `gen_ai.system`, `gen_ai.request.model`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `metrics.ttfb`, tool config |
| **STT** | `gen_ai.system`, `gen_ai.request.model`, `transcript`, `is_final`, `language`, `metrics.ttfb` |
| **TTS** | `gen_ai.system`, `gen_ai.request.model`, `voice_id`, `text`, `metrics.character_count`, `metrics.ttfb` |

## Associating Sessions and Metadata

Wrap the function that runs your pipeline in `@Tracer.observe` and use Judgment's
tracer to attach a `session_id`, a `customer_id`, and any custom attributes. These
are set on the active trace and propagate to every Pipecat span created during the
run, so the whole conversation is grouped and attributed in Judgment.

```python
@Tracer.observe(span_type="function") # [!code highlight]
async def run_bot(session_id: str, customer_id: str):
    Tracer.set_session_id(session_id)    # group related turns into one session # [!code highlight]
    Tracer.set_customer_id(customer_id)  # attribute the conversation to a user # [!code highlight]
    Tracer.set_attributes({"channel": "voice", "room": "support-1"}) # [!code highlight]

    runner = PipelineRunner()
    await runner.run(task)
```

> **Tip**
>
> Use `Tracer.set_session_id()` to group multiple traces into a session-level view,
> `Tracer.set_customer_id()` to tie traces to a specific user, and
> `Tracer.set_attributes()` for any custom metadata. See the
> [attributes and context](/documentation/tracing/attributes) guide for details.

## Next Steps

- [Monitor a behavior](/documentation/monitoring) - Monitor your Pipecat agents in production with behavioral scoring.
- [Instrument your agent](/documentation/tracing/instrumentation) - Configure and verify Judgment tracing beyond this integration.
- [Pipecat OpenTelemetry Docs](https://docs.pipecat.ai/server/utilities/opentelemetry) - Full reference for Pipecat's built-in tracing.