From c54232bdb40c28c7e28e1600bd3492212ff4b6cc Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 28 Feb 2026 14:04:21 -0500 Subject: [PATCH] Add StartupTimingObserver for measuring processor start() times Tracks how long each processor start method takes during pipeline startup by measuring StartFrame arrive/leave deltas. Emits a timing report via the on_startup_timing_report event and auto-logs a summary. Internal pipeline processors are excluded from reports by default. --- .../foundational/29-turn-tracking-observer.py | 12 +- .../observers/startup_timing_observer.py | 232 ++++++++++++++++++ tests/test_startup_timing_observer.py | 186 ++++++++++++++ 3 files changed, 427 insertions(+), 3 deletions(-) create mode 100644 src/pipecat/observers/startup_timing_observer.py create mode 100644 tests/test_startup_timing_observer.py diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 321197db2..3e85ddfb8 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -12,6 +12,7 @@ from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame +from pipecat.observers.startup_timing_observer import StartupTimingObserver from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner @@ -87,8 +88,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ] ) - # Create latency tracking observer latency_observer = UserBotLatencyObserver() + startup_observer = StartupTimingObserver() task = PipelineTask( pipeline, @@ -97,14 +98,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): enable_usage_metrics=True, ), idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - observers=[latency_observer], + observers=[latency_observer, startup_observer], ) - # Log latency measurements using the event handler @latency_observer.event_handler("on_latency_measured") async def on_latency_measured(observer, latency_seconds): logger.info(f"⏱️ User-to-bot latency: {latency_seconds:.3f}s") + @startup_observer.event_handler("on_startup_timing_report") + async def on_startup_timing_report(observer, report): + logger.info(f"Total startup: {report.total_duration_secs:.3f}s") + for timing in report.processor_timings: + logger.info(f" {timing.processor_name}: {timing.duration_secs:.3f}s") + turn_observer = task.turn_tracking_observer if turn_observer: diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py new file mode 100644 index 000000000..0f3ad0b7a --- /dev/null +++ b/src/pipecat/observers/startup_timing_observer.py @@ -0,0 +1,232 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Observer for tracking pipeline startup timing. + +This module provides an observer that measures how long each processor's +``start()`` method takes during pipeline startup. It works by tracking +when a ``StartFrame`` arrives at a processor (``on_process_frame``) versus +when it leaves (``on_push_frame``), giving the exact ``start()`` duration +for each processor in the pipeline. + +Example:: + + observer = StartupTimingObserver() + + @observer.event_handler("on_startup_timing_report") + async def on_report(observer, report): + for t in report.processor_timings: + print(f"{t.processor_name}: {t.duration_secs:.3f}s") + + task = PipelineTask(pipeline, observers=[observer]) +""" + +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple, Type + +from loguru import logger + +from pipecat.frames.frames import StartFrame +from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed +from pipecat.pipeline.base_pipeline import BasePipeline +from pipecat.pipeline.pipeline import PipelineSink, PipelineSource +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +# Internal pipeline types excluded from tracking by default. +_INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline) + + +@dataclass +class ProcessorStartupTiming: + """Startup timing for a single processor. + + Parameters: + processor_name: The name of the processor. + duration_secs: How long the processor's start() took, in seconds. + """ + + processor_name: str + duration_secs: float + + +@dataclass +class StartupTimingReport: + """Report of startup timings for all measured processors. + + Parameters: + total_duration_secs: Total wall-clock time from first to last processor start. + processor_timings: Per-processor timing data, in pipeline order. + """ + + total_duration_secs: float + processor_timings: List[ProcessorStartupTiming] = field(default_factory=list) + + +class StartupTimingObserver(BaseObserver): + """Observer that measures processor startup times during pipeline initialization. + + Tracks how long each processor's ``start()`` method takes by measuring the + time between when a ``StartFrame`` arrives at a processor and when it is + pushed downstream. This captures WebSocket connections, API authentication, + model loading, and other initialization work. + + By default, internal pipeline processors (``PipelineSource``, ``PipelineSink``, + ``Pipeline``) are excluded from the report. Pass ``processor_types`` to + measure only specific types. + + Event handlers available: + + - on_startup_timing_report: Called once after startup completes with the full + timing report. + + Example:: + + observer = StartupTimingObserver( + processor_types=(STTService, TTSService) + ) + + @observer.event_handler("on_startup_timing_report") + async def on_report(observer, report): + for t in report.processor_timings: + logger.info(f"{t.processor_name}: {t.duration_secs:.3f}s") + + task = PipelineTask(pipeline, observers=[observer]) + + Args: + processor_types: Optional tuple of processor types to measure. If None, + all non-internal processors are measured. + """ + + def __init__( + self, + *, + processor_types: Optional[Tuple[Type[FrameProcessor], ...]] = None, + **kwargs, + ): + """Initialize the startup timing observer. + + Args: + processor_types: Optional tuple of processor types to measure. + If None, all non-internal processors are measured. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(**kwargs) + self._processor_types = processor_types + + # Map processor ID -> (processor, arrival_timestamp_ns) + self._arrivals: Dict[int, Tuple[FrameProcessor, int]] = {} + + # Collected timings in pipeline order. + self._timings: List[ProcessorStartupTiming] = [] + + # Lock onto the first StartFrame we see (by frame ID). + self._start_frame_id: Optional[str] = None + + # Whether we've already emitted the report. + self._reported = False + + self._register_event_handler("on_startup_timing_report") + + def _should_track(self, processor: FrameProcessor) -> bool: + """Check if a processor should be tracked for timing. + + Args: + processor: The processor to check. + + Returns: + True if the processor matches the filter or no filter is set. + """ + if self._processor_types is not None: + return isinstance(processor, self._processor_types) + # Default: exclude internal pipeline plumbing. + return not isinstance(processor, _INTERNAL_TYPES) + + async def on_process_frame(self, data: FrameProcessed): + """Record when a StartFrame arrives at a processor. + + When a ``StartFrame`` reaches a ``PipelineSink``, startup is complete + (the frame has traversed the entire pipeline) and the report is emitted. + + Args: + data: The frame processing event data. + """ + if self._reported: + return + + if not isinstance(data.frame, StartFrame): + return + + if data.direction != FrameDirection.DOWNSTREAM: + return + + # Lock onto the first StartFrame. + if self._start_frame_id is None: + self._start_frame_id = data.frame.id + elif data.frame.id != self._start_frame_id: + return + + # When the StartFrame reaches a PipelineSink, all processors have + # completed start(). PipelineSinks use direct mode so the outermost + # sink fires last within the same synchronous call chain. + if isinstance(data.processor, PipelineSink): + if self._timings: + await self._emit_report() + return + + if self._should_track(data.processor): + self._arrivals[data.processor.id] = (data.processor, data.timestamp) + + async def on_push_frame(self, data: FramePushed): + """Record when a StartFrame leaves a processor and compute the delta. + + Args: + data: The frame push event data. + """ + if self._reported: + return + + if not isinstance(data.frame, StartFrame): + return + + if data.direction != FrameDirection.DOWNSTREAM: + return + + if self._start_frame_id is not None and data.frame.id != self._start_frame_id: + return + + arrival = self._arrivals.pop(data.source.id, None) + if arrival is None: + return + + processor, arrival_ts = arrival + duration_ns = data.timestamp - arrival_ts + duration_secs = duration_ns / 1e9 + + self._timings.append( + ProcessorStartupTiming( + processor_name=processor.name, + duration_secs=duration_secs, + ) + ) + + async def _emit_report(self): + """Build and emit the startup timing report.""" + if self._reported: + return + self._reported = True + + total = sum(t.duration_secs for t in self._timings) + + report = StartupTimingReport( + total_duration_secs=total, + processor_timings=self._timings, + ) + + logger.debug(f"Pipeline startup completed in {total:.3f}s") + for t in self._timings: + logger.debug(f" {t.processor_name}: {t.duration_secs:.3f}s") + + await self._call_event_handler("on_startup_timing_report", report) diff --git a/tests/test_startup_timing_observer.py b/tests/test_startup_timing_observer.py new file mode 100644 index 000000000..e3cd7c2b7 --- /dev/null +++ b/tests/test_startup_timing_observer.py @@ -0,0 +1,186 @@ +import asyncio +import unittest + +from pipecat.frames.frames import Frame, StartFrame, TextFrame +from pipecat.observers.startup_timing_observer import ( + StartupTimingObserver, + StartupTimingReport, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.tests.utils import run_test + + +class SlowStartProcessor(FrameProcessor): + """A processor that sleeps during start to simulate slow initialization.""" + + def __init__(self, delay: float = 0.1, **kwargs): + super().__init__(**kwargs) + self._delay = delay + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartFrame): + await asyncio.sleep(self._delay) + await self.push_frame(frame, direction) + + +class FastProcessor(FrameProcessor): + """A processor with no start delay.""" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + + +class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): + """Tests for StartupTimingObserver.""" + + async def test_timing_reported(self): + """Test that startup timing is measured and reported.""" + observer = StartupTimingObserver() + processor = SlowStartProcessor(delay=0.1) + + reports = [] + + @observer.event_handler("on_startup_timing_report") + async def on_report(obs, report): + reports.append(report) + + frames_to_send = [TextFrame(text="hello")] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=[TextFrame], + observers=[observer], + ) + + self.assertEqual(len(reports), 1) + report = reports[0] + self.assertGreater(report.total_duration_secs, 0) + self.assertGreater(len(report.processor_timings), 0) + + # Find our slow processor in the timings. + slow_timings = [ + t for t in report.processor_timings if "SlowStartProcessor" in t.processor_name + ] + self.assertEqual(len(slow_timings), 1) + self.assertGreaterEqual(slow_timings[0].duration_secs, 0.05) + + async def test_processor_types_filter(self): + """Test that processor_types filter limits which processors appear.""" + observer = StartupTimingObserver(processor_types=(SlowStartProcessor,)) + processor = SlowStartProcessor(delay=0.05) + + reports = [] + + @observer.event_handler("on_startup_timing_report") + async def on_report(obs, report): + reports.append(report) + + frames_to_send = [TextFrame(text="hello")] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=[TextFrame], + observers=[observer], + ) + + self.assertEqual(len(reports), 1) + report = reports[0] + + # Only SlowStartProcessor should be in the timings. + for t in report.processor_timings: + self.assertIn("SlowStartProcessor", t.processor_name) + + async def test_report_emits_once(self): + """Test that the report is emitted only once even with multiple frames.""" + observer = StartupTimingObserver() + processor = FastProcessor() + + reports = [] + + @observer.event_handler("on_startup_timing_report") + async def on_report(obs, report): + reports.append(report) + + frames_to_send = [ + TextFrame(text="first"), + TextFrame(text="second"), + TextFrame(text="third"), + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=[TextFrame, TextFrame, TextFrame], + observers=[observer], + ) + + self.assertEqual(len(reports), 1) + + async def test_event_handler_receives_report(self): + """Test that the event handler receives a proper StartupTimingReport.""" + observer = StartupTimingObserver() + processor = SlowStartProcessor(delay=0.05) + + reports = [] + + @observer.event_handler("on_startup_timing_report") + async def on_report(obs, report): + reports.append(report) + + frames_to_send = [TextFrame(text="hello")] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=[TextFrame], + observers=[observer], + ) + + self.assertEqual(len(reports), 1) + report = reports[0] + self.assertIsInstance(report, StartupTimingReport) + self.assertIsInstance(report.total_duration_secs, float) + for timing in report.processor_timings: + self.assertIsInstance(timing.processor_name, str) + self.assertIsInstance(timing.duration_secs, float) + + async def test_excludes_internal_processors(self): + """Test that internal pipeline processors are excluded by default.""" + observer = StartupTimingObserver() + processor = FastProcessor() + + reports = [] + + @observer.event_handler("on_startup_timing_report") + async def on_report(obs, report): + reports.append(report) + + frames_to_send = [TextFrame(text="hello")] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=[TextFrame], + observers=[observer], + ) + + self.assertEqual(len(reports), 1) + report = reports[0] + + # No internal processors (PipelineSource, PipelineSink, Pipeline) in the report. + internal_names = ("Pipeline#", "PipelineTask#") + for t in report.processor_timings: + for prefix in internal_names: + self.assertNotIn( + prefix, + t.processor_name, + f"Internal processor {t.processor_name} should be excluded by default", + ) + + +if __name__ == "__main__": + unittest.main()