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

turns: rename bot turn start to user turn stop strategies

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-30 14:09:00 -08:00
parent fb9a772e33
commit eb5a797b12
161 changed files with 921 additions and 891 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/pipecat/audio/interruptions/min_words_interruption_strategy.py
View File

@@ -21,7 +21,7 @@ class MinWordsInterruptionStrategy(BaseInterruptionStrategy):
.. deprecated:: 0.0.99
This class is deprecated, use
`pipecat.turns.user.MinWordsUserTurnStartStrategy` with `PipelineTask`'s
`pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s
new `turn_start_strategies` parameter instead.
"""
@@ -42,7 +42,7 @@ class MinWordsInterruptionStrategy(BaseInterruptionStrategy):
warnings.simplefilter("always")
warnings.warn(
"'pipecat.audio.interruptions' is deprecated. "
"Use `pipecat.turns.user.MinWordsUserTurnStartStrategy` with `PipelineTask`'s "
"Use `pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s "
"new `turn_start_strategies` parameter instead.",
DeprecationWarning,
)

6
src/pipecat/extensions/voicemail/voicemail_detector.py
View File

@@ -43,7 +43,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.services.llm_service import LLMService
from pipecat.turns.turn_start_strategies import ExternalTurnStartStrategies
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies
from pipecat.utils.sync.base_notifier import BaseNotifier
from pipecat.utils.sync.event_notifier import EventNotifier
@@ -629,9 +629,7 @@ VOICEMAIL SYSTEM (respond "VOICEMAIL"):
self._context = LLMContext(self._messages)
self._context_aggregator = LLMContextAggregatorPair(
self._context,
user_params=LLMUserAggregatorParams(
turn_start_strategies=ExternalTurnStartStrategies()
),
user_params=LLMUserAggregatorParams(user_turn_strategies=ExternalUserTurnStrategies()),
)
# Create notification system for coordinating between components

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

