diff --git a/CHANGELOG.md b/CHANGELOG.md index ee2a534f0..e15588e73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `SpeechControlParamsFrame`, a new `SystemFrame` that notifies + downstream processors of the VAD and Turn analyzer params. This frame is + pushed by the `BaseInputTransport` at Start and any time a + `VADParamsUpdateFrame` is received. + ### Changed - Two package dependencies have been updated: @@ -25,6 +32,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue in ParallelPipeline that caused errors when attempting to drain the queues. +- Fixed an issue with emulated VAD timeout inconsistency in + `LLMUserContextAggregator`. Previously, emulated VAD scenarios (where + transcription is received without VAD detection) used a hardcoded + `aggregation_timeout` (default 0.5s) instead of matching the VAD's + `stop_secs` parameter (default 0.8s). This created different user experiences + between real VAD and emulated VAD scenarios. Now, emulated VAD timeouts + automatically synchronize with the VAD's `stop_secs` parameter. + - Fix a pipeline freeze when using AWS Nova Sonic, which would occur if the user started early, while the bot was still working through `trigger_assistant_response()`. diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index 642173852..f64655fb6 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -76,6 +76,16 @@ class BaseTurnAnalyzer(ABC): """ pass + @property + @abstractmethod + def params(self): + """Get the current turn analyzer parameters. + + Returns: + Current turn analyzer configuration parameters. + """ + pass + @abstractmethod def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: """Appends audio data for analysis. diff --git a/src/pipecat/audio/turn/smart_turn/base_smart_turn.py b/src/pipecat/audio/turn/smart_turn/base_smart_turn.py index 38b5410ec..f72ac7d84 100644 --- a/src/pipecat/audio/turn/smart_turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/base_smart_turn.py @@ -87,6 +87,15 @@ class BaseSmartTurn(BaseTurnAnalyzer): """ return self._speech_triggered + @property + def params(self) -> SmartTurnParams: + """Get the current smart turn parameters. + + Returns: + Current smart turn configuration parameters. + """ + return self._params + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: """Append audio data for turn analysis. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index c7d7b4c75..9d73b44ff 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -28,6 +28,7 @@ from typing import ( ) from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.metrics.metrics import MetricsData from pipecat.transcriptions.language import Language @@ -1145,6 +1146,23 @@ class OutputDTMFUrgentFrame(DTMFFrame, SystemFrame): pass +@dataclass +class SpeechControlParamsFrame(SystemFrame): + """Frame for notifying processors of speech control parameter changes. + + This includes parameters for both VAD (Voice Activity Detection) and + turn-taking analysis. It allows downstream processors to adjust their + behavior based on updated interaction control settings. + + Parameters: + vad_params: Current VAD parameters. + turn_params: Current turn-taking analysis parameters. + """ + + vad_params: Optional[VADParams] = None + turn_params: Optional[SmartTurnParams] = None + + # # Control frames # diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 99834447f..8c9212ea0 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -19,6 +19,8 @@ from typing import Dict, List, Literal, Optional, Set from loguru import logger from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( BotInterruptionFrame, BotStartedSpeakingFrame, @@ -43,6 +45,7 @@ from pipecat.frames.frames import ( LLMSetToolsFrame, LLMTextFrame, OpenAILLMContextAssistantTimestampFrame, + SpeechControlParamsFrame, StartFrame, StartInterruptionFrame, TextFrame, @@ -67,9 +70,13 @@ class LLMUserAggregatorParams: aggregation_timeout: Maximum time in seconds to wait for additional transcription content before pushing aggregated result. This timeout is used only when the transcription is slow to arrive. + turn_emulated_vad_timeout: Maximum time in seconds to wait for emulated + VAD when using turn-based analysis. Applied when transcription is + received but VAD didn't detect speech (e.g., whispered utterances). """ aggregation_timeout: float = 0.5 + turn_emulated_vad_timeout: float = 0.8 @dataclass @@ -390,6 +397,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): """ super().__init__(context=context, role="user", **kwargs) self._params = params or LLMUserAggregatorParams() + self._vad_params: Optional[VADParams] = None + self._turn_params: Optional[SmartTurnParams] = None + if "aggregation_timeout" in kwargs: import warnings @@ -477,6 +487,10 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) + elif isinstance(frame, SpeechControlParamsFrame): + self._vad_params = frame.vad_params + self._turn_params = frame.turn_params + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -618,9 +632,40 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): async def _aggregation_task_handler(self): while True: try: - await asyncio.wait_for( - self._aggregation_event.wait(), self._params.aggregation_timeout - ) + # The _aggregation_task_handler handles two distinct timeout scenarios: + # + # 1. When emulating_vad=True: Wait for emulated VAD timeout before + # pushing aggregation (simulating VAD behavior when no actual VAD + # detection occurred). + # + # 2. When emulating_vad=False: Use aggregation_timeout as a buffer + # to wait for potential late-arriving transcription frames after + # a real VAD event. + # + # For emulated VAD scenarios, the timeout strategy depends on whether + # a turn analyzer is configured: + # + # - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because + # the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech + # chunking to feed the turn analyzer. This low value is too fast + # for emulated VAD scenarios where we need to allow users time to + # finish speaking (e.g. 0.8s). + # + # - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain + # consistent user experience between real VAD detection and + # emulated VAD scenarios. + if not self._emulating_vad: + timeout = self._params.aggregation_timeout + elif self._turn_params: + timeout = self._params.turn_emulated_vad_timeout + else: + # Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params + timeout = ( + self._vad_params.stop_secs + if self._vad_params + else self._params.turn_emulated_vad_timeout + ) + await asyncio.wait_for(self._aggregation_event.wait(), timeout) await self._maybe_emulate_user_speaking() except asyncio.TimeoutError: if not self._user_speaking: diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 5b71ba69c..bbb9179bc 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -34,6 +34,7 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputImageRawFrame, MetricsFrame, + SpeechControlParamsFrame, StartFrame, StartInterruptionFrame, StopFrame, @@ -195,6 +196,13 @@ class BaseInputTransport(FrameProcessor): if self._params.turn_analyzer: self._params.turn_analyzer.set_sample_rate(self._sample_rate) + if self._params.vad_analyzer or self._params.turn_analyzer: + vad_params = self._params.vad_analyzer.params if self._params.vad_analyzer else None + turn_params = self._params.turn_analyzer.params if self._params.turn_analyzer else None + + speech_frame = SpeechControlParamsFrame(vad_params=vad_params, turn_params=turn_params) + await self.push_frame(speech_frame) + # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) @@ -310,6 +318,13 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, VADParamsUpdateFrame): if self.vad_analyzer: self.vad_analyzer.set_params(frame.params) + speech_frame = SpeechControlParamsFrame( + vad_params=frame.params, + turn_params=self._params.turn_analyzer.params + if self._params.turn_analyzer + else None, + ) + await self.push_frame(speech_frame) elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter: await self._params.audio_in_filter.process_frame(frame) # Other frames diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index 75a81aaac..232fa59cc 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -8,6 +8,8 @@ import json import unittest from typing import Any +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame, @@ -18,6 +20,7 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, OpenAILLMContextAssistantTimestampFrame, + SpeechControlParamsFrame, StartInterruptionFrame, TextFrame, TranscriptionFrame, @@ -284,6 +287,7 @@ class BaseTestUserContextAggregator: context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) ) frames_to_send = [ + SpeechControlParamsFrame(vad_params=VADParams(stop_secs=AGGREGATION_TIMEOUT)), UserStartedSpeakingFrame(), TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""), SleepFrame(), @@ -292,6 +296,7 @@ class BaseTestUserContextAggregator: SleepFrame(sleep=AGGREGATION_SLEEP), ] expected_down_frames = [ + SpeechControlParamsFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, *self.EXPECTED_CONTEXT_FRAMES, @@ -368,14 +373,51 @@ class BaseTestUserContextAggregator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) - ) + context + ) # No aggregation timeout; this tests VAD emulation + frames_to_send = [ + SpeechControlParamsFrame(vad_params=VADParams(stop_secs=AGGREGATION_TIMEOUT)), TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""), SleepFrame(sleep=AGGREGATION_SLEEP), ] - expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES] + expected_down_frames = [ + SpeechControlParamsFrame, + *self.EXPECTED_CONTEXT_FRAMES, + ] expected_up_frames = [EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame] + + await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + self.check_message_content(context, 0, "Hello!") + + async def test_t_with_turn_analyzer(self): + assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass" + assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" + + context = self.CONTEXT_CLASS() + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(turn_emulated_vad_timeout=AGGREGATION_TIMEOUT) + ) + + frames_to_send = [ + SpeechControlParamsFrame( + vad_params=VADParams(stop_secs=0.2), + turn_params=SmartTurnParams(stop_secs=3.0), # Turn analyzer present + ), + TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""), + SleepFrame(sleep=AGGREGATION_SLEEP), + ] + expected_down_frames = [ + SpeechControlParamsFrame, + *self.EXPECTED_CONTEXT_FRAMES, + ] + expected_up_frames = [EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame] + await run_test( aggregator, frames_to_send=frames_to_send, @@ -390,15 +432,16 @@ class BaseTestUserContextAggregator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) - ) + context + ) # No aggregation timeout; this tests VAD emulation frames_to_send = [ + SpeechControlParamsFrame(vad_params=VADParams(stop_secs=AGGREGATION_TIMEOUT)), InterimTranscriptionFrame(text="Hello ", user_id="cat", timestamp=""), SleepFrame(), TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""), SleepFrame(sleep=AGGREGATION_SLEEP), ] - expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES] + expected_down_frames = [SpeechControlParamsFrame, *self.EXPECTED_CONTEXT_FRAMES] expected_up_frames = [EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame] await run_test( aggregator,