wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3744 from pipecat-ai/mb/user-idle-timeout-frame

Browse Source 
Redesign UserIdleController to use BotStoppedSpeakingFrame
This commit is contained in:
Aleix Conchillo Flaqué
2026-02-19 11:34:42 -08:00
committed by GitHub
parent 4f8cacc769 cb7023681f
commit 20509e8f96
6 changed files with 278 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3744.fixed.md Normal file
View File

@@ -0,0 +1 @@
- Fixed `UserIdleController` false idle triggers caused by gaps between user and bot activity frames. The idle timer now starts only after `BotStoppedSpeakingFrame` and is suppressed during active user turns and function calls.

54
examples/foundational/17-detect-user-idle.py
View File

@@ -5,11 +5,14 @@
#
import asyncio
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import (
EndTaskFrame,
@@ -30,6 +33,7 @@ from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -74,6 +78,17 @@ class IdleHandler:
await aggregator.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
async def fetch_weather_from_api(params: FunctionCallParams):
# Simulate a slow API call, waiting longer than the user idle timeout.
await asyncio.sleep(3)
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await asyncio.sleep(6)
await params.result_callback({"name": "The Golden Dragon"})
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
@@ -104,6 +119,42 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
messages = [
{
"role": "system",
@@ -111,7 +162,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
},
]
context = LLMContext(messages)
context = LLMContext(messages, tools)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(
@@ -146,6 +197,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@user_aggregator.event_handler("on_user_turn_idle")
async def on_user_turn_idle(aggregator):
logger.info(f"User turn idle")
await idle_handler.handle_idle(aggregator)
@user_aggregator.event_handler("on_user_turn_started")

6
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -696,6 +696,9 @@ class LLMUserAggregator(LLMContextAggregator):
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStartedSpeakingFrame)
if self._user_idle_controller:
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
if params.enable_interruptions and self._allow_interruptions:
await self.push_interruption_task_frame_and_wait()
@@ -712,6 +715,9 @@ class LLMUserAggregator(LLMContextAggregator):
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStoppedSpeakingFrame)
if self._user_idle_controller:
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
await self._maybe_emit_user_turn_stopped(strategy)
async def _on_user_turn_stop_timeout(self, controller):

146
src/pipecat/turns/user_idle_controller.py
View File