@@ -63,10 +63,10 @@ from pipecat.processors.aggregators.llm_context import (
NotGiven,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.turns.bot import BaseBotTurnStartStrategy, BotTurnStartedParams
from pipecat.turns.mute import BaseUserMuteStrategy
from pipecat.turns.turn_start_strategies import ExternalTurnStartStrategies, TurnStartStrategies
from pipecat.turns.user import BaseUserTurnStartStrategy, UserTurnStartedParams
from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
from pipecat.utils.time import time_now_iso8601
@@ -76,15 +76,15 @@ class LLMUserAggregatorParams:
"""Parameters for configuring LLM user aggregation behavior.
Parameters:
turn_start_strategies: User and bot turn start strategies.
user_turn_strategies: User turn start and stop strategies.
user_mute_strategies: List of user mute strategies.
user_turn_end_timeout: Time in seconds to wait before considering the
user's turn finished and starting the bot turn.
user_turn_stop_timeout: Time in seconds to wait before considering the
user's turn finished.
"""
turn_start_strategies: Optional[TurnStartStrategies] = None
user_turn_strategies: Optional[UserTurnStrategies] = None
user_mute_strategies: List[BaseUserMuteStrategy] = field(default_factory=list)
user_turn_end_timeout: float = 5.0
user_turn_stop_timeout: float = 5.0
@dataclass
@@ -232,8 +232,8 @@ class LLMUserAggregator(LLMContextAggregator):
Event handlers available:
- on_user_turn_started: Called when the user turn starts
- on_bot_turn_started: Called when the user turn ends and it is now the bot’s turn
- on_user_turn_end_timeout: Called when no bot turn start strategy triggers
- on_user_turn_stopped: Called when the user turn ends
- on_user_turn_stop_timeout: Called when no user turn stop strategy triggers
Example::
@@ -241,12 +241,12 @@ class LLMUserAggregator(LLMContextAggregator):
async def on_user_turn_started(aggregator, Optional[strategy]):
...
@aggregator.event_handler("on_bot_turn_started")
async def on_bot_turn_started(aggregator, Optional[strategy]):
@aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(aggregator, Optional[strategy]):
...
@aggregator.event_handler("on_user_turn_end_timeout")
async def on_user_turn_end_timeout(aggregator):
@aggregator.event_handler("on_user_turn_stop_timeout")
async def on_user_turn_stop_timeout(aggregator):
...
"""
@@ -268,19 +268,19 @@ class LLMUserAggregator(LLMContextAggregator):
super().__init__(context=context, role="user", **kwargs)
self._params = params or LLMUserAggregatorParams()
# Initialize default user and bot turn start strategies.
self._turn_start_strategies = self._params.turn_start_strategies or TurnStartStrategies()
# Initialize default user turn strategies.
self._user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
self._vad_user_speaking = False
self._user_turn = False
self._user_is_muted = False
self._user_turn_end_timeout_event = asyncio.Event()
self._user_turn_end_timeout_task: Optional[asyncio.Task] = None
self._user_turn_stop_timeout_event = asyncio.Event()
self._user_turn_stop_timeout_task: Optional[asyncio.Task] = None
self._register_event_handler("on_user_turn_started")
self._register_event_handler("on_user_turn_end_timeout")
self._register_event_handler("on_bot_turn_started")
self._register_event_handler("on_user_turn_stopped")
self._register_event_handler("on_user_turn_stop_timeout")
async def cleanup(self):
"""Clean up processor resources."""
@@ -341,7 +341,7 @@ class LLMUserAggregator(LLMContextAggregator):
else:
await self.push_frame(frame, direction)
await self._turn_start_strategies_process_frame(frame)
await self._user_turn_strategies_process_frame(frame)
async def push_aggregation(self):
"""Push the current aggregation."""
@@ -354,32 +354,32 @@ class LLMUserAggregator(LLMContextAggregator):
await self.push_context_frame()
async def _start(self, frame: StartFrame):
if not self._user_turn_end_timeout_task:
self._user_turn_end_timeout_task = self.create_task(
self._user_turn_end_timeout_task_handler()
if not self._user_turn_stop_timeout_task:
self._user_turn_stop_timeout_task = self.create_task(
self._user_turn_stop_timeout_task_handler()
)
await self._setup_turn_start_strategies()
await self._setup_user_turn_strategies()
await self._setup_user_mute_strategies()
async def _setup_user_mute_strategies(self):
for s in self._params.user_mute_strategies:
await s.setup(self.task_manager)
async def _setup_turn_start_strategies(self):
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
async def _setup_user_turn_strategies(self):
if self._user_turn_strategies.start:
for s in self._user_turn_strategies.start:
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)
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
if self._user_turn_strategies.stop:
for s in self._user_turn_strategies.stop:
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_bot_turn_started", self._on_bot_turn_started)
s.add_event_handler("on_user_turn_stopped", self._on_user_turn_stopped)
async def _stop(self, frame: EndFrame):
await self._cleanup()
@@ -388,20 +388,20 @@ class LLMUserAggregator(LLMContextAggregator):
await self._cleanup()
async def _cleanup(self):
if self._user_turn_end_timeout_task:
await self.cancel_task(self._user_turn_end_timeout_task)
self._user_turn_end_timeout_task = None
if self._user_turn_stop_timeout_task:
await self.cancel_task(self._user_turn_stop_timeout_task)
self._user_turn_stop_timeout_task = None
await self._cleanup_turn_start_strategies()
await self._cleanup_user_turn_strategies()
await self._cleanup_user_mute_strategies()
async def _cleanup_turn_start_strategies(self):
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
async def _cleanup_user_turn_strategies(self):
if self._user_turn_strategies.start:
for s in self._user_turn_strategies.start:
await s.cleanup()
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
if self._user_turn_strategies.stop:
for s in self._user_turn_strategies.stop:
await s.cleanup()
async def _cleanup_user_mute_strategies(self):
@@ -436,13 +436,13 @@ class LLMUserAggregator(LLMContextAggregator):
return should_mute_frame
async def _turn_start_strategies_process_frame(self, frame: Frame):
if self._turn_start_strategies.user:
for strategy in self._turn_start_strategies.user:
async def _user_turn_strategies_process_frame(self, frame: Frame):
if self._user_turn_strategies.start:
for strategy in self._user_turn_strategies.start:
await strategy.process_frame(frame)
if self._turn_start_strategies.bot:
for strategy in self._turn_start_strategies.bot:
if self._user_turn_strategies.stop:
for strategy in self._user_turn_strategies.stop:
await strategy.process_frame(frame)
async def _handle_llm_run(self, frame: LLMRunFrame):
@@ -465,15 +465,15 @@ class LLMUserAggregator(LLMContextAggregator):
logger.warning(
f"{self}: `turn_analyzer` in base input transport is deprecated and "
"might result in unexpected behavior. Use `LLMUserAggregator`'s new `turn_start_strategies` "
"parameter with `TurnAnalyzerBotTurnStartStrategy` instead:\n"
"parameter with `TurnAnalyzerUserTurnStopStrategy` instead:\n"
"\n"
" context_aggregator = LLMContextAggregatorPair(\n"
" context,\n"
" user_params=LLMUserAggregatorParams(\n"
" ...,\n"
" turn_start_strategies=TurnStartStrategies(\n"
" bot=[\n"
" TurnAnalyzerBotTurnStartStrategy(\n"
" user_turn_strategies=UserTurnStrategies(\n"
" stop=[\n"
" TurnAnalyzerUserTurnStopStrategy(\n"
" turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams())\n"
" )\n"
" ],\n"
@@ -482,21 +482,21 @@ class LLMUserAggregator(LLMContextAggregator):
" )"
)
await self._cleanup_turn_start_strategies()
self._turn_start_strategies = ExternalTurnStartStrategies()
await self._setup_turn_start_strategies()
await self._cleanup_user_turn_strategies()
self._turn_strategies = ExternalUserTurnStrategies()
await self._setup_user_turn_strategies()
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_end_timeout_event.set()
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_end_timeout_event.set()
self._user_turn_stop_timeout_event.set()
async def _handle_transcription(self, frame: TranscriptionFrame):
text = frame.text
@@ -506,7 +506,7 @@ class LLMUserAggregator(LLMContextAggregator):
return
# We have creceived a transcription, let's reset the user turn timeout.
self._user_turn_end_timeout_event.set()
self._user_turn_stop_timeout_event.set()
# Transcriptions never include inter-part spaces (so far).
self._aggregation.append(
@@ -522,14 +522,14 @@ class LLMUserAggregator(LLMContextAggregator):
):
await self._trigger_user_turn_start(strategy, params)
async def _on_bot_turn_started(
self, strategy: BaseBotTurnStartStrategy, params: BotTurnStartedParams
async def _on_user_turn_stopped(
self, strategy: BaseUserTurnStopStrategy, params: UserTurnStoppedParams
):
await self._trigger_bot_turn_start(strategy, params)
await self._trigger_user_turn_stop(strategy, params)
async def _on_push_frame(
self,
strategy: BaseUserTurnStartStrategy | BaseBotTurnStartStrategy,
strategy: BaseUserTurnStartStrategy | BaseUserTurnStopStrategy,
frame: Frame,
direction: FrameDirection = FrameDirection.DOWNSTREAM,
):
@@ -537,7 +537,7 @@ class LLMUserAggregator(LLMContextAggregator):
async def _on_broadcast_frame(
self,
strategy: BaseUserTurnStartStrategy | BaseBotTurnStartStrategy,
strategy: BaseUserTurnStartStrategy | BaseUserTurnStopStrategy,
frame_cls: Type[Frame],
**kwargs,
):
@@ -553,11 +553,11 @@ class LLMUserAggregator(LLMContextAggregator):
logger.debug(f"User started speaking (user turn start strategy: {strategy})")
self._user_turn = True
self._user_turn_end_timeout_event.set()
self._user_turn_stop_timeout_event.set()
# Reset all user turn start strategies to start fresh.
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
if self._user_turn_strategies.start:
for s in self._user_turn_strategies.start:
await s.reset()
if params.enable_user_speaking_frames:
@@ -570,45 +570,45 @@ class LLMUserAggregator(LLMContextAggregator):
await self._call_event_handler("on_user_turn_started", strategy)
async def _trigger_bot_turn_start(
self, strategy: Optional[BaseBotTurnStartStrategy], params: BotTurnStartedParams
async def _trigger_user_turn_stop(
self, strategy: Optional[BaseUserTurnStopStrategy], params: UserTurnStoppedParams
):
# Prevent two consecutive bot turn starts.
# Prevent two consecutive user turn stops.
if not self._user_turn:
return
logger.debug(f"User stopped speaking (bot turn start strategy: {strategy})")
logger.debug(f"User stopped speaking (user turn stop strategy: {strategy})")
self._user_turn = False
self._user_turn_end_timeout_event.set()
self._user_turn_stop_timeout_event.set()
# Reset all bot turn start strategies to start fresh.
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
# Reset all user turn stop strategies to start fresh.
if self._user_turn_strategies.stop:
for s in self._user_turn_strategies.stop:
await s.reset()
if params.enable_user_speaking_frames:
# TODO(aleix): This frame should really come from the top of the pipeline.
await self.broadcast_frame(UserStoppedSpeakingFrame)
await self._call_event_handler("on_bot_turn_started", strategy)
await self._call_event_handler("on_user_turn_stopped", strategy)
# Always push context frame.
await self.push_aggregation()
async def _user_turn_end_timeout_task_handler(self):
async def _user_turn_stop_timeout_task_handler(self):
while True:
try:
await asyncio.wait_for(
self._user_turn_end_timeout_event.wait(),
timeout=self._params.user_turn_end_timeout,
self._user_turn_stop_timeout_event.wait(),
timeout=self._params.user_turn_stop_timeout,
)
self._user_turn_end_timeout_event.clear()
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_end_timeout")
await self._trigger_bot_turn_start(
None, BotTurnStartedParams(enable_user_speaking_frames=True)
await self._call_event_handler("on_user_turn_stop_timeout")
await self._trigger_user_turn_stop(
None, UserTurnStoppedParams(enable_user_speaking_frames=True)
)

15
src/pipecat/turns/bot/__init__.py
View File

@@ -1,15 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.turns.bot.base_bot_turn_start_strategy import (
BaseBotTurnStartStrategy,
BotTurnStartedParams,
)
from pipecat.turns.bot.external_bot_turn_start_strategy import ExternalBotTurnStartStrategy
from pipecat.turns.bot.transcription_bot_turn_start_strategy import (
TranscriptionBotTurnStartStrategy,
)
from pipecat.turns.bot.turn_analyzer_bot_turn_start_strategy import TurnAnalyzerBotTurnStartStrategy

75
src/pipecat/turns/turn_start_strategies.py
View File

@@ -1,75 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Turn start strategy configuration."""
from dataclasses import dataclass
from typing import List, Optional
from pipecat.turns.bot import (
BaseBotTurnStartStrategy,
ExternalBotTurnStartStrategy,
TranscriptionBotTurnStartStrategy,
)
from pipecat.turns.user import (
BaseUserTurnStartStrategy,
ExternalUserTurnStartStrategy,
TranscriptionUserTurnStartStrategy,
VADUserTurnStartStrategy,
)
@dataclass
class TurnStartStrategies:
"""Container for user and bot turn start strategies.
This class groups the configured turn start strategies for both the user
and the bot.
If no strategies are specified for the user or the bot, the following
defaults are used:
user: [VADUserTurnStartStrategy, TranscriptionUserTurnStartStrategy]
bot: [TranscriptionBotTurnStartStrategy]
Attributes:
user: A list of user turn start strategies used to detect when the
user starts speaking.
bot: A list of bot turn start strategies used to decide when the bot
should start speaking.
"""
user: Optional[List[BaseUserTurnStartStrategy]] = None
bot: Optional[List[BaseBotTurnStartStrategy]] = None
def __post_init__(self):
if not self.user:
self.user = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
if not self.bot:
self.bot = [TranscriptionBotTurnStartStrategy()]
@dataclass
class ExternalTurnStartStrategies(TurnStartStrategies):
"""Default container for external user and bot turn start strategies.
This class provides a convenience default for configuring external turn
control. It preconfigures `TurnStartStrategies` with
`ExternalUserTurnStartStrategy` and `ExternalBotTurnStartStrategy`, allowing
external processors (such as services) to control when user and bot turns
start.
When using this container, the user aggregator does not push
`UserStartedSpeakingFrame` or `UserStoppedSpeakingFrame` frames, and does
not generate interruptions. These signals are expected to be provided by an
external processor.
"""
def __post_init__(self):
self.user = [ExternalUserTurnStartStrategy()]
self.bot = [ExternalBotTurnStartStrategy()]

16
src/pipecat/turns/user/__init__.py
View File

@@ -1,16 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.turns.user.base_user_turn_start_strategy import (
BaseUserTurnStartStrategy,
UserTurnStartedParams,
)
from pipecat.turns.user.external_user_turn_start_strategy import ExternalUserTurnStartStrategy
from pipecat.turns.user.min_words_user_turn_start_strategy import MinWordsUserTurnStartStrategy
from pipecat.turns.user.transcription_user_turn_start_strategy import (
TranscriptionUserTurnStartStrategy,
)
from pipecat.turns.user.vad_user_turn_start_strategy import VADUserTurnStartStrategy

18
src/pipecat/turns/user_start/__init__.py Normal file
View File

@@ -0,0 +1,18 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.turns.user_start.base_user_turn_start_strategy import (
BaseUserTurnStartStrategy,
UserTurnStartedParams,
)
from pipecat.turns.user_start.external_user_turn_start_strategy import ExternalUserTurnStartStrategy
from pipecat.turns.user_start.min_words_user_turn_start_strategy import (
MinWordsUserTurnStartStrategy,
)
from pipecat.turns.user_start.transcription_user_turn_start_strategy import (
TranscriptionUserTurnStartStrategy,
)
from pipecat.turns.user_start.vad_user_turn_start_strategy import VADUserTurnStartStrategy

0
src/pipecat/turns/user/base_user_turn_start_strategy.py → src/pipecat/turns/user_start/base_user_turn_start_strategy.py
View File

2
src/pipecat/turns/user/external_user_turn_start_strategy.py → src/pipecat/turns/user_start/external_user_turn_start_strategy.py
View File

@@ -7,7 +7,7 @@
"""User turn start strategy triggered by externally emitted frames."""
from pipecat.frames.frames import Frame, UserStartedSpeakingFrame
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
class ExternalUserTurnStartStrategy(BaseUserTurnStartStrategy):

2
src/pipecat/turns/user/min_words_user_turn_start_strategy.py → src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py
View File

@@ -15,7 +15,7 @@ from pipecat.frames.frames import (
InterimTranscriptionFrame,
TranscriptionFrame,
)
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):

