More updates—added new voicemail module

examples/foundational/voicemail_test.py

@@ -6,38 +6,27 @@
 
 import asyncio
 import os
-from typing import Optional
 
 from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import (
+from pipecat.frames.frames import EndFrame, EndTaskFrame, TTSSpeakFrame
-    BotInterruptionFrame,
+from pipecat.observers.loggers.debug_log_observer import DebugLogObserver
-    CancelFrame,
-    EndFrame,
-    Frame,
-    LLMTextFrame,
-    StartFrame,
-    TTSSpeakFrame,
-)
-from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frame_processor import FrameDirection
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
-from pipecat.services.llm_service import LLMService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.sync.base_notifier import BaseNotifier
-from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
+from pipecat.utils.voicemail.voicemail_detector import VoicemailDetector
 
 load_dotenv(override=True)
 
@@ -63,130 +52,23 @@ transport_params = {
 }
 
 
-class VoicemailDetector(ParallelPipeline):
+async def handle_voicemail(processor):
-    def __init__(self, llm: LLMService):
+    """Called when a voicemail is detected.
-        # Initialize LLM
-        self._classifier_llm = llm
-        self._messages = [
-            {
-                "role": "system",
-                "content": """You are a voicemail detection classifier. Your job is to determine if the caller is leaving a voicemail message or trying to have a live conversation.
 
-VOICEMAIL INDICATORS (respond "YES"):
+    Args:
-- One-way communication (caller talks without expecting immediate responses)
+        processor: The VoicemailProcessor instance. processor.push_frame() is
-- Messages like "Hi, this is [name], please call me back"
+            available to push frames.
-- "I'm calling about..." followed by details without pausing for response
+    """
-- "Leave me a message" or "call me when you get this"
+    logger.info("Voicemail detected! Playing greeting...")
-- Monologue-style speech patterns
-- Mentions of time/date when they're calling
-- Business-like messages with contact information
 
-CONVERSATION INDICATORS (respond "NO"):
+    # Wait a moment for interruption to clear
-- Interactive speech ("Hello?", "Are you there?", "Can you hear me?")
+    await asyncio.sleep(1)
-- Questions directed at the recipient expecting immediate answers
-- Responses to prompts or questions
-- Back-and-forth dialogue patterns
-- Greetings expecting responses ("Hi, how are you?")
-- Real-time problem solving or discussion
 
-Respond with ONLY "YES" if it's a voicemail, or "NO" if it's a conversation attempt. Do not explain your reasoning.""",
+    # Push frames using standard Pipecat pattern
-            },
+    await processor.push_frame(
-        ]
+        TTSSpeakFrame("This is Mattie. Call me back when you can!"),
-        self._context = OpenAILLMContext(self._messages)
+    )
-        self._context_aggregator = llm.create_context_aggregator(self._context)
+    await processor.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
-        self._conversation_notifier = EventNotifier()
-        self._classifier_gate = self.ClassifierGate(self._conversation_notifier)
-        self._voicemail_processor = self.VoicemailProcessor(self._conversation_notifier)
-        self._passthrough_processor = self.PassThroughProcessor()
-
-        super().__init__(
-            # Conversation branch
-            [self._passthrough_processor],
-            # Classifer branch
-            [
-                self._classifier_gate,
-                self._context_aggregator.user(),
-                self._classifier_llm,
-                self._voicemail_processor,
-                self._context_aggregator.assistant(),
-            ],
-        )
-
-    class ClassifierGate(FrameProcessor):
-        def __init__(self, notifier: BaseNotifier):
-            super().__init__()
-            self._notifier = notifier
-            self._gate_opened = True
-            self._gate_task: Optional[asyncio.Task] = None
-
-        async def process_frame(self, frame: Frame, direction: FrameDirection):
-            await super().process_frame(frame, direction)
-
-            if isinstance(frame, StartFrame):
-                # Start the task immediately, don't wait for other conditions
-                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)):
-                if self._gate_task:
-                    await self.cancel_task(self._gate_task)
-                    self._gate_task = None
-
-            if self._gate_opened:
-                await self.push_frame(frame, direction)
-            elif not self._gate_opened and isinstance(frame, BotInterruptionFrame):
-                await self.push_frame(frame, direction)
-
-        async def _wait_for_notification(self):
-            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")
-            except asyncio.CancelledError:
-                logger.debug(f"{self}: Gate task was cancelled")
-                raise
-            except Exception as e:
-                logger.exception(f"{self}: Error in gate task: {e}")
-                raise
-
-    class VoicemailProcessor(FrameProcessor):
-        def __init__(self, notifier: BaseNotifier):
-            super().__init__()
-            self._notifier = notifier
-
-        async def process_frame(self, frame: Frame, direction: FrameDirection):
-            await super().process_frame(frame, direction)
-            if isinstance(frame, LLMTextFrame):
-                # Check if the frame is a NO response, notify the notifier
-                response = frame.text.strip().upper()
-                print(f"Response from LLM: {response}")
-                if "NO" in response:
-                    logger.info(f"{self}: User conversation, notifying to close gate")
-                    await self._notifier.notify()
-                elif "YES" in response:
-                    logger.info(f"{self}: User is leaving a voicemail, push BotInterruptionFrame")
-                    # If the user is leaving a voicemail, we push a BotInterruptionFrame
-                    await self._notifier.notify()
-                    await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
-                    # How do we know when to send this?!
-                    await asyncio.sleep(3)
-                    await self.push_frame(TTSSpeakFrame("This is Mark. Call me back later."))
-
-            else:
-                # Push the frame
-                await self.push_frame(frame, direction)
-
-    class PassThroughProcessor(FrameProcessor):
-        def __init__(self):
-            super().__init__()
-
-        async def process_frame(self, frame: Frame, direction: FrameDirection):
-            await super().process_frame(frame, direction)
-            await self.push_frame(frame, direction)
 
 
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@@ -202,7 +84,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
     voicemail_detector_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-    voicemail_detector = VoicemailDetector(voicemail_detector_llm)
