Pipeline Idle Detection

Automatically detect and handle idle pipelines with no bot activity

Overview

Pipeline idle detection monitors activity in your pipeline and can automatically cancel tasks when no meaningful interactions are occurring. This helps prevent pipelines from running indefinitely when a conversation has naturally ended but wasn’t properly terminated.

How It Works

The system monitors specific “activity frames” that indicate the bot is actively engaged in the conversation. By default, these are:

• BotSpeakingFrame - When the bot is speaking
• UserSpeakingFrame - When the user is speaking

If no activity frames are detected within the configured timeout period (5 minutes by default), the system considers the pipeline idle and can automatically terminate it.

Idle detection only starts after the pipeline has begun processing frames. The idle timer resets whenever an activity frame (as specified in idle_timeout_frames) is received.

Configuration

You can configure idle detection behavior when creating a PipelineWorker:

from pipecat.pipeline.worker import PipelineParams, PipelineWorker

# Default configuration - cancel after 5 minutes of inactivity
task = PipelineWorker(pipeline)

# Custom configuration
task = PipelineWorker(
    pipeline,
    idle_timeout_secs=600,  # 10 minute timeout
    idle_timeout_frames=(BotSpeakingFrame,),  # Only monitor bot speaking
    cancel_on_idle_timeout=False,  # Don't auto-cancel, just notify
)

Configuration Parameters

idle_timeout_secs · float | None · default: "300"

Timeout in seconds before considering the pipeline idle. Set to None to disable idle detection.

idle_timeout_frames · Tuple[Type[Frame], ...] · default: "(BotSpeakingFrame, UserSpeakingFrame)"

Frame types that should prevent the pipeline from being considered idle.

cancel_on_idle_timeout · bool · default: "True"

Whether to automatically cancel the pipeline task when idle timeout is reached.

Handling Idle Timeouts

You can respond to idle timeout events by adding an event handler:

@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
    logger.info("Pipeline has been idle for too long")
    # Perform any custom cleanup or logging
    # Note: If cancel_on_idle_timeout=True, the pipeline will be cancelled after this handler runs

Example Implementation

Here’s a complete example showing how to configure idle detection with custom handling:

from pipecat.frames.frames import EndFrame, TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.workers.runner import WorkerRunner
from pipecat.pipeline.worker import PipelineWorker

# Create pipeline
pipeline = Pipeline([...])

# Configure task with custom idle settings
task = PipelineWorker(
    pipeline,
    idle_timeout_secs=180,  # 3 minutes
    cancel_on_idle_timeout=False  # Don't auto-cancel
)

# Add event handler for idle timeout
@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
    logger.info("Conversation has been idle for 3 minutes")

    # Add a farewell message
    await task.queue_frame(TTSSpeakFrame("I haven't heard from you in a while. Goodbye!"))

    # Then end the conversation gracefully
    await task.queue_frame(EndFrame())

runner = WorkerRunner()

await runner.run(task)