User-Bot Latency Observer

Measure response time between user speech and bot responses in Pipecat

The UserBotLatencyObserver measures the time between when a user stops speaking and when the bot starts responding, emitting events for custom handling and optional OpenTelemetry tracing integration. It also tracks first-bot-speech latency and provides detailed latency breakdowns when metrics are enabled, including per-service metrics and contributions that name every part of the interval.

Features

• Tracks user speech start/stop timing using VAD frames
• Measures bot response latency from the actual moment the user started speaking
• Measures first bot speech latency (client connection to first speech)
• Provides detailed latency breakdown with per-service TTFB, text aggregation, user turn duration, and function call metrics
• Provides contributions timeline that names every part of the user-to-bot interval, including time no service measures
• Emits on_latency_measured events for custom processing
• Emits on_latency_breakdown events with detailed per-service metrics
• Emits on_first_bot_speech_latency event for greeting latency measurement
• Automatically records latency as OpenTelemetry span attributes when tracing is enabled
• Automatically resets between conversation turns

Usage

Basic Latency Monitoring

Add latency monitoring to your pipeline and handle the event:

from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver

latency_observer = UserBotLatencyObserver()

@latency_observer.event_handler("on_latency_measured")
async def on_latency_measured(observer, latency):
    print(f"User-to-bot latency: {latency:.3f}s")

worker = PipelineWorker(
    pipeline,
    observers=[latency_observer],
)

Detailed Latency Breakdown

Enable metrics to collect per-service latency breakdown:

from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver

latency_observer = UserBotLatencyObserver()

@latency_observer.event_handler("on_latency_breakdown")
async def on_latency_breakdown(observer, breakdown):
    # Each line names one part of the turn and they sum to the total latency
    for line in breakdown.turn_contribution_lines():
        print(f"  {line}")

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(
        enable_metrics=True,  # Required for breakdown metrics
    ),
    observers=[latency_observer],
)

OpenTelemetry Integration

When tracing is enabled, latency measurements are automatically recorded as turn.user_bot_latency_seconds attributes on OpenTelemetry turn spans. No additional configuration is needed.

How It Works

The observer tracks conversation flow through these key events:

1. Client connects (ClientConnectedFrame) → Records timestamp for first-bot-speech measurement
2. User starts speaking (VADUserStartedSpeakingFrame) → Resets latency tracking
3. User stops speaking (VADUserStoppedSpeakingFrame) → Records timestamp, accounting for VAD stop_secs delay
4. Bot starts speaking (BotStartedSpeakingFrame) → Calculates latency and emits on_latency_measured and on_latency_breakdown events

When enable_metrics=True in PipelineParams, the observer also collects per-service metrics (TTFB, text aggregation, function call latency) from MetricsFrame instances and builds a contributions timeline that names every part of the user-to-bot interval, including the parts no service measures (VAD silence, turn detection, turn-completion markers).

Event Handlers

on_latency_measured

Called each time a user-to-bot latency measurement is captured.

@latency_observer.event_handler("on_latency_measured")
async def on_latency_measured(observer, latency):
    # latency is a float representing seconds
    logger.info(f"Response latency: {latency:.3f}s")

on_latency_breakdown

Called alongside on_latency_measured with detailed per-service metrics collected during the user→bot cycle. The breakdown includes TTFB from each service, text aggregation latency, user turn duration, and function call timings.

@latency_observer.event_handler("on_latency_breakdown")
async def on_latency_breakdown(observer, breakdown):
    # breakdown is a LatencyBreakdown object
    for line in breakdown.turn_contribution_lines():
        logger.info(f"  {line}")
    # Output:
    #   0.200s  endpointing wait     [config: VAD stop_secs]
    #   0.125s  transcription        [DeepgramSTTService#0]
    #   0.336s  LLM inference        [OpenAILLMService#0]
    #   0.359s  speech synthesis     [CartesiaTTSService#0]
    #   1.020s  TOTAL

LatencyBreakdown fields:

Field	Type	Description
contributions	List[LatencyContribution]	Named parts of the interval in chronological order, summing to the measured latency
measured_from	Optional[MeasuredFrom]	Where the interval was anchored: USER_SILENCE for turns, CLIENT_CONNECTED for greetings
total_secs	float	The measured interval that the contributions sum to
ttfb	List[TTFBBreakdownMetrics]	Time-to-first-byte metrics from each service
text_aggregation	Optional[TextAggregationBreakdownMetrics]	First text aggregation measurement (sentence aggregation latency)
user_turn_start_time	Optional[float]	Unix timestamp when user turn started (adjusted for VAD stop_secs)
user_turn_secs	Optional[float]	User turn duration including VAD silence detection, STT finalization, and turn analyzer wait
function_calls	List[FunctionCallMetrics]	Latency for each function call executed during the cycle

LatencyContribution fields:

Each contribution names one part of the user-to-bot interval:

Field	Type	Description
key	str	Stable identifier for this part (safe to group on, survives label rewording)
label	str	What the time was spent on
owner	str	What spent it — a processor name or a config: tag naming the setting that governs it
owner_kind	LatencyOwnerKind	What kind of thing the owner is: SERVICE, SETTING, BOT, or PIPELINE
start_time	float	Unix timestamp when it started
duration_secs	float	How long it took in seconds

Methods:

• turn_contribution_lines(by_cost=False) — Format contributions for logging, one per line plus a total. Pass by_cost=True to order by duration rather than chronologically.
• chronological_events() — Deprecated since 1.9.0. Use turn_contribution_lines() instead, which names every part of the interval. Will be removed in 2.0.0.

on_first_bot_speech_latency

Called once when the bot first speaks after client connection. Measures the time from ClientConnectedFrame to the first BotStartedSpeakingFrame. This is particularly useful for measuring greeting latency.

@latency_observer.event_handler("on_first_bot_speech_latency")
async def on_first_bot_speech_latency(observer, latency):
    logger.info(f"First bot speech latency: {latency:.3f}s")

The on_latency_breakdown event is also emitted for the first bot speech, allowing you to see the detailed breakdown of what contributed to the greeting latency.

Configuration

Constructor Parameters

max_frames · int · default: "100"

Maximum number of frame IDs to keep in history for duplicate detection. Prevents unbounded memory growth in long conversations.

min_contribution_secs · float · default: "0.005"

Contributions shorter than this are rolled into the single pipeline entry rather than listed separately. Pass 0 to list every contribution, including individual frame hops.

time_source · Callable[[], float] · default: "time.time"

Reads the current time in seconds. Supplying one lets a test drive a cycle without waiting out the intervals it describes.

Limitations

• Requires proper frame sequencing to work accurately
• Per-service metrics are only collected when enable_metrics=True in PipelineParams