Merge pull request #2578 from pipecat-ai/aleix/user-speaking-frame

add UserSpeakingFrame and UserStartedSpeakingFrame/UserStopeedSpeakingFrame updates

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.
 

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.

src/pipecat/pipeline/task.py

@@ -32,15 +32,11 @@ from pipecat.frames.frames import (
     Frame,
     HeartbeatFrame,
     InputAudioRawFrame,
-    InterimTranscriptionFrame,
-    LLMFullResponseEndFrame,
     MetricsFrame,
     StartFrame,
     StopFrame,
     StopTaskFrame,
-    TranscriptionFrame,
+    UserSpeakingFrame,
-    UserStartedSpeakingFrame,
-    UserStoppedSpeakingFrame,
 )
 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], ...] = (
+        idle_timeout_frames: Tuple[Type[Frame], ...] = (BotSpeakingFrame, UserSpeakingFrame),
-            BotSpeakingFrame,
-            InterimTranscriptionFrame,
-            LLMFullResponseEndFrame,
-            TranscriptionFrame,
-            UserStartedSpeakingFrame,
-            UserStoppedSpeakingFrame,
-        ),
         idle_timeout_secs: Optional[float] = IDLE_TIMEOUT_SECS,
         observers: Optional[List[BaseObserver]] = None,
         task_manager: Optional[BaseTaskManager] = None,

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:
+            if interruption_state:
-                await self._handle_user_interruption(frame)
+                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."""