> ## Documentation Index
> Fetch the complete documentation index at: https://arize-ax.mintlify.site/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Pipecat

## Pipecat Voice & Multimodal Agent Tracing

[Pipecat](https://www.pipecat.ai) 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`](https://github.com/Arize-ai/openinference/tree/main/python/instrumentation/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:

```bash theme={null}
pip install openinference-instrumentation-pipecat pipecat-ai arize-otel
```

<Info>
  `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:

  ```bash theme={null}
  pip install 'openinference-instrumentation-pipecat<=0.1.4' 'pipecat-ai<1.0'
  ```
</Info>

#### Instrumentation Setup

Configure the `PipecatInstrumentor` and tracer to send traces to Arize:

```python theme={null}
# 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:

```python theme={null}
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](https://github.com/Arize-ai/openinference/tree/main/python/instrumentation/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.

<Tabs>
  <Tab title="Arize skill (agent)">
    Install the [Arize Skills](https://github.com/Arize-ai/arize-skills) plugin and let your coding agent check for you:

    ```bash theme={null}
    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.
  </Tab>

  <Tab title="AX CLI">
    Export recent spans for your project — any rows mean traces are landing:

    ```bash theme={null}
    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](/api-clients/cli/spans).
  </Tab>

  <Tab title="SDK">
    Query the project's spans and check that at least one came back.

    <CodeGroup>
      ```python Python theme={null}
      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"
      )
      ```

      ```typescript TypeScript theme={null}
      // Reads ARIZE_API_KEY from the environment.
      import { listSpans } from "@arizeai/ax-client";

      const { data: spans } = await listSpans({
        project: process.env.ARIZE_PROJECT_NAME!,
        space: process.env.ARIZE_SPACE_ID!,
        limit: 5,
      });
      const count = spans.length;
      console.log(
        count ? `${count} span(s) found` : "No spans yet — recheck setup",
      );
      ```

      ```go Go theme={null}
      client, err := arize.NewClient(
          arize.Config{APIKey: os.Getenv("ARIZE_API_KEY")},
      )
      if err != nil {
          log.Fatal(err)
      }
      resp, err := client.Spans.List(ctx, spans.ListRequest{
          Project: os.Getenv("ARIZE_PROJECT_NAME"),
          Space:   os.Getenv("ARIZE_SPACE_ID"),
          Limit:   5,
      })
      if err != nil {
          log.Fatal(err)
      }
      fmt.Printf("%d span(s) found\n", len(resp.Spans))
      ```
    </CodeGroup>

    SDK span references: [Python](/api-clients/python/version-8/client-resources/spans) · [TypeScript](/api-clients/typescript/version-1/client-resources/spans) · [Go](/api-clients/go/version-2/client-resources/spans).
  </Tab>
</Tabs>