2
src/pipecat/turns/user/transcription_user_turn_start_strategy.py → src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py
View File

@@ -7,7 +7,7 @@
"""User turn start strategy based on transcriptions."""
from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy):

2
src/pipecat/turns/user/vad_user_turn_start_strategy.py → src/pipecat/turns/user_start/vad_user_turn_start_strategy.py
View File

@@ -7,7 +7,7 @@
"""User turn start strategy based on VAD events."""
from pipecat.frames.frames import Frame, VADUserStartedSpeakingFrame
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
class VADUserTurnStartStrategy(BaseUserTurnStartStrategy):

17
src/pipecat/turns/user_stop/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.turns.user_stop.base_user_turn_stop_strategy import (
BaseUserTurnStopStrategy,
UserTurnStoppedParams,
)
from pipecat.turns.user_stop.external_user_turn_stop_strategy import ExternalUserTurnStopStrategy
from pipecat.turns.user_stop.transcription_user_turn_stop_strategy import (
TranscriptionUserTurnStopStrategy,
)
from pipecat.turns.user_stop.turn_analyzer_user_turn_stop_strategy import (
TurnAnalyzerUserTurnStopStrategy,
)

48
src/pipecat/turns/bot/base_bot_turn_start_strategy.py → src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base turn start strategy for determining when the bot should start speaking."""
"""Base user turn stop strategy for determining when the user stopped speaking."""
from dataclasses import dataclass
from typing import Optional, Type
@@ -16,41 +16,41 @@ from pipecat.utils.base_object import BaseObject
@dataclass
class BotTurnStartedParams:
"""Parameters emitted when a bot turn starts.
class UserTurnStoppedParams:
"""Parameters emitted when a user turn stops.
These parameters are passed to the `on_bot_turn_started` event and provide
contextual information about how the bot turn should be handled by the user
aggregator.
These parameters are passed to the `on_user_turn_stopped` event and provide
contextual information about how the end of user turn should be handled by
the user aggregator.
Attributes:
enable_user_speaking_frames: Whether the user aggregator should emit
frames indicating user speaking state (e.g., user stopped speaking)
during the bot's turn. This is typically enabled by default, but may
be disabled when another component (such as an STT service) is already
responsible for generating user speaking frames.
frames indicating user speaking state (e.g., user stopped speaking).
This is typically enabled by default, but may be disabled when another
component (such as an STT service) is already responsible for
generating user speaking frames.
"""
enable_user_speaking_frames: bool
class BaseBotTurnStartStrategy(BaseObject):
"""Base class for strategies that determine when the bot should start speaking.
class BaseUserTurnStopStrategy(BaseObject):
"""Base class for strategies that determine when the user stops speaking.
Subclasses should implement logic to detect when the bot should start
Subclasses should implement logic to detect when the user stops
speaking. This could be based on analyzing incoming frames (such as
transcriptions), conversation state, or other heuristics.
Events triggered by bot turn start strategies:
Events triggered by strategies:
- `on_push_frame`: Indicates the strategy wants to push a frame.
- `on_bot_turn_started`: Signals that the bot should start speaking.
- `on_user_turn_stopped`: Signals that the user stopped speaking.
"""
def __init__(self, *, enable_user_speaking_frames: bool = True, **kwargs):
"""Initialize the base bot turn start strategy.
"""Initialize the base user turn stop strategy.
Args:
enable_user_speaking_frames: If True, the aggregator will emit frames
@@ -64,13 +64,13 @@ class BaseBotTurnStartStrategy(BaseObject):
self._task_manager: Optional[BaseTaskManager] = None
self._register_event_handler("on_push_frame", sync=True)
self._register_event_handler("on_broadcast_frame", sync=True)
self._register_event_handler("on_bot_turn_started", sync=True)
self._register_event_handler("on_user_turn_stopped", sync=True)
@property
def task_manager(self) -> BaseTaskManager:
"""Returns the configured task manager."""
if not self._task_manager:
raise RuntimeError(f"{self} bot turn start strategy was not properly setup")
raise RuntimeError(f"{self} user turn stop strategy was not properly setup")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
@@ -90,10 +90,10 @@ class BaseBotTurnStartStrategy(BaseObject):
pass
async def process_frame(self, frame: Frame):
"""Process an incoming frame to decide whether the bot should speak.
"""Process an incoming frame to decide whether the user stopped speaking.
Subclasses should override this to implement logic that decides whether
the bot turn has started.
the user has stopped speaking.
Args:
frame: The frame to be analyzed.
@@ -118,9 +118,9 @@ class BaseBotTurnStartStrategy(BaseObject):
"""
await self._call_event_handler("on_broadcast_frame", frame_cls, **kwargs)
async def trigger_bot_turn_started(self):
"""Trigger the `on_bot_turn_started` event."""
async def trigger_user_turn_stopped(self):
"""Trigger the `on_user_turn_stopped` event."""
await self._call_event_handler(
"on_bot_turn_started",
BotTurnStartedParams(enable_user_speaking_frames=self._enable_user_speaking_frames),
"on_user_turn_stopped",
UserTurnStoppedParams(enable_user_speaking_frames=self._enable_user_speaking_frames),
)

