Python auto-instrumentation library for Pipecat. This library allows you to convert Pipecat traces to OpenInference, which is OpenTelemetry compatible, and view those traces in Arize Phoenix.
openinference-instrumentation-pipecat |
pipecat-ai |
Python |
|---|---|---|
>=1.0 |
>=1.0 |
>=3.11 |
<=0.1.4 |
<1.0 (e.g. 0.0.99) |
>=3.10 |
Pipecat 1.0 introduced breaking changes (renamed observers, removed
LLMMessagesFrame, dropped Python 3.10). If you’re still on pipecat-ai<1.0,
pin this instrumentor to <=0.1.4:
pip install 'openinference-instrumentation-pipecat<=0.1.4' 'pipecat-ai<1.0'
pip install openinference-instrumentation-pipecat
This quickstart shows you how to setup tracing in your Pipecat application:
from phoenix.otel import register
from openinference.instrumentation.pipecat import PipecatInstrumentor
# Set up the tracer provider
tracer_provider = register(
project_name="default" #Phoenix project name
)
# Add auto-instrumentor at the top of the application
PipecatInstrumentor().instrument(tracer_provider=tracer_provider)
# Build your Pipecat pipeline
# ... (add your Pipecat pipeline code here)
### CONFIGURATION SETUP (Transport, LLM, STT, TTS) ###
### PIPELINE ###
pipeline = Pipeline(...)
### TASK ###
task = PipelineTask(
pipeline,
conversation_id=conversation_id, # conversation id is used for session tracking in Arize or Phoenix
)
### EVENT HANDLING
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
await task.queue_frames([LLMRunFrame()])
### PIPELINE RUNNER ###
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
After configuring tracing, exchanges in the running application are logged to your project in Phoenix or Arize AX.
uv pip install -e '.[examples]'
or
uv pip install -r examples/trace/requirements.txt
uv run python examples/trace/001-trace.py