Pipecat

Overview

Pipecat is an open-source Python framework for building realtime voice and multimodal agents.

The integration works through OpenTelemetry. Pipecat emits native spans for each conversation, its turns, and the speech-to-text, LLM, and text-to-speech calls inside them; confident-trace, Confident AI's OpenTelemetry-native tracing SDK, exports those spans to Confident AI with their parent-child relationships intact.

Call init() at the top of your bot file. You don't need Pipecat's setup_tracing(), an OTLP exporter, or enable_tracing=True on each pipeline.

Auto-Instrument

1. Install Dependencies

   Add confident-trace to your existing Pipecat project. Keep the service extras your bot already uses.

   pip install confident-trace 'pipecat-ai==1.13.0'

   If you're starting a new bot, follow the Pipecat quickstart first, then add the tracing setup below.

2. Set Your API Key

   Get your Confident AI Project API key and make it available to your bot:

   export CONFIDENT_API_KEY="<your-confident-project-key>"

   Keep your existing transport credentials and model provider keys configured as usual.

3. Instrument Your Bot

   Add init() at module scope in your bot file. Keep your existing transport, services, and pipeline.

   from confident_trace import init
   from pipecat.pipeline.pipeline import Pipeline
   from pipecat.pipeline.worker import PipelineParams, PipelineWorker
   from pipecat.workers.runner import WorkerRunner
   
   init()
   
   
   async def bot(runner_args):
       # Keep your existing transport, services, and context aggregators here.
       pipeline = Pipeline([
           transport.input(), stt, user_aggregator, llm, tts,
           transport.output(), assistant_aggregator,
       ])
       worker = PipelineWorker(
           pipeline,
           params=PipelineParams(enable_metrics=True),
           conversation_id=runner_args.session_id,
       )
       runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
       await runner.add_workers(worker)
       await runner.run()
   
   
   if __name__ == "__main__":
       from pipecat.runner.run import main
   
       main()

   Pipecat's tracing is off unless a pipeline asks for it. Once init() has run, every PipelineWorker (and the deprecated PipelineTask) traces by default. To keep a pipeline out of Confident AI, pass enable_tracing=False to it.

4. Run Your Bot

   python bot.py

   Connect to your bot as usual, then open the Observatory in your Confident AI project to inspect its traces.

   Confident flushes pending spans when the pipeline finishes, after Pipecat ends the conversation span, so bots that exit as soon as a call ends (such as on Pipecat Cloud) still deliver the whole trace. You don't need to add a separate flush. The flush waits at most five seconds, and a failed export never blocks the pipeline from shutting down.

What Gets Captured

Each call becomes one trace, labelled Pipecat:

conversation (trace)
├── turn
│   ├── stt
│   ├── llm
│   └── tts
└── turn
    └── ...

• Conversation: the whole call, with its conversation.id and any attributes you pass in additional_span_attributes.
• Turns: one span per user and bot exchange, with its duration, whether the user interrupted, and the user-to-bot latency.
• Speech-to-text: the transcript, language, and time to first byte.
• LLM calls: model, messages, output, tools, and token usage.
• Text-to-speech: the text being spoken, character count, and time to first byte.
• Speech-to-speech models: with OpenAI Realtime or Gemini Live, the session setup and its instructions, each model response with its output and token usage, and tool calls and results as tool spans.

Pipecat's llm span already records each model call, so the integration suppresses the overlapping OpenAI or Anthropic provider span and each call's cost is counted once. Supported provider calls made outside Pipecat's services are still traced separately.

Call Transcripts

Confident AI reads the conversation from Pipecat's speech spans and maps it into the trace, so you can read a call without opening each span. With OpenAI Realtime or Gemini Live, the bot's side comes from the transcript of each spoken model response:

• Trace input and output: the trace output is the bot's last reply, and the trace input is the conversation before it.
• Turn spans: each turn span shows what the user said and how the bot answered. A reply spoken across several text-to-speech spans is shown as one message.
• Voice timing: when a call belongs to a thread, each turn shows how long the bot took to answer and whether the user interrupted it. A turn cut off because the call ended doesn't count as an interruption.

Calls often send spans in several pieces, so the transcript fills in as later spans arrive. Thread evaluations are held back while a call is still running, so online evals score the full conversation instead of a partial one.

Record Call Audio

Pipecat has no built-in recording. Add Pipecat's AudioBufferProcessor to your pipeline after the transport output so it hears both sides, and Confident uploads the call recording to the call's trace when the pipeline ends. You can then listen to the whole call, and each turn, in the Observatory.

from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor


audio = AudioBufferProcessor(num_channels=2, auto_start_recording=True)
pipeline = Pipeline([
    transport.input(), stt, user_aggregator, llm, tts,
    transport.output(),
    audio,
    assistant_aggregator,
])

