wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add support for OpenTelemetry tracing (#1729)

Browse Source 
* Also added TurnTrackingObserver, TurnTraceObserver, foundational 29, open-telemetry-example
This commit is contained in:
Mark Backman
2025-05-13 17:18:11 -04:00
committed by GitHub
parent 6f4d94f91b
commit 50f6235edb
52 changed files with 2797 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
CHANGELOG.md
View File

@@ -16,6 +16,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`CancelFrame` are pushed from the beginning of the pipeline and finally
`FrameProcessor.cleanup()` is called.
- Added support for OpenTelemetry tracing in Pipecat. This initial
implementation includes:
- A `setup_tracing` method where you can specify your OpenTelemetry exporter
- Service decorators for STT (`@traced_stt`), LLM (`@traced_llm`), and TTS
(`@traced_tts`) which trace the execution and collect properties and
metrics (TTFB, token usage, character counts, etc.)
- Class decorators that provide execution tracking; these are generic and can
be used for service tracking as needed
- Spans that help track traces on a per conversations and turn basis:
```
conversation-uuid
├── turn-1
│ ├── stt_deepgramsttservice
│ ├── llm_openaillmservice
│ └── tts_cartesiattsservice
...
└── turn-n
└── ...
```
By default, Pipecat has implemented service decorators to trace execution of
STT, LLM, and TTS services. You can enable tracing by setting `enable_tracing`
to `True` in the PipelineTask.
- Added `TurnTrackingObserver`, which tracks the start and end of a user/bot
turn pair and emits events `on_turn_started` and `on_turn_stopped`
corresponding to the start and end of a turn, respectively.
- Allow passing observers to `run_test()` while running unit tests.
### Changed
@@ -23,6 +53,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `GoogleLLMService` has been updated to use `google-genai` instead of the
deprecated `google-generativeai`.
### Other
- Added an `open-telemetry-tracing` example, showing how to setup tracing. The
example also includes Jaeger as an open source OpenTelemetry client to review
traces from the example runs.
- Added foundational example `29-turn-tracking-observer.py` to show how to use
the `TurnTrackingObserver.
## [0.0.67] - 2025-05-07
### Added

119
examples/foundational/29-turn-tracking-observer.py Normal file
View File

@@ -0,0 +1,119 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
from dotenv import load_dotenv
from loguru import logger
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.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
report_only_initial_ttfb=True,
),
)
turn_observer = task.turn_tracking_observer
if turn_observer:
@turn_observer.event_handler("on_turn_started")
async def on_turn_started(observer, turn_number):
logger.info(f"🔄 Turn {turn_number} started")
@turn_observer.event_handler("on_turn_ended")
async def on_turn_ended(observer, turn_number, duration, was_interrupted):
if was_interrupted:
logger.info(f"🔄 Turn {turn_number} interrupted after {duration:.2f}s")
else:
logger.info(f"🏁 Turn {turn_number} completed in {duration:.2f}s")
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
from run import main
main()

169
examples/open-telemetry-tracing/README.md Normal file
View File

@@ -0,0 +1,169 @@
# OpenTelemetry Tracing for Pipecat
This demo showcases OpenTelemetry tracing integration for Pipecat services, allowing you to visualize service calls, performance metrics, and dependencies in a Jaeger dashboard.
## Features
- **Hierarchical Tracing**: Track entire conversations, turns, and service calls
- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context
- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis
- **Usage Statistics**: Track character counts for TTS and token usage for LLMs
- **Flexible Exporters**: Use Jaeger, Zipkin, or any OpenTelemetry-compatible backend
## Trace Structure
Traces are organized hierarchically:
```
Conversation (conversation-uuid)
├── turn-1
│ ├── stt_deepgramsttservice
│ ├── llm_openaillmservice
│ └── tts_cartesiattsservice
└── turn-2
├── stt_deepgramsttservice
├── llm_openaillmservice
└── tts_cartesiattsservice
turn-N
└── ...
```
This organization helps you track conversation-to-conversation and turn-to-turn.
## Setup Instructions
### 1. Start the Jaeger Container
Run Jaeger in Docker to collect and visualize traces:
```bash
docker run -d --name jaeger \
-e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
-p 16686:16686 \
-p 4317:4317 \
-p 4318:4318 \
jaegertracing/all-in-one:latest
```
### 2. Environment Configuration
Create a `.env` file with your API keys and enable tracing:
```
ENABLE_TRACING=true
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 # Point to your preferred backend
# OTEL_CONSOLE_EXPORT=true # Set to any value for debug output to console
# Service API keys
DEEPGRAM_API_KEY=your_key_here
CARTESIA_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
```
### 3. Configure Your Pipeline Task
Enable tracing in your Pipecat application:
```python
# Initialize OpenTelemetry with your chosen exporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
endpoint="http://localhost:4317", # Jaeger OTLP endpoint
insecure=True,
)
setup_tracing(
service_name="pipecat-demo",
exporter=exporter,
console_export=os.getenv("OTEL_CONSOLE_EXPORT", "false").lower() == "true",
)
# Enable tracing in your PipelineTask
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True, # Required for some service metrics
),
enable_tracing=True, # Enables both turn and conversation tracing
conversation_id="customer-123", # Optional - will auto-generate if not provided
)
```
### 4. Exporter Options
While this demo uses Jaeger, you can configure any OpenTelemetry-compatible exporter:
#### Jaeger (Default for the demo)
```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
endpoint="http://localhost:4317", # Jaeger OTLP endpoint
insecure=True,
)
```
#### Cloud Providers
Many cloud providers offer OpenTelemetry-compatible observability services:
- AWS X-Ray
- Google Cloud Trace
- Azure Monitor
- Datadog APM
See the OpenTelemetry documentation for specific exporter configurations:
https://opentelemetry.io/ecosystem/vendors/
### 5. Install Dependencies
```bash
pip install -r requirements.txt
```
### 6. Run the Demo
```bash
python bot.py
```
### 7. View Traces in Jaeger
Open your browser to [http://localhost:16686](http://localhost:16686) and select the "pipecat-demo" service to view traces.
## Understanding the Traces
- **Conversation Spans**: The top-level span representing an entire conversation
- **Turn Spans**: Child spans of conversations that represent each turn in the dialog
- **Service Spans**: Detailed service operations nested under turns
- **Service Attributes**: Each service includes rich context about its operation:
- **TTS**: Voice ID, character count, service type
- **STT**: Transcription text, language, model
- **LLM**: Messages, tokens used, model, service configuration
- **Metrics**: Performance data like `metrics.ttfb_ms` and processing durations
## How It Works
The tracing system consists of:
1. **TurnTrackingObserver**: Detects conversation turns
2. **TurnTraceObserver**: Creates spans for turns and conversations
3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing
4. **Context Providers**: Share context between different parts of the pipeline
## Troubleshooting
- **No Traces in Jaeger**: Ensure the Docker container is running and the OTLP endpoint is correct
- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging
- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams
- **Connection Errors**: Verify network connectivity to the Jaeger container
- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works
- **Other Backends**: If using a different backend, ensure you've configured the correct exporter and endpoint
## References
- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/)
- [Jaeger Documentation](https://www.jaegertracing.io/docs/latest/)

159
examples/open-telemetry-tracing/bot.py Normal file
View File

@@ -0,0 +1,159 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
from dotenv import load_dotenv
from loguru import logger
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
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.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
from pipecat.utils.tracing.setup import setup_tracing
load_dotenv(override=True)
IS_TRACING_ENABLED = bool(os.getenv("ENABLE_TRACING"))
# Initialize tracing if enabled
if IS_TRACING_ENABLED:
# Create the exporter
otlp_exporter = OTLPSpanExporter(
endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
insecure=True,
)
# Set up tracing with the exporter
setup_tracing(
service_name="pipecat-demo",
exporter=otlp_exporter,
console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")),
)
logger.info("OpenTelemetry tracing initialized")
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), params=OpenAILLMService.InputParams(temperature=0.5)
)
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
tools = ToolsSchema(standard_tools=[weather_function])
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = OpenAILLMContext(messages, tools)
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(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
),
enable_tracing=IS_TRACING_ENABLED,
# Optionally, add a conversation ID to track the conversation
# conversation_id="8df26cc1-6db0-4a7a-9930-1e037c8f1fa2",
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
from run import main
main()

10
examples/open-telemetry-tracing/env.example Normal file
View File

@@ -0,0 +1,10 @@
DEEPGRAM_API_KEY=your_deepgram_key
CARTESIA_API_KEY=your_cartesia_key
OPENAI_API_KEY=your_openai_key
# Set to any value to enable tracing
ENABLE_TRACING=true
# OTLP endpoint (defaults to localhost:4317 if not set)
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
# Set to any value to enable console output for debugging
# OTEL_CONSOLE_EXPORT=true

6
examples/open-telemetry-tracing/requirements.txt Normal file
View File

@@ -0,0 +1,6 @@
fastapi
uvicorn
python-dotenv
pipecat-ai[webrtc,silero,cartesia,deepgram,openai,tracing]
pipecat-ai-small-webrtc-prebuilt
opentelemetry-exporter-otlp-proto-grpc

205
examples/open-telemetry-tracing/run.py Normal file
View File

@@ -0,0 +1,205 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import importlib.util
import os
import sys
from contextlib import asynccontextmanager
from inspect import iscoroutinefunction, signature
from typing import Any, Callable, Dict, Optional, Tuple
import uvicorn
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI
from fastapi.responses import RedirectResponse
from loguru import logger
from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
# Load environment variables
load_dotenv(override=True)
app = FastAPI()
# Store connections by pc_id
pcs_map: Dict[str, SmallWebRTCConnection] = {}
ice_servers = [
IceServer(
urls="stun:stun.l.google.com:19302",
)
]
# Mount the frontend at /
app.mount("/client", SmallWebRTCPrebuiltUI)
# Store program arguments
args: argparse.Namespace = argparse.Namespace()
# Store the bot module and function info
bot_module: Any = None
run_bot_func: Optional[Callable] = None
is_webrtc_bot: bool = True
def import_bot_file(file_path: str) -> Tuple[Any, Callable, bool]:
"""Dynamically import the bot file and determine how to run it.
Returns:
tuple: (module, run_function, is_webrtc_bot)
- module: The imported module
- run_function: Either run_bot or main function
- is_webrtc_bot: True if run_bot function exists and accepts a WebRTC connection
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"Bot file not found: {file_path}")
# Extract module name without extension
module_name = os.path.splitext(os.path.basename(file_path))[0]
# Load the module
spec = importlib.util.spec_from_file_location(module_name, file_path)
if not spec or not spec.loader:
raise ImportError(f"Could not load spec for {file_path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# Check for run_bot function first
if hasattr(module, "run_bot"):
run_func = module.run_bot
# Check if the function accepts a WebRTC connection
sig = signature(run_func)
is_webrtc = len(sig.parameters) > 0
return module, run_func, is_webrtc
# Fall back to main function
if hasattr(module, "main") and iscoroutinefunction(module.main):
return module, module.main, False
raise AttributeError(f"No run_bot or async main function found in {file_path}")
@app.get("/", include_in_schema=False)
async def root_redirect():
return RedirectResponse(url="/client/")
@app.post("/api/offer")
async def offer(request: dict, background_tasks: BackgroundTasks):
global run_bot_func, is_webrtc_bot
if not run_bot_func:
raise RuntimeError("No bot file has been loaded")
if not is_webrtc_bot:
return {
"error": "This bot doesn't support WebRTC connections, it's running in standalone mode"
}
pc_id = request.get("pc_id")
if pc_id and pc_id in pcs_map:
pipecat_connection = pcs_map[pc_id]
logger.info(f"Reusing existing connection for pc_id: {pc_id}")
await pipecat_connection.renegotiate(
sdp=request["sdp"], type=request["type"], restart_pc=request.get("restart_pc", False)
)
else:
pipecat_connection = SmallWebRTCConnection(ice_servers)
await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
@pipecat_connection.event_handler("closed")
async def handle_disconnected(webrtc_connection: SmallWebRTCConnection):
logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}")
pcs_map.pop(webrtc_connection.pc_id, None)
# We've already checked that run_bot_func exists
assert run_bot_func is not None
background_tasks.add_task(run_bot_func, pipecat_connection, args)
answer = pipecat_connection.get_answer()
# Updating the peer connection inside the map
pcs_map[answer["pc_id"]] = pipecat_connection
return answer
@asynccontextmanager
async def lifespan(app: FastAPI):
yield # Run app
coros = [pc.close() for pc in pcs_map.values()]
await asyncio.gather(*coros)
pcs_map.clear()
async def run_standalone_bot() -> None:
"""Run a standalone bot that doesn't require WebRTC"""
global run_bot_func
if run_bot_func is not None:
await run_bot_func()
else:
raise RuntimeError("No bot function available to run")
def main(parser: Optional[argparse.ArgumentParser] = None):
global args
if not parser:
parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
parser.add_argument("bot_file", nargs="?", help="Path to the bot file", default=None)
parser.add_argument(
"--host", default="localhost", help="Host for HTTP server (default: localhost)"
)
parser.add_argument(
"--port", type=int, default=7860, help="Port for HTTP server (default: 7860)"
)
parser.add_argument("--verbose", "-v", action="count", default=0)
args = parser.parse_args()
logger.remove(0)
if args.verbose:
logger.add(sys.stderr, level="TRACE")
else:
logger.add(sys.stderr, level="DEBUG")
# Infer the bot file from the caller if not provided explicitly
bot_file = args.bot_file
if bot_file is None:
# Get the __file__ of the script that called main()
import inspect
caller_frame = inspect.stack()[1]
caller_globals = caller_frame.frame.f_globals
bot_file = caller_globals.get("__file__")
if not bot_file:
print("❌ Could not determine the bot file. Pass it explicitly to main().")
sys.exit(1)
# Import the bot file
try:
global run_bot_func, bot_module, is_webrtc_bot
bot_module, run_bot_func, is_webrtc_bot = import_bot_file(bot_file)
logger.info(f"Successfully loaded bot from {bot_file}")
if is_webrtc_bot:
logger.info("Detected WebRTC-compatible bot, starting web server...")
uvicorn.run(app, host=args.host, port=args.port)
else:
logger.info("Detected standalone bot, running directly...")
asyncio.run(run_standalone_bot())
except Exception as e:
logger.error(f"Error loading bot file: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

1
pyproject.toml
View File

@@ -87,6 +87,7 @@ simli = [ "simli-ai~=0.1.10"]
soundfile = [ "soundfile~=0.13.0" ]
tavus=[]
together = []
tracing = [ "opentelemetry-sdk>=1.33.0", "opentelemetry-api>=1.33.0", "opentelemetry-instrumentation>=0.54b0" ]
ultravox = [ "transformers~=4.48.0", "vllm~=0.7.3" ]
webrtc = [ "aiortc~=1.11.0", "opencv-python~=4.11.0.86" ]
websocket = [ "websockets~=13.1", "fastapi~=0.115.6" ]

26
src/pipecat/observers/loggers/debug_log_observer.py
View File

@@ -42,7 +42,9 @@ class DebugLogObserver(BaseObserver):
Log specific frame types from any source/destination:
```python
from pipecat.frames.frames import TranscriptionFrame, InterimTranscriptionFrame
observers = DebugLogObserver(frame_types=(TranscriptionFrame, InterimTranscriptionFrame))
observers=[
DebugLogObserver(frame_types=(LLMTextFrame,TranscriptionFrame,)),
],
```
Log frames with specific source/destination filters:
@@ -51,16 +53,18 @@ class DebugLogObserver(BaseObserver):
from pipecat.transports.base_output_transport import BaseOutputTransport
from pipecat.services.stt_service import STTService
observers = DebugLogObserver(frame_types={
# Only log StartInterruptionFrame when source is BaseOutputTransport
StartInterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE),
# Only log UserStartedSpeakingFrame when destination is STTService
UserStartedSpeakingFrame: (STTService, FrameEndpoint.DESTINATION),
# Log LLMTextFrame regardless of source or destination type
LLMTextFrame: None
})
observers=[
DebugLogObserver(
frame_types={
# Only log StartInterruptionFrame when source is BaseOutputTransport
StartInterruptionFrame: (BaseOutputTransport, FrameEndpoint.SOURCE),
# Only log UserStartedSpeakingFrame when destination is STTService
UserStartedSpeakingFrame: (STTService, FrameEndpoint.DESTINATION),
# Log LLMTextFrame regardless of source or destination type
LLMTextFrame: None,
}
),
],
```
"""

185
src/pipecat/observers/turn_tracking_observer.py Normal file
View File

@@ -0,0 +1,185 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
from collections import deque
from loguru import logger
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
StartFrame,
UserStartedSpeakingFrame,
)
from pipecat.observers.base_observer import BaseObserver, FramePushed
class TurnTrackingObserver(BaseObserver):
"""Observer that tracks conversation turns in a pipeline.
Turn tracking logic:
- The first turn starts immediately when the pipeline starts (StartFrame)
- Subsequent turns start when the user starts speaking
- A turn ends when the bot stops speaking and either:
- The user starts speaking again
- A timeout period elapses with no more bot speech
"""
def __init__(self, max_frames=100, turn_end_timeout_secs=2.5, **kwargs):
super().__init__()
self._turn_count = 0
self._is_turn_active = False
self._is_bot_speaking = False
self._has_bot_spoken = False
self._turn_start_time = 0
self._turn_end_timeout_secs = turn_end_timeout_secs
self._end_turn_timer = None
# Track processed frames to avoid duplicates
self._processed_frames = set()
self._frame_history = deque(maxlen=max_frames)
self._register_event_handler("on_turn_started")
self._register_event_handler("on_turn_ended")
async def on_push_frame(self, data: FramePushed):
"""Process frame events for turn tracking."""
# Skip already processed frames
if data.frame.id in self._processed_frames:
return
self._processed_frames.add(data.frame.id)
self._frame_history.append(data.frame.id)
# If we've exceeded our history size, remove the oldest frame ID
# from the set of processed frames.
if len(self._processed_frames) > len(self._frame_history):
# Rebuild the set from the current deque contents
self._processed_frames = set(self._frame_history)
if isinstance(data.frame, StartFrame):
# Start the first turn immediately when the pipeline starts
if self._turn_count == 0:
await self._start_turn(data)
elif isinstance(data.frame, UserStartedSpeakingFrame):
await self._handle_user_started_speaking(data)
elif isinstance(data.frame, BotStartedSpeakingFrame):
await self._handle_bot_started_speaking(data)
# A BotStoppedSpeakingFrame can arrive after a UserStartedSpeakingFrame following an interruption
# We only want to end the turn if the bot was previously speaking
elif isinstance(data.frame, BotStoppedSpeakingFrame) and self._is_bot_speaking:
await self._handle_bot_stopped_speaking(data)
def _schedule_turn_end(self, data: FramePushed):
"""Schedule turn end with a timeout."""
# Cancel any existing timer
self._cancel_turn_end_timer()
# Create a new timer
loop = asyncio.get_event_loop()
self._end_turn_timer = loop.call_later(
self._turn_end_timeout_secs,
lambda: asyncio.create_task(self._end_turn_after_timeout(data)),
)
def _cancel_turn_end_timer(self):
"""Cancel the turn end timer if it exists."""
if self._end_turn_timer:
self._end_turn_timer.cancel()
self._end_turn_timer = None
async def _end_turn_after_timeout(self, data: FramePushed):
"""End turn after timeout has expired."""
if self._is_turn_active and not self._is_bot_speaking:
logger.trace(f"Turn {self._turn_count} ending due to timeout")
await self._end_turn(data, was_interrupted=False)
self._end_turn_timer = None
async def _handle_user_started_speaking(self, data: FramePushed):
"""Handle user speaking events, including interruptions."""
if self._is_bot_speaking:
# Handle interruption - end current turn and start a new one
self._cancel_turn_end_timer() # Cancel any pending end turn timer
await self._end_turn(data, was_interrupted=True)
self._is_bot_speaking = False # Bot is considered interrupted
await self._start_turn(data)
elif self._is_turn_active and self._has_bot_spoken:
# User started speaking during the turn_end_timeout_secs period after bot speech
self._cancel_turn_end_timer() # Cancel any pending end turn timer
await self._end_turn(data, was_interrupted=False)
await self._start_turn(data)
elif not self._is_turn_active:
# Start a new turn after previous one ended
await self._start_turn(data)
else:
# User is speaking within the same turn (before bot has responded)
logger.trace(f"User is already speaking in Turn {self._turn_count}")
async def _handle_bot_started_speaking(self, data: FramePushed):
"""Handle bot speaking events."""
self._is_bot_speaking = True
self._has_bot_spoken = True
# Cancel any pending turn end timer when bot starts speaking again
self._cancel_turn_end_timer()
async def _handle_bot_stopped_speaking(self, data: FramePushed):
"""Handle bot stopped speaking events."""
self._is_bot_speaking = False
# Schedule turn end with timeout
# This is needed to handle cases where the bot's speech ends and then resumes
# This can happen with HTTP TTS services or function calls
self._schedule_turn_end(data)
async def _start_turn(self, data: FramePushed):
"""Start a new turn."""
self._is_turn_active = True
self._has_bot_spoken = False
self._turn_count += 1
self._turn_start_time = data.timestamp
logger.trace(f"Turn {self._turn_count} started")
await self._call_event_handler("on_turn_started", self._turn_count)
async def _end_turn(self, data: FramePushed, was_interrupted: bool):
"""End the current turn."""
if not self._is_turn_active:
return
duration = (data.timestamp - self._turn_start_time) / 1_000_000_000 # Convert to seconds
self._is_turn_active = False
status = "interrupted" if was_interrupted else "completed"
logger.trace(f"Turn {self._turn_count} {status} after {duration:.2f}s")
await self._call_event_handler("on_turn_ended", self._turn_count, duration, was_interrupted)
def _register_event_handler(self, event_name):
"""Register an event handler."""
if not hasattr(self, "_event_handlers"):
self._event_handlers = {}
if event_name not in self._event_handlers:
self._event_handlers[event_name] = []
async def _call_event_handler(self, event_name, *args, **kwargs):
"""Call registered event handlers."""
if not hasattr(self, "_event_handlers"):
return
if event_name in self._event_handlers:
for handler in self._event_handlers[event_name]:
await handler(self, *args, **kwargs)
def event_handler(self, event_name):
"""Decorator for registering event handlers."""
def decorator(func):
if not hasattr(self, "_event_handlers"):
self._event_handlers = {}
if event_name not in self._event_handlers:
self._event_handlers[event_name] = []
self._event_handlers[event_name].append(func)
return func
return decorator

33
src/pipecat/pipeline/task.py
View File

@@ -30,11 +30,14 @@ from pipecat.frames.frames import (
)
from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
from pipecat.observers.base_observer import BaseObserver
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.base_task import BaseTask
from pipecat.pipeline.task_observer import TaskObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.utils.asyncio import BaseTaskManager, TaskManager
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
HEARTBEAT_SECONDS = 1.0
HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
@@ -157,6 +160,8 @@ class PipelineTask(BaseTask):
timeout if not received withing `idle_timeout_seconds`.
cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
the idle timeout is reached.
enable_turn_tracking: Whether to enable turn tracking.
enable_turn_tracing: Whether to enable turn tracing.
"""
@@ -175,6 +180,9 @@ class PipelineTask(BaseTask):
LLMFullResponseEndFrame,
),
cancel_on_idle_timeout: bool = True,
enable_turn_tracking: bool = True,
enable_tracing: bool = False,
conversation_id: Optional[str] = None,
):
super().__init__()
self._pipeline = pipeline
@@ -184,6 +192,9 @@ class PipelineTask(BaseTask):
self._idle_timeout_secs = idle_timeout_secs
self._idle_timeout_frames = idle_timeout_frames
self._cancel_on_idle_timeout = cancel_on_idle_timeout
self._enable_turn_tracking = enable_turn_tracking
self._enable_tracing = enable_tracing and is_tracing_available()
self._conversation_id = conversation_id
if self._params.observers:
import warnings
@@ -194,6 +205,14 @@ class PipelineTask(BaseTask):
DeprecationWarning,
)
observers = self._params.observers
if self._enable_turn_tracking:
self._turn_tracking_observer = TurnTrackingObserver()
observers = [self._turn_tracking_observer] + list(observers)
if self._enable_turn_tracking and self._enable_tracing:
self._turn_trace_observer = TurnTraceObserver(
self._turn_tracking_observer, conversation_id=self._conversation_id
)
observers = [self._turn_trace_observer] + list(observers)
self._finished = False
# This queue receives frames coming from the pipeline upstream.
@@ -251,6 +270,16 @@ class PipelineTask(BaseTask):
"""Returns the pipeline parameters of this task."""
return self._params
@property
def turn_tracking_observer(self) -> Optional[TurnTrackingObserver]:
"""Return the turn tracking observer if enabled."""
return getattr(self, "_turn_tracking_observer", None)
@property
def turn_trace_observer(self) -> Optional[TurnTraceObserver]:
"""Return the turn trace observer if enabled."""
return getattr(self, "_turn_trace_observer", None)
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
self._task_manager.set_event_loop(loop)
@@ -426,6 +455,10 @@ class PipelineTask(BaseTask):
# Cleanup base object.
await self.cleanup()
# End conversation tracing if it's active - this will also close any active turn span
if self._enable_tracing and hasattr(self, "_turn_trace_observer"):
self._turn_trace_observer.end_conversation_tracing()
# Cleanup pipeline processors.
await self._source.cleanup()
if cleanup_pipeline:

