Overview
Pipeline idle detection is a feature that monitors activity in your pipeline and can automatically cancel tasks when no meaningful bot 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
LLMFullResponseEndFrame
- When the LLM has completed a response
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 PipelineTask
:
from pipecat.pipeline.task import PipelineParams, PipelineTask
# Default configuration - cancel after 5 minutes of inactivity
task = PipelineTask(pipeline)
# Custom configuration
task = PipelineTask(
pipeline,
params=PipelineParams(allow_interruptions=True),
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
Optional[float]
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, LLMFullResponseEndFrame)"
Frame types that should prevent the pipeline from being considered idle.
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 BotSpeakingFrame, LLMFullResponseEndFrame, TTSSpeakFrame
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
# Create pipeline
pipeline = Pipeline([...])
# Configure task with custom idle settings
task = PipelineTask(
pipeline,
params=PipelineParams(allow_interruptions=True),
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.stop_when_done()
runner = PipelineRunner()
await runner.run(task)