Pipecat

Pipecat Voice & Multimodal Agent Tracing

Pipecat is an open-source Python framework for building real-time voice and multimodal conversational agents. It orchestrates the audio transport, speech-to-text, LLM, and text-to-speech stages of a conversation into a single streaming pipeline.

Arize provides first-class support for instrumenting Pipecat pipelines using the openinference-instrumentation-pipecat package. Each pipeline run is captured as a session in Arize with LLM, STT, TTS, and tool spans tied to a conversation_id so you can review the full audio-in to audio-out path of any turn.

Quick Start: Pipecat Python Integration

Installation & Setup

Install the OpenInference Pipecat instrumentor along with Pipecat and the Arize OTel helper:

pip install openinference-instrumentation-pipecat pipecat-ai arize-otel

openinference-instrumentation-pipecat >=1.0 requires pipecat-ai >=1.0 and Python >=3.11. Pipecat 1.0 renamed observers, removed LLMMessagesFrame, and dropped Python 3.10 support. If you’re still on pipecat-ai<1.0, pin the instrumentor:

pip install 'openinference-instrumentation-pipecat<=0.1.4' 'pipecat-ai<1.0'

Instrumentation Setup

Configure the PipecatInstrumentor and tracer to send traces to Arize:

# Import open-telemetry dependencies
from arize.otel import register

# Setup OTel via our convenience function
tracer_provider = register(
    space_id = "your-space-id", # in app space settings page
    api_key = "your-api-key", # in app space settings page
    project_name = "your-project-name", # name this to whatever you would like
)

# Import openinference instrumentor to map Pipecat traces to a standard format
from openinference.instrumentation.pipecat import PipecatInstrumentor

# Turn on the instrumentor
PipecatInstrumentor().instrument(tracer_provider=tracer_provider)

Example: Basic Pipecat Pipeline Usage

Build your pipeline as usual. Pass a conversation_id to PipelineTask so each conversation is grouped into a session in Arize:

from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask

# Build your pipeline (transport, STT, LLM, TTS, ...)
pipeline = Pipeline([
    # your stages here
])

# Conversation id groups turns under one session in Arize
task = PipelineTask(
    pipeline,
    conversation_id=conversation_id,
)

@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    await task.queue_frames([LLMRunFrame()])

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

Start running your Pipecat application and monitor traces in Arize. For advanced examples, explore the openinference-instrumentation-pipecat examples.

What is covered by the Instrumentation

Arize provides comprehensive observability for Pipecat’s real-time voice and multimodal pipelines, automatically tracing:

Conversation & Session Tracking

• Conversation Sessions: All turns sharing a conversation_id grouped into a single session

• Turn Boundaries: Each user-to-assistant exchange captured as a parent span

• End-to-End Latency: The full audio-in to audio-out path for every turn

Pipeline Stage Spans

• LLM Calls: Prompts, responses, token counts, and model metadata from your LLM service

• Speech-to-Text (STT): Input audio transcription with latency

• Text-to-Speech (TTS): Output audio synthesis with latency

• Tool / Function Calls: When the LLM service invokes tools, their inputs, outputs, and duration

Performance & Reliability Monitoring

• Stage Latency: Per-stage timing to identify bottlenecks in the audio pipeline

• Token Usage: Prompt, completion, and total tokens across the conversation

• Errors: Failures inside any pipeline stage surfaced as span errors

Verify in Arize AX

Check from the skill, CLI, or SDK

Confirm spans are actually reaching your Arize AX project. Use whichever fits your workflow — the skill and CLI work for any framework; the SDK check is shown for each language.

Arize skill (agent)

Install the Arize Skills plugin and let your coding agent check for you:

npx skills add Arize-ai/arize-skills

Then prompt your agent:

Use the arize-trace skill to export and analyze recent traces from my project. Confirm spans are arriving, and summarize any errors or latency issues.

AX CLI

Export recent spans for your project — any rows mean traces are landing:

ax spans export "$ARIZE_PROJECT_NAME" --space "$ARIZE_SPACE_ID" \
  --limit 5 --stdout | jq 'length'

A non-zero count confirms spans reached Arize AX. Run ax auth login first if you have not authenticated. See the ax spans reference.

SDK

Query the project’s spans and check that at least one came back.

Python

import os
from arize import ArizeClient

client = ArizeClient(api_key=os.environ["ARIZE_API_KEY"])
resp = client.spans.list(
    project=os.environ["ARIZE_PROJECT_NAME"],
    space=os.environ["ARIZE_SPACE_ID"],
    limit=5,
)
count = len(resp.spans)
print(
    f"{count} span(s) found" if count else "No spans yet — recheck setup"
)

SDK span references: Python · TypeScript · Go.