async def _stream_audio(self):
        """Handle bi-directional streaming with Google STT."""
        try:
            while True:
                try:
                    # Start bi-directional streaming
                    streaming_recognize = await self._client.streaming_recognize(
                        requests=self._request_generator()
                    )

                    # Process responses
                    await self._process_responses(streaming_recognize)

                    # If we're here, check if we need to reconnect
                    if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
                        logger.debug("Reconnecting stream after timeout")
                        # Reset stream start time
                        self._stream_start_time = int(time.time() * 1000)
                        continue
                    else:
                        # Normal stream end
                        break

                except Exception as e:
                    logger.warning(f"{self} Reconnecting: {e}")

                    await asyncio.sleep(1)  # Brief delay before reconnecting
                    self._stream_start_time = int(time.time() * 1000)
                    continue

        except Exception as e:
            logger.error(f"Error in streaming task: {e}")
            await self.push_frame(ErrorFrame(str(e)))
Handle bi-directional streaming with Google STT.
src/pipecat/services/google/stt.py