From 9da33f3897fdaf15022bc2459ddfba76c65d4dd6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 9 Aug 2025 07:23:51 -0400 Subject: [PATCH] Handle multiple user inputs from the user when a voicemail is detected; add a configurable timeout to emitting the callback --- .../foundational/44-voicemail-detection.py | 3 - .../utils/voicemail/voicemail_detector.py | 191 ++++++++++++++++-- 2 files changed, 176 insertions(+), 18 deletions(-) diff --git a/examples/foundational/44-voicemail-detection.py b/examples/foundational/44-voicemail-detection.py index 24f5d1762..64e78465f 100644 --- a/examples/foundational/44-voicemail-detection.py +++ b/examples/foundational/44-voicemail-detection.py @@ -61,9 +61,6 @@ async def handle_voicemail(processor): """ logger.info("Voicemail detected! Playing greeting...") - # Wait a moment for interruption to clear - await asyncio.sleep(1) - # Push frames using standard Pipecat pattern await processor.push_frame( TTSSpeakFrame("This is Mattie. Call me back when you can!"), diff --git a/src/pipecat/utils/voicemail/voicemail_detector.py b/src/pipecat/utils/voicemail/voicemail_detector.py index 97144a4e7..4a6b070c2 100644 --- a/src/pipecat/utils/voicemail/voicemail_detector.py +++ b/src/pipecat/utils/voicemail/voicemail_detector.py @@ -28,8 +28,12 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, StartFrame, + StartInterruptionFrame, + StopInterruptionFrame, TTSAudioRawFrame, TTSTextFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext @@ -48,18 +52,19 @@ class ClassifierGate(FrameProcessor): preventing unnecessary LLM calls and maintaining system efficiency. The gate allows all frames to pass through while open, but once closed, only - allows system frames (interruptions, end frames, cancel frames) to continue. + allows system frames and user speaking frames to continue. Speaking frames + are needed for voicemail timing control. """ - def __init__(self, notifier: BaseNotifier): + def __init__(self, gate_notifier: BaseNotifier): """Initialize the classifier gate. Args: - notifier: Notifier that signals when a classification decision has + gate_notifier: Notifier that signals when a classification decision has been made and the gate should close. """ super().__init__() - self._notifier = notifier + self._gate_notifier = gate_notifier self._gate_opened = True self._gate_task: Optional[asyncio.Task] = None @@ -86,7 +91,18 @@ class ClassifierGate(FrameProcessor): if self._gate_opened: await self.push_frame(frame, direction) elif not self._gate_opened and isinstance( - frame, (BotInterruptionFrame, EndTaskFrame, EndFrame, CancelTaskFrame, CancelFrame) + frame, + ( + BotInterruptionFrame, + EndTaskFrame, + EndFrame, + CancelTaskFrame, + CancelFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + StartInterruptionFrame, + StopInterruptionFrame, + ), ): await self.push_frame(frame, direction) @@ -98,7 +114,7 @@ class ClassifierGate(FrameProcessor): closes permanently to stop further classification processing. """ try: - await self._notifier.wait() + await self._gate_notifier.wait() if self._gate_opened: self._gate_opened = False @@ -111,6 +127,86 @@ class ClassifierGate(FrameProcessor): raise +class ConversationGate(FrameProcessor): + """Gate processor that blocks conversation flow when voicemail is detected. + + This gate starts open to allow normal conversation processing but closes + permanently when voicemail is detected. This prevents the main conversation + LLM from processing additional input after voicemail classification. + """ + + def __init__(self, voicemail_notifier: BaseNotifier): + """Initialize the conversation gate. + + Args: + voicemail_notifier: Notifier that signals when voicemail has been + detected and the conversation should be blocked. + """ + super().__init__() + self._voicemail_notifier = voicemail_notifier + self._gate_opened = True + self._voicemail_task: Optional[asyncio.Task] = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and control gate state based on voicemail detection. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + # Start the notification waiting task immediately + self._voicemail_task = self.create_task(self._wait_for_voicemail()) + + elif isinstance(frame, (EndFrame, CancelFrame)): + # Clean up the task when pipeline ends or is cancelled + if self._voicemail_task: + await self.cancel_task(self._voicemail_task) + self._voicemail_task = None + + # Gate logic: open gate allows all frames, closed gate blocks everything + if self._gate_opened: + await self.push_frame(frame, direction) + elif not self._gate_opened and isinstance( + frame, + ( + BotInterruptionFrame, + EndTaskFrame, + EndFrame, + CancelTaskFrame, + CancelFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + StartInterruptionFrame, + StopInterruptionFrame, + ), + ): + # Only allow system frames and user speaking frames through when closed + await self.push_frame(frame, direction) + # When closed, don't push any frames (complete conversation blocking) + + async def _wait_for_voicemail(self): + """Wait for voicemail detection notification and close the gate. + + This method blocks until voicemail is detected, then closes the gate + permanently to prevent any further conversation processing. + """ + try: + await self._voicemail_notifier.wait() + + if self._gate_opened: + self._gate_opened = False + logger.debug(f"{self}: Conversation gate closed - voicemail detected") + except asyncio.CancelledError: + logger.debug(f"{self}: Conversation gate task was cancelled") + raise + except Exception as e: + logger.exception(f"{self}: Error in conversation gate task: {e}") + raise + + class VoicemailProcessor(FrameProcessor): """Processor that handles LLM classification responses and triggers callbacks. @@ -123,6 +219,9 @@ class VoicemailProcessor(FrameProcessor): The processor expects responses containing either "CONVERSATION" (indicating a human answered) or "VOICEMAIL" (indicating an automated system). Once a decision is made, it triggers the appropriate notifications and callbacks. + + For voicemail detection, the callback timer starts immediately and is cancelled + and restarted based on user speech patterns to ensure proper timing. """ def __init__( @@ -132,6 +231,7 @@ class VoicemailProcessor(FrameProcessor): conversation_notifier: BaseNotifier, voicemail_notifier: BaseNotifier, on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None, + voicemail_response_delay: float, ): """Initialize the voicemail processor. @@ -145,18 +245,26 @@ class VoicemailProcessor(FrameProcessor): on_voicemail_detected: Optional callback function called when voicemail is detected. The callback receives this processor instance and can use it to push custom frames (like voicemail greetings). + voicemail_response_delay: Delay in seconds after user stops speaking + before triggering the voicemail callback. This ensures the voicemail + greeting or user message is complete before responding. """ super().__init__() self._gate_notifier = gate_notifier self._conversation_notifier = conversation_notifier self._voicemail_notifier = voicemail_notifier self._on_voicemail_detected = on_voicemail_detected + self._voicemail_response_delay = voicemail_response_delay # Aggregation state for collecting complete LLM responses self._processing_response = False self._response_buffer = "" self._decision_made = False + # Voicemail timing state + self._voicemail_detected = False + self._voicemail_callback_task: Optional[asyncio.Task] = None + async def process_frame(self, frame: Frame, direction: FrameDirection): """Process frames and handle LLM classification responses. @@ -164,6 +272,7 @@ class VoicemailProcessor(FrameProcessor): 1. LLMFullResponseStartFrame: Begin collecting tokens 2. LLMTextFrame: Accumulate text tokens into buffer 3. LLMFullResponseEndFrame: Process complete response and make decision + 4. UserStartedSpeakingFrame/UserStoppedSpeakingFrame: Manage voicemail timing Args: frame: The frame to process. @@ -189,9 +298,24 @@ class VoicemailProcessor(FrameProcessor): self._response_buffer += frame.text logger.trace(f"{self}: Buffer: '{self._response_buffer}'") + elif isinstance(frame, UserStartedSpeakingFrame): + # User started speaking - cancel voicemail callback timer + if self._voicemail_callback_task: + logger.debug(f"{self}: User started speaking, cancelling voicemail callback") + await self.cancel_task(self._voicemail_callback_task) + self._voicemail_callback_task = None + + elif isinstance(frame, UserStoppedSpeakingFrame): + # User stopped speaking - restart voicemail callback timer if voicemail detected + if self._voicemail_detected and not self._voicemail_callback_task: + logger.debug( + f"{self}: User stopped speaking, restarting voicemail callback timer ({self._voicemail_response_delay}s)" + ) + self._voicemail_callback_task = self.create_task(self._delayed_voicemail_callback()) + else: # Pass all non-LLM frames through - # Blocking LLM frames prevents interferences with the downsteram LLM + # Blocking LLM frames prevents interference with the downstream LLM await self.push_frame(frame, direction) async def _process_classification(self, full_response: str): @@ -214,29 +338,58 @@ class VoicemailProcessor(FrameProcessor): if "CONVERSATION" in response: # Human answered - continue normal conversation flow self._decision_made = True - logger.info(f"{self}: CONVERSATION detected - releasing TTS buffer") + logger.info(f"{self}: CONVERSATION detected") await self._gate_notifier.notify() # Close the classifier gate await self._conversation_notifier.notify() # Release buffered TTS frames elif "VOICEMAIL" in response: # Voicemail detected - trigger voicemail handling self._decision_made = True - logger.info(f"{self}: VOICEMAIL detected - clearing TTS buffer and triggering callback") + self._voicemail_detected = True + logger.info(f"{self}: VOICEMAIL detected") await self._gate_notifier.notify() # Close the classifier gate await self._voicemail_notifier.notify() # Clear buffered TTS frames # Interrupt the current pipeline to stop any ongoing processing await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - # Execute developer callback for custom voicemail handling + # Always start the callback timer immediately + # It will be cancelled and restarted if user starts/stops speaking + if not self._voicemail_callback_task: + logger.debug( + f"{self}: Starting voicemail callback timer ({self._voicemail_response_delay}s)" + ) + self._voicemail_callback_task = self.create_task(self._delayed_voicemail_callback()) + + else: + # Unexpected response - log warning but don't crash + logger.warning(f"{self}: Unexpected classification response: '{full_response}'") + + async def _delayed_voicemail_callback(self): + """Execute the voicemail callback after the configured delay. + + This method waits for the specified delay period, then triggers the + developer's voicemail callback. The timer can be cancelled and restarted + based on user speech patterns to ensure proper timing. + """ + try: + logger.debug( + f"{self}: Waiting {self._voicemail_response_delay}s before voicemail callback" + ) + await asyncio.sleep(self._voicemail_response_delay) + + logger.info(f"{self}: Executing voicemail callback") if self._on_voicemail_detected: try: await self._on_voicemail_detected(self) except Exception as e: logger.exception(f"{self}: Error in voicemail callback: {e}") - else: - # Unexpected response - log warning but don't crash - logger.warning(f"{self}: Unexpected classification response: '{full_response}'") + + except asyncio.CancelledError: + logger.debug(f"{self}: Voicemail callback timer was cancelled") + raise + finally: + self._voicemail_callback_task = None class VoicemailBuffer(FrameProcessor): @@ -432,6 +585,7 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo llm: LLMService, on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None, system_prompt: Optional[str] = None, + voicemail_response_delay: float = 2.0, ): """Initialize the voicemail detector with classification and buffering components. @@ -445,9 +599,14 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo uses the default prompt optimized for outbound calling scenarios. Custom prompts should instruct the LLM to respond with exactly "CONVERSATION" or "VOICEMAIL" for proper detection functionality. + voicemail_response_delay: Delay in seconds after user stops speaking + before triggering the voicemail callback. This allows voicemail + responses to be played back after a short delay to ensure the response + occurs during the voicemail recording. Default is 2.0 seconds. """ self._classifier_llm = llm self._prompt = system_prompt if system_prompt is not None else self.DEFAULT_SYSTEM_PROMPT + self._voicemail_response_delay = voicemail_response_delay # Validate custom prompts to ensure they work with the detection logic if system_prompt is not None: @@ -472,11 +631,13 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo # Create the processor components self._classifier_gate = ClassifierGate(self._gate_notifier) + self._conversation_gate = ConversationGate(self._voicemail_notifier) self._voicemail_processor = VoicemailProcessor( gate_notifier=self._gate_notifier, conversation_notifier=self._conversation_notifier, voicemail_notifier=self._voicemail_notifier, on_voicemail_detected=on_voicemail_detected, + voicemail_response_delay=voicemail_response_delay, ) self._voicemail_buffer = VoicemailBuffer( self._conversation_notifier, self._voicemail_notifier @@ -484,8 +645,8 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo # Initialize the parallel pipeline with conversation and classifier branches super().__init__( - # Conversation branch: empty pipeline for normal frame flow - [], + # Conversation branch: gate to blocks after voicemail detection + [self._conversation_gate], # Classification branch: gate -> context -> LLM -> processor -> context [ self._classifier_gate,