26
src/pipecat/turns/bot/external_bot_turn_start_strategy.py → src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bot turn start strategy triggered by externally emitted frames."""
"""User turn stop strategy triggered by externally emitted frames."""
import asyncio
from typing import Optional
@@ -16,12 +16,12 @@ from pipecat.frames.frames import (
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy
from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
from pipecat.utils.asyncio.task_manager import BaseTaskManager
class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""Bot turn start strategy controlled by an external processor.
class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy):
"""User turn stop strategy controlled by an external processor.
This strategy does not determine when a user turn ends on its own, it relies
on a different processor in the pipeline which is responsible for emitting
@@ -30,7 +30,7 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""
def __init__(self, *, timeout: float = 0.5, **kwargs):
"""Initialize the external bot turn start strategy.
"""Initialize the external user turn stop strategy.
Args:
timeout: A short delay used internally to handle consecutive or
@@ -72,8 +72,8 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
async def process_frame(self, frame: Frame):
"""Process an incoming frame to update strategy state.
Updates internal transcription text and VAD state. The bot turn will be
triggered when appropriate based on the collected frames.
Updates internal transcription text and VAD state. The user end turn
will be triggered when appropriate based on the collected frames.
Args:
frame: The frame to be analyzed.
@@ -95,7 +95,7 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
"""Handle when the external service indicates the user has stopped speaking."""
self._user_speaking = False
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _handle_interim_transcription(self, frame: InterimTranscriptionFrame):
self._seen_interim_results = True
@@ -109,10 +109,10 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
self._event.set()
async def _task_handler(self):
"""Asynchronously monitor transcriptions and trigger bot turn when ready.
"""Asynchronously monitor transcriptions and trigger user end turn when ready.
If transcription text exists and the user is not currently speaking,
triggers the bot turn. Handles multiple or delayed transcriptions
triggers the user end turn. Handles multiple or delayed transcriptions
gracefully.
"""
@@ -121,8 +121,8 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
self._event.clear()
except asyncio.TimeoutError:
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _maybe_trigger_bot_turn_started(self):
async def _maybe_trigger_user_turn_stopped(self):
if not self._user_speaking and not self._seen_interim_results and self._text:
await self.trigger_bot_turn_started()
await self.trigger_user_turn_stopped()

