diff --git a/changelog/3885.added.md b/changelog/3885.added.md index 87cbd7824..96f8cc2cd 100644 --- a/changelog/3885.added.md +++ b/changelog/3885.added.md @@ -1 +1 @@ -- Added `on_latency_breakdown` event to `UserBotLatencyObserver` providing per-service TTFB, text aggregation, and user turn duration metrics for each user-to-bot response cycle. +- Added `on_latency_breakdown` event to `UserBotLatencyObserver` providing per-service TTFB, text aggregation, user turn duration, and function call latency metrics for each user-to-bot response cycle. diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 2c8ec1f84..8bec9e2bc 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -5,11 +5,14 @@ # +import asyncio import os from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame from pipecat.observers.startup_timing_observer import StartupTimingObserver @@ -26,6 +29,7 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -33,6 +37,17 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) + +async def fetch_weather_from_api(params: FunctionCallParams): + await asyncio.sleep(0.25) + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await asyncio.sleep(0.1) + await params.result_callback({"name": "The Golden Dragon"}) + + # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { @@ -63,6 +78,38 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + messages = [ { "role": "system", @@ -70,7 +117,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = LLMContext(messages) + context = LLMContext(messages, tools) user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), @@ -147,9 +194,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): stt_note = f" (STT: {stt_ttfb.value:.3f}s)" if stt_ttfb else "" logger.info(f" User turn: {breakdown.user_turn_secs:.3f}s{stt_note}") - for ttfb in breakdown.ttfb: - if ttfb is not stt_ttfb: - logger.info(f" {ttfb.processor}: TTFB {ttfb.value:.3f}s") + # Show non-STT TTFBs, inserting function calls after the first + # LLM TTFB (which triggered the calls) for a chronological waterfall. + non_stt = [t for t in breakdown.ttfb if t is not stt_ttfb] + fc_shown = False + for ttfb in non_stt: + logger.info(f" {ttfb.processor}: TTFB {ttfb.value:.3f}s") + if not fc_shown and breakdown.function_calls: + for fc in breakdown.function_calls: + logger.info(f" {fc.function_name}: {fc.duration_secs:.3f}s") + fc_shown = True if breakdown.text_aggregation: ta = breakdown.text_aggregation diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index cdc7c43d0..46dbbd0f1 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -15,11 +15,13 @@ is measured. Optionally collects per-service latency breakdown metrics import time from collections import deque from dataclasses import dataclass, field -from typing import List, Optional +from typing import Dict, List, Optional from pipecat.frames.frames import ( BotStartedSpeakingFrame, ClientConnectedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -34,6 +36,19 @@ from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.processors.frame_processor import FrameDirection +@dataclass +class FunctionCallMetrics: + """Latency for a single function call execution. + + Parameters: + function_name: Name of the function that was called. + duration_secs: Time in seconds from execution start to result. + """ + + function_name: str + duration_secs: float + + @dataclass class LatencyBreakdown: """Per-service latency breakdown for a single user-to-bot cycle. @@ -52,11 +67,14 @@ class LatencyBreakdown: 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. """ ttfb: List[TTFBMetricsData] = field(default_factory=list) text_aggregation: Optional[TextAggregationMetricsData] = None user_turn_secs: Optional[float] = None + function_calls: List[FunctionCallMetrics] = field(default_factory=list) class UserBotLatencyObserver(BaseObserver): @@ -113,6 +131,8 @@ class UserBotLatencyObserver(BaseObserver): # Per-cycle metric accumulators self._ttfb: List[TTFBMetricsData] = [] self._text_aggregation: Optional[TextAggregationMetricsData] = None + self._function_call_starts: Dict[str, tuple[str, float]] = {} + self._function_call_metrics: List[FunctionCallMetrics] = [] self._register_event_handler("on_latency_measured") self._register_event_handler("on_latency_breakdown") @@ -171,6 +191,21 @@ class UserBotLatencyObserver(BaseObserver): elif isinstance(data.frame, InterruptionFrame): # Discard stale metrics from cancelled LLM/TTS cycles self._reset_accumulators() + elif isinstance(data.frame, FunctionCallInProgressFrame): + self._function_call_starts[data.frame.tool_call_id] = ( + data.frame.function_name, + time.time(), + ) + elif isinstance(data.frame, FunctionCallResultFrame): + start = self._function_call_starts.pop(data.frame.tool_call_id, None) + if start is not None: + function_name, start_time = start + self._function_call_metrics.append( + FunctionCallMetrics( + function_name=function_name, + duration_secs=time.time() - start_time, + ) + ) elif isinstance(data.frame, MetricsFrame): self._handle_metrics_frame(data.frame) elif isinstance(data.frame, BotStartedSpeakingFrame): @@ -198,6 +233,7 @@ class UserBotLatencyObserver(BaseObserver): ttfb=list(self._ttfb), text_aggregation=self._text_aggregation, user_turn_secs=self._user_turn, + function_calls=list(self._function_call_metrics), ) await self._call_event_handler("on_latency_breakdown", breakdown) self._reset_accumulators() @@ -229,3 +265,5 @@ class UserBotLatencyObserver(BaseObserver): self._ttfb = [] self._text_aggregation = None self._user_turn = None + self._function_call_starts = {} + self._function_call_metrics = [] diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index ab00bfd61..8e63647cf 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -3,6 +3,8 @@ import unittest from pipecat.frames.frames import ( BotStartedSpeakingFrame, ClientConnectedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -463,6 +465,85 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(first_speech_latencies), 0) + async def test_function_call_latency_in_breakdown(self): + """Test that function call duration appears in the latency breakdown.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + tool_call_id = "call_abc123" + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + FunctionCallInProgressFrame( + function_name="get_weather", + tool_call_id=tool_call_id, + arguments={"location": "Atlanta"}, + ), + SleepFrame(sleep=0.1), + FunctionCallResultFrame( + function_name="get_weather", + tool_call_id=tool_call_id, + arguments={"location": "Atlanta"}, + result={"temperature": "75"}, + ), + BotStartedSpeakingFrame(), + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertEqual(len(breakdowns[0].function_calls), 1) + fc = breakdowns[0].function_calls[0] + self.assertEqual(fc.function_name, "get_weather") + self.assertGreaterEqual(fc.duration_secs, 0.1) + + async def test_function_call_reset_on_interruption(self): + """Test that function call metrics are cleared on interruption.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + FunctionCallInProgressFrame( + function_name="get_weather", + tool_call_id="call_1", + arguments={}, + ), + FunctionCallResultFrame( + function_name="get_weather", + tool_call_id="call_1", + arguments={}, + result={}, + ), + InterruptionFrame(), + BotStartedSpeakingFrame(), + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertEqual(len(breakdowns[0].function_calls), 0) + if __name__ == "__main__": unittest.main()