Final docstrings, comments, and cleanup

examples/foundational/44-voicemail-detection.py

@@ -82,9 +82,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-    voicemail_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    classifier_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-    voicemail = VoicemailDetector(llm=voicemail_llm, on_voicemail_detected=handle_voicemail)
+    voicemail = VoicemailDetector(llm=classifier_llm, on_voicemail_detected=handle_voicemail)
 
     messages = [
         {
@@ -98,15 +98,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     pipeline = Pipeline(
         [
-            transport.input(),  # Transport user input
+            transport.input(),
             stt,
-            voicemail.detector(),
+            voicemail.detector(),  # Voicemail detection
-            context_aggregator.user(),  # User responses
+            context_aggregator.user(),
-            llm,  # LLM
+            llm,
-            tts,  # TTS
+            tts,
-            voicemail.buffer(),
+            voicemail.buffer(),  # TTS buffering
-            transport.output(),  # Transport bot output
+            transport.output(),
-            context_aggregator.assistant(),  # Assistant spoken responses
+            context_aggregator.assistant(),
         ]
     )
 

src/pipecat/utils/voicemail/voicemail_detector.py

@@ -8,7 +8,8 @@
 
 This module provides voicemail detection capabilities using parallel pipeline
 processing to classify incoming calls as either voicemail messages or live
-conversations.
+conversations. It's specifically designed for outbound calling scenarios where
+a bot needs to determine if a human answered or if the call went to voicemail.
 """
 
 import asyncio
@@ -41,16 +42,21 @@ from pipecat.sync.event_notifier import EventNotifier
 class ClassifierGate(FrameProcessor):
     """Gate processor that controls frame flow based on classification decisions.
 
-    The gate starts open and closes permanently once a classification decision
+    The gate starts open to allow initial classification processing and closes
-    is made (CONVERSATION or VOICEMAIL). This ensures the classifier only runs until
+    permanently once a classification decision is made (CONVERSATION or VOICEMAIL).
-    a definitive decision is reached.
+    This ensures the classifier only runs until a definitive decision is reached,
+    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.
     """
 
     def __init__(self, notifier: BaseNotifier):
         """Initialize the classifier gate.
 
         Args:
-            notifier: Notifier that signals when to close the gate.
+            notifier: Notifier that signals when a classification decision has
+                been made and the gate should close.
         """
         super().__init__()
         self._notifier = notifier
@@ -58,24 +64,25 @@ class ClassifierGate(FrameProcessor):
         self._gate_task: Optional[asyncio.Task] = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames and control gate state.
+        """Process frames and control gate state based on classification decisions.
 
         Args:
             frame: The frame to process.
-            direction: The direction of frame flow.
+            direction: The direction of frame flow in the pipeline.
         """
         await super().process_frame(frame, direction)
 
         if isinstance(frame, StartFrame):
-            # Start the task immediately, don't wait for other conditions
+            # Start the notification waiting task immediately
             self._gate_task = self.create_task(self._wait_for_notification())
-            logger.info(f"{self}: Gate task started, waiting for notification")
 
         elif isinstance(frame, (EndFrame, CancelFrame)):
+            # Clean up the gate task when pipeline ends or is cancelled
             if self._gate_task:
                 await self.cancel_task(self._gate_task)
                 self._gate_task = None
 
+        # Gate logic: open gate allows all frames, closed gate only allows specific system frames
         if self._gate_opened:
             await self.push_frame(frame, direction)
         elif not self._gate_opened and isinstance(
@@ -84,15 +91,18 @@ class ClassifierGate(FrameProcessor):
             await self.push_frame(frame, direction)
 
     async def _wait_for_notification(self):
-        """Wait for classification decision notification."""
+        """Wait for classification decision notification and close the gate.
+
+        This method blocks until the VoicemailProcessor makes a classification
+        decision and signals through the notifier. Once notified, the gate
+        closes permanently to stop further classification processing.
+        """
         try:
-            logger.info(f"{self}: Waiting for notification...")
             await self._notifier.wait()
-            logger.info(f"{self}: Received notification!")
 
             if self._gate_opened:
                 self._gate_opened = False
-                logger.info(f"{self}: Gate closed")
+                logger.debug(f"{self}: Gate closed - classification complete")
         except asyncio.CancelledError:
             logger.debug(f"{self}: Gate task was cancelled")
             raise
@@ -104,26 +114,37 @@ class ClassifierGate(FrameProcessor):
 class VoicemailProcessor(FrameProcessor):
     """Processor that handles LLM classification responses and triggers callbacks.
 
-    Processes LLM text responses to determine if the call is a voicemail (VOICEMAIL)
+    This processor aggregates LLM text tokens into complete responses and analyzes
-    or conversation (CONVERSATION), then triggers appropriate actions including developer
+    them to determine if the call reached a voicemail system or a live person.
-    callbacks for voicemail detection.
+    It uses the LLM response frame delimiters (LLMFullResponseStartFrame and
+    LLMFullResponseEndFrame) to ensure complete token aggregation regardless
+    of how the LLM tokenizes the response words.
+
+    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.
     """
 
     def __init__(
         self,
         *,
         gate_notifier: BaseNotifier,
-        conversation_notifier: BaseNotifier,  # Buffer should release frames
+        conversation_notifier: BaseNotifier,
-        voicemail_notifier: BaseNotifier,  # Buffer should clear frames
+        voicemail_notifier: BaseNotifier,
         on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None,
     ):
         """Initialize the voicemail processor.
 
         Args:
-            gate_notifier: Notifier to signal gate about classification decisions.
+            gate_notifier: Notifier to signal the ClassifierGate about classification
-            conversation_notifier: Notifier to signal buffer to release frames.
+                decisions so it can close and stop processing.
-            voicemail_notifier: Notifier to signal buffer to clear frames.
+            conversation_notifier: Notifier to signal the VoicemailBuffer to release
-            on_voicemail_detected: Callback function called when voicemail is detected.
+                all buffered TTS frames for normal conversation flow.
+            voicemail_notifier: Notifier to signal the VoicemailBuffer to clear
+                buffered TTS frames since voicemail was detected.
+            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).
         """
         super().__init__()
         self._gate_notifier = gate_notifier
@@ -131,90 +152,117 @@ class VoicemailProcessor(FrameProcessor):
         self._voicemail_notifier = voicemail_notifier
         self._on_voicemail_detected = on_voicemail_detected
 
-        # Aggregation state
+        # Aggregation state for collecting complete LLM responses
-        self._aggregating = False
+        self._processing_response = False
         self._response_buffer = ""
         self._decision_made = False
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames and handle LLM classification responses.
 
+        This method implements a state machine for aggregating LLM responses:
+        1. LLMFullResponseStartFrame: Begin collecting tokens
+        2. LLMTextFrame: Accumulate text tokens into buffer
+        3. LLMFullResponseEndFrame: Process complete response and make decision
+
         Args:
             frame: The frame to process.
-            direction: The direction of frame flow.
+            direction: The direction of frame flow in the pipeline.
         """
         await super().process_frame(frame, direction)
 
         if isinstance(frame, LLMFullResponseStartFrame):
-            # Start aggregating the LLM response
+            # Begin aggregating a new LLM response
-            self._aggregating = True
+            self._processing_response = True
             self._response_buffer = ""
             logger.debug(f"{self}: Starting LLM response aggregation")
 
         elif isinstance(frame, LLMFullResponseEndFrame):
-            # End of LLM response - make decision
+            # Complete response received - make classification decision
-            if self._aggregating and not self._decision_made:
+            if self._processing_response and not self._decision_made:
                 await self._process_classification(self._response_buffer.strip())
-            self._aggregating = False
+            self._processing_response = False
             self._response_buffer = ""
 
-        elif isinstance(frame, LLMTextFrame) and self._aggregating:
+        elif isinstance(frame, LLMTextFrame) and self._processing_response:
-            # Accumulate text tokens
+            # Accumulate text tokens from the streaming LLM response
             self._response_buffer += frame.text
-            logger.debug(f"{self}: Accumulated: '{self._response_buffer}'")
+            logger.trace(f"{self}: Buffer: '{self._response_buffer}'")
 
         else:
-            # Always push the frame downstream (for context aggregator)
+            # Pass all non-LLM frames through
+            # Blocking LLM frames prevents interferences with the downsteram LLM
             await self.push_frame(frame, direction)
 
     async def _process_classification(self, full_response: str):
-        """Process the complete LLM classification response.
+        """Process the complete LLM classification response and trigger actions.
+
+        Analyzes the aggregated response text to determine if it contains
+        "CONVERSATION" or "VOICEMAIL" and triggers the appropriate notifications
+        and callbacks based on the classification result.
 
         Args:
-            full_response: The complete aggregated response from the LLM.
+            full_response: The complete aggregated response text from the LLM.
         """
         if self._decision_made:
+            logger.debug(f"{self}: Decision already made, ignoring response")
             return
 
         response = full_response.upper()
-        logger.info(f"{self}: Processing classification: '{full_response}'")
+        logger.info(f"{self}: Classifying response: '{full_response}'")
 
         if "CONVERSATION" in response:
+            # Human answered - continue normal conversation flow
             self._decision_made = True
-            logger.info(f"{self}: CONVERSATION detected - releasing buffer")
+            logger.info(f"{self}: CONVERSATION detected - releasing TTS buffer")
-            await self._gate_notifier.notify()
+            await self._gate_notifier.notify()  # Close the classifier gate
-            await self._conversation_notifier.notify()
+            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 - triggering callback")
+            logger.info(f"{self}: VOICEMAIL detected - clearing TTS buffer and triggering callback")
-            await self._gate_notifier.notify()
+            await self._gate_notifier.notify()  # Close the classifier gate
-            await self._voicemail_notifier.notify()
+            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
             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}'")
 
 
 class VoicemailBuffer(FrameProcessor):
     """Buffers TTS frames until voicemail classification decision is made.
 
-    Holds TTS frames in a buffer while voicemail classification is in progress.
+    This processor holds TTS output frames in a buffer while the voicemail
-    Releases all buffered frames when conversation is detected, or keeps them
+    classification is in progress. This prevents audio from being played
-    buffered when voicemail is detected.
+    to the caller before determining if they're human or a voicemail system.
+
+    The buffer operates in two modes based on the classification result:
+
+    - CONVERSATION: Releases all buffered frames to continue normal dialogue
+    - VOICEMAIL: Clears buffered frames since they're not needed for voicemail
+
+    The buffering only applies to TTS-related frames (TTSTextFrame, TTSAudioRawFrame).
+    All other frames pass through immediately to maintain proper pipeline flow.
     """
 
     def __init__(self, conversation_notifier: BaseNotifier, voicemail_notifier: BaseNotifier):
         """Initialize the voicemail buffer.
 
         Args:
-            conversation_notifier: Notifier that signals when to release buffered frames.
+            conversation_notifier: Notifier that signals when a conversation is
-            voicemail_notifier: Notifier that signals when to keep buffered frames.
+                detected and buffered frames should be released for playback.
+            voicemail_notifier: Notifier that signals when voicemail is detected
+                and buffered frames should be cleared (not played).
         """
         super().__init__()
         self._conversation_notifier = conversation_notifier
@@ -225,20 +273,26 @@ class VoicemailBuffer(FrameProcessor):
         self._voicemail_task: Optional[asyncio.Task] = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames and handle buffering logic.
+        """Process frames and handle buffering logic based on frame type.
+
+        TTS frames are buffered while classification is active. All other frames
+        pass through immediately. The buffering state is controlled by the
+        classification notifications.
 
         Args:
             frame: The frame to process.
-            direction: The direction of frame flow.
+            direction: The direction of frame flow in the pipeline.
         """
         await super().process_frame(frame, direction)
 
         if isinstance(frame, StartFrame):
+            # Start notification waiting tasks for both conversation and voicemail
             self._conversation_task = self.create_task(self._wait_for_conversation())
             self._voicemail_task = self.create_task(self._wait_for_voicemail())
-            logger.info(f"{self}: Buffer tasks started")
             await self.push_frame(frame, direction)
+
         elif isinstance(frame, (EndFrame, CancelFrame)):
+            # Clean up notification tasks when pipeline ends
             if self._conversation_task:
                 await self.cancel_task(self._conversation_task)
                 self._conversation_task = None
@@ -247,62 +301,109 @@ class VoicemailBuffer(FrameProcessor):
                 self._voicemail_task = None
             await self.push_frame(frame, direction)
 
-        # Buffer TTS frames while buffering is active
+        # Core buffering logic: hold TTS frames, pass everything else through
-        if self._buffering_active and isinstance(frame, (TTSTextFrame, TTSAudioRawFrame)):
+        elif self._buffering_active and isinstance(frame, (TTSTextFrame, TTSAudioRawFrame)):
+            # Buffer TTS frames while waiting for classification decision
             self._frame_buffer.append((frame, direction))
         else:
+            # Pass through all non-TTS frames immediately
             await self.push_frame(frame, direction)
 
     async def _wait_for_conversation(self):
-        """Wait for conversation detection - release buffered frames."""
+        """Wait for conversation detection notification and release buffered frames.
+
+        When a conversation is detected, all buffered TTS frames are released
+        in order to continue normal dialogue flow. This allows the bot to
+        respond naturally to the human caller.
+        """
         try:
             await self._conversation_notifier.wait()
-            logger.info(f"{self}: CONVERSATION - releasing frames")
 
+            # Release all buffered frames in original order
             self._buffering_active = False
             for frame, direction in self._frame_buffer:
                 await self.push_frame(frame, direction)
             self._frame_buffer.clear()
 
-            # Cancel the other task
+            # Cancel the voicemail task since decision is final
             if self._voicemail_task:
                 await self.cancel_task(self._voicemail_task)
                 self._voicemail_task = None
 
         except asyncio.CancelledError:
+            logger.debug(f"{self}: Conversation task was cancelled")
             raise
 
     async def _wait_for_voicemail(self):
-        """Wait for voicemail detection - clear buffered frames."""
+        """Wait for voicemail detection notification and clear buffered frames.
+
+        When voicemail is detected, all buffered TTS frames are discarded
+        since they were intended for human conversation and are not appropriate
+        for voicemail systems. The developer callback will handle voicemail-
+        specific audio output.
+        """
         try:
             await self._voicemail_notifier.wait()
-            logger.info(f"{self}: VOICEMAIL - clearing frames")
 
+            # Clear buffered frames without playing them
             self._buffering_active = False
             self._frame_buffer.clear()
 
-            # Cancel the other task
+            # Cancel the conversation task since decision is final
             if self._conversation_task:
                 await self.cancel_task(self._conversation_task)
                 self._conversation_task = None
 
         except asyncio.CancelledError:
+            logger.debug(f"{self}: Voicemail task was cancelled")
             raise
 
 
 class VoicemailDetector(ParallelPipeline):
-    """Parallel pipeline for detecting voicemail vs. live conversation.
+    """Parallel pipeline for detecting voicemail vs. live conversation in outbound calls.
 
-    Uses a parallel pipeline architecture with two branches:
+    This detector uses a parallel pipeline architecture to perform real-time
-    1. Conversation branch: Normal frame flow for live conversations
+    classification of outbound phone calls without interrupting the conversation
-    2. Classification branch: LLM-based classification that can interrupt for voicemail
+    flow. It determines whether a human answered the phone or if the call went
+    to a voicemail system.
 
-    The classifier runs in parallel and makes a one-time decision to either:
+    Architecture:
-    - Continue normal conversation flow (CONVERSATION response)
+
-    - Interrupt and trigger voicemail handling (VOICEMAIL response)
+    - Conversation branch: Empty pipeline that allows normal frame flow
+    - Classification branch: Contains the LLM classifier and decision logic
+
+    The system uses a gate mechanism to control when classification runs and
+    a buffering system to prevent TTS output until classification is complete.
+    Once a decision is made, the appropriate action is taken:
+
+    - CONVERSATION: Continue normal bot dialogue
+    - VOICEMAIL: Trigger developer callback for custom voicemail handling
+
+    Example::
+
+        classification_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        async def handle_voicemail(processor):
+            await processor.push_frame(TTSSpeakFrame("Please leave a message."))
+
+        detector = VoicemailDetector(
+            llm=classification_llm,
+            on_voicemail_detected=handle_voicemail
+        )
+
+        pipeline = Pipeline([
+            transport.input(),
+            stt,
+            detector.detector(),          # Classification
+            context_aggregator.user(),
+            llm,
+            tts,
+            detector.buffer(),            # TTS buffering
+            transport.output(),
+            context_aggregator.assistant(),
+        ])
     """
 
-    # Default prompt
     DEFAULT_SYSTEM_PROMPT = """You are a voicemail detection classifier for an OUTBOUND calling system. A bot has called a phone number and you need to determine if a human answered or if the call went to voicemail based on the provided text.
 
 HUMAN ANSWERED - LIVE CONVERSATION (respond "CONVERSATION"):
@@ -329,25 +430,30 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
         self,
         *,
         llm: LLMService,
-        on_voicemail_detected: Optional[Callable[[], Awaitable[None]]] = None,
+        on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None,
         system_prompt: Optional[str] = None,
     ):
-        """Initialize the voicemail detector.
+        """Initialize the voicemail detector with classification and buffering components.
 
         Args:
-            llm: LLM service for classification.
+            llm: LLM service used for voicemail vs conversation classification.
-            on_voicemail_detected: Callback function called when voicemail is detected.
+                Should be fast and reliable for real-time classification.
-            system_prompt: Optional custom system prompt for classification. If None, uses
+            on_voicemail_detected: Optional callback function invoked when voicemail
-                default prompt optimized for outbound calling scenarios. If providing a
+                is detected. Receives the VoicemailProcessor instance which can be
-                custom prompt, ensure it results in a clear "CONVERSATION" or "VOICEMAIL" response,
+                used to push frames (like custom voicemail greetings).
-                where "CONVERSATION" indicates a human answered and "VOICEMAIL" indicates voicemail.
+            system_prompt: Optional custom system prompt for classification. If None,
+                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.
         """
         self._classifier_llm = llm
         self._prompt = system_prompt if system_prompt is not None else self.DEFAULT_SYSTEM_PROMPT
 
+        # Validate custom prompts to ensure they work with the detection logic
         if system_prompt is not None:
             self._validate_prompt(system_prompt)
 
+        # Set up the LLM context with the classification prompt
         self._messages = [
             {
                 "role": "system",
@@ -355,11 +461,16 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
             },
         ]
 
+        # Create the LLM context and aggregators for conversation management
         self._context = OpenAILLMContext(self._messages)
         self._context_aggregator = llm.create_context_aggregator(self._context)
-        self._gate_notifier = EventNotifier()
+
-        self._conversation_notifier = EventNotifier()
+        # Create notification system for coordinating between components
-        self._voicemail_notifier = EventNotifier()
+        self._gate_notifier = EventNotifier()  # Signals classification completion
+        self._conversation_notifier = EventNotifier()  # Signals conversation detected
+        self._voicemail_notifier = EventNotifier()  # Signals voicemail detected
+
+        # Create the processor components
         self._classifier_gate = ClassifierGate(self._gate_notifier)
         self._voicemail_processor = VoicemailProcessor(
             gate_notifier=self._gate_notifier,
@@ -371,10 +482,11 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
             self._conversation_notifier, self._voicemail_notifier
         )
 
+        # Initialize the parallel pipeline with conversation and classifier branches
         super().__init__(
-            # Conversation branch
+            # Conversation branch: empty pipeline for normal frame flow
             [],
-            # Classifer branch
+            # Classification branch: gate -> context -> LLM -> processor -> context
             [
                 self._classifier_gate,
                 self._context_aggregator.user(),
@@ -385,12 +497,15 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
         )
 
     def _validate_prompt(self, prompt: str) -> None:
-        """Validate custom prompt contains essential instructions.
+        """Validate custom prompt contains required response format instructions.
+
+        Custom prompts must instruct the LLM to respond with exactly "CONVERSATION"
+        or "VOICEMAIL" for the detection logic to work properly. This method
+        checks for the presence of these keywords and warns if they're missing.
 
         Args:
             prompt: The custom system prompt to validate.
         """
-        # Check for exact response format requirements
         has_conversation = "CONVERSATION" in prompt
         has_voicemail = "VOICEMAIL" in prompt
 
@@ -402,15 +517,21 @@ Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it's vo
             )
 
     def detector(self) -> "VoicemailDetector":
-        """Get the detector pipeline (for placement after STT).
+        """Get the detector pipeline for placement after STT in the main pipeline.
+
+        This should be placed after the STT service and before the context
+        aggregator in your main pipeline to enable voicemail classification.
 
         Returns:
-            The VoicemailDetector instance itself.
+            The VoicemailDetector instance itself (which is a ParallelPipeline).
         """
         return self
 
     def buffer(self) -> VoicemailBuffer:
-        """Get the buffer processor (for placement after TTS).
+        """Get the buffer processor for placement after TTS in the main pipeline.
+
+        This should be placed after the TTS service and before the transport
+        output to enable TTS frame buffering during classification.
 
         Returns:
             The VoicemailBuffer processor instance.