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 = "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: Optional[float] = getattr(getattr(self, "_metrics", None), "ttfb", None)
                    if ttfb is not None:
                        span.set_attribute("metrics.ttfb", ttfb)
Async context manager for TTS tracing.
src/pipecat/utils/tracing/service_decorators.py