async def _stream(
        self, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None
    ) -> AsyncIterator[Mapping[str, Any]]:
        """
        Send a POST request expecting a streaming response with retry logic.

        Args:
            payload (Dict[str, Any]): Parameters for the request payload.
            headers (Optional[Dict[str, str]]): HTTP headers to include in the request.

        Yields:
            Mapping[str, Any]: Decoded JSON response line-by-line.

        Raises:
            ExtendedRequestException: Raised when the request fails.

        """
        logger.debug(f"Starting asynchronous streaming request with payload: {payload}")
        last_exception_text = None
        for attempt in range(1, self.retries + 2):  # retries + initial attempt
            logger.debug(f"Attempt {attempt} for asynchronous streaming request.")
            try:
                async with self._client.stream(
                    "POST",
                    self.endpoint,
                    headers=self._prepare_headers(stream=True, headers=headers),
                    auth=self.auth,
                    json={**payload, "stream": True},
                ) as response:
                    try:
                        logger.debug(
                            f"Received streaming response with status code: {response.status_code}"
                        )
                        response.raise_for_status()
                        async for line in response.aiter_lines():
                            if not line:  # Skip empty lines
                                continue
                            parsed_line = self._parse_streaming_line(line)
                            if parsed_line:
                                logger.debug(f"Yielding parsed line: {parsed_line}")
                                yield parsed_line
                        return
                    except Exception as e:
                        if hasattr(e, "response") and e.response:
                            content = await e.response.aread()
                            last_exception_text = content.decode(
                                e.response.encoding or DEFAULT_ENCODING
                            )
                        raise
            except Exception as e:
                if attempt <= self.retries and _should_retry_exception(e):
                    delay = self.backoff_factor * (2 ** (attempt - 1))
                    logger.warning(
                        f"Streaming attempt {attempt} failed: {e}. Retrying in {delay} seconds..."
                    )
                    await asyncio.sleep(delay)
                else:
                    logger.error(
                        f"Streaming request failed. Error: {e!s}. Details: {last_exception_text}"
                    )
                    raise ExtendedRequestException(
                        f"Streaming request failed: {e!s}. Details: {last_exception_text}",
                        e,
                        last_exception_text,
                    ) from e

        Send a POST request expecting a streaming response with retry logic.

        Args:
            payload (Dict[str, Any]): Parameters for the request payload.
            headers (Optional[Dict[str, str]]): HTTP headers to include in the request.

        Yields:
            Mapping[str, Any]: Decoded JSON response line-by-line.

        Raises:
            ExtendedRequestException: Raised when the request fails.

        
llama-index-integrations/llms/llama-index-llms-oci-data-science/llama_index/llms/oci_data_science/client.py