+    voicemail_detector = VoicemailDetector(
+        llm=voicemail_detector_llm, on_voicemail_detected=handle_voicemail
+    )
 
     messages = [
         {
@@ -234,6 +118,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             enable_usage_metrics=True,
         ),
         idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        observers=[
+            DebugLogObserver(
+                frame_types={
+                    EndFrame: None,
+                    EndTaskFrame: None,
+                }
+            ),
+        ],
     )
 
     @transport.event_handler("on_client_connected")

src/pipecat/utils/voicemail/voicemail_detector.py

@@ -0,0 +1,224 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Voicemail detection module for Pipecat.
+
+This module provides voicemail detection capabilities using parallel pipeline
+processing to classify incoming calls as either voicemail messages or live
+conversations.
+"""
+
+import asyncio
+from typing import Awaitable, Callable, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotInterruptionFrame,
+    CancelFrame,
+    CancelTaskFrame,
+    EndFrame,
+    EndTaskFrame,
+    Frame,
+    LLMTextFrame,
+    StartFrame,
+)
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.llm_service import LLMService
+from pipecat.sync.base_notifier import BaseNotifier
+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
+    is made (YES or NO). This ensures the classifier only runs until a definitive
+    decision is reached.
+    """
+
+    def __init__(self, notifier: BaseNotifier):
+        """Initialize the classifier gate.
+
+        Args:
+            notifier: Notifier that signals when to close the gate.
+        """
+        super().__init__()
+        self._notifier = notifier
+        self._gate_opened = True
+        self._gate_task: Optional[asyncio.Task] = None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames and control gate state.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            # Start the task immediately, don't wait for other conditions
+            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)):
+            if self._gate_task:
+                await self.cancel_task(self._gate_task)
+                self._gate_task = None
+
+        if self._gate_opened:
+            await self.push_frame(frame, direction)
+        elif not self._gate_opened and isinstance(
+            frame, (BotInterruptionFrame, EndTaskFrame, EndFrame, CancelTaskFrame, CancelFrame)
+        ):
+            await self.push_frame(frame, direction)
+
+    async def _wait_for_notification(self):
+        """Wait for classification decision notification."""
+        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")
+        except asyncio.CancelledError:
+            logger.debug(f"{self}: Gate task was cancelled")
+            raise
+        except Exception as e:
+            logger.exception(f"{self}: Error in gate task: {e}")
+            raise
+
+
+class VoicemailProcessor(FrameProcessor):
+    """Processor that handles LLM classification responses and triggers callbacks.
+
+    Processes LLM text responses to determine if the call is a voicemail (YES)
+    or conversation (NO), then triggers appropriate actions including developer
+    callbacks for voicemail detection.
+    """
+
+    def __init__(
+        self,
+        notifier: BaseNotifier,
+        on_voicemail_detected: Optional[Callable[["VoicemailProcessor"], Awaitable[None]]] = None,
+    ):
+        """Initialize the voicemail processor.
+
+        Args:
+            notifier: Notifier to signal classification decisions.
+            on_voicemail_detected: Callback function called when voicemail is detected.
+        """
+        super().__init__()
+        self._notifier = notifier
+        self._on_voicemail_detected = on_voicemail_detected
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames and handle LLM classification responses.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMTextFrame):
+            response = frame.text.strip().upper()
+            if "NO" in response:
+                logger.info(f"{self}: CONVERSATION detected - notifying to close gate")
+                await self._notifier.notify()
+            elif "YES" in response:
+                logger.info(f"{self}: VOICEMAIL detected - triggering callback")
+                # Notify gate to close (decision is final)
+                await self._notifier.notify()
+                # Push BotInterruptionFrame to clear the pipeline
+                await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+                # Call developer callback if provided
+                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:
+            # Push the frame
+            await self.push_frame(frame, direction)
+
+
+class VoicemailDetector(ParallelPipeline):
+    """Parallel pipeline for detecting voicemail vs. live conversation.
+
+    Uses a parallel pipeline architecture with two branches:
+    1. Conversation branch: Normal frame flow for live conversations
+    2. Classification branch: LLM-based classification that can interrupt for voicemail
+
+    The classifier runs in parallel and makes a one-time decision to either:
+    - Continue normal conversation flow (NO response)
+    - Interrupt and trigger voicemail handling (YES response)
+    """
+
+    def __init__(
+        self,
+        *,
+        llm: LLMService,
+        on_voicemail_detected: Optional[Callable[[], Awaitable[None]]] = None,
+    ):
+        """Initialize the voicemail detector.
+
+        Args:
+            llm: LLM service for classification.
+            on_voicemail_detected: Callback function called when voicemail is detected.
+        """
+        self._classifier_llm = llm
+        self._messages = [
+            {
+                "role": "system",
+                "content": """You are a voicemail detection classifier. Your job is to determine if the caller is leaving a voicemail message or trying to have a live conversation.
+
+VOICEMAIL INDICATORS (respond "YES"):
+- One-way communication (caller talks without expecting immediate responses)
+- Messages like "Hi, this is [name], please call me back"
+- "I'm calling about..." followed by details without pausing for response
+- "Leave me a message" or "call me when you get this"
+- Monologue-style speech patterns
+- Mentions of time/date when they're calling
+- Business-like messages with contact information
+
+CONVERSATION INDICATORS (respond "NO"):
+- Interactive speech ("Hello?", "Are you there?", "Can you hear me?")
+- Questions directed at the recipient expecting immediate answers
+- Responses to prompts or questions
+- Back-and-forth dialogue patterns
+- Greetings expecting responses ("Hi, how are you?")
+- Real-time problem solving or discussion
+
+Respond with ONLY "YES" if it's a voicemail, or "NO" if it's a conversation attempt. Do not explain your reasoning.""",
+            },
+        ]
+        self._context = OpenAILLMContext(self._messages)
+        self._context_aggregator = llm.create_context_aggregator(self._context)
+        self._conversation_notifier = EventNotifier()
+        self._classifier_gate = ClassifierGate(self._conversation_notifier)
+        self._voicemail_processor = VoicemailProcessor(
+            self._conversation_notifier, on_voicemail_detected
+        )
+
+        super().__init__(
+            # Conversation branch
+            [],
+            # Classifer branch
+            [
+                self._classifier_gate,
+                self._context_aggregator.user(),
+                self._classifier_llm,
+                self._voicemail_processor,
+                self._context_aggregator.assistant(),
+            ],
+        )