wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3355 from itsderek23/user-bot-latency

Browse Source 
Add `user_bot_latency_seconds` to OpenTelemetry turn spans
This commit is contained in:
Mark Backman
2026-02-05 15:23:15 -05:00
committed by GitHub
parent 5a5a98b497 56d8ef2bf4
commit 9b43cde128
8 changed files with 286 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3355.added.md Normal file
View File

@@ -0,0 +1 @@
- Added `UserBotLatencyObserver` for tracking user-to-bot response latency. When tracing is enabled, latency measurements are automatically recorded as `turn.user_bot_latency_seconds` attributes on OpenTelemetry turn spans.

1
changelog/3355.deprecated.md Normal file
View File

@@ -0,0 +1 @@
- Deprecated `UserBotLatencyLogObserver`. Use `UserBotLatencyObserver` directly with its `on_latency_measured` event handler instead.

12
examples/foundational/29-turn-tracking-observer.py
View File

@@ -14,7 +14,7 @@ from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnal
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame
from pipecat.observers.loggers.user_bot_latency_log_observer import UserBotLatencyLogObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -96,6 +96,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
]
)
# Create latency tracking observer
latency_observer = UserBotLatencyObserver()
task = PipelineTask(
pipeline,
params=PipelineParams(
@@ -103,9 +106,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
observers=[UserBotLatencyLogObserver()],
observers=[latency_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")
turn_observer = task.turn_tracking_observer
if turn_observer:

22
src/pipecat/observers/loggers/user_bot_latency_log_observer.py
View File

@@ -4,9 +4,15 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Observer for measuring user-to-bot response latency."""
"""Observer for measuring user-to-bot response latency.
.. deprecated:: 0.0.102
This module is deprecated. Use :class:`UserBotLatencyObserver` directly
with its ``on_latency_measured`` event handler instead.
"""
import time
import warnings
from statistics import mean
from loguru import logger
@@ -27,6 +33,10 @@ class UserBotLatencyLogObserver(BaseObserver):
This helps measure how quickly the AI services respond by tracking
conversation turn timing and logging latency metrics.
.. deprecated:: 0.0.102
This class is deprecated. Use :class:`UserBotLatencyObserver` directly
with its ``on_latency_measured`` event handler for custom logging.
"""
def __init__(self):
@@ -34,7 +44,17 @@ class UserBotLatencyLogObserver(BaseObserver):
Sets up tracking for processed frames and user speech timing
to calculate response latencies.
.. deprecated:: 0.0.102
This class is deprecated. Use :class:`UserBotLatencyObserver`
directly with its ``on_latency_measured`` event handler.
"""
warnings.warn(
"UserBotLatencyLogObserver is deprecated and will be removed in a future version. "
"Use UserBotLatencyObserver directly with its on_latency_measured event handler instead.",
DeprecationWarning,
stacklevel=2,
)
super().__init__()
self._user_bot_latency_processed_frames = set()
self._user_stopped_time = 0

81
src/pipecat/observers/user_bot_latency_observer.py Normal file
View File

@@ -0,0 +1,81 @@
"""Observer 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.
"""
import time
from typing import Optional, Set
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.processors.frame_processor import FrameDirection
class UserBotLatencyObserver(BaseObserver):
"""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.
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 user-to-bot
latency is calculated. Includes the latency value in seconds as a float.
"""
def __init__(self, **kwargs):
"""Initialize the user-bot latency observer.
Sets up tracking for processed frames and user speech timing
to calculate response latencies.
Args:
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._user_stopped_time: Optional[float] = None
self._processed_frames: Set[str] = set()
self._register_event_handler("on_latency_measured")
async def on_push_frame(self, data: FramePushed):
"""Process 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.
Args:
data: Frame push event containing the frame and direction information.
"""
# Only process downstream frames
if data.direction != FrameDirection.DOWNSTREAM:
return
# Skip already processed frames
if data.frame.id in self._processed_frames:
return
self._processed_frames.add(data.frame.id)
# Track VAD and bot speaking events for latency
if isinstance(data.frame, VADUserStartedSpeakingFrame):
# Reset when user starts speaking
self._user_stopped_time = None
elif isinstance(data.frame, VADUserStoppedSpeakingFrame):
# Record timestamp when user stops speaking
self._user_stopped_time = time.time()
elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time:
# Calculate and emit latency
latency = time.time() - self._user_stopped_time
self._user_stopped_time = None
await self._call_event_handler("on_latency_measured", latency)

7
src/pipecat/pipeline/task.py
View File

@@ -43,6 +43,7 @@ from pipecat.frames.frames import (
from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams
from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
@@ -287,13 +288,19 @@ class PipelineTask(BasePipelineTask):
observers = self._params.observers
observers = observers or []
self._turn_tracking_observer: Optional[TurnTrackingObserver] = None
self._user_bot_latency_observer: Optional[UserBotLatencyObserver] = None
self._turn_trace_observer: Optional[TurnTraceObserver] = None
if self._enable_turn_tracking:
self._turn_tracking_observer = TurnTrackingObserver()
observers.append(self._turn_tracking_observer)
if self._enable_tracing and self._turn_tracking_observer:
# Create latency observer for tracing
self._user_bot_latency_observer = UserBotLatencyObserver()
observers.append(self._user_bot_latency_observer)
# Create turn trace observer with latency tracking
self._turn_trace_observer = TurnTraceObserver(
self._turn_tracking_observer,
latency_tracker=self._user_bot_latency_observer,
conversation_id=self._conversation_id,
additional_span_attributes=self._additional_span_attributes,
)

35
src/pipecat/utils/tracing/turn_trace_observer.py
View File

@@ -18,6 +18,7 @@ from loguru import logger
from pipecat.frames.frames import StartFrame
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_context_provider import TurnContextProvider
@@ -45,6 +46,7 @@ class TurnTraceObserver(BaseObserver):
def __init__(
self,
turn_tracker: TurnTrackingObserver,
latency_tracker: UserBotLatencyObserver,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
**kwargs,
@@ -53,12 +55,14 @@ class TurnTraceObserver(BaseObserver):
Args:
turn_tracker: The turn tracking observer to monitor.
latency_tracker: The latency tracking observer for user-bot latency.
conversation_id: Optional conversation ID for grouping turns.
additional_span_attributes: Additional attributes to add to spans.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._turn_tracker = turn_tracker
self._latency_tracker = latency_tracker
self._current_span: Optional["Span"] = None
self._current_turn_number: int = 0
self._trace_context_map: Dict[int, "SpanContext"] = {}
@@ -69,15 +73,32 @@ class TurnTraceObserver(BaseObserver):
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
if turn_tracker:
@turn_tracker.event_handler("on_turn_started")
async def on_turn_started(tracker, turn_number):
await self._handle_turn_started(turn_number)
@turn_tracker.event_handler("on_turn_started")
async def on_turn_started(tracker, turn_number):
await self._handle_turn_started(turn_number)
@turn_tracker.event_handler("on_turn_ended")
async def on_turn_ended(tracker, turn_number, duration, was_interrupted):
await self._handle_turn_ended(turn_number, duration, was_interrupted)
@turn_tracker.event_handler("on_turn_ended")
async def on_turn_ended(tracker, turn_number, duration, was_interrupted):
await self._handle_turn_ended(turn_number, duration, was_interrupted)
@latency_tracker.event_handler("on_latency_measured")
async def on_latency_measured(tracker, latency_seconds):
await self._handle_latency_measured(latency_seconds)
async def _handle_latency_measured(self, latency_seconds: float):
"""Handle latency measurement events.
Called when the latency tracker measures user-to-bot latency.
Adds the latency as an attribute to the current turn span.
Args:
latency_seconds: The measured latency in seconds.
"""
if self._current_span and is_tracing_available():
self._current_span.set_attribute("turn.user_bot_latency_seconds", latency_seconds)
logger.debug(
f"Turn {self._current_turn_number} user-bot latency: {latency_seconds:.3f}s"
)
async def on_push_frame(self, data: FramePushed):
"""Process a frame without modifying it.

137
tests/test_user_bot_latency_observer.py Normal file
View File

@@ -0,0 +1,137 @@
import unittest
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.processors.filters.identity_filter import IdentityFilter
from pipecat.tests.utils import run_test
class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase):
"""Tests for UserBotLatencyObserver."""
async def test_normal_latency_measurement(self):
"""Test basic latency measurement from user stop to bot start."""
# Create observer
observer = UserBotLatencyObserver()
# Create identity filter (passes all frames through)
processor = IdentityFilter()
# Capture latency events
latencies = []
@observer.event_handler("on_latency_measured")
async def on_latency(obs, latency_seconds):
latencies.append(latency_seconds)
# Define frame sequence
frames_to_send = [
VADUserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
]
expected_down_frames = [
VADUserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
]
# Run test
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[observer],
)
# Verify latency was measured
self.assertEqual(len(latencies), 1)
self.assertGreater(latencies[0], 0)
self.assertLess(latencies[0], 1.0) # Should be very quick
async def test_multiple_latency_measurements(self):
"""Test that multiple user-bot exchanges produce separate latency events."""
# Create observer
observer = UserBotLatencyObserver()
# Create identity filter
processor = IdentityFilter()
# Capture latency events
latencies = []
@observer.event_handler("on_latency_measured")
async def on_latency(obs, latency_seconds):
latencies.append(latency_seconds)
# Define frame sequence with two complete cycles
frames_to_send = [
# First cycle
VADUserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
# Second cycle
VADUserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
]
expected_down_frames = [
VADUserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
]
# Run test
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[observer],
)
# Verify two separate latencies were measured
self.assertEqual(len(latencies), 2)
self.assertGreater(latencies[0], 0)
self.assertGreater(latencies[1], 0)
async def test_no_measurement_without_user_stop(self):
"""Test that latency is not measured if bot starts without user stopping first."""
# Create observer
observer = UserBotLatencyObserver()
# Create identity filter
processor = IdentityFilter()
# Capture latency events
latencies = []
@observer.event_handler("on_latency_measured")
async def on_latency(obs, latency_seconds):
latencies.append(latency_seconds)
# Define frame sequence - bot starts without user stop
frames_to_send = [
BotStartedSpeakingFrame(),
]
expected_down_frames = [
BotStartedSpeakingFrame,
]
# Run test
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[observer],
)
# Verify no latency was measured
self.assertEqual(len(latencies), 0)
if __name__ == "__main__":
unittest.main()