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

Handle multiple user inputs from the user when a voicemail is detected; add a configurable timeout to emitting the callback

Browse Source
This commit is contained in:
Mark Backman
2025-08-09 07:23:51 -04:00
parent 5ca82ec61e
commit 9da33f3897
2 changed files with 176 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
examples/foundational/44-voicemail-detection.py
View File

@@ -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!"),

191
src/pipecat/utils/voicemail/voicemail_detector.py
View File

@@ -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,