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

LLMUserAggregator: move turn_start_strategies from PipelineTask

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-24 15:12:18 -08:00
parent e5bd55d1d5
commit 8b861d9143
149 changed files with 1691 additions and 757 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/pipecat/frames/frames.py
View File

@@ -38,9 +38,8 @@ from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id
if TYPE_CHECKING:
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage, NotGiven
from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.turns.turn_start_strategies import TurnStartStrategies
class DeprecatedKeypadEntry:
@@ -958,7 +957,7 @@ class StartFrame(SystemFrame):
interruption_strategies: List of interruption handling strategies.
.. deprecated:: 0.0.99
Use the `turn_start_strategies` instead.
Use `LLMUserAggregator`'s new `turn_start_strategies` parameter instead.
report_only_initial_ttfb: Whether to report only initial time-to-first-byte.
"""
@@ -970,7 +969,6 @@ class StartFrame(SystemFrame):
enable_tracing: bool = False
enable_usage_metrics: bool = False
interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list)
turn_start_strategies: Optional["TurnStartStrategies"] = None
report_only_initial_ttfb: bool = False

10
src/pipecat/pipeline/task.py
View File

@@ -47,7 +47,6 @@ from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams
from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
from pipecat.pipeline.task_observer import TaskObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.turns.turn_start_strategies import TurnStartStrategies
from pipecat.utils.asyncio.task_manager import BaseTaskManager, TaskManager, TaskManagerParams
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
@@ -115,9 +114,8 @@ class PipelineParams(BaseModel):
interruption_strategies: [deprecated] Strategies for bot interruption behavior.
.. deprecated:: 0.0.99
Use the `turn_start_strategies` instead.
Use `LLMUserAggregator`'s new `turn_start_strategies` parameter instead.
turn_start_strategies: User and bot turn start strategies.
observers: [deprecated] Use `observers` arg in `PipelineTask` class.
.. deprecated:: 0.0.58
@@ -138,7 +136,6 @@ class PipelineParams(BaseModel):
enable_usage_metrics: bool = False
heartbeats_period_secs: float = HEARTBEAT_SECS
interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
turn_start_strategies: Optional[TurnStartStrategies] = None
observers: List[BaseObserver] = Field(default_factory=list)
report_only_initial_ttfb: bool = False
send_initial_empty_metrics: bool = True
@@ -286,10 +283,6 @@ class PipelineTask(BasePipelineTask):
)
observers.append(self._turn_trace_observer)
# Initialize default user and bot turn start strategies.
if not self._params.turn_start_strategies:
self._params.turn_start_strategies = TurnStartStrategies()
self._finished = False
self._cancelled = False
@@ -706,7 +699,6 @@ class PipelineTask(BasePipelineTask):
enable_usage_metrics=self._params.enable_usage_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
interruption_strategies=self._params.interruption_strategies,
turn_start_strategies=self._params.turn_start_strategies,
)
start_frame.metadata = self._params.start_metadata
await self._pipeline.queue_frame(start_frame)

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

@@ -65,6 +65,7 @@ from pipecat.processors.aggregators.llm_context import (
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy
from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
from pipecat.turns.turn_start_strategies import TurnStartStrategies
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
from pipecat.utils.time import time_now_iso8601
@@ -80,12 +81,14 @@ class LLMUserAggregatorParams:
interruption frames. This is enabled by default, but you may want
to disable it if another component (e.g., an STT service) is already
generating these frames.
turn_start_strategies: User and bot turn start 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.
"""
enable_user_speaking_frames: bool = True
turn_start_strategies: Optional[TurnStartStrategies] = None
user_mute_strategies: List[BaseUserMuteStrategy] = field(default_factory=list)
user_turn_end_timeout: float = 5.0
@@ -271,6 +274,9 @@ 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()
self._vad_user_speaking = False
self._user_turn = False
@@ -362,15 +368,15 @@ class LLMUserAggregator(LLMContextAggregator):
for s in self._params.user_mute_strategies:
await s.setup(self.task_manager)
if self.turn_start_strategies and self.turn_start_strategies.user:
for s in self.turn_start_strategies.user:
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
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 and self.turn_start_strategies.bot:
for s in self.turn_start_strategies.bot:
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
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)
@@ -390,12 +396,12 @@ class LLMUserAggregator(LLMContextAggregator):
for s in self._params.user_mute_strategies:
await s.cleanup()
if self.turn_start_strategies and self.turn_start_strategies.user:
for s in self.turn_start_strategies.user:
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
await s.cleanup()
if self.turn_start_strategies and self.turn_start_strategies.bot:
for s in self.turn_start_strategies.bot:
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
await s.cleanup()
async def _maybe_mute_frame(self, frame: Frame):
@@ -427,12 +433,12 @@ class LLMUserAggregator(LLMContextAggregator):
return should_mute_frame
async def _turn_start_strategies_process_frame(self, frame: Frame):
if self.turn_start_strategies and self.turn_start_strategies.user:
for strategy in self.turn_start_strategies.user:
if self._turn_start_strategies.user:
for strategy in self._turn_start_strategies.user:
await strategy.process_frame(frame)
if self.turn_start_strategies and self.turn_start_strategies.bot:
for strategy in self.turn_start_strategies.bot:
if self._turn_start_strategies.bot:
for strategy in self._turn_start_strategies.bot:
await strategy.process_frame(frame)
async def _handle_llm_run(self, frame: LLMRunFrame):
@@ -454,17 +460,21 @@ class LLMUserAggregator(LLMContextAggregator):
logger.warning(
f"{self}: `turn_analyzer` in base input transport is deprecated and "
"might result in unexpected behavior. Use `PipelineTask`'s `turn_start_strategies` with "
"`TurnAnalyzerBotTurnStartStrategy` instead.:\n\n"
" task = PipelineTask(\n"
" pipeline,\n"
" params=PipelineParams(\n"
"might result in unexpected behavior. Use `LLMUserAggregator`'s new `turn_start_strategies` "
"parameter with `TurnAnalyzerBotTurnStartStrategy` instead:\n"
"\n"
" context_aggregator = LLMContextAggregatorPair(\n"
" context,\n"
" user_params=LLMUserAggregatorParams(\n"
" ...,\n"
" turn_start_strategies=TurnStartStrategies(\n"
" bot=[TurnAnalyzerBotTurnStartStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]\n"
" ),\n"
" bot=[\n"
" TurnAnalyzerBotTurnStartStrategy(\n"
" turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams())\n"
" )\n"
" ],\n"
" )\n"
" ),\n"
" ...,\n"
" )"
)
@@ -528,8 +538,8 @@ class LLMUserAggregator(LLMContextAggregator):
self._user_turn_end_timeout_event.set()
# Reset all user turn start strategies to start fresh.
if self.turn_start_strategies and self.turn_start_strategies.user:
for s in self.turn_start_strategies.user:
if self._turn_start_strategies.user:
for s in self._turn_start_strategies.user:
await s.reset()
if self._params.enable_user_speaking_frames:
@@ -549,8 +559,8 @@ class LLMUserAggregator(LLMContextAggregator):
self._user_turn_end_timeout_event.set()
# Reset all bot turn start strategies to start fresh.
if self.turn_start_strategies and self.turn_start_strategies.bot:
for s in self.turn_start_strategies.bot:
if self._turn_start_strategies.bot:
for s in self._turn_start_strategies.bot:
await s.reset()
if self._params.enable_user_speaking_frames:

