Redesign UserIdleController to use BotStoppedSpeakingFrame

Replace the continuous heartbeat-based timer (UserSpeakingFrame/BotSpeakingFrame
+ asyncio.Event loop) with a simple one-shot timer that starts when
BotStoppedSpeakingFrame is received and cancels on UserStartedSpeakingFrame or
BotStartedSpeakingFrame. This eliminates false idle triggers caused by gaps
between the user finishing speaking and the bot starting to speak (LLM/TTS
latency).

Guard the timer start with two conditions to prevent false triggers:
- User turn in progress: during interruptions, BotStoppedSpeaking arrives
  while the user is still speaking mid-turn.
- Function calls in progress: FunctionCallsStarted arrives before
  BotStoppedSpeaking because the bot speaks concurrently with the function
  call starting, so the timer must wait for the result and subsequent bot
  response.

Author: markbackman

examples/foundational/17-detect-user-idle.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 (
     EndTaskFrame,
@@ -30,6 +33,7 @@
 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
@@ -74,6 +78,17 @@ async def handle_idle(self, aggregator):
             await aggregator.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
 
 
+async def fetch_weather_from_api(params: FunctionCallParams):
+    # Simulate a slow API call, waiting longer than the user idle timeout.
+    await asyncio.sleep(3)
+    await params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+async def fetch_restaurant_recommendation(params: FunctionCallParams):
+    await asyncio.sleep(6)
+    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 = {
@@ -104,14 +119,50 @@ 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)
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
+
+    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",
             "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
         },
     ]
 
-    context = LLMContext(messages)
+    context = LLMContext(messages, tools)
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
         context,
         user_params=LLMUserAggregatorParams(
@@ -146,6 +197,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     @user_aggregator.event_handler("on_user_turn_idle")
     async def on_user_turn_idle(aggregator):
+        logger.info(f"User turn idle")
         await idle_handler.handle_idle(aggregator)
 
     @user_aggregator.event_handler("on_user_turn_started")

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -689,6 +689,9 @@ async def _on_user_turn_started(
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStartedSpeakingFrame)
 
+        if self._user_idle_controller:
+            await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
+
         if params.enable_interruptions and self._allow_interruptions:
             await self.push_interruption_task_frame_and_wait()
 
@@ -705,6 +708,9 @@ async def _on_user_turn_stopped(
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStoppedSpeakingFrame)
 
+        if self._user_idle_controller:
+            await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
+
         await self._maybe_emit_user_turn_stopped(strategy)
 
     async def _on_user_turn_stop_timeout(self, controller):

src/pipecat/turns/user_idle_controller.py

@@ -10,12 +10,14 @@
 from typing import Optional
 
 from pipecat.frames.frames import (
-    BotSpeakingFrame,
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
     Frame,
+    FunctionCallCancelFrame,
     FunctionCallResultFrame,
     FunctionCallsStartedFrame,
-    UserSpeakingFrame,
     UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
 )
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
@@ -25,14 +27,14 @@ class UserIdleController(BaseObject):
     """Controller for managing user idle detection.
 
     This class monitors user activity and triggers an event when the user has been
-    idle (not speaking) for a configured timeout period. It only starts monitoring
-    after the first conversation activity and does not trigger while the bot is
-    speaking or function calls are in progress.
+    idle (not speaking) for a configured timeout period after the bot finishes
+    speaking. The timer starts when BotStoppedSpeakingFrame is received and is
+    cancelled when someone starts speaking again (UserStartedSpeakingFrame or
+    BotStartedSpeakingFrame).
 
-    The controller tracks activity using continuous frames (UserSpeakingFrame and
-    BotSpeakingFrame) which are emitted repeatedly while speaking is happening, and
-    state-based tracking for function calls (FunctionCallsStartedFrame and
-    FunctionCallResultFrame) which are only sent at start and end.
+    The timer is suppressed while a user turn is in progress to avoid false
+    triggers during interruptions (where BotStoppedSpeakingFrame arrives while
+    the user is still speaking).
 
     Event handlers available:
 
@@ -62,11 +64,9 @@ def __init__(
 
         self._task_manager: Optional[BaseTaskManager] = None
 
-        self._conversation_started = False
-        self._function_call_in_progress = False
-
-        self.user_idle_event = asyncio.Event()
-        self.user_idle_task: Optional[asyncio.Task] = None
+        self._user_turn_in_progress: bool = False
+        self._function_calls_in_progress: int = 0
+        self._idle_timer_task: Optional[asyncio.Task] = None
 
         self._register_event_handler("on_user_turn_idle", sync=True)
 
@@ -85,89 +85,63 @@ async def setup(self, task_manager: BaseTaskManager):
         """
         self._task_manager = task_manager
 
-        if not self.user_idle_task:
-            self.user_idle_task = self.task_manager.create_task(
-                self.user_idle_task_handler(),
-                f"{self}::user_idle_task_handler",
-            )
-
     async def cleanup(self):
         """Cleanup the controller."""
         await super().cleanup()
-
-        if self.user_idle_task:
-            await self.task_manager.cancel_task(self.user_idle_task)
-            self.user_idle_task = None
+        await self._cancel_idle_timer()
 
     async def process_frame(self, frame: Frame):
         """Process an incoming frame to track user activity state.
 
         Args:
             frame: The frame to be processed.
         """
-        # Start monitoring on first conversation activity
-        if not self._conversation_started:
-            if isinstance(frame, (UserStartedSpeakingFrame, BotSpeakingFrame)):
-                self._conversation_started = True
-                self.user_idle_event.set()
-            else:
-                return
-
-        # Reset idle timer on continuous activity frames
-        if isinstance(frame, (UserSpeakingFrame, BotSpeakingFrame)):
-            await self._handle_activity(frame)
-        # Track function call state (start/end frames, not continuous)
+        if isinstance(frame, BotStoppedSpeakingFrame):
+            # Only start the timer if the user isn't mid-turn and no function
+            # calls are pending.
+            #
+            # Interruption case: the frame order is UserStartedSpeaking →
+            # BotStoppedSpeaking → (user keeps talking) → UserStoppedSpeaking.
+            # Without the user-turn guard the timer would start while the user
+            # is still speaking.
+            #
+            # Function call case: normally FunctionCallsStarted arrives after
+            # BotStoppedSpeaking and cancels the timer directly. But a race
+            # condition can cause FunctionCallsStarted to arrive before
+            # BotStoppedSpeaking when pushing a TTSSpeakFrame in the
+            # on_function_calls_started event handler, so the counter guard
+            # prevents the timer from starting while a function call is in progress.
+            if not self._user_turn_in_progress and self._function_calls_in_progress == 0:
+                await self._start_idle_timer()
+        elif isinstance(frame, BotStartedSpeakingFrame):
+            await self._cancel_idle_timer()
+        elif isinstance(frame, UserStartedSpeakingFrame):
+            self._user_turn_in_progress = True
+            await self._cancel_idle_timer()
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self._user_turn_in_progress = False
         elif isinstance(frame, FunctionCallsStartedFrame):
-            await self._handle_function_calls_started(frame)
-        elif isinstance(frame, FunctionCallResultFrame):
-            await self._handle_function_call_result(frame)
-
-    async def _handle_activity(self, _: UserSpeakingFrame | BotSpeakingFrame):
-        """Handle continuous activity frames that should reset the idle timer.
-
-        These frames are emitted continuously while the user or bot is speaking,
-        so we simply reset the timer whenever we receive them.
-
-        Args:
-            frame: The activity frame to process.
-        """
-        self.user_idle_event.set()
-
-    async def _handle_function_calls_started(self, _: FunctionCallsStartedFrame):
-        """Handle function calls started event.
-
-        Function calls can take longer than the timeout, so we track their state
-        to prevent idle callbacks while they're in progress.
-
-        Args:
-            frame: The FunctionCallsStartedFrame to process.
-        """
-        self._function_call_in_progress = True
-        self.user_idle_event.set()
-
-    async def _handle_function_call_result(self, _: FunctionCallResultFrame):
-        """Handle function call result event.
-
-        Args:
-            frame: The FunctionCallResultFrame to process.
-        """
-        self._function_call_in_progress = False
-        self.user_idle_event.set()
-
-    async def user_idle_task_handler(self):
-        """Monitors for idle timeout and triggers events.
-
-        Runs in a loop until cancelled. The idle timer is reset whenever activity
-        frames are received (UserSpeakingFrame or BotSpeakingFrame). Function calls
-        are tracked via state since they only send start/end frames. If no activity
-        is detected for the configured timeout period and no function call is in
-        progress, the on_user_turn_idle event is triggered.
-        """
-        while True:
-            try:
-                await asyncio.wait_for(self.user_idle_event.wait(), timeout=self._user_idle_timeout)
-                self.user_idle_event.clear()
-            except asyncio.TimeoutError:
-                # Only trigger if conversation has started and no function call is in progress
-                if self._conversation_started and not self._function_call_in_progress:
-                    await self._call_event_handler("on_user_turn_idle")
+            self._function_calls_in_progress += len(frame.function_calls)
+            await self._cancel_idle_timer()
+        elif isinstance(frame, (FunctionCallResultFrame, FunctionCallCancelFrame)):
+            self._function_calls_in_progress = max(0, self._function_calls_in_progress - 1)
+
+    async def _start_idle_timer(self):
+        """Start (or restart) the idle timer."""
+        await self._cancel_idle_timer()
+        self._idle_timer_task = self.task_manager.create_task(
+            self._idle_timer_expired(),
+            f"{self}::idle_timer",
+        )
+
+    async def _cancel_idle_timer(self):
+        """Cancel the idle timer if running."""
+        if self._idle_timer_task:
+            await self.task_manager.cancel_task(self._idle_timer_task)
+            self._idle_timer_task = None
+
+    async def _idle_timer_expired(self):
+        """Sleep for the timeout duration then fire the idle event."""
+        await asyncio.sleep(self._user_idle_timeout)
+        self._idle_timer_task = None
+        await self._call_event_handler("on_user_turn_idle")

src/pipecat/turns/user_turn_processor.py

@@ -189,6 +189,9 @@ async def _on_user_turn_started(
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStartedSpeakingFrame)
 
+        if self._user_idle_controller:
+            await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
+
         if params.enable_interruptions and self._allow_interruptions:
             await self.push_interruption_task_frame_and_wait()
 
@@ -205,6 +208,9 @@ async def _on_user_turn_stopped(
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStoppedSpeakingFrame)
 
+        if self._user_idle_controller:
+            await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
+
         await self._call_event_handler("on_user_turn_stopped", strategy)
 
     async def _on_user_turn_stop_timeout(self, controller):