That is the only change. Confident never adds a recorder to your pipeline for you. If you already save audio with @audio.event_handler("on_audio_data"), your handler keeps working; Confident's runs alongside it.

num_channels=2 records the user on the left channel and the bot on the right, which makes interruptions and overlapping speech easy to hear. auto_start_recording=True starts recording when the pipeline starts; without it, call await audio.start_recording() yourself, for example after the caller consents.

Confident encodes the audio to Ogg/Opus as Pipecat delivers it. By default Pipecat holds the whole call in memory until the pipeline ends; for long calls, set buffer_size so audio is handed over in pieces, for example every second:

audio = AudioBufferProcessor(
    num_channels=2,
    buffer_size=24000 * 2 * 2,
    auto_start_recording=True,
)

buffer_size is in bytes: sample rate × 2 bytes per sample × channels. Pipecat uses your pipeline's output sample rate, 24 kHz by default.

Uploads follow a few limits:

• No recording: nothing is uploaded if the pipeline has no AudioBufferProcessor or recording never started.
• Size: recordings up to 18 MB, about 34 minutes of stereo or 69 minutes of mono audio, are uploaded. Larger recordings are skipped, and Confident logs a warning.
• Sample rate: Opus encodes 8, 12, 16, 24 and 48 kHz audio. At other sample rates the recording is skipped, and Confident logs a warning.
• Tracing off: pipelines with enable_tracing=False, or with Pipecat left out of init(instrumentations=...), upload nothing.
• Content capture: capture_content=False in init() turns off recording uploads.
• Export: recordings go to the same Confident AI endpoint and API key as your traces, so EU projects need no extra setup. They're only uploaded when traces are exported over OTLP HTTP, the default, not over gRPC or through your own exporter.

Set Trace Span Properties

Use a trace context to add tags, metadata, a thread, user ID, and customer ID around the pipeline run. It creates no extra span; the properties are applied to the conversation and every span inside it.

Keep init() at module scope as shown above. Replace the sample IDs with values from your application.

from confident_trace import trace_context


async def bot(runner_args):
    # Build your transport, pipeline, and worker as above.
    async with trace_context(
        tags=["support"],
        metadata={"release": "2026-10"},
        thread_id=runner_args.session_id,
        user_id="user-42",
        customer_id="customer-7",
    ):
        await runner.add_workers(worker)
        await runner.run()

See threads, users and customers to set an optional display name alongside each ID, and trace context for every supported trace property and update behavior.

Tracer Provider

Pipecat creates its spans through the global OpenTelemetry tracer provider. If none is set, init() sets Confident's, or the one you pass as tracer_provider=. If your app set its own global provider before init() without passing it in, Pipecat's spans stay on that provider and don't reach Confident AI.

Calling Pipecat's setup_tracing() after init() has no effect, because OpenTelemetry keeps the first global provider. You can remove it.

Content Privacy

Pipecat writes the conversation into its own span attributes: messages and output on LLM spans, transcripts on speech-to-text spans, spoken text on text-to-speech spans, and system instructions and tool arguments and results where the service records them. Confident applies your content settings to those attributes as spans are exported.

To keep the call structure, timing, and token usage but remove the conversation, turn off content capture:

from confident_trace import init

init(capture_content=False)

To keep the conversation but mask parts of it, pass a redactor. It receives each value: messages arrive as a list of message dictionaries, and transcripts and spoken text as strings. Return the value with sensitive parts replaced.

import re

from confident_trace import init

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")


def redact(value):
    if isinstance(value, str):
        return EMAIL.sub("[email]", value)
    if isinstance(value, list):
        return [redact(item) for item in value]
    if isinstance(value, dict):
        return {key: redact(item) for key, item in value.items()}
    return value


init(redact=redact)

If the redactor raises an error, that value is left out rather than sent unredacted. max_content_bytes does not shorten Pipecat's attributes.

To leave a pipeline out of Confident AI entirely, pass enable_tracing=False to it. None of its Pipecat spans are created, and its model calls are traced as standalone OpenAI or Anthropic spans instead.

worker = PipelineWorker(pipeline, enable_tracing=False)

Disable Pipecat Instrumentation

Pass init() a list of integration identifiers to enable only those integrations. The Pipecat identifier is "pipecat". Omit it to disable the adapter; an empty list disables all automatic instrumentation:

from confident_trace import init

init(instrumentations=())

This leaves Pipecat's own default in place, so pipelines no longer trace unless they pass enable_tracing=True, and turns off the flush when a pipeline finishes. Pipecat is optional: applications that don't install it can use init() normally.

Next Steps

Last updated on October 22, 2026