Fix traced_tts so metrics.ttfb reflects the real TTFB
Previously @traced_tts scoped the span to the lifetime of run_tts(). For streaming TTS services run_tts() returns as soon as the synthesis request is sent, long before audio chunks arrive, so: - The span duration measured the WebSocket-send time, not synthesis time. - The first synthesis recorded the WS-send duration as metrics.ttfb (via the in-progress fallback in FrameProcessorMetrics.ttfb). - Subsequent syntheses recorded the previous call's TTFB on the current span (off-by-one). The decorator now uses a __set_name__ descriptor to wrap the owning class's setup() at class definition time. setup() installs per-instance patches on create_audio_context, append_to_audio_context, remove_audio_context, on_audio_context_completed, and reset_active_audio_context. These patches own the span lifetime: - create_audio_context: open span, set baseline attributes. - append_to_audio_context: record metrics.ttfb on the first TTSAudioRawFrame (when stop_ttfb_metrics has produced a real value), end span on appended TTSStoppedFrame. - on_audio_context_completed: end span on natural completion (handles services that auto-push TTSStoppedFrame via push_frame, bypassing append_to_audio_context). - remove_audio_context: safety net for explicit removal paths. - reset_active_audio_context: interruption hook (always reached from _handle_interruption); marks the span tts.interrupted=true only when nothing else has closed it. The run_tts wrapper now only attaches per-call attributes (text, metrics.character_count) to the already-open span. No changes required in tts_service.py or in any of the per-service files.
This commit is contained in:
@@ -11,7 +11,6 @@ rich information about service execution including configuration,
|
|||||||
parameters, and performance metrics.
|
parameters, and performance metrics.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import contextlib
|
|
||||||
import functools
|
import functools
|
||||||
import inspect
|
import inspect
|
||||||
import json
|
import json
|
||||||
@@ -24,6 +23,8 @@ if TYPE_CHECKING:
|
|||||||
from opentelemetry import context as context_api
|
from opentelemetry import context as context_api
|
||||||
from opentelemetry import trace
|
from opentelemetry import trace
|
||||||
|
|
||||||
|
from pipecat.frames.frames import MetricsFrame, TTSStoppedFrame
|
||||||
|
from pipecat.metrics.metrics import TTFBMetricsData
|
||||||
from pipecat.processors.aggregators.llm_context import NOT_GIVEN
|
from pipecat.processors.aggregators.llm_context import NOT_GIVEN
|
||||||
from pipecat.utils.tracing.service_attributes import (
|
from pipecat.utils.tracing.service_attributes import (
|
||||||
add_gemini_live_span_attributes,
|
add_gemini_live_span_attributes,
|
||||||
@@ -175,6 +176,13 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call
|
|||||||
- Character count and text content
|
- Character count and text content
|
||||||
- Performance metrics like TTFB
|
- Performance metrics like TTFB
|
||||||
|
|
||||||
|
The span is scoped to the full synthesis operation, from
|
||||||
|
``create_audio_context`` until ``TTSStoppedFrame`` (or
|
||||||
|
``remove_audio_context`` as a safety net), so TTFB and any other
|
||||||
|
runtime-computed metrics land on the correct span even when audio
|
||||||
|
chunks are delivered after ``run_tts`` returns (e.g. WebSocket
|
||||||
|
streaming TTS services).
|
||||||
|
|
||||||
Works with both async functions and generators.
|
Works with both async functions and generators.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@@ -190,99 +198,217 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call
|
|||||||
def decorator(f):
|
def decorator(f):
|
||||||
is_async_generator = inspect.isasyncgenfunction(f)
|
is_async_generator = inspect.isasyncgenfunction(f)
|
||||||
|
|
||||||
@contextlib.asynccontextmanager
|
def end_tts_span(service, context_id, *, interrupted=False):
|
||||||
async def tracing_context(self, text):
|
"""End the TTS span for ``context_id`` if still open. Idempotent."""
|
||||||
"""Async context manager for TTS tracing.
|
entry = service._tts_spans.pop(context_id, None)
|
||||||
|
if not entry:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
span = entry["span"]
|
||||||
|
if interrupted:
|
||||||
|
span.set_attribute("tts.interrupted", True)
|
||||||
|
span.end()
|
||||||
|
except Exception as e:
|
||||||
|
logging.warning(f"Error closing TTS span: {e}")
|
||||||
|
|
||||||
Args:
|
def install_audio_context_patches(service):
|
||||||
self: The TTS service instance.
|
"""Install per-instance wrappers on the audio-context methods.
|
||||||
text: The text being synthesized.
|
|
||||||
|
|
||||||
Yields:
|
The wrappers own the lifetime of the TTS span:
|
||||||
The active span for the TTS operation.
|
|
||||||
|
- ``create_audio_context``: opens the span and records
|
||||||
|
baseline attributes.
|
||||||
|
- ``append_to_audio_context``: ends the span on
|
||||||
|
``TTSStoppedFrame``.
|
||||||
|
- ``push_frame``: records ``metrics.ttfb`` from the
|
||||||
|
canonical ``TTFBMetricsData`` payload of any
|
||||||
|
``MetricsFrame`` pushed by ``stop_ttfb_metrics``. Reading
|
||||||
|
the value from the metrics event (instead of polling
|
||||||
|
``_metrics.ttfb`` when the first audio is queued) avoids
|
||||||
|
the ``ttfb`` property's in-progress fallback, which would
|
||||||
|
otherwise report an under-estimate whenever a context's
|
||||||
|
audio waits behind earlier queued audio before
|
||||||
|
``_handle_audio_context`` actually stops the TTFB
|
||||||
|
measurement.
|
||||||
|
- ``remove_audio_context``: ends any still-open span as a
|
||||||
|
safety net for error and cancellation paths.
|
||||||
|
- ``on_audio_context_completed``: ends the span on natural
|
||||||
|
completion. Needed because services that rely on the
|
||||||
|
base class to auto-push ``TTSStoppedFrame`` (via
|
||||||
|
``push_frame`` in ``_handle_audio_context``) bypass the
|
||||||
|
``append_to_audio_context`` hook entirely.
|
||||||
|
- ``reset_active_audio_context``: ends the currently
|
||||||
|
playing context's span if still open. Always called from
|
||||||
|
``_handle_interruption``, so this is the interruption
|
||||||
|
hook.
|
||||||
|
|
||||||
|
The patches check ``_tracing_enabled`` at invocation time,
|
||||||
|
so they are safe to install regardless of whether tracing
|
||||||
|
is enabled.
|
||||||
"""
|
"""
|
||||||
# Check if tracing is enabled for this service instance
|
if getattr(service, "__tts_tracing_patches_installed__", False):
|
||||||
if not getattr(self, "_tracing_enabled", False):
|
return
|
||||||
yield None
|
service.__tts_tracing_patches_installed__ = True
|
||||||
|
service._tts_spans = {}
|
||||||
|
|
||||||
|
orig_create = service.create_audio_context
|
||||||
|
orig_append = service.append_to_audio_context
|
||||||
|
orig_remove = service.remove_audio_context
|
||||||
|
orig_completed = service.on_audio_context_completed
|
||||||
|
orig_reset_active = service.reset_active_audio_context
|
||||||
|
orig_push_frame = service.push_frame
|
||||||
|
|
||||||
|
async def traced_create_audio_context(context_id):
|
||||||
|
if getattr(service, "_tracing_enabled", False):
|
||||||
|
try:
|
||||||
|
parent = _get_turn_context(service) or _get_parent_service_context(service)
|
||||||
|
tracer = trace.get_tracer("pipecat")
|
||||||
|
span = tracer.start_span("tts", context=parent)
|
||||||
|
service._tts_spans[context_id] = {"span": span, "ttfb_recorded": False}
|
||||||
|
|
||||||
|
settings = getattr(service, "_settings", None)
|
||||||
|
add_tts_span_attributes(
|
||||||
|
span=span,
|
||||||
|
service_name=service.__class__.__name__,
|
||||||
|
model=_get_model_name(service),
|
||||||
|
voice_id=getattr(settings, "voice", "unknown"),
|
||||||
|
settings=settings,
|
||||||
|
operation_name="tts",
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logging.warning(f"Error opening TTS span: {e}")
|
||||||
|
return await orig_create(context_id)
|
||||||
|
|
||||||
|
async def traced_append_to_audio_context(context_id, frame):
|
||||||
|
entry = service._tts_spans.get(context_id)
|
||||||
|
if entry and frame is not None:
|
||||||
|
try:
|
||||||
|
if isinstance(frame, TTSStoppedFrame):
|
||||||
|
entry["span"].end()
|
||||||
|
service._tts_spans.pop(context_id, None)
|
||||||
|
except Exception as e:
|
||||||
|
logging.warning(f"Error updating TTS span: {e}")
|
||||||
|
return await orig_append(context_id, frame)
|
||||||
|
|
||||||
|
async def traced_push_frame(frame, direction=FrameDirection.DOWNSTREAM):
|
||||||
|
await orig_push_frame(frame, direction)
|
||||||
|
if not getattr(service, "_tracing_enabled", False):
|
||||||
|
return
|
||||||
|
if not isinstance(frame, MetricsFrame):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
playing_id = getattr(service, "_playing_context_id", None)
|
||||||
|
if playing_id is None:
|
||||||
|
return
|
||||||
|
entry = service._tts_spans.get(playing_id)
|
||||||
|
if not entry or entry["ttfb_recorded"]:
|
||||||
|
return
|
||||||
|
for data in frame.data:
|
||||||
|
if isinstance(data, TTFBMetricsData):
|
||||||
|
entry["span"].set_attribute("metrics.ttfb", data.value)
|
||||||
|
entry["ttfb_recorded"] = True
|
||||||
|
break
|
||||||
|
except Exception as e:
|
||||||
|
logging.warning(f"Error recording TTS ttfb from MetricsFrame: {e}")
|
||||||
|
|
||||||
|
async def traced_remove_audio_context(context_id):
|
||||||
|
entry = service._tts_spans.pop(context_id, None)
|
||||||
|
if entry:
|
||||||
|
try:
|
||||||
|
entry["span"].end()
|
||||||
|
except Exception as e:
|
||||||
|
logging.warning(f"Error closing TTS span: {e}")
|
||||||
|
return await orig_remove(context_id)
|
||||||
|
|
||||||
|
async def traced_on_audio_context_completed(context_id):
|
||||||
|
end_tts_span(service, context_id)
|
||||||
|
return await orig_completed(context_id)
|
||||||
|
|
||||||
|
def traced_reset_active_audio_context():
|
||||||
|
playing_id = getattr(service, "_playing_context_id", None)
|
||||||
|
if playing_id is not None:
|
||||||
|
end_tts_span(service, playing_id, interrupted=True)
|
||||||
|
return orig_reset_active()
|
||||||
|
|
||||||
|
service.create_audio_context = traced_create_audio_context
|
||||||
|
service.append_to_audio_context = traced_append_to_audio_context
|
||||||
|
service.push_frame = traced_push_frame
|
||||||
|
service.remove_audio_context = traced_remove_audio_context
|
||||||
|
service.on_audio_context_completed = traced_on_audio_context_completed
|
||||||
|
service.reset_active_audio_context = traced_reset_active_audio_context
|
||||||
|
|
||||||
|
def patch_setup(owner):
|
||||||
|
"""Wrap ``owner.setup`` so audio-context patches install per-instance.
|
||||||
|
|
||||||
|
Idempotent: if a parent class has already been wrapped,
|
||||||
|
skip. The patches check ``_tracing_enabled`` at invocation
|
||||||
|
time, so wrapping is always safe.
|
||||||
|
"""
|
||||||
|
original_setup = owner.setup
|
||||||
|
if getattr(original_setup, "__tts_tracing_setup_wrapped__", False):
|
||||||
return
|
return
|
||||||
|
|
||||||
service_class_name = self.__class__.__name__
|
@functools.wraps(original_setup)
|
||||||
span_name = "tts"
|
async def patched_setup(self, setup):
|
||||||
|
await original_setup(self, setup)
|
||||||
|
install_audio_context_patches(self)
|
||||||
|
|
||||||
# Get parent context
|
patched_setup.__tts_tracing_setup_wrapped__ = True
|
||||||
parent_context = _get_turn_context(self) or _get_parent_service_context(self)
|
owner.setup = patched_setup
|
||||||
|
|
||||||
# Create span
|
def attach_run_tts_attributes(service, text, args, kwargs):
|
||||||
tracer = trace.get_tracer("pipecat")
|
"""Attach text-specific attributes to the in-flight TTS span."""
|
||||||
with tracer.start_as_current_span(span_name, context=parent_context) as span:
|
if not getattr(service, "_tracing_enabled", False):
|
||||||
try:
|
return
|
||||||
settings = getattr(self, "_settings", None)
|
try:
|
||||||
add_tts_span_attributes(
|
context_id = args[0] if args else kwargs.get("context_id")
|
||||||
span=span,
|
entry = getattr(service, "_tts_spans", {}).get(context_id)
|
||||||
service_name=service_class_name,
|
if entry and text:
|
||||||
model=_get_model_name(self),
|
span = entry["span"]
|
||||||
voice_id=getattr(settings, "voice", "unknown"),
|
span.set_attribute("text", text)
|
||||||
text=text,
|
span.set_attribute("metrics.character_count", len(text))
|
||||||
settings=settings,
|
except Exception as e:
|
||||||
character_count=len(text),
|
logging.warning(f"Error attaching TTS text to span: {e}")
|
||||||
operation_name="tts",
|
|
||||||
cartesia_version=getattr(self, "_cartesia_version", None),
|
|
||||||
context_id=getattr(self, "_context_id", None),
|
|
||||||
)
|
|
||||||
|
|
||||||
yield span
|
def make_run_tts_wrapper():
|
||||||
|
"""Build the wrapper around ``run_tts`` that adds per-call attributes.
|
||||||
|
|
||||||
except Exception as e:
|
Span lifetime is owned by the audio-context patches. This
|
||||||
logging.warning(f"Error in TTS tracing: {e}")
|
wrapper only attaches the text and character count to the
|
||||||
raise
|
span that was opened by ``create_audio_context`` just
|
||||||
finally:
|
before ``run_tts`` was invoked.
|
||||||
# Update TTFB metric at the end
|
"""
|
||||||
ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None)
|
if is_async_generator:
|
||||||
if ttfb is not None:
|
|
||||||
span.set_attribute("metrics.ttfb", ttfb)
|
|
||||||
|
|
||||||
if is_async_generator:
|
@functools.wraps(f)
|
||||||
|
async def gen_wrapper(self, text, *args, **kwargs):
|
||||||
@functools.wraps(f)
|
attach_run_tts_attributes(self, text, args, kwargs)
|
||||||
async def gen_wrapper(self, text, *args, **kwargs):
|
|
||||||
if not getattr(self, "_tracing_enabled", False):
|
|
||||||
async for item in f(self, text, *args, **kwargs):
|
|
||||||
yield item
|
|
||||||
return
|
|
||||||
|
|
||||||
fn_called = False
|
|
||||||
try:
|
|
||||||
async with tracing_context(self, text):
|
|
||||||
fn_called = True
|
|
||||||
async for item in f(self, text, *args, **kwargs):
|
|
||||||
yield item
|
|
||||||
except Exception as e:
|
|
||||||
if fn_called:
|
|
||||||
raise
|
|
||||||
logging.error(f"Error in TTS tracing (continuing without tracing): {e}")
|
|
||||||
async for item in f(self, text, *args, **kwargs):
|
async for item in f(self, text, *args, **kwargs):
|
||||||
yield item
|
yield item
|
||||||
|
|
||||||
return gen_wrapper
|
return gen_wrapper
|
||||||
else:
|
|
||||||
|
|
||||||
@functools.wraps(f)
|
@functools.wraps(f)
|
||||||
async def wrapper(self, text, *args, **kwargs):
|
async def coro_wrapper(self, text, *args, **kwargs):
|
||||||
if not getattr(self, "_tracing_enabled", False):
|
attach_run_tts_attributes(self, text, args, kwargs)
|
||||||
return await f(self, text, *args, **kwargs)
|
return await f(self, text, *args, **kwargs)
|
||||||
|
|
||||||
fn_called = False
|
return coro_wrapper
|
||||||
try:
|
|
||||||
async with tracing_context(self, text):
|
|
||||||
fn_called = True
|
|
||||||
return await f(self, text, *args, **kwargs)
|
|
||||||
except Exception as e:
|
|
||||||
if fn_called:
|
|
||||||
raise
|
|
||||||
logging.error(f"Error in TTS tracing (continuing without tracing): {e}")
|
|
||||||
return await f(self, text, *args, **kwargs)
|
|
||||||
|
|
||||||
return wrapper
|
class _TracedTTSDescriptor:
|
||||||
|
"""Class-level descriptor that wires up TTS tracing at class definition time.
|
||||||
|
|
||||||
|
``__set_name__`` fires when the class body finishes evaluating,
|
||||||
|
giving us a chance to wrap the owner's ``setup()`` so that the
|
||||||
|
audio-context patches install on every instance before any
|
||||||
|
``create_audio_context`` call (including the very first one).
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __set_name__(self, owner, attr_name):
|
||||||
|
patch_setup(owner)
|
||||||
|
setattr(owner, attr_name, make_run_tts_wrapper())
|
||||||
|
|
||||||
|
return _TracedTTSDescriptor()
|
||||||
|
|
||||||
if func is not None:
|
if func is not None:
|
||||||
return decorator(func)
|
return decorator(func)
|
||||||
|
|||||||
Reference in New Issue
Block a user