@@ -10,12 +10,14 @@ import asyncio
from typing import Optional
from pipecat.frames.frames import (
BotSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
UserSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
@@ -25,14 +27,14 @@ class UserIdleController(BaseObject):
"""Controller for managing user idle detection.
This class monitors user activity and triggers an event when the user has been
idle (not speaking) for a configured timeout period. It only starts monitoring
after the first conversation activity and does not trigger while the bot is
speaking or function calls are in progress.
idle (not speaking) for a configured timeout period after the bot finishes
speaking. The timer starts when BotStoppedSpeakingFrame is received and is
cancelled when someone starts speaking again (UserStartedSpeakingFrame or
BotStartedSpeakingFrame).
The controller tracks activity using continuous frames (UserSpeakingFrame and
BotSpeakingFrame) which are emitted repeatedly while speaking is happening, and
state-based tracking for function calls (FunctionCallsStartedFrame and
FunctionCallResultFrame) which are only sent at start and end.
The timer is suppressed while a user turn is in progress to avoid false
triggers during interruptions (where BotStoppedSpeakingFrame arrives while
the user is still speaking).
Event handlers available:
@@ -62,11 +64,9 @@ class UserIdleController(BaseObject):
self._task_manager: Optional[BaseTaskManager] = None
self._conversation_started = False
self._function_call_in_progress = False
self.user_idle_event = asyncio.Event()
self.user_idle_task: Optional[asyncio.Task] = None
self._user_turn_in_progress: bool = False
self._function_calls_in_progress: int = 0
self._idle_timer_task: Optional[asyncio.Task] = None
self._register_event_handler("on_user_turn_idle", sync=True)
@@ -85,19 +85,10 @@ class UserIdleController(BaseObject):
"""
self._task_manager = task_manager
if not self.user_idle_task:
self.user_idle_task = self.task_manager.create_task(
self.user_idle_task_handler(),
f"{self}::user_idle_task_handler",
)
async def cleanup(self):
"""Cleanup the controller."""
await super().cleanup()
if self.user_idle_task:
await self.task_manager.cancel_task(self.user_idle_task)
self.user_idle_task = None
await self._cancel_idle_timer()
async def process_frame(self, frame: Frame):
"""Process an incoming frame to track user activity state.
@@ -105,69 +96,52 @@ class UserIdleController(BaseObject):
Args:
frame: The frame to be processed.
"""
# Start monitoring on first conversation activity
if not self._conversation_started:
if isinstance(frame, (UserStartedSpeakingFrame, BotSpeakingFrame)):
self._conversation_started = True
self.user_idle_event.set()
else:
return
# Reset idle timer on continuous activity frames
if isinstance(frame, (UserSpeakingFrame, BotSpeakingFrame)):
await self._handle_activity(frame)
# Track function call state (start/end frames, not continuous)
if isinstance(frame, BotStoppedSpeakingFrame):
# Only start the timer if the user isn't mid-turn and no function
# calls are pending.
#
# Interruption case: the frame order is UserStartedSpeaking →
# BotStoppedSpeaking → (user keeps talking) → UserStoppedSpeaking.
# Without the user-turn guard the timer would start while the user
# is still speaking.
#
# Function call case: normally FunctionCallsStarted arrives after
# BotStoppedSpeaking and cancels the timer directly. But a race
# condition can cause FunctionCallsStarted to arrive before
# BotStoppedSpeaking when pushing a TTSSpeakFrame in the
# on_function_calls_started event handler, so the counter guard
# prevents the timer from starting while a function call is in progress.
if not self._user_turn_in_progress and self._function_calls_in_progress == 0:
await self._start_idle_timer()
elif isinstance(frame, BotStartedSpeakingFrame):
await self._cancel_idle_timer()
elif isinstance(frame, UserStartedSpeakingFrame):
self._user_turn_in_progress = True
await self._cancel_idle_timer()
elif isinstance(frame, UserStoppedSpeakingFrame):
self._user_turn_in_progress = False
elif isinstance(frame, FunctionCallsStartedFrame):
await self._handle_function_calls_started(frame)
elif isinstance(frame, FunctionCallResultFrame):
await self._handle_function_call_result(frame)
self._function_calls_in_progress += len(frame.function_calls)
await self._cancel_idle_timer()
elif isinstance(frame, (FunctionCallResultFrame, FunctionCallCancelFrame)):
self._function_calls_in_progress = max(0, self._function_calls_in_progress - 1)
async def _handle_activity(self, _: UserSpeakingFrame | BotSpeakingFrame):
"""Handle continuous activity frames that should reset the idle timer.
async def _start_idle_timer(self):
"""Start (or restart) the idle timer."""
await self._cancel_idle_timer()
self._idle_timer_task = self.task_manager.create_task(
self._idle_timer_expired(),
f"{self}::idle_timer",
)
These frames are emitted continuously while the user or bot is speaking,
so we simply reset the timer whenever we receive them.
async def _cancel_idle_timer(self):
"""Cancel the idle timer if running."""
if self._idle_timer_task:
await self.task_manager.cancel_task(self._idle_timer_task)
self._idle_timer_task = None
Args:
frame: The activity frame to process.
"""
self.user_idle_event.set()
async def _handle_function_calls_started(self, _: FunctionCallsStartedFrame):
"""Handle function calls started event.
Function calls can take longer than the timeout, so we track their state
to prevent idle callbacks while they're in progress.
Args:
frame: The FunctionCallsStartedFrame to process.
"""
self._function_call_in_progress = True
self.user_idle_event.set()
async def _handle_function_call_result(self, _: FunctionCallResultFrame):
"""Handle function call result event.
Args:
frame: The FunctionCallResultFrame to process.
"""
self._function_call_in_progress = False
self.user_idle_event.set()
async def user_idle_task_handler(self):
"""Monitors for idle timeout and triggers events.
Runs in a loop until cancelled. The idle timer is reset whenever activity
frames are received (UserSpeakingFrame or BotSpeakingFrame). Function calls
are tracked via state since they only send start/end frames. If no activity
is detected for the configured timeout period and no function call is in
progress, the on_user_turn_idle event is triggered.
"""
while True:
try:
await asyncio.wait_for(self.user_idle_event.wait(), timeout=self._user_idle_timeout)
self.user_idle_event.clear()
except asyncio.TimeoutError:
# Only trigger if conversation has started and no function call is in progress
if self._conversation_started and not self._function_call_in_progress:
await self._call_event_handler("on_user_turn_idle")
async def _idle_timer_expired(self):
"""Sleep for the timeout duration then fire the idle event."""
await asyncio.sleep(self._user_idle_timeout)
self._idle_timer_task = None
await self._call_event_handler("on_user_turn_idle")

6
src/pipecat/turns/user_turn_processor.py
View File

@@ -189,6 +189,9 @@ class UserTurnProcessor(FrameProcessor):
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStartedSpeakingFrame)
if self._user_idle_controller:
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
if params.enable_interruptions and self._allow_interruptions:
await self.push_interruption_task_frame_and_wait()
@@ -205,6 +208,9 @@ class UserTurnProcessor(FrameProcessor):
if params.enable_user_speaking_frames:
await self.broadcast_frame(UserStoppedSpeakingFrame)
if self._user_idle_controller:
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
await self._call_event_handler("on_user_turn_stopped", strategy)
async def _on_user_turn_stop_timeout(self, controller):