25
src/pipecat/processors/metrics/frame_processor_metrics.py
View File

@@ -5,6 +5,7 @@
#
import time
from typing import Optional
from loguru import logger
@@ -23,8 +24,25 @@ class FrameProcessorMetrics:
def __init__(self):
self._start_ttfb_time = 0
self._start_processing_time = 0
self._last_ttfb_time = 0
self._should_report_ttfb = True
@property
def ttfb_ms(self) -> Optional[float]:
"""Get the current TTFB value in seconds.
Returns:
Optional[float]: The TTFB value in seconds, or None if not measured
"""
if self._last_ttfb_time > 0:
return self._last_ttfb_time
# If TTFB is in progress, calculate current value
if self._start_ttfb_time > 0:
return time.time() - self._start_ttfb_time
return None
def _processor_name(self):
return self._core_metrics_data.processor
@@ -40,16 +58,17 @@ class FrameProcessorMetrics:
async def start_ttfb_metrics(self, report_only_initial_ttfb):
if self._should_report_ttfb:
self._start_ttfb_time = time.time()
self._last_ttfb_time = 0
self._should_report_ttfb = not report_only_initial_ttfb
async def stop_ttfb_metrics(self):
if self._start_ttfb_time == 0:
return None
value = time.time() - self._start_ttfb_time
logger.debug(f"{self._processor_name()} TTFB: {value}")
self._last_ttfb_time = time.time() - self._start_ttfb_time
logger.debug(f"{self._processor_name()} TTFB: {self._last_ttfb_time}")
ttfb = TTFBMetricsData(
processor=self._processor_name(), value=value, model=self._model_name()
processor=self._processor_name(), value=self._last_ttfb_time, model=self._model_name()
)
self._start_ttfb_time = 0
return MetricsFrame(data=[ttfb])

2
src/pipecat/services/anthropic/llm.py
View File

@@ -46,6 +46,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.utils.tracing.service_decorators import traced_llm
try:
from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
@@ -147,6 +148,7 @@ class AnthropicLLMService(LLMService):
assistant = AnthropicAssistantContextAggregator(context, params=assistant_params)
return AnthropicContextAggregatorPair(_user=user, _assistant=assistant)
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
# Usage tracking. We track the usage reported by Anthropic in prompt_tokens and
# completion_tokens. We also estimate the completion tokens from output text

32
src/pipecat/services/assemblyai/stt.py
View File

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import assemblyai as aai
@@ -51,6 +52,9 @@ class AssemblyAISTTService(STTService):
"language": language,
}
def can_generate_metrics(self) -> bool:
return True
async def set_language(self, language: Language):
logger.info(f"Switching STT language to: [{language}]")
self._settings["language"] = language
@@ -77,18 +81,25 @@ class AssemblyAISTTService(STTService):
:yield: None (transcription frames are pushed via self.push_frame in callbacks)
"""
if self._transcriber:
await self.start_ttfb_metrics()
await self.start_processing_metrics()
self._transcriber.stream(audio)
await self.stop_processing_metrics()
yield None
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
async def _connect(self):
"""Establish a connection to the AssemblyAI real-time transcription service.
This method sets up the necessary callback functions and initializes the
AssemblyAI transcriber.
"""
if self._transcriber:
return
@@ -107,15 +118,18 @@ class AssemblyAISTTService(STTService):
return
timestamp = time_now_iso8601()
is_final = isinstance(transcript, aai.RealtimeFinalTranscript)
language = self._settings["language"]
if isinstance(transcript, aai.RealtimeFinalTranscript):
frame = TranscriptionFrame(
transcript.text, "", timestamp, self._settings["language"]
)
if is_final:
frame = TranscriptionFrame(transcript.text, "", timestamp, language)
else:
frame = InterimTranscriptionFrame(
transcript.text, "", timestamp, self._settings["language"]
)
frame = InterimTranscriptionFrame(transcript.text, "", timestamp, language)
asyncio.run_coroutine_threadsafe(
self._handle_transcription(transcript.text, is_final, language),
self.get_event_loop(),
)
# Schedule the coroutine to run in the main event loop
# This is necessary because this callback runs in a different thread

2
src/pipecat/services/aws/llm.py
View File

@@ -44,6 +44,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.utils.tracing.service_decorators import traced_llm
try:
import boto3
@@ -603,6 +604,7 @@ class AWSBedrockLLMService(LLMService):
assistant = AWSBedrockAssistantContextAggregator(context, params=assistant_params)
return AWSBedrockContextAggregatorPair(_user=user, _assistant=assistant)
@traced_llm
async def _process_context(self, context: AWSBedrockLLMContext):
# Usage tracking
prompt_tokens = 0

12
src/pipecat/services/aws/stt.py
View File

@@ -26,6 +26,7 @@ from pipecat.services.aws.utils import build_event_message, decode_event, get_pr
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import websockets
@@ -268,6 +269,12 @@ class AWSTranscribeSTTService(STTService):
}
return language_map.get(language)
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[str] = None
):
pass
async def _receive_loop(self):
"""Background task to receive and process messages from AWS Transcribe."""
while True:
@@ -300,6 +307,11 @@ class AWSTranscribeSTTService(STTService):
self._settings["language"],
)
)
await self._handle_transcription(
transcript,
is_final,
self._settings["language"],
)
await self.stop_processing_metrics()
else:
await self.push_frame(

2
src/pipecat/services/aws/tts.py
View File

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
import boto3
@@ -207,6 +208,7 @@ class AWSPollyTTSService(TTSService):
return ssml
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
def read_audio_data(**args):
response = self._polly_client.synthesize_speech(**args)

25
src/pipecat/services/azure/stt.py
View File

@@ -20,6 +20,7 @@ from pipecat.services.azure.common import language_to_azure_language
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
from azure.cognitiveservices.speech import (
@@ -58,12 +59,20 @@ class AzureSTTService(STTService):
self._audio_stream = None
self._speech_recognizer = None
self._settings = {
"region": region,
"language": language_to_azure_language(language),
"sample_rate": sample_rate,
}
def can_generate_metrics(self) -> bool:
return True
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()
await self.start_ttfb_metrics()
if self._audio_stream:
self._audio_stream.write(audio)
await self.stop_processing_metrics()
yield None
async def start(self, frame: StartFrame):
@@ -101,7 +110,19 @@ class AzureSTTService(STTService):
if self._audio_stream:
self._audio_stream.close()
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
def _on_handle_recognized(self, event):
if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:
frame = TranscriptionFrame(event.result.text, "", time_now_iso8601())
language = getattr(event.result, "language", None) or self._settings.get("language")
frame = TranscriptionFrame(event.result.text, "", time_now_iso8601(), language)
asyncio.run_coroutine_threadsafe(
self._handle_transcription(event.result.text, True, language), self.get_event_loop()
)
asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop())

3
src/pipecat/services/azure/tts.py
View File

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
from pipecat.services.azure.common import language_to_azure_language
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
from azure.cognitiveservices.speech import (
@@ -196,6 +197,7 @@ class AzureTTSService(AzureBaseTTSService):
async def flush_audio(self):
logger.trace(f"{self}: flushing audio")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -263,6 +265,7 @@ class AzureHttpTTSService(AzureBaseTTSService):
speech_config=self._speech_config, audio_config=None
)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

3
src/pipecat/services/cartesia/tts.py
View File

@@ -28,6 +28,7 @@ from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
# See .env.example for Cartesia configuration needed
try:
@@ -274,6 +275,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
else:
logger.error(f"{self} error, unknown message type: {msg}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -360,6 +362,7 @@ class CartesiaHttpTTSService(TTSService):
await super().cancel(frame)
await self._client.close()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

10
src/pipecat/services/deepgram/stt.py
View File

@@ -22,6 +22,7 @@ from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
from deepgram import (
@@ -187,6 +188,13 @@ class DeepgramSTTService(STTService):
async def _on_utterance_end(self, *args, **kwargs):
await self._call_event_handler("on_utterance_end", *args, **kwargs)
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def _on_message(self, *args, **kwargs):
result: LiveResultResponse = kwargs["result"]
if len(result.channel.alternatives) == 0:
@@ -203,8 +211,10 @@ class DeepgramSTTService(STTService):
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), language)
)
await self._handle_transcription(transcript, is_final, language)
await self.stop_processing_metrics()
else:
# For interim transcriptions, just push the frame without tracing
await self.push_frame(
InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language)
)

2
src/pipecat/services/deepgram/tts.py
View File

@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.utils.tracing.service_decorators import traced_tts
try:
from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions
@@ -49,6 +50,7 @@ class DeepgramTTSService(TTSService):
def can_generate_metrics(self) -> bool:
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

3
src/pipecat/services/elevenlabs/tts.py
View File

@@ -32,6 +32,7 @@ from pipecat.services.tts_service import (
WordTTSService,
)
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
# See .env.example for ElevenLabs configuration needed
try:
@@ -445,6 +446,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
msg = {"text": text, "context_id": self._context_id}
await self._websocket.send(json.dumps(msg))
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -645,6 +647,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
return word_times
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs streaming API with timestamps.

13
src/pipecat/services/fal/stt.py
View File

@@ -14,6 +14,7 @@ from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import fal_client
@@ -211,6 +212,14 @@ class FalSTTService(SegmentedSTTService):
await super().set_model(model)
logger.info(f"Switching STT model to: [{model}]")
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[str] = None
):
"""Handle a transcription result with tracing."""
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Transcribes an audio segment using Fal's Wizper API.
@@ -225,6 +234,9 @@ class FalSTTService(SegmentedSTTService):
Only non-empty transcriptions are yielded.
"""
try:
await self.start_processing_metrics()
await self.start_ttfb_metrics()
# Send to Fal directly (audio is already in WAV format from base class)
data_uri = fal_client.encode(audio, "audio/x-wav")
response = await self._fal_client.run(
@@ -235,6 +247,7 @@ class FalSTTService(SegmentedSTTService):
if response and "text" in response:
text = response["text"].strip()
if text: # Only yield non-empty text
await self._handle_transcription(text, True, self._settings["language"])
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(
text, "", time_now_iso8601(), Language(self._settings["language"])

2
src/pipecat/services/fish/tts.py
View File

@@ -24,6 +24,7 @@ from pipecat.frames.frames import (
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
import ormsgpack
@@ -186,6 +187,7 @@ class FishAudioTTSService(InterruptibleTTSService):
except Exception as e:
logger.error(f"Error processing message: {e}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating Fish TTS: [{text}]")
try:

25
src/pipecat/services/gladia/stt.py
View File

@@ -26,6 +26,7 @@ from pipecat.services.gladia.config import GladiaInputParams
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import websockets
@@ -227,6 +228,10 @@ class GladiaSTTService(STTService):
self._websocket = None
self._receive_task = None
self._keepalive_task = None
self._settings = {}
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert pipecat Language enum to Gladia's language code."""
@@ -278,6 +283,9 @@ class GladiaSTTService(STTService):
if self._params.messages_config:
settings["messages_config"] = self._params.messages_config.model_dump(exclude_none=True)
# Store settings for tracing
self._settings = settings
return settings
async def start(self, frame: StartFrame):
@@ -328,9 +336,9 @@ class GladiaSTTService(STTService):
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Run speech-to-text on audio data."""
await self.start_ttfb_metrics()
await self.start_processing_metrics()
await self._send_audio(audio)
await self.stop_processing_metrics()
yield None
async def _setup_gladia(self, settings: Dict[str, Any]):
@@ -351,6 +359,13 @@ class GladiaSTTService(STTService):
f"Failed to initialize Gladia session: {response.status} - {error_text}"
)
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[str] = None
):
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
async def _send_audio(self, audio: bytes):
data = base64.b64encode(audio).decode("utf-8")
message = {"type": "audio_chunk", "data": {"chunk": data}}
@@ -387,11 +402,17 @@ class GladiaSTTService(STTService):
confidence = utterance.get("confidence", 0)
language = utterance["language"]
transcript = utterance["text"]
is_final = content["data"]["is_final"]
if confidence >= self._confidence:
if content["data"]["is_final"]:
if is_final:
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), language)
)
await self._handle_transcription(
transcript=transcript,
is_final=is_final,
language=language,
)
else:
await self.push_frame(
InterimTranscriptionFrame(

2
src/pipecat/services/google/llm.py
View File

@@ -47,6 +47,7 @@ from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
)
from pipecat.utils.tracing.service_decorators import traced_llm
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
@@ -493,6 +494,7 @@ class GoogleLLMService(LLMService):
def _create_client(self, api_key: str):
self._client = genai.Client(api_key=api_key)
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
await self.push_frame(LLMFullResponseStartFrame())

20
src/pipecat/services/google/stt.py
View File

@@ -9,6 +9,8 @@ import json
import os
import time
from pipecat.utils.tracing.service_decorators import traced_stt
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
@@ -496,6 +498,9 @@ class GoogleSTTService(STTService):
"enable_voice_activity_events": params.enable_voice_activity_events,
}
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language | List[Language]) -> str | List[str]:
"""Convert Language enum(s) to Google STT language code(s).
@@ -773,9 +778,17 @@ class GoogleSTTService(STTService):
"""Process an audio chunk for STT transcription."""
if self._streaming_task:
# Queue the audio data
await self.start_ttfb_metrics()
await self.start_processing_metrics()
await self._request_queue.put(audio)
yield None
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[str] = None
):
pass
async def _process_responses(self, streaming_recognize):
"""Process streaming recognition responses."""
try:
@@ -803,8 +816,15 @@ class GoogleSTTService(STTService):
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), primary_language)
)
await self.stop_processing_metrics()
await self._handle_transcription(
transcript,
is_final=True,
language=primary_language,
)
else:
self._last_transcript_was_final = False
await self.stop_ttfb_metrics()
await self.push_frame(
InterimTranscriptionFrame(
transcript, "", time_now_iso8601(), primary_language

3
src/pipecat/services/google/tts.py
View File

@@ -8,6 +8,8 @@ import asyncio
import json
import os
from pipecat.utils.tracing.service_decorators import traced_tts
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
@@ -318,6 +320,7 @@ class GoogleTTSService(TTSService):
return ssml
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

12
src/pipecat/services/groq/tts.py
View File

@@ -12,6 +12,7 @@ from pydantic import BaseModel
from pipecat.frames.frames import Frame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
from groq import AsyncGroq
@@ -25,7 +26,6 @@ class GroqTTSService(TTSService):
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
speed: Optional[float] = 1.0
seed: Optional[int] = None
GROQ_SAMPLE_RATE = 48000 # Groq TTS only supports 48kHz sample rate
@@ -54,11 +54,21 @@ class GroqTTSService(TTSService):
self._voice_id = voice_id
self._params = params
self._settings = {
"model": model_name,
"voice_id": voice_id,
"output_format": output_format,
"language": str(params.language) if params.language else "en",
"speed": params.speed,
"sample_rate": sample_rate,
}
self._client = AsyncGroq(api_key=self._api_key)
def can_generate_metrics(self) -> bool:
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
measuring_ttfb = True

2
src/pipecat/services/lmnt/tts.py
View File

@@ -23,6 +23,7 @@ from pipecat.frames.frames import (
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
# See .env.example for LMNT configuration needed
try:
@@ -198,6 +199,7 @@ class LmntTTSService(InterruptibleTTSService):
except json.JSONDecodeError:
logger.error(f"Invalid JSON message: {message}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio from text."""
logger.debug(f"{self}: Generating TTS [{text}]")

3
src/pipecat/services/neuphonic/tts.py
View File

@@ -29,6 +29,7 @@ from pipecat.frames.frames import (
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
import websockets
@@ -239,6 +240,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
logger.debug(f"Sending text to websocket: {msg}")
await self._websocket.send(json.dumps(msg))
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
@@ -315,6 +317,7 @@ class NeuphonicHttpTTSService(TTSService):
async def flush_audio(self):
pass
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Neuphonic streaming API.

2
src/pipecat/services/openai/base_llm.py
View File

@@ -35,6 +35,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.utils.tracing.service_decorators import traced_llm
class OpenAIUnhandledFunctionException(Exception):
@@ -176,6 +177,7 @@ class BaseOpenAILLMService(LLMService):
return chunks
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
functions_list = []
arguments_list = []

2
src/pipecat/services/openai/tts.py
View File

@@ -18,6 +18,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.utils.tracing.service_decorators import traced_tts
ValidVoice = Literal[
"alloy", "ash", "ballad", "coral", "echo", "fable", "onyx", "nova", "sage", "shimmer", "verse"
@@ -94,6 +95,7 @@ class OpenAITTSService(TTSService):
f"Current rate of {self.sample_rate}Hz may cause issues."
)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
try:

2
src/pipecat/services/piper/tts.py
View File

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.utils.tracing.service_decorators import traced_tts
# This assumes a running TTS service running: https://github.com/rhasspy/piper/blob/master/src/python_run/README_http.md
@@ -54,6 +55,7 @@ class PiperTTSService(TTSService):
def can_generate_metrics(self) -> bool:
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Piper API.

3
src/pipecat/services/playht/tts.py
View File

@@ -29,6 +29,7 @@ from pipecat.frames.frames import (
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
from pyht.async_client import AsyncClient
@@ -268,6 +269,7 @@ class PlayHTTTSService(InterruptibleTTSService):
except json.JSONDecodeError:
logger.error(f"Invalid JSON message: {message}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -391,6 +393,7 @@ class PlayHTHttpTTSService(TTSService):
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_playht_language(language)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

3
src/pipecat/services/rime/tts.py
View File

@@ -29,6 +29,7 @@ from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
try:
import websockets
@@ -310,6 +311,7 @@ class RimeTTSService(AudioContextWordTTSService):
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("Reset", 0)])
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text.
@@ -385,6 +387,7 @@ class RimeHttpTTSService(TTSService):
def can_generate_metrics(self) -> bool:
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

37
src/pipecat/services/riva/stt.py
View File

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
from pipecat.services.stt_service import SegmentedSTTService, STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import riva.client
@@ -118,6 +119,15 @@ class RivaSTTService(STTService):
self._custom_configuration = ""
self._function_id = model_function_map.get("function_id")
self._settings = {
"language": str(params.language),
"profanity_filter": self._profanity_filter,
"automatic_punctuation": self._automatic_punctuation,
"verbatim_transcripts": not self._no_verbatim_transcripts,
"boosted_lm_words": self._boosted_lm_words,
"boosted_lm_score": self._boosted_lm_score,
}
self.set_model_name(model_function_map.get("model_name"))
metadata = [
@@ -225,6 +235,13 @@ class RivaSTTService(STTService):
self._thread_running = False
raise
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def _handle_response(self, response):
for result in response.results:
if result and not result.alternatives:
@@ -236,11 +253,18 @@ class RivaSTTService(STTService):
if result.is_final:
await self.stop_processing_metrics()
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), None)
TranscriptionFrame(transcript, "", time_now_iso8601(), self._language_code)
)
await self._handle_transcription(
transcript=transcript,
is_final=result.is_final,
language=self._language_code,
)
else:
await self.push_frame(
InterimTranscriptionFrame(transcript, "", time_now_iso8601(), None)
InterimTranscriptionFrame(
transcript, "", time_now_iso8601(), self._language_code
)
)
async def _response_task_handler(self):
@@ -249,6 +273,8 @@ class RivaSTTService(STTService):
await self._handle_response(response)
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_ttfb_metrics()
await self.start_processing_metrics()
await self._queue.put(audio)
yield None
@@ -418,6 +444,11 @@ class RivaSegmentedSTTService(SegmentedSTTService):
if self._config:
self._config.language_code = self._language
@traced_stt
async def _handle_transcription(self, transcript: str, language: Optional[Language] = None):
"""Handle a transcription result with tracing."""
pass
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Transcribe an audio segment.
@@ -475,6 +506,8 @@ class RivaSegmentedSTTService(SegmentedSTTService):
)
transcription_found = True
await self._handle_transcription(text, True, self._language_enum)
if not transcription_found:
logger.debug("No transcription results found in Riva response")

3
src/pipecat/services/riva/tts.py
View File

@@ -8,6 +8,8 @@ import asyncio
import os
from typing import AsyncGenerator, Mapping, Optional
from pipecat.utils.tracing.service_decorators import traced_tts
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
@@ -83,6 +85,7 @@ class RivaTTSService(TTSService):
f"{self.__class__.__name__}(api_key=<api_key>, model_function_map={example})"
)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
def read_audio_responses(queue: asyncio.Queue):
def add_response(r):

16
src/pipecat/services/whisper/base_stt.py
View File

@@ -14,6 +14,7 @@ from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_whisper_language(language: Language) -> Optional[str]:
@@ -126,6 +127,13 @@ class BaseWhisperSTTService(SegmentedSTTService):
self._prompt = prompt
self._temperature = temperature
self._settings = {
"base_url": base_url,
"language": self._language,
"prompt": self._prompt,
"temperature": self._temperature,
}
def _create_client(self, api_key: Optional[str], base_url: Optional[str]):
return AsyncOpenAI(api_key=api_key, base_url=base_url)
@@ -147,6 +155,13 @@ class BaseWhisperSTTService(SegmentedSTTService):
logger.info(f"Switching STT language to: [{language}]")
self._language = language
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
try:
await self.start_processing_metrics()
@@ -160,6 +175,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
text = response.text.strip()
if text:
await self._handle_transcription(text, True, self._language)
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(text, "", time_now_iso8601())
else:

23
src/pipecat/services/whisper/stt.py
View File

@@ -18,6 +18,7 @@ from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
if TYPE_CHECKING:
try:
@@ -291,6 +292,9 @@ class WhisperSTTService(SegmentedSTTService):
self._settings = {
"language": language,
"device": self._device,
"compute_type": self._compute_type,
"no_speech_prob": self._no_speech_prob,
}
self._load()
@@ -343,6 +347,13 @@ class WhisperSTTService(SegmentedSTTService):
logger.error("In order to use Whisper, you need to `pip install pipecat-ai[whisper]`.")
self._model = None
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Transcribes given audio using Whisper.
@@ -381,6 +392,7 @@ class WhisperSTTService(SegmentedSTTService):
await self.stop_processing_metrics()
if text:
await self._handle_transcription(text, True, self._settings["language"])
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(text, "", time_now_iso8601(), self._settings["language"])
@@ -422,6 +434,9 @@ class WhisperSTTServiceMLX(WhisperSTTService):
self._settings = {
"language": language,
"no_speech_prob": self._no_speech_prob,
"temperature": self._temperature,
"engine": "mlx",
}
# No need to call _load() as MLX Whisper loads models on demand
@@ -431,6 +446,13 @@ class WhisperSTTServiceMLX(WhisperSTTService):
"""MLX Whisper loads models on demand, so this is a no-op."""
pass
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
@override
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Transcribes given audio using MLX Whisper.
@@ -479,6 +501,7 @@ class WhisperSTTServiceMLX(WhisperSTTService):
await self.stop_processing_metrics()
if text:
await self._handle_transcription(text, True, self._settings["language"])
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(text, "", time_now_iso8601(), self._settings["language"])

2
src/pipecat/services/xtts/tts.py
View File

@@ -20,6 +20,7 @@ from pipecat.frames.frames import (
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
# The server below can connect to XTTS through a local running docker
#
@@ -117,6 +118,7 @@ class XTTSService(TTSService):
return
self._studio_speakers = await r.json()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")

7
src/pipecat/utils/tracing/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenTelemetry tracing utilities for Pipecat."""

219
src/pipecat/utils/tracing/class_decorators.py Normal file
View File

@@ -0,0 +1,219 @@
#
# Copyright (c) 2024–2025, Daily
# Portions Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base OpenTelemetry tracing decorators and utilities for Pipecat.
This module provides class and method level tracing capabilities
similar to the original NVIDIA implementation.
"""
import asyncio
import contextlib
import enum
import functools
import inspect
from typing import Callable, Optional, TypeVar
from pipecat.utils.tracing.setup import OPENTELEMETRY_AVAILABLE
# Import OpenTelemetry if available
if OPENTELEMETRY_AVAILABLE:
import opentelemetry.trace
from opentelemetry import metrics, trace
# Type variables for better typing support
T = TypeVar("T")
C = TypeVar("C", bound=type)
class AttachmentStrategy(enum.Enum):
"""Controls how spans are attached to the trace hierarchy.
Attributes:
CHILD: Attached to class span if no parent, otherwise to parent.
LINK: Attached to class span with link to parent.
NONE: Always attached to class span regardless of context.
"""
CHILD = enum.auto()
LINK = enum.auto()
NONE = enum.auto()
class Traceable:
"""Base class for objects that can be traced with OpenTelemetry.
Provides the foundational tracing capabilities used by @traced methods.
"""
def __init__(self, name: str, **kwargs):
"""Initialize a traceable object.
Args:
name: Name of the traceable object for the span.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
if not OPENTELEMETRY_AVAILABLE:
self._tracer = self._meter = self._parent_span_id = self._span = None
return
self._tracer = trace.get_tracer("pipecat")
self._meter = metrics.get_meter("pipecat")
self._parent_span_id = trace.get_current_span().get_span_context().span_id
self._span = self._tracer.start_span(name)
self._span.end()
@property
def meter(self):
"""Returns the OpenTelemetry meter instance.
Returns:
Meter: The OpenTelemetry meter instance for this object.
"""
return self._meter
@contextlib.contextmanager
def __traced_context_manager(
self: Traceable, func: Callable, name: str | None, attachment_strategy: AttachmentStrategy
):
"""Internal context manager for the traced decorator."""
if not isinstance(self, Traceable):
raise RuntimeError(
"@traced annotation can only be used in classes inheriting from Traceable"
)
stack = contextlib.ExitStack()
try:
current_span = trace.get_current_span()
is_span_class_parent_span = current_span.get_span_context().span_id == self._parent_span_id
match attachment_strategy:
case AttachmentStrategy.CHILD if not is_span_class_parent_span:
stack.enter_context(
self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore
)
case AttachmentStrategy.LINK:
if is_span_class_parent_span:
link = trace.Link(self._span.get_span_context()) # type: ignore
else:
link = trace.Link(current_span.get_span_context())
stack.enter_context(
opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore
)
stack.enter_context(
self._tracer.start_as_current_span( # type: ignore
func.__name__ if name is None else name, links=[link]
)
)
case AttachmentStrategy.NONE | AttachmentStrategy.CHILD:
stack.enter_context(
opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore
)
stack.enter_context(
self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore
)
yield
finally:
stack.close()
def __traced_decorator(func, name, attachment_strategy: AttachmentStrategy):
"""Implementation of the traced decorator."""
@functools.wraps(func)
async def coroutine_wrapper(self: Traceable, *args, **kwargs):
exception = None
with __traced_context_manager(self, func, name, attachment_strategy):
try:
return await func(self, *args, **kwargs)
except asyncio.CancelledError as e:
exception = e
if exception:
raise exception
@functools.wraps(func)
async def generator_wrapper(self: Traceable, *args, **kwargs):
exception = None
with __traced_context_manager(self, func, name, attachment_strategy):
try:
async for v in func(self, *args, **kwargs):
yield v
except asyncio.CancelledError as e:
exception = e
if exception:
raise exception
if inspect.iscoroutinefunction(func):
return coroutine_wrapper
if inspect.isasyncgenfunction(func):
return generator_wrapper
raise ValueError("@traced annotation can only be used on async or async generator functions")
def traced(
func: Optional[Callable] = None,
*,
name: Optional[str] = None,
attachment_strategy: AttachmentStrategy = AttachmentStrategy.CHILD,
) -> Callable:
"""Adds tracing to an async function in a Traceable class.
Args:
func: The async function to trace.
name: Custom span name. Defaults to function name.
attachment_strategy: How to attach this span (CHILD, LINK, NONE).
Returns:
Wrapped async function with tracing.
Raises:
RuntimeError: If used in a class not inheriting from Traceable.
ValueError: If used on a non-async function.
"""
if not OPENTELEMETRY_AVAILABLE:
# Just return the original function or a simple decorator
def decorator(f):
return f
return decorator if func is None else func
if func is not None:
return __traced_decorator(func, name=name, attachment_strategy=attachment_strategy)
else:
return functools.partial(
__traced_decorator, name=name, attachment_strategy=attachment_strategy
)
def traceable(cls: C) -> C:
"""Makes a class traceable for OpenTelemetry.
Creates a new class that inherits from both the original class
and Traceable, enabling tracing for class methods.
Args:
cls: The class to make traceable.
Returns:
A new class with tracing capabilities.
"""
if not OPENTELEMETRY_AVAILABLE:
return cls
@functools.wraps(cls, updated=())
class TracedClass(cls, Traceable):
def __init__(self, *args, **kwargs):
cls.__init__(self, *args, **kwargs)
if hasattr(self, "name"):
Traceable.__init__(self, self.name)
else:
Traceable.__init__(self, cls.__name__)
return TracedClass

99
src/pipecat/utils/tracing/conversation_context_provider.py Normal file
View File

@@ -0,0 +1,99 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import uuid
from typing import Optional
from pipecat.utils.tracing.setup import is_tracing_available
if is_tracing_available():
from opentelemetry.context import Context
from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context
class ConversationContextProvider:
"""Provides access to the current conversation's tracing context.
This is a singleton that can be used to get the current conversation's
span context to create child spans (like turns).
"""
_instance = None
_current_conversation_context: Optional[Context] = None
_conversation_id: Optional[str] = None
@classmethod
def get_instance(cls):
"""Get the singleton instance."""
if cls._instance is None:
cls._instance = ConversationContextProvider()
return cls._instance
def set_current_conversation_context(
self, span_context: Optional[SpanContext], conversation_id: Optional[str] = None
):
"""Set the current conversation context.
Args:
span_context: The span context for the current conversation or None to clear it.
conversation_id: Optional ID for the conversation.
"""
if not is_tracing_available():
return
self._conversation_id = conversation_id
if span_context:
# Create a non-recording span from the span context
non_recording_span = NonRecordingSpan(span_context)
self._current_conversation_context = set_span_in_context(non_recording_span)
else:
self._current_conversation_context = None
def get_current_conversation_context(self) -> Optional[Context]:
"""Get the OpenTelemetry context for the current conversation.
Returns:
The current conversation context or None if not available.
"""
return self._current_conversation_context
def get_conversation_id(self) -> Optional[str]:
"""Get the ID for the current conversation.
Returns:
The current conversation ID or None if not available.
"""
return self._conversation_id
def generate_conversation_id(self) -> str:
"""Generate a new conversation ID.
Returns:
A new randomly generated UUID string.
"""
return str(uuid.uuid4())
# Create a simple helper function to get the current conversation context
def get_current_conversation_context() -> Optional[Context]:
"""Get the OpenTelemetry context for the current conversation.
Returns:
The current conversation context or None if not available.
"""
provider = ConversationContextProvider.get_instance()
return provider.get_current_conversation_context()
def get_conversation_id() -> Optional[str]:
"""Get the ID for the current conversation.
Returns:
The current conversation ID or None if not available.
"""
provider = ConversationContextProvider.get_instance()
return provider.get_conversation_id()

195
src/pipecat/utils/tracing/service_attributes.py Normal file
View File

@@ -0,0 +1,195 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Functions for adding attributes to OpenTelemetry spans."""
from typing import Any, Dict, Optional
from opentelemetry.trace import Span
def add_tts_span_attributes(
span: Span,
service_name: str,
model: str,
voice_id: str,
text: Optional[str] = None,
settings: Optional[Dict[str, Any]] = None,
character_count: Optional[int] = None,
operation_name: str = "tts",
ttfb_ms: Optional[float] = None,
**kwargs,
) -> None:
"""Add TTS-specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the TTS service (e.g., "cartesia")
model: Model name/identifier
voice_id: Voice identifier
text: The text being synthesized
settings: Service configuration settings
character_count: Number of characters in the text
operation_name: Name of the operation (default: "tts")
ttfb_ms: Time to first byte in milliseconds
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("voice_id", voice_id)
span.set_attribute("operation", operation_name)
# Add optional attributes
if text:
span.set_attribute("text", text)
if character_count is not None:
span.set_attribute("metrics.tts.character_count", character_count)
if ttfb_ms is not None:
span.set_attribute("metrics.ttfb_ms", ttfb_ms)
# Add settings if provided
if settings:
for key, value in settings.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"settings.{key}", value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_stt_span_attributes(
span: Span,
service_name: str,
model: str,
transcript: Optional[str] = None,
is_final: Optional[bool] = None,
language: Optional[str] = None,
settings: Optional[Dict[str, Any]] = None,
vad_enabled: bool = False,
ttfb_ms: Optional[float] = None,
**kwargs,
) -> None:
"""Add STT-specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the STT service (e.g., "deepgram")
model: Model name/identifier
transcript: The transcribed text
is_final: Whether this is a final transcript
language: Detected or configured language
settings: Service configuration settings
vad_enabled: Whether voice activity detection is enabled
ttfb_ms: Time to first byte in milliseconds
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("vad_enabled", vad_enabled)
# Add optional attributes
if transcript:
span.set_attribute("transcript", transcript)
if is_final is not None:
span.set_attribute("is_final", is_final)
if language:
span.set_attribute("language", language)
if ttfb_ms is not None:
span.set_attribute("metrics.ttfb_ms", ttfb_ms)
# Add settings if provided
if settings:
for key, value in settings.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"settings.{key}", value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_llm_span_attributes(
span: Span,
service_name: str,
model: str,
stream: bool = True,
messages: Optional[str] = None,
tools: Optional[str] = None,
tool_count: Optional[int] = None,
tool_choice: Optional[str] = None,
system: Optional[str] = None,
parameters: Optional[Dict[str, Any]] = None,
extra_parameters: Optional[Dict[str, Any]] = None,
ttfb_ms: Optional[float] = None,
**kwargs,
) -> None:
"""Add LLM-specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the LLM service (e.g., "openai")
model: Model name/identifier
stream: Whether streaming is enabled
messages: JSON-serialized messages
tools: JSON-serialized tools configuration
tool_count: Number of tools available
tool_choice: Tool selection configuration
system: System message
parameters: Service parameters
extra_parameters: Additional parameters
ttfb_ms: Time to first byte in milliseconds
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("stream", stream)
# Add optional attributes
if messages:
span.set_attribute("messages", messages)
if tools:
span.set_attribute("tools", tools)
if tool_count is not None:
span.set_attribute("tool_count", tool_count)
if tool_choice:
span.set_attribute("tool_choice", tool_choice)
if system:
span.set_attribute("system", system)
if ttfb_ms is not None:
span.set_attribute("metrics.ttfb_ms", ttfb_ms)
# Add parameters if provided
if parameters:
for key, value in parameters.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"param.{key}", value)
# Add extra parameters if provided
if extra_parameters:
for key, value in extra_parameters.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"extra.{key}", value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)

446
src/pipecat/utils/tracing/service_decorators.py Normal file
View File

@@ -0,0 +1,446 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Service-specific OpenTelemetry tracing decorators for Pipecat.
This module provides specialized decorators that automatically capture
rich information about service execution including configuration,
parameters, and performance metrics.
"""
import contextlib
import functools
import inspect
import json
import logging
from typing import Callable, Optional, TypeVar
from opentelemetry import context as context_api
from opentelemetry import trace
from pipecat.utils.tracing.service_attributes import (
add_llm_span_attributes,
add_stt_span_attributes,
add_tts_span_attributes,
)
from pipecat.utils.tracing.setup import OPENTELEMETRY_AVAILABLE, is_tracing_available
from pipecat.utils.tracing.turn_context_provider import get_current_turn_context
T = TypeVar("T")
R = TypeVar("R")
# Internal helper functions
def _noop_decorator(func):
"""No-op fallback decorator when tracing is unavailable."""
return func
def _get_parent_service_context(self):
"""Get the parent service span context (internal use only).
This looks for the service span that was created when the service was initialized.
Args:
self: The service instance
Returns:
Context or None: The parent service context, or None if unavailable
"""
if not is_tracing_available():
return None
# The parent span was created when Traceable was initialized and stored as self._span
if hasattr(self, "_span") and self._span:
return trace.set_span_in_context(self._span)
# If we can't find a stored span, default to current context
return context_api.get_current()
def _get_service_name(self, service_prefix: str) -> str:
"""Generate a default span name using service type and class name.
Args:
self: The service instance.
service_prefix: The service type (e.g., 'llm', 'stt', 'tts').
Returns:
A default span name string like "type_classname" (e.g. llm_openaillmservice).
"""
service_class_name = self.__class__.__name__.lower()
return f"{service_prefix}_{service_class_name}"
def _add_token_usage_to_span(span, token_usage):
"""Add token usage metrics to a span (internal use only).
Args:
span: The span to add token metrics to
token_usage: Dictionary or object containing token usage information
"""
if not is_tracing_available() or not token_usage:
return
if isinstance(token_usage, dict):
if "prompt_tokens" in token_usage:
span.set_attribute("llm.prompt_tokens", token_usage["prompt_tokens"])
if "completion_tokens" in token_usage:
span.set_attribute("llm.completion_tokens", token_usage["completion_tokens"])
else:
# Handle LLMTokenUsage object
span.set_attribute("llm.prompt_tokens", getattr(token_usage, "prompt_tokens", 0))
span.set_attribute("llm.completion_tokens", getattr(token_usage, "completion_tokens", 0))
def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) -> Callable:
"""Traces TTS service methods with TTS-specific attributes.
Automatically captures and records:
- Service name and model information
- Voice ID and settings
- Character count and text content
- Performance metrics like TTFB
Works with both async functions and generators.
Args:
func: The TTS method to trace.
name: Custom span name. Defaults to service type and class name.
Returns:
Wrapped method with TTS-specific tracing.
"""
if not OPENTELEMETRY_AVAILABLE:
return _noop_decorator if func is None else _noop_decorator(func)
def decorator(f):
is_async_generator = inspect.isasyncgenfunction(f)
@contextlib.asynccontextmanager
async def tracing_context(self, text):
"""Async context manager for TTS tracing."""
if not is_tracing_available():
yield None
return
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "tts")
# Get parent context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(span_name, context=parent_context) as span:
try:
add_tts_span_attributes(
span=span,
service_name=service_class_name,
model=getattr(self, "model_name", "unknown"),
voice_id=getattr(self, "_voice_id", "unknown"),
text=text,
settings=getattr(self, "_settings", {}),
character_count=len(text),
operation_name="tts",
cartesia_version=getattr(self, "_cartesia_version", None),
context_id=getattr(self, "_context_id", None),
)
yield span
except Exception as e:
logging.warning(f"Error in TTS tracing: {e}")
raise
finally:
# Update TTFB metric at the end
ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None)
if ttfb_ms is not None:
span.set_attribute("metrics.ttfb_ms", ttfb_ms)
if is_async_generator:
@functools.wraps(f)
async def gen_wrapper(self, text, *args, **kwargs):
try:
if not is_tracing_available():
async for item in f(self, text, *args, **kwargs):
yield item
return
async with tracing_context(self, text):
async for item in f(self, text, *args, **kwargs):
yield item
except Exception as e:
logging.error(f"Error in TTS tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
async for item in f(self, text, *args, **kwargs):
yield item
return gen_wrapper
else:
@functools.wraps(f)
async def wrapper(self, text, *args, **kwargs):
try:
if not is_tracing_available():
return await f(self, text, *args, **kwargs)
async with tracing_context(self, text):
return await f(self, text, *args, **kwargs)
except Exception as e:
logging.error(f"Error in TTS tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await f(self, text, *args, **kwargs)
return wrapper
if func is not None:
return decorator(func)
return decorator
def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) -> Callable:
"""Traces STT service methods with transcription attributes.
Automatically captures and records:
- Service name and model information
- Transcription text and final status
- Language information
- Performance metrics like TTFB
Args:
func: The STT method to trace.
name: Custom span name. Defaults to function name.
Returns:
Wrapped method with STT-specific tracing.
"""
if not OPENTELEMETRY_AVAILABLE:
return _noop_decorator if func is None else _noop_decorator(func)
def decorator(f):
@functools.wraps(f)
async def wrapper(self, transcript, is_final, language=None):
try:
if not is_tracing_available():
return await f(self, transcript, is_final, language)
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "stt")
# Get the turn context first, then fall back to service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# Get TTFB metric if available
ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None)
# Use settings from the service if available
settings = getattr(self, "_settings", {})
add_stt_span_attributes(
span=current_span,
service_name=service_class_name,
model=getattr(self, "model_name", settings.get("model", "unknown")),
transcript=transcript,
is_final=is_final,
language=str(language) if language else None,
vad_enabled=getattr(self, "vad_enabled", False),
settings=settings,
ttfb_ms=ttfb_ms,
)
# Call the original function
return await f(self, transcript, is_final, language)
except Exception as e:
# Log any exception but don't disrupt the main flow
logging.warning(f"Error in STT transcription tracing: {e}")
raise
except Exception as e:
logging.error(f"Error in STT tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await f(self, transcript, is_final, language)
return wrapper
if func is not None:
return decorator(func)
return decorator
def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -> Callable:
"""Traces LLM service methods with LLM-specific attributes.
Automatically captures and records:
- Service name and model information
- Context content and messages
- Tool configurations
- Token usage metrics
- Performance metrics like TTFB
Args:
func: The LLM method to trace.
name: Custom span name. Defaults to service type and class name.
Returns:
Wrapped method with LLM-specific tracing.
"""
if not OPENTELEMETRY_AVAILABLE:
return _noop_decorator if func is None else _noop_decorator(func)
def decorator(f):
@functools.wraps(f)
async def wrapper(self, context, *args, **kwargs):
try:
if not is_tracing_available():
return await f(self, context, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "llm")
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# For token usage monitoring
original_start_llm_usage_metrics = None
if hasattr(self, "start_llm_usage_metrics"):
original_start_llm_usage_metrics = self.start_llm_usage_metrics
# Override the method to capture token usage
@functools.wraps(original_start_llm_usage_metrics)
async def wrapped_start_llm_usage_metrics(tokens):
# Call the original method
await original_start_llm_usage_metrics(tokens)
# Add token usage to the current span
_add_token_usage_to_span(current_span, tokens)
# Replace the method temporarily
self.start_llm_usage_metrics = wrapped_start_llm_usage_metrics
try:
# Detect if we're using Google's service
is_google_service = "google" in service_class_name.lower()
# Try to get messages based on service type
messages = None
serialized_messages = None
# TODO: Revisit once we unify the messages across services
if is_google_service:
# Handle Google service specifically
if hasattr(context, "get_messages_for_logging"):
messages = context.get_messages_for_logging()
else:
# Handle other services like OpenAI
if hasattr(context, "get_messages"):
messages = context.get_messages()
elif hasattr(context, "messages"):
messages = context.messages
# Serialize messages if available
if messages:
try:
serialized_messages = json.dumps(messages)
except Exception as e:
serialized_messages = f"Error serializing messages: {str(e)}"
# Get tools, system message, etc. based on the service type
tools = getattr(context, "tools", None)
serialized_tools = None
tool_count = 0
if tools:
try:
serialized_tools = json.dumps(tools)
tool_count = len(tools) if isinstance(tools, list) else 1
except Exception as e:
serialized_tools = f"Error serializing tools: {str(e)}"
# Handle system message for different services
system_message = None
if hasattr(context, "system"):
system_message = context.system
elif hasattr(context, "system_message"):
system_message = context.system_message
elif hasattr(self, "_system_instruction"):
system_message = self._system_instruction
# Get settings from the service
params = {}
if hasattr(self, "_settings"):
for key, value in self._settings.items():
if key == "extra":
continue
# Add value directly if it's a basic type
if isinstance(value, (int, float, bool, str)):
params[key] = value
elif value is None or (
hasattr(value, "__name__") and value.__name__ == "NOT_GIVEN"
):
params[key] = "NOT_GIVEN"
# Add all available attributes to the span
attribute_kwargs = {
"service_name": service_class_name,
"model": getattr(self, "model_name", "unknown"),
"stream": True, # Most LLM services use streaming
"parameters": params,
}
# Add optional attributes only if they exist
if serialized_messages:
attribute_kwargs["messages"] = serialized_messages
if serialized_tools:
attribute_kwargs["tools"] = serialized_tools
attribute_kwargs["tool_count"] = tool_count
if system_message:
attribute_kwargs["system"] = system_message
# Add all gathered attributes to the span
add_llm_span_attributes(span=current_span, **attribute_kwargs)
except Exception as e:
logging.warning(f"Error adding initial LLM attributes: {e}")
# Call the original function
return await f(self, context, *args, **kwargs)
finally:
# Restore the original methods if we overrode them
if (
"original_start_llm_usage_metrics" in locals()
and original_start_llm_usage_metrics
):
self.start_llm_usage_metrics = original_start_llm_usage_metrics
# Update TTFB metric
ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None)
if ttfb_ms is not None:
current_span.set_attribute("metrics.ttfb_ms", ttfb_ms)
except Exception as e:
logging.error(f"Error in LLM tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await f(self, context, *args, **kwargs)
return wrapper
if func is not None:
return decorator(func)
return decorator

84
src/pipecat/utils/tracing/setup.py Normal file
View File

@@ -0,0 +1,84 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Core OpenTelemetry tracing utilities and setup for Pipecat."""
import os
# Check if OpenTelemetry is available
try:
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
OPENTELEMETRY_AVAILABLE = True
except ImportError:
OPENTELEMETRY_AVAILABLE = False
def is_tracing_available() -> bool:
"""Returns True if OpenTelemetry tracing is available and configured.
Returns:
bool: True if tracing is available, False otherwise.
"""
return OPENTELEMETRY_AVAILABLE
def setup_tracing(
service_name: str = "pipecat",
exporter=None, # User-provided exporter
console_export: bool = False,
) -> bool:
"""Set up OpenTelemetry tracing with a user-provided exporter.
Args:
service_name: The name of the service for traces
exporter: A pre-configured OpenTelemetry span exporter instance.
If None, only console export will be available if enabled.
console_export: Whether to also export traces to console (useful for debugging)
Returns:
bool: True if setup was successful, False otherwise
Example:
# With OTLP exporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
setup_tracing("my-service", exporter=exporter)
"""
if not OPENTELEMETRY_AVAILABLE:
return False
try:
# Create a resource with service info
resource = Resource.create(
{
"service.name": service_name,
"service.instance.id": os.getenv("HOSTNAME", "unknown"),
"deployment.environment": os.getenv("ENVIRONMENT", "development"),
}
)
# Set up the tracer provider with the resource
tracer_provider = TracerProvider(resource=resource)
trace.set_tracer_provider(tracer_provider)
# Add console exporter if requested (good for debugging)
if console_export:
console_exporter = ConsoleSpanExporter()
tracer_provider.add_span_processor(BatchSpanProcessor(console_exporter))
# Add user-provided exporter if available
if exporter:
tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
return True
except Exception as e:
print(f"Error setting up tracing: {e}")
return False

66
src/pipecat/utils/tracing/turn_context_provider.py Normal file
View File

@@ -0,0 +1,66 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Optional
from pipecat.utils.tracing.setup import is_tracing_available
if is_tracing_available():
from opentelemetry.context import Context
from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context
class TurnContextProvider:
"""Provides access to the current turn's tracing context.
This is a singleton that services can use to get the current turn's
span context to create child spans.
"""
_instance = None
_current_turn_context: Optional[Context] = None
@classmethod
def get_instance(cls):
"""Get the singleton instance."""
if cls._instance is None:
cls._instance = TurnContextProvider()
return cls._instance
def set_current_turn_context(self, span_context: Optional[SpanContext]):
"""Set the current turn context.
Args:
span_context: The span context for the current turn or None to clear it.
"""
if not is_tracing_available():
return
if span_context:
# Create a non-recording span from the span context
non_recording_span = NonRecordingSpan(span_context)
self._current_turn_context = set_span_in_context(non_recording_span)
else:
self._current_turn_context = None
def get_current_turn_context(self) -> Optional[Context]:
"""Get the OpenTelemetry context for the current turn.
Returns:
The current turn context or None if not available.
"""
return self._current_turn_context
# Create a simple helper function to get the current turn context
def get_current_turn_context() -> Optional[Context]:
"""Get the OpenTelemetry context for the current turn.
Returns:
The current turn context or None if not available.
"""
provider = TurnContextProvider.get_instance()
return provider.get_current_turn_context()

201
src/pipecat/utils/tracing/turn_trace_observer.py Normal file
View File

@@ -0,0 +1,201 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Dict, Optional
from loguru import logger
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_context_provider import TurnContextProvider
if is_tracing_available():
from opentelemetry import trace
from opentelemetry.trace import Span, SpanContext
class TurnTraceObserver(BaseObserver):
"""Observer that creates trace spans for each conversation turn.
This observer uses TurnTrackingObserver to track turns and creates
OpenTelemetry spans for each turn. Service spans (STT, LLM, TTS)
become children of the turn spans.
If conversation tracing is enabled, turns become children of a
conversation span that encapsulates the entire session.
"""
def __init__(self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None):
super().__init__()
self._turn_tracker = turn_tracker
self._current_span: Optional[Span] = None
self._current_turn_number: int = 0
self._trace_context_map: Dict[int, SpanContext] = {}
self._tracer = trace.get_tracer("pipecat.turn") if is_tracing_available() else None
# Conversation tracking properties
self._conversation_span: Optional[Span] = None
self._conversation_id = conversation_id
if turn_tracker:
@turn_tracker.event_handler("on_turn_started")
async def on_turn_started(tracker, turn_number):
await self._handle_turn_started(turn_number)
@turn_tracker.event_handler("on_turn_ended")
async def on_turn_ended(tracker, turn_number, duration, was_interrupted):
await self._handle_turn_ended(turn_number, duration, was_interrupted)
async def on_push_frame(self, data: FramePushed):
"""Process a frame without modifying it.
This observer doesn't need to process individual frames as it
relies on turn start/end events from the turn tracker.
"""
pass
def start_conversation_tracing(self, conversation_id: Optional[str] = None):
"""Start a new conversation span.
Args:
conversation_id: Optional custom ID for the conversation. If None, a UUID will be generated.
"""
if not is_tracing_available() or not self._tracer:
return
# Generate a conversation ID if not provided
context_provider = ConversationContextProvider.get_instance()
if conversation_id is None:
conversation_id = context_provider.generate_conversation_id()
logger.debug(f"Generated new conversation ID: {conversation_id}")
self._conversation_id = conversation_id
# Create a new span for this conversation
self._conversation_span = self._tracer.start_span(f"conversation-{conversation_id}")
# Set span attributes
self._conversation_span.set_attribute("conversation.id", conversation_id)
self._conversation_span.set_attribute("conversation.type", "voice")
# Update the conversation context provider
context_provider.set_current_conversation_context(
self._conversation_span.get_span_context(), conversation_id
)
logger.debug(f"Started tracing for Conversation {conversation_id}")
def end_conversation_tracing(self):
"""End the current conversation span and ensure the last turn is closed."""
if not is_tracing_available():
return
# First, ensure any active turn is closed properly
if self._current_span:
# If we have an active turn span, end it with a standard duration
logger.debug(f"Ending Turn {self._current_turn_number} due to conversation end")
self._current_span.set_attribute("turn.was_interrupted", True)
self._current_span.set_attribute("turn.ended_by_conversation_end", True)
self._current_span.end()
self._current_span = None
# Clear the turn context provider
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(None)
# Now end the conversation span if it exists
if self._conversation_span:
# End the span
self._conversation_span.end()
self._conversation_span = None
# Clear the context provider
context_provider = ConversationContextProvider.get_instance()
context_provider.set_current_conversation_context(None)
logger.debug(f"Ended tracing for Conversation {self._conversation_id}")
self._conversation_id = None
async def _handle_turn_started(self, turn_number: int):
"""Handle a turn start event by creating a new span."""
if not is_tracing_available() or not self._tracer:
return
# If this is the first turn and no conversation span exists yet,
# start the conversation tracing (will generate ID if needed)
if turn_number == 1 and not self._conversation_span:
self.start_conversation_tracing(self._conversation_id)
# Get the parent context - conversation if available, otherwise use root context
parent_context = None
if self._conversation_span:
context_provider = ConversationContextProvider.get_instance()
parent_context = context_provider.get_current_conversation_context()
# Create a new span for this turn
self._current_span = self._tracer.start_span(f"turn-{turn_number}", context=parent_context)
self._current_turn_number = turn_number
# Set span attributes
self._current_span.set_attribute("turn.number", turn_number)
self._current_span.set_attribute("turn.type", "conversation")
# Add conversation ID attribute if available
if self._conversation_id:
self._current_span.set_attribute("conversation.id", self._conversation_id)
# Store the span context so services can become children of this span
self._trace_context_map[turn_number] = self._current_span.get_span_context()
# Update the context provider so services can access this span
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(self._current_span.get_span_context())
logger.debug(f"Started tracing for Turn {turn_number}")
async def _handle_turn_ended(self, turn_number: int, duration: float, was_interrupted: bool):
"""Handle a turn end event by ending the current span."""
if not is_tracing_available() or not self._current_span:
return
# Only end the span if it matches the current turn
if turn_number == self._current_turn_number:
# Set additional attributes
self._current_span.set_attribute("turn.duration_seconds", duration)
self._current_span.set_attribute("turn.was_interrupted", was_interrupted)
# End the span
self._current_span.end()
self._current_span = None
# Clear the context provider
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(None)
logger.debug(f"Ended tracing for Turn {turn_number}")
def get_current_turn_context(self) -> Optional[SpanContext]:
"""Get the span context for the current turn.
This can be used by services to create child spans.
"""
if not is_tracing_available() or not self._current_span:
return None
return self._current_span.get_span_context()
def get_turn_context(self, turn_number: int) -> Optional[SpanContext]:
"""Get the span context for a specific turn.
This can be used by services to create child spans.
"""
if not is_tracing_available():
return None
return self._trace_context_map.get(turn_number)

2
test-requirements.txt
View File

@@ -1 +1 @@
-e ".[anthropic,aws,google,langchain]"
-e ".[anthropic,aws,google,langchain,tracing]"

261
tests/test_turn_tracking_observer.py Normal file
View File

@@ -0,0 +1,261 @@
#
# Copyright (c) 2024-2025 Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.processors.filters.identity_filter import IdentityFilter
from pipecat.tests.utils import SleepFrame, run_test
class TestTurnTrackingObserver(unittest.IsolatedAsyncioTestCase):
"""Tests for TurnTrackingObserver."""
async def test_normal_conversation_flow(self):
"""Test a normal conversation with two complete turns."""
# Create observer with a short timeout
turn_observer = TurnTrackingObserver(turn_end_timeout_secs=0.2)
# Create identity filter (passes all frames through)
processor = IdentityFilter()
# Record start/end events with turn numbers
turn_events = []
@turn_observer.event_handler("on_turn_started")
async def on_turn_started(observer, turn_number):
turn_events.append(f"Turn {turn_number} started")
@turn_observer.event_handler("on_turn_ended")
async def on_turn_ended(observer, turn_number, duration, was_interrupted):
turn_events.append(f"Turn {turn_number} ended (interrupted: {was_interrupted})")
frames_to_send = [
# Turn 1
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.05), # < 0.2 seconds turn_end_timeout
# Turn 2
UserStartedSpeakingFrame(), # New turn starts
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
# Add a sleep frame to allow turn timeout to occur
SleepFrame(sleep=0.4), # > 0.2 seconds turn_end_timeout
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[turn_observer],
)
# Verify turn events
expected_events = [
"Turn 1 started",
"Turn 1 ended (interrupted: False)",
"Turn 2 started",
"Turn 2 ended (interrupted: False)",
]
self.assertEqual(turn_events, expected_events)
self.assertEqual(turn_observer._turn_count, 2)
async def test_user_speaks_twice_before_bot(self):
"""Test when user speaks twice before bot responds, should be same turn."""
# Create observer with a short timeout
turn_observer = TurnTrackingObserver(turn_end_timeout_secs=0.2)
# Create identity filter (passes all frames through)
processor = IdentityFilter()
# Record start/end events with turn numbers
turn_events = []
@turn_observer.event_handler("on_turn_started")
async def on_turn_started(observer, turn_number):
turn_events.append(f"Turn {turn_number} started")
@turn_observer.event_handler("on_turn_ended")
async def on_turn_ended(observer, turn_number, duration, was_interrupted):
turn_events.append(f"Turn {turn_number} ended (interrupted: {was_interrupted})")
frames_to_send = [
# Turn 1 - User speaks twice before bot responds
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
UserStartedSpeakingFrame(), # Second user speaking event should not start a new turn
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
# Turn 2
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
# Add a sleep frame to allow turn timeout to occur
SleepFrame(sleep=0.4), # > 0.2 seconds turn_end_timeout
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[turn_observer],
)
# Verify turn events - should only see two turns despite user speaking twice
expected_events = [
"Turn 1 started",
"Turn 1 ended (interrupted: False)",
"Turn 2 started",
"Turn 2 ended (interrupted: False)",
]
self.assertEqual(turn_events, expected_events)
self.assertEqual(turn_observer._turn_count, 2)
async def test_user_interrupts_bot(self):
"""Test when user interrupts bot speaking, should end current turn and start new one."""
# Create observer with a short timeout
turn_observer = TurnTrackingObserver(turn_end_timeout_secs=0.2)
# Create identity filter (passes all frames through)
processor = IdentityFilter()
# Record start/end events with turn numbers
turn_events = []
@turn_observer.event_handler("on_turn_started")
async def on_turn_started(observer, turn_number):
turn_events.append(f"Turn {turn_number} started")
@turn_observer.event_handler("on_turn_ended")
async def on_turn_ended(observer, turn_number, duration, was_interrupted):
turn_events.append(f"Turn {turn_number} ended (interrupted: {was_interrupted})")
frames_to_send = [
# Turn 1
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
# Interruption here - user starts speaking while bot is still speaking
UserStartedSpeakingFrame(), # This should end Turn 1 and start Turn 2
SleepFrame(sleep=0.4), # > 0.2 seconds turn_end_timeout
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
UserStartedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[turn_observer],
)
# Verify turn events - should see Turn 1 interrupted
expected_events = [
"Turn 1 started",
"Turn 1 ended (interrupted: True)", # First turn was interrupted
"Turn 2 started", # New turn started after interruption
]
self.assertEqual(turn_events, expected_events)
self.assertEqual(turn_observer._turn_count, 2)
async def test_bot_starts_stops_multiple_times(self):
"""Test that multiple bot start/stop frames in the same turn work correctly."""
# Create observer with a short timeout
turn_observer = TurnTrackingObserver(turn_end_timeout_secs=0.2)
# Create identity filter (passes all frames through)
processor = IdentityFilter()
turn_events = []
@turn_observer.event_handler("on_turn_started")
async def on_turn_started(observer, turn_number):
turn_events.append(f"Turn {turn_number} started")
@turn_observer.event_handler("on_turn_ended")
async def on_turn_ended(observer, turn_number, duration, was_interrupted):
turn_events.append(f"Turn {turn_number} ended (interrupted: {was_interrupted})")
frames_to_send = [
# Start turn with user speaking
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
# Bot speaks, stops, speaks again (simulating HTTP TTS or function calls)
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
BotStartedSpeakingFrame(), # Bot speaks again, should not end turn
BotStoppedSpeakingFrame(),
# Add a sleep frame to allow turn timeout to occur
SleepFrame(sleep=0.4), # > 0.2 seconds turn_end_timeout
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[turn_observer],
)
# Should only be one turn with a normal end
expected_events = [
"Turn 1 started",
"Turn 1 ended (interrupted: False)",
]
self.assertEqual(turn_events, expected_events)
self.assertEqual(turn_observer._turn_count, 1)
if __name__ == "__main__":
unittest.main()