qi9dZddlZddlmZddlmZmZmZddlm Z m Z ddl m Z m Z mZmZmZmZmZmZmZddlmZmZddlmZmZdd lmZGd d e ZGd d e ZGdde ZGdde Z GddeZ!y)aMObserver for tracking user-to-bot response latency. This module provides an observer that monitors the time between when a user stops speaking and when the bot starts speaking, emitting events when latency is measured. Optionally collects per-service latency breakdown metrics (TTFB, text aggregation) when ``enable_metrics=True``. N)deque)DictListOptional) BaseModelField) BotStartedSpeakingFrameClientConnectedFrameFunctionCallInProgressFrameFunctionCallResultFrameInterruptionFrame MetricsFrameUserStoppedSpeakingFrameVADUserStartedSpeakingFrameVADUserStoppedSpeakingFrame)TextAggregationMetricsDataTTFBMetricsData) BaseObserver FramePushed)FrameDirectioncDeZdZUdZeed<dZeeed<eed<eed<y)TTFBBreakdownMetricsaDTTFB measurement with timestamp for timeline placement. Parameters: processor: Name of the processor that reported the TTFB. model: Optional model name associated with the metric. start_time: Unix timestamp when the TTFB measurement started. duration_secs: TTFB duration in seconds. processorNmodel start_time duration_secs) __name__ __module__ __qualname____doc__str__annotations__rrfloat]/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/observers/user_bot_latency_observer.pyrr(s(NE8C=r%rc0eZdZUdZeed<eed<eed<y)TextAggregationBreakdownMetricsaText aggregation measurement with timestamp for timeline placement. Parameters: processor: Name of the processor that reported the metric. start_time: Unix timestamp when text aggregation started. duration_secs: Aggregation duration in seconds. rrrNrrrr r!r"r#r$r%r&r(r(8sNr%r(c0eZdZUdZeed<eed<eed<y)FunctionCallMetricsaLatency for a single function call execution. Parameters: function_name: Name of the function that was called. start_time: Unix timestamp when execution started. duration_secs: Time in seconds from execution start to result. function_namerrNr)r$r%r&r+r+Fsr%r+ceZdZUdZeeZeee d<dZ e e e d<dZ e ee d<dZe ee d<eeZeee d<d eefd Zy) LatencyBreakdownaPer-service latency breakdown for a single user-to-bot cycle. Collected between ``VADUserStoppedSpeakingFrame`` and ``BotStartedSpeakingFrame`` when ``enable_metrics=True`` in :class:`~pipecat.pipeline.task.PipelineParams`. Parameters: ttfb: Time-to-first-byte metrics from each service in the pipeline. text_aggregation: First text aggregation measurement, representing the latency cost of sentence aggregation in the TTS pipeline. user_turn_start_time: Unix timestamp when the user turn started (actual user silence, adjusted for VAD stop_secs). ``None`` if no ``VADUserStoppedSpeakingFrame`` was observed. user_turn_secs: Duration in seconds of the user's turn, measured from when the user actually stopped speaking to when the turn was released (``UserStoppedSpeakingFrame``). This includes VAD silence detection, STT finalization, and any turn analyzer wait. ``None`` if no ``UserStoppedSpeakingFrame`` was observed (e.g. no turn analyzer configured). function_calls: Latency for each function call executed during this cycle. Empty if no function calls occurred. )default_factoryttfbNtext_aggregationuser_turn_start_timeuser_turn_secsfunction_callsreturncg}|j8|j,|j|jd|jddf|jD]:}|j|j|j d|j ddf<|jD]:}|j|j|jd|j ddf<|jrD|j}|j|j|j d|j ddf|jd|Dcgc]\}}| c}}Scc}}w) a*Return human-readable event labels sorted by start time. Collects all sub-metrics into a flat list, sorts by ``start_time``, and returns formatted strings suitable for logging. Returns: List of formatted strings, one per event, in chronological order. z User turn: z.3fsz: TTFB z: z: text aggregation c |dS)Nrr$)es r&z7LatencyBreakdown.chronological_events..s !A$r%)key) r2r3appendr0rrrr4r,r1sort)selfeventstfcta_labels r&chronological_eventsz%LatencyBreakdown.chronological_eventsrsO!  $ $ 0T5H5H5T MM444 DDWDWX[C\\]6^_ ` YA MM1<bAQAQRU@VVW*XY Z [  &&B MM2<<.0CBDTDTUXCYYZ [\    '&,-(!U---s E)rrrr rlistr0rrr"r1rr(r2r#r3r4r+r!rEr$r%r&r.r.Tsq.(-T'BD$# $BBFh>?F,0(5/0&*NHUO*05d0KND,-K.d3i.r%r.cLeZdZdZddfd ZdefdZdZdefd Z d Z xZ S) UserBotLatencyObserveru}Observer that tracks user-to-bot response latency. Measures the time between when a user stops speaking (VADUserStoppedSpeakingFrame) and when the bot starts speaking (BotStartedSpeakingFrame). Emits events when latency is measured, allowing consumers to log, trace, or otherwise process the latency data. When ``enable_metrics=True`` in pipeline params, also collects per-service latency breakdown (TTFB, text aggregation) and emits an ``on_latency_breakdown`` event alongside the existing latency measurement. This observer follows the composition pattern used by TurnTrackingObserver, acting as a reusable component for latency measurement. Events: on_latency_measured(observer, latency_seconds): Emitted when time-to-first-bot-speech is calculated. Measures the time from when the user stopped speaking to when the bot starts speaking. on_latency_breakdown(observer, breakdown): Emitted at each ``BotStartedSpeakingFrame`` with a :class:`LatencyBreakdown` containing per-service metrics collected during the user→bot cycle. on_first_bot_speech_latency(observer, latency_seconds): Emitted once, the first time ``BotStartedSpeakingFrame`` arrives after ``ClientConnectedFrame``. Measures the time from client connection to the first bot speech. d) max_framesc Ht|di|d|_d|_d|_d|_d|_t|_t||_ g|_ d|_ i|_ g|_|jd|jd|jdy)atInitialize the user-bot latency observer. Sets up tracking for processed frames and user speech timing to calculate response latencies. Args: max_frames: Maximum number of frame IDs to keep in history for duplicate detection. Defaults to 100. **kwargs: Additional arguments passed to parent class. NF)maxlenon_latency_measuredon_latency_breakdownon_first_bot_speech_latencyr$)super__init___user_stopped_time_user_turn_start_time _user_turn_client_connected_time_first_bot_speech_measuredset_processed_framesr_frame_history_ttfb_text_aggregation_function_call_starts_function_call_metrics_register_event_handler)r>rJkwargs __class__s r&rQzUserBotLatencyObserver.__init__s "6"376:"+/8<#05''*e%**%=24 LPCE"AC# $$%:; $$%;< $$%BCr%datacPK|jtjk7ry|jj|j vry|j j |jj|jj|jjt|j t|jkDrt|j|_t|jtr&|jtj|_ yt|jtr-d|_d|_d|_|j'd|_yt|jt*rD|jj,|jj.z |_|j |_yt|jt0r4|j 'tj|j z |_yyt|jt2r|j'yt|jt4rL|jj6tjf|j8|jj:<yt|jt<ru|j8j?|jj:d}|B|\}}|j@jtC||tj|z yyt|jtDr|jG|jyt|jtHr|jKd{yy7w)azProcess frames to track speech timing and calculate latency. Tracks VAD events and bot speaking events to measure the time between user stopping speech and bot starting speech. Also accumulates metrics from MetricsFrame for the latency breakdown. Args: data: Frame push event containing the frame and direction information. NT)r,rr)& directionr DOWNSTREAMframeidrXaddrYr<lenrW isinstancer rUtimerrRrSrT_reset_accumulatorsrVr timestamp stop_secsrr r r,r\ tool_call_idr popr]r+r_handle_metrics_framer _handle_bot_started_speaking)r>rastartr,rs r& on_push_framez$UserBotLatencyObserver.on_push_frames >>^66 6  ::==D22 2  ""4::==1 ""4::==1 t%% &T-@-@)A A%()<)<%=D " djj"6 7**2.2iik+  djj"= >&*D #)-D &"DO  $ $ &/3D +  $? @'+jj&:&:TZZ=Q=Q&QD #)-)@)@D &  $< =&&2"&))+0G0G"G3  $5 6  $ $ &  $? @ (( CD & &tzz'>'> ? $; <..224::3J3JDQE ,1) z++22'&3#-&*iikJ&>! L 1  & &tzz 2  $; <335 5 5= 5sNN&N$N&c vKd}|jP|jsDd|_tj|jz }|jd|d{d}|jDtj|jz }d|_|jd|d{d}|rt t |j|j|j|jt |j}|jd|d{|jyy777w)z=Handle BotStartedSpeakingFrame to emit latency and breakdown.FNTrOrM)r0r1r2r3r4rN) rUrVrj_call_event_handlerrRr.rFrZr[rSrTr]rk)r>emit_breakdownlatency breakdowns r&rqz3UserBotLatencyObserver._handle_bot_started_speakings  & & 24;Z;Z.2D +iikD$?$??G**+H'R R R!N  " " .iikD$;$;;G&*D #**+@'J J J!N ($**%!%!7!7%)%?%?##D$?$?@ I**+A9M M M  $ $ &  S K Ns8AD9D3AD9*D5+A1D9D7D95D97D9rec 0|jduxr |j }|j|sytj}|jD]}t |t rc|jdkDrT|jjt|j|j||jz |jvt |ts|jt|j||jz |j|_y)uExtract latency metrics from a MetricsFrame. Accumulates metrics when a measurement is in progress: either a user→bot cycle (after ``VADUserStoppedSpeakingFrame``) or the first-bot-speech window (after ``ClientConnectedFrame``). Nr)rrrr)rrr)rUrVrRrjrarirvaluerZr<rrrrr[r()r>rewaiting_for_first_speechnow metrics_datas r&rpz,UserBotLatencyObserver._handle_metrics_frame6s  ' 't 3 [D<[<[8[ !  " " *3K iik!JJ L,8\=O=ORS=S !!("."8"8*00#&););#;&2&8&8 L*DE))1-L"."8"8#&););#;&2&8&8.D* r%cXg|_d|_d|_d|_i|_g|_y)z$Clear per-cycle metric accumulators.N)rZr[rSrTr\r])r>s r&rkz*UserBotLatencyObserver._reset_accumulatorsXs0 !%%)"%'"&(#r%) rrrr rQrrsrqrrprk __classcell__)r`s@r&rHrHs:6&) DDJ6 J6X'8 < D)r%rH)"r rj collectionsrtypingrrrpydanticrrpipecat.frames.framesr r r r r rrrrpipecat.metrics.metricsrrpipecat.observers.base_observerrr"pipecat.processors.frame_processorrrr(r+r.rHr$r%r&rsw ''%   F= 9  i  ) 9.y9.xO)\O)r%