15
src/pipecat/processors/frame_processor.py
View File

@@ -16,7 +16,6 @@ import traceback
from dataclasses import dataclass
from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
Awaitable,
Callable,
@@ -52,9 +51,6 @@ from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMet
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
if TYPE_CHECKING:
from pipecat.turns.turn_start_strategies import TurnStartStrategies
class FrameDirection(Enum):
"""Direction of frame flow in the processing pipeline.
@@ -199,7 +195,6 @@ class FrameProcessor(BaseObject):
self._enable_usage_metrics = False
self._report_only_initial_ttfb = False
self._interruption_strategies: List[BaseInterruptionStrategy] = []
self._turn_start_strategies: Optional["TurnStartStrategies"] = None
# Indicates whether we have received the StartFrame.
self.__started = False
@@ -368,15 +363,6 @@ class FrameProcessor(BaseObject):
"""
return self._interruption_strategies
@property
def turn_start_strategies(self) -> Optional["TurnStartStrategies"]:
"""Get the user and bot turn start strategies for this processor.
Returns:
The user and bot turn start strategies.
"""
return self._turn_start_strategies
@property
def task_manager(self) -> BaseTaskManager:
"""Get the task manager for this processor.
@@ -791,7 +777,6 @@ class FrameProcessor(BaseObject):
self._enable_metrics = frame.enable_metrics
self._enable_usage_metrics = frame.enable_usage_metrics
self._interruption_strategies = frame.interruption_strategies
self._turn_start_strategies = frame.turn_start_strategies
self._report_only_initial_ttfb = frame.report_only_initial_ttfb
self.__create_process_task()

11
src/pipecat/transports/base_input.py
View File

@@ -133,7 +133,7 @@ class BaseInputTransport(FrameProcessor):
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'turn_analyzer' is deprecated, use `PipelineTask`'s new "
"Parameter 'turn_analyzer' is deprecated, use `LLMUserAggregator`'s new "
"`turn_start_strategies` parameter instead.",
DeprecationWarning,
)
@@ -178,7 +178,7 @@ class BaseInputTransport(FrameProcessor):
.. deprecated:: 0.0.99
This method is deprecated and will be removed in a future version.
Use `PipelineTask`'s new `turn_start_strategies` parameter instead.
Use `LLMUserAggregator`'s new `turn_start_strategies` parameter instead.
Returns:
The turn analyzer instance if configured, None otherwise.
@@ -188,16 +188,11 @@ class BaseInputTransport(FrameProcessor):
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Method 'turn_analyzer' is deprecated. Use `PipelineTask`'s new "
"Method 'turn_analyzer' is deprecated. Use `LLMUserAggregator`'s new "
" `turn_start_strategies` parameter instead.",
DeprecationWarning,
)
logger.warning(
f"{self}: method 'turn_analyzer' is deprecated. Use `PipelineTask`'s new "
"`turn_start_strategies` parameter instead."
)
return self._params.turn_analyzer
async def start(self, frame: StartFrame):

2
src/pipecat/transports/base_transport.py
View File

@@ -115,7 +115,7 @@ class TransportParams(BaseModel):
turn_analyzer: Turn-taking analyzer instance for conversation management.
.. deprecated:: 0.0.99
The `turn_analyzer` parameter is deprecated, use `PipelineTask`'s
The `turn_analyzer` parameter is deprecated, use `LLMUSerAggregator`'s
new `turn_start_strategies` parameter instead.
"""