diff --git a/examples/foundational/31-user-idle.py b/examples/foundational/31-user-idle.py new file mode 100644 index 000000000..d9dbc92e7 --- /dev/null +++ b/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()) diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 8f40fd52e..9fc2912c7 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/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 - within a given timeout. If the timeout is reached before any interaction - occurred the provided callback will be called. + """This class is useful to check if the user is interacting with the bot within a given timeout. + 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: