From 199986815c29c4905156de2490436c26a97a26a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 7 Jan 2026 17:56:10 -0800 Subject: [PATCH] turns: add UserTurnController for user turn management --- changelog/3372.added.md | 1 + src/pipecat/turns/user_turn_controller.py | 267 ++++++++++++++++++++++ 2 files changed, 268 insertions(+) create mode 100644 changelog/3372.added.md create mode 100644 src/pipecat/turns/user_turn_controller.py diff --git a/changelog/3372.added.md b/changelog/3372.added.md new file mode 100644 index 000000000..3468f7ac1 --- /dev/null +++ b/changelog/3372.added.md @@ -0,0 +1 @@ +- Added `UserTurnController` to manage user turns. It emits `on_user_turn_started`, `on_user_turn_stopped`, and `on_user_turn_stop_timeout` events, and can be integrated into processors to detect and handle user turns. `LLMUserAggregator` and `UserTurnProcessor` are implemented using this controller. diff --git a/src/pipecat/turns/user_turn_controller.py b/src/pipecat/turns/user_turn_controller.py new file mode 100644 index 000000000..a9b267ae0 --- /dev/null +++ b/src/pipecat/turns/user_turn_controller.py @@ -0,0 +1,267 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""This module defines a controller for managing user turn lifecycle.""" + +import asyncio +from typing import Optional, Type + +from pipecat.frames.frames import ( + Frame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams +from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams +from pipecat.turns.user_turn_strategies import UserTurnStrategies +from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.base_object import BaseObject + + +class UserTurnController(BaseObject): + """Controller for managing user turn lifecycle. + + This class manages user turn state (active/inactive), handles start and stop + strategies, and emits events when user turns begin, end, or timeout occurs. + + Event handlers available: + + - on_user_turn_started: Emitted when a user turn starts. + - on_user_turn_stopped: Emitted when a user turn stops. + - on_user_turn_stop_timeout: Emitted if no stop strategy triggers before timeout. + - on_push_frame: Emitted when a strategy wants to push a frame. + - on_broadcast_frame: Emitted when a strategy wants to broadcast a frame. + + Example:: + + @controller.event_handler("on_user_turn_started") + async def on_user_turn_started(controller, strategy: BaseUserTurnStartStrategy, params: UserTurnStartedParams): + ... + + @controller.event_handler("on_user_turn_stopped") + async def on_user_turn_stopped(controller, strategy: BaseUserTurnStopStrategy, params: UserTurnStoppedParams): + ... + + @controller.event_handler("on_user_turn_stop_timeout") + async def on_user_turn_stop_timeout(controller): + ... + + @controller.event_handler("on_push_frame") + async def on_push_frame(controller, frame: Frame, direction: FrameDirection): + ... + + @controller.event_handler("on_broadcast_frame") + async def on_broadcast_frame(controller, frame_cls: Type[Frame], **kwargs): + ... + """ + + def __init__( + self, + *, + user_turn_strategies: UserTurnStrategies, + user_turn_stop_timeout: float = 5.0, + ): + """Initialize the user turn controller. + + Args: + user_turn_strategies: Configured strategies for starting and stopping user turns. + user_turn_stop_timeout: Timeout in seconds to automatically stop a user turn + if no activity is detected. + """ + super().__init__() + + self._user_turn_strategies = user_turn_strategies + self._user_turn_stop_timeout = user_turn_stop_timeout + + self._task_manager: Optional[BaseTaskManager] = None + + self._vad_user_speaking = False + + self._user_turn = False + self._user_turn_stop_timeout_event = asyncio.Event() + self._user_turn_stop_timeout_task: Optional[asyncio.Task] = None + + self._register_event_handler("on_push_frame", sync=True) + self._register_event_handler("on_broadcast_frame", sync=True) + self._register_event_handler("on_user_turn_started", sync=True) + self._register_event_handler("on_user_turn_stopped", sync=True) + self._register_event_handler("on_user_turn_stop_timeout", sync=True) + + @property + def task_manager(self) -> BaseTaskManager: + """Returns the configured task manager.""" + if not self._task_manager: + raise RuntimeError(f"{self} user turn controller was not properly setup") + return self._task_manager + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the controller with the given task manager. + + Args: + task_manager: The task manager to be associated with this instance. + """ + self._task_manager = task_manager + + if not self._user_turn_stop_timeout_task: + self._user_turn_stop_timeout_task = self.task_manager.create_task( + self._user_turn_stop_timeout_task_handler(), + f"{self}::_user_turn_stop_timeout_task_handler", + ) + + await self._setup_strategies() + + async def cleanup(self): + """Cleanup the controller.""" + await super().cleanup() + + if self._user_turn_stop_timeout_task: + await self.task_manager.cancel_task(self._user_turn_stop_timeout_task) + self._user_turn_stop_timeout_task = None + + await self._cleanup_strategies() + + async def update_strategies(self, strategies: UserTurnStrategies): + """Replace the current strategies with the given ones. + + Args: + strategies: The new user turn strategies the controller should use. + """ + await self._cleanup_strategies() + self._user_turn_strategies = strategies + await self._setup_strategies() + + async def process_frame(self, frame: Frame): + """Process an incoming frame to detect user turn start or stop. + + The frame is passed to the configured user turn strategies, which are + responsible for deciding when a user turn starts or stops and emitting + the corresponding events. + + Args: + frame: The frame to be processed. + + """ + if isinstance(frame, VADUserStartedSpeakingFrame): + await self._handle_vad_user_started_speaking(frame) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._handle_vad_user_stopped_speaking(frame) + elif isinstance(frame, TranscriptionFrame): + await self._handle_transcription(frame) + + for strategy in self._user_turn_strategies.start or []: + await strategy.process_frame(frame) + + for strategy in self._user_turn_strategies.stop or []: + await strategy.process_frame(frame) + + async def _setup_strategies(self): + for s in self._user_turn_strategies.start or []: + await s.setup(self.task_manager) + s.add_event_handler("on_push_frame", self._on_push_frame) + s.add_event_handler("on_broadcast_frame", self._on_broadcast_frame) + s.add_event_handler("on_user_turn_started", self._on_user_turn_started) + + for s in self._user_turn_strategies.stop or []: + await s.setup(self.task_manager) + s.add_event_handler("on_push_frame", self._on_push_frame) + s.add_event_handler("on_broadcast_frame", self._on_broadcast_frame) + s.add_event_handler("on_user_turn_stopped", self._on_user_turn_stopped) + + async def _cleanup_strategies(self): + for s in self._user_turn_strategies.start or []: + await s.cleanup() + + for s in self._user_turn_strategies.stop or []: + await s.cleanup() + + async def _handle_vad_user_started_speaking(self, frame: VADUserStartedSpeakingFrame): + self._vad_user_speaking = True + + # The user started talking, let's reset the user turn timeout. + self._user_turn_stop_timeout_event.set() + + async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame): + self._vad_user_speaking = False + + # The user stopped talking, let's reset the user turn timeout. + self._user_turn_stop_timeout_event.set() + + async def _handle_transcription(self, frame: TranscriptionFrame): + # We have creceived a transcription, let's reset the user turn timeout. + self._user_turn_stop_timeout_event.set() + + async def _on_push_frame( + self, + strategy: BaseUserTurnStartStrategy | BaseUserTurnStopStrategy, + frame: Frame, + direction: FrameDirection = FrameDirection.DOWNSTREAM, + ): + await self._call_event_handler("on_push_frame", frame, direction) + + async def _on_broadcast_frame( + self, + strategy: BaseUserTurnStartStrategy | BaseUserTurnStopStrategy, + frame_cls: Type[Frame], + **kwargs, + ): + await self._call_event_handler("on_broadcast_frame", frame_cls, **kwargs) + + async def _on_user_turn_started( + self, + strategy: BaseUserTurnStartStrategy, + params: UserTurnStartedParams, + ): + await self._trigger_user_turn_start(strategy, params) + + async def _on_user_turn_stopped( + self, strategy: BaseUserTurnStopStrategy, params: UserTurnStoppedParams + ): + await self._trigger_user_turn_stop(strategy, params) + + async def _trigger_user_turn_start( + self, strategy: Optional[BaseUserTurnStartStrategy], params: UserTurnStartedParams + ): + # Prevent two consecutive user turn starts. + if self._user_turn: + return + + self._user_turn = True + self._user_turn_stop_timeout_event.set() + + await self._call_event_handler("on_user_turn_started", strategy, params) + + async def _trigger_user_turn_stop( + self, strategy: Optional[BaseUserTurnStopStrategy], params: UserTurnStoppedParams + ): + # Prevent two consecutive user turn stops. + if not self._user_turn: + return + + self._user_turn = False + self._user_turn_stop_timeout_event.set() + + # Reset all user turn stop strategies to start fresh. + for s in self._user_turn_strategies.stop or []: + await s.reset() + + await self._call_event_handler("on_user_turn_stopped", strategy, params) + + async def _user_turn_stop_timeout_task_handler(self): + while True: + try: + await asyncio.wait_for( + self._user_turn_stop_timeout_event.wait(), + timeout=self._user_turn_stop_timeout, + ) + self._user_turn_stop_timeout_event.clear() + except asyncio.TimeoutError: + if self._user_turn and not self._vad_user_speaking: + await self._call_event_handler("on_user_turn_stop_timeout") + await self._trigger_user_turn_stop( + None, UserTurnStoppedParams(enable_user_speaking_frames=True) + )