Refactor to create task on StartFrame, also cleanup

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

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. 
+            "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.",
-
-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.""",
         },
     ]
 
@@ -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(
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            {
-                "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")

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)):
+        if isinstance(frame, StartFrame):
-            # Flush any pending aggregation before stopping
+            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
+        elif isinstance(frame, InputDTMFFrame):
-
+            # Push the DTMF frame downstream first
-        # Push all other frames downstream immediately
+            await self.push_frame(frame, direction)
-        await self.push_frame(frame, direction)
+            # Then handle it in order for the TranscriptionFrame to be emitted
-
+            # after the InputDTMFFrame
-        if isinstance(frame, InputDTMFFrame):
             await self._handle_dtmf_frame(frame)
+        else:
+            await self.push_frame(frame, direction)
 
-    async def _handle_dtmf_frame(
+    async def _handle_dtmf_frame(self, frame: InputDTMFFrame):
-        self,
+        """Handle DTMF input frame."""
-        frame: InputDTMFFrame,
+        is_first_digit = not self._aggregation
-    ):
-        # Create task on first DTMF input if needed
-        if not self._aggregation_task:
-            self._create_aggregation_task()
 
         digit_value = frame.button.value
         self._aggregation += digit_value
 
-        # If this is the first digit, send BotInterruptionFrame upstream
+        # For first digit, schedule interruption in separate task
-        # But use a separate task to avoid interfering with current frame processing
+        if is_first_digit:
-        if len(self._aggregation) == 1:
+            asyncio.create_task(self._send_interruption_task())
-            # 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()
         else:
-            # Signal new digit received
+            # Signal digit received for timeout handling
             self._digit_event.set()
 
-    async def _send_interruption_frame(self):
+    async def _send_interruption_task(self):
-        """Send an interruption frame in a separate task to avoid queue issues.
+        """Send interruption frame safely in a separate task."""
-
-        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:
+            # 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:
-            while True:
+            try:
-                try:
+                await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout)
-                    await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout)
+                self._digit_event.clear()
-                except asyncio.TimeoutError:
+            except asyncio.TimeoutError:
-                    if self._aggregation:
+                if self._aggregation:
-                        await self._flush_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."""
@@ -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()

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,