From 0ed46f457e9bacf02441d8d8dd33bb5385a77691 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 24 May 2025 10:20:35 -0400 Subject: [PATCH 1/4] Add DTMFAggregator --- CHANGELOG.md | 7 +- .../processors/aggregators/dtmf_aggregator.py | 128 ++++++++++++++++++ 2 files changed, 133 insertions(+), 2 deletions(-) create mode 100644 src/pipecat/processors/aggregators/dtmf_aggregator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 03bc4dae9..89cca2ef5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `OutputDTMFUrgentFrame` to send a DTMF keypress quickly. The previous `OutputDTMFFrame` queues the keypress with the rest of data frames. +- Added `DTMFAggregator`, which aggregates keypad presses into + `TranscriptionFrame`s. Aggregation occurs after a timeout, termination key + press, or user interruption. You can specify the prefix of the + `TranscriptionFrame`. + - Added new functions `DailyTransport.start_transcription()` and `DailyTransport.stop_transcription()` to be able to start and stop Daily transcription dynamically (maybe with different settings). @@ -45,8 +50,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [0.0.68] - 2025-05-28 -### Added - - Added `GoogleHttpTTSService` which uses Google's HTTP TTS API. - Added `TavusTransport`, a new transport implementation compatible with any diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py new file mode 100644 index 000000000..46dd7db7b --- /dev/null +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -0,0 +1,128 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +from typing import Optional + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InputDTMFFrame, + KeypadEntry, + StartInterruptionFrame, + TranscriptionFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.time import time_now_iso8601 + + +class DTMFAggregator(FrameProcessor): + """Aggregates DTMF frames into meaningful sequences for LLM processing. + + The aggregator accumulates digits from InputDTMFFrame instances and flushes + when: + - Timeout occurs (configurable idle period) + - Termination digit is received (default: '#') + - Interruption occurs + + Emits TranscriptionFrame for compatibility with existing LLM context aggregators. + + Args: + timeout: Idle timeout in seconds before flushing + termination_digit: Digit that triggers immediate flush + prefix: Prefix added to DTMF sequence in transcription + """ + + def __init__( + self, + timeout: float = 2.0, + termination_digit: KeypadEntry = KeypadEntry.POUND, + prefix: str = "DTMF: ", + **kwargs, + ): + super().__init__(**kwargs) + + self._aggregation = "" + self._idle_timeout = timeout + self._termination_digit = termination_digit + self._prefix = prefix + + self._digit_event = asyncio.Event() + self._aggregation_task = None + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + + if isinstance(frame, InputDTMFFrame): + await self._handle_dtmf_frame(frame, direction) + elif isinstance(frame, StartInterruptionFrame): + # Flush on interruption + if self._aggregation: + await self._flush_aggregation(direction) + elif isinstance(frame, (EndFrame, CancelFrame)): + # Flush any pending aggregation + if self._aggregation: + await self._flush_aggregation(direction) + await self._stop_aggregation_task() + + # Push frame + await self.push_frame(frame, direction) + + async def _handle_dtmf_frame(self, frame: InputDTMFFrame, direction: FrameDirection): + # Start aggregation task if not running + if self._aggregation_task is None: + self._aggregation_task = self.create_task(self._aggregation_task_handler(direction)) + + # Add digit to aggregation + digit_value = frame.button.value + self._aggregation += digit_value + + # Check for immediate flush conditions + if frame.button == self._termination_digit: + await self._flush_aggregation(direction) + else: + # Signal new digit received + self._digit_event.set() + + async def _aggregation_task_handler(self, direction: FrameDirection): + """Background task that handles timeout-based flushing.""" + while True: + try: + await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) + self._digit_event.clear() + except asyncio.TimeoutError: + if self._aggregation: + await self._flush_aggregation(direction) + + async def _flush_aggregation(self, direction: FrameDirection): + """Flush the current aggregation as a TranscriptionFrame.""" + if not self._aggregation: + return + + sequence = self._aggregation + + # Create transcription with prefix for LLM context + transcription_text = f"{self._prefix}{sequence}" + + # Create and push transcription frame + transcription_frame = TranscriptionFrame( + text=transcription_text, user_id="", timestamp=time_now_iso8601() + ) + + await self.push_frame(transcription_frame, direction) + + # Reset aggregation + self._aggregation = "" + + async def _stop_aggregation_task(self): + if self._aggregation_task: + await self.cancel_task(self._aggregation_task) + self._aggregation_task = None + + async def cleanup(self) -> None: + await super().cleanup() + await self._stop_aggregation_task() From 74827f983fff46f7974e1f3f34434c6bb56a14a7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 24 May 2025 10:43:09 -0400 Subject: [PATCH 2/4] Add tests, improve frame timing --- .../processors/aggregators/dtmf_aggregator.py | 76 +++--- tests/test_dtmf_aggregator.py | 257 ++++++++++++++++++ 2 files changed, 301 insertions(+), 32 deletions(-) create mode 100644 tests/test_dtmf_aggregator.py diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py index 46dd7db7b..9b9ef6dc7 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -45,60 +45,81 @@ class DTMFAggregator(FrameProcessor): **kwargs, ): super().__init__(**kwargs) - self._aggregation = "" self._idle_timeout = timeout self._termination_digit = termination_digit self._prefix = prefix self._digit_event = asyncio.Event() - self._aggregation_task = None + self._aggregation_task: Optional[asyncio.Task] = None + + def _create_aggregation_task(self) -> None: + """Creates the aggregation task if it hasn't been created yet.""" + if not self._aggregation_task: + self._aggregation_task = self.create_task(self._aggregation_task_handler()) + + async def _stop(self) -> None: + """Stops and cleans up the aggregation task.""" + if self._aggregation_task: + await self.cancel_task(self._aggregation_task) + self._aggregation_task = None + + async def cleanup(self) -> None: + await super().cleanup() + if self._aggregation_task: + await self._stop() async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) - if isinstance(frame, InputDTMFFrame): - await self._handle_dtmf_frame(frame, direction) + if isinstance(frame, (EndFrame, CancelFrame)): + # Flush any pending aggregation before stopping + if self._aggregation: + await self._flush_aggregation() + await self._stop() + await self.push_frame(frame, direction) + return elif isinstance(frame, StartInterruptionFrame): - # Flush on interruption if self._aggregation: - await self._flush_aggregation(direction) - elif isinstance(frame, (EndFrame, CancelFrame)): - # Flush any pending aggregation - if self._aggregation: - await self._flush_aggregation(direction) - await self._stop_aggregation_task() + await self._flush_aggregation() + await self.push_frame(frame, direction) + return - # Push frame await self.push_frame(frame, direction) - async def _handle_dtmf_frame(self, frame: InputDTMFFrame, direction: FrameDirection): - # Start aggregation task if not running - if self._aggregation_task is None: - self._aggregation_task = self.create_task(self._aggregation_task_handler(direction)) + if isinstance(frame, InputDTMFFrame): + await self._handle_dtmf_frame(frame) + + async def _handle_dtmf_frame( + self, + frame: InputDTMFFrame, + ): + # Create task on first DTMF input if needed + if not self._aggregation_task: + self._create_aggregation_task() - # Add digit to aggregation digit_value = frame.button.value self._aggregation += digit_value # Check for immediate flush conditions if frame.button == self._termination_digit: - await self._flush_aggregation(direction) + await self._flush_aggregation() else: # Signal new digit received self._digit_event.set() - async def _aggregation_task_handler(self, direction: FrameDirection): + 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) - self._digit_event.clear() except asyncio.TimeoutError: if self._aggregation: - await self._flush_aggregation(direction) + await self._flush_aggregation() + finally: + self._digit_event.clear() - async def _flush_aggregation(self, direction: FrameDirection): + async def _flush_aggregation(self): """Flush the current aggregation as a TranscriptionFrame.""" if not self._aggregation: return @@ -113,16 +134,7 @@ class DTMFAggregator(FrameProcessor): text=transcription_text, user_id="", timestamp=time_now_iso8601() ) - await self.push_frame(transcription_frame, direction) + await self.push_frame(transcription_frame) # Reset aggregation self._aggregation = "" - - async def _stop_aggregation_task(self): - if self._aggregation_task: - await self.cancel_task(self._aggregation_task) - self._aggregation_task = None - - async def cleanup(self) -> None: - await super().cleanup() - await self._stop_aggregation_task() diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py new file mode 100644 index 000000000..4f176aa65 --- /dev/null +++ b/tests/test_dtmf_aggregator.py @@ -0,0 +1,257 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.frames.frames import ( + EndFrame, + InputDTMFFrame, + KeypadEntry, + StartInterruptionFrame, + TranscriptionFrame, +) +from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator +from pipecat.tests.utils import SleepFrame, run_test + + +class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): + async def test_basic_aggregation_with_pound(self): + """Test basic DTMF aggregation ending with pound key.""" + aggregator = DTMFAggregator() + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.THREE), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + # Find and verify the TranscriptionFrame + 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: 123#") + + async def test_timeout_aggregation(self): + """Test DTMF aggregation with timeout flush.""" + aggregator = DTMFAggregator(timeout=0.1) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + SleepFrame(sleep=0.2), # This should trigger timeout + InputDTMFFrame(button=KeypadEntry.THREE), + SleepFrame(sleep=0.2), # This should trigger another timeout + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # First aggregation "12" + InputDTMFFrame, + TranscriptionFrame, # Second aggregation "3" + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + # Find the TranscriptionFrames + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 2) + self.assertEqual(transcription_frames[0].text, "DTMF: 12") + self.assertEqual(transcription_frames[1].text, "DTMF: 3") + + async def test_multiple_aggregations(self): + """Test multiple DTMF sequences with pound termination.""" + aggregator = DTMFAggregator(timeout=0.1) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.POUND), # First sequence + InputDTMFFrame(button=KeypadEntry.FOUR), + InputDTMFFrame(button=KeypadEntry.FIVE), + SleepFrame(sleep=0.2), # Second sequence via timeout + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # "12#" + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # "45" + ] + + 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), 2) + self.assertEqual(transcription_frames[0].text, "DTMF: 12#") + self.assertEqual(transcription_frames[1].text, "DTMF: 45") + + async def test_end_frame_flush(self): + """Test that EndFrame 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 + EndFrame(), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # Should flush before EndFrame + EndFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + send_end_frame=False, # We're sending one in the test to test EndFrame logic + ) + + 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_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: ") + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + 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, "Menu: 1#") + + async def test_custom_termination_digit(self): + """Test custom termination digit configuration.""" + aggregator = DTMFAggregator(termination_digit=KeypadEntry.STAR) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.STAR), # Custom terminator + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + 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_all_keypad_entries(self): + """Test all possible keypad entries.""" + aggregator = DTMFAggregator() + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ZERO), + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.THREE), + InputDTMFFrame(button=KeypadEntry.FOUR), + InputDTMFFrame(button=KeypadEntry.FIVE), + InputDTMFFrame(button=KeypadEntry.SIX), + InputDTMFFrame(button=KeypadEntry.SEVEN), + InputDTMFFrame(button=KeypadEntry.EIGHT), + InputDTMFFrame(button=KeypadEntry.NINE), + InputDTMFFrame(button=KeypadEntry.STAR), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + + # All the InputDTMFFrames plus one TranscriptionFrame + expected_down_frames = [InputDTMFFrame] * 12 + [TranscriptionFrame] + + 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: 0123456789*#") From 0bec7db03bb76b948d9dabb878157ce1983d75a0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 28 May 2025 19:36:00 -0400 Subject: [PATCH 3/4] 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: ") From 4d2a02f318984825d8682dbfbdd92c90885ca1fd Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 28 May 2025 22:23:19 -0400 Subject: [PATCH 4/4] Refactor to create task on StartFrame, also cleanup --- CHANGELOG.md | 2 + examples/twilio-chatbot/bot.py | 24 +--- .../processors/aggregators/dtmf_aggregator.py | 115 ++++++++---------- tests/test_dtmf_aggregator.py | 5 +- 4 files changed, 55 insertions(+), 91 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 89cca2ef5..b1f496127 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [0.0.68] - 2025-05-28 +### Added + - Added `GoogleHttpTTSService` which uses Google's HTTP TTS API. - Added `TavusTransport`, a new transport implementation compatible with any diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index 01fbbc59b..8aa73a2be 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -19,7 +19,6 @@ 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 @@ -84,23 +83,10 @@ 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. - -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.""", + "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.", }, ] @@ -114,7 +100,6 @@ Respond to both voice and keypad input in short sentences.""", 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 @@ -139,12 +124,7 @@ Respond to both voice and keypad input in short sentences.""", # Start recording. await audiobuffer.start_recording() # Kick off the conversation. - 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.", - } - ) + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) 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 889966966..cc6218a1f 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -14,6 +14,7 @@ from pipecat.frames.frames import ( Frame, InputDTMFFrame, KeypadEntry, + StartFrame, TranscriptionFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -53,93 +54,73 @@ class DTMFAggregator(FrameProcessor): self._digit_event = asyncio.Event() self._aggregation_task: Optional[asyncio.Task] = None - def _create_aggregation_task(self) -> None: - """Creates the aggregation task if it hasn't been created yet.""" - if not self._aggregation_task: - self._aggregation_task = self.create_task(self._aggregation_task_handler()) - - async def _stop(self) -> None: - """Stops and cleans up the aggregation task.""" - if self._aggregation_task: - await self.cancel_task(self._aggregation_task) - self._aggregation_task = None - - async def cleanup(self) -> None: - await super().cleanup() - if self._aggregation_task: - await self._stop() - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) - if isinstance(frame, (EndFrame, CancelFrame)): - # Flush any pending aggregation before stopping + if isinstance(frame, StartFrame): + self._create_aggregation_task() + await self.push_frame(frame, direction) + elif isinstance(frame, (EndFrame, CancelFrame)): if self._aggregation: await self._flush_aggregation() - await self._stop() + await self._stop_aggregation_task() await self.push_frame(frame, direction) - return - - # Push all other frames downstream immediately - await self.push_frame(frame, direction) - - if isinstance(frame, InputDTMFFrame): + elif isinstance(frame, InputDTMFFrame): + # Push the DTMF frame downstream first + await self.push_frame(frame, direction) + # Then handle it in order for the TranscriptionFrame to be emitted + # after the InputDTMFFrame await self._handle_dtmf_frame(frame) + else: + await self.push_frame(frame, direction) - async def _handle_dtmf_frame( - self, - frame: InputDTMFFrame, - ): - # Create task on first DTMF input if needed - if not self._aggregation_task: - self._create_aggregation_task() + async def _handle_dtmf_frame(self, frame: InputDTMFFrame): + """Handle DTMF input frame.""" + is_first_digit = not self._aggregation 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()) + # For first digit, schedule interruption in separate task + if is_first_digit: + asyncio.create_task(self._send_interruption_task()) # Check for immediate flush conditions if frame.button == self._termination_digit: await self._flush_aggregation() else: - # Signal new digit received + # Signal digit received for timeout handling 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. - """ + async def _send_interruption_task(self): + """Send interruption frame safely in a separate task.""" try: + # Send the interruption frame await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) except Exception as e: - print(f"Error sending interruption frame: {e}") + # Log error but don't propagate + print(f"Error sending interruption: {e}") + + def _create_aggregation_task(self) -> None: + """Creates the aggregation task if it hasn't been created yet.""" + if not self._aggregation_task: + self._aggregation_task = self.create_task(self._aggregation_task_handler()) + + async def _stop_aggregation_task(self) -> None: + """Stops the aggregation task.""" + if self._aggregation_task: + await self.cancel_task(self._aggregation_task) + self._aggregation_task = None async def _aggregation_task_handler(self): """Background task that handles timeout-based flushing.""" - 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}") + while True: + try: + await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) + self._digit_event.clear() + except asyncio.TimeoutError: + if self._aggregation: + await self._flush_aggregation() async def _flush_aggregation(self): """Flush the current aggregation as a TranscriptionFrame.""" @@ -147,16 +128,16 @@ class DTMFAggregator(FrameProcessor): return sequence = self._aggregation - - # Create transcription with prefix for LLM context transcription_text = f"{self._prefix}{sequence}" - # Create and push transcription frame transcription_frame = TranscriptionFrame( text=transcription_text, user_id="", timestamp=time_now_iso8601() ) - await self.push_frame(transcription_frame) - # Reset aggregation self._aggregation = "" + + async def cleanup(self) -> None: + """Clean up resources.""" + await super().cleanup() + await self._stop_aggregation_task() diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py index 4fe23550a..d2e1cc9aa 100644 --- a/tests/test_dtmf_aggregator.py +++ b/tests/test_dtmf_aggregator.py @@ -81,14 +81,15 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): async def test_multiple_aggregations(self): """Test multiple DTMF sequences with pound termination.""" - aggregator = DTMFAggregator(timeout=0.1) + aggregator = DTMFAggregator(timeout=0.2) frames_to_send = [ InputDTMFFrame(button=KeypadEntry.ONE), InputDTMFFrame(button=KeypadEntry.TWO), InputDTMFFrame(button=KeypadEntry.POUND), # First sequence + SleepFrame(sleep=0.1), InputDTMFFrame(button=KeypadEntry.FOUR), InputDTMFFrame(button=KeypadEntry.FIVE), - SleepFrame(sleep=0.2), # Second sequence via timeout + SleepFrame(sleep=0.3), # Second sequence via timeout ] expected_down_frames = [ InputDTMFFrame,