diff --git a/CHANGELOG.md b/CHANGELOG.md index e562995fc..b47a26116 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `UserSpeakingFrame`. This will be sent upstream and downstream while VAD + detects the user is speaking. + - Expanded support for universal `LLMContext` to more LLM services. Using the universal `LLMContext` and associated `LLMContextAggregatorPair` is a pre-requisite for using `LLMSwitcher` to switch between LLMs at runtime. @@ -79,6 +82,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `UserStartedSpeakingFrame` and `UserStoppedSpeakingFrame` are also pushed + upstream. + - `ParallelPipeline` now waits for `CancelFrame` to finish in all branches before pushing it downstream. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ac939f0fa..0d8471756 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -898,6 +898,16 @@ class UserStoppedSpeakingFrame(SystemFrame): emulated: bool = False +@dataclass +class UserSpeakingFrame(SystemFrame): + """Frame indicating the user is speaking. + + Emitted by VAD to indicate the user is speaking. + """ + + pass + + @dataclass class EmulateUserStartedSpeakingFrame(SystemFrame): """Frame to emulate user started speaking behavior. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 46c6b8312..8c1e4c044 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -32,15 +32,11 @@ from pipecat.frames.frames import ( Frame, HeartbeatFrame, InputAudioRawFrame, - InterimTranscriptionFrame, - LLMFullResponseEndFrame, MetricsFrame, StartFrame, StopFrame, StopTaskFrame, - TranscriptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, + UserSpeakingFrame, ) from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData from pipecat.observers.base_observer import BaseObserver @@ -145,14 +141,7 @@ class PipelineTask(BasePipelineTask): conversation_id: Optional[str] = None, enable_tracing: bool = False, enable_turn_tracking: bool = True, - idle_timeout_frames: Tuple[Type[Frame], ...] = ( - BotSpeakingFrame, - InterimTranscriptionFrame, - LLMFullResponseEndFrame, - TranscriptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - ), + idle_timeout_frames: Tuple[Type[Frame], ...] = (BotSpeakingFrame, UserSpeakingFrame), idle_timeout_secs: Optional[float] = IDLE_TIMEOUT_SECS, observers: Optional[List[BaseObserver]] = None, task_manager: Optional[BaseTaskManager] = None, diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 2052eed00..cb3808bf9 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -39,6 +39,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StopFrame, SystemFrame, + UserSpeakingFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, VADParamsUpdateFrame, @@ -298,10 +299,10 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, EmulateUserStartedSpeakingFrame): logger.debug("Emulating user started speaking") - await self._handle_user_interruption(UserStartedSpeakingFrame(emulated=True)) + await self._handle_user_interruption(VADState.SPEAKING, emulated=True) elif isinstance(frame, EmulateUserStoppedSpeakingFrame): logger.debug("Emulating user stopped speaking") - await self._handle_user_interruption(UserStoppedSpeakingFrame(emulated=True)) + await self._handle_user_interruption(VADState.QUIET, emulated=True) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) @@ -341,12 +342,16 @@ class BaseInputTransport(FrameProcessor): await self._start_interruption() await self.push_frame(StartInterruptionFrame()) - async def _handle_user_interruption(self, frame: Frame): + async def _handle_user_interruption(self, vad_state: VADState, emulated: bool = False): """Handle user interruption events based on speaking state.""" - if isinstance(frame, UserStartedSpeakingFrame): + if vad_state == VADState.SPEAKING: logger.debug("User started speaking") self._user_speaking = True - await self.push_frame(frame) + + upstream_frame = UserStartedSpeakingFrame(emulated=emulated) + downstream_frame = UserStartedSpeakingFrame(emulated=emulated) + await self.push_frame(downstream_frame) + await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) # Only push StartInterruptionFrame if: # 1. No interruption config is set, OR @@ -367,10 +372,15 @@ class BaseInputTransport(FrameProcessor): "User started speaking while bot is speaking with interruption config - " "deferring interruption to aggregator" ) - elif isinstance(frame, UserStoppedSpeakingFrame): + elif vad_state == VADState.QUIET: logger.debug("User stopped speaking") self._user_speaking = False - await self.push_frame(frame) + + upstream_frame = UserStoppedSpeakingFrame(emulated=emulated) + downstream_frame = UserStoppedSpeakingFrame(emulated=emulated) + await self.push_frame(downstream_frame) + await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) + if self.interruptions_allowed: await self._stop_interruption() @@ -411,7 +421,7 @@ class BaseInputTransport(FrameProcessor): ) return state - async def _handle_vad(self, audio_frame: InputAudioRawFrame, vad_state: VADState): + async def _handle_vad(self, audio_frame: InputAudioRawFrame, vad_state: VADState) -> VADState: """Handle Voice Activity Detection results and generate appropriate frames.""" new_vad_state = await self._vad_analyze(audio_frame) if ( @@ -419,7 +429,8 @@ class BaseInputTransport(FrameProcessor): and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING ): - frame = None + interruption_state = None + # If the turn analyser is enabled, this will prevent: # - Creating the UserStoppedSpeakingFrame # - Creating the UserStartedSpeakingFrame multiple times @@ -430,14 +441,14 @@ class BaseInputTransport(FrameProcessor): if new_vad_state == VADState.SPEAKING: await self.push_frame(VADUserStartedSpeakingFrame()) if can_create_user_frames: - frame = UserStartedSpeakingFrame() + interruption_state = VADState.SPEAKING elif new_vad_state == VADState.QUIET: await self.push_frame(VADUserStoppedSpeakingFrame()) if can_create_user_frames: - frame = UserStoppedSpeakingFrame() + interruption_state = VADState.QUIET - if frame: - await self._handle_user_interruption(frame) + if interruption_state: + await self._handle_user_interruption(interruption_state) vad_state = new_vad_state return vad_state @@ -452,7 +463,7 @@ class BaseInputTransport(FrameProcessor): async def _handle_end_of_turn_complete(self, state: EndOfTurnState): """Handle completion of end-of-turn analysis.""" if state == EndOfTurnState.COMPLETE: - await self._handle_user_interruption(UserStoppedSpeakingFrame()) + await self._handle_user_interruption(VADState.QUIET) async def _run_turn_analyzer( self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState @@ -489,6 +500,10 @@ class BaseInputTransport(FrameProcessor): if self._params.turn_analyzer: await self._run_turn_analyzer(frame, vad_state, previous_vad_state) + if vad_state == VADState.SPEAKING: + await self.push_frame(UserSpeakingFrame()) + await self.push_frame(UserSpeakingFrame(), FrameDirection.UPSTREAM) + # Push audio downstream if passthrough is set. if self._params.audio_in_passthrough: await self.push_frame(frame) @@ -502,7 +517,7 @@ class BaseInputTransport(FrameProcessor): vad_state = VADState.QUIET if self._params.turn_analyzer: self._params.turn_analyzer.clear() - await self._handle_user_interruption(UserStoppedSpeakingFrame()) + await self._handle_user_interruption(VADState.QUIET) async def _handle_prediction_result(self, result: MetricsData): """Handle a prediction result event from the turn analyzer."""