Add function call latency tracking to LatencyBreakdown

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.

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

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 = []

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()