From 0bec7db03bb76b948d9dabb878157ce1983d75a0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 28 May 2025 19:36:00 -0400 Subject: [PATCH] Emit a BotInterruptionFrame when the first keypress of a sequence is received --- examples/twilio-chatbot/bot.py | 24 ++++++++- .../processors/aggregators/dtmf_aggregator.py | 52 +++++++++++++------ tests/test_dtmf_aggregator.py | 29 ----------- 3 files changed, 59 insertions(+), 46 deletions(-) diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index 8aa73a2be..01fbbc59b 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -19,6 +19,7 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.serializers.twilio import TwilioFrameSerializer @@ -83,10 +84,23 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, call_sid: str, t push_silence_after_stop=testing, ) + # Create DTMF aggregator + dtmf_aggregator = DTMFAggregator( + timeout=3.0, # 3 second timeout + prefix="Menu selection: ", # Helpful prefix for LLM + ) + messages = [ { "role": "system", - "content": "You are an elementary teacher in an audio call. Your output will be converted to audio so don't include special characters in your answers. Respond to what the student said in a short short sentence.", + "content": """You are an elementary teacher in an audio call. Your output will be converted to audio so don't include special characters in your answers. + +When you receive input starting with "Menu selection:", this represents button presses on the phone keypad. For example: +- "Menu selection: 1" means they pressed button 1 +- "Menu selection: 123#" means they pressed 1, 2, 3, then # (pound) +- Common patterns: single digits for menu choices, sequences ending with # for completed entries + +Respond to both voice and keypad input in short sentences.""", }, ] @@ -100,6 +114,7 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, call_sid: str, t pipeline = Pipeline( [ transport.input(), # Websocket input from client + dtmf_aggregator, # DTMF aggregator (processes DTMF before STT) stt, # Speech-To-Text context_aggregator.user(), llm, # LLM @@ -124,7 +139,12 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, call_sid: str, t # Start recording. await audiobuffer.start_recording() # Kick off the conversation. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + messages.append( + { + "role": "system", + "content": "Please introduce yourself to the user and mention they can use voice or press numbers on their phone keypad to interact.", + } + ) await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_client_disconnected") diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py index 9b9ef6dc7..889966966 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -8,12 +8,12 @@ import asyncio from typing import Optional from pipecat.frames.frames import ( + BotInterruptionFrame, CancelFrame, EndFrame, Frame, InputDTMFFrame, KeypadEntry, - StartInterruptionFrame, TranscriptionFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -27,7 +27,7 @@ class DTMFAggregator(FrameProcessor): when: - Timeout occurs (configurable idle period) - Termination digit is received (default: '#') - - Interruption occurs + - EndFrame or CancelFrame is received Emits TranscriptionFrame for compatibility with existing LLM context aggregators. @@ -79,12 +79,8 @@ class DTMFAggregator(FrameProcessor): await self._stop() await self.push_frame(frame, direction) return - elif isinstance(frame, StartInterruptionFrame): - if self._aggregation: - await self._flush_aggregation() - await self.push_frame(frame, direction) - return + # Push all other frames downstream immediately await self.push_frame(frame, direction) if isinstance(frame, InputDTMFFrame): @@ -101,6 +97,12 @@ class DTMFAggregator(FrameProcessor): digit_value = frame.button.value self._aggregation += digit_value + # If this is the first digit, send BotInterruptionFrame upstream + # But use a separate task to avoid interfering with current frame processing + if len(self._aggregation) == 1: + # Use create_task to avoid queue issues + self.create_task(self._send_interruption_frame()) + # Check for immediate flush conditions if frame.button == self._termination_digit: await self._flush_aggregation() @@ -108,16 +110,36 @@ class DTMFAggregator(FrameProcessor): # Signal new digit received self._digit_event.set() + async def _send_interruption_frame(self): + """Send an interruption frame in a separate task to avoid queue issues. + + This interruption frame allows for the user to interrupt the bot's speaking + with a keypress. Without this, the TranscriptionFrame generated is accompanied + by EmulatedUserStarted/StoppedSpeakingFrames, which the bot currently ignores. + This treats the keypress as an explicit input which the bot should respond to. + """ + try: + await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + except Exception as e: + print(f"Error sending interruption frame: {e}") + async def _aggregation_task_handler(self): """Background task that handles timeout-based flushing.""" - while True: - try: - await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) - except asyncio.TimeoutError: - if self._aggregation: - await self._flush_aggregation() - finally: - self._digit_event.clear() + try: + while True: + try: + await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) + except asyncio.TimeoutError: + if self._aggregation: + await self._flush_aggregation() + finally: + self._digit_event.clear() + except asyncio.CancelledError: + # Task was cancelled - exit cleanly + pass + except Exception as e: + # Log unexpected errors but don't crash + print(f"Unexpected error in DTMF aggregation task: {e}") async def _flush_aggregation(self): """Flush the current aggregation as a TranscriptionFrame.""" diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py index 4f176aa65..4fe23550a 100644 --- a/tests/test_dtmf_aggregator.py +++ b/tests/test_dtmf_aggregator.py @@ -10,7 +10,6 @@ from pipecat.frames.frames import ( EndFrame, InputDTMFFrame, KeypadEntry, - StartInterruptionFrame, TranscriptionFrame, ) from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator @@ -143,34 +142,6 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(transcription_frames), 1) self.assertEqual(transcription_frames[0].text, "DTMF: 12") - async def test_interruption_frame_flush(self): - """Test that StartInterruptionFrame flushes pending aggregation.""" - aggregator = DTMFAggregator(timeout=1.0) - frames_to_send = [ - InputDTMFFrame(button=KeypadEntry.ONE), - InputDTMFFrame(button=KeypadEntry.TWO), - SleepFrame(sleep=0.1), # Allow time for aggregation - StartInterruptionFrame(), - ] - expected_down_frames = [ - InputDTMFFrame, - InputDTMFFrame, - TranscriptionFrame, # Should flush before interruption - StartInterruptionFrame, - ] - - received_down_frames, _ = await run_test( - aggregator, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - transcription_frames = [ - f for f in received_down_frames if isinstance(f, TranscriptionFrame) - ] - self.assertEqual(len(transcription_frames), 1) - self.assertEqual(transcription_frames[0].text, "DTMF: 12") - async def test_custom_prefix(self): """Test custom prefix configuration.""" aggregator = DTMFAggregator(prefix="Menu: ")