diff --git a/CHANGELOG.md b/CHANGELOG.md index ff46e3430..561161be3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `pipecat.extensions.voicemail`, a module for detecting voicemail vs. + live conversation, primarily intended for use in outbound calling scenarios. + The voicemail module is optimized for text LLMs only. + - Added new frames to the `idle_timeout_frames` arg: `TranscriptionFrame`, `InterimTranscriptionFrame`, `UserStartedSpeakingFrame`, and `UserStoppedSpeakingFrame`. These additions serve as indicators of user diff --git a/examples/foundational/44-voicemail-detection.py b/examples/foundational/44-voicemail-detection.py new file mode 100644 index 000000000..1834346ba --- /dev/null +++ b/examples/foundational/44-voicemail-detection.py @@ -0,0 +1,139 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetector +from pipecat.frames.frames import EndTaskFrame, TTSSpeakFrame +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 +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.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + classifier_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + voicemail = VoicemailDetector(llm=classifier_llm) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + stt, + voicemail.detector(), # Voicemail detection — between STT and User context aggregator + context_aggregator.user(), + llm, + tts, + voicemail.gate(), # TTS gating — Immediately after the TTS service + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + @voicemail.event_handler("on_voicemail_detected") + async def handle_voicemail(processor): + logger.info("Voicemail detected! Leaving a message...") + + # Push frames using standard Pipecat pattern + await processor.push_frame( + TTSSpeakFrame( + "Hello, this is Jamie calling about your appointment. Please call me back at 555-0123 when you get this." + ) + ) + + # NOTE: A common pattern is to end pipeline after the voicemail is left. + # Uncomment the following line to end the pipeline after leaving the voicemail. + # await processor.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM) + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/scripts/evals/eval.py b/scripts/evals/eval.py index b91f27c6e..bbc170803 100644 --- a/scripts/evals/eval.py +++ b/scripts/evals/eval.py @@ -89,7 +89,13 @@ class EvalRunner: async def assert_eval_false(self): await self._queue.put(False) - async def run_eval(self, example_file: str, prompt: EvalPrompt, eval: Optional[str] = None): + async def run_eval( + self, + example_file: str, + prompt: EvalPrompt, + eval: Optional[str] = None, + user_speaks_first: bool = False, + ): if not re.match(self._pattern, example_file): return @@ -106,7 +112,9 @@ class EvalRunner: try: tasks = [ asyncio.create_task(run_example_pipeline(script_path)), - asyncio.create_task(run_eval_pipeline(self, example_file, prompt, eval)), + asyncio.create_task( + run_eval_pipeline(self, example_file, prompt, eval, user_speaks_first) + ), ] _, pending = await asyncio.wait(tasks, timeout=EVAL_TIMEOUT_SECS) if pending: @@ -196,6 +204,7 @@ async def run_eval_pipeline( example_file: str, prompt: EvalPrompt, eval: Optional[str], + user_speaks_first: bool = False, ): logger.info(f"Starting eval bot") @@ -225,7 +234,7 @@ async def run_eval_pipeline( tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + voice_id="97f4b8fb-f2fe-444b-bb9a-c109783a857a", # Nathan ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) @@ -260,12 +269,17 @@ async def run_eval_pipeline( # See if we need to include an eval prompt. eval_prompt = "" if eval: - eval_prompt = f"The answer is correct if the user says [{eval}]." + if user_speaks_first: + eval_prompt = f"After the user responds, evaluate if their response is appropriate for the context and matches: [{eval}]." + system_prompt = f"You will start the conversation by saying: '{prompt}'. {eval_prompt} Then call the eval function with your assessment." + else: + eval_prompt = f"The answer is correct if the user says [{eval}]." + system_prompt = f"You are an LLM eval, be extremly brief. Your goal is to only ask one question: {example_prompt}. Call the eval function only if the user answers the question and check if the answer is correct (words as numbers are valid). {eval_prompt}" messages = [ { "role": "system", - "content": f"You are an LLM eval, be extremly brief. Your goal is to only ask one question: {example_prompt}. Call the eval function only if the user answers the question and check if the answer is correct (words as numbers are valid). {eval_prompt}", + "content": system_prompt, }, ] @@ -313,6 +327,14 @@ async def run_eval_pipeline( ) await audio_buffer.start_recording() + # Default behavior is for the bot to speak first + # If the eval bot speaks first, we append the prompt to the messages + if user_speaks_first: + messages.append( + {"role": "user", "content": f"Start by saying this exactly: '{prompt}'"} + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 49acc432a..938200464 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -24,6 +24,9 @@ ASSETS_DIR = SCRIPT_DIR / "assets" FOUNDATIONAL_DIR = SCRIPT_DIR.parent.parent / "examples" / "foundational" +# Speaking order constants +USER_SPEAKS_FIRST = True +BOT_SPEAKS_FIRST = False # Math PROMPT_SIMPLE_MATH = "A simple math addition." @@ -46,115 +49,136 @@ EVAL_SWITCH_LANGUAGE = "Check if the user is now talking in Spanish." PROMPT_VISION = ("What do you see?", Image.open(ASSETS_DIR / "cat.jpg")) EVAL_VISION = "A cat description." +# Voicemail +PROMPT_VOICEMAIL = "Please leave a message after the beep." +EVAL_VOICEMAIL = "Assess the conversation and determine if it is a voicemail." +PROMPT_CONVERSATION = "Hello, this is Mark." +EVAL_CONVERSATION = "A start of a conversation, not a voicemail." + TESTS_07 = [ # 07 series - ("07-interruptible.py", PROMPT_SIMPLE_MATH, None), - ("07-interruptible-cartesia-http.py", PROMPT_SIMPLE_MATH, None), - ("07a-interruptible-speechmatics.py", PROMPT_SIMPLE_MATH, None), - ("07aa-interruptible-soniox.py", PROMPT_SIMPLE_MATH, None), - ("07ab-interruptible-inworld-http.py", PROMPT_SIMPLE_MATH, None), - ("07ac-interruptible-asyncai.py", PROMPT_SIMPLE_MATH, None), - ("07ac-interruptible-asyncai-http.py", PROMPT_SIMPLE_MATH, None), - ("07b-interruptible-langchain.py", PROMPT_SIMPLE_MATH, None), - ("07c-interruptible-deepgram.py", PROMPT_SIMPLE_MATH, None), - ("07d-interruptible-elevenlabs.py", PROMPT_SIMPLE_MATH, None), - ("07d-interruptible-elevenlabs-http.py", PROMPT_SIMPLE_MATH, None), - ("07e-interruptible-playht.py", PROMPT_SIMPLE_MATH, None), - ("07e-interruptible-playht-http.py", PROMPT_SIMPLE_MATH, None), - ("07f-interruptible-azure.py", PROMPT_SIMPLE_MATH, None), - ("07g-interruptible-openai.py", PROMPT_SIMPLE_MATH, None), - ("07h-interruptible-openpipe.py", PROMPT_SIMPLE_MATH, None), - ("07j-interruptible-gladia.py", PROMPT_SIMPLE_MATH, None), - ("07k-interruptible-lmnt.py", PROMPT_SIMPLE_MATH, None), - ("07l-interruptible-groq.py", PROMPT_SIMPLE_MATH, None), - ("07m-interruptible-aws.py", PROMPT_SIMPLE_MATH, None), - ("07n-interruptible-gemini.py", PROMPT_SIMPLE_MATH, None), - ("07n-interruptible-google.py", PROMPT_SIMPLE_MATH, None), - ("07o-interruptible-assemblyai.py", PROMPT_SIMPLE_MATH, None), - ("07q-interruptible-rime.py", PROMPT_SIMPLE_MATH, None), - ("07q-interruptible-rime-http.py", PROMPT_SIMPLE_MATH, None), - ("07r-interruptible-riva-nim.py", PROMPT_SIMPLE_MATH, None), - ("07s-interruptible-google-audio-in.py", PROMPT_SIMPLE_MATH, None), - ("07t-interruptible-fish.py", PROMPT_SIMPLE_MATH, None), - ("07v-interruptible-neuphonic.py", PROMPT_SIMPLE_MATH, None), - ("07v-interruptible-neuphonic-http.py", PROMPT_SIMPLE_MATH, None), - ("07w-interruptible-fal.py", PROMPT_SIMPLE_MATH, None), - ("07y-interruptible-minimax.py", PROMPT_SIMPLE_MATH, None), - ("07z-interruptible-sarvam.py", PROMPT_SIMPLE_MATH, None), + ("07-interruptible.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07-interruptible-cartesia-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07a-interruptible-speechmatics.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07aa-interruptible-soniox.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07ab-interruptible-inworld-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07ac-interruptible-asyncai.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07ac-interruptible-asyncai-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07b-interruptible-langchain.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07c-interruptible-deepgram.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07d-interruptible-elevenlabs.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07d-interruptible-elevenlabs-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07e-interruptible-playht.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07e-interruptible-playht-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07f-interruptible-azure.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07g-interruptible-openai.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07h-interruptible-openpipe.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07j-interruptible-gladia.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07k-interruptible-lmnt.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07l-interruptible-groq.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07m-interruptible-aws.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07n-interruptible-gemini.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07n-interruptible-google.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07o-interruptible-assemblyai.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07q-interruptible-rime.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07q-interruptible-rime-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07r-interruptible-riva-nim.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07s-interruptible-google-audio-in.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07t-interruptible-fish.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07v-interruptible-neuphonic.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07v-interruptible-neuphonic-http.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07w-interruptible-fal.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07y-interruptible-minimax.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("07z-interruptible-sarvam.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), # Needs a local XTTS docker instance running. - # ("07i-interruptible-xtts.py", PROMPT_SIMPLE_MATH, None), + # ("07i-interruptible-xtts.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), # Needs a Krisp license. - # ("07p-interruptible-krisp.py", PROMPT_SIMPLE_MATH, None), + # ("07p-interruptible-krisp.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), # Needs GPU resources. - # ("07u-interruptible-ultravox.py", PROMPT_SIMPLE_MATH, None), + # ("07u-interruptible-ultravox.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), ] TESTS_12 = [ - ("12-describe-video.py", PROMPT_VISION, EVAL_VISION), - ("12a-describe-video-gemini-flash.py", PROMPT_VISION, EVAL_VISION), - ("12b-describe-video-gpt-4o.py", PROMPT_VISION, EVAL_VISION), - ("12c-describe-video-anthropic.py", PROMPT_VISION, EVAL_VISION), + ("12-describe-video.py", PROMPT_VISION, EVAL_VISION, BOT_SPEAKS_FIRST), + ("12a-describe-video-gemini-flash.py", PROMPT_VISION, EVAL_VISION, BOT_SPEAKS_FIRST), + ("12b-describe-video-gpt-4o.py", PROMPT_VISION, EVAL_VISION, BOT_SPEAKS_FIRST), + ("12c-describe-video-anthropic.py", PROMPT_VISION, EVAL_VISION, BOT_SPEAKS_FIRST), ] TESTS_14 = [ - ("14-function-calling.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14a-function-calling-anthropic.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14b-function-calling-anthropic-video.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14d-function-calling-video.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14e-function-calling-google.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14f-function-calling-groq.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14g-function-calling-grok.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14h-function-calling-azure.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14i-function-calling-fireworks.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14j-function-calling-nim.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14m-function-calling-openrouter.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14n-function-calling-perplexity.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14p-function-calling-gemini-vertex-ai.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14q-function-calling-qwen.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14r-function-calling-aws.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14v-function-calling-openai.py", PROMPT_WEATHER, EVAL_WEATHER), - ("14w-function-calling-mistral.py", PROMPT_WEATHER, EVAL_WEATHER), + ("14-function-calling.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14a-function-calling-anthropic.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14b-function-calling-anthropic-video.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14d-function-calling-video.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14e-function-calling-google.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14f-function-calling-groq.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14g-function-calling-grok.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14h-function-calling-azure.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14i-function-calling-fireworks.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14j-function-calling-nim.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14m-function-calling-openrouter.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14n-function-calling-perplexity.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14p-function-calling-gemini-vertex-ai.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14q-function-calling-qwen.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14r-function-calling-aws.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14v-function-calling-openai.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("14w-function-calling-mistral.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), # Currently not working. - # ("14c-function-calling-together.py", PROMPT_WEATHER, EVAL_WEATHER), - # ("14k-function-calling-cerebras.py", PROMPT_WEATHER, EVAL_WEATHER), - # ("14l-function-calling-deepseek.py", PROMPT_WEATHER, EVAL_WEATHER), - # ("14o-function-calling-gemini-openai-format.py", PROMPT_WEATHER, EVAL_WEATHER), + # ("14c-function-calling-together.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + # ("14k-function-calling-cerebras.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + # ("14l-function-calling-deepseek.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + # ("14o-function-calling-gemini-openai-format.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), ] TESTS_15 = [ - ("15a-switch-languages.py", PROMPT_SWITCH_LANGUAGE, EVAL_SWITCH_LANGUAGE), + ("15a-switch-languages.py", PROMPT_SWITCH_LANGUAGE, EVAL_SWITCH_LANGUAGE, BOT_SPEAKS_FIRST), ] TESTS_19 = [ - ("19-openai-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER), - ("19a-azure-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER), - ("19b-openai-realtime-beta-text.py", PROMPT_WEATHER, EVAL_WEATHER), + ("19-openai-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("19a-azure-realtime-beta.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), + ("19b-openai-realtime-beta-text.py", PROMPT_WEATHER, EVAL_WEATHER, BOT_SPEAKS_FIRST), ] TESTS_21 = [ - ("21a-tavus-video-service.py", PROMPT_SIMPLE_MATH, None), + ("21a-tavus-video-service.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), ] TESTS_26 = [ - ("26-gemini-multimodal-live.py", PROMPT_SIMPLE_MATH, None), - ("26a-gemini-multimodal-live-transcription.py", PROMPT_SIMPLE_MATH, None), - ("26b-gemini-multimodal-live-function-calling.py", PROMPT_WEATHER, EVAL_WEATHER), - ("26c-gemini-multimodal-live-video.py", PROMPT_SIMPLE_MATH, None), - ("26e-gemini-multimodal-google-search.py", PROMPT_ONLINE_SEARCH, EVAL_ONLINE_SEARCH), + ("26-gemini-multimodal-live.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ("26a-gemini-multimodal-live-transcription.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ( + "26b-gemini-multimodal-live-function-calling.py", + PROMPT_WEATHER, + EVAL_WEATHER, + BOT_SPEAKS_FIRST, + ), + ("26c-gemini-multimodal-live-video.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), + ( + "26e-gemini-multimodal-google-search.py", + PROMPT_ONLINE_SEARCH, + EVAL_ONLINE_SEARCH, + BOT_SPEAKS_FIRST, + ), # Currently not working. - # ("26d-gemini-multimodal-live-text.py", PROMPT_SIMPLE_MATH, None), + # ("26d-gemini-multimodal-live-text.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), ] TESTS_27 = [ - ("27-simli-layer.py", PROMPT_SIMPLE_MATH, None), + ("27-simli-layer.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), ] TESTS_40 = [ - ("40-aws-nova-sonic.py", PROMPT_SIMPLE_MATH, None), + ("40-aws-nova-sonic.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), ] TESTS_43 = [ - ("43a-heygen-video-service.py", PROMPT_SIMPLE_MATH, None), + ("43a-heygen-video-service.py", PROMPT_SIMPLE_MATH, None, BOT_SPEAKS_FIRST), +] + +TESTS_44 = [ + ("44-voicemail-detection.py", PROMPT_VOICEMAIL, EVAL_VOICEMAIL, USER_SPEAKS_FIRST), + ("44-voicemail-detection.py", PROMPT_CONVERSATION, EVAL_CONVERSATION, USER_SPEAKS_FIRST), ] TESTS = [ @@ -168,6 +192,7 @@ TESTS = [ *TESTS_27, *TESTS_40, *TESTS_43, + *TESTS_44, ] @@ -189,8 +214,11 @@ async def main(args: argparse.Namespace): log_level=log_level, ) - for test, prompt, eval in TESTS: - await runner.run_eval(test, prompt, eval) + # Parse test config: (test, prompt, eval, user_speaks_first) + for test_config in TESTS: + test, prompt, eval, user_speaks_first = test_config + + await runner.run_eval(test, prompt, eval, user_speaks_first) runner.print_results() diff --git a/src/pipecat/extensions/__init__.py b/src/pipecat/extensions/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/extensions/voicemail/__init__.py b/src/pipecat/extensions/voicemail/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/extensions/voicemail/voicemail_detector.py b/src/pipecat/extensions/voicemail/voicemail_detector.py new file mode 100644 index 000000000..12c429a3f --- /dev/null +++ b/src/pipecat/extensions/voicemail/voicemail_detector.py @@ -0,0 +1,707 @@ +# +# 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. 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. + +Note: + The voicemail module is optimized for text LLMs only. +""" + +import asyncio +from typing import List, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + BotInterruptionFrame, + EndFrame, + Frame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + StopFrame, + SystemFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, + TTSTextFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.services.llm_service import LLMService +from pipecat.sync.base_notifier import BaseNotifier +from pipecat.sync.event_notifier import EventNotifier + + +class NotifierGate(FrameProcessor): + """Base gate processor that controls frame flow based on notifier signals. + + This base class provides common gate functionality for processors that need to + start open and close permanently when a notifier signals. Subclasses define + which frames are allowed through when the gate is closed. + + The gate starts open to allow initial processing and closes permanently once + the notifier signals. This ensures controlled frame flow based on external + decisions or events. + """ + + def __init__(self, notifier: BaseNotifier, task_name: str = "gate"): + """Initialize the notifier gate. + + Args: + notifier: Notifier that signals when the gate should close. + task_name: Name for the notification waiting task (for debugging). + """ + super().__init__() + self._notifier = notifier + self._task_name = task_name + self._gate_opened = True + self._gate_task: Optional[asyncio.Task] = None + + async def setup(self, setup: FrameProcessorSetup): + """Set up the processor with required components. + + Args: + setup: Configuration object containing setup parameters. + """ + await super().setup(setup) + self._gate_task = self.create_task(self._wait_for_notification()) + + async def cleanup(self): + """Clean up the processor resources.""" + await super().cleanup() + if self._gate_task: + await self.cancel_task(self._gate_task) + self._gate_task = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and control gate state based on notifier signals. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await super().process_frame(frame, direction) + + # Gate logic: open gate allows all frames, closed gate filters frames + if self._gate_opened: + await self.push_frame(frame, direction) + elif isinstance( + frame, + (SystemFrame, EndFrame, StopFrame), + ): + await self.push_frame(frame, direction) + + async def _wait_for_notification(self): + """Wait for notifier signal and close the gate. + + This method blocks until the notifier signals, then closes the gate + permanently to change frame filtering behavior. + """ + await self._notifier.wait() + + if self._gate_opened: + self._gate_opened = False + + +class ClassifierGate(NotifierGate): + """Gate processor that controls frame flow based on classification decisions. + + Inherits from NotifierGate and starts open to allow initial classification + processing. Closes permanently once a classification decision is made + (CONVERSATION or VOICEMAIL). This ensures the classifier only runs until a + definitive decision is reached, preventing unnecessary LLM calls and maintaining + system efficiency. + + When closed, only allows system frames and user speaking frames to continue. + Speaking frames are needed for voicemail timing control, but not for conversation. + """ + + def __init__(self, gate_notifier: BaseNotifier, conversation_notifier: BaseNotifier): + """Initialize the classifier gate. + + Args: + gate_notifier: Notifier that signals when a classification decision has + been made and the gate should close. + conversation_notifier: Notifier that signals when conversation is detected. + """ + super().__init__(gate_notifier, task_name="classifier_gate") + self._conversation_notifier = conversation_notifier + self._conversation_detected = False + self._conversation_task: Optional[asyncio.Task] = None + + async def setup(self, setup: FrameProcessorSetup): + """Set up the processor with required components. + + Args: + setup: Configuration object containing setup parameters. + """ + await super().setup(setup) + self._conversation_task = self.create_task(self._wait_for_conversation()) + + async def cleanup(self): + """Clean up the processor resources.""" + await super().cleanup() + if self._conversation_task: + await self.cancel_task(self._conversation_task) + self._conversation_task = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and control gate state based on notifier signals. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await FrameProcessor.process_frame(self, frame, direction) + + # Gate logic: open gate allows all frames, closed gate filters frames + if self._gate_opened: + await self.push_frame(frame, direction) + elif isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)): + # Only allow speaking frames if conversation was NOT detected (i.e., voicemail case) + # This prevents the UserContextAggregator from issuing a warning about no aggregation + # to push. + if not self._conversation_detected: + await self.push_frame(frame, direction) + elif isinstance(frame, (SystemFrame, EndFrame, StopFrame)): + # Always allow system frames through + # This includes the UserStartedSpeakingFrame and UserStoppedSpeakingFrame + # which are used to detect voicemail timing. + await self.push_frame(frame, direction) + + async def _wait_for_conversation(self): + """Wait for conversation detection notification and mark conversation detected.""" + await self._conversation_notifier.wait() + self._conversation_detected = True + + +class ConversationGate(NotifierGate): + """Gate processor that blocks conversation flow when voicemail is detected. + + Inherits from NotifierGate and starts open to allow normal conversation + processing. Closes permanently when voicemail is detected to prevent the + main conversation LLM from processing additional input after voicemail + classification. + + When closed, only allows system frames and user speaking frames to continue. + """ + + 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__(voicemail_notifier, task_name="conversation_gate") + + +class ClassificationProcessor(FrameProcessor): + """Processor that handles LLM classification responses and triggers events. + + This processor aggregates LLM text tokens into complete responses and analyzes + them to determine if the call reached a voicemail system or a live person. + 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 event handlers. + + For voicemail detection, the event handler timer starts immediately and is cancelled + and restarted based on user speech patterns to ensure proper timing. + """ + + def __init__( + self, + *, + gate_notifier: BaseNotifier, + conversation_notifier: BaseNotifier, + voicemail_notifier: BaseNotifier, + voicemail_response_delay: float, + ): + """Initialize the voicemail processor. + + Args: + gate_notifier: Notifier to signal the ClassifierGate about classification + decisions so it can close and stop processing. + conversation_notifier: Notifier to signal the TTSGate to release + all gated TTS frames for normal conversation flow. + voicemail_notifier: Notifier to signal the TTSGate to clear + gated TTS frames since voicemail was detected. + voicemail_response_delay: Delay in seconds after user stops speaking + before triggering the voicemail event handler. 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._voicemail_response_delay = voicemail_response_delay + + # Register the voicemail detected event + self._register_event_handler("on_voicemail_detected") + + # 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_task: Optional[asyncio.Task] = None + self._voicemail_event = asyncio.Event() + self._voicemail_event.set() + + async def setup(self, setup: FrameProcessorSetup): + """Set up the processor with required components. + + Args: + setup: Configuration object containing setup parameters. + """ + await super().setup(setup) + self._voicemail_task = self.create_task(self._delayed_voicemail_handler()) + + async def cleanup(self): + """Clean up the processor resources.""" + await super().cleanup() + if self._voicemail_task: + await self.cancel_task(self._voicemail_task) + self._voicemail_task = None + + 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 + 4. UserStartedSpeakingFrame/UserStoppedSpeakingFrame: Manage voicemail timing + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, LLMFullResponseStartFrame): + # Begin aggregating a new LLM response + self._processing_response = True + self._response_buffer = "" + + elif isinstance(frame, LLMFullResponseEndFrame): + # Complete response received - make classification decision + if self._processing_response and not self._decision_made: + await self._process_classification(self._response_buffer.strip()) + self._processing_response = False + self._response_buffer = "" + + elif isinstance(frame, LLMTextFrame) and self._processing_response: + # Accumulate text tokens from the streaming LLM response + self._response_buffer += frame.text + + elif isinstance(frame, UserStartedSpeakingFrame): + # User started speaking - set the voicemail event + if self._voicemail_detected: + self._voicemail_event.set() + + elif isinstance(frame, UserStoppedSpeakingFrame): + # User stopped speaking - clear the voicemail event + if self._voicemail_detected: + self._voicemail_event.clear() + + else: + # Pass all non-LLM frames through + # Blocking LLM frames prevents interference with the downstream LLM + await self.push_frame(frame, direction) + + async def _process_classification(self, full_response: str): + """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 text from the LLM. + """ + if self._decision_made: + return + + response = full_response.upper() + logger.debug(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") + 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 + 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) + + # Set the voicemail event to trigger the voicemail handler + self._voicemail_event.clear() + + else: + # This can happen if the LLM is interrupted before completing the response + logger.debug(f"{self}: No classification found: '{full_response}'") + + async def _delayed_voicemail_handler(self): + """Execute the voicemail event handler after the configured delay. + + This method waits for the specified delay period, then triggers the + developer's voicemail event handler. The timer can be cancelled and restarted + based on user speech patterns to ensure proper timing. + """ + while True: + try: + await asyncio.wait_for( + self._voicemail_event.wait(), timeout=self._voicemail_response_delay + ) + await asyncio.sleep(0.1) + except asyncio.TimeoutError: + await self._call_event_handler("on_voicemail_detected") + break + + +class TTSGate(FrameProcessor): + """Gates TTS frames until voicemail classification decision is made. + + This processor holds TTS output frames in a gate while the voicemail + classification is in progress. This prevents audio from being played + to the caller before determining if they're human or a voicemail system. + + The gate operates in two modes based on the classification result: + + - CONVERSATION: Opens the gate to release all held frames for normal dialogue + - VOICEMAIL: Clears held frames since they're not needed for voicemail + + The gating 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 TTS gate. + + Args: + conversation_notifier: Notifier that signals when a conversation is + detected and gated frames should be released for playback. + voicemail_notifier: Notifier that signals when voicemail is detected + and gated frames should be cleared (not played). + """ + super().__init__() + self._conversation_notifier = conversation_notifier + self._voicemail_notifier = voicemail_notifier + self._frame_buffer: List[tuple[Frame, FrameDirection]] = [] + self._gating_active = True + self._conversation_task: Optional[asyncio.Task] = None + self._voicemail_task: Optional[asyncio.Task] = None + + async def setup(self, setup: FrameProcessorSetup): + """Set up the processor with required components. + + Args: + setup: Configuration object containing setup parameters. + """ + await super().setup(setup) + + self._conversation_task = self.create_task(self._wait_for_conversation()) + self._voicemail_task = self.create_task(self._wait_for_voicemail()) + + async def cleanup(self): + """Clean up the processor resources.""" + await super().cleanup() + if self._conversation_task: + await self.cancel_task(self._conversation_task) + self._conversation_task = None + if self._voicemail_task: + await self.cancel_task(self._voicemail_task) + self._voicemail_task = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and handle gating logic based on frame type. + + TTS frames are gated while classification is active. All other frames + pass through immediately. The gating state is controlled by the + classification notifications. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await super().process_frame(frame, direction) + + # Core gating logic: hold TTS frames, pass everything else through + if self._gating_active and isinstance( + frame, (TTSStartedFrame, TTSStoppedFrame, TTSTextFrame, TTSAudioRawFrame) + ): + # Gate 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 notification and release gated frames. + + When a conversation is detected, all gated TTS frames are released + in order to continue normal dialogue flow. This allows the bot to + respond naturally to the human caller. + """ + await self._conversation_notifier.wait() + + # Release all gated frames in original order + self._gating_active = False + for frame, direction in self._frame_buffer: + await self.push_frame(frame, direction) + self._frame_buffer.clear() + + async def _wait_for_voicemail(self): + """Wait for voicemail detection notification and clear gated frames. + + When voicemail is detected, all gated TTS frames are discarded + since they were intended for human conversation and are not appropriate + for voicemail systems. The developer event handlers will handle voicemail- + specific audio output. + """ + await self._voicemail_notifier.wait() + + # Clear gated frames without playing them + self._gating_active = False + self._frame_buffer.clear() + + +class VoicemailDetector(ParallelPipeline): + """Parallel pipeline for detecting voicemail vs. live conversation in outbound calls. + + This detector uses a parallel pipeline architecture to perform real-time + classification of outbound phone calls without interrupting the conversation + flow. It determines whether a human answered the phone or if the call went + to a voicemail system. + + Architecture: + + - 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 gating 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 event handler for custom voicemail handling + + Example:: + + classification_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + detector = VoicemailDetector(llm=classification_llm) + + @detector.event_handler("on_voicemail_detected") + async def handle_voicemail(processor): + await processor.push_frame(TTSSpeakFrame("Please leave a message.")) + + pipeline = Pipeline([ + transport.input(), + stt, + detector.detector(), # Classification + context_aggregator.user(), + llm, + tts, + detector.gate(), # TTS gating + transport.output(), + context_aggregator.assistant(), + ]) + + # For custom prompts, append the required response instruction: + custom_prompt = "Your custom classification logic here. " + VoicemailDetector.CLASSIFIER_RESPONSE_INSTRUCTION + + Events: + on_voicemail_detected: Triggered when voicemail is detected after the configured + delay. The event handler receives one argument: the ClassificationProcessor + instance which can be used to push frames. + + Constants: + CLASSIFIER_RESPONSE_INSTRUCTION: The exact text that must be included in custom + system prompts to ensure proper classification functionality. + """ + + CLASSIFIER_RESPONSE_INSTRUCTION = 'Respond with ONLY "CONVERSATION" if a person answered, or "VOICEMAIL" if it\'s voicemail/recording.' + + 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"): +- Personal greetings: "Hello?", "Hi", "Yeah?", "John speaking" +- Interactive responses: "Who is this?", "What do you want?", "Can I help you?" +- Conversational tone expecting back-and-forth dialogue +- Questions directed at the caller: "Hello? Anyone there?" +- Informal responses: "Yep", "What's up?", "Speaking" +- Natural, spontaneous speech patterns +- Immediate acknowledgment of the call + +VOICEMAIL SYSTEM (respond "VOICEMAIL"): +- Automated voicemail greetings: "Hi, you've reached [name], please leave a message" +- Phone carrier messages: "The number you have dialed is not in service", "Please leave a message", "All circuits are busy" +- Professional voicemail: "This is [name], I'm not available right now" +- Instructions about leaving messages: "leave a message", "leave your name and number" +- References to callback or messaging: "call me back", "I'll get back to you" +- Carrier system messages: "mailbox is full", "has not been set up" +- Business hours messages: "our office is currently closed" + +""" + + CLASSIFIER_RESPONSE_INSTRUCTION + ) + + def __init__( + self, + *, + llm: LLMService, + voicemail_response_delay: float = 2.0, + custom_system_prompt: Optional[str] = None, + ): + """Initialize the voicemail detector with classification and buffering components. + + Args: + llm: LLM service used for voicemail vs conversation classification. + Should be fast and reliable for real-time classification. + voicemail_response_delay: Delay in seconds after user stops speaking + before triggering the voicemail event handler. 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. + custom_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 = ( + custom_system_prompt if custom_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 custom_system_prompt is not None: + self._validate_prompt(custom_system_prompt) + + # Set up the LLM context with the classification prompt + self._messages = [ + { + "role": "system", + "content": self._prompt, + }, + ] + + # Create the LLM context and aggregators for conversation management + self._context = OpenAILLMContext(self._messages) + self._context_aggregator = llm.create_context_aggregator(self._context) + + # Create notification system for coordinating between components + 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._conversation_notifier) + self._conversation_gate = ConversationGate(self._voicemail_notifier) + self._classification_processor = ClassificationProcessor( + gate_notifier=self._gate_notifier, + conversation_notifier=self._conversation_notifier, + voicemail_notifier=self._voicemail_notifier, + voicemail_response_delay=voicemail_response_delay, + ) + self._voicemail_gate = TTSGate(self._conversation_notifier, self._voicemail_notifier) + + # Initialize the parallel pipeline with conversation and classifier branches + super().__init__( + # Conversation branch: gate to blocks after voicemail detection + [self._conversation_gate], + # Classification branch: gate -> context -> LLM -> processor -> context + [ + self._classifier_gate, + self._context_aggregator.user(), + self._classifier_llm, + self._classification_processor, + self._context_aggregator.assistant(), + ], + ) + + # Register the voicemail detected event after super().__init__() + self._register_event_handler("on_voicemail_detected") + + def _validate_prompt(self, prompt: str) -> None: + """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. + """ + has_conversation = "CONVERSATION" in prompt + has_voicemail = "VOICEMAIL" in prompt + + if not has_conversation or not has_voicemail: + logger.warning( + "Custom system prompt should instruct the LLM to respond with exactly " + '"CONVERSATION" or "VOICEMAIL" for proper detection functionality. ' + f"Consider appending VoicemailDetector.CLASSIFIER_RESPONSE_INSTRUCTION to your prompt: " + f'"{self.CLASSIFIER_RESPONSE_INSTRUCTION}"' + ) + + def detector(self) -> "VoicemailDetector": + """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 (which is a ParallelPipeline). + """ + return self + + def gate(self) -> TTSGate: + """Get the gate 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 gating during classification. + + Returns: + The TTSGate processor instance. + """ + return self._voicemail_gate + + def add_event_handler(self, event_name: str, handler): + """Add an event handler for voicemail detection events. + + Args: + event_name: The name of the event to handle. + handler: The function to call when the event occurs. + """ + if event_name == "on_voicemail_detected": + self._classification_processor.add_event_handler(event_name, handler) + else: + super().add_event_handler(event_name, handler)