A processor that monitors user inactivity and triggers callbacks after specified timeout periods
The UserIdleProcessor is a specialized frame processor that monitors user activity in a conversation and executes callbacks when the user becomes idle. It’s particularly useful for maintaining engagement by detecting periods of user inactivity and providing escalating responses to inactivity.
The processor starts monitoring for inactivity only after the first conversation activity (either UserStartedSpeakingFrame or BotSpeakingFrame). It manages idle state based on the following rules:
Resets idle timer when user starts or stops speaking
Pauses idle monitoring while user is speaking
Resets idle timer when bot is speaking
Stops monitoring on conversation end or cancellation
Manages a retry count for the retry callback
Stops monitoring when retry callback returns False
Here are two example showing how to use the UserIdleProcessor: one with the basic callback and one with the retry callback:
Copy
from pipecat.frames.frames import LLMMessagesFramefrom pipecat.pipeline.pipeline import Pipelinefrom pipecat.processors.user_idle_processor import UserIdleProcessorasync def handle_idle(user_idle: UserIdleProcessor) -> None: messages.append({ "role": "system", "content": "Ask the user if they are still there and try to prompt for some input." }) await user_idle.push_frame(LLMMessagesFrame(messages))# Create the processoruser_idle = UserIdleProcessor( callback=handle_idle, timeout=5.0)# Add to pipelinepipeline = Pipeline([ transport.input(), user_idle, # Add the processor to monitor user activity context_aggregator.user(), # ... rest of pipeline])
Copy
from pipecat.frames.frames import LLMMessagesFramefrom pipecat.pipeline.pipeline import Pipelinefrom pipecat.processors.user_idle_processor import UserIdleProcessorasync def handle_idle(user_idle: UserIdleProcessor) -> None: messages.append({ "role": "system", "content": "Ask the user if they are still there and try to prompt for some input." }) await user_idle.push_frame(LLMMessagesFrame(messages))# Create the processoruser_idle = UserIdleProcessor( callback=handle_idle, timeout=5.0)# Add to pipelinepipeline = Pipeline([ transport.input(), user_idle, # Add the processor to monitor user activity context_aggregator.user(), # ... rest of pipeline])
Copy
from pipecat.frames.frames import EndFrame, LLMMessagesFrame, TTSSpeakFramefrom pipecat.pipeline.pipeline import Pipelinefrom pipecat.processors.user_idle_processor import UserIdleProcessorasync def handle_user_idle(user_idle: UserIdleProcessor, retry_count: int) -> bool: if retry_count == 1: # First attempt: Gentle reminder messages.append({ "role": "system", "content": "The user has been quiet. Politely and briefly ask if they're still there." }) await user_idle.push_frame(LLMMessagesFrame(messages)) return True elif retry_count == 2: # Second attempt: Direct prompt messages.append({ "role": "system", "content": "The user is still inactive. Ask if they'd like to continue our conversation." }) await user_idle.push_frame(LLMMessagesFrame(messages)) return True else: # Third attempt: End conversation await user_idle.push_frame( TTSSpeakFrame("It seems like you're busy right now. Have a nice day!") ) await task.queue_frame(EndFrame()) return False # Stop monitoring# Create the processoruser_idle = UserIdleProcessor( callback=handle_user_idle, timeout=5.0)# Add to pipelinepipeline = Pipeline([ transport.input(), user_idle, # Add the processor to monitor user activity context_aggregator.user(), # ... rest of pipeline])
