Add callbacks to the UserIdleProcessor

examples/foundational/31-user-idle.py

@@ -0,0 +1,148 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import EndFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.user_idle_processor import UserIdleProcessor
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+# Example callback using the new style with retry control
+async def handle_user_idle(processor: UserIdleProcessor, retry_count: int) -> bool:
+    if retry_count == 1:
+        # First attempt: Add a gentle prompt to the conversation
+        messages.append(
+            {
+                "role": "system",
+                "content": "The user has been quiet for a while. Politely ask if they're still there.",
+            }
+        )
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+        return True
+    elif retry_count == 2:
+        # Second attempt: More direct prompt
+        messages.append(
+            {
+                "role": "system",
+                "content": "The user is still inactive. Ask if they would like to continue the conversation.",
+            }
+        )
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+        return True
+    else:
+        # Third attempt: End the conversation
+        messages.append(
+            {
+                "role": "system",
+                "content": "The user has been inactive for too long. Politely end the conversation.",
+            }
+        )
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+        await task.queue_frame(EndFrame())
+        return False
+
+
+async def main():
+    global task, messages, context_aggregator  # Make these accessible to the idle handler
+
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4")
+
+        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 converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        # Create the idle processor
+        idle_processor = UserIdleProcessor(
+            callback=handle_user_idle,
+            timeout=5.0,  # 5 seconds of inactivity before triggering
+        )
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                idle_processor,  # Add the idle processor
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            await task.queue_frame(EndFrame())
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/pipecat/processors/user_idle_processor.py

@@ -5,7 +5,8 @@
 #
 
 import asyncio
-from typing import Awaitable, Callable
+from functools import wraps
+from typing import Awaitable, Callable, Union
 
 from pipecat.frames.frames import (
     BotSpeakingFrame,
@@ -20,24 +21,59 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
 class UserIdleProcessor(FrameProcessor):
-    """This class is useful to check if the user is interacting with the bot
+    """This class is useful to check if the user is interacting with the bot within a given timeout.
-    within a given timeout. If the timeout is reached before any interaction
-    occurred the provided callback will be called.
 
+    If the timeout is reached before any interaction occurred the provided callback will be called.
+
+    The callback can be either:
+    - async def callback(processor: UserIdleProcessor) -> None  # Legacy
+    - async def callback(processor: UserIdleProcessor, retry_count: int) -> bool  # New
+
+    The new style callback receives the current retry count and should return True
+    to continue monitoring or False to stop.
     """
 
     def __init__(
         self,
         *,
-        callback: Callable[["UserIdleProcessor"], Awaitable[None]],
+        callback: Union[
+            Callable[["UserIdleProcessor"], Awaitable[None]],  # Old signature
+            Callable[["UserIdleProcessor", int], Awaitable[bool]],  # New signature
+        ],
         timeout: float,
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self._callback = callback
+        self._callback = self._wrap_callback(callback)
         self._timeout = timeout
+        self._retry_count = 0
         self._interrupted = False
 
+    def _wrap_callback(
+        self,
+        callback: Union[
+            Callable[["UserIdleProcessor"], Awaitable[None]],
+            Callable[["UserIdleProcessor", int], Awaitable[bool]],
+        ],
+    ) -> Callable[["UserIdleProcessor", int], Awaitable[bool]]:
+        @wraps(callback)
+        async def wrapper(processor: "UserIdleProcessor", retry_count: int) -> bool:
+            # Check callback signature
+            import inspect
+
+            sig = inspect.signature(callback)
+            param_count = len(sig.parameters)
+
+            if param_count == 1:
+                # Old callback
+                await callback(processor)  # type: ignore
+                return True  # Always continue for backwards compatibility
+            else:
+                # New callback
+                return await callback(processor, retry_count)  # type: ignore
+
+        return wrapper
+
     async def _stop(self):
         self._idle_task.cancel()
         await self._idle_task
@@ -76,7 +112,11 @@ class UserIdleProcessor(FrameProcessor):
                 await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout)
             except asyncio.TimeoutError:
                 if not self._interrupted:
-                    await self._callback(self)
+                    self._retry_count += 1
+                    should_continue = await self._callback(self, self._retry_count)
+                    if not should_continue:
+                        await self._stop()
+                        break
             except asyncio.CancelledError:
                 break
             finally: