Pipecat

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.

Quickstart

Install Dependencies

uv add judgeval pipecat-ai

pip install judgeval pipecat-ai

Install the Judgment Tracer Provider

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

from judgeval import JudgmentTracerProvider, Tracer

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

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.

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.

from pipecat.pipeline.task import PipelineParams, PipelineTask

task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True), 
    enable_tracing=True, 
    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.

bot.py

import asyncio
import os

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  

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


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),  
        enable_tracing=True,  
        conversation_id="customer-123",
    )

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


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

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.

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

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

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 Tracing reference for details.

Next Steps

Agent Behavior Monitoring - Monitor your Pipecat agents in production with behavioral scoring.
Tracing Reference - Learn more about Judgment's tracing capabilities and advanced configuration.
Pipecat OpenTelemetry Docs - Full reference for Pipecat's built-in tracing.

Pipecat

On this page