Skip to content
Judgment Labs
Esc
↑↓navigate↵open⌘Jpreview
On this page

Pipecat Tracing

Automatically trace Pipecat voice and multimodal agent pipelines.

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()

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.


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())

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)

Next Steps

Was this page helpful?