Overview

In conversational applications, it’s important to handle situations where users go silent or inactive. Pipecat’s UserIdleProcessor helps you detect when users haven’t spoken for a defined period, allowing your bot to respond appropriately.

How It Works

The UserIdleProcessor monitors user activity and:

  1. Starts tracking after the first interaction (user or bot speaking)
  2. Resets a timer whenever the user speaks
  3. Calls your custom callback function when the user is idle for longer than the timeout period
  4. Provides a retry counter to track consecutive idle periods
  5. Allows you to implement escalating responses or gracefully end the conversation

The processor uses speech events (not audio frames) to detect activity, so it requires an active speech-to-text service or a transport with built-in speech detection.

Basic Implementation

Step 1: Create a Handler Function

First, create a callback function that will be triggered when the user is idle:

# Simple handler that doesn't use retry count
async def handle_user_idle(processor):
    # Send a reminder to the user
    await processor.push_frame(TTSSpeakFrame("Are you still there?"))

# OR

# Advanced handler with retry logic
async def handle_user_idle(processor, retry_count):
    if retry_count == 1:
        # First attempt - gentle reminder
        await processor.push_frame(TTSSpeakFrame("Are you still there?"))
        return True  # Continue monitoring
    elif retry_count == 2:
        # Second attempt - more direct prompt
        await processor.push_frame(TTSSpeakFrame("Would you like to continue our conversation?"))
        return True  # Continue monitoring
    else:
        # Third attempt - end conversation
        await processor.push_frame(TTSSpeakFrame("I'll leave you for now. Have a nice day!"))
        await processor.push_frame(EndFrame(), FrameDirection.UPSTREAM)
        return False  # Stop monitoring

Step 2: Create the Idle Processor

Initialize the processor with your callback and desired timeout:

from pipecat.processors.user_idle_processor import UserIdleProcessor

user_idle = UserIdleProcessor(
    callback=handle_user_idle,  # Your callback function
    timeout=5.0,               # Seconds of inactivity before triggering
)

Step 3: Add to Your Pipeline

Place the processor after speech detection but before context handling:

pipeline = Pipeline(
    [
        transport.input(),
        stt,                         # Speech-to-text
        user_idle,                   # Add idle detection here
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ]
)

Best Practices

  • Set appropriate timeouts: Shorter timeouts (5-10 seconds) work well for voice conversations
  • Use escalating responses: Start with gentle reminders and gradually become more direct
  • Limit retry attempts: After 2-3 unsuccessful attempts, consider ending the conversation gracefully by pushing an EndFrame

Next Steps

Implementing idle user detection improves the conversational experience by ensuring your bot can handle periods of user inactivity gracefully, either by prompting for re-engagement or politely ending the conversation when appropriate.