diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bb74c358..60081503c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py new file mode 100644 index 000000000..0e9c7acec --- /dev/null +++ b/examples/foundational/29-turn-tracking-observer.py @@ -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() diff --git a/examples/open-telemetry-tracing/README.md b/examples/open-telemetry-tracing/README.md new file mode 100644 index 000000000..0635450f4 --- /dev/null +++ b/examples/open-telemetry-tracing/README.md @@ -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/) diff --git a/examples/open-telemetry-tracing/bot.py b/examples/open-telemetry-tracing/bot.py new file mode 100644 index 000000000..0b44c3865 --- /dev/null +++ b/examples/open-telemetry-tracing/bot.py @@ -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() diff --git a/examples/open-telemetry-tracing/env.example b/examples/open-telemetry-tracing/env.example new file mode 100644 index 000000000..68b2ef6b0 --- /dev/null +++ b/examples/open-telemetry-tracing/env.example @@ -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 \ No newline at end of file diff --git a/examples/open-telemetry-tracing/requirements.txt b/examples/open-telemetry-tracing/requirements.txt new file mode 100644 index 000000000..7f5abb16d --- /dev/null +++ b/examples/open-telemetry-tracing/requirements.txt @@ -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 \ No newline at end of file diff --git a/examples/open-telemetry-tracing/run.py b/examples/open-telemetry-tracing/run.py new file mode 100644 index 000000000..e7012c9e9 --- /dev/null +++ b/examples/open-telemetry-tracing/run.py @@ -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() diff --git a/pyproject.toml b/pyproject.toml index b37314496..78bd78773 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" ] diff --git a/src/pipecat/observers/loggers/debug_log_observer.py b/src/pipecat/observers/loggers/debug_log_observer.py index 575a31683..18048890e 100644 --- a/src/pipecat/observers/loggers/debug_log_observer.py +++ b/src/pipecat/observers/loggers/debug_log_observer.py @@ -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, + } + ), + ], ``` """ diff --git a/src/pipecat/observers/turn_tracking_observer.py b/src/pipecat/observers/turn_tracking_observer.py new file mode 100644 index 000000000..99abdaff6 --- /dev/null +++ b/src/pipecat/observers/turn_tracking_observer.py @@ -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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 679bb02e4..b45999784 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -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: diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py index 4f592bce3..c9be3e3de 100644 --- a/src/pipecat/processors/metrics/frame_processor_metrics.py +++ b/src/pipecat/processors/metrics/frame_processor_metrics.py @@ -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]) diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index eba6a5041..43fe5ab6f 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -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 diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index e6705a4a7..2153d969e 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -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 diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 921d3c790..f1a24bb18 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -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 diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index a02625f81..5016d5e78 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -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( diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index 40d746514..cba2acc5f 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -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) diff --git a/src/pipecat/services/azure/stt.py b/src/pipecat/services/azure/stt.py index 95f3dcae1..9c9b386eb 100644 --- a/src/pipecat/services/azure/stt.py +++ b/src/pipecat/services/azure/stt.py @@ -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()) diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index cf50c9fba..9dde55496 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -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}]") diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 3f7df063b..3a7d6c696 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -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}]") diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 7b5209e0b..07cc6a9d2 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -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) ) diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index c9a026aa1..a684a340e 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -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}]") diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index b5859e26c..5e5a97e75 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -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. diff --git a/src/pipecat/services/fal/stt.py b/src/pipecat/services/fal/stt.py index 477d50f3b..950d74d06 100644 --- a/src/pipecat/services/fal/stt.py +++ b/src/pipecat/services/fal/stt.py @@ -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"]) diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index abf609177..39c85281a 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -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: diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index f27664a25..8208d01e2 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -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( diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 50f1ad680..e5fcf56db 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -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()) diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index cf0385857..d7373aa71 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -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 diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 6ecf6d762..487aec7a6 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -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}]") diff --git a/src/pipecat/services/groq/tts.py b/src/pipecat/services/groq/tts.py index f4f0f308b..828edf72b 100644 --- a/src/pipecat/services/groq/tts.py +++ b/src/pipecat/services/groq/tts.py @@ -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 diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index b9e0bc79f..4a3be761c 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -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}]") diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index 1819482b8..de12c8231 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -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. diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 1aba1b159..e24ad1d21 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -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 = [] diff --git a/src/pipecat/services/openai/tts.py b/src/pipecat/services/openai/tts.py index 634b0a22e..61fb3e77c 100644 --- a/src/pipecat/services/openai/tts.py +++ b/src/pipecat/services/openai/tts.py @@ -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: diff --git a/src/pipecat/services/piper/tts.py b/src/pipecat/services/piper/tts.py index 3b5d0fa06..7686196de 100644 --- a/src/pipecat/services/piper/tts.py +++ b/src/pipecat/services/piper/tts.py @@ -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. diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py index eb1b8d3f1..865b267c9 100644 --- a/src/pipecat/services/playht/tts.py +++ b/src/pipecat/services/playht/tts.py @@ -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}]") diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index a26a54805..09a633456 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -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}]") diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index 2b263f03e..b70d63e4e 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -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") diff --git a/src/pipecat/services/riva/tts.py b/src/pipecat/services/riva/tts.py index 4fc8b8489..a61155685 100644 --- a/src/pipecat/services/riva/tts.py +++ b/src/pipecat/services/riva/tts.py @@ -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=, 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): diff --git a/src/pipecat/services/whisper/base_stt.py b/src/pipecat/services/whisper/base_stt.py index 95d14bbe5..018dae85a 100644 --- a/src/pipecat/services/whisper/base_stt.py +++ b/src/pipecat/services/whisper/base_stt.py @@ -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: diff --git a/src/pipecat/services/whisper/stt.py b/src/pipecat/services/whisper/stt.py index 4026dbea1..d6920ed6c 100644 --- a/src/pipecat/services/whisper/stt.py +++ b/src/pipecat/services/whisper/stt.py @@ -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"]) diff --git a/src/pipecat/services/xtts/tts.py b/src/pipecat/services/xtts/tts.py index 5e08732a9..18528f0ea 100644 --- a/src/pipecat/services/xtts/tts.py +++ b/src/pipecat/services/xtts/tts.py @@ -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}]") diff --git a/src/pipecat/utils/tracing/__init__.py b/src/pipecat/utils/tracing/__init__.py new file mode 100644 index 000000000..c9cf04597 --- /dev/null +++ b/src/pipecat/utils/tracing/__init__.py @@ -0,0 +1,7 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""OpenTelemetry tracing utilities for Pipecat.""" diff --git a/src/pipecat/utils/tracing/class_decorators.py b/src/pipecat/utils/tracing/class_decorators.py new file mode 100644 index 000000000..d84605839 --- /dev/null +++ b/src/pipecat/utils/tracing/class_decorators.py @@ -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 diff --git a/src/pipecat/utils/tracing/conversation_context_provider.py b/src/pipecat/utils/tracing/conversation_context_provider.py new file mode 100644 index 000000000..d4a7e8205 --- /dev/null +++ b/src/pipecat/utils/tracing/conversation_context_provider.py @@ -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() diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py new file mode 100644 index 000000000..cebabc4bf --- /dev/null +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -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) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py new file mode 100644 index 000000000..7c9912c65 --- /dev/null +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -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 diff --git a/src/pipecat/utils/tracing/setup.py b/src/pipecat/utils/tracing/setup.py new file mode 100644 index 000000000..ab74530cb --- /dev/null +++ b/src/pipecat/utils/tracing/setup.py @@ -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 diff --git a/src/pipecat/utils/tracing/turn_context_provider.py b/src/pipecat/utils/tracing/turn_context_provider.py new file mode 100644 index 000000000..868e2cf4a --- /dev/null +++ b/src/pipecat/utils/tracing/turn_context_provider.py @@ -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() diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py new file mode 100644 index 000000000..dedf87696 --- /dev/null +++ b/src/pipecat/utils/tracing/turn_trace_observer.py @@ -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) diff --git a/test-requirements.txt b/test-requirements.txt index fec8adf52..1abcabfcf 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1 +1 @@ --e ".[anthropic,aws,google,langchain]" +-e ".[anthropic,aws,google,langchain,tracing]" diff --git a/tests/test_turn_tracking_observer.py b/tests/test_turn_tracking_observer.py new file mode 100644 index 000000000..14cfd472f --- /dev/null +++ b/tests/test_turn_tracking_observer.py @@ -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()