async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Process incoming frames.

        This method collects audio frames and processes them when speech ends.

        Args:
            frame: The frame to process
            direction: Direction of the frame (input/output)
        """
        await super().process_frame(frame, direction)

        if isinstance(frame, UserStartedSpeakingFrame):
            logger.info("Speech started")
            self._buffer = AudioBuffer()
            self._buffer.started_at = time.time()

        elif isinstance(frame, AudioRawFrame) and self._buffer.started_at is not None:
            self._buffer.frames.append(frame)

        elif isinstance(frame, UserStoppedSpeakingFrame):
            if self._buffer.frames and not self._buffer.is_processing:
                logger.info("Speech ended, processing buffer...")
                await self.process_generator(self._process_audio_buffer())
                return  # Return early to avoid pushing None frame

        # Only push the original frame if we haven't processed audio
        if frame is not None:
            await self.push_frame(frame, direction)
Process incoming frames.

        This method collects audio frames and processes them when speech ends.

        Args:
            frame: The frame to process
            direction: Direction of the frame (input/output)
        
src/pipecat/services/ultravox/stt.py