Implement Sentry instrumentation for performance and error tracking (#470)
* feat: Add Sentry support in FrameProcessor This update add optional Sentry integration for performance tracking and error monitoring. Key changes include: - Add conditional Sentry import and initialization check - Implement Sentry spans in FrameProcessorMetrics to measure TTFB (Time To First Byte) and processing time when Sentry is available - Maintain existing metrics functionality with MetricsFrame regardless of Sentry availability * feat: Enable metrics in DeepgramSTTService for Sentry This commit enhances the DeepgramSTTService class to enable metrics generation for use with Sentry. Key changes include: 1. Enable general metrics generation: - Implement `can_generate_metrics` method, returning True when VAD is enabled - This allows metrics to be collected and used by both Sentry and the metrics system in frame_processor.py 2. Integrate Sentry-compatible performance tracking: - Add start_ttfb_metrics and start_processing_metrics calls in the VAD speech detection handler - Implement stop_ttfb_metrics call when receiving transcripts - Add stop_processing_metrics for final transcripts 3. Enhance VAD support for metrics: - Add `vad_enabled` property to check VAD event availability - Implement VAD-based speech detection handler for precise metric timing These changes enable detailed performance tracking via both Sentry and the general metrics system when VAD is active. This allows for better monitoring and analysis of the speech-to-text process, providing valuable insights through Sentry and any other metrics consumers in the pipeline. * Update frame_processor.py * Refactor to support flexible metrics implementation - Modified the __init__ method to accept a metrics parameter that is either FrameProcessorMetrics or one of its subclasses - Updated the metrics initialization to create an instance with the processor's name - Moved all FrameProcessorMetrics-related logic to a new processors\metrics\base.py file * Implement flexible metrics system with Sentry integration 1. Created a new metrics module in processors/metrics/ 2. Implemented FrameProcessorMetrics base class in base.py: 3. Implemented SentryMetrics class in sentry.py: - Inherits from FrameProcessorMetrics - Integrates with Sentry SDK for advanced metrics tracking - Implements Sentry-specific span creation and management for TTFB and processing metrics - Handles cases where Sentry is not available or initialized
This commit is contained in:
Cyril S.
@@ -14,18 +14,14 @@ from pipecat.frames.frames import (
|
||||
EndFrame,
|
||||
ErrorFrame,
|
||||
Frame,
|
||||
MetricsFrame,
|
||||
StartFrame,
|
||||
StartInterruptionFrame,
|
||||
StopInterruptionFrame,
|
||||
SystemFrame)
|
||||
from pipecat.metrics.metrics import (
|
||||
LLMTokenUsage,
|
||||
LLMUsageMetricsData,
|
||||
MetricsData,
|
||||
ProcessingMetricsData,
|
||||
TTFBMetricsData,
|
||||
TTSUsageMetricsData)
|
||||
MetricsData)
|
||||
from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
|
||||
from pipecat.utils.utils import obj_count, obj_id
|
||||
|
||||
from loguru import logger
|
||||
@@ -36,78 +32,13 @@ class FrameDirection(Enum):
|
||||
UPSTREAM = 2
|
||||
|
||||
|
||||
class FrameProcessorMetrics:
|
||||
def __init__(self, name: str):
|
||||
self._core_metrics_data = MetricsData(processor=name)
|
||||
self._start_ttfb_time = 0
|
||||
self._start_processing_time = 0
|
||||
self._should_report_ttfb = True
|
||||
|
||||
def _processor_name(self):
|
||||
return self._core_metrics_data.processor
|
||||
|
||||
def _model_name(self):
|
||||
return self._core_metrics_data.model
|
||||
|
||||
def set_core_metrics_data(self, data: MetricsData):
|
||||
self._core_metrics_data = data
|
||||
|
||||
async def start_ttfb_metrics(self, report_only_initial_ttfb):
|
||||
if self._should_report_ttfb:
|
||||
self._start_ttfb_time = time.time()
|
||||
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}")
|
||||
ttfb = TTFBMetricsData(
|
||||
processor=self._processor_name(),
|
||||
value=value,
|
||||
model=self._model_name())
|
||||
self._start_ttfb_time = 0
|
||||
return MetricsFrame(data=[ttfb])
|
||||
|
||||
async def start_processing_metrics(self):
|
||||
self._start_processing_time = time.time()
|
||||
|
||||
async def stop_processing_metrics(self):
|
||||
if self._start_processing_time == 0:
|
||||
return None
|
||||
|
||||
value = time.time() - self._start_processing_time
|
||||
logger.debug(f"{self._processor_name()} processing time: {value}")
|
||||
processing = ProcessingMetricsData(
|
||||
processor=self._processor_name(), value=value, model=self._model_name())
|
||||
self._start_processing_time = 0
|
||||
return MetricsFrame(data=[processing])
|
||||
|
||||
async def start_llm_usage_metrics(self, tokens: LLMTokenUsage):
|
||||
logger.debug(
|
||||
f"{self._processor_name()} prompt tokens: {tokens.prompt_tokens}, completion tokens: {tokens.completion_tokens}")
|
||||
value = LLMUsageMetricsData(
|
||||
processor=self._processor_name(),
|
||||
model=self._model_name(),
|
||||
value=tokens)
|
||||
return MetricsFrame(data=[value])
|
||||
|
||||
async def start_tts_usage_metrics(self, text: str):
|
||||
characters = TTSUsageMetricsData(
|
||||
processor=self._processor_name(),
|
||||
model=self._model_name(),
|
||||
value=len(text))
|
||||
logger.debug(f"{self._processor_name()} usage characters: {characters.value}")
|
||||
return MetricsFrame(data=[characters])
|
||||
|
||||
|
||||
class FrameProcessor:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
name: str | None = None,
|
||||
metrics: FrameProcessorMetrics | None = None,
|
||||
sync: bool = True,
|
||||
loop: asyncio.AbstractEventLoop | None = None,
|
||||
**kwargs):
|
||||
@@ -129,7 +60,8 @@ class FrameProcessor:
|
||||
self._report_only_initial_ttfb = False
|
||||
|
||||
# Metrics
|
||||
self._metrics = FrameProcessorMetrics(name=self.name)
|
||||
self._metrics = metrics or FrameProcessorMetrics()
|
||||
self._metrics.set_processor_name(self.name)
|
||||
|
||||
# Every processor in Pipecat should only output frames from a single
|
||||
# task. This avoid problems like audio overlapping. System frames are
|
||||
@@ -262,14 +194,16 @@ class FrameProcessor:
|
||||
logger.trace(f"Pushing {frame} from {self} to {self._next}")
|
||||
await self._next.process_frame(frame, direction)
|
||||
elif direction == FrameDirection.UPSTREAM and self._prev:
|
||||
logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
|
||||
logger.trace(f"Pushing {frame} upstream from {
|
||||
self} to {self._prev}")
|
||||
await self._prev.process_frame(frame, direction)
|
||||
except Exception as e:
|
||||
logger.exception(f"Uncaught exception in {self}: {e}")
|
||||
|
||||
def __create_push_task(self):
|
||||
self.__push_queue = asyncio.Queue()
|
||||
self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler())
|
||||
self.__push_frame_task = self.get_event_loop(
|
||||
).create_task(self.__push_frame_task_handler())
|
||||
|
||||
async def __push_frame_task_handler(self):
|
||||
running = True
|
||||
|
||||
Reference in New Issue
Block a user