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

Class name changes, add TTSStarted/StoppedFrame to the TTSBuffer

Browse Source
This commit is contained in:
Mark Backman
2025-08-09 08:10:31 -04:00
parent 9da33f3897
commit 5a07b30c7a
2 changed files with 26 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -97,11 +97,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
[ [
transport.input(), transport.input(),
stt, stt,
voicemail.detector(), # Voicemail detection voicemail.detector(), # Voicemail detection — between STT and User context aggregator
context_aggregator.user(), context_aggregator.user(),
llm, llm,
tts, tts,
voicemail.buffer(), # TTS buffering voicemail.buffer(), # TTS buffering — Immediately after the TTS service
transport.output(), transport.output(),
context_aggregator.assistant(), context_aggregator.assistant(),
] ]

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

@@ -31,6 +31,8 @@ from pipecat.frames.frames import (
StartInterruptionFrame, StartInterruptionFrame,
StopInterruptionFrame, StopInterruptionFrame,
TTSAudioRawFrame, TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
TTSTextFrame, TTSTextFrame,
UserStartedSpeakingFrame, UserStartedSpeakingFrame,
UserStoppedSpeakingFrame, UserStoppedSpeakingFrame,
@@ -109,7 +111,7 @@ class ClassifierGate(FrameProcessor):
async def _wait_for_notification(self): async def _wait_for_notification(self):
"""Wait for classification decision notification and close the gate. """Wait for classification decision notification and close the gate.
This method blocks until the VoicemailProcessor makes a classification This method blocks until the ClassificationProcessor makes a classification
decision and signals through the notifier. Once notified, the gate decision and signals through the notifier. Once notified, the gate
closes permanently to stop further classification processing. closes permanently to stop further classification processing.
""" """
@@ -207,7 +209,7 @@ class ConversationGate(FrameProcessor):
raise raise
class VoicemailProcessor(FrameProcessor): class ClassificationProcessor(FrameProcessor):
"""Processor that handles LLM classification responses and triggers callbacks. """Processor that handles LLM classification responses and triggers callbacks.
This processor aggregates LLM text tokens into complete responses and analyzes This processor aggregates LLM text tokens into complete responses and analyzes
@@ -230,7 +232,9 @@ class VoicemailProcessor(FrameProcessor):
gate_notifier: BaseNotifier, gate_notifier: BaseNotifier,
conversation_notifier: BaseNotifier, conversation_notifier: BaseNotifier,
voicemail_notifier: BaseNotifier, voicemail_notifier: BaseNotifier,
on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None, on_voicemail_detected: Optional[
Callable[["ClassificationProcessor"], Awaitable[None]]
] = None,
voicemail_response_delay: float, voicemail_response_delay: float,
): ):
"""Initialize the voicemail processor. """Initialize the voicemail processor.
@@ -238,9 +242,9 @@ class VoicemailProcessor(FrameProcessor):
Args: Args:
gate_notifier: Notifier to signal the ClassifierGate about classification gate_notifier: Notifier to signal the ClassifierGate about classification
decisions so it can close and stop processing. decisions so it can close and stop processing.
conversation_notifier: Notifier to signal the VoicemailBuffer to release conversation_notifier: Notifier to signal the TTSBuffer to release
all buffered TTS frames for normal conversation flow. all buffered TTS frames for normal conversation flow.
voicemail_notifier: Notifier to signal the VoicemailBuffer to clear voicemail_notifier: Notifier to signal the TTSBuffer to clear
buffered TTS frames since voicemail was detected. buffered TTS frames since voicemail was detected.
on_voicemail_detected: Optional callback function called when voicemail on_voicemail_detected: Optional callback function called when voicemail
is detected. The callback receives this processor instance and can is detected. The callback receives this processor instance and can
@@ -392,7 +396,7 @@ class VoicemailProcessor(FrameProcessor):
self._voicemail_callback_task = None self._voicemail_callback_task = None
class VoicemailBuffer(FrameProcessor): class TTSBuffer(FrameProcessor):
"""Buffers TTS frames until voicemail classification decision is made. """Buffers TTS frames until voicemail classification decision is made.
This processor holds TTS output frames in a buffer while the voicemail This processor holds TTS output frames in a buffer while the voicemail
@@ -455,7 +459,9 @@ class VoicemailBuffer(FrameProcessor):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
# Core buffering logic: hold TTS frames, pass everything else through # Core buffering logic: hold TTS frames, pass everything else through
elif self._buffering_active and isinstance(frame, (TTSTextFrame, TTSAudioRawFrame)): elif self._buffering_active and isinstance(
frame, (TTSStartedFrame, TTSStoppedFrame, TTSTextFrame, TTSAudioRawFrame)
):
# Buffer TTS frames while waiting for classification decision # Buffer TTS frames while waiting for classification decision
self._frame_buffer.append((frame, direction)) self._frame_buffer.append((frame, direction))
else: else:
@@ -583,22 +589,24 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
self, self,
*, *,
llm: LLMService, llm: LLMService,
on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None,
system_prompt: Optional[str] = None, system_prompt: Optional[str] = None,
voicemail_response_delay: float = 2.0, on_voicemail_detected: Optional[
Callable[["ClassificationProcessor"], Awaitable[None]]
] = None,
voicemail_response_delay: Optional[float] = 2.0,
): ):
"""Initialize the voicemail detector with classification and buffering components. """Initialize the voicemail detector with classification and buffering components.
Args: Args:
llm: LLM service used for voicemail vs conversation classification. llm: LLM service used for voicemail vs conversation classification.
Should be fast and reliable for real-time classification. Should be fast and reliable for real-time classification.
on_voicemail_detected: Optional callback function invoked when voicemail
is detected. Receives the VoicemailProcessor instance which can be
used to push frames (like custom voicemail greetings).
system_prompt: Optional custom system prompt for classification. If None, system_prompt: Optional custom system prompt for classification. If None,
uses the default prompt optimized for outbound calling scenarios. uses the default prompt optimized for outbound calling scenarios.
Custom prompts should instruct the LLM to respond with exactly Custom prompts should instruct the LLM to respond with exactly
"CONVERSATION" or "VOICEMAIL" for proper detection functionality. "CONVERSATION" or "VOICEMAIL" for proper detection functionality.
on_voicemail_detected: Optional callback function invoked when voicemail
is detected. Receives the ClassificationProcessor instance which can be
used to push frames (like custom voicemail greetings).
voicemail_response_delay: Delay in seconds after user stops speaking voicemail_response_delay: Delay in seconds after user stops speaking
before triggering the voicemail callback. This allows voicemail before triggering the voicemail callback. This allows voicemail
responses to be played back after a short delay to ensure the response responses to be played back after a short delay to ensure the response
@@ -632,16 +640,14 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
# Create the processor components # Create the processor components
self._classifier_gate = ClassifierGate(self._gate_notifier) self._classifier_gate = ClassifierGate(self._gate_notifier)
self._conversation_gate = ConversationGate(self._voicemail_notifier) self._conversation_gate = ConversationGate(self._voicemail_notifier)
self._voicemail_processor = VoicemailProcessor( self._voicemail_processor = ClassificationProcessor(
gate_notifier=self._gate_notifier, gate_notifier=self._gate_notifier,
conversation_notifier=self._conversation_notifier, conversation_notifier=self._conversation_notifier,
voicemail_notifier=self._voicemail_notifier, voicemail_notifier=self._voicemail_notifier,
on_voicemail_detected=on_voicemail_detected, on_voicemail_detected=on_voicemail_detected,
voicemail_response_delay=voicemail_response_delay, voicemail_response_delay=voicemail_response_delay,
) )
self._voicemail_buffer = VoicemailBuffer( self._voicemail_buffer = TTSBuffer(self._conversation_notifier, self._voicemail_notifier)
self._conversation_notifier, self._voicemail_notifier
)
# Initialize the parallel pipeline with conversation and classifier branches # Initialize the parallel pipeline with conversation and classifier branches
super().__init__( super().__init__(
@@ -688,13 +694,13 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
""" """
return self return self
def buffer(self) -> VoicemailBuffer: def buffer(self) -> TTSBuffer:
"""Get the buffer processor for placement after TTS in the main pipeline. """Get the buffer processor for placement after TTS in the main pipeline.
This should be placed after the TTS service and before the transport This should be placed after the TTS service and before the transport
output to enable TTS frame buffering during classification. output to enable TTS frame buffering during classification.
Returns: Returns:
The VoicemailBuffer processor instance. The TTSBuffer processor instance.
""" """
return self._voicemail_buffer return self._voicemail_buffer