32
src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py → src/pipecat/turns/user_stop/transcription_user_turn_stop_strategy.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Transcription time-based bot turn start strategy."""
"""Transcription time-based user turn stop strategy."""
import asyncio
from typing import Optional
@@ -16,20 +16,20 @@ from pipecat.frames.frames import (
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy
from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
from pipecat.utils.asyncio.task_manager import BaseTaskManager
class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""Bot turn start strategy based on transcriptions.
class TranscriptionUserTurnStopStrategy(BaseUserTurnStopStrategy):
"""User turn stop strategy based on transcriptions.
This strategy assumes the user stops speaking once a transcription has been
received. It handles multiple or delayed transcription frames gracefully.
This strategy assumes the bot should start speaking once a transcription
has been received and the user is not actively speaking. It handles
multiple or delayed transcription frames gracefully.
"""
def __init__(self, *, timeout: float = 0.5, **kwargs):
"""Initialize the transcription-based bot turn start strategy.
"""Initialize the transcription-based user turn stop strategy.
Args:
timeout: A short delay used internally to handle consecutive or
@@ -71,8 +71,8 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
async def process_frame(self, frame: Frame):
"""Process an incoming frame to update strategy state.
Updates internal transcription text and VAD state. The bot turn will be
triggered when appropriate based on the collected frames.
Updates internal transcription text and VAD state. The user end turn
will be triggered when appropriate based on the collected frames.
Args:
frame: The frame to be analyzed.
@@ -94,7 +94,7 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
async def _handle_vad_user_stopped_speaking(self, _: VADUserStoppedSpeakingFrame):
"""Handle when the VAD indicates the user has stopped speaking."""
self._vad_user_speaking = False
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _handle_interim_transcription(self, frame: InterimTranscriptionFrame):
self._seen_interim_results = True
@@ -108,10 +108,10 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
self._event.set()
async def _task_handler(self):
"""Asynchronously monitor transcriptions and trigger bot turn when ready.
"""Asynchronously monitor transcriptions and trigger user end turn when ready.
If transcription text exists and the user is not currently speaking,
triggers the bot turn. Handles multiple or delayed transcriptions
triggers the user end turn. Handles multiple or delayed transcriptions
gracefully.
"""
@@ -120,8 +120,8 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
self._event.clear()
except asyncio.TimeoutError:
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _maybe_trigger_bot_turn_started(self):
async def _maybe_trigger_user_turn_stopped(self):
if not self._vad_user_speaking and not self._seen_interim_results and self._text:
await self.trigger_bot_turn_started()
await self.trigger_user_turn_stopped()

27
src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py → src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bot turn start strategy based on turn detection analyzers."""
"""User turn stop strategy based on turn detection analyzers."""
import asyncio
from typing import Optional
@@ -22,21 +22,21 @@ from pipecat.frames.frames import (
VADUserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import MetricsData
from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy
from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
from pipecat.utils.asyncio.task_manager import BaseTaskManager
class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""Bot turn start strategy using a turn detection model to detect end of user turn.
class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
"""User turn stop strategy using a turn detection model to detect end of user turn.
This strategy uses the turn detection models to determine when the user has
finished speaking, combining audio, VAD, and transcription frames. Once the
turn is considered complete, the bot turn is triggered.
turn is considered complete, the user end of turn is triggered.
"""
def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5, **kwargs):
"""Initialize the bot turn start strategy.
"""Initialize the user turn stop strategy.
Args:
turn_analyzer: The turn detection analyzer instance to detect end of user turn.
@@ -107,10 +107,11 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy):
state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking)
# If at this point the model says the turn is complete it will be due to
# a timeout, so we mark turn as complete and we trigger the bot turn.
# a timeout, so we mark turn as complete and we trigger the user end of
# turn.
if state == EndOfTurnState.COMPLETE:
self._turn_complete = True
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame):
"""Handle when the VAD indicates the user is speaking."""
@@ -149,11 +150,11 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy):
await self.push_frame(MetricsFrame(data=[result]))
async def _task_handler(self):
"""Asynchronously monitor events and trigger bot turn when appropriate.
"""Asynchronously monitor events and trigger user end of turn when appropriate.
If we have not received a transcription in the specified amount of time
(and we initially received one) and the turn analyzer said the turn is
done, then the bot is ready to speak.
done, then the user is done speaking.
"""
while True:
@@ -161,8 +162,8 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy):
await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
self._event.clear()
except asyncio.TimeoutError:
await self._maybe_trigger_bot_turn_started()
await self._maybe_trigger_user_turn_stopped()
async def _maybe_trigger_bot_turn_started(self):
async def _maybe_trigger_user_turn_stopped(self):
if self._text and self._turn_complete:
await self.trigger_bot_turn_started()
await self.trigger_user_turn_stopped()

71
src/pipecat/turns/user_turn_strategies.py Normal file
View File

@@ -0,0 +1,71 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Turn start strategy configuration."""
from dataclasses import dataclass
from typing import List, Optional
from pipecat.turns.user_start import (
BaseUserTurnStartStrategy,
ExternalUserTurnStartStrategy,
TranscriptionUserTurnStartStrategy,
VADUserTurnStartStrategy,
)
from pipecat.turns.user_stop import (
BaseUserTurnStopStrategy,
ExternalUserTurnStopStrategy,
TranscriptionUserTurnStopStrategy,
)
@dataclass
class UserTurnStrategies:
"""Container for user turn start and stop strategies.
If no strategies are specified, the following defaults are used:
start: [VADUserTurnStartStrategy, TranscriptionUserTurnStartStrategy]
stop: [TranscriptionUserTurnStopStrategy]
Attributes:
start: A list of user turn start strategies used to detect when
the user starts speaking.
stop: A list of user turn stop strategies used to decide when
the user stops speaking.
"""
start: Optional[List[BaseUserTurnStartStrategy]] = None
stop: Optional[List[BaseUserTurnStopStrategy]] = None
def __post_init__(self):
if not self.start:
self.start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
if not self.stop:
self.stop = [TranscriptionUserTurnStopStrategy()]
@dataclass
class ExternalUserTurnStrategies(UserTurnStrategies):
"""Default container for external user turn start and stop strategies.
This class provides a convenience default for configuring external turn
control. It preconfigures `UserTurnStrategies` with
`ExternalUserTurnStartStrategy` and `ExternalUserTurnStopStrategy`, allowing
external processors (such as services) to control when user turn starts and
stops.
When using this container, the user aggregator does not push
`UserStartedSpeakingFrame` or `UserStoppedSpeakingFrame` frames, and does
not generate interruptions. These signals are expected to be provided by an
external processor.
"""
def __post_init__(self):
self.start = [ExternalUserTurnStartStrategy()]
self.stop = [ExternalUserTurnStopStrategy()]