272
tests/test_user_idle_controller.py
View File

@@ -6,12 +6,13 @@
import asyncio
import unittest
import unittest.mock
from pipecat.frames.frames import (
BotSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
UserSpeakingFrame,
UserStartedSpeakingFrame,
)
from pipecat.turns.user_idle_controller import UserIdleController
@@ -25,8 +26,8 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
self.task_manager = TaskManager()
self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
async def test_basic_idle_detection(self):
"""Test that idle event is triggered after timeout when no activity."""
async def test_idle_after_bot_stops_speaking(self):
"""Test that idle event fires after BotStoppedSpeakingFrame + timeout."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
@@ -37,18 +38,16 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
nonlocal idle_triggered
idle_triggered = True
# Start conversation
await controller.process_frame(UserStartedSpeakingFrame())
await controller.process_frame(BotStoppedSpeakingFrame())
# Wait for idle timeout
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertTrue(idle_triggered)
await controller.cleanup()
async def test_user_speaking_resets_idle_timer(self):
"""Test that continuous UserSpeakingFrame frames reset the idle timer."""
async def test_user_speaking_cancels_timer(self):
"""Test that UserStartedSpeakingFrame cancels the idle timer."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
@@ -59,20 +58,18 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
nonlocal idle_triggered
idle_triggered = True
# Start conversation
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
await controller.process_frame(UserStartedSpeakingFrame())
# Send UserSpeakingFrame continuously to reset timer
for _ in range(5):
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.5) # 50% of timeout period
await controller.process_frame(UserSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_bot_speaking_resets_idle_timer(self):
"""Test that BotSpeakingFrame frames reset the idle timer."""
async def test_bot_speaking_cancels_timer(self):
"""Test that BotStartedSpeakingFrame cancels the idle timer."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
@@ -83,102 +80,61 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
nonlocal idle_triggered
idle_triggered = True
# Start conversation
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
await controller.process_frame(BotStartedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_no_idle_before_bot_speaks(self):
"""Test that idle does not fire if no BotStoppedSpeakingFrame is received."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# Wait without any frames
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_interruption_no_false_trigger(self):
"""Test that BotStoppedSpeakingFrame during a user turn does not start the timer."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# User starts speaking (interruption)
await controller.process_frame(UserStartedSpeakingFrame())
# Bot stops speaking due to interruption
await controller.process_frame(BotStoppedSpeakingFrame())
# Bot speaking should reset timer
for _ in range(5):
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.6) # 60% of timeout
await controller.process_frame(BotSpeakingFrame())
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_function_call_prevents_idle(self):
"""Test that function calls in progress prevent idle event."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# Start conversation
await controller.process_frame(UserStartedSpeakingFrame())
# Start function call
await controller.process_frame(FunctionCallsStartedFrame(function_calls=[]))
# Wait longer than idle timeout
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
# Should not trigger idle because function call is in progress
self.assertFalse(idle_triggered)
# Complete function call
await controller.process_frame(
FunctionCallResultFrame(
function_name="test",
tool_call_id="123",
arguments={},
result=None,
run_llm=False,
)
)
# Now idle should trigger
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertTrue(idle_triggered)
await controller.cleanup()
async def test_no_idle_before_conversation_starts(self):
"""Test that idle monitoring doesn't start before first conversation activity."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# Wait without starting conversation
# Wait - timer should NOT have started because user turn is in progress
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_idle_starts_with_bot_speaking(self):
"""Test that monitoring starts with BotSpeakingFrame, not just user speech."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# Start conversation with bot speaking
await controller.process_frame(BotSpeakingFrame())
# Wait for idle timeout
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertTrue(idle_triggered)
await controller.cleanup()
async def test_multiple_idle_events(self):
"""Test that idle event can trigger multiple times."""
async def test_idle_cycle(self):
"""Test that idle fires, then can fire again after another bot speaking cycle."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
@@ -189,29 +145,105 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
nonlocal idle_count
idle_count += 1
# Start conversation
await controller.process_frame(UserStartedSpeakingFrame())
# First idle
# First cycle: bot stops → idle fires
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
first_count = idle_count
self.assertGreaterEqual(first_count, 1)
self.assertEqual(idle_count, 1)
# Second idle
# Second cycle: bot starts → bot stops → idle fires again
await controller.process_frame(BotStartedSpeakingFrame())
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
second_count = idle_count
self.assertGreater(second_count, first_count)
self.assertEqual(idle_count, 2)
# User activity resets timer
await controller.process_frame(UserSpeakingFrame())
await controller.cleanup()
# Give a moment for the timer to reset
await asyncio.sleep(0.1)
async def test_cleanup_cancels_timer(self):
"""Test that cleanup cancels a pending idle timer."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
await controller.cleanup()
# Third idle
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
third_count = idle_count
self.assertGreater(third_count, second_count)
self.assertFalse(idle_triggered)
async def test_function_call_cancels_timer(self):
"""Test normal ordering: BotStopped starts timer, FunctionCallsStarted cancels it."""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# Bot finishes speaking, timer starts
await controller.process_frame(BotStoppedSpeakingFrame())
# Function call starts shortly after, cancels the timer
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
await controller.process_frame(
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
)
# Wait longer than timeout — should not fire
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_function_call_suppresses_timer(self):
"""Test race condition: FunctionCallsStarted arrives before BotStopped.
A race condition can cause FunctionCallsStarted to arrive before
BotStoppedSpeaking. The counter guard prevents the timer from starting
while a function call is in progress.
"""
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
await controller.setup(self.task_manager)
idle_triggered = False
@controller.event_handler("on_user_turn_idle")
async def on_user_turn_idle(controller):
nonlocal idle_triggered
idle_triggered = True
# LLM emits function call and "let me check" concurrently
await controller.process_frame(
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
)
await controller.process_frame(BotStartedSpeakingFrame())
await controller.process_frame(BotStoppedSpeakingFrame())
# Wait longer than timeout — should not fire (function call in progress)
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
# Function call completes, bot speaks result
await controller.process_frame(
FunctionCallResultFrame(
function_name="test", tool_call_id="123", arguments={}, result="ok"
)
)
await controller.process_frame(BotStartedSpeakingFrame())
await controller.process_frame(BotStoppedSpeakingFrame())
# Now the timer should start and fire
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertTrue(idle_triggered)
await